fix(codex): wait for compaction turn identity

agent/transports/codex_app_server_session.py

@@ -92,25 +92,15 @@ class TurnResult:
 _TURN_ABORTED_MARKERS = ("<turn_aborted>", "<turn_aborted/>")
 
 
-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,

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(