From 23d1472435d5ea8fd561df08d7cbc26cb62885a1 Mon Sep 17 00:00:00 2001 From: call-me-ram Date: Mon, 31 Aug 2026 17:30:51 +0000 Subject: [PATCH 1/3] fix(codex): opt into the deferred-init direct status probe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The deferred-assign confirm loop polls the cached event-driven status, which detects only at rising-edge/quiescence; a repainting Working spinner defers quiescence, so the cache can sit IDLE past the whole confirm window while the pane visibly works. The loop's capture-pane fallback exists for exactly this lag but is gated on supports_direct_status_probe, which CodexProvider never set — so the loop re-delivered the task into the working pane up to three times and then tore down the active worker. Codex's get_status() is line-oriented analysis of exactly the rendered shape a capture provides (the same frames supports_screen_detection feeds it in production) with no dispatch bookkeeping to bypass, so the opt-in is valid; every dropped-submit shape still reads IDLE, keeping the probe fail-toward-recovery. Regression pins the opt-in end to end with the real provider and a real Working frame. Fixes #659 --- src/cli_agent_orchestrator/providers/codex.py | 12 ++++ .../test_deferred_submit_verification.py | 66 +++++++++++++++++++ 2 files changed, 78 insertions(+) diff --git a/src/cli_agent_orchestrator/providers/codex.py b/src/cli_agent_orchestrator/providers/codex.py index ac568e624..9a8be09ad 100644 --- a/src/cli_agent_orchestrator/providers/codex.py +++ b/src/cli_agent_orchestrator/providers/codex.py @@ -792,6 +792,18 @@ class CodexProvider(BaseProvider): # the live frame rather than stale redraw history. supports_screen_detection = True + # Opt-in for the deferred-init direct status probe (capture-pane bypass, + # #659). The event-driven cache detects only at rising-edge/quiescence, so + # a repainting Working spinner can leave the cached status IDLE past the + # whole confirm window — the retry loop then re-delivers the task into the + # already-working pane and eventually tears the worker down. get_status() + # is line-oriented text analysis of exactly this rendered shape (the same + # frames the screen-detection route above feeds it in production), so a + # live capture-pane snapshot is a valid input; there is no dispatch + # bookkeeping that a fresh capture would bypass (the kiro_cli-style + # disqualifier documented on _worker_is_started_direct). + supports_direct_status_probe = True + def __init__( self, terminal_id: str, diff --git a/test/services/test_deferred_submit_verification.py b/test/services/test_deferred_submit_verification.py index 659f1106b..bc3ad1b3f 100644 --- a/test/services/test_deferred_submit_verification.py +++ b/test/services/test_deferred_submit_verification.py @@ -373,3 +373,69 @@ def test_returns_false_when_status_is_idle(self): patch.object(ts, "get_backend") as mock_be, ): assert ts._worker_is_started_direct("t1", provider) is False + + +class TestCodexDirectProbeOptIn: + """#659: Codex deferred assign — the cached status can sit IDLE for the whole + confirm window while the real pane already shows the TUI Working spinner + (detection fires only at rising-edge/quiescence, and a repainting spinner + defers quiescence). Without the direct-probe opt-in the confirm loop + re-delivers the task into the working pane up to three times and then tears + the worker down. These pin the opt-in end to end with the REAL provider and + a real rendered frame, so removing the flag (or breaking the detector on + this shape) goes red here — not just in a unit assert on the attribute. + """ + + # The shape from the issue report: handoff prompt in the transcript, live + # Working spinner, TUI footer. Same frame family the codex provider unit + # tests pin as PROCESSING. + _WORKING_FRAME = ( + "› [CAO Handoff] Supervisor terminal ID: sup-123. Do the task.\n" + "\n" + "• Working (3s • esc to interrupt)\n" + "\n" + "› Use /skills to list available skills\n" + "\n" + " ? for shortcuts 100% context left\n" + ) + + def test_codex_opts_into_direct_status_probe(self): + from cli_agent_orchestrator.providers.codex import CodexProvider + + assert CodexProvider.supports_direct_status_probe is True + + @pytest.mark.asyncio + async def test_codex_confirm_succeeds_from_live_frame_without_redelivery(self): + from cli_agent_orchestrator.providers.codex import CodexProvider + + provider = CodexProvider("t1", "s1", "w0") + backend = MagicMock() + backend.get_history.return_value = self._WORKING_FRAME + with ( + # Cached status stays IDLE past every poll — the #496-class lag. + patch.object(ts, "wait_until_status", new=AsyncMock(return_value=False)), + patch.object( + ts, + "get_terminal_metadata", + return_value={"tmux_session": "s1", "tmux_window": "w0"}, + ), + patch.object(ts, "get_backend", return_value=backend), + patch.object(ts, "send_special_key") as key, + patch.object(ts, "send_input") as send, + ): + ok = await ts._confirm_worker_started_or_resubmit( + "t1", + "Do the task", + None, + "sup", + None, + provider=provider, + ) + + # Started: the caller must not classify this as a dropped submit, so the + # delete_worker teardown arm never fires... + assert ok is True + # ...and nothing was typed into the already-working pane: no full + # re-delivery (which would run the task twice) and no blind Enter. + send.assert_not_called() + key.assert_not_called() From dca3dd34b5fbf385461c0a15d7ef195b0bf965a9 Mon Sep 17 00:00:00 2001 From: call-me-ram Date: Wed, 2 Sep 2026 07:23:36 +0000 Subject: [PATCH 2/3] fix(codex): bind the direct status probe to the submitted message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A started status read off the live pane is only half the verdict: the pane must also attribute it to the message being confirmed. get_status classifies the frame as a whole, and Codex's startup chrome shares the shape of its task activity — the "Starting MCP servers (4s • esc to interrupt)" spinner is the TUI progress pattern and any startup bullet is an assistant marker. initialize() returns once the bottom 15 lines are activity-free, but get_status scans 25 lines for the spinner and the whole capture for a marker, so a task-less pane with that residue 16+ lines above the idle composer reads PROCESSING (or COMPLETED once the spinner leaves the tail). The unqualified opt-in accepted that as "started", skipped the redelivery a dropped paste needs, and left the supervisor waiting on a task that was never submitted. Add a provider hook, direct_probe_confirms_submission(output, message), that the probe consults after a started read. The default keeps the status-only verdict for providers whose indicators are turn-scoped (the existing opt-ins are unchanged). Codex overrides it with the causal trace a submitted turn leaves: the message is echoed as a transcript line and the turn's activity (spinner or bullet) renders below it, whereas startup residue sits above the composer and an unsubmitted paste has nothing but the footer beneath it. Bullets that are part of the pasted message text do not count as activity, and a message too short to match reliably cannot bind, so every unattributed read falls through to the box check and redelivery. Regressions pin the finding's frame (residue at both the PROCESSING and COMPLETED distances, no task → full redelivery on every attempt), residue over an unsubmitted paste (bare Enter), residue over an accepted turn (started, nothing sent), an accepted turn parked on the 0.147 approval menu (started, no blind Enter into the menu), a paste whose own lines are bullets (no self-attribution), and the short- message floor. Neutralizing the hook, the probe's use of it, or the message-internal bullet exclusion each turns the corresponding tests red. --- src/cli_agent_orchestrator/providers/base.py | 19 ++ src/cli_agent_orchestrator/providers/codex.py | 60 +++++- .../services/terminal_service.py | 18 +- .../test_deferred_submit_verification.py | 175 +++++++++++++++++- 4 files changed, 266 insertions(+), 6 deletions(-) diff --git a/src/cli_agent_orchestrator/providers/base.py b/src/cli_agent_orchestrator/providers/base.py index a8a2d0612..365ec4b45 100644 --- a/src/cli_agent_orchestrator/providers/base.py +++ b/src/cli_agent_orchestrator/providers/base.py @@ -179,6 +179,25 @@ def get_status(self, buffer: str) -> TerminalStatus: # this False — their COMPLETED/IDLE split is not screen-detectable. supports_direct_status_probe: bool = False + def direct_probe_confirms_submission(self, output: str, message: str) -> bool: + """Whether a started status read from ``output`` is attributable to + ``message`` — the text the direct status probe is confirming. + + The deferred-submit probe (``supports_direct_status_probe``) reads a + started status off a live capture-pane snapshot and, when it holds, + skips redelivery. For a TUI whose activity indicators are turn-scoped + that status alone is evidence the submission was accepted, and the + default keeps that behavior. A provider whose startup chrome shares the + shape of its task-activity indicators (a startup spinner or bullet + that ``get_status`` reads as PROCESSING/COMPLETED before any task was + ever submitted) must override this and bind the verdict to evidence of + the current submission — otherwise the probe suppresses the recovery + it exists to perform and a dropped task is silently lost. Return False + whenever attribution cannot be established: the probe then falls + through to the redelivery decision. + """ + return True + def get_status_from_screen(self, screen_lines: List[str]) -> TerminalStatus: """Detect status from a pyte-rendered screen (composited viewport). diff --git a/src/cli_agent_orchestrator/providers/codex.py b/src/cli_agent_orchestrator/providers/codex.py index 9a8be09ad..1254dd614 100644 --- a/src/cli_agent_orchestrator/providers/codex.py +++ b/src/cli_agent_orchestrator/providers/codex.py @@ -801,9 +801,67 @@ class CodexProvider(BaseProvider): # frames the screen-detection route above feeds it in production), so a # live capture-pane snapshot is a valid input; there is no dispatch # bookkeeping that a fresh capture would bypass (the kiro_cli-style - # disqualifier documented on _worker_is_started_direct). + # disqualifier documented on _worker_is_started_direct). The status alone + # is NOT the verdict, though — see direct_probe_confirms_submission below. supports_direct_status_probe = True + def direct_probe_confirms_submission(self, output: str, message: str) -> bool: + """Bind a started status to the submission being confirmed. + + Codex's startup chrome shares the shape of its task activity: a + ``• Starting MCP servers (4s • esc to interrupt)`` spinner is the + TUI_PROGRESS_PATTERN, and any startup ``•`` line is an assistant + marker. ``initialize()`` returns once the bottom + STARTUP_PROMPT_BOTTOM_LINES are activity-free, but ``get_status`` + scans a wider tail for the spinner and the whole capture for a + marker, so a task-less pane with that residue 16+ lines above the + idle composer reads PROCESSING (or COMPLETED once the spinner leaves + the tail). Accepting that as "started" would skip the redelivery a + dropped paste needs and lose the task silently. + + A submitted turn leaves a causal trace instead: Codex echoes the + message as a ``›`` transcript line and renders the turn's activity + (spinner, ``•`` reply/tool bullets) BELOW it, whereas startup residue + sits ABOVE the composer and an unsubmitted paste sits in the composer + with nothing but the footer beneath. So the verdict is: the message + is visible, and a turn-activity line follows it. Bullets that are + part of the message text itself do not count — a pasted, unsubmitted + message containing its own ``•`` lines must not attribute activity + to itself. Anything short of that returns False and the probe falls + through to the redelivery decision (fail toward recovery). + + Matching collapses both sides to ``[a-z0-9]`` and uses the message's + leading 24 characters, the same collapse ``_message_visible_in_box`` + applies, so pane-width wrapping, unicode punctuation, and whitespace + cannot defeat the match. + """ + probe = re.sub(r"[^a-z0-9]", "", message.lower())[:24] + if len(probe) < 8: + return False + lines = strip_terminal_escapes(output).splitlines() + # Collapse the pane line by line, recording which line each kept + # character came from so the echo can be located by line. + collapsed: list[str] = [] + owner: list[int] = [] + for index, line in enumerate(lines): + kept = re.sub(r"[^a-z0-9]", "", line.lower()) + collapsed.append(kept) + owner.extend([index] * len(kept)) + hit = "".join(collapsed).find(probe) + if hit == -1: + return False + echo_line = owner[hit] + message_text = re.sub(r"[^a-z0-9]", "", message.lower()) + for line in lines[echo_line + 1 :]: + if re.search(TUI_PROGRESS_PATTERN, line): + return True + if re.match(STARTUP_ACTIVITY_PATTERN, line): + kept = re.sub(r"[^a-z0-9]", "", line.lower()) + if kept and kept in message_text: + continue # a bullet inside the pasted message, not a reply + return True + return False + def __init__( self, terminal_id: str, diff --git a/src/cli_agent_orchestrator/services/terminal_service.py b/src/cli_agent_orchestrator/services/terminal_service.py index 27caf8ec2..9cdf7a0c8 100644 --- a/src/cli_agent_orchestrator/services/terminal_service.py +++ b/src/cli_agent_orchestrator/services/terminal_service.py @@ -1178,7 +1178,7 @@ def _notify_caller_of_deferred_failure( } -def _worker_is_started_direct(terminal_id: str, provider) -> bool: +def _worker_is_started_direct(terminal_id: str, provider, message: str = "") -> bool: """Direct visible-screen status check bypassing the event-driven status cache. The deferred-init retry loop polls ``status_monitor.get_status()`` which @@ -1197,6 +1197,16 @@ def _worker_is_started_direct(terminal_id: str, provider) -> bool: providers (e.g. kiro_cli, antigravity_cli, cursor_cli) relies on dispatch bookkeeping and cannot distinguish IDLE from COMPLETED on a rendered capture-pane snapshot. + + A started status is only half the verdict: the pane must also attribute it + to ``message``, the submission being confirmed. ``get_status`` classifies + the frame as a whole, so activity that predates the submission (a + provider's startup spinner or bullet still on screen) reads as started for + a pane whose task paste was dropped — and accepting it here would suppress + the redelivery this probe exists to gate. The provider decides attribution + through ``direct_probe_confirms_submission`` (default: the status alone, + for TUIs whose indicators are turn-scoped); a False there falls through + to the box check and redelivery like any other not-started read. """ try: metadata = get_terminal_metadata(terminal_id) @@ -1208,6 +1218,9 @@ def _worker_is_started_direct(terminal_id: str, provider) -> bool: return False output = get_backend().get_history(session_name, window_name, tail_lines=200) status = provider.get_status(output) + if status not in _DEFERRED_STARTED_STATUSES: + return False + return bool(provider.direct_probe_confirms_submission(output, message)) except Exception: logger.debug( "Direct status probe for %s failed (falling through to cached path)", @@ -1215,7 +1228,6 @@ def _worker_is_started_direct(terminal_id: str, provider) -> bool: exc_info=True, ) return False - return status in _DEFERRED_STARTED_STATUSES def _message_visible_in_box(terminal_id: str, message: str) -> bool: @@ -1301,7 +1313,7 @@ def redeliver_dropped_message( provider, "supports_direct_status_probe", False ) if probe_capable: - if _worker_is_started_direct(terminal_id, provider): + if _worker_is_started_direct(terminal_id, provider, message): return True if _message_visible_in_box(terminal_id, message): logger.warning( diff --git a/test/services/test_deferred_submit_verification.py b/test/services/test_deferred_submit_verification.py index bc3ad1b3f..a073372ea 100644 --- a/test/services/test_deferred_submit_verification.py +++ b/test/services/test_deferred_submit_verification.py @@ -57,7 +57,8 @@ def test_resolves_provider_from_registry_for_direct_probe(self): started = ts.redeliver_dropped_message("t1", "Analyze the logs", 1) assert started is True mgr.get_provider.assert_called_once_with("t1") - probe.assert_called_once_with("t1", provider) + # The message rides along: the probe binds its verdict to it. + probe.assert_called_once_with("t1", provider, "Analyze the logs") key.assert_not_called() send.assert_not_called() @@ -386,6 +387,8 @@ class TestCodexDirectProbeOptIn: this shape) goes red here — not just in a unit assert on the attribute. """ + _MESSAGE = "[CAO Handoff] Supervisor terminal ID: sup-123. Do the task." + # The shape from the issue report: handoff prompt in the transcript, live # Working spinner, TUI footer. Same frame family the codex provider unit # tests pin as PROCESSING. @@ -399,6 +402,56 @@ class TestCodexDirectProbeOptIn: " ? for shortcuts 100% context left\n" ) + # Startup chrome as Codex renders it before any task exists. The MCP + # startup spinner IS the TUI progress pattern, and any startup bullet is + # an assistant marker. + _STARTUP_BANNER = ( + "╭──────────────────────────────────────────────╮\n" + "│ >_ OpenAI Codex (v0.145.0) │\n" + "│ │\n" + "│ model: gpt-5.6-sol medium │\n" + "│ directory: ~/project │\n" + "│ permissions: YOLO mode │\n" + "╰──────────────────────────────────────────────╯\n" + "\n" + "• Starting MCP servers (4s • esc to interrupt)\n" + ) + _IDLE_COMPOSER = ( + "\n" "› Write tests for @filename\n" "\n" " gpt-5.6-sol medium · Context 100% left\n" + ) + + @staticmethod + def _residue_frame(gap: int, composer: str) -> str: + """Startup spinner ``gap`` blank lines above the composer: outside the + bottom-15 window initialize() vetoes activity in, so the provider + reports ready, yet inside (or above) the wider tail get_status scans.""" + return TestCodexDirectProbeOptIn._STARTUP_BANNER + "\n" * gap + composer + + @staticmethod + async def _run_confirm(frame: str, message: str): + from cli_agent_orchestrator.providers.codex import CodexProvider + + provider = CodexProvider("t1", "s1", "w0") + backend = MagicMock() + backend.get_history.return_value = frame + with ( + # Cached status stays IDLE past every poll — the #496-class lag. + patch.object(ts, "wait_until_status", new=AsyncMock(return_value=False)), + patch.object( + ts, + "get_terminal_metadata", + return_value={"tmux_session": "s1", "tmux_window": "w0"}, + ), + patch.object(ts, "get_backend", return_value=backend), + patch.object(ts, "get_output", return_value=frame), + patch.object(ts, "send_special_key") as key, + patch.object(ts, "send_input") as send, + ): + ok = await ts._confirm_worker_started_or_resubmit( + "t1", message, None, "sup", None, provider=provider + ) + return ok, key, send + def test_codex_opts_into_direct_status_probe(self): from cli_agent_orchestrator.providers.codex import CodexProvider @@ -425,7 +478,7 @@ async def test_codex_confirm_succeeds_from_live_frame_without_redelivery(self): ): ok = await ts._confirm_worker_started_or_resubmit( "t1", - "Do the task", + self._MESSAGE, None, "sup", None, @@ -439,3 +492,121 @@ async def test_codex_confirm_succeeds_from_live_frame_without_redelivery(self): # re-delivery (which would run the task twice) and no blind Enter. send.assert_not_called() key.assert_not_called() + + # --- the verdict is bound to the submission, not to the pane's status ----- + # get_status classifies the frame as a whole, so startup residue reads as + # started for a pane whose task paste was dropped. The probe must not take + # that as acceptance: it would skip the redelivery this path exists for and + # the task would be silently lost with the supervisor waiting forever. + + @pytest.mark.parametrize( + "gap, expected_status", + [ + (12, "processing"), # spinner inside get_status's 25-line spinner tail + (20, "processing"), # ...at its far edge + (28, "completed"), # spinner out of the tail: the bullet is an assistant marker + ], + ) + @pytest.mark.asyncio + async def test_startup_residue_on_a_dropped_task_still_redelivers(self, gap, expected_status): + from cli_agent_orchestrator.providers.codex import CodexProvider, _has_startup_idle_composer + + frame = self._residue_frame(gap, self._IDLE_COMPOSER) + # Precondition — this is exactly the frame the finding describes: the + # provider reports ready (initialize() would have returned) while the + # whole-frame status says started, and the message is nowhere. + assert _has_startup_idle_composer(frame) is True + assert CodexProvider("t1", "s1", "w0").get_status(frame).value == expected_status + assert self._MESSAGE[:12] not in frame + + ok, key, send = await self._run_confirm(frame, self._MESSAGE) + + # Not started: the paste was dropped, so every attempt re-delivers the + # full message and the caller gets to classify the outcome. + assert ok is False + assert send.call_count == ts._DEFERRED_SUBMIT_MAX_RESUBMITS + key.assert_not_called() + + @pytest.mark.asyncio + async def test_startup_residue_with_unsubmitted_text_sends_enter(self): + # Paste landed, Enter was swallowed: the message sits in the composer + # under the residue. The bare-Enter recovery must still fire. + composer = "\n› " + self._MESSAGE + "\n\n gpt-5.6-sol medium · Context 100% left\n" + frame = self._residue_frame(18, composer) + + ok, key, send = await self._run_confirm(frame, self._MESSAGE) + + assert ok is False + assert key.call_count == ts._DEFERRED_SUBMIT_MAX_RESUBMITS + send.assert_not_called() + + @pytest.mark.asyncio + async def test_startup_residue_above_an_accepted_task_is_started(self): + # Residue AND a real accepted turn: the echo of our message with the + # turn's activity below it is the causal evidence; residue above it + # neither adds nor subtracts. + frame = self._residue_frame(18, self._WORKING_FRAME) + + ok, key, send = await self._run_confirm(frame, self._MESSAGE) + + assert ok is True + send.assert_not_called() + key.assert_not_called() + + @pytest.mark.asyncio + async def test_accepted_turn_on_an_approval_prompt_is_started(self): + # WAITING_USER_ANSWER after our turn (codex 0.147 approval menu, as in + # test/providers/fixtures/codex_approval_modal_raw.txt): the activity + # bullet below the echo binds it; the probe must not blind-Enter into + # the menu (that would select "Yes, proceed"). + frame = ( + "› " + self._MESSAGE + "\n" + "• Running mkdir -p /tmp/work/subdir\n" + " Would you like to run the following command?\n" + " $ mkdir -p /tmp/work/subdir\n" + "› 1. Yes, proceed (y)\n" + " 2. Yes, and don't ask again for commands that start with `mkdir` (p)\n" + " 3. No, and tell Codex what to do differently (esc)\n" + " Press enter to confirm or esc to cancel\n" + ) + ok, key, send = await self._run_confirm(frame, self._MESSAGE) + + assert ok is True + send.assert_not_called() + key.assert_not_called() + + def test_bullets_inside_the_pasted_message_do_not_self_attribute(self): + from cli_agent_orchestrator.providers.codex import CodexProvider + + # An unsubmitted multi-line paste whose own lines are bullets: the + # bullets below the echo line belong to the message, not to a reply. + message = "Review these findings for me:\n• first finding\n• second finding" + frame = self._residue_frame( + 18, + "\n› Review these findings for me:\n • first finding\n • second finding\n" + "\n gpt-5.6-sol medium · Context 100% left\n", + ) + provider = CodexProvider("t1", "s1", "w0") + assert provider.direct_probe_confirms_submission(frame, message) is False + # ...while a reply bullet under the same paste does bind it. + assert ( + provider.direct_probe_confirms_submission( + frame.replace( + " • second finding\n", " • second finding\n• Reviewing the findings\n" + ), + message, + ) + is True + ) + + def test_short_message_cannot_bind(self): + from cli_agent_orchestrator.providers.codex import CodexProvider + + # Below the 8-character floor the collapse cannot match reliably; the + # hook must refuse rather than guess (same floor as the box check). + assert ( + CodexProvider("t1", "s1", "w0").direct_probe_confirms_submission( + "› hi\n• Working (3s • esc to interrupt)\n", "hi" + ) + is False + ) From 510eea3ba0d0619a5ec03a9be0643d0e86cf0b0e Mon Sep 17 00:00:00 2001 From: call-me-ram Date: Thu, 3 Sep 2026 03:14:17 +0000 Subject: [PATCH 3/3] fix(codex): bind the direct probe to post-dispatch output, not message text MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 bound the probe's verdict to the submitted message by matching that message against the rendered pane. haofeif showed the matching itself is unsound, and reproducing each case confirmed all of it: - The leading 24 collapsed characters are not message-unique. Every orchestrated handoff opens with the same banner, so an OLD handoff still in the transcript satisfied the match and its completion bullet vouched for a NEW message that was never delivered. - A message absent from the bounded 200-line capture was read as a dropped paste, but an ACCEPTED turn can out-scroll its own echo while a spinner still runs. The full re-send then submitted the task a second time into the pane already working on it. - Excluding bullets whose text appears in the message discarded real replies: "Reply with Done" answered by "• Done" was rejected, so a completed worker got three bare Enters and a teardown. - A non-ASCII bullet in a pasted message normalizes to nothing, so an unsubmitted paste confirmed itself. Text cannot establish causality. Use the dispatch boundary the code already maintains instead: send_input empties the StatusMonitor rolling buffer immediately BEFORE sending keystrokes, precisely so the turn's first bytes are captured, so every byte in that buffer arrived after the dispatch by construction. It cannot be forged by a shared prefix, evicted by a bounded capture tail, or confused with leftover chrome. The provider hook becomes direct_probe_confirms_dispatch(post_dispatch _output) and inspects no message text at all. The base default keeps the status-only verdict, so the opencode and minimax opt-ins are unchanged. Codex accepts one piece of evidence: the progress spinner, whose "(s • esc to interrupt)" shape a live turn necessarily repaints and a stopped one cannot emit. A bullet is deliberately not accepted, because the composer echoes a pasted message as it renders. Because a False verdict now means unproven rather than idle, the full re-send is also withheld from a probe-capable provider whenever any output arrived after the dispatch without proving the turn started. That is exactly the accepted-turn-past-its-echo shape, and re-pasting there duplicates work. A genuinely dropped paste emits nothing after the dispatch, so it still re-sends. Regressions cover every reproduction above plus the round-1 startup residue at both the PROCESSING and COMPLETED distances, escape-laden spinner bytes, and the herdr backend that feeds no buffer. Five mutations each redden their tests: neutralizing the hook, accepting bullets as evidence, ignoring the hook, dropping the withheld-resend guard, and sourcing the evidence from the pane instead of the buffer. The pane-text prefix collision also affects _message_visible_in_box on main, independently of this probe; filed as #727 rather than widened into this PR, since the step path (#562) shares that helper. --- src/cli_agent_orchestrator/providers/base.py | 39 ++- src/cli_agent_orchestrator/providers/codex.py | 89 +++--- .../services/terminal_service.py | 85 ++++-- .../test_deferred_submit_verification.py | 278 +++++++++--------- 4 files changed, 260 insertions(+), 231 deletions(-) diff --git a/src/cli_agent_orchestrator/providers/base.py b/src/cli_agent_orchestrator/providers/base.py index 365ec4b45..667c6a767 100644 --- a/src/cli_agent_orchestrator/providers/base.py +++ b/src/cli_agent_orchestrator/providers/base.py @@ -179,22 +179,29 @@ def get_status(self, buffer: str) -> TerminalStatus: # this False — their COMPLETED/IDLE split is not screen-detectable. supports_direct_status_probe: bool = False - def direct_probe_confirms_submission(self, output: str, message: str) -> bool: - """Whether a started status read from ``output`` is attributable to - ``message`` — the text the direct status probe is confirming. - - The deferred-submit probe (``supports_direct_status_probe``) reads a - started status off a live capture-pane snapshot and, when it holds, - skips redelivery. For a TUI whose activity indicators are turn-scoped - that status alone is evidence the submission was accepted, and the - default keeps that behavior. A provider whose startup chrome shares the - shape of its task-activity indicators (a startup spinner or bullet - that ``get_status`` reads as PROCESSING/COMPLETED before any task was - ever submitted) must override this and bind the verdict to evidence of - the current submission — otherwise the probe suppresses the recovery - it exists to perform and a dropped task is silently lost. Return False - whenever attribution cannot be established: the probe then falls - through to the redelivery decision. + def direct_probe_confirms_dispatch(self, post_dispatch_output: str) -> bool: + """Whether ``post_dispatch_output`` proves this provider's TUI actually + began working on the submission the direct probe is confirming. + + ``post_dispatch_output`` is the StatusMonitor rolling byte buffer, which + ``send_input`` empties immediately BEFORE it sends the keystrokes (see + ``clear_rolling_buffer``). Every byte in it therefore arrived after the + dispatch, by construction — it cannot contain pre-dispatch scrollback, + cannot be defeated by a shared message prefix, and cannot be evicted by + a bounded capture-pane tail. + + The direct probe reads a started status off the live rendered pane + (``get_status``), which classifies the frame as a whole and so cannot + tell a running turn from leftover chrome. This hook supplies the + causality that status alone lacks. The default keeps the status-only + verdict, for TUIs whose activity indicators are turn-scoped. A provider + whose startup chrome renders like task activity must override it and + name the byte-level evidence that only a real turn produces. + + Returning False is not "not started": it means unproven, and the caller + treats it as such — the bare-Enter recovery still runs when the message + is on screen, and a full re-send is withheld whenever any post-dispatch + output exists, because absence of proof is not proof of a dropped paste. """ return True diff --git a/src/cli_agent_orchestrator/providers/codex.py b/src/cli_agent_orchestrator/providers/codex.py index 1254dd614..360ce4f3b 100644 --- a/src/cli_agent_orchestrator/providers/codex.py +++ b/src/cli_agent_orchestrator/providers/codex.py @@ -802,65 +802,44 @@ class CodexProvider(BaseProvider): # live capture-pane snapshot is a valid input; there is no dispatch # bookkeeping that a fresh capture would bypass (the kiro_cli-style # disqualifier documented on _worker_is_started_direct). The status alone - # is NOT the verdict, though — see direct_probe_confirms_submission below. + # is NOT the verdict, though — see direct_probe_confirms_dispatch below. supports_direct_status_probe = True - def direct_probe_confirms_submission(self, output: str, message: str) -> bool: - """Bind a started status to the submission being confirmed. - - Codex's startup chrome shares the shape of its task activity: a - ``• Starting MCP servers (4s • esc to interrupt)`` spinner is the - TUI_PROGRESS_PATTERN, and any startup ``•`` line is an assistant - marker. ``initialize()`` returns once the bottom - STARTUP_PROMPT_BOTTOM_LINES are activity-free, but ``get_status`` - scans a wider tail for the spinner and the whole capture for a - marker, so a task-less pane with that residue 16+ lines above the - idle composer reads PROCESSING (or COMPLETED once the spinner leaves - the tail). Accepting that as "started" would skip the redelivery a - dropped paste needs and lose the task silently. - - A submitted turn leaves a causal trace instead: Codex echoes the - message as a ``›`` transcript line and renders the turn's activity - (spinner, ``•`` reply/tool bullets) BELOW it, whereas startup residue - sits ABOVE the composer and an unsubmitted paste sits in the composer - with nothing but the footer beneath. So the verdict is: the message - is visible, and a turn-activity line follows it. Bullets that are - part of the message text itself do not count — a pasted, unsubmitted - message containing its own ``•`` lines must not attribute activity - to itself. Anything short of that returns False and the probe falls - through to the redelivery decision (fail toward recovery). - - Matching collapses both sides to ``[a-z0-9]`` and uses the message's - leading 24 characters, the same collapse ``_message_visible_in_box`` - applies, so pane-width wrapping, unicode punctuation, and whitespace - cannot defeat the match. + def direct_probe_confirms_dispatch(self, post_dispatch_output: str) -> bool: + """Prove a codex turn actually started, from post-dispatch bytes alone. + + Codex's startup chrome renders like its task activity: the + ``Starting MCP servers (4s - esc to interrupt)`` spinner IS + ``TUI_PROGRESS_PATTERN`` and any startup bullet is an assistant marker. + On a whole-frame ``get_status`` read that residue is indistinguishable + from a running turn, so the probe needs evidence tied to time rather + than to text. + + ``post_dispatch_output`` supplies it: ``send_input`` empties the rolling + buffer immediately before sending the keystrokes, so a spinner that + stopped BEFORE the dispatch contributes no bytes here no matter where it + still sits on screen, while a live one repaints its elapsed counter and + necessarily does. The evidence is therefore the progress spinner alone. + + A bullet is deliberately NOT accepted: the composer echoes the pasted + message as it renders, so a multi-line paste containing its own ``.`` + bullet would emit one without any turn having started. The spinner's + ``(s - esc to interrupt)`` shape cannot be produced that way. + + False here is "unproven", never "idle" — see the base docstring. It is + also the honest answer on an event-inbox backend (herdr) that never + feeds a byte buffer: the probe then declines to vouch for the turn and + the caller falls back to its pre-existing behavior. """ - probe = re.sub(r"[^a-z0-9]", "", message.lower())[:24] - if len(probe) < 8: + if not post_dispatch_output: return False - lines = strip_terminal_escapes(output).splitlines() - # Collapse the pane line by line, recording which line each kept - # character came from so the echo can be located by line. - collapsed: list[str] = [] - owner: list[int] = [] - for index, line in enumerate(lines): - kept = re.sub(r"[^a-z0-9]", "", line.lower()) - collapsed.append(kept) - owner.extend([index] * len(kept)) - hit = "".join(collapsed).find(probe) - if hit == -1: - return False - echo_line = owner[hit] - message_text = re.sub(r"[^a-z0-9]", "", message.lower()) - for line in lines[echo_line + 1 :]: - if re.search(TUI_PROGRESS_PATTERN, line): - return True - if re.match(STARTUP_ACTIVITY_PATTERN, line): - kept = re.sub(r"[^a-z0-9]", "", line.lower()) - if kept and kept in message_text: - continue # a bullet inside the pasted message, not a reply - return True - return False + return bool( + re.search( + TUI_PROGRESS_PATTERN, + strip_terminal_escapes(post_dispatch_output), + re.MULTILINE, + ) + ) def __init__( self, diff --git a/src/cli_agent_orchestrator/services/terminal_service.py b/src/cli_agent_orchestrator/services/terminal_service.py index 9cdf7a0c8..571247041 100644 --- a/src/cli_agent_orchestrator/services/terminal_service.py +++ b/src/cli_agent_orchestrator/services/terminal_service.py @@ -1178,7 +1178,7 @@ def _notify_caller_of_deferred_failure( } -def _worker_is_started_direct(terminal_id: str, provider, message: str = "") -> bool: +def _worker_is_started_direct(terminal_id: str, provider, post_dispatch_output: str) -> bool: """Direct visible-screen status check bypassing the event-driven status cache. The deferred-init retry loop polls ``status_monitor.get_status()`` which @@ -1198,15 +1198,23 @@ def _worker_is_started_direct(terminal_id: str, provider, message: str = "") -> dispatch bookkeeping and cannot distinguish IDLE from COMPLETED on a rendered capture-pane snapshot. - A started status is only half the verdict: the pane must also attribute it - to ``message``, the submission being confirmed. ``get_status`` classifies - the frame as a whole, so activity that predates the submission (a - provider's startup spinner or bullet still on screen) reads as started for - a pane whose task paste was dropped — and accepting it here would suppress - the redelivery this probe exists to gate. The provider decides attribution - through ``direct_probe_confirms_submission`` (default: the status alone, - for TUIs whose indicators are turn-scoped); a False there falls through - to the box check and redelivery like any other not-started read. + A started status is only half the verdict, and the weaker half: it is read + from a rendered frame, which ``get_status`` classifies as a whole, so + activity that PREDATES the submission (a provider's startup spinner or + bullet still on screen) reads as started for a pane whose task paste was + dropped. Accepting that would suppress the very redelivery this probe + gates, silently losing the task. + + ``post_dispatch_output`` supplies the missing causality. It is the + StatusMonitor rolling buffer, which ``send_input`` empties immediately + before sending the keystrokes, so its contents arrived after the dispatch + by construction — evidence that cannot be forged by a shared message + prefix, evicted by a bounded capture tail, or confused with leftover + chrome. The provider judges it through ``direct_probe_confirms_dispatch`` + (default: the status alone, for TUIs whose indicators are turn-scoped). + + False means UNPROVEN rather than idle; the caller keeps the recoveries + that cannot duplicate work and withholds the one that can. """ try: metadata = get_terminal_metadata(terminal_id) @@ -1220,7 +1228,7 @@ def _worker_is_started_direct(terminal_id: str, provider, message: str = "") -> status = provider.get_status(output) if status not in _DEFERRED_STARTED_STATUSES: return False - return bool(provider.direct_probe_confirms_submission(output, message)) + return bool(provider.direct_probe_confirms_dispatch(post_dispatch_output)) except Exception: logger.debug( "Direct status probe for %s failed (falling through to cached path)", @@ -1276,14 +1284,25 @@ def redeliver_dropped_message( path (#479) and the synchronous step path (#562). First, when the provider opts in via ``supports_direct_status_probe``, a live capture-pane check catches a worker that IS already running but whose cached status lags - behind (#496) — returns True (started) without sending anything. A caller - that already holds the provider instance passes it; otherwise it is - resolved from the registry, best-effort (a resolution failure means no - probe, never a failed redelivery). Then the box check picks the - redelivery: if the delivered text is still visible in the rendered pane - only the Enter was swallowed (send a bare Enter); if it is absent the - paste itself was dropped (re-deliver in full). See - ``_message_visible_in_box`` for why guessing wrong must be avoided. + behind (#496) — returns True (started) without sending anything. That + check is bound to THIS submission by the StatusMonitor rolling buffer, + which ``send_input`` empties at the dispatch boundary, so leftover chrome + from before the send can never vouch for it (see + ``_worker_is_started_direct``). A caller that already holds the provider + instance passes it; otherwise it is resolved from the registry, + best-effort (a resolution failure means no probe, never a failed + redelivery). Then the box check picks the redelivery: if the delivered + text is still visible in the rendered pane only the Enter was swallowed + (send a bare Enter); if it is absent the paste itself was dropped + (re-deliver in full). See ``_message_visible_in_box`` for why guessing + wrong must be avoided. + + The full re-send is the one branch that can duplicate work, so it is also + withheld from a probe-capable provider whenever ANY output arrived after + the dispatch without proving the turn started: an accepted turn that has + out-scrolled its own echo presents exactly like a dropped paste to a + bounded capture, and absence of proof is not proof of a drop. A genuinely + dropped paste produces no post-dispatch output and so still re-sends. ``full_resend_requires_probe`` gates the full re-send on the provider being probe-capable. Reason: ``_message_visible_in_box`` scans the whole @@ -1312,8 +1331,13 @@ def redeliver_dropped_message( probe_capable = provider is not None and getattr( provider, "supports_direct_status_probe", False ) + post_dispatch_output = "" if probe_capable: - if _worker_is_started_direct(terminal_id, provider, message): + try: + post_dispatch_output = status_monitor.get_buffer(terminal_id) + except Exception: # noqa: BLE001 — evidence is best-effort + post_dispatch_output = "" + if _worker_is_started_direct(terminal_id, provider, post_dispatch_output): return True if _message_visible_in_box(terminal_id, message): logger.warning( @@ -1323,6 +1347,23 @@ def redeliver_dropped_message( ) send_special_key(terminal_id, "Enter") return False + if probe_capable and post_dispatch_output: + # The turn was not PROVEN started, but the terminal did emit output + # after the dispatch and our text is not on screen. Those are exactly + # the observations an ACCEPTED turn produces once its own echo has + # scrolled out of the captured tail, and re-pasting into it would run + # the task twice. Absence of proof is not proof of a dropped paste: + # withhold the full re-send (the only irreversible branch) and let the + # caller's deadline classify. A genuinely dropped paste emits nothing + # after the dispatch and so never reaches here. + logger.warning( + "Delivery to %s unconfirmed but the terminal emitted output after " + "dispatch; withholding a full re-send that could duplicate the task " + "(attempt %d)", + terminal_id, + attempt, + ) + return False if full_resend_requires_probe and not probe_capable: # No probe → cannot rule out a working worker whose prompt left the # pane; a full re-send could silently duplicate the task. Skip the @@ -1373,8 +1414,8 @@ async def _confirm_worker_started_or_resubmit( for attempt in range(1, _DEFERRED_SUBMIT_MAX_RESUBMITS + 1): # The redelivery decision (box check + #496's direct-probe guard for - # providers that opt in) lives in ``redeliver_dropped_message`` — - # shared with the synchronous step path (#562). + # providers that opt in, bound to post-dispatch output) lives in + # ``redeliver_dropped_message`` — shared with the step path (#562). already_started = await asyncio.to_thread( redeliver_dropped_message, terminal_id, diff --git a/test/services/test_deferred_submit_verification.py b/test/services/test_deferred_submit_verification.py index a073372ea..4075fd730 100644 --- a/test/services/test_deferred_submit_verification.py +++ b/test/services/test_deferred_submit_verification.py @@ -50,6 +50,7 @@ def test_resolves_provider_from_registry_for_direct_probe(self): with ( patch.object(ts, "provider_manager") as mgr, patch.object(ts, "_worker_is_started_direct", return_value=True) as probe, + patch.object(ts.status_monitor, "get_buffer", return_value="• Working (1s)"), patch.object(ts, "send_special_key") as key, patch.object(ts, "send_input") as send, ): @@ -57,8 +58,9 @@ def test_resolves_provider_from_registry_for_direct_probe(self): started = ts.redeliver_dropped_message("t1", "Analyze the logs", 1) assert started is True mgr.get_provider.assert_called_once_with("t1") - # The message rides along: the probe binds its verdict to it. - probe.assert_called_once_with("t1", provider, "Analyze the logs") + # The post-dispatch bytes ride along: they are what binds the verdict + # to THIS submission, so the probe must receive them, not the message. + probe.assert_called_once_with("t1", provider, "• Working (1s)") key.assert_not_called() send.assert_not_called() @@ -298,15 +300,15 @@ class TestWorkerIsStartedDirect: def test_returns_false_when_metadata_is_none(self): with patch.object(ts, "get_terminal_metadata", return_value=None): - assert ts._worker_is_started_direct("t1", MagicMock()) is False + assert ts._worker_is_started_direct("t1", MagicMock(), "evidence") is False def test_returns_false_when_session_key_missing(self): with patch.object(ts, "get_terminal_metadata", return_value={"tmux_window": "w1"}): - assert ts._worker_is_started_direct("t1", MagicMock()) is False + assert ts._worker_is_started_direct("t1", MagicMock(), "evidence") is False def test_returns_false_when_window_key_missing(self): with patch.object(ts, "get_terminal_metadata", return_value={"tmux_session": "s1"}): - assert ts._worker_is_started_direct("t1", MagicMock()) is False + assert ts._worker_is_started_direct("t1", MagicMock(), "evidence") is False def test_returns_false_when_get_history_raises(self): with ( @@ -321,7 +323,7 @@ def test_returns_false_when_get_history_raises(self): patch.object(ts, "get_backend") as mock_be, ): mock_be.return_value.get_history.side_effect = Exception("capture failed") - assert ts._worker_is_started_direct("t1", MagicMock()) is False + assert ts._worker_is_started_direct("t1", MagicMock(), "evidence") is False def test_returns_false_when_get_status_raises(self): provider = MagicMock() @@ -337,7 +339,7 @@ def test_returns_false_when_get_status_raises(self): ), patch.object(ts, "get_backend") as mock_be, ): - assert ts._worker_is_started_direct("t1", provider) is False + assert ts._worker_is_started_direct("t1", provider, "evidence") is False def test_returns_true_when_status_is_processing(self): from cli_agent_orchestrator.models.terminal import TerminalStatus @@ -355,7 +357,7 @@ def test_returns_true_when_status_is_processing(self): ), patch.object(ts, "get_backend") as mock_be, ): - assert ts._worker_is_started_direct("t1", provider) is True + assert ts._worker_is_started_direct("t1", provider, "evidence") is True def test_returns_false_when_status_is_idle(self): from cli_agent_orchestrator.models.terminal import TerminalStatus @@ -373,7 +375,7 @@ def test_returns_false_when_status_is_idle(self): ), patch.object(ts, "get_backend") as mock_be, ): - assert ts._worker_is_started_direct("t1", provider) is False + assert ts._worker_is_started_direct("t1", provider, "evidence") is False class TestCodexDirectProbeOptIn: @@ -382,16 +384,27 @@ class TestCodexDirectProbeOptIn: (detection fires only at rising-edge/quiescence, and a repainting spinner defers quiescence). Without the direct-probe opt-in the confirm loop re-delivers the task into the working pane up to three times and then tears - the worker down. These pin the opt-in end to end with the REAL provider and - a real rendered frame, so removing the flag (or breaking the detector on - this shape) goes red here — not just in a unit assert on the attribute. + the worker down. + + The opt-in alone is not safe, though: ``get_status`` classifies a rendered + frame as a whole, and Codex's startup chrome renders like its task activity + (the ``Starting MCP servers`` spinner IS ``TUI_PROGRESS_PATTERN``; any + startup bullet is an assistant marker). So the verdict is bound to + POST-DISPATCH BYTES: ``send_input`` empties the StatusMonitor rolling buffer + immediately before sending keystrokes, so a spinner that stopped before the + dispatch contributes nothing there no matter where it still sits on screen, + while a live one repaints its counter and necessarily does. + + These drive the REAL provider through the real redelivery decision, so a + regression in either half goes red here — not just in a unit assert. """ _MESSAGE = "[CAO Handoff] Supervisor terminal ID: sup-123. Do the task." + _FOOTER = " gpt-5.6-sol medium · Context 100% left\n" + _SPINNER = "• Working (3s • esc to interrupt)\n" # The shape from the issue report: handoff prompt in the transcript, live - # Working spinner, TUI footer. Same frame family the codex provider unit - # tests pin as PROCESSING. + # Working spinner, TUI footer. _WORKING_FRAME = ( "› [CAO Handoff] Supervisor terminal ID: sup-123. Do the task.\n" "\n" @@ -402,41 +415,38 @@ class TestCodexDirectProbeOptIn: " ? for shortcuts 100% context left\n" ) - # Startup chrome as Codex renders it before any task exists. The MCP - # startup spinner IS the TUI progress pattern, and any startup bullet is - # an assistant marker. + # Startup chrome as Codex renders it before any task exists. _STARTUP_BANNER = ( "╭──────────────────────────────────────────────╮\n" "│ >_ OpenAI Codex (v0.145.0) │\n" - "│ │\n" - "│ model: gpt-5.6-sol medium │\n" - "│ directory: ~/project │\n" "│ permissions: YOLO mode │\n" "╰──────────────────────────────────────────────╯\n" "\n" "• Starting MCP servers (4s • esc to interrupt)\n" ) - _IDLE_COMPOSER = ( - "\n" "› Write tests for @filename\n" "\n" " gpt-5.6-sol medium · Context 100% left\n" - ) + _IDLE_COMPOSER = "\n› Write tests for @filename\n\n" + _FOOTER - @staticmethod - def _residue_frame(gap: int, composer: str) -> str: + @classmethod + def _residue_frame(cls, gap: int, composer: str) -> str: """Startup spinner ``gap`` blank lines above the composer: outside the bottom-15 window initialize() vetoes activity in, so the provider reports ready, yet inside (or above) the wider tail get_status scans.""" - return TestCodexDirectProbeOptIn._STARTUP_BANNER + "\n" * gap + composer + return cls._STARTUP_BANNER + "\n" * gap + composer @staticmethod - async def _run_confirm(frame: str, message: str): + def _provider(): from cli_agent_orchestrator.providers.codex import CodexProvider - provider = CodexProvider("t1", "s1", "w0") + return CodexProvider("t1", "s1", "w0") + + @classmethod + def _redeliver(cls, frame: str, message: str, post_dispatch: str): + """One redelivery decision against a rendered ``frame`` and the bytes + that arrived since the dispatch. Returns (started, enter, full_resend).""" + provider = cls._provider() backend = MagicMock() backend.get_history.return_value = frame with ( - # Cached status stays IDLE past every poll — the #496-class lag. - patch.object(ts, "wait_until_status", new=AsyncMock(return_value=False)), patch.object( ts, "get_terminal_metadata", @@ -444,13 +454,12 @@ async def _run_confirm(frame: str, message: str): ), patch.object(ts, "get_backend", return_value=backend), patch.object(ts, "get_output", return_value=frame), + patch.object(ts.status_monitor, "get_buffer", return_value=post_dispatch), patch.object(ts, "send_special_key") as key, patch.object(ts, "send_input") as send, ): - ok = await ts._confirm_worker_started_or_resubmit( - "t1", message, None, "sup", None, provider=provider - ) - return ok, key, send + started = ts.redeliver_dropped_message("t1", message, 1, provider) + return started, key.called, send.called def test_codex_opts_into_direct_status_probe(self): from cli_agent_orchestrator.providers.codex import CodexProvider @@ -459,9 +468,9 @@ def test_codex_opts_into_direct_status_probe(self): @pytest.mark.asyncio async def test_codex_confirm_succeeds_from_live_frame_without_redelivery(self): - from cli_agent_orchestrator.providers.codex import CodexProvider - - provider = CodexProvider("t1", "s1", "w0") + """End to end through the confirm loop: cached status stays IDLE past + every poll, the pane shows a live turn, nothing is typed into it.""" + provider = self._provider() backend = MagicMock() backend.get_history.return_value = self._WORKING_FRAME with ( @@ -473,16 +482,12 @@ async def test_codex_confirm_succeeds_from_live_frame_without_redelivery(self): return_value={"tmux_session": "s1", "tmux_window": "w0"}, ), patch.object(ts, "get_backend", return_value=backend), + patch.object(ts.status_monitor, "get_buffer", return_value=self._SPINNER * 3), patch.object(ts, "send_special_key") as key, patch.object(ts, "send_input") as send, ): ok = await ts._confirm_worker_started_or_resubmit( - "t1", - self._MESSAGE, - None, - "sup", - None, - provider=provider, + "t1", self._MESSAGE, None, "sup", None, provider=provider ) # Started: the caller must not classify this as a dropped submit, so the @@ -493,120 +498,117 @@ async def test_codex_confirm_succeeds_from_live_frame_without_redelivery(self): send.assert_not_called() key.assert_not_called() - # --- the verdict is bound to the submission, not to the pane's status ----- - # get_status classifies the frame as a whole, so startup residue reads as - # started for a pane whose task paste was dropped. The probe must not take - # that as acceptance: it would skip the redelivery this path exists for and - # the task would be silently lost with the supervisor waiting forever. + # --- the verdict is bound to post-dispatch bytes, not to the frame ------- @pytest.mark.parametrize( "gap, expected_status", [ - (12, "processing"), # spinner inside get_status's 25-line spinner tail + (12, "processing"), # spinner inside get_status's 25-line tail (20, "processing"), # ...at its far edge - (28, "completed"), # spinner out of the tail: the bullet is an assistant marker + (28, "completed"), # spinner out of the tail: the bullet is a marker ], ) - @pytest.mark.asyncio - async def test_startup_residue_on_a_dropped_task_still_redelivers(self, gap, expected_status): - from cli_agent_orchestrator.providers.codex import CodexProvider, _has_startup_idle_composer + def test_startup_residue_on_a_dropped_task_still_redelivers(self, gap, expected_status): + """A task-less pane reads STARTED on the frame at every distance, yet + emitted nothing since the dispatch — so the task really was dropped.""" + from cli_agent_orchestrator.providers.codex import _has_startup_idle_composer frame = self._residue_frame(gap, self._IDLE_COMPOSER) - # Precondition — this is exactly the frame the finding describes: the - # provider reports ready (initialize() would have returned) while the - # whole-frame status says started, and the message is nowhere. + # Precondition — the frame alone is genuinely misleading: the provider + # reports ready (initialize() would have returned) while the whole-frame + # status says started, and the message is nowhere. assert _has_startup_idle_composer(frame) is True - assert CodexProvider("t1", "s1", "w0").get_status(frame).value == expected_status - assert self._MESSAGE[:12] not in frame + assert self._provider().get_status(frame).value == expected_status + assert "sup-123" not in frame - ok, key, send = await self._run_confirm(frame, self._MESSAGE) - - # Not started: the paste was dropped, so every attempt re-delivers the - # full message and the caller gets to classify the outcome. - assert ok is False - assert send.call_count == ts._DEFERRED_SUBMIT_MAX_RESUBMITS - key.assert_not_called() - - @pytest.mark.asyncio - async def test_startup_residue_with_unsubmitted_text_sends_enter(self): - # Paste landed, Enter was swallowed: the message sits in the composer - # under the residue. The bare-Enter recovery must still fire. - composer = "\n› " + self._MESSAGE + "\n\n gpt-5.6-sol medium · Context 100% left\n" - frame = self._residue_frame(18, composer) - - ok, key, send = await self._run_confirm(frame, self._MESSAGE) - - assert ok is False - assert key.call_count == ts._DEFERRED_SUBMIT_MAX_RESUBMITS - send.assert_not_called() + started, enter, full_resend = self._redeliver(frame, self._MESSAGE, "") - @pytest.mark.asyncio - async def test_startup_residue_above_an_accepted_task_is_started(self): - # Residue AND a real accepted turn: the echo of our message with the - # turn's activity below it is the causal evidence; residue above it - # neither adds nor subtracts. - frame = self._residue_frame(18, self._WORKING_FRAME) + assert started is False + assert full_resend is True + assert enter is False - ok, key, send = await self._run_confirm(frame, self._MESSAGE) + def test_stale_activity_of_a_previous_turn_does_not_confirm_a_new_message(self): + """A completed EARLIER handoff, still on screen, must not vouch for a + message that was never delivered. Its bullet predates the dispatch, so + it contributes no post-dispatch bytes.""" + frame = ( + "› [CAO Handoff] Supervisor terminal ID: OLD-999. Do the old task.\n" + "• Finished the old task\n" + "\n› Write tests for @filename\n\n" + self._FOOTER + ) + assert self._provider().get_status(frame).value == "completed" - assert ok is True - send.assert_not_called() - key.assert_not_called() + provider = self._provider() + assert provider.direct_probe_confirms_dispatch("") is False + assert ts._worker_is_started_direct("t1", provider, "") is False - @pytest.mark.asyncio - async def test_accepted_turn_on_an_approval_prompt_is_started(self): - # WAITING_USER_ANSWER after our turn (codex 0.147 approval menu, as in - # test/providers/fixtures/codex_approval_modal_raw.txt): the activity - # bullet below the echo binds it; the probe must not blind-Enter into - # the menu (that would select "Yes, proceed"). + def test_accepted_turn_is_confirmed_even_after_its_echo_scrolls_away(self): + """An accepted turn can out-scroll its own echo while a spinner remains. + The frame no longer contains the message, but the live spinner is + post-dispatch output, so the turn is confirmed and NOT re-sent.""" frame = ( - "› " + self._MESSAGE + "\n" - "• Running mkdir -p /tmp/work/subdir\n" - " Would you like to run the following command?\n" - " $ mkdir -p /tmp/work/subdir\n" - "› 1. Yes, proceed (y)\n" - " 2. Yes, and don't ask again for commands that start with `mkdir` (p)\n" - " 3. No, and tell Codex what to do differently (esc)\n" - " Press enter to confirm or esc to cancel\n" + "\n".join(f" output line {i}" for i in range(190)) + + "\n" + + self._SPINNER + + self._FOOTER ) - ok, key, send = await self._run_confirm(frame, self._MESSAGE) + assert self._provider().get_status(frame).value == "processing" + assert "sup-123" not in frame - assert ok is True - send.assert_not_called() - key.assert_not_called() + started, enter, full_resend = self._redeliver(frame, self._MESSAGE, self._SPINNER * 4) - def test_bullets_inside_the_pasted_message_do_not_self_attribute(self): - from cli_agent_orchestrator.providers.codex import CodexProvider + assert started is True + assert full_resend is False, "re-pasting here would run the task twice" + assert enter is False - # An unsubmitted multi-line paste whose own lines are bullets: the - # bullets below the echo line belong to the message, not to a reply. - message = "Review these findings for me:\n• first finding\n• second finding" - frame = self._residue_frame( - 18, - "\n› Review these findings for me:\n • first finding\n • second finding\n" - "\n gpt-5.6-sol medium · Context 100% left\n", - ) - provider = CodexProvider("t1", "s1", "w0") - assert provider.direct_probe_confirms_submission(frame, message) is False - # ...while a reply bullet under the same paste does bind it. - assert ( - provider.direct_probe_confirms_submission( - frame.replace( - " • second finding\n", " • second finding\n• Reviewing the findings\n" - ), - message, - ) - is True - ) + def test_reply_wording_that_echoes_the_prompt_still_confirms(self): + """A valid fast reply whose words appear in the prompt must not be + discarded — the verdict never inspects the message text.""" + message = "Reply with Done" + frame = "› Reply with Done\n• Done\n\n› Write tests for @filename\n\n" + self._FOOTER + assert self._provider().get_status(frame).value == "completed" - def test_short_message_cannot_bind(self): - from cli_agent_orchestrator.providers.codex import CodexProvider + started, enter, full_resend = self._redeliver(frame, message, self._SPINNER + "• Done\n") - # Below the 8-character floor the collapse cannot match reliably; the - # hook must refuse rather than guess (same floor as the box check). - assert ( - CodexProvider("t1", "s1", "w0").direct_probe_confirms_submission( - "› hi\n• Working (3s • esc to interrupt)\n", "hi" - ) - is False - ) + assert started is True + assert enter is False and full_resend is False + + def test_pasted_bullets_cannot_confirm_their_own_delivery(self): + """The composer echoes a paste as it renders, so a multi-line message + carrying its own bullet emits one without any turn starting. Only the + spinner's ``(s • esc to interrupt)`` shape counts as evidence.""" + provider = self._provider() + echoed_paste = "› Review these findings:\n • first finding\n • second finding\n" + + assert provider.direct_probe_confirms_dispatch(echoed_paste) is False + assert provider.direct_probe_confirms_dispatch(echoed_paste + self._SPINNER) is True + + def test_unicode_and_empty_post_dispatch_output_are_unproven(self): + provider = self._provider() + assert provider.direct_probe_confirms_dispatch("") is False + assert provider.direct_probe_confirms_dispatch("› Review this:\n • 日本語\n") is False + + def test_spinner_is_recognized_through_terminal_escapes(self): + """The buffer holds the RAW byte stream, so the evidence must survive + the SGR/cursor sequences a repainting TUI interleaves.""" + provider = self._provider() + raw = "\x1b[2K\x1b[1;32m• Working (12s • esc to interrupt)\x1b[0m\r\n" + assert provider.direct_probe_confirms_dispatch(raw) is True + + def test_unproven_delivery_with_post_dispatch_output_withholds_the_resend(self): + """Absence of proof is not proof of a dropped paste: when the terminal + emitted output after the dispatch and our text is not on screen, the + full re-send (the only branch that can duplicate work) is withheld.""" + frame = "\n".join(f" output line {i}" for i in range(190)) + "\n" + self._FOOTER + + started, enter, full_resend = self._redeliver(frame, self._MESSAGE, " a line of output\n") + + assert started is False + assert full_resend is False + assert enter is False + + def test_probe_declines_when_the_backend_feeds_no_buffer(self): + """Event-inbox backends (herdr) never push a byte buffer. The probe then + vouches for nothing and the caller keeps its pre-existing behavior.""" + provider = self._provider() + assert ts._worker_is_started_direct("t1", provider, "") is False