fix(agent): preserve successful siblings during orphan recovery

agent/replay_cleanup.py

@@ -20,6 +20,9 @@ from __future__ import annotations
 import logging
 from typing import Any, Dict, List
 
+from agent.tool_dispatch_helpers import make_tool_result_message
+from agent.tool_result_classification import tool_may_have_side_effect
+
 logger = logging.getLogger(__name__)
 
 
@@ -64,8 +67,40 @@ def strip_interrupted_tool_tails(
                 is_interrupted_tool_result(m.get("content", ""))
                 for m in tool_results
             ):
+                calls = msg.get("tool_calls") or []
+                if any(
+                    tool_may_have_side_effect(
+                        str((call.get("function") or {}).get("name") or "")
+                    )
+                    for call in calls
+                ):
+                    call_names = {
+                        str(call.get("id") or call.get("call_id") or ""): str(
+                            (call.get("function") or {}).get("name") or ""
+                        )
+                        for call in calls
+                    }
+                    cleaned.append(msg)
+                    for tool_result in tool_results:
+                        if not is_interrupted_tool_result(tool_result.get("content", "")):
+                            cleaned.append(tool_result)
+                            continue
+                        recovered = dict(tool_result)
+                        name = call_names.get(str(tool_result.get("tool_call_id") or ""), "")
+                        recovered["effect_disposition"] = (
+                            "unknown" if tool_may_have_side_effect(name) else "none"
+                        )
+                        recovered["content"] = (
+                            "[Orphan recovery: interrupted side-effecting tool may have "
+                            "executed; its effect is UNKNOWN. Inspect state before retrying.]"
+                            if recovered["effect_disposition"] == "unknown"
+                            else "[Orphan recovery: interrupted read-only tool did not complete.]"
+                        )
+                        cleaned.append(recovered)
+                    i = j
+                    continue
                 logger.debug(
-                    "Stripping interrupted assistant→tool replay block "
+                    "Stripping interrupted read-only assistant→tool replay block "
                     "(indices %d–%d, tool_results=%d)",
                     i, j - 1, len(tool_results),
                 )
@@ -116,11 +151,34 @@ def strip_dangling_tool_call_tail(
     ):
         return agent_history
 
+    tool_calls = last.get("tool_calls") or []
+    if any(
+        tool_may_have_side_effect(
+            str((call.get("function") or {}).get("name") or "")
+        )
+        for call in tool_calls
+    ):
+        recovered = list(agent_history)
+        for call in tool_calls:
+            function = call.get("function") or {}
+            name = str(function.get("name") or "unknown")
+            call_id = str(call.get("id") or call.get("call_id") or "")
+            disposition = "unknown" if tool_may_have_side_effect(name) else "none"
+            recovered.append(make_tool_result_message(
+                name,
+                "[Orphan recovery: this tool may have executed before Hermes stopped; "
+                "its effect is UNKNOWN. Inspect current state before retrying.]",
+                call_id,
+                effect_disposition=disposition,
+            ))
+        logger.warning(
+            "Recovered dangling side-effecting tool call(s) as UNKNOWN instead of erasing them"
+        )
+        return recovered
+
     logger.debug(
-        "Stripping dangling unanswered assistant(tool_calls) tail "
-        "(%d call(s)) — process likely killed mid-tool-call by a "
-        "restart/shutdown command (#49201)",
-        len(last.get("tool_calls") or []),
+        "Stripping dangling unanswered read-only assistant(tool_calls) tail (%d call(s))",
+        len(tool_calls),
     )
     return agent_history[:-1]
 

agent/tool_dispatch_helpers.py

@@ -358,7 +358,13 @@ def _trajectory_normalize_msg(msg: Dict[str, Any]) -> Dict[str, Any]:
     return msg
 
 
-def make_tool_result_message(name: str, content: Any, tool_call_id: str) -> dict:
+def make_tool_result_message(
+    name: str,
+    content: Any,
+    tool_call_id: str,
+    *,
+    effect_disposition: str | None = None,
+) -> dict:
     """Build a tool-result message dict with both the OpenAI-format ``name``
     field (required by the wire format and provider adapters) and the internal
     ``tool_name`` field (written to the session DB messages table).
@@ -379,13 +385,16 @@ def make_tool_result_message(name: str, content: Any, tool_call_id: str) -> dict
     callers should compare by value, not by ``is``.
     """
     wrapped = _maybe_wrap_untrusted(name, content)
