fix: preserve worker holds and resume on operator replies

docs/pipeline-setup.md

@@ -148,6 +148,21 @@ results from work that already started; the lock is never held while waiting for
 The supervisor yields after a turn leaves detached work running, instead of immediately
 requesting another agent turn to start the same work.
 
+Pending `wander ask` questions also hold new agent and stage launches. Questions share the
+admission lock with launches; already-running work keeps its completion records. Answering
+wakes the supervisor on its next pass, including a reply that arrives as the prior turn ends.
+A stalled session is discarded so the next turn can recover from the journal.
+
+To watch just one run while answering questions, use:
+
+```sh
+uv run --locked python -m orchestrator.supervisor --run-id RUN_ID --watch
+```
+
+Without `--watch`, `--run-id` still makes one pass and exits. `--once` and `--watch` are
+mutually exclusive. The worker's `codex --version` and `codex exec --help` must start normally;
+an API doctor check does not test that executable or the worker's model sign-in.
+
 An executor that disappears without recording completion remains unresolved. Inspect its log,
 provider receipt and retained outputs before reconciling it; restarting the supervisor, removing
 the lock file, or inventing a new attempt must not resubmit uncertain work. Older unresolved starts

docs/pipeline-validation.md

@@ -92,6 +92,22 @@ normalized only in API responses when their event history proves the hold came f
 ambiguous or missing history remains held. This verifies the operator reply path, not autonomous
 agent execution or fresh inference. Evidence: `.context/evidence/review-answer-handoff/`.
 
+A subsequent local check used the real Codex CLI worker with the retained bottle source.
+The agent verified the source/config hashes and asked a question; a reply through the admin
+woke the named supervisor watch and resumed the same saved session. The agent recorded the
+reply token in 9.1 seconds with the quiet retry timer set to one hour, rechecked both hashes,
+and asked for final review. The diagnostic was then paused and its supervisor stopped.
+No stage attempt, provider command, or quality acceptance occurred; the paid ledger and
+retained hashes were unchanged. This proves the live question/reply/resume path, not fresh
+reconstruction. The check used an isolated official CLI 0.163.0 with the existing ChatGPT
+sign-in; inherited unrelated MCP sign-in and duplicate-hook warnings did not prevent either
+turn. The broken system CLI was then updated to that version and its exec/resume startup
+checks passed. Evidence: `.context/evidence/worker-reply-resume/`.
+
+Regression tests cover questions arriving before stage/agent launch, blocked watchdog resumes,
+retained transcript/session evidence, and answers arriving between a turn's final journal read
+and the supervisor's next pass. Quiet turns do not repeatedly wake on their own status notes.
+
 A real admin-to-viewer check reused the retained six-second scene in an isolated paused run.
 The preview loaded all 75 samples, sought to frames 0, 36 and 74, and advanced playback with the
 source video. Delivered world and compact-motion hashes matched the retained files. No requests

orchestrator/agent/harness.py

@@ -15,6 +15,7 @@
 import subprocess
 import time
 from collections.abc import Callable
+from contextlib import AbstractContextManager, nullcontext
 from dataclasses import dataclass, field
 from pathlib import Path
 
