Merge e5c1add7a5 into ff5c8cc8e4

strix/interface/tui/live_view.py

@@ -332,19 +332,28 @@ class TuiLiveView:
         call_id = call["call_id"]
         event_key = (agent_id, call_id)
         existing = self._tool_event_by_agent_and_call_id.get(event_key)
-        tool_data = {
-            "tool_name": call["tool_name"],
-            "args": call["args"],
-            "status": "running",
-            "agent_id": agent_id,
-            "call_id": call_id,
-            **self._mcp_tool_fields(call["tool_name"], call["args"]),
-        }
+        mcp_fields = self._mcp_tool_fields(call["tool_name"], call["args"])
         if existing is None:
+            tool_data = {
+                "tool_name": call["tool_name"],
+                "args": call["args"],
+                "status": "running",
+                "agent_id": agent_id,
+                "call_id": call_id,
+                **mcp_fields,
+            }
             event = self._append_event(agent_id, "tool", tool_data, timestamp=timestamp)
             self._tool_event_by_agent_and_call_id[event_key] = event
         else:
-            existing["data"].update(tool_data)
+            # A replayed or duplicated tool_call_item (duplicate stream event,
+            # hydration replay) must not resurrect a tool that has already
+            # reached a terminal state back to "running" - its result would
+            # then sit next to a status that says the call is still in
+            # flight.
+            update = {"tool_name": call["tool_name"], "args": call["args"], **mcp_fields}
+            if existing["data"].get("status") not in {"completed", "failed"}:
+                update["status"] = "running"
+            existing["data"].update(update)
             self._bump_event(existing, timestamp=timestamp)
 
     def _record_tool_output(self, agent_id: str, item: Any) -> None:

tests/test_tui_live_view_tool_events.py

@@ -0,0 +1,75 @@
+"""Tool call/output events must stay isolated per agent and never regress
+from a terminal status back to "running" on replay.
+
+Covers usestrix/strix#660: two bugs in TuiLiveView's tool-event bookkeeping.
+Bug 1 (events keyed only by call_id, colliding across agents) was already
+fixed by fade370 ("fix viewer tool call collisions across agents", #917) -
+this file adds the regression test that fix never got. Bug 2 (a replayed
+tool_call_item resetting a completed/failed event back to "running") was
+still present and is fixed here.
+"""
+
+from __future__ import annotations
+
+from strix.interface.tui.live_view import TuiLiveView
+
+
+def _call(call_id: str, tool_name: str = "shell") -> dict[str, object]:
+    return {"call_id": call_id, "tool_name": tool_name, "args": {}}
+
+
+def _output(call_id: str, output: object, tool_name: str = "shell") -> dict[str, object]:
+    return {"call_id": call_id, "tool_name": tool_name, "output": output}
+
+
+def test_tool_events_do_not_collide_across_agents_sharing_a_call_id() -> None:
+    view = TuiLiveView()
+
+    view._record_tool_call_data("agent-A", _call("shared-id", tool_name="nmap"))
+    view._record_tool_call_data("agent-B", _call("shared-id", tool_name="curl"))
+
+    a_events = view.events_for_agent("agent-A")
+    b_events = view.events_for_agent("agent-B")
+
+    assert len(a_events) == 1
+    assert len(b_events) == 1
+    assert a_events[0]["data"]["tool_name"] == "nmap"
+    assert b_events[0]["data"]["tool_name"] == "curl"
+
+
+def test_replayed_tool_call_does_not_revert_completed_status_to_running() -> None:
+    view = TuiLiveView()
+
+    view._record_tool_call_data("agent-A", _call("id-1"))
+    view._record_tool_output_data("agent-A", _output("id-1", {"success": True}))
+
+    event = view.events_for_agent("agent-A")[0]
+    assert event["data"]["status"] == "completed"
+
+    # Duplicate stream event / hydration replay of the same call.
+    view._record_tool_call_data("agent-A", _call("id-1"))
+
+    replayed = view.events_for_agent("agent-A")[0]
+    assert replayed["data"]["status"] == "completed"
+    assert replayed["data"]["result"] == {"success": True}
+
+
+def test_replayed_tool_call_does_not_revert_failed_status_to_running() -> None:
+    view = TuiLiveView()
+
+    view._record_tool_call_data("agent-A", _call("id-1"))
+    view._record_tool_output_data("agent-A", _output("id-1", {"success": False, "error": "x"}))
+
+    assert view.events_for_agent("agent-A")[0]["data"]["status"] == "failed"
+
+    view._record_tool_call_data("agent-A", _call("id-1"))
+
+    assert view.events_for_agent("agent-A")[0]["data"]["status"] == "failed"
+
+
+def test_first_tool_call_is_still_recorded_as_running() -> None:
+    view = TuiLiveView()
+
+    view._record_tool_call_data("agent-A", _call("id-1"))
+
+    assert view.events_for_agent("agent-A")[0]["data"]["status"] == "running"