-    return {
+    message = {
         "role": "tool",
         "name": name,
         "tool_name": name,
         "content": wrapped,
         "tool_call_id": tool_call_id,
     }
+    if effect_disposition is not None:
+        message["effect_disposition"] = effect_disposition
+    return message
 
 
 # Tools whose results carry attacker-controllable content.  Wrapping their

agent/tool_executor.py

@@ -324,6 +324,7 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
                 tc.function.name,
                 f"[Tool execution cancelled — {tc.function.name} was skipped due to user interrupt]",
                 tc.id,
+                effect_disposition="none",
             ))
             _flush_session_db_after_tool_progress(
                 agent,
@@ -798,9 +799,11 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
         # deadline snapshot (timed_out_indices, taken from not_done) and this
         # loop. Prefer that real result over a fabricated timeout message — the
         # tool genuinely succeeded, just slightly late.
+        effect_disposition = None
         if i in timed_out_indices and r is None:
             suffix = f"{timeout_s:.1f}s" if timeout_s is not None else "the configured timeout"
             function_result = f"Error executing tool '{name}': timed out after {suffix}"
+            effect_disposition = "unknown"
             _emit_terminal_post_tool_call(
                 agent,
                 function_name=name,
@@ -847,6 +850,8 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
             tool_duration = 0.0
         else:
             function_name, function_args, function_result, tool_duration, is_error, blocked, middleware_trace = r
+            if blocked:
+                effect_disposition = "none"
 
             if not blocked:
                 function_result = agent._append_guardrail_observation(
@@ -935,7 +940,12 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
         # image tool result never poisons canonical session history.
         # String results pass through unchanged.
         _tool_content = agent._tool_result_content_for_active_model(name, function_result)
-        messages.append(make_tool_result_message(name, _tool_content, tc.id))
+        messages.append(make_tool_result_message(
+            name,
+            _tool_content,
+            tc.id,
+            effect_disposition=effect_disposition,
+        ))
         _flush_session_db_after_tool_progress(
             agent,
             messages,
@@ -980,6 +990,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
                     skipped_name,
                     f"[Tool execution cancelled — {skipped_name} was skipped due to user interrupt]",
                     skipped_tc.id,
+                    effect_disposition="none",
                 ))
                 _flush_session_db_after_tool_progress(
                     agent,
@@ -1615,6 +1626,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
                     skipped_name,
                     f"[Tool execution skipped — {skipped_name} was not started. User sent a new message]",
                     skipped_tc.id,
+                    effect_disposition="none",
                 ))
                 _flush_session_db_after_tool_progress(
                     agent,

agent/tool_result_classification.py

@@ -3,12 +3,60 @@
 from __future__ import annotations
 
 import json
+from enum import Enum
 from typing import Any
 
 
 FILE_MUTATING_TOOL_NAMES = frozenset({"write_file", "patch"})
 
 
+class ToolEffectDisposition(str, Enum):
+    """What Hermes knows about a tool call's externally visible effects."""
+
+    COMMITTED = "committed"
+    NONE = "none"
+    UNKNOWN = "unknown"
+    PARTIAL = "partial"
+    ROLLED_BACK = "rolled_back"
+
+
+# Known observational tools. Unknown/plugin/MCP tools are conservatively treated
+# as effect-capable so an orphan is not silently presented as never having run.
+NO_EFFECT_TOOL_NAMES = frozenset({
+    "read_file", "search_files", "session_search", "skill_view", "skills_list",
+    "web_extract", "web_search", "vision_analyze", "browser_snapshot",
+    "browser_get_images", "browser_console", "todo", "read_terminal",
+})
+
+
+def tool_may_have_side_effect(tool_name: str) -> bool:
+    return tool_name not in NO_EFFECT_TOOL_NAMES
+
+
+def classify_tool_effect(
+    tool_name: str,
+    result: Any,
+    *,
+    status: str = "completed",
+) -> ToolEffectDisposition:
+    """Classify effects independently from a tool's success/failure status."""
+    normalized = (status or "completed").strip().lower()
+    if normalized in {"blocked", "skipped", "not_started"}:
+        return ToolEffectDisposition.NONE
+    if not tool_may_have_side_effect(tool_name):
+        return ToolEffectDisposition.NONE
+    if normalized in {
+        "timeout", "timed_out", "detached", "cancelled", "interrupted",
+        "error", "failed",
+    }:
+        return ToolEffectDisposition.UNKNOWN
+    if normalized == "partial":
+        return ToolEffectDisposition.PARTIAL
+    if normalized in {"rolled_back", "rollback"}:
+        return ToolEffectDisposition.ROLLED_BACK
+    return ToolEffectDisposition.COMMITTED
+
+
 def file_mutation_result_landed(tool_name: str, result: Any) -> bool:
     """Return True when a file mutation result proves the write landed."""
     if tool_name not in FILE_MUTATING_TOOL_NAMES or not isinstance(result, str):

agent/transports/chat_completions.py

@@ -171,6 +171,7 @@ class ChatCompletionsTransport(ProviderTransport):
                 "codex_reasoning_items" in msg
                 or "codex_message_items" in msg
                 or "tool_name" in msg
+                or "effect_disposition" in msg
                 or "timestamp" in msg  # #47868 — strict providers reject this
             ):
                 needs_sanitize = True
@@ -212,12 +213,14 @@ class ChatCompletionsTransport(ProviderTransport):
                 "codex_reasoning_items" in msg
                 or "codex_message_items" in msg
                 or "tool_name" in msg
+                or "effect_disposition" in msg
                 or "timestamp" in msg  # #47868 — leak into strict providers
             ):
                 out_msg = mutable_msg()
                 out_msg.pop("codex_reasoning_items", None)
                 out_msg.pop("codex_message_items", None)
                 out_msg.pop("tool_name", None)
