From 63e363f3066fdc7a572768d8ff8426eb185a7716 Mon Sep 17 00:00:00 2001 From: Tanmay Dixit Date: Sun, 12 Jul 2026 19:04:45 -0400 Subject: [PATCH] fix(codex): scope app-server notifications to active turn --- agent/transports/codex_app_server_session.py | 92 +++++++ .../test_codex_app_server_session.py | 233 ++++++++++++++++++ 2 files changed, 325 insertions(+) diff --git a/agent/transports/codex_app_server_session.py b/agent/transports/codex_app_server_session.py index 16c6905c07a..17ce87d25c4 100644 --- a/agent/transports/codex_app_server_session.py +++ b/agent/transports/codex_app_server_session.py @@ -92,6 +92,67 @@ class TurnResult: _TURN_ABORTED_MARKERS = ("", "") +def _notification_belongs_to_turn( + note: dict, + *, + thread_id: Optional[str], + turn_id: Optional[str], +) -> bool: + """Return whether a multiplexed notification belongs to this turn. + + Codex app-server can carry parent and hosted subagent threads over one + JSON-RPC connection. An explicitly foreign child or + stale-turn event must not mutate the active parent's transcript or mark + its turn complete. Unscoped notifications remain accepted for protocol + compatibility. + """ + if not isinstance(note, dict): + return False + params = note.get("params") or {} + if not isinstance(params, dict): + return True + + nested_turn = params.get("turn") or {} + nested_item = params.get("item") or {} + + observed_thread_id = params.get("threadId") or params.get("thread_id") + if observed_thread_id is None and isinstance(nested_turn, dict): + observed_thread_id = ( + nested_turn.get("threadId") + or nested_turn.get("thread_id") + ) + if observed_thread_id is None and isinstance(nested_item, dict): + observed_thread_id = ( + nested_item.get("threadId") + or nested_item.get("thread_id") + ) + + if ( + thread_id is not None + and observed_thread_id is not None + and str(observed_thread_id) != str(thread_id) + ): + return False + + observed_turn_id = params.get("turnId") or params.get("turn_id") + if observed_turn_id is None and isinstance(nested_turn, dict): + observed_turn_id = nested_turn.get("id") or nested_turn.get("turnId") + if observed_turn_id is None and isinstance(nested_item, dict): + observed_turn_id = ( + nested_item.get("turnId") + or nested_item.get("turn_id") + ) + + if ( + turn_id is not None + and observed_turn_id is not None + and str(observed_turn_id) != str(turn_id) + ): + return False + + return True + + def _coerce_turn_input_text(user_input: Any) -> str: """Collapse Hermes/OpenAI rich content into app-server text input. @@ -505,6 +566,17 @@ class CodexAppServerSession: pending = self._client.take_notification(timeout=0) if pending is None: break + if not _notification_belongs_to_turn( + pending, + thread_id=self._thread_id, + turn_id=result.turn_id, + ): + logger.debug( + "ignoring foreign codex notification while draining " + "server request: method=%s", + pending.get("method"), + ) + continue # Mirror the main notification-handling block below so # display events surface and stay in step with projector # state. Without this, item/started / item/completed @@ -550,6 +622,16 @@ class CodexAppServerSession: continue method = note.get("method", "") + if not _notification_belongs_to_turn( + note, + thread_id=self._thread_id, + turn_id=result.turn_id, + ): + logger.debug( + "ignoring foreign codex notification: method=%s", method + ) + continue + if self._on_event is not None: try: self._on_event(note) @@ -737,6 +819,16 @@ class CodexAppServerSession: continue method = note.get("method", "") + if not _notification_belongs_to_turn( + note, + thread_id=self._thread_id, + turn_id=result.turn_id, + ): + logger.debug( + "ignoring foreign codex notification: method=%s", method + ) + continue + if self._on_event is not None: try: self._on_event(note) diff --git a/tests/agent/transports/test_codex_app_server_session.py b/tests/agent/transports/test_codex_app_server_session.py index 956709e9ef4..8c281872069 100644 --- a/tests/agent/transports/test_codex_app_server_session.py +++ b/tests/agent/transports/test_codex_app_server_session.py @@ -95,6 +95,17 @@ class FakeClient: # Test helpers def queue_notification(self, method: str, **params): + # Keep legacy fixture shorthand aligned with the IDs returned by the + # fake thread/start and turn/start responses. + if params.get("threadId") in {"t", "th"}: + params["threadId"] = "thread-fake-001" + if params.get("turnId") == "tu1": + params["turnId"] = "turn-fake-001" + turn = params.get("turn") + if isinstance(turn, dict) and turn.get("id") == "tu1": + turn = dict(turn) + turn["id"] = "turn-fake-001" + params["turn"] = turn self._notifications.append({"method": method, "params": params}) def queue_server_request(self, method: str, request_id: Any = "srv-1", **params): @@ -196,6 +207,173 @@ class TestRunTurn: # turn_id propagated for downstream session-DB linkage assert r.turn_id == "turn-fake-001" + def test_subagent_completion_does_not_end_parent_turn(self): + """A child completion must not become the parent's response.""" + client = FakeClient() + client.queue_notification( + "item/completed", + threadId="thread-child-001", + turnId="turn-child-001", + item={ + "type": "agentMessage", + "id": "child-message-1", + "text": "child summary", + }, + ) + client.queue_notification( + "turn/completed", + threadId="thread-child-001", + turn={ + "id": "turn-child-001", + "status": "completed", + "error": None, + }, + ) + client.queue_notification( + "item/completed", + threadId="thread-fake-001", + turnId="turn-fake-001", + item={ + "type": "agentMessage", + "id": "parent-message-1", + "text": "parent synthesis", + }, + ) + client.queue_notification( + "turn/completed", + threadId="thread-fake-001", + turn={ + "id": "turn-fake-001", + "status": "completed", + "error": None, + }, + ) + + result = make_session(client).run_turn("delegate this", turn_timeout=2.0) + + assert result.final_text == "parent synthesis" + assert result.interrupted is False + assert result.error is None + assert result.projected_messages == [ + {"role": "assistant", "content": "parent synthesis"} + ] + + def test_stale_completion_on_parent_thread_is_ignored(self): + """A late completion from the previous parent turn is not terminal.""" + client = FakeClient() + client.queue_notification( + "item/completed", + threadId="thread-fake-001", + turnId="turn-previous", + item={ + "type": "agentMessage", + "id": "stale-message", + "text": "stale previous answer", + }, + ) + client.queue_notification( + "turn/completed", + threadId="thread-fake-001", + turn={ + "id": "turn-previous", + "status": "completed", + "error": None, + }, + ) + client.queue_notification( + "item/completed", + threadId="thread-fake-001", + turnId="turn-fake-001", + item={ + "type": "agentMessage", + "id": "current-message", + "text": "current parent answer", + }, + ) + client.queue_notification( + "turn/completed", + threadId="thread-fake-001", + turn={ + "id": "turn-fake-001", + "status": "completed", + "error": None, + }, + ) + + result = make_session(client).run_turn("new prompt", turn_timeout=2.0) + + assert result.final_text == "current parent answer" + assert result.projected_messages == [ + {"role": "assistant", "content": "current parent answer"} + ] + + def test_foreign_completion_in_server_request_drain_is_ignored(self): + """Approval draining must not project a child result into the parent.""" + client = FakeClient() + client.queue_server_request( + "item/commandExecution/requestApproval", + request_id="approval-1", + command="pwd", + cwd="/tmp", + ) + client.queue_notification( + "item/completed", + threadId="thread-child-001", + turnId="turn-child-001", + item={ + "type": "agentMessage", + "id": "child-message", + "text": "child drain summary", + }, + ) + client.queue_notification( + "turn/completed", + threadId="thread-child-001", + turn={ + "id": "turn-child-001", + "status": "completed", + "error": None, + }, + ) + + original_respond = client.respond + + def respond_and_release_parent(request_id, response): + original_respond(request_id, response) + client.queue_notification( + "item/completed", + threadId="thread-fake-001", + turnId="turn-fake-001", + item={ + "type": "agentMessage", + "id": "parent-message", + "text": "parent after approval", + }, + ) + client.queue_notification( + "turn/completed", + threadId="thread-fake-001", + turn={ + "id": "turn-fake-001", + "status": "completed", + "error": None, + }, + ) + + client.respond = respond_and_release_parent + session = make_session( + client, + request_routing=_ServerRequestRouting(auto_approve_exec=True), + ) + + result = session.run_turn("delegate then continue", turn_timeout=2.0) + + assert client.responses == [("approval-1", {"decision": "accept"})] + assert result.final_text == "parent after approval" + assert result.projected_messages == [ + {"role": "assistant", "content": "parent after approval"} + ] + def test_token_usage_notification_is_captured(self): client = FakeClient() client.queue_notification( @@ -528,6 +706,61 @@ class TestCompactThread: assert r.token_usage_last["totalTokens"] == 12 assert r.model_context_window == 200000 + def test_compact_thread_ignores_foreign_child_completion(self): + client = FakeClient() + client.queue_notification( + "item/completed", + threadId="thread-child-001", + turnId="child-compact-turn", + item={ + "type": "agentMessage", + "id": "child-compact-message", + "text": "child compact summary", + }, + ) + client.queue_notification( + "turn/completed", + threadId="thread-child-001", + turn={ + "id": "child-compact-turn", + "status": "completed", + "error": None, + }, + ) + client.queue_notification( + "turn/started", + threadId="thread-fake-001", + turn={"id": "compact-turn-1"}, + ) + client.queue_notification( + "item/completed", + threadId="thread-fake-001", + turnId="compact-turn-1", + item={ + "type": "agentMessage", + "id": "parent-compact-message", + "text": "parent compacted", + }, + ) + client.queue_notification( + "turn/completed", + threadId="thread-fake-001", + turn={ + "id": "compact-turn-1", + "status": "completed", + "error": None, + }, + ) + + result = make_session(client).compact_thread(turn_timeout=2.0) + + assert result.error is None + assert result.turn_id == "compact-turn-1" + assert result.final_text == "parent compacted" + assert result.projected_messages == [ + {"role": "assistant", "content": "parent compacted"} + ] + def test_compact_thread_interrupted_returns_non_success(self): client = FakeClient() client.queue_notification(