@@ -244,6 +245,7 @@ def run(
         instructions: str,
         session: str | None = None,
         heartbeat: Callable[[], None] | None = None,
+        spawn_guard: Callable[[], AbstractContextManager] | None = None,
     ) -> HarnessOutcome:
         """Review one attempt, nudging the agent back to work if it goes silent.
 
@@ -263,7 +265,6 @@ def run(
         if packet is not None:
             atomic_json(packet_path, packet.model_dump(mode="json"))
         instructions_path.write_text(instructions)
-        decision_path.unlink(missing_ok=True)
         environment = {key: os.environ[key] for key in self.policy.passthrough if key in os.environ}
         environment.update(self.policy.extra_environment)
 
@@ -284,6 +285,7 @@ def run(
                     mode,
                     budget,
                     heartbeat,
+                    spawn_guard,
                 )
             except subprocess.TimeoutExpired:
                 result = AgentResult(
@@ -358,43 +360,50 @@ def _spawn(
         mode: str,
         budget: float,
         heartbeat: Callable[[], None] | None = None,
+        spawn_guard: Callable[[], AbstractContextManager] | None = None,
     ) -> int:
         """Run one harness process, raising :class:`Stalled` if it stops writing.
 
         The prompt is a file rather than a pipe we write into, so a long brief cannot deadlock
         against a child that has not started reading yet. Silence is measured by the size of the
         transcript, which the child appends to as it works.
         """
-        with prompt_path.open("rb") as prompt, transcript.open(mode) as output:
-            process = subprocess.Popen(
-                argv,
-                cwd=workspace,
-                env=environment,
-                stdin=prompt,
-                stdout=output,
-                stderr=subprocess.STDOUT,
-                start_new_session=True,
-            )
-            started = time.monotonic()
-            quiet_since = started
-            written = transcript.stat().st_size
-            while True:
-                try:
-                    return process.wait(timeout=POLL_SECONDS)
-                except subprocess.TimeoutExpired:
-                    pass
-                if heartbeat:
-                    heartbeat()
-                now = time.monotonic()
-                size = transcript.stat().st_size
-                if size != written:
-                    written, quiet_since = size, now
-                if now - started > budget:
-                    _halt(process)
-                    raise subprocess.TimeoutExpired(argv, budget)
-                if self.policy.idle_seconds and now - quiet_since > self.policy.idle_seconds:
-                    _halt(process)
-                    raise Stalled()
+        with prompt_path.open("rb") as prompt:
+            # A run may acquire a hold while the prompt is prepared, including before
+            # a watchdog resume. Serialize admission, artifact reset and spawn only.
+            with spawn_guard() if spawn_guard is not None else nullcontext():
+                if mode == "wb":
+                    (workspace / DECISION_FILE).unlink(missing_ok=True)
+                with transcript.open(mode) as output:
+                    process = subprocess.Popen(
+                        argv,
+                        cwd=workspace,
+                        env=environment,
+                        stdin=prompt,
+                        stdout=output,
+                        stderr=subprocess.STDOUT,
+                        start_new_session=True,
+                    )
+        started = time.monotonic()
+        quiet_since = started
+        written = transcript.stat().st_size
+        while True:
+            try:
+                return process.wait(timeout=POLL_SECONDS)
+            except subprocess.TimeoutExpired:
+                pass
+            if heartbeat:
+                heartbeat()
+            now = time.monotonic()
+            size = transcript.stat().st_size
+            if size != written:
+                written, quiet_since = size, now
+            if now - started > budget:
+                _halt(process)
+                raise subprocess.TimeoutExpired(argv, budget)
+            if self.policy.idle_seconds and now - quiet_since > self.policy.idle_seconds:
+                _halt(process)
+                raise Stalled()
 
 
 def _halt(process: subprocess.Popen) -> None:

orchestrator/cli.py

@@ -183,14 +183,21 @@ def source_problem(context: RunContext) -> str | None:
     return None
 
 
+def stage_hold_reason(journal: Journal) -> str | None:
+    if status := journal.control_status():
+        return f"Run is {status}; an operator must resume it before more work can start."
+    if journal.unanswered():
+        return "Run is waiting for an operator answer; answer the pending question before more work can start."
+    return None
+
+
 def refuse_controlled_step(
     context: RunContext, journal: Journal, step: str, attempt: str | None = None
 ) -> bool:
     """Reject new work, closing only a pending local attempt that has not launched yet."""
-    status = journal.control_status()
-    if status is None:
+    problem = stage_hold_reason(journal)
+    if problem is None:
         return False
-    problem = f"Run is {status}; an operator must resume it before more work can start."
     if attempt:
         for entry in reversed(journal.entries()):
             if entry.data.get("attempt") != attempt:
@@ -349,8 +356,8 @@ def command_step(context: RunContext, args: argparse.Namespace) -> int:
             # Operator controls take this same short lock. Once a pause/cancel has
             # returned, no new stage can slip between the final check and launch.
             with journal.admission_lock():
-                if control := journal.control_status():
-                    status, error = "blocked", f"Run is {control}; no subprocess was launched."
+                if problem := stage_hold_reason(journal):
+                    status, error = "blocked", problem
                     return 2
                 process = subprocess.Popen(
                     command,
@@ -698,7 +705,12 @@ def command_note(context: RunContext, args: argparse.Namespace) -> int:
 
 
 def command_ask(context: RunContext, args: argparse.Namespace) -> int:
-    context.journal().append("question", question=args.question, step=args.step)
+    if not args.question.strip():
+        print("A question must contain non-whitespace text.", file=sys.stderr)
+        return 2
+    journal = context.journal()
+    with journal.admission_lock():
+        journal.append("question", question=args.question, step=args.step)
     print(
         "asked, and the run is waiting. An operator answers with `wander answer`; "
         "stop working on this and finish your turn.",

orchestrator/session.py

@@ -13,14 +13,15 @@
 from __future__ import annotations
 
 import json
+from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path
 
 import dataclasses
 import os
 import sys
 
-from orchestrator.agent.harness import HarnessAgent
+from orchestrator.agent.harness import HarnessAgent, TRANSCRIPT_FILE, read_session
 from orchestrator.journal import Journal, RunProjection
 from orchestrator.steps import STEPS, people_steps, planned_steps
 
@@ -163,12 +164,19 @@ class SessionOutcome:
     error: str | None = None
     # Steps started and not yet finished when the turn ended.
     running: int = 0
+    # Journal length observed before deriving the outcome. Later events must wake the run.
+    observed_entries: int = 0
 
     @property
     def wants_another_turn(self) -> bool:
         return self.finished is None and self.waiting is None and self.error is None
 
 
+class AdmissionStopped(Exception):
+    def __init__(self, outcome: SessionOutcome):
+        self.outcome = outcome
+
+
 class RunSession:
     """The agent working on one run, one turn at a time."""
 
@@ -244,16 +252,24 @@ def read_session(self) -> str | None:
 
     def admission_outcome(self, journal: Journal) -> SessionOutcome | None:
         """Check durable controls again on every turn, including a session already in memory."""
+        observed_entries = len(journal.entries())
         control = journal.control_status()
         finished = "canceled" if control == "canceled" else journal.finished()
         if finished:
-            return SessionOutcome(finished=finished, waiting=None, steps_run=0)
+            return SessionOutcome(
+                finished=finished, waiting=None, steps_run=0, observed_entries=observed_entries
+            )
         if control == "paused":
             return SessionOutcome(
                 finished=None,
                 waiting=None,
                 steps_run=0,
                 error="Run is paused; an operator must resume it before another turn.",
+                observed_entries=observed_entries,
+            )
+        if waiting := journal.unanswered():
+            return SessionOutcome(
+                finished=None, waiting=waiting, steps_run=0, observed_entries=observed_entries
             )
         return None
 
@@ -276,19 +292,41 @@ def turn(self) -> SessionOutcome:
         install_skill(self.run_dir)
         if stopped := self.admission_outcome(journal):
             return stopped
-        outcome = self.harness.run(
-            None,
-            self.run_dir,
-            brief,
-            session=self.read_session(),
-        )
-        if outcome.session:
-            (self.run_dir / SESSION_FILE).write_text(outcome.session + "\n")
-        elif outcome.result.status == "stalled":
+
+        spawned = False
+
+        @contextmanager
+        def spawn_guard():
+            nonlocal spawned
+            with journal.admission_lock():
+                if stopped := self.admission_outcome(journal):
+                    raise AdmissionStopped(stopped)
+                yield
+                spawned = True
+
+        try:
+            outcome = self.harness.run(
+                None,
+                self.run_dir,
+                brief,
+                session=self.read_session(),
+                spawn_guard=spawn_guard,
+            )
+        except AdmissionStopped as stopped:
+            # A question can stop a watchdog nudge after the first process has run.
+            # Preserve that session for the answer, without reviving an older transcript.
+            if spawned and (session := read_session(self.run_dir / TRANSCRIPT_FILE)):
+                (self.run_dir / SESSION_FILE).write_text(session + "\n")
+            return stopped.outcome
+        if outcome.result.status == "stalled":
             # A session that stopped answering is not the one to ask again; the journal is
             # this run's memory, not the conversation.
             (self.run_dir / SESSION_FILE).unlink(missing_ok=True)
+        elif outcome.session:
+            (self.run_dir / SESSION_FILE).write_text(outcome.session + "\n")
         after = self.journal()
+        entries = after.entries()
+        observed_entries = len(entries)
         error = None
         if outcome.result.status not in {"completed", "stalled"}:
             error = outcome.result.error
@@ -297,9 +335,10 @@ def turn(self) -> SessionOutcome:
         return SessionOutcome(
             finished=after.finished(),
             waiting=after.unanswered(),
-            steps_run=len([e for e in after.entries() if e.kind == "step.started"]) - before,
+            steps_run=len([e for e in entries if e.kind == "step.started"]) - before,
             error=error,
             running=len(after.active_attempts()),
+            observed_entries=observed_entries,
         )
 
     def work(self, turns: int = 12) -> SessionOutcome:
@@ -309,7 +348,12 @@ def work(self, turns: int = 12) -> SessionOutcome:
         lost the thread or has nothing left to do, and either way another identical turn is
         unlikely to differ.
         """
-        outcome = SessionOutcome(finished=None, waiting=None, steps_run=0)
+        outcome = SessionOutcome(
+            finished=None,
+            waiting=None,
+            steps_run=0,
+            observed_entries=len(self.journal().entries()),
+        )
         for _ in range(turns):
             outcome = self.turn()
             if not outcome.wants_another_turn:
@@ -319,9 +363,13 @@ def work(self, turns: int = 12) -> SessionOutcome:
                 # change instead of immediately asking it to launch the same step.
                 return outcome
             if outcome.steps_run == 0:
-                self.journal().append(
+                journal = self.journal()
+                note = journal.append(
                     "note",
                     text="the agent took a turn without running a step or deciding anything",
                 )
+                entries = journal.entries()
+                if len(entries) == outcome.observed_entries + 1 and entries[-1] == note:
+                    outcome = dataclasses.replace(outcome, observed_entries=len(entries))
                 return outcome
         return outcome