+                out_msg.pop("effect_disposition", None)
                 out_msg.pop("timestamp", None)  # #47868 — leak into strict providers
 
 

hermes_state.py

@@ -753,6 +753,7 @@ CREATE TABLE IF NOT EXISTS messages (
     tool_call_id TEXT,
     tool_calls TEXT,
     tool_name TEXT,
+    effect_disposition TEXT,
     timestamp REAL NOT NULL,
     token_count INTEGER,
     finish_reason TEXT,
@@ -3455,6 +3456,7 @@ class SessionDB:
         codex_message_items: Any = None,
         platform_message_id: str = None,
         observed: bool = False,
+        effect_disposition: Optional[str] = None,
         timestamp: Any = None,
     ) -> int:
         """
@@ -3505,10 +3507,10 @@ class SessionDB:
         def _do(conn):
             cursor = conn.execute(
                 """INSERT INTO messages (session_id, role, content, tool_call_id,
-                   tool_calls, tool_name, timestamp, token_count, finish_reason,
+                   tool_calls, tool_name, effect_disposition, timestamp, token_count, finish_reason,
                    reasoning, reasoning_content, reasoning_details, codex_reasoning_items,
                    codex_message_items, platform_message_id, observed, active)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                 (
                     session_id,
                     role,
@@ -3516,6 +3518,7 @@ class SessionDB:
                     tool_call_id,
                     tool_calls_json,
                     tool_name,
+                    effect_disposition,
                     message_timestamp,
                     token_count,
                     finish_reason,
@@ -3597,10 +3600,10 @@ class SessionDB:
 
             conn.execute(
                 """INSERT INTO messages (session_id, role, content, tool_call_id,
-                   tool_calls, tool_name, timestamp, token_count, finish_reason,
+                   tool_calls, tool_name, effect_disposition, timestamp, token_count, finish_reason,
                    reasoning, reasoning_content, reasoning_details, codex_reasoning_items,
                    codex_message_items, platform_message_id, observed, active)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                 (
                     session_id,
                     role,
@@ -3608,6 +3611,7 @@ class SessionDB:
                     msg.get("tool_call_id"),
                     tool_calls_json,
                     msg.get("tool_name"),
+                    msg.get("effect_disposition"),
                     message_timestamp,
                     msg.get("token_count"),
                     msg.get("finish_reason"),
@@ -4103,7 +4107,7 @@ class SessionDB:
         with self._lock:
             placeholders = ",".join("?" for _ in session_ids)
             rows = self._conn.execute(
-                "SELECT role, content, tool_call_id, tool_calls, tool_name, "
+                "SELECT role, content, tool_call_id, tool_calls, tool_name, effect_disposition, "
                 "finish_reason, reasoning, reasoning_content, reasoning_details, "
                 "codex_reasoning_items, codex_message_items, platform_message_id, observed, timestamp "
                 f"FROM messages WHERE session_id IN ({placeholders})"
@@ -4131,6 +4135,8 @@ class SessionDB:
                 msg["tool_call_id"] = row["tool_call_id"]
             if row["tool_name"]:
                 msg["tool_name"] = row["tool_name"]
+            if row["effect_disposition"]:
+                msg["effect_disposition"] = row["effect_disposition"]
             if row["tool_calls"]:
                 try:
                     msg["tool_calls"] = json.loads(row["tool_calls"])

tests/agent/test_replay_cleanup.py

@@ -40,24 +40,46 @@ def test_is_interrupted_tool_result_markers():
     assert not is_interrupted_tool_result(None)
 
 
-def test_strip_dangling_tool_call_tail_removes_unanswered_tail():
-    history = [_user("hi"), _assistant_tc("write_file")]
+def test_strip_dangling_tool_call_tail_removes_unanswered_read_only_tail():
+    history = [_user("hi"), _assistant_tc("read_file")]
     out = strip_dangling_tool_call_tail(history)
     assert out == [_user("hi")]
 
 
+def test_dangling_side_effect_is_recovered_as_unknown_not_erased():
+    history = [_user("hi"), _assistant_tc("write_file")]
+
+    out = strip_dangling_tool_call_tail(history)
+
+    assert out[:-1] == history
+    assert out[-1]["role"] == "tool"
+    assert out[-1]["tool_call_id"] == "c1"
+    assert out[-1]["effect_disposition"] == "unknown"
+    assert "may have executed" in out[-1]["content"].lower()
+
+
 def test_strip_dangling_tool_call_tail_preserves_answered_pair():
     history = [_user("hi"), _assistant_tc("read_file"), _tool("contents")]
     out = strip_dangling_tool_call_tail(history)
     assert out == history  # answered -> untouched
 
 
-def test_strip_interrupted_tool_tails_removes_interrupted_block():
-    history = [_user("hi"), _assistant_tc("terminal"), _tool("[Command interrupted]")]
+def test_strip_interrupted_tool_tails_removes_interrupted_read_only_block():
+    history = [_user("hi"), _assistant_tc("read_file"), _tool("[Command interrupted]")]
     out = strip_interrupted_tool_tails(history)
     assert out == [_user("hi")]
 
 
+def test_interrupted_side_effect_is_preserved_as_unknown():
+    history = [_user("hi"), _assistant_tc("terminal"), _tool("[Command interrupted]")]
+
+    out = strip_interrupted_tool_tails(history)
+
+    assert out[:-1] == history[:-1]
+    assert out[-1]["role"] == "tool"
+    assert out[-1]["effect_disposition"] == "unknown"
+
+
 def test_strip_interrupted_tool_tails_preserves_successful_block():
     history = [_user("hi"), _assistant_tc("read_file"), _tool("ok"),
                {"role": "assistant", "content": "done"}]
@@ -72,15 +94,20 @@ def test_strip_interrupted_tool_tails_removes_orphan_interrupted_tool():
 
 
 def test_sanitize_replay_history_combines_both():
-    # interrupted block in the middle + dangling tail at the end
+    # interrupted block is removed; a dangling read-only call is safe to erase
     history = [
         _user("first"),
         _assistant_tc("terminal"), _tool("[Command interrupted]"),
         _user("second"),
-        _assistant_tc("write_file"),  # dangling
+        _assistant_tc("read_file"),  # dangling
     ]
     out = sanitize_replay_history(history)
-    assert out == [_user("first"), _user("second")]
+    assert out[:2] == [
+        _user("first"),
+        _assistant_tc("terminal"),
+    ]
+    assert out[2]["effect_disposition"] == "unknown"
+    assert out[-1] == _user("second")
 
 
 def test_sanitize_replay_history_noop_on_clean_history():

tests/agent/test_tool_dispatch_helpers.py

@@ -222,6 +222,12 @@ class TestMakeToolResultMessage:
             "tool_call_id": "call_1",
         }
 
+    def test_effect_disposition_is_internal_message_metadata(self):
+        msg = make_tool_result_message(
+            "terminal", "timed out", "call_effect", effect_disposition="unknown"
+        )
+        assert msg["effect_disposition"] == "unknown"
+
     def test_high_risk_message_content_wrapped(self):
         msg = make_tool_result_message("web_extract", SAMPLE_LONG_TEXT, "call_2")
         assert msg["role"] == "tool"

tests/agent/test_tool_result_classification.py

@@ -2,7 +2,11 @@
 
 import json
 
-from agent.tool_result_classification import file_mutation_result_landed
+from agent.tool_result_classification import (
+    ToolEffectDisposition,
+    classify_tool_effect,
+    file_mutation_result_landed,
+)
 
 
 def test_write_file_with_nested_lint_error_counts_as_landed():
@@ -28,3 +32,21 @@ def test_top_level_file_mutation_error_does_not_count_as_landed():
     result = json.dumps({"success": True, "error": "post-write verification failed"})
 
     assert file_mutation_result_landed("patch", result) is False
+
+
+def test_effect_classifier_distinguishes_all_internal_dispositions():
+    assert classify_tool_effect("patch", '{"success": true}') is ToolEffectDisposition.COMMITTED
+    assert classify_tool_effect("read_file", "contents") is ToolEffectDisposition.NONE
+    assert classify_tool_effect("terminal", "timed out", status="timeout") is ToolEffectDisposition.UNKNOWN
+    assert classify_tool_effect("patch", "some hunks applied", status="partial") is ToolEffectDisposition.PARTIAL
+    assert classify_tool_effect("patch", "restored checkpoint", status="rolled_back") is ToolEffectDisposition.ROLLED_BACK
+
+
+def test_blocked_or_skipped_tool_never_claims_an_effect():
+    assert classify_tool_effect("terminal", "blocked", status="blocked") is ToolEffectDisposition.NONE
+    assert classify_tool_effect("write_file", "skipped", status="skipped") is ToolEffectDisposition.NONE
+
+
+def test_detached_or_timed_out_side_effect_is_unknown():
+    assert classify_tool_effect("terminal", "started", status="detached") is ToolEffectDisposition.UNKNOWN
+    assert classify_tool_effect("write_file", "late worker", status="timeout") is ToolEffectDisposition.UNKNOWN

tests/agent/transports/test_chat_completions.py

@@ -30,6 +30,19 @@ class TestChatCompletionsBasic:
         result = transport.convert_messages(msgs)
         assert result is msgs  # no copy needed
 
+    def test_convert_messages_strips_internal_effect_disposition(self, transport):
+        msgs = [{
+            "role": "tool",
+            "content": "uncertain",
+            "tool_call_id": "c1",
+            "effect_disposition": "unknown",
+        }]
+
+        result = transport.convert_messages(msgs)
+
+        assert "effect_disposition" not in result[0]
+        assert msgs[0]["effect_disposition"] == "unknown"
+
     def test_convert_messages_strips_codex_fields(self, transport):
         msgs = [
             {"role": "assistant", "content": "ok", "codex_reasoning_items": [{"id": "rs_1"}],

tests/gateway/test_auto_continue.py

@@ -96,7 +96,7 @@ class TestAutoDetection:
 
 
 class TestInterruptedReplayFiltering:
-    def test_interrupted_tool_tail_is_removed_from_agent_history(self):
+    def test_interrupted_side_effect_is_replayed_as_unknown(self):
         from gateway.run import _build_gateway_agent_history
 
         history = [
@@ -118,9 +118,12 @@ class TestInterruptedReplayFiltering:
         agent_history, observed_context = _build_gateway_agent_history(history)
 
         assert observed_context is None
-        assert agent_history == [{"role": "user", "content": "transcribe this video"}]
+        assert agent_history[:2] == history[:2]
+        assert agent_history[-1]["role"] == "tool"
+        assert agent_history[-1]["tool_call_id"] == "call_1"
+        assert agent_history[-1]["effect_disposition"] == "unknown"
 
-    def test_mixed_tail_with_one_interrupted_result_is_removed(self):
+    def test_mixed_tail_preserves_results_and_marks_interrupted_effect_unknown(self):
         from gateway.run import _build_gateway_agent_history
 
         history = [
@@ -143,7 +146,10 @@ class TestInterruptedReplayFiltering:
 
         agent_history, _observed_context = _build_gateway_agent_history(history)
 
-        assert agent_history == [{"role": "user", "content": "search and transcribe"}]
+        assert agent_history[:3] == history[:3]
+        assert agent_history[-1]["role"] == "tool"
+        assert agent_history[-1]["tool_call_id"] == "call_2"
+        assert agent_history[-1]["effect_disposition"] == "unknown"
 
     def test_successful_tool_tail_is_preserved(self):
         from gateway.run import _build_gateway_agent_history
@@ -165,14 +171,14 @@ class TestInterruptedReplayFiltering:
         assert agent_history[-1]["role"] == "tool"
         assert agent_history[-1]["content"] == "deployed successfully"
 
-    def test_dangling_unanswered_tool_call_tail_is_removed(self):
-        """A trailing assistant(tool_calls) with NO tool answers is stripped.
+    def test_dangling_unanswered_side_effect_is_replayed_as_unknown(self):
+        """A trailing side-effecting call gets an UNKNOWN result, not a retry.
 
         This is the SIGKILL signature from #49201: the tool itself ran a
         restart/shutdown command and killed the gateway before its result was
         persisted. The transcript tail is an assistant message with tool_calls
-        and zero matching tool rows. Without stripping it, the model re-issues
-        the unanswered call on resume and loops the restart forever.
+        and zero matching tool rows. A synthetic UNKNOWN result closes the tool
+        pair without claiming the restart did not happen or inviting a retry.
         """
         from gateway.run import _build_gateway_agent_history
 
@@ -195,13 +201,16 @@ class TestInterruptedReplayFiltering:
 
         agent_history, _observed_context = _build_gateway_agent_history(history)
 
-        assert agent_history == [{"role": "user", "content": "restart the container"}]
+        assert agent_history[:2] == history
+        assert agent_history[-1]["role"] == "tool"
+        assert agent_history[-1]["tool_call_id"] == "call_1"
+        assert agent_history[-1]["effect_disposition"] == "unknown"
 
-    def test_dangling_tail_after_completed_pair_is_removed_only_at_tail(self):
-        """Only the trailing unanswered tool-call block is stripped.
+    def test_dangling_tail_after_completed_pair_gets_unknown_result(self):
+        """The completed pair survives and the trailing call becomes UNKNOWN.
 
-        An earlier completed assistant→tool pair must survive — we only drop
-        the final assistant(tool_calls) that has no answers.
+        An earlier completed assistant→tool pair must survive, and the final
+        assistant(tool_calls) receives a matching UNKNOWN result.
         """
         from gateway.run import _build_gateway_agent_history
 
@@ -232,18 +241,19 @@ class TestInterruptedReplayFiltering:
 
         agent_history, _observed_context = _build_gateway_agent_history(history)
 
-        # The completed call_1 pair survives; the dangling call_2 tail is gone.
+        # The completed call_1 pair survives; call_2 is closed truthfully.
         assert agent_history[-1]["role"] == "tool"
-        assert agent_history[-1]["content"] == "found it"
-        # The surviving assistant(tool_calls) is the completed call_1 (which
-        # has a matching tool answer), not the stripped dangling call_2.
+        assert agent_history[-1]["tool_call_id"] == "call_2"
+        assert agent_history[-1]["effect_disposition"] == "unknown"
+        assert agent_history[2]["content"] == "found it"
+        # Both assistant calls survive with matching tool results.
         _surviving_calls = [
             tc.get("id")
             for m in agent_history
             if m.get("role") == "assistant" and m.get("tool_calls")
             for tc in m["tool_calls"]
         ]
-        assert _surviving_calls == ["call_1"]
+        assert _surviving_calls == ["call_1", "call_2"]
 
     def test_persisted_auto_continue_note_is_not_replayed(self):
         from gateway.run import _build_gateway_agent_history

tests/run_agent/test_run_agent.py

@@ -2776,6 +2776,7 @@ class TestConcurrentToolExecution:
         assert "fast-result" in messages[0]["content"]
         assert messages[1]["tool_call_id"] == "c2"
         assert "timed out after" in messages[1]["content"]
+        assert messages[1]["effect_disposition"] == "unknown"
         assert [batch[-1]["tool_call_id"] for batch in flushed] == ["c1", "c2"]
         assert "fast-result" in flushed[0][-1]["content"]
         assert "timed out after" in flushed[1][-1]["content"]

tests/test_hermes_state.py

@@ -934,6 +934,19 @@ class TestMessageStorage:
         tool_msg = next(m for m in msgs if m["role"] == "tool")
         assert tool_msg["tool_name"] == "web_search"
 
+    def test_tool_effect_disposition_round_trips_through_session_db(self, db):
+        from agent.tool_dispatch_helpers import make_tool_result_message
+
+        db.create_session(session_id="s1", source="cli")
+        db.replace_messages(
+            "s1",
+            [make_tool_result_message(
+                "write_file", "worker detached", "c1", effect_disposition="unknown"
+            )],
+        )
+
+        assert db.get_messages_as_conversation("s1")[0]["effect_disposition"] == "unknown"
+
     def test_replace_messages_handles_multimodal_content(self, db):
         """`replace_messages` (used by /retry, /undo, /compress) must also
         handle list content without crashing."""