diff --git a/CHANGELOG.md b/CHANGELOG.md index b13586fbe..cecdbc6a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **`cao launch` could drop the initial task, or tear down a worker that had + already done it.** The initial message is now delivered by the server as + part of `POST /sessions` instead of a second request that raced provider + startup, and the terminal reads as not-yet-completable until that delivery + has been made and the worker has produced output for it. Confirmation gates + on an output-only generation sampled at the dispatch boundary, inside the + send, so neither a completion cached from provider startup nor a redelivery's + own keystrokes can pass for this task starting, and a worker fast enough to + finish before the send returns is confirmed rather than resubmitted to and + deleted (#566) + - **enabling `CAO_MEMORY_API_URL` rejected memory keys that work without it.** The `/internal/memory/store` and `/forget` routes validated the wire `key` as the strict `MemoryKey` (`^[a-z0-9-]{1,60}$`), while the MCP tools have always diff --git a/src/cli_agent_orchestrator/cli/commands/launch.py b/src/cli_agent_orchestrator/cli/commands/launch.py index 457ea65e5..938ede54b 100644 --- a/src/cli_agent_orchestrator/cli/commands/launch.py +++ b/src/cli_agent_orchestrator/cli/commands/launch.py @@ -1,7 +1,6 @@ """Launch command for CLI Agent Orchestrator CLI.""" import os -import time import click import requests @@ -47,6 +46,53 @@ # the two client paths cannot drift) and are mirrored server-side in # ``TmuxClient._merge_extra_env``. See issue #248. +# How long the CLI allows for server-side provider init: the pre-attach +# readiness poll on the non-headless path, and the init allowance folded into +# the headless wait (where init now runs inside ``poll_until_done``'s window +# rather than in a separate poll before a client-side send). +_READINESS_WAIT_TIMEOUT = 120 + +# How long the agent gets to finish MESSAGE on the headless non-async path, +# on top of ``_READINESS_WAIT_TIMEOUT``. +_HEADLESS_TASK_TIMEOUT = 300 + + +def _is_waiting_on_user(terminal_id: str) -> bool: + """Return True when the terminal's live status is WAITING_USER_ANSWER. + + Read separately because ``wait_until_terminal_status`` reports only whether + one of its target statuses was reached, not which one, and the pre-attach + poll accepts three. + + Best-effort by design: this only decides which advisory line to print before + attaching, so a transport blip must not turn a successful launch into a + ``ClickException``. Hence the local except rather than letting it reach the + caller's ``RequestException`` handler, which reports "Failed to connect to + cao-server" — untrue here, since the poll above just talked to it. + + An *unparseable* body is covered by the except: + ``requests.exceptions.JSONDecodeError`` subclasses ``RequestException`` (and + ``ValueError``) and has since requests 2.27, below this project's + ``requests>=2.32.0`` floor. A body that parses but isn't an object is not — + ``[].get`` raises ``AttributeError``, which is no kind of + ``RequestException`` — so the shape is checked rather than assumed. Without + that check a 200 carrying a JSON array, string or ``null`` escapes to the + caller's generic handler and aborts the launch with ``exit 1`` *after* the + session exists, leaving it orphaned in tmux: the precise failure this + function's local except is here to prevent. + """ + try: + resp = requests.get(f"{API_BASE_URL}/terminals/{terminal_id}", timeout=5.0) + if resp.status_code == 200: + payload = resp.json() + if isinstance(payload, dict): + # bool(): ``payload.get`` is Any, so the comparison is too, and + # this function is annotated ``-> bool``. + return bool(payload.get("status") == TerminalStatus.WAITING_USER_ANSWER.value) + except requests.exceptions.RequestException: + pass + return False + def _parse_env_pairs(pairs): """Parse repeated ``KEY=VALUE`` entries into a validated dict. @@ -289,13 +335,45 @@ def launch( if resume_session_id: params["resume_session_id"] = resume_session_id + # Hand MESSAGE to the server rather than sending it ourselves. + # ``initial_message`` on ``POST /sessions`` puts the initial terminal on + # the existing deferred-init path (``session_service.create_session`` -> + # ``create_terminal(defer_init=True)``): the server responds as soon as + # the terminal record exists, then finishes provider init, delivers the + # message, and confirms/re-submits if the TUI swallowed it. + # + # The CLI used to create the session and then issue a SEPARATE + # ``POST /terminals/{id}/input``. Because ``POST /sessions`` ran the + # provider's full ``initialize()`` inline, a slow cold start outlived + # the client's read timeout: ``requests`` raised ``ReadTimeout``, + # ``launch`` reported "Failed to connect to cao-server", and that second + # request never happened — MESSAGE was silently dropped even though the + # session, the terminal and a healthy idle TUI all existed server-side, + # and nothing retried because from the server's point of view the launch + # had succeeded. Server-side delivery closes the window for every + # provider at once; ``cao launch`` was the last client still doing its + # own create-then-send (mcp_server and ops_mcp_server already pass + # ``initial_message``). + # + # Headless only: that is the path that used to send MESSAGE. A + # non-headless launch attaches instead and has never delivered MESSAGE, + # and deferring init there would make the pre-attach readiness poll + # below race the agent's first turn. + server_delivers_message = bool(message) and headless + # Forwarded env vars travel in the JSON body so values (which may # contain secrets) don't end up in cao-server's HTTP access log. - # See issue #248. - request_timeout = get_server_settings()["mcp_request_timeout"] - post_kwargs: dict = {"params": params, "timeout": request_timeout} + # MESSAGE rides in the body for the same reason, plus URL-length. See + # issue #248 and ``CreateSessionBody``. + settings = get_server_settings() + post_kwargs: dict = {"params": params, "timeout": settings["mcp_request_timeout"]} + body: dict = {} if forwarded_env: - post_kwargs["json"] = {"env_vars": forwarded_env} + body["env_vars"] = forwarded_env + if server_delivers_message: + body["initial_message"] = message + if body: + post_kwargs["json"] = body response = requests.post(url, **post_kwargs) response.raise_for_status() @@ -311,6 +389,21 @@ def launch( # silently drops keystrokes. See issue #220. The wait is advisory: # if it times out we still attach so the user can inspect the # half-initialized session rather than orphan it in tmux. + # + # WAITING_USER_ANSWER counts as settled here, not as a stall. A provider + # can finish initializing on a screen that legitimately needs the + # operator — Codex's first-run login menu is the case that forced this — + # and such a screen never becomes IDLE on its own, so waiting for IDLE + # burned the full ``_READINESS_WAIT_TIMEOUT`` and then blamed init for a + # pane that was simply waiting for a human. Safe because non-headless + # ``POST /sessions`` initializes synchronously (no ``initial_message``, + # see ``server_delivers_message`` above), so by the time this poll runs + # every provider's startup handler has already returned: a + # WAITING_USER_ANSWER here is a settled prompt, not a dialog caught + # mid-dismissal. If non-headless init is ever deferred, this poll would + # race the startup handler and attaching early would resize the pty + # mid-init — issue #220 again — so that change must gate attach on init + # completion rather than reuse this set. if not headless: # Align the CLI's backend singleton with the running server. # Without this, ``cao-server --terminal herdr`` + no config.json @@ -318,45 +411,55 @@ def launch( sync_backend_from_server() ready = wait_until_terminal_status( terminal["id"], - {TerminalStatus.IDLE, TerminalStatus.COMPLETED}, - timeout=120, + { + TerminalStatus.IDLE, + TerminalStatus.COMPLETED, + TerminalStatus.WAITING_USER_ANSWER, + }, + timeout=_READINESS_WAIT_TIMEOUT, ) if not ready: click.echo( click.style( - f" Warning: {terminal['id']} did not reach idle within 120s — " - "attaching anyway; input may be unreliable until init completes.", + f" Warning: {terminal['id']} did not reach idle within " + f"{_READINESS_WAIT_TIMEOUT}s — attaching anyway; input may be " + "unreliable until init completes.", + fg="yellow", + ) + ) + elif _is_waiting_on_user(terminal["id"]): + click.echo( + click.style( + f" {terminal['id']} is waiting for an answer in the pane " + "(a first-run sign-in, for example) — complete it after " + "attaching.", fg="yellow", ) ) get_backend().attach_session(terminal["session_name"]) elif message: - ready = wait_until_terminal_status( - terminal["id"], - {TerminalStatus.IDLE, TerminalStatus.COMPLETED}, - timeout=120, - ) - if not ready: - raise click.ClickException( - f"Conductor {terminal['id']} did not become ready within 120s" - ) - request_timeout = get_server_settings()["mcp_request_timeout"] - response = requests.post( - f"{API_BASE_URL}/terminals/{terminal['id']}/input", - params={"message": message}, - timeout=request_timeout, - ) - response.raise_for_status() - time.sleep(3) + # Nothing to send: the server took MESSAGE in the create body above + # and owns init, delivery and re-submission. There is also nothing + # left to wait for before delivery — waiting for IDLE here is what + # used to gate a send that no longer happens. if is_async: - click.echo(f"Message sent to {terminal['name']}. Running in background.") + click.echo( + f"Message accepted for {terminal['name']}; the server delivers it " + "once provider init completes. Running in background." + ) return - poll_until_done(terminal["id"], timeout=300) - request_timeout = get_server_settings()["mcp_request_timeout"] + # Provider init now happens inside this wait instead of in a + # separate readiness poll before the send, so the budget covers + # both. A deferred terminal reports UNKNOWN until init finishes, + # which ``poll_until_done`` deliberately does not count as "started" + # — it returns only once the agent has been observed working. + poll_until_done( + terminal["id"], timeout=_READINESS_WAIT_TIMEOUT + _HEADLESS_TASK_TIMEOUT + ) output_resp = requests.get( f"{API_BASE_URL}/terminals/{terminal['id']}/output", params={"mode": "last"}, - timeout=request_timeout, + timeout=settings["mcp_request_timeout"], ) output_resp.raise_for_status() output = output_resp.json().get("output", "") diff --git a/src/cli_agent_orchestrator/services/session_service.py b/src/cli_agent_orchestrator/services/session_service.py index b75bb8809..b442dcb52 100644 --- a/src/cli_agent_orchestrator/services/session_service.py +++ b/src/cli_agent_orchestrator/services/session_service.py @@ -295,9 +295,15 @@ def get_session(session_name: str) -> Dict: # single source of truth and is backend-aware (tmux push vs herdr # native), so derive it here rather than persisting a stale column. from cli_agent_orchestrator.services.status_monitor import status_monitor + from cli_agent_orchestrator.services.terminal_service import reported_status for terminal in terminals: - terminal["status"] = status_monitor.get_status(terminal["id"]).value + # reported_status keeps this in step with GET /terminals/{id}: a + # terminal whose accepted initial message has not been dispatched yet + # must not read IDLE/COMPLETED anywhere a client can see it (#566). + terminal["status"] = reported_status( + terminal["id"], status_monitor.get_status(terminal["id"]) + ).value return {"session": session_data, "terminals": terminals} except Exception as e: diff --git a/src/cli_agent_orchestrator/services/status_monitor.py b/src/cli_agent_orchestrator/services/status_monitor.py index cd424b3f2..ae24cf55d 100644 --- a/src/cli_agent_orchestrator/services/status_monitor.py +++ b/src/cli_agent_orchestrator/services/status_monitor.py @@ -142,6 +142,20 @@ def __init__(self): # applied across that boundary would consume the arm and latch-block the new # turn's genuine PROCESSING. self._capture_generation: Dict[str, int] = {} + # Per-terminal OUTPUT-ONLY generation. Bumped under the lock by + # _process_chunk alone -- never by notify_input_sent -- so a strictly + # greater value than one sampled at a dispatch boundary means real output + # has landed since that boundary. This is what output_generation() exposes + # and what delivery confirmation gates on (PR #566). It is kept separate + # from _capture_generation on purpose: that counter must ALSO advance on + # notify_input_sent (a new turn invalidates in-flight capture verdicts), + # and a counter that moves on arm cannot distinguish "the worker produced + # output" from "we sent keys again" -- a resubmit's own arm would satisfy + # the gate on a still-cached pre-dispatch COMPLETED. The byte-buffer epoch + # that clear_rolling_buffer hands stateful providers is not a substitute + # either: it advances on CLEAR, never on output, so it cannot say whether + # anything arrived after the boundary. + self._output_generation: Dict[str, int] = {} # --- pyte rendered-screen detection state (only used when CAO_PYTE_STATUS # is on AND the provider opts in via supports_screen_detection) --- # Per-terminal pyte Screen+Stream that composites the raw byte stream @@ -230,6 +244,7 @@ def _process_chunk(self, terminal_id: str, chunk: str) -> None: # by a later read. self._buffer_changed_at[terminal_id] = time.monotonic() self._capture_generation[terminal_id] = self._capture_generation.get(terminal_id, 0) + 1 + self._output_generation[terminal_id] = self._output_generation.get(terminal_id, 0) + 1 self._pending_stale_capture.pop(terminal_id, None) if use_screen: self._feed_screen_locked(terminal_id, chunk) @@ -623,6 +638,51 @@ def _cancel_quiesce_handle(self, handle: Optional[asyncio.TimerHandle]) -> None: except RuntimeError: pass # loop already closed during shutdown — the timer is moot + def output_generation(self, terminal_id: str) -> int: + """Return the terminal's current OUTPUT generation. + + Advances in exactly one place: ``_process_chunk``, when a chunk of output + from the terminal lands. It does NOT advance in ``notify_input_sent``, so + a value strictly greater than one sampled at a dispatch boundary means the + terminal emitted output after that boundary; in particular a redelivery's + own arm cannot make it read that way. + + What that does and does not prove: it proves the FIFO reader delivered a + chunk after the sample, not that the chunk belongs to the dispatched task. + A late startup frame (spinner redraw, MCP startup line) landing after the + sample but before the pasted keys reach the pane counts too. That is the + same approximation the rolling-buffer status detection already makes -- + it parses whatever bytes have landed since the clear -- and it is the + conservative side of the previous defect, where a real completion was + rejected. Callers gate a started STATUS on this; they do not treat the + counter alone as completion evidence. + + This exists because a status VALUE cannot carry recency information. + ``notify_input_sent`` deliberately leaves ``_last_status`` alone while + arming the revert, so a ready status cached BEFORE a send is + indistinguishable from one earned after it — the defect PR #566 hit when + a pre-dispatch COMPLETED satisfied delivery confirmation instantly. + + The boundary must be sampled INSIDE the send, after the monitor is armed + and the rolling buffer cleared but before any key reaches the pane (see + ``terminal_service.dispatch_input``). Sampling after the send returns is + too late: ``send_keys`` includes the provider's submit delay, during which + a fast worker can emit and complete, and those chunks would then sit + inside the baseline -- a genuine completion would read as pre-dispatch + and the worker would be resubmitted to and finally torn down. + + Read-only and lock-guarded. Returns 0 for an unknown terminal, which is + below any real generation and so never reads as "something happened". + + NOT meaningful for event-inbox backends (herdr): they start no FIFO + reader, so ``_process_chunk`` never runs and this never advances. + ``get_status`` derives their status on demand instead, which is why they + have no staleness problem to solve and callers must not gate on this for + them. + """ + with self._lock: + return self._output_generation.get(terminal_id, 0) + def notify_input_sent(self, terminal_id: str, *, assume_processing: bool = False) -> None: """Arm the next PROCESSING transition. @@ -695,6 +755,7 @@ def clear_terminal(self, terminal_id: str) -> None: self._buffer_changed_at.pop(terminal_id, None) self._pending_stale_capture.pop(terminal_id, None) self._capture_generation.pop(terminal_id, None) + self._output_generation.pop(terminal_id, None) handle = self._quiesce_handle.pop(terminal_id, None) self._cancel_quiesce_handle(handle) @@ -720,6 +781,7 @@ def reset_buffer(self, terminal_id: str) -> None: self._buffer_changed_at.pop(terminal_id, None) self._pending_stale_capture.pop(terminal_id, None) self._capture_generation.pop(terminal_id, None) + self._output_generation.pop(terminal_id, None) handle = self._quiesce_handle.pop(terminal_id, None) self._cancel_quiesce_handle(handle) diff --git a/src/cli_agent_orchestrator/services/terminal_service.py b/src/cli_agent_orchestrator/services/terminal_service.py index 5cce19e11..cade2df4a 100644 --- a/src/cli_agent_orchestrator/services/terminal_service.py +++ b/src/cli_agent_orchestrator/services/terminal_service.py @@ -180,6 +180,76 @@ class TerminalRecordCorruptError(Exception): # silently leaving a worker uninitialized. Tasks drop themselves on completion. _deferred_init_tasks: set = set() +# Terminals whose deferred initial message has been ACCEPTED but not yet +# dispatched to the pane. Written on the event loop by _schedule_deferred_init, +# read from worker threads by the status-reporting helpers below, hence the lock +# (mirrors _memory_injected_lock above). +_pending_initial_delivery: set = set() +_pending_initial_delivery_lock = threading.Lock() + +# Statuses a poller may read as "there is nothing left to wait for". While an +# initial message is still pending these are exactly the readings that must not +# escape: see reported_status(). +_COMPLETABLE_STATUSES = (TerminalStatus.IDLE, TerminalStatus.COMPLETED) + + +def _mark_initial_delivery_pending(terminal_id: str) -> None: + """Record that an accepted initial message has not been dispatched yet.""" + with _pending_initial_delivery_lock: + _pending_initial_delivery.add(terminal_id) + + +def _clear_initial_delivery_pending(terminal_id: str) -> None: + """Release the pending mark. Idempotent — every exit path may call it.""" + with _pending_initial_delivery_lock: + _pending_initial_delivery.discard(terminal_id) + + +def initial_delivery_pending(terminal_id: str) -> bool: + """True while an accepted initial message has not reached the pane yet.""" + with _pending_initial_delivery_lock: + return terminal_id in _pending_initial_delivery + + +def reported_status(terminal_id: str, status: TerminalStatus) -> TerminalStatus: + """Mask a completable status while the initial message is still undispatched. + + PR #566 review (haofeif), P1. A client that polls for completion decides + "done" from the terminal's status alone, and every such poller needs the same + thing: evidence that is causally DOWNSTREAM of the send it is waiting on. + Provider startup does not qualify. On the deferred path the pane can sit at a + perfectly genuine IDLE for seconds after ``initialize()`` returns while + ``_schedule_deferred_init`` resolves shell_baseline and metadata and + ``send_input`` runs ``inject_memory_context`` — all strictly BEFORE any + keystroke is dispatched. A poller sampling that window sees a stable IDLE and + concludes the agent finished a task it was never given: the synchronous + ``cao launch`` printed empty output and exited 0. + + Reporting UNKNOWN closes that window at the source, for every client at once, + rather than asking each one to remember to wait for a separate handshake. + UNKNOWN is the honest reading — the terminal exists but is not tracking the + task yet — and pollers already treat it as neither progress nor completion + (``utils.terminal.poll_until_done``, ``services.agent_step``), so no number + of pre-dispatch samples can satisfy an idle gate. + + Deliberately NOT masked: + + - **WAITING_USER_ANSWER / PROCESSING / ERROR.** Masking WAITING_USER_ANSWER + would hide a pane genuinely parked on a prompt, which is the one state an + operator has to see to clear it (and which ``send_input``'s own guard turns + into a TerminalInputBlockedError that releases the pending mark anyway). + Letting it through is harmless here: it flips a poller's "has started" flag + early, but with IDLE masked no idle streak can accumulate to act on it. + - **The internal callers of ``status_monitor.get_status``.** send_input's + ERROR/WAITING_USER_ANSWER guards, inbox/flow/memory services and the + deferred path's own retry loop all need the RAW state; masking there would + change delivery decisions, not just reporting. This is a reporting-layer + concern only, applied where a status crosses the API boundary. + """ + if status in _COMPLETABLE_STATUSES and initial_delivery_pending(terminal_id): + return TerminalStatus.UNKNOWN + return status + def inject_memory_context( first_message: str, terminal_id: str, frozen_memory: str | None = None @@ -1898,6 +1968,54 @@ def redeliver_dropped_message( return False +async def _wait_for_post_dispatch_start( + terminal_id: str, + dispatch_generation: Optional[int], + timeout: float, + polling_interval: float = 0.5, +) -> bool: + """Wait for a started status that is EVIDENCE OF THIS TURN, not a cached one. + + Round-6 review (haofeif), P1. ``wait_until_status`` returns on the first poll + whose value is in the target set, and a status VALUE cannot say when it was + earned. Provider startup output can legitimately parse as COMPLETED, which + then latches (``_STICKY_READY_STATUSES``), and ``send_input`` only ARMS the + next transition without touching the cached value. So a pre-dispatch + COMPLETED satisfied confirmation instantly -- measured at 0.008s against an + exact head -- releasing the pending-delivery mask before the new task had + emitted anything. Two earlier release points failed for the same underlying + reason: they keyed on a status value rather than on its recency. + + ``dispatch_generation`` is ``status_monitor.output_generation()`` sampled at + the dispatch boundary INSIDE ``dispatch_input`` -- armed, buffer cleared, no + key sent yet. Requiring the current generation to EXCEED it means real output + has landed since dispatch, because that counter advances in ``_process_chunk`` + alone. Two things are deliberately NOT in it: the arm bump (so a redelivery's + own ``notify_input_sent`` cannot satisfy the gate on a still-cached + COMPLETED), and output emitted during ``send_keys``' submit delay (so a fast + worker that completes before the send returns is confirmed, not resubmitted + to and torn down). That is the causal evidence the mask needs. + + ``dispatch_generation=None`` disables the recency requirement, which is + necessary for event-inbox backends (herdr): they start no FIFO reader, so the + generation never advances from output and gating on it could never succeed -- + the worker would burn every resubmit and then be torn down. They need no gate + anyway, because ``get_status`` derives their status on demand at call time, so + there is no cached value to go stale. + """ + deadline = time.monotonic() + timeout + while True: + status = status_monitor.get_status(terminal_id) + if status in _DEFERRED_STARTED_STATUSES and ( + dispatch_generation is None + or status_monitor.output_generation(terminal_id) > dispatch_generation + ): + return True + if time.monotonic() >= deadline: + return False + await asyncio.sleep(polling_interval) + + async def _confirm_worker_started_or_resubmit( terminal_id: str, message: str, @@ -1905,18 +2023,19 @@ async def _confirm_worker_started_or_resubmit( sender_id: Optional[str], orchestration_type: Optional[OrchestrationType], provider=None, + dispatch_generation: Optional[int] = None, ) -> bool: """Confirm a deferred-init worker began processing; re-submit if not. - Returns True once the terminal reaches a started status, False if it is - still stuck at IDLE after all resubmit attempts. Blocking tmux/DB I/O runs - off the loop via to_thread so concurrent deferred inits aren't frozen. + Returns True once the terminal reaches a started status EARNED AFTER dispatch + (see ``_wait_for_post_dispatch_start``), False if it never does after all + resubmit attempts. Blocking tmux/DB I/O runs off the loop via to_thread so + concurrent deferred inits aren't frozen. """ - if await wait_until_status( + if await _wait_for_post_dispatch_start( terminal_id, - _DEFERRED_STARTED_STATUSES, + dispatch_generation, timeout=_DEFERRED_SUBMIT_CONFIRM_TIMEOUT, - polling_interval=0.5, ): return True @@ -1936,11 +2055,12 @@ async def _confirm_worker_started_or_resubmit( ) if already_started: return True - if await wait_until_status( + # Same recency requirement as the first wait: a resubmit that lands on a + # still-cached pre-dispatch COMPLETED must not read as success either. + if await _wait_for_post_dispatch_start( terminal_id, - _DEFERRED_STARTED_STATUSES, + dispatch_generation, timeout=_DEFERRED_SUBMIT_CONFIRM_TIMEOUT, - polling_interval=0.5, ): return True @@ -2002,18 +2122,68 @@ async def _run() -> None: effective_orchestration_type = orchestration_type or OrchestrationType.ASSIGN # send_input is blocking tmux I/O — off the loop so it can't # freeze the server for concurrent requests. - await asyncio.to_thread( - send_input, + dispatch_boundary = await asyncio.to_thread( + dispatch_input, terminal_id, initial_message, registry=registry, sender_id=caller_id, orchestration_type=effective_orchestration_type, ) + # The pending mark is deliberately NOT released here. + # + # Round-4 review (haofeif), P1. An earlier revision cleared it at + # this dispatch boundary, reasoning that a keystroke had been + # issued so a poller's evidence was now downstream of the send. It + # isn't. send_input only calls status_monitor.notify_input_sent(), + # which ARMS the next transition without changing the cached + # status, and no current provider enables + # assume_processing_on_dispatch. The status a poller reads right + # after this line is therefore still the pre-send IDLE, and stays + # that way until the agent's first output chunk is detected. The + # reviewer measured ~1.4s of it against a real worker; the + # regression test here reproduced 0.93s. Either way it is far + # longer than idle_stable_polls samples, so releasing at this line + # only moved the false-completion window from before the send to + # after it. + # + # _run's outer ``finally`` releases it instead, which is reached + # only after _confirm_worker_started_or_resubmit has observed a + # status in _DEFERRED_STARTED_STATUSES. That set includes + # COMPLETED, so holding the mark across the confirm window does + # NOT hide a genuine early completion (the concern that motivated + # releasing here): confirm returns as soon as the completion is + # visible, and the mark lifts with it. + # + # Cost of the change: when NO started status is ever observed the + # mark is now held for the confirm loop's whole budget + # (_DEFERRED_SUBMIT_CONFIRM_TIMEOUT x (1 + max resubmits) = ~32s) + # instead of being dropped at dispatch. That is inside `cao + # launch`'s 420s headless budget, and the worker is torn down at + # the end of it anyway, so UNKNOWN is the honest reading for a + # delivery we cannot confirm. # Delivery can be silently dropped (Enter swallowed / paste lost) # when the TUI isn't input-ready. Confirm the worker actually # started and re-submit if not; if it never starts, surface the # failure so the supervisor re-routes instead of waiting forever. + # The output generation at the dispatch boundary, captured INSIDE + # dispatch_input between arming the monitor and sending the first + # key. Confirmation requires the generation to exceed it, i.e. real + # output arrived after the send -- so a COMPLETED cached during + # provider startup can no longer read as this task starting. + # It is not sampled here, after the send returned: send_keys' + # submit delay is long enough for a fast worker to emit and + # complete, and a baseline taken afterwards would contain that + # output and reject the genuine completion (reviewer-reproduced). + # + # None for event-inbox backends (herdr): they run no FIFO reader, + # so the generation never advances from output and the requirement + # could never be met -- every resubmit would burn and the worker + # would be torn down. Their status is derived on demand instead, so + # there is no stale cached value for the gate to protect against. + dispatch_generation = ( + None if get_backend().supports_event_inbox() else dispatch_boundary + ) started = await _confirm_worker_started_or_resubmit( terminal_id, initial_message, @@ -2024,6 +2194,7 @@ async def _run() -> None: # must not silently drop back to the unguarded original type. effective_orchestration_type, provider=provider_instance, + dispatch_generation=dispatch_generation, ) if not started: logger.error( @@ -2086,13 +2257,70 @@ async def _run() -> None: registry, delete_worker=True, ) + finally: + # The single release point for every path, and on the happy path the + # first moment a completion poller's evidence is genuinely downstream + # of the send: reached only once _confirm_worker_started_or_resubmit + # has seen PROCESSING, COMPLETED or WAITING_USER_ANSWER. + # + # It also covers every abnormal exit, so a terminal can never be left + # permanently masked: initialize() raising, send_input raising + # TerminalInputBlockedError (the pane is parked on a prompt — the + # terminal is then honestly WAITING_USER_ANSWER and the operator has + # to see it), the worker never starting after all resubmits, or the + # loop being torn down. Idempotent, so double-clearing is harmless. + _clear_initial_delivery_pending(terminal_id) try: loop = asyncio.get_running_loop() except RuntimeError: logger.error(f"Deferred init for {terminal_id}: no running event loop; init skipped") return - task = loop.create_task(_run()) + # Mirrors _run's own ``if initial_message`` condition: defer_init is also used + # with no message at all (see the create_terminal call site), and there is + # nothing pending to mask on that path. + # + # Set here rather than inside _run so the mark is established by the time this + # function returns — the point at which the message has been accepted and the + # caller is free to start polling. That keeps the invariant independent of when + # the event loop first schedules _run instead of relying on it winning a race. + if initial_message: + _mark_initial_delivery_pending(terminal_id) + # Only _run's finally releases the mark, so a create_task that raises would + # leave it set with nothing left to clear it — create_terminal's own exception + # cleanup does not know about this mark. Cheap to close, so closed. + # + # Deliberately NOT claiming the obvious trigger: "the loop closed between the + # check above and this line" cannot happen. get_running_loop() only succeeds + # on the loop thread, so reaching here means we ARE the running loop, and + # loop.close() on a running loop raises "Cannot close a running event loop". + # What remains is the unglamorous set: _run() not being a coroutine + # (programming error), MemoryError, or a KeyboardInterrupt landing in the gap. + # + # Known and deliberately unguarded: after loop.stop(), create_task SUCCEEDS + # and the coroutine is never run, so the mark leaks with nothing raised for + # this to catch. That only arises while the loop is being torn down, and + # _pending_initial_delivery is module state that dies with the process, so + # there is nothing for it to leak into. Guarding it would mean a shutdown hook + # for state that cannot outlive shutdown. + # BaseException rather than Exception is deliberate, and matches the rollback + # guards at :415 and :796. KeyboardInterrupt is one of the two realistic + # triggers named above and is not an Exception subclass, so narrowing this + # would skip the case it exists for. The bare ``raise`` leaves shutdown + # semantics intact — this only cleans up on the way past. + coro = _run() + try: + task = loop.create_task(coro) + except BaseException: + # The coroutine object exists but was never handed to a task, so close it + # or the interpreter warns "coroutine was never awaited" when it is GC'd. + coro.close() + # Only clear what this call may have set: with no initial_message no mark + # was taken, and an unconditional discard would be reaching for state this + # call does not own. + if initial_message: + _clear_initial_delivery_pending(terminal_id) + raise _deferred_init_tasks.add(task) task.add_done_callback(_deferred_init_tasks.discard) @@ -2104,7 +2332,7 @@ def get_terminal(terminal_id: str) -> Dict: if not metadata: raise ValueError(f"Terminal '{terminal_id}' not found") - status = status_monitor.get_status(terminal_id).value + status = reported_status(terminal_id, status_monitor.get_status(terminal_id)).value return { "id": metadata["id"], @@ -2198,7 +2426,9 @@ def list_siblings( caller_id, prefix, caller_session=caller_session, cross_session=cross_session ) for sibling in siblings: - sibling["status"] = status_monitor.get_status(sibling["id"]).value + sibling["status"] = reported_status( + sibling["id"], status_monitor.get_status(sibling["id"]) + ).value return siblings @@ -2240,6 +2470,46 @@ def send_input( ) -> bool: """Send input to terminal via tmux paste buffer. + Thin wrapper over :func:`dispatch_input` that keeps the ``bool`` contract. + The contract is load-bearing, not ceremony: ``POST /terminals/{id}/input`` + returns this value on the wire as ``{"success": ...}``, so returning the + dispatch boundary here would change the API response -- and a first-ever + dispatch, whose boundary is 0, would read as ``"success": 0``. A caller that + must later prove the terminal produced output FOR THIS SEND (the deferred + initial-message path) calls ``dispatch_input`` directly and keeps the + boundary it returns. + """ + dispatch_input( + terminal_id, + message, + registry=registry, + sender_id=sender_id, + orchestration_type=orchestration_type, + frozen_memory=frozen_memory, + ) + return True + + +def dispatch_input( + terminal_id: str, + message: str, + registry: PluginRegistry | None = None, + sender_id: str | None = None, + orchestration_type: OrchestrationType | None = None, + frozen_memory: str | None = None, +) -> int: + """Send input to terminal via tmux paste buffer and return its dispatch boundary. + + The return value is ``status_monitor.output_generation()`` sampled at the + dispatch boundary: after the monitor has been armed and the rolling buffer + cleared, and before any key reaches the pane. Any output generation strictly + greater than it is output the terminal produced after this send -- the + evidence ``_wait_for_post_dispatch_start`` needs. It has to be captured here + rather than by the caller after this function returns, because ``send_keys`` + includes the provider's submit delay, during which a fast worker can already + emit and complete; sampled afterwards, those chunks would be inside the + baseline and a genuine completion would read as pre-dispatch. + Uses bracketed paste mode (-p) to bypass TUI hotkey handling. The number of Enter keys sent after pasting is determined by the provider's ``paste_enter_count`` property (e.g., some TUIs need 2 Enters because @@ -2334,6 +2604,22 @@ def send_input( # byte-identical completion from a retained completion screen. status_monitor.clear_rolling_buffer(terminal_id, provider) + # THE DISPATCH BOUNDARY. Sampled here -- armed, buffer cleared, no key + # sent yet -- and returned to the caller. Output that lands from this + # point on is output produced after this send; output that landed before + # it is already inside the value and can never count as evidence. This + # is the only place the sample is correct: one line later send_keys + # starts, and its submit delay is exactly the window in which a fast + # worker emits and completes (reviewer-reproduced on PR #566). + # + # Deliberately a separate lock acquisition AFTER the clear, not atomic + # with it. A chunk that lands between the two is pre-dispatch output (no + # key has been sent), and this ordering folds it INTO the baseline, so it + # cannot pass for evidence. Sampling under the clear's own lock would + # leave that chunk outside the baseline and count it -- the false + # confirmation this gate exists to prevent. + dispatch_boundary = status_monitor.output_generation(terminal_id) + # Mark the provider before send_keys rather than after it. send_keys # includes the provider-specific submit delay, during which a fast CLI # can already emit its first processing and completion frames. Those @@ -2382,7 +2668,7 @@ def send_input( traceparent=inject_traceparent(), ), ) - return True + return dispatch_boundary except Exception as e: logger.error(f"Failed to send input to terminal {terminal_id}: {e}") diff --git a/test/cli/commands/test_launch.py b/test/cli/commands/test_launch.py index 48c42dd0b..565836940 100644 --- a/test/cli/commands/test_launch.py +++ b/test/cli/commands/test_launch.py @@ -6,7 +6,14 @@ import pytest from click.testing import CliRunner -from cli_agent_orchestrator.cli.commands.launch import _parse_env_pairs, launch +from cli_agent_orchestrator.cli.commands.launch import ( + _HEADLESS_TASK_TIMEOUT, + _READINESS_WAIT_TIMEOUT, + _is_waiting_on_user, + _parse_env_pairs, + launch, +) +from cli_agent_orchestrator.models.terminal import TerminalStatus # ── Backend auto-detection (issue #308) ────────────────────────────── @@ -156,14 +163,20 @@ def test_launch_passes_explicit_kiro_engine(): def test_launch_headless_message_sends_to_terminal(): - """Test headless mode with message waits for IDLE then sends and polls for output.""" + """Headless+message hands MESSAGE to the server in the create body, then polls. + + Regression guard for caom-7it: the CLI used to create the session and then + issue a SEPARATE ``POST /terminals/{id}/input``. When provider init outlived + the client's read timeout on ``POST /sessions``, that second request never + happened and MESSAGE was silently dropped. There must be exactly ONE POST, + carrying ``initial_message``. + """ runner = CliRunner() with ( patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get, patch("cli_agent_orchestrator.cli.commands.launch.wait_until_terminal_status") as mock_wait, - patch("cli_agent_orchestrator.cli.commands.launch.time.sleep"), ): mock_post.return_value.json.return_value = { "session_name": "test-session", @@ -196,9 +209,15 @@ def test_launch_headless_message_sends_to_terminal(): assert result.exit_code == 0 assert "task done" in result.output - mock_wait.assert_called_once() - # Two POST calls: create session + send message - assert mock_post.call_count == 2 + # ONE POST: create session, with MESSAGE in the body. No follow-up + # /terminals/{id}/input — the server delivers it off the deferred-init + # path and re-submits if the TUI swallowed it. + assert mock_post.call_count == 1 + assert mock_post.call_args.kwargs["json"]["initial_message"] == "do something" + assert "/input" not in str(mock_post.call_args) + # No client-side readiness gate either: the create response returns + # before init finishes (status UNKNOWN), and poll_until_done handles it. + mock_wait.assert_not_called() def test_launch_invalid_provider(): @@ -360,6 +379,172 @@ def test_launch_non_headless_attaches_even_if_wait_times_out(): mock_get_backend.return_value.attach_session.assert_called_once_with("test-session") +# ── A screen that legitimately needs the operator is settled, not a stall ── +# +# A provider can finish init on a prompt only a human can answer (Codex's +# first-run login menu is the case that forced this). Such a screen never +# becomes IDLE on its own, so a readiness poll that accepts only +# {IDLE, COMPLETED} burned the full _READINESS_WAIT_TIMEOUT and then blamed +# init for a pane that was merely waiting — while telling the operator nothing +# about the sign-in they had to complete. + + +def test_readiness_poll_accepts_waiting_user_answer(): + """WAITING_USER_ANSWER is in the pre-attach target set. + + This is the assertion that matters: it pins the set itself, so narrowing it + back to {IDLE, COMPLETED} fails here rather than only showing up as a 120s + stall nothing in the suite waits around for. + """ + runner = CliRunner() + + with ( + patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, + patch("cli_agent_orchestrator.cli.commands.launch.get_backend") as mock_get_backend, + patch("cli_agent_orchestrator.cli.commands.launch.wait_until_terminal_status") as mock_wait, + patch("cli_agent_orchestrator.cli.commands.launch._is_waiting_on_user") as mock_waiting, + ): + mock_post.return_value.json.return_value = { + "session_name": "test-session", + "id": "test-terminal-id", + "name": "test-terminal", + } + mock_post.return_value.raise_for_status.return_value = None + mock_wait.return_value = True + mock_waiting.return_value = False + mock_get_backend.return_value.attach_session.return_value = None + + result = runner.invoke(launch, ["--agents", "test-agent", "--yolo"]) + + assert result.exit_code == 0 + target_set = mock_wait.call_args[0][1] + assert TerminalStatus.WAITING_USER_ANSWER in target_set, ( + "pre-attach poll must treat WAITING_USER_ANSWER as settled, or a " + f"first-run sign-in stalls for {_READINESS_WAIT_TIMEOUT}s; got {target_set}" + ) + # The pre-existing members must survive the widening. + assert {TerminalStatus.IDLE, TerminalStatus.COMPLETED} <= target_set + + +def test_launch_tells_the_operator_when_the_pane_awaits_an_answer(): + """On a settled WAITING_USER_ANSWER the operator is told to finish it, not warned. + + The "did not reach idle" warning would be actively wrong here: init did + finish, and the only thing outstanding is the human's answer. + """ + runner = CliRunner() + + with ( + patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, + patch("cli_agent_orchestrator.cli.commands.launch.get_backend") as mock_get_backend, + patch("cli_agent_orchestrator.cli.commands.launch.wait_until_terminal_status") as mock_wait, + patch("cli_agent_orchestrator.cli.commands.launch._is_waiting_on_user") as mock_waiting, + ): + mock_post.return_value.json.return_value = { + "session_name": "test-session", + "id": "test-terminal-id", + "name": "test-terminal", + } + mock_post.return_value.raise_for_status.return_value = None + mock_wait.return_value = True + mock_waiting.return_value = True + mock_get_backend.return_value.attach_session.return_value = None + + result = runner.invoke(launch, ["--agents", "test-agent", "--yolo"]) + + assert result.exit_code == 0 + assert "waiting for an answer in the pane" in result.output + assert "did not reach idle" not in result.output + mock_get_backend.return_value.attach_session.assert_called_once_with("test-session") + + +def test_launch_stays_quiet_when_the_terminal_settles_idle(): + """The new hint must not fire on the ordinary path.""" + runner = CliRunner() + + with ( + patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, + patch("cli_agent_orchestrator.cli.commands.launch.get_backend") as mock_get_backend, + patch("cli_agent_orchestrator.cli.commands.launch.wait_until_terminal_status") as mock_wait, + patch("cli_agent_orchestrator.cli.commands.launch._is_waiting_on_user") as mock_waiting, + ): + mock_post.return_value.json.return_value = { + "session_name": "test-session", + "id": "test-terminal-id", + "name": "test-terminal", + } + mock_post.return_value.raise_for_status.return_value = None + mock_wait.return_value = True + mock_waiting.return_value = False + mock_get_backend.return_value.attach_session.return_value = None + + result = runner.invoke(launch, ["--agents", "test-agent", "--yolo"]) + + assert result.exit_code == 0 + assert "waiting for an answer" not in result.output + assert "did not reach idle" not in result.output + + +def test_is_waiting_on_user_reads_the_live_status(): + """True only for WAITING_USER_ANSWER.""" + with patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get: + mock_get.return_value.status_code = 200 + mock_get.return_value.json.return_value = { + "status": TerminalStatus.WAITING_USER_ANSWER.value + } + assert _is_waiting_on_user("t1") is True + + mock_get.return_value.json.return_value = {"status": TerminalStatus.IDLE.value} + assert _is_waiting_on_user("t1") is False + + +def test_is_waiting_on_user_swallows_transport_errors(): + """A blip must not turn a successful launch into "Failed to connect to cao-server". + + This helper only decides which advisory line to print, and it runs inside the + command's ``RequestException`` handler — so letting one escape would report a + connection failure to a server the poll just finished talking to. + """ + import requests as _requests + + with patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get: + mock_get.side_effect = _requests.exceptions.ConnectionError("boom") + assert _is_waiting_on_user("t1") is False + + +@pytest.mark.parametrize("status_code", [201, 204, 400, 401, 404, 500, 503]) +def test_is_waiting_on_user_ignores_a_non_200_even_with_a_waiting_body(status_code): + """Only a 200 is believed — and the body must be a real dict to prove it. + + Leaving ``json()`` as an unconfigured MagicMock would make this pass on the + isinstance guard alone, so deleting the status check would not fail any test. + A concrete waiting body is what makes the status check the thing under test. + """ + with patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get: + mock_get.return_value.status_code = status_code + mock_get.return_value.json.return_value = { + "status": TerminalStatus.WAITING_USER_ANSWER.value + } + assert _is_waiting_on_user("t1") is False + + +@pytest.mark.parametrize("payload", [[], ["a"], None, "waiting_user_answer", 3]) +def test_is_waiting_on_user_survives_a_200_that_is_not_a_json_object(payload): + """A parseable-but-non-object body must not abort the launch. + + ``AttributeError`` from ``[].get`` is no kind of ``RequestException``, so + without the isinstance check it escapes this function, reaches the command's + generic handler, and exits 1 *after* ``POST /sessions`` has already created + the session — orphaning it in tmux, which is the exact outcome the local + except exists to prevent. Parametrized because each JSON scalar type reaches + ``.get`` by a different route. + """ + with patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get: + mock_get.return_value.status_code = 200 + mock_get.return_value.json.return_value = payload + assert _is_waiting_on_user("t1") is False + + def test_launch_workspace_confirmation_accepted(): """Test workspace confirmation is shown for claude_code provider and accepted.""" runner = CliRunner() @@ -536,12 +721,20 @@ def test_launch_builtin_profile_resolves_role_defaults(): assert "@cao-mcp-server" in params["allowed_tools"] -def test_launch_headless_message_conductor_not_ready(): - """Test headless+message raises when conductor does not become ready.""" +def test_launch_headless_message_does_not_gate_delivery_on_client_readiness(): + """Headless+message must not abort just because the CLI never saw IDLE. + + The old flow waited for IDLE client-side and raised "did not become ready" + before sending — a launch that hit that branch dropped MESSAGE outright even + though the terminal was alive and would have accepted it moments later. The + server now owns readiness, so a never-IDLE reading from the CLI's own poll + helper is irrelevant: MESSAGE is already in the create body. + """ runner = CliRunner() with ( patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, + patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get, patch("cli_agent_orchestrator.cli.commands.launch.wait_until_terminal_status") as mock_wait, ): mock_post.return_value.json.return_value = { @@ -552,6 +745,14 @@ def test_launch_headless_message_conductor_not_ready(): mock_post.return_value.raise_for_status.return_value = None mock_wait.return_value = False + poll_resp = MagicMock() + poll_resp.raise_for_status.return_value = None + poll_resp.json.return_value = {"status": "completed"} + output_resp = MagicMock() + output_resp.raise_for_status.return_value = None + output_resp.json.return_value = {"output": "task done"} + mock_get.side_effect = [poll_resp, output_resp] + result = runner.invoke( launch, [ @@ -563,8 +764,9 @@ def test_launch_headless_message_conductor_not_ready(): ], ) - assert result.exit_code != 0 - assert "did not become ready" in result.output + assert result.exit_code == 0 + assert "did not become ready" not in result.output + assert mock_post.call_args.kwargs["json"]["initial_message"] == "do something" def test_launch_headless_message_poll_error_status(): @@ -575,7 +777,6 @@ def test_launch_headless_message_poll_error_status(): patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get, patch("cli_agent_orchestrator.cli.commands.launch.wait_until_terminal_status") as mock_wait, - patch("cli_agent_orchestrator.cli.commands.launch.time.sleep"), ): mock_post.return_value.json.return_value = { "session_name": "test-session", @@ -613,7 +814,6 @@ def test_launch_headless_message_poll_processing_then_completed(): patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get, patch("cli_agent_orchestrator.cli.commands.launch.wait_until_terminal_status") as mock_wait, - patch("cli_agent_orchestrator.cli.commands.launch.time.sleep"), ): mock_post.return_value.json.return_value = { "session_name": "test-session", @@ -1023,3 +1223,221 @@ def test_minimax_code_requires_workspace_access_confirmation(): ) assert "mcode" in PROVIDERS_REQUIRING_WORKSPACE_ACCESS + + +# ── Server-side initial_message delivery (caom-7it) ──────────────────── + + +def _create_session_response(): + resp = MagicMock() + resp.raise_for_status.return_value = None + resp.json.return_value = { + "session_name": "test-session", + "id": "test-terminal-id", + "name": "test-terminal", + } + return resp + + +def test_launch_create_call_keeps_mcp_request_timeout(): + """``POST /sessions`` stays on the ordinary ``mcp_request_timeout``. + + Widening this instead of moving delivery server-side was the wrong fix: it + raises requests' CONNECT timeout as well (``timeout=`` is a single scalar, + so a dead server takes minutes to report), and it only helps callers that + happen to be this CLI. With ``initial_message`` in the body the server + returns before init even starts, so no widening is needed at all. + """ + runner = CliRunner() + + with ( + patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, + patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get, + patch("cli_agent_orchestrator.cli.commands.launch.get_server_settings") as mock_settings, + ): + mock_settings.return_value = {"mcp_request_timeout": 30} + mock_post.return_value = _create_session_response() + + poll_resp = MagicMock() + poll_resp.raise_for_status.return_value = None + poll_resp.json.return_value = {"status": "completed"} + output_resp = MagicMock() + output_resp.raise_for_status.return_value = None + output_resp.json.return_value = {"output": "task done"} + mock_get.side_effect = [poll_resp, output_resp] + + result = runner.invoke( + launch, ["--agents", "test-agent", "--headless", "--yolo", "do something"] + ) + + assert result.exit_code == 0 + assert mock_post.call_args.kwargs["timeout"] == 30 + # The /output read is an ordinary request too. + assert mock_get.call_args_list[-1].kwargs["timeout"] == 30 + + +def test_launch_headless_message_travels_in_body_not_query_string(): + """MESSAGE must not land in the URL: prompts are large (414 risk) and are + routinely captured verbatim in HTTP access logs and traces. Same reason + ``--env`` values were moved into the body in issue #248.""" + runner = CliRunner() + + with ( + patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, + patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get, + ): + mock_post.return_value = _create_session_response() + + poll_resp = MagicMock() + poll_resp.raise_for_status.return_value = None + poll_resp.json.return_value = {"status": "completed"} + output_resp = MagicMock() + output_resp.raise_for_status.return_value = None + output_resp.json.return_value = {"output": ""} + mock_get.side_effect = [poll_resp, output_resp] + + result = runner.invoke( + launch, ["--agents", "test-agent", "--headless", "--yolo", "secret prompt"] + ) + + assert result.exit_code == 0 + assert mock_post.call_args.kwargs["json"]["initial_message"] == "secret prompt" + assert "message" not in mock_post.call_args.kwargs["params"] + assert "secret prompt" not in str(mock_post.call_args.kwargs["params"]) + + +def test_launch_headless_message_shares_body_with_forwarded_env(): + """``initial_message`` and ``env_vars`` are both body fields on + ``CreateSessionBody`` — passing both must not drop either.""" + runner = CliRunner() + + with ( + patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, + patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get, + ): + mock_post.return_value = _create_session_response() + + poll_resp = MagicMock() + poll_resp.raise_for_status.return_value = None + poll_resp.json.return_value = {"status": "completed"} + output_resp = MagicMock() + output_resp.raise_for_status.return_value = None + output_resp.json.return_value = {"output": ""} + mock_get.side_effect = [poll_resp, output_resp] + + result = runner.invoke( + launch, + [ + "--agents", + "test-agent", + "--headless", + "--yolo", + "--env", + "AWS_PROFILE=dev", + "do something", + ], + ) + + assert result.exit_code == 0 + body = mock_post.call_args.kwargs["json"] + assert body["initial_message"] == "do something" + assert body["env_vars"] == {"AWS_PROFILE": "dev"} + + +def test_launch_async_returns_without_sending_input(): + """``--async`` reports and returns. It must not fall back to a client-side + send, and it must not wait for IDLE first — the server delivers MESSAGE + once init completes.""" + runner = CliRunner() + + with ( + patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, + patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get, + patch("cli_agent_orchestrator.cli.commands.launch.wait_until_terminal_status") as mock_wait, + patch("cli_agent_orchestrator.cli.commands.launch.poll_until_done") as mock_poll, + ): + mock_post.return_value = _create_session_response() + + result = runner.invoke( + launch, + ["--agents", "test-agent", "--headless", "--async", "--yolo", "do something"], + ) + + assert result.exit_code == 0 + assert "Running in background" in result.output + assert mock_post.call_count == 1 + assert mock_post.call_args.kwargs["json"]["initial_message"] == "do something" + mock_wait.assert_not_called() + mock_poll.assert_not_called() + mock_get.assert_not_called() + + +def test_launch_headless_no_message_omits_initial_message(): + """A headless launch with no MESSAGE must not put the terminal on the + deferred-init path — it has nothing to deliver, and deferring would report + UNKNOWN instead of IDLE to a caller that expects a ready terminal.""" + runner = CliRunner() + + with patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post: + mock_post.return_value = _create_session_response() + + result = runner.invoke(launch, ["--agents", "test-agent", "--headless", "--yolo"]) + + assert result.exit_code == 0 + assert "json" not in mock_post.call_args.kwargs + + +def test_launch_non_headless_does_not_defer_init(): + """The attach path never sent MESSAGE and must not start now. + + Deferring init here would leave the pre-attach readiness poll racing the + agent's first turn: the terminal reports UNKNOWN until init finishes and + then goes straight to PROCESSING, so the wait for IDLE would usually expire + and print a spurious "did not reach idle" warning before attaching. + """ + runner = CliRunner() + + with ( + patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, + patch("cli_agent_orchestrator.cli.commands.launch.get_backend"), + patch("cli_agent_orchestrator.cli.commands.launch.sync_backend_from_server"), + patch("cli_agent_orchestrator.cli.commands.launch.wait_until_terminal_status") as mock_wait, + ): + mock_post.return_value = _create_session_response() + mock_wait.return_value = True + + result = runner.invoke(launch, ["--agents", "test-agent", "--yolo", "do something"]) + + assert result.exit_code == 0 + assert "json" not in mock_post.call_args.kwargs + assert mock_post.call_count == 1 + + +def test_launch_headless_poll_budget_covers_server_side_init(): + """The wait budget must cover init as well as the task. + + Init used to happen in a separate readiness poll BEFORE the client-side + send; it now runs inside this wait, so charging the task's own 300s for it + would silently shrink how long an agent gets to finish. + """ + runner = CliRunner() + + with ( + patch("cli_agent_orchestrator.cli.commands.launch.requests.post") as mock_post, + patch("cli_agent_orchestrator.cli.commands.launch.requests.get") as mock_get, + patch("cli_agent_orchestrator.cli.commands.launch.poll_until_done") as mock_poll, + ): + mock_post.return_value = _create_session_response() + output_resp = MagicMock() + output_resp.raise_for_status.return_value = None + output_resp.json.return_value = {"output": "task done"} + mock_get.return_value = output_resp + + result = runner.invoke( + launch, ["--agents", "test-agent", "--headless", "--yolo", "do something"] + ) + + assert result.exit_code == 0 + mock_poll.assert_called_once_with( + "test-terminal-id", timeout=_READINESS_WAIT_TIMEOUT + _HEADLESS_TASK_TIMEOUT + ) diff --git a/test/services/test_deferred_submit_verification.py b/test/services/test_deferred_submit_verification.py index b8b784f23..5e910b867 100644 --- a/test/services/test_deferred_submit_verification.py +++ b/test/services/test_deferred_submit_verification.py @@ -218,7 +218,7 @@ def test_gate_off_default_keeps_deferred_init_behavior(self): class TestConfirmWorkerStartedOrResubmit: async def test_started_on_first_confirm_no_resubmit(self): with ( - patch.object(ts, "wait_until_status", new=AsyncMock(return_value=True)), + patch.object(ts, "_wait_for_post_dispatch_start", new=AsyncMock(return_value=True)), patch.object(ts, "send_special_key") as key, patch.object(ts, "send_input") as send, ): @@ -233,7 +233,9 @@ async def test_enter_resubmit_when_message_in_box(self): # First confirm fails, box shows our text (Enter swallowed) → bare Enter, # second confirm succeeds. with ( - patch.object(ts, "wait_until_status", new=AsyncMock(side_effect=[False, True])), + patch.object( + ts, "_wait_for_post_dispatch_start", new=AsyncMock(side_effect=[False, True]) + ), patch.object(ts, "_message_visible_in_box", return_value=True), patch.object(ts, "send_special_key") as key, patch.object(ts, "send_input") as send, @@ -248,7 +250,9 @@ async def test_enter_resubmit_when_message_in_box(self): async def test_full_redeliver_when_box_empty(self): # First confirm fails, box empty (paste dropped) → re-deliver full msg. with ( - patch.object(ts, "wait_until_status", new=AsyncMock(side_effect=[False, True])), + patch.object( + ts, "_wait_for_post_dispatch_start", new=AsyncMock(side_effect=[False, True]) + ), patch.object(ts, "_message_visible_in_box", return_value=False), patch.object(ts, "send_special_key") as key, patch.object(ts, "send_input") as send, @@ -265,7 +269,7 @@ async def test_full_redeliver_when_box_empty(self): async def test_returns_false_when_worker_never_starts(self): # Every confirm fails through all resubmit attempts. with ( - patch.object(ts, "wait_until_status", new=AsyncMock(return_value=False)), + patch.object(ts, "_wait_for_post_dispatch_start", new=AsyncMock(return_value=False)), patch.object(ts, "_message_visible_in_box", return_value=True), patch.object(ts, "send_special_key"), patch.object(ts, "send_input"), @@ -280,7 +284,7 @@ async def test_direct_probe_short_circuits_when_worker_started(self): # returns True without calling send_input or send_special_key. provider = MagicMock(supports_direct_status_probe=True) with ( - patch.object(ts, "wait_until_status", new=AsyncMock(return_value=False)), + patch.object(ts, "_wait_for_post_dispatch_start", new=AsyncMock(return_value=False)), patch.object(ts, "_worker_is_started_direct", return_value=True), patch.object(ts, "send_special_key") as key, patch.object(ts, "send_input") as send, @@ -301,7 +305,9 @@ async def test_direct_probe_falls_through_when_worker_not_started(self): # Direct probe returns False → continues to existing resubmit logic. provider = MagicMock(supports_direct_status_probe=True) with ( - patch.object(ts, "wait_until_status", new=AsyncMock(side_effect=[False, True])), + patch.object( + ts, "_wait_for_post_dispatch_start", new=AsyncMock(side_effect=[False, True]) + ), patch.object(ts, "_worker_is_started_direct", return_value=False), patch.object(ts, "_message_visible_in_box", return_value=True), patch.object(ts, "send_special_key") as key, @@ -324,7 +330,9 @@ async def test_direct_probe_skipped_when_provider_not_opted_in(self): # invoked; falls through to existing resubmit logic. provider = MagicMock(supports_direct_status_probe=False) with ( - patch.object(ts, "wait_until_status", new=AsyncMock(side_effect=[False, True])), + patch.object( + ts, "_wait_for_post_dispatch_start", new=AsyncMock(side_effect=[False, True]) + ), patch.object(ts, "_worker_is_started_direct") as probe, patch.object(ts, "_message_visible_in_box", return_value=True), patch.object(ts, "send_special_key"), @@ -344,7 +352,9 @@ async def test_direct_probe_skipped_when_provider_not_opted_in(self): async def test_provider_none_skips_direct_probe(self): # The existing None-provider path still works unchanged. with ( - patch.object(ts, "wait_until_status", new=AsyncMock(side_effect=[False, True])), + patch.object( + ts, "_wait_for_post_dispatch_start", new=AsyncMock(side_effect=[False, True]) + ), patch.object(ts, "_worker_is_started_direct") as probe, patch.object(ts, "_message_visible_in_box", return_value=True), patch.object(ts, "send_special_key"), diff --git a/test/services/test_session_service.py b/test/services/test_session_service.py index 6a6b1ad8f..70f381852 100644 --- a/test/services/test_session_service.py +++ b/test/services/test_session_service.py @@ -861,6 +861,56 @@ async def test_same_name_relaunch_purges_stale_terminal_metadata( delete_session(second.session_name) +class TestGetSessionMasksPendingDelivery: + """#566: the masking WIRING here, not just ``reported_status`` in isolation. + + ``TestReportedStatusMasking`` pins the function; nothing pinned that + ``get_session`` actually calls it. Deleting the call left 162 tests passing + (gutosantos82's mutation check), so the whole invariant -- a terminal whose + accepted initial message is undelivered must never read completable anywhere a + client can see it -- rested on an unguarded call site. ``GET /sessions/{name}`` + is one of the three outward surfaces, reached by ``examples/fleet/panel`` and + by ``ops_mcp_server.get_session_info``. + """ + + @patch("cli_agent_orchestrator.services.session_service.list_terminals_by_session") + @patch("cli_agent_orchestrator.services.session_service.get_backend") + def test_pending_delivery_is_masked_in_the_session_listing( + self, mock_get_backend, mock_list_terminals + ): + from cli_agent_orchestrator.models.terminal import TerminalStatus + from cli_agent_orchestrator.services import terminal_service + + mock_get_backend.return_value.session_exists.return_value = True + mock_get_backend.return_value.list_sessions.return_value = [ + {"id": "cao-test", "name": "Test Session"} + ] + mock_list_terminals.return_value = [ + {"id": "pend1234", "session": "cao-test"}, + {"id": "free5678", "session": "cao-test"}, + ] + + fake_monitor = MagicMock() + fake_monitor.get_status.return_value = TerminalStatus.IDLE + + # Only pend1234 has an undelivered initial message. + with ( + patch.object(terminal_service, "_pending_initial_delivery", {"pend1234"}), + patch("cli_agent_orchestrator.services.status_monitor.status_monitor", fake_monitor), + ): + result = get_session("cao-test") + + by_id = {term["id"]: term["status"] for term in result["terminals"]} + assert by_id["pend1234"] == TerminalStatus.UNKNOWN.value, ( + "get_session reported a completable status for a terminal whose initial " + "message has not been dispatched -- the mask is not wired into this surface, " + "so a client polling GET /sessions/{name} can still conclude 'done'" + ) + assert ( + by_id["free5678"] == TerminalStatus.IDLE.value + ), "masking leaked to a terminal with no pending delivery; it is per-terminal" + + class TestGetSession: """Tests for get_session function.""" diff --git a/test/services/test_status_monitor.py b/test/services/test_status_monitor.py index c9071045d..77ddbdd7a 100644 --- a/test/services/test_status_monitor.py +++ b/test/services/test_status_monitor.py @@ -1409,3 +1409,66 @@ def test_erased_composer_is_not_work_evidence_and_the_real_spinner_still_lands(s sm._bursting["t1"] = True sm._schedule_screen_detection("t1", provider) assert sm._last_status["t1"] == TerminalStatus.PROCESSING + + +class TestOutputGenerationIsOutputOnly: + """PR #566: the delivery-confirmation gate needs a counter only OUTPUT can move. + + ``_capture_generation`` must advance on ``notify_input_sent`` too (a new turn + invalidates in-flight capture verdicts), so it cannot serve: a redelivery's own + arm would satisfy "output arrived since dispatch" on a still-cached COMPLETED. + ``output_generation()`` therefore exposes a separate counter that only + ``_process_chunk`` bumps. + """ + + def test_unknown_terminal_reads_zero(self): + assert StatusMonitor().output_generation("never-seen") == 0 + + def test_arming_a_turn_does_not_advance_it(self): + sm = StatusMonitor() + sm.notify_input_sent("t1") + sm.notify_input_sent("t1", assume_processing=False) + assert sm.output_generation("t1") == 0, ( + "notify_input_sent moved the output generation: a redelivery's own arm would " + "now pass for post-dispatch output and confirm a stale COMPLETED" + ) + # The capture generation, by contrast, MUST have moved -- that is its job. + assert sm._capture_generation["t1"] == 2 + + @patch("cli_agent_orchestrator.services.status_monitor.get_server_settings") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_each_real_chunk_advances_it_by_one(self, mock_pm, mock_settings): + mock_settings.return_value = {"state_buffer_max": 32768} + provider = MagicMock() + provider.supports_screen_detection = False + provider.get_status.return_value = TerminalStatus.PROCESSING + mock_pm.get_provider.return_value = provider + + sm = StatusMonitor() + sm.notify_input_sent("t1") # dispatch: arms, does not count + boundary = sm.output_generation("t1") + for chunk in ("thinking ", "done.\n", "> "): + sm._process_chunk("t1", chunk) + + assert sm.output_generation("t1") == boundary + 3 + # Interleaving another arm (a redelivery) still adds nothing. + sm.notify_input_sent("t1") + assert sm.output_generation("t1") == boundary + 3 + + @patch("cli_agent_orchestrator.services.status_monitor.get_server_settings") + @patch("cli_agent_orchestrator.services.status_monitor.provider_manager") + def test_forgetting_a_terminal_resets_it(self, mock_pm, mock_settings): + mock_settings.return_value = {"state_buffer_max": 32768} + provider = MagicMock() + provider.supports_screen_detection = False + provider.get_status.return_value = TerminalStatus.PROCESSING + mock_pm.get_provider.return_value = provider + + sm = StatusMonitor() + sm._process_chunk("t1", "x") + assert sm.output_generation("t1") == 1 + sm.reset_buffer("t1") + assert sm.output_generation("t1") == 0 + sm._process_chunk("t1", "y") + sm.clear_terminal("t1") + assert sm.output_generation("t1") == 0 diff --git a/test/services/test_terminal_service_full.py b/test/services/test_terminal_service_full.py index c3959855c..b6596323b 100644 --- a/test/services/test_terminal_service_full.py +++ b/test/services/test_terminal_service_full.py @@ -3146,3 +3146,825 @@ async def test_no_orchestration_type_still_blocked_post_sessions_bypass( mock_tmux.send_keys.assert_not_called() mock_notify.assert_called_once() assert mock_notify.call_args.kwargs["delete_worker"] is False + + +class TestDeferredDeliveryNotCompletableBeforeDispatch: + """PR #566 review (haofeif), BLOCKING P1: the synchronous headless + ``cao launch`` path must not be able to return before the initial message + has actually been dispatched. + + ``launch``'s ``poll_until_done`` starts the moment the deferred + ``POST /sessions`` returns, so its ``observed_working`` evidence is not + causally downstream of the send. Provider startup on its own reports + WAITING_USER_ANSWER (kiro-cli's consent dialog is the common case), which + flips ``observed_working=True``; once ``initialize()`` finishes the pane + reads IDLE while ``_schedule_deferred_init`` still has to resolve + shell_baseline and metadata and then run ``inject_memory_context`` inside + ``send_input`` (a curated-memory lookup widens that to seconds). Three IDLE + samples inside that window and the CLI returns, reads empty output, and + exits 0 with the task never delivered. + + The invariant under test is the ordering one, not a wall-clock one: while an + initial delivery is still pending, the terminal must not report a + *completable* status, so no number of samples taken before dispatch can + satisfy the idle gate. + """ + + @pytest.mark.asyncio + @patch("cli_agent_orchestrator.services.terminal_service._notify_caller_of_deferred_failure") + @patch("cli_agent_orchestrator.services.terminal_service.update_terminal_shell_command") + @patch( + "cli_agent_orchestrator.services.terminal_service._confirm_worker_started_or_resubmit", + new_callable=AsyncMock, + ) + @patch("cli_agent_orchestrator.services.terminal_service.dispatch_input") + @patch("cli_agent_orchestrator.services.terminal_service.status_monitor") + @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata") + async def test_poll_until_done_cannot_return_before_initial_send_is_issued( + self, + mock_meta, + mock_status_monitor, + mock_send_input, + mock_confirm_started, + mock_update_shell, + mock_notify, + ): + """The reviewer's reproduction, driven through the real reporting path. + + Real ``_schedule_deferred_init``, real ``terminal_service.get_terminal`` + (the status every client polls), and the real ``poll_until_done`` the CLI + calls, run concurrently against one valid startup sequence: + WAITING_USER_ANSWER while initializing, then IDLE, then PROCESSING once + the message is dispatched. + + Against the unfixed head this fails: ``poll_until_done`` returns while + ``dispatched`` is still False. The polling interval is compressed only to + keep the test fast — what makes the bug reproduce is that the + pre-dispatch IDLE window outlasts ``idle_stable_polls`` samples, which is + exactly the real-world condition. + """ + import asyncio + import threading + import time as _time + + from cli_agent_orchestrator.services.terminal_service import ( + _deferred_init_tasks, + _schedule_deferred_init, + get_terminal, + ) + from cli_agent_orchestrator.utils.terminal import poll_until_done + + TERMINAL_ID = "abcd1234" + INIT_SECONDS = 0.10 + PRE_DISPATCH_SECONDS = 0.60 + POST_DISPATCH_WORK_SECONDS = 0.10 + POLL_INTERVAL = 0.02 + + mock_meta.return_value = { + "id": TERMINAL_ID, + "tmux_window": "developer-abcd", + "tmux_session": "cao-session", + "provider": "kiro_cli", + "agent_profile": "developer", + "caller_id": None, + "allowed_tools": None, + "engine": None, + "group": None, + "metadata": None, + "last_active": datetime.now(), + } + + init_done = threading.Event() + dispatched = threading.Event() + dispatched_at = {} + + def fake_get_status(terminal_id): + # One valid startup sequence, then one ordinary turn: + # WAITING_USER_ANSWER (consent dialog, real startup activity but + # NOT evidence the assigned task was picked up) + # -> IDLE (init finished; nothing dispatched yet) + # -> PROCESSING (the agent is working on the delivered task) + # -> IDLE (turn finished, the only legitimate exit) + if dispatched.is_set(): + if _time.monotonic() - dispatched_at["t"] < POST_DISPATCH_WORK_SECONDS: + return TerminalStatus.PROCESSING + return TerminalStatus.IDLE + if init_done.is_set(): + return TerminalStatus.IDLE + return TerminalStatus.WAITING_USER_ANSWER + + mock_status_monitor.get_status.side_effect = fake_get_status + + def fake_send_input(terminal_id, message, **kwargs): + # Stands in for dispatch_input's pre-dispatch work — most of it + # inject_memory_context — before any keystroke reaches the pane. + _time.sleep(PRE_DISPATCH_SECONDS) + dispatched_at["t"] = _time.monotonic() + dispatched.set() + return 0 # the dispatch boundary dispatch_input returns + + mock_send_input.side_effect = fake_send_input + mock_confirm_started.return_value = True + + async def fake_initialize(): + await asyncio.sleep(INIT_SECONDS) + init_done.set() + return True + + provider_instance = AsyncMock() + provider_instance.initialize.side_effect = fake_initialize + provider_instance.shell_baseline = None + + # Route the CLI's status poll through the real server-side reporting + # path instead of a scripted list, so this test pins the behaviour of + # get_terminal rather than a restatement of the fixture. + def fake_requests_get(url, **kwargs): + resp = MagicMock() + resp.raise_for_status.return_value = None + resp.json.return_value = {"status": get_terminal(TERMINAL_ID)["status"]} + return resp + + before_tasks = set(_deferred_init_tasks) + _schedule_deferred_init( + provider_instance, TERMINAL_ID, "do the task", OrchestrationType.ASSIGN, None + ) + (task,) = set(_deferred_init_tasks) - before_tasks + + dispatched_when_poll_returned = {} + + def run_poll(): + with patch("cli_agent_orchestrator.utils.terminal.requests.get", fake_requests_get): + poll_until_done( + TERMINAL_ID, + timeout=30.0, + polling_interval=POLL_INTERVAL, + ) + dispatched_when_poll_returned["value"] = dispatched.is_set() + + await asyncio.gather(task, asyncio.to_thread(run_poll)) + + mock_notify.assert_not_called() + assert dispatched.is_set(), "fixture bug: the initial send never ran" + assert dispatched_when_poll_returned["value"] is True, ( + "poll_until_done returned before the initial message was dispatched — " + "the synchronous `cao launch` would print empty output and exit 0 " + "with the task never delivered" + ) + + @pytest.mark.asyncio + @patch("cli_agent_orchestrator.services.terminal_service._notify_caller_of_deferred_failure") + @patch("cli_agent_orchestrator.services.terminal_service.update_terminal_shell_command") + @patch("cli_agent_orchestrator.services.terminal_service.redeliver_dropped_message") + @patch("cli_agent_orchestrator.services.terminal_service.dispatch_input") + @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata") + async def test_poll_cannot_complete_on_the_stale_post_dispatch_idle( + self, mock_meta, mock_send_input, mock_redeliver, mock_update_shell, mock_notify + ): + """Round-4 review (haofeif), P1: issuing the send is not evidence of it. + + An earlier revision released the mask when ``send_input()`` returned, on + the reasoning that a keystroke had been dispatched so a poller's evidence + was now causal. It isn't. ``send_input`` only calls + ``notify_input_sent()``, which arms the next transition without touching + the cached status, and no provider enables + ``assume_processing_on_dispatch`` — so the reading right after dispatch is + still the pre-send IDLE, for as long as it takes the agent's first output + chunk to be detected (~1.4s measured against the real scheduler). + + This runs the REAL ``_confirm_worker_started_or_resubmit`` against a status + fixture with that post-dispatch IDLE lag, and fails on the head that + released at the dispatch boundary: ``poll_until_done`` returned 0.93s + before the first PROCESSING signal ever appeared. + """ + import asyncio + import threading + import time as _time + + from cli_agent_orchestrator.services.terminal_service import ( + _deferred_init_tasks, + _schedule_deferred_init, + get_terminal, + ) + from cli_agent_orchestrator.utils.terminal import poll_until_done + + TERMINAL_ID = "abcd1234" + INIT_SECONDS = 0.10 + PRE_DISPATCH_SECONDS = 0.20 + POST_DISPATCH_IDLE_LAG = 0.60 + POST_DISPATCH_WORK_SECONDS = 3.00 + POLL_INTERVAL = 0.02 + + mock_meta.return_value = { + "id": TERMINAL_ID, + "tmux_window": "developer-abcd", + "tmux_session": "cao-session", + "provider": "kiro_cli", + "agent_profile": "developer", + "caller_id": None, + "allowed_tools": None, + "engine": None, + "group": None, + "metadata": None, + "last_active": datetime.now(), + } + + init_done = threading.Event() + dispatched = threading.Event() + dispatched_at = {} + first_processing_at = {} + + def fake_get_status(terminal_id): + if dispatched.is_set(): + elapsed = _time.monotonic() - dispatched_at["t"] + if elapsed < POST_DISPATCH_IDLE_LAG: + return TerminalStatus.IDLE # the stale pre-send reading + if elapsed < POST_DISPATCH_IDLE_LAG + POST_DISPATCH_WORK_SECONDS: + first_processing_at.setdefault("t", _time.monotonic()) + return TerminalStatus.PROCESSING + return TerminalStatus.IDLE + if init_done.is_set(): + return TerminalStatus.IDLE + return TerminalStatus.WAITING_USER_ANSWER + + fake_monitor = MagicMock() + fake_monitor.get_status.side_effect = fake_get_status + + def fake_send_input(terminal_id, message, **kwargs): + _time.sleep(PRE_DISPATCH_SECONDS) + dispatched_at["t"] = _time.monotonic() + dispatched.set() + return 0 # the dispatch boundary dispatch_input returns + + mock_send_input.side_effect = fake_send_input + mock_redeliver.return_value = False + + async def fake_initialize(): + await asyncio.sleep(INIT_SECONDS) + init_done.set() + return True + + provider_instance = AsyncMock() + provider_instance.initialize.side_effect = fake_initialize + provider_instance.shell_baseline = None + + def fake_requests_get(url, **kwargs): + resp = MagicMock() + resp.raise_for_status.return_value = None + resp.json.return_value = {"status": get_terminal(TERMINAL_ID)["status"]} + return resp + + poll_returned_at = {} + + def run_poll(): + with patch("cli_agent_orchestrator.utils.terminal.requests.get", fake_requests_get): + poll_until_done(TERMINAL_ID, timeout=30.0, polling_interval=POLL_INTERVAL) + poll_returned_at["t"] = _time.monotonic() + + # wait_until_status imports the singleton at call time, so patching + # terminal_service.status_monitor alone would not reach the confirm loop. + with ( + patch("cli_agent_orchestrator.services.status_monitor.status_monitor", fake_monitor), + patch("cli_agent_orchestrator.services.terminal_service.status_monitor", fake_monitor), + ): + before_tasks = set(_deferred_init_tasks) + _schedule_deferred_init( + provider_instance, TERMINAL_ID, "do the task", OrchestrationType.ASSIGN, None + ) + (task,) = set(_deferred_init_tasks) - before_tasks + await asyncio.gather(task, asyncio.to_thread(run_poll)) + + assert dispatched.is_set(), "fixture bug: the initial send never ran" + assert ( + "t" in first_processing_at + ), "fixture bug: the agent never reached PROCESSING, so the test proves nothing" + assert poll_returned_at["t"] > first_processing_at["t"], ( + "poll_until_done returned before the first post-dispatch PROCESSING signal — " + "it completed on the stale pre-send IDLE that survives send_input()'s return, " + "so `cao launch` exits 0 with empty output while the agent is about to start" + ) + + @pytest.mark.asyncio + @patch("cli_agent_orchestrator.services.terminal_service._notify_caller_of_deferred_failure") + @patch("cli_agent_orchestrator.services.terminal_service.update_terminal_shell_command") + @patch("cli_agent_orchestrator.services.terminal_service.redeliver_dropped_message") + @patch("cli_agent_orchestrator.services.terminal_service.send_input") + @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata") + async def test_a_genuine_early_completion_is_not_hidden_by_the_mask( + self, mock_meta, mock_send_input, mock_redeliver, mock_update_shell, mock_notify + ): + """The mirror-image regression, pinned with the REAL confirm loop. + + Replaces an earlier test that asserted the mask had to be released at the + dispatch boundary. That test's premise was a fixture artifact: it mocked + ``_confirm_worker_started_or_resubmit`` as a flat 1.5s sleep that returned + True regardless of status, so holding the mark across it necessarily + stranded the poller. The real function's first action is + ``wait_until_status(_DEFERRED_STARTED_STATUSES, polling_interval=0.5)``, + and that set contains COMPLETED — so it returns as soon as a completion is + visible, and the mark lifts with it. + + Here a turn finishes without the pipeline ever publishing PROCESSING + (IDLE -> COMPLETED directly), which is the case that release point existed + to protect. The poll must still return promptly rather than sit out the + confirm window. + """ + import asyncio + import threading + import time as _time + + from cli_agent_orchestrator.services.terminal_service import ( + _deferred_init_tasks, + _schedule_deferred_init, + get_terminal, + ) + from cli_agent_orchestrator.utils.terminal import poll_until_done + + TERMINAL_ID = "beef5678" + INIT_SECONDS = 0.10 + COMPLETION_LAG = 0.30 + POLL_INTERVAL = 0.02 + + mock_meta.return_value = { + "id": TERMINAL_ID, + "tmux_window": "developer-beef", + "tmux_session": "cao-session", + "provider": "kiro_cli", + "agent_profile": "developer", + "caller_id": None, + "allowed_tools": None, + "engine": None, + "group": None, + "metadata": None, + "last_active": datetime.now(), + } + + init_done = threading.Event() + dispatched = threading.Event() + dispatched_at = {} + saw_processing = {"value": False} + + def fake_get_status(terminal_id): + if dispatched.is_set(): + if _time.monotonic() - dispatched_at["t"] < COMPLETION_LAG: + return TerminalStatus.IDLE + return TerminalStatus.COMPLETED # never PROCESSING + if init_done.is_set(): + return TerminalStatus.IDLE + return TerminalStatus.WAITING_USER_ANSWER + + fake_monitor = MagicMock() + fake_monitor.get_status.side_effect = fake_get_status + + def fake_send_input(terminal_id, message, **kwargs): + dispatched_at["t"] = _time.monotonic() + dispatched.set() + return True + + mock_send_input.side_effect = fake_send_input + mock_redeliver.return_value = False + + async def fake_initialize(): + await asyncio.sleep(INIT_SECONDS) + init_done.set() + return True + + provider_instance = AsyncMock() + provider_instance.initialize.side_effect = fake_initialize + provider_instance.shell_baseline = None + + def fake_requests_get(url, **kwargs): + resp = MagicMock() + resp.raise_for_status.return_value = None + status = get_terminal(TERMINAL_ID)["status"] + if status == TerminalStatus.PROCESSING.value: + saw_processing["value"] = True + resp.json.return_value = {"status": status} + return resp + + elapsed = {} + + def run_poll(): + started = _time.monotonic() + with patch("cli_agent_orchestrator.utils.terminal.requests.get", fake_requests_get): + poll_until_done(TERMINAL_ID, timeout=30.0, polling_interval=POLL_INTERVAL) + elapsed["t"] = _time.monotonic() - started + + with ( + patch("cli_agent_orchestrator.services.status_monitor.status_monitor", fake_monitor), + patch("cli_agent_orchestrator.services.terminal_service.status_monitor", fake_monitor), + ): + before_tasks = set(_deferred_init_tasks) + _schedule_deferred_init( + provider_instance, TERMINAL_ID, "do the task", OrchestrationType.ASSIGN, None + ) + (task,) = set(_deferred_init_tasks) - before_tasks + await asyncio.gather(task, asyncio.to_thread(run_poll)) + + assert ( + saw_processing["value"] is False + ), "fixture bug: this must exercise the no-PROCESSING path" + assert elapsed["t"] < 3.0, ( + f"poll_until_done took {elapsed['t']:.2f}s for a turn that completed " + f"{COMPLETION_LAG:.2f}s after dispatch — the mask is stranding a genuine " + "early completion" + ) + + +class TestListSiblingsMasksPendingDelivery: + """#566: the masking WIRING in list_siblings, not ``reported_status`` alone. + + The existing sibling tests use empty sibling lists, so deleting the + ``reported_status`` call left 130 tests passing (gutosantos82's mutation + check). ``list_siblings`` is one of the three outward surfaces, and it is the + one a supervisor reads to decide whether a sibling is finished -- an unmasked + IDLE there invites a caller to treat an undelivered worker as done. + """ + + @patch("cli_agent_orchestrator.services.terminal_service.list_siblings_by_group_prefix") + @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata") + def test_pending_delivery_is_masked_for_siblings(self, mock_meta, mock_by_prefix): + from cli_agent_orchestrator.services import terminal_service + from cli_agent_orchestrator.services.terminal_service import list_siblings + + mock_meta.return_value = {"group": ["root", "child"], "tmux_session": "cao-session"} + mock_by_prefix.return_value = [ + {"id": "pend1234", "group": ["root", "child"], "metadata": None}, + {"id": "free5678", "group": ["root", "child"], "metadata": None}, + ] + + fake_monitor = MagicMock() + fake_monitor.get_status.return_value = TerminalStatus.COMPLETED + + with ( + patch.object(terminal_service, "_pending_initial_delivery", {"pend1234"}), + patch.object(terminal_service, "status_monitor", fake_monitor), + ): + siblings = list_siblings("caller99") + + by_id = {s["id"]: s["status"] for s in siblings} + assert by_id["pend1234"] == TerminalStatus.UNKNOWN.value, ( + "list_siblings reported COMPLETED for a sibling whose initial message has " + "not been dispatched -- a supervisor would treat an undelivered worker as " + "finished" + ) + assert ( + by_id["free5678"] == TerminalStatus.COMPLETED.value + ), "masking leaked to a sibling with no pending delivery; it is per-terminal" + + +class TestConfirmationRequiresPostDispatchEvidence: + """Round-6 review (haofeif), P1: a cached pre-dispatch COMPLETED is not evidence. + + Provider startup output can legitimately parse as COMPLETED, which then latches + (``_STICKY_READY_STATUSES``), and ``send_input`` only ARMS the next transition + without touching the cached value. Confirmation therefore used to succeed + instantly on a status earned BEFORE the send -- measured at 0.008s on an exact + head -- releasing the pending-delivery mask before the task emitted anything. + + Every prior test in this area seeded IDLE and produced COMPLETED afterwards, so + none of them could see this. These seed the completion FIRST. + """ + + @pytest.mark.asyncio + async def test_pre_dispatch_completed_does_not_confirm_the_send(self): + """The reviewer's case: COMPLETED cached before dispatch, no new output.""" + from cli_agent_orchestrator.services.terminal_service import ( + _wait_for_post_dispatch_start, + ) + + fake_monitor = MagicMock() + # Status is COMPLETED throughout, and the generation NEVER advances -- + # exactly the shape of a completion left over from provider startup. + fake_monitor.get_status.return_value = TerminalStatus.COMPLETED + fake_monitor.output_generation.return_value = 7 + + with patch("cli_agent_orchestrator.services.terminal_service.status_monitor", fake_monitor): + confirmed = await _wait_for_post_dispatch_start( + "abcd1234", dispatch_generation=7, timeout=0.25, polling_interval=0.05 + ) + + assert confirmed is False, ( + "confirmation accepted a COMPLETED cached before dispatch: the generation " + "never advanced, so no output arrived for this task, yet the send read as " + "started and the delivery mask would clear on the previous turn's result" + ) + + @pytest.mark.asyncio + async def test_post_dispatch_output_does_confirm(self): + """The discriminating half: same status, but the generation advanced.""" + from cli_agent_orchestrator.services.terminal_service import ( + _wait_for_post_dispatch_start, + ) + + fake_monitor = MagicMock() + fake_monitor.get_status.return_value = TerminalStatus.COMPLETED + fake_monitor.output_generation.return_value = 8 # real output landed + + with patch("cli_agent_orchestrator.services.terminal_service.status_monitor", fake_monitor): + confirmed = await _wait_for_post_dispatch_start( + "abcd1234", dispatch_generation=7, timeout=0.25, polling_interval=0.05 + ) + + assert confirmed is True, ( + "a COMPLETED with an advanced generation IS this turn's completion and " + "must confirm -- otherwise a genuine fast turn burns every resubmit and " + "the worker is torn down" + ) + + @pytest.mark.asyncio + async def test_event_inbox_backends_are_not_gated_on_generation(self): + """herdr runs no FIFO reader, so the generation never advances from output. + + Gating it would make confirmation unsatisfiable and tear down working + workers. ``dispatch_generation=None`` opts out, and their status is derived + on demand so there is no stale cached value to defend against. + """ + from cli_agent_orchestrator.services.terminal_service import ( + _wait_for_post_dispatch_start, + ) + + fake_monitor = MagicMock() + fake_monitor.get_status.return_value = TerminalStatus.COMPLETED + fake_monitor.output_generation.return_value = 0 # never advances for herdr + + with patch("cli_agent_orchestrator.services.terminal_service.status_monitor", fake_monitor): + confirmed = await _wait_for_post_dispatch_start( + "abcd1234", dispatch_generation=None, timeout=0.25, polling_interval=0.05 + ) + + assert ( + confirmed is True + ), "an event-inbox backend was gated on a generation it can never advance" + + +class TestPendingMarkNeverLeaks: + """Only ``_run``'s finally releases the mark, so a ``create_task`` that raises + would leave it set with nothing left to clear it. + + Scope, stated honestly because the tempting version of this claim is false: + the trigger is NOT a closed loop. ``get_running_loop()`` only succeeds on the + loop thread, so reaching ``create_task`` means we are the running loop, and + closing a running loop raises. The reachable triggers are ``_run()`` not being + a coroutine, MemoryError, or a KeyboardInterrupt in the gap. This pins the + invariant rather than any one of them. + + Not covered, by choice: after ``loop.stop()`` ``create_task`` succeeds and the + coroutine never runs, leaking the mark with nothing raised. That is + loop-teardown only, and this is module state that dies with the process. + """ + + @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata") + def test_create_task_raising_does_not_leak_the_mark(self, mock_meta): + import asyncio + + from cli_agent_orchestrator.services.terminal_service import ( + _schedule_deferred_init, + initial_delivery_pending, + ) + + TERMINAL_ID = "1eak1eak" + mock_meta.return_value = None + + captured = {} + + class RefusingLoop: + """Stands in for any create_task failure, not for a closed loop. + + Deliberately does NOT close the coroutine: a real loop doesn't either, + and the production guard is what has to close it. Leaving that to the + fake would hide an un-awaited-coroutine warning in real use. The + coroutine is captured so the assertion below can check production + closed it, rather than inferring that from a GC-timed warning. + """ + + def create_task(self, coro): + captured["coro"] = coro + raise RuntimeError("create_task refused") + + provider_instance = AsyncMock() + provider_instance.shell_baseline = None + + assert not initial_delivery_pending(TERMINAL_ID) + with patch.object(asyncio, "get_running_loop", return_value=RefusingLoop()): + with pytest.raises(RuntimeError, match="create_task refused"): + _schedule_deferred_init( + provider_instance, + TERMINAL_ID, + "do the task", + OrchestrationType.ASSIGN, + None, + ) + # A closed coroutine has cr_frame is None; a never-started, never-closed one + # does not. Deterministic, unlike asserting on the un-awaited RuntimeWarning + # that pytest only surfaces as a non-failing PytestUnraisableExceptionWarning + # at GC -- which is why deleting production's coro.close() used to fail + # nothing (gutosantos82). + assert captured["coro"].cr_frame is None, ( + "the orphaned coroutine was left open: create_task raised after _run() was " + "constructed, so nothing will ever await it and the interpreter warns " + "'coroutine was never awaited' when it is collected" + ) + assert not initial_delivery_pending(TERMINAL_ID), ( + "the pending mark leaked: create_task raised after the mark was set, so " + "_run never ran and its finally never released it, leaving this " + "terminal_id reporting UNKNOWN with nothing able to clear it" + ) + + +class TestReportedStatusMasking: + """The masking matrix ``reported_status`` documents, pinned. + + Which statuses are masked is the whole safety argument: mask too little and + the #566 P1 is back; mask too much and a pane parked on a real prompt, or a + dead provider, becomes invisible to the operator who has to act on it. + """ + + def _pending(self, terminal_id): + from cli_agent_orchestrator.services import terminal_service + + return patch.object(terminal_service, "_pending_initial_delivery", {terminal_id}) + + @pytest.mark.parametrize( + "raw", + [TerminalStatus.IDLE, TerminalStatus.COMPLETED], + ) + def test_completable_statuses_are_masked_while_delivery_pending(self, raw): + from cli_agent_orchestrator.services.terminal_service import reported_status + + with self._pending("aaaa1111"): + assert reported_status("aaaa1111", raw) is TerminalStatus.UNKNOWN + + @pytest.mark.parametrize( + "raw", + [ + TerminalStatus.WAITING_USER_ANSWER, + TerminalStatus.PROCESSING, + TerminalStatus.ERROR, + TerminalStatus.UNKNOWN, + ], + ) + def test_actionable_statuses_are_never_masked(self, raw): + """WAITING_USER_ANSWER in particular: masking it would hide the one state + an operator must see to unblock the pane, and send_input's own guard + converts it into a TerminalInputBlockedError that releases the mark.""" + from cli_agent_orchestrator.services.terminal_service import reported_status + + with self._pending("aaaa1111"): + assert reported_status("aaaa1111", raw) is raw + + @pytest.mark.parametrize( + "raw", + [ + TerminalStatus.IDLE, + TerminalStatus.COMPLETED, + TerminalStatus.WAITING_USER_ANSWER, + TerminalStatus.PROCESSING, + TerminalStatus.ERROR, + TerminalStatus.UNKNOWN, + ], + ) + def test_nothing_is_masked_once_no_delivery_is_pending(self, raw): + """The mask is scoped to the pending window only — the ordinary steady + state of every terminal must be reported verbatim.""" + from cli_agent_orchestrator.services.terminal_service import reported_status + + assert reported_status("aaaa1111", raw) is raw + + def test_mask_is_per_terminal_not_global(self): + """A pending delivery on one terminal must not mask a sibling's IDLE.""" + from cli_agent_orchestrator.services.terminal_service import reported_status + + with self._pending("aaaa1111"): + assert reported_status("aaaa1111", TerminalStatus.IDLE) is TerminalStatus.UNKNOWN + assert reported_status("bbbb2222", TerminalStatus.IDLE) is TerminalStatus.IDLE + + +class TestDispatchBoundaryIsSampledBeforeKeys: + """Round-8 review (haofeif), P1: the boundary must be captured inside the send. + + ``send_keys`` includes the provider's submit delay, and a fast worker can emit + and complete inside it. A baseline sampled by the CALLER after ``send_input`` + returned therefore contained that output, so ``_wait_for_post_dispatch_start`` + rejected a genuine completion, resubmitted, and could delete a worker that had + done the task. ``dispatch_input`` now returns the generation sampled between + arming the monitor and sending the first key. + """ + + def _metadata(self): + return {"tmux_session": "cao-session", "tmux_window": "developer-abcd"} + + @patch("cli_agent_orchestrator.services.terminal_service.MemoryService") + @patch("cli_agent_orchestrator.services.terminal_service.status_monitor") + @patch("cli_agent_orchestrator.services.terminal_service.update_last_active") + @patch("cli_agent_orchestrator.services.terminal_service.provider_manager") + @patch("cli_agent_orchestrator.backends.registry._backend") + @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata") + def test_output_emitted_during_send_keys_is_not_inside_the_boundary( + self, + mock_get_metadata, + mock_tmux, + mock_pm, + mock_update, + mock_status_monitor, + mock_memory_service, + ): + from cli_agent_orchestrator.services.terminal_service import dispatch_input + + mock_memory_service.return_value.get_curated_memory_context.return_value = "" + mock_get_metadata.return_value = self._metadata() + mock_provider = mock_pm.get_provider.return_value + mock_provider.paste_enter_count = 1 + mock_provider.paste_submit_delay = 0.0 + mock_provider.assume_processing_on_dispatch = False + mock_provider.blocks_orchestrated_input_while_waiting_user_answer = False + mock_status_monitor.get_status.return_value = TerminalStatus.IDLE + + generation = {"n": 3} + mock_status_monitor.output_generation.side_effect = lambda terminal_id: generation["n"] + + def worker_emits_and_completes_during_the_submit_delay(*args, **kwargs): + # The FIFO reader lands the worker's first (and last) chunk while + # send_keys is still blocking on the submit delay. + generation["n"] += 1 + + mock_tmux.send_keys.side_effect = worker_emits_and_completes_during_the_submit_delay + + boundary = dispatch_input("test1234", "a task the worker finishes instantly") + + assert boundary == 3, ( + "the boundary must be the generation BEFORE any key was sent; a value of 4 " + "means it was sampled after send_keys and contains this task's own output" + ) + # The old caller-side sample, taken after the send returned, would have + # been 4 -- equal to the current generation -- and would have rejected + # the completion. + assert mock_status_monitor.output_generation("test1234") == 4 + # Ordering: armed and cleared BEFORE the sample, keys AFTER it. + mock_status_monitor.notify_input_sent.assert_called_once() + mock_status_monitor.clear_rolling_buffer.assert_called_once() + order = [name for name, _args, _kwargs in mock_status_monitor.mock_calls] + assert order.index("clear_rolling_buffer") < order.index("output_generation") + + @pytest.mark.asyncio + async def test_fast_completion_during_send_confirms_with_the_inner_boundary(self): + """The reviewer's schedule, end to end through the confirmation gate.""" + from cli_agent_orchestrator.services.terminal_service import ( + _wait_for_post_dispatch_start, + ) + + fake_monitor = MagicMock() + # After the send: the worker emitted once (3 -> 4) and completed; nothing + # further will ever arrive. + fake_monitor.get_status.return_value = TerminalStatus.COMPLETED + fake_monitor.output_generation.return_value = 4 + inner_boundary = 3 # what dispatch_input returned + post_send_sample = 4 # what the caller used to sample after send_input + + with patch("cli_agent_orchestrator.services.terminal_service.status_monitor", fake_monitor): + with_inner = await _wait_for_post_dispatch_start( + "abcd1234", dispatch_generation=inner_boundary, timeout=0.25, polling_interval=0.05 + ) + with_post_send = await _wait_for_post_dispatch_start( + "abcd1234", + dispatch_generation=post_send_sample, + timeout=0.25, + polling_interval=0.05, + ) + + assert with_inner is True, ( + "a worker that completed during the submit delay is a genuine completion and " + "must confirm; rejecting it burns every resubmit and deletes a worker that did " + "the task" + ) + assert with_post_send is False, ( + "documents the defect: a baseline sampled after the send contains the task's " + "own output and can never be exceeded by a worker that emits nothing more" + ) + + @patch("cli_agent_orchestrator.services.terminal_service.MemoryService") + @patch("cli_agent_orchestrator.services.terminal_service.status_monitor") + @patch("cli_agent_orchestrator.services.terminal_service.update_last_active") + @patch("cli_agent_orchestrator.services.terminal_service.provider_manager") + @patch("cli_agent_orchestrator.backends.registry._backend") + @patch("cli_agent_orchestrator.services.terminal_service.get_terminal_metadata") + def test_send_input_keeps_its_bool_contract( + self, + mock_get_metadata, + mock_tmux, + mock_pm, + mock_update, + mock_status_monitor, + mock_memory_service, + ): + """POST /terminals/{id}/input echoes this value as {"success": ...}.""" + mock_memory_service.return_value.get_curated_memory_context.return_value = "" + mock_get_metadata.return_value = self._metadata() + # A first-ever dispatch has boundary 0; the wrapper must still say True. + mock_status_monitor.output_generation.return_value = 0 + mock_status_monitor.get_status.return_value = TerminalStatus.IDLE + mock_pm.get_provider.return_value.blocks_orchestrated_input_while_waiting_user_answer = ( + False + ) + + assert send_input("test1234", "hello") is True + mock_tmux.send_keys.assert_called_once()