fix(launch): deliver the initial message via POST /sessions instead of a dropped second request

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

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,52 +389,77 @@ 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
             # entry causes the CLI to default to tmux. See issue #308.
             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", "")

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:

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)