diff --git a/agent/transports/codex_app_server_session.py b/agent/transports/codex_app_server_session.py index 17ce87d25c45..2e1cc64bdcbc 100644 --- a/agent/transports/codex_app_server_session.py +++ b/agent/transports/codex_app_server_session.py @@ -92,25 +92,15 @@ class TurnResult: _TURN_ABORTED_MARKERS = ("", "") -def _notification_belongs_to_turn( +def _notification_scope_ids( 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. - """ +) -> tuple[Optional[str], Optional[str]]: + """Extract the thread/turn identity carried by a notification.""" if not isinstance(note, dict): - return False + return None, None params = note.get("params") or {} if not isinstance(params, dict): - return True + return None, None nested_turn = params.get("turn") or {} nested_item = params.get("item") or {} @@ -127,13 +117,6 @@ def _notification_belongs_to_turn( 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") @@ -143,6 +126,35 @@ def _notification_belongs_to_turn( or nested_item.get("turn_id") ) + return observed_thread_id, observed_turn_id + + +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 + + observed_thread_id, observed_turn_id = _notification_scope_ids(note) + + if ( + thread_id is not None + and observed_thread_id is not None + and str(observed_thread_id) != str(thread_id) + ): + return False + if ( turn_id is not None and observed_turn_id is not None @@ -819,6 +831,38 @@ class CodexAppServerSession: continue method = note.get("method", "") + observed_thread_id, observed_turn_id = _notification_scope_ids(note) + if result.turn_id is None: + if method == "turn/started": + if ( + observed_thread_id is not None + and str(observed_thread_id) != str(self._thread_id) + ): + logger.debug( + "ignoring foreign compact turn/started: thread=%s", + observed_thread_id, + ) + continue + if observed_turn_id is None: + logger.debug( + "ignoring compact turn/started without a turn id" + ) + continue + result.turn_id = str(observed_turn_id) + elif observed_turn_id is not None or method in { + "item/completed", + "turn/completed", + }: + # thread/compact/start does not return a turn id. Until the + # new turn/started arrives, any terminal/projectable event + # is stale or cannot be safely attributed to this compaction. + logger.debug( + "ignoring codex notification before compact turn start: " + "method=%s", + method, + ) + continue + if not _notification_belongs_to_turn( note, thread_id=self._thread_id, diff --git a/tests/agent/transports/test_codex_app_server_session.py b/tests/agent/transports/test_codex_app_server_session.py index 8c2818720694..5479c14ae21e 100644 --- a/tests/agent/transports/test_codex_app_server_session.py +++ b/tests/agent/transports/test_codex_app_server_session.py @@ -708,6 +708,11 @@ class TestCompactThread: def test_compact_thread_ignores_foreign_child_completion(self): client = FakeClient() + client.queue_notification( + "turn/started", + threadId="thread-child-001", + turn={"id": "child-compact-turn"}, + ) client.queue_notification( "item/completed", threadId="thread-child-001", @@ -761,6 +766,61 @@ class TestCompactThread: {"role": "assistant", "content": "parent compacted"} ] + def test_compact_thread_ignores_stale_same_thread_completion_before_start(self): + client = FakeClient() + client.queue_notification( + "item/completed", + threadId="thread-fake-001", + turnId="previous-compact-turn", + item={ + "type": "agentMessage", + "id": "stale-compact-message", + "text": "stale compact result", + }, + ) + client.queue_notification( + "turn/completed", + threadId="thread-fake-001", + turn={ + "id": "previous-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": "current-compact-message", + "text": "current compact result", + }, + ) + 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 == "current compact result" + assert result.projected_messages == [ + {"role": "assistant", "content": "current compact result"} + ] + def test_compact_thread_interrupted_returns_non_success(self): client = FakeClient() client.queue_notification(