From 272b36567c41c4eed1d12f3b454f2d5c55435488 Mon Sep 17 00:00:00 2001 From: Austin Jian Date: Fri, 16 Oct 2026 16:19:38 -0400 Subject: [PATCH 1/2] fix: preserve worker holds and resume on operator replies --- docs/pipeline-setup.md | 15 +++ docs/pipeline-validation.md | 16 +++ orchestrator/agent/harness.py | 71 ++++++----- orchestrator/cli.py | 24 +++- orchestrator/session.py | 76 +++++++++-- orchestrator/supervisor.py | 53 +++++--- scripts/tests/test_admin_live_commands.py | 2 +- scripts/tests/test_orchestrator_session.py | 142 ++++++++++++++++++++- scripts/tests/test_run_controls.py | 93 +++++++++++++- scripts/tests/test_step_admission.py | 19 ++- scripts/tests/test_supervisor_wakeup.py | 85 ++++++++++++ 11 files changed, 519 insertions(+), 77 deletions(-) create mode 100644 scripts/tests/test_supervisor_wakeup.py diff --git a/docs/pipeline-setup.md b/docs/pipeline-setup.md index f90287b..4d4ac34 100644 --- a/docs/pipeline-setup.md +++ b/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 diff --git a/docs/pipeline-validation.md b/docs/pipeline-validation.md index 213e6d3..23d64c1 100644 --- a/docs/pipeline-validation.md +++ b/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 diff --git a/orchestrator/agent/harness.py b/orchestrator/agent/harness.py index d777e7e..f8021a8 100644 --- a/orchestrator/agent/harness.py +++ b/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,6 +360,7 @@ 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. @@ -365,36 +368,42 @@ def _spawn( 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: diff --git a/orchestrator/cli.py b/orchestrator/cli.py index 74d0bf5..45feda2 100644 --- a/orchestrator/cli.py +++ b/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.", diff --git a/orchestrator/session.py b/orchestrator/session.py index 464b3c8..9185122 100644 --- a/orchestrator/session.py +++ b/orchestrator/session.py @@ -13,6 +13,7 @@ from __future__ import annotations import json +from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path @@ -20,7 +21,7 @@ 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 diff --git a/orchestrator/supervisor.py b/orchestrator/supervisor.py index cd771dd..174b94e 100644 --- a/orchestrator/supervisor.py +++ b/orchestrator/supervisor.py @@ -14,6 +14,7 @@ import argparse import os import time +from dataclasses import dataclass from pathlib import Path from orchestrator.agent.harness import HarnessAgent, HarnessPolicy @@ -79,13 +80,21 @@ def due(run_dir: Path, pushed: dict[str, tuple[float, int]], push_after: float) return happened != seen or (time.monotonic() - when) >= push_after -def work_on(run_dir: Path, repository, policy: HarnessPolicy, turns: int) -> str: +@dataclass(frozen=True) +class WorkOutcome: + message: str + observed_entries: int + + +def work_on(run_dir: Path, repository, policy: HarnessPolicy, turns: int) -> WorkOutcome: import json described = json.loads((run_dir / RUN_FILE).read_text()) run_id = described["runId"] - if status := Journal(run_dir).control_status(): - return f"{run_id}: {status} by the operator" + journal = Journal(run_dir) + observed_entries = len(journal.entries()) + if status := journal.control_status(): + return WorkOutcome(f"{run_id}: {status} by the operator", observed_entries) if repository is not None: # A run being picked up again is a run being worked on, whatever it last said. try: @@ -101,12 +110,14 @@ def work_on(run_dir: Path, repository, policy: HarnessPolicy, turns: int) -> str ) outcome = session.work(turns=turns) if outcome.finished: - return f"{run_id}: finished {outcome.finished}" - if outcome.waiting: - return f"{run_id}: waiting on an operator — {outcome.waiting}" - if outcome.error: - return f"{run_id}: session ended without deciding — {outcome.error}" - return f"{run_id}: still going after {turns} turns" + message = f"{run_id}: finished {outcome.finished}" + elif outcome.waiting: + message = f"{run_id}: waiting on an operator — {outcome.waiting}" + elif outcome.error: + message = f"{run_id}: session ended without deciding — {outcome.error}" + else: + message = f"{run_id}: still going after {turns} turns" + return WorkOutcome(message, outcome.observed_entries) def main(argv: list[str] | None = None) -> int: @@ -116,11 +127,19 @@ def main(argv: list[str] | None = None) -> int: default=str(resolve_runs_root()), help="directory holding one folder per run", ) - parser.add_argument("--run-id", help="work on one run and stop") - parser.add_argument("--turns", type=int, default=12, help="turns per run before moving on") parser.add_argument( + "--run-id", help="select one run; stop after one pass unless --watch is set" + ) + parser.add_argument("--turns", type=int, default=12, help="turns per run before moving on") + mode = parser.add_mutually_exclusive_group() + mode.add_argument( "--once", action="store_true", help="make one pass over the runs instead of watching" ) + mode.add_argument( + "--watch", + action="store_true", + help="keep watching, including when --run-id selects one run", + ) parser.add_argument("--interval", type=float, default=15.0, help="seconds between passes") parser.add_argument( "--push-after", @@ -129,6 +148,7 @@ def main(argv: list[str] | None = None) -> int: help="seconds before a run that went nowhere is given a fresh agent", ) args = parser.parse_args(argv) + single_pass = args.once or (args.run_id and not args.watch) root = resolve_runs_root(args.runs) root.mkdir(parents=True, exist_ok=True) @@ -157,11 +177,14 @@ def main(argv: list[str] | None = None) -> int: # started it, so a run whose journal has not moved is a run whose agent is # waiting on something -- handing it a second agent is how a 1600-credit world # gets submitted twice. - if not (args.once or args.run_id) and not due(run_dir, pushed, args.push_after): + if not single_pass and not due(run_dir, pushed, args.push_after): continue - print(work_on(run_dir, repository, policy, args.turns), flush=True) - pushed[str(run_dir)] = (time.monotonic(), len(Journal(run_dir).entries())) - if args.once or args.run_id: + outcome = work_on(run_dir, repository, policy, args.turns) + print(outcome.message, flush=True) + # Keep the cursor the session actually observed. A reply can arrive between + # its final read and this assignment; reading again here would swallow it. + pushed[str(run_dir)] = (time.monotonic(), outcome.observed_entries) + if single_pass: return 0 time.sleep(args.interval) diff --git a/scripts/tests/test_admin_live_commands.py b/scripts/tests/test_admin_live_commands.py index cd7ddcd..f71eb89 100644 --- a/scripts/tests/test_admin_live_commands.py +++ b/scripts/tests/test_admin_live_commands.py @@ -125,7 +125,7 @@ def test_stale_supervisor_selection_cannot_reopen_operator_stopped_run( session = Mock() monkeypatch.setattr(supervisor, "RunSession", session) result = supervisor.work_on(selected[0], repository, Mock(), turns=1) - assert "operator" in result + assert "operator" in result.message repository.reopen_run.assert_not_called() session.assert_not_called() summary = api.get("/api/pipeline/runs/operator-test").json() diff --git a/scripts/tests/test_orchestrator_session.py b/scripts/tests/test_orchestrator_session.py index 0a623e6..eed72e8 100644 --- a/scripts/tests/test_orchestrator_session.py +++ b/scripts/tests/test_orchestrator_session.py @@ -20,11 +20,12 @@ ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) -from orchestrator.agent.harness import HarnessAgent, HarnessPolicy +from orchestrator.agent.contracts import AgentResult +from orchestrator.agent.harness import HarnessAgent, HarnessOutcome, HarnessPolicy from orchestrator.journal import Journal from orchestrator.session import RunSession, SessionOutcome, install_wander, render_brief from orchestrator.steps import STEPS, people_steps, planned_steps -from orchestrator.supervisor import open_run, runs_needing_work, work_on +from orchestrator.supervisor import due, open_run, runs_needing_work, work_on AGENT = """ import os, pathlib, subprocess @@ -183,6 +184,141 @@ def test_a_question_ends_the_turn_and_holds_the_run(self): self.assertIsNone(outcome.finished) self.assertEqual(runs_needing_work(self.runs), [], "a waiting run is not picked up again") + def test_a_held_session_stays_idle_until_the_answer_reaches_its_saved_session(self): + run_dir = self.open() + journal = Journal(run_dir) + journal.append("question", question="Use the retained clip?", step="clean") + (run_dir / "session.id").write_text("saved-thread\n") + session = RunSession("run-1", run_dir, HarnessAgent(self.agent("pass"))) + session.harness.run = Mock( + return_value=HarnessOutcome( + AgentResult(status="completed", transcript_path="inert"), + session="saved-thread", + ) + ) + + held = session.turn() + self.assertEqual(held.waiting, "Use the retained clip?") + session.harness.run.assert_not_called() + + journal.append("answer", text="Yes; use the retained clip without new inference") + resumed = session.turn() + self.assertIsNone(resumed.waiting) + session.harness.run.assert_called_once() + args, kwargs = session.harness.run.call_args + self.assertIsNone(args[0]) + self.assertEqual(kwargs["session"], "saved-thread") + self.assertIn("answered: Yes; use the retained clip without new inference", args[2]) + self.assertEqual((run_dir / "session.id").read_text(), "saved-thread\n") + + def test_operator_controls_take_precedence_over_a_question_hold(self): + run_dir = self.open() + journal = Journal(run_dir) + journal.append("question", question="Use the retained clip?", step="clean") + session = RunSession("run-1", run_dir, HarnessAgent(self.agent("pass"))) + session.harness.run = Mock(side_effect=AssertionError("held run spawned an agent")) + + (run_dir / "paused").touch() + paused = session.turn() + self.assertIn("paused", paused.error) + self.assertIsNone(paused.waiting) + + (run_dir / "paused").unlink() + journal.append("run.finished", status="canceled") + canceled = session.turn() + self.assertEqual(canceled.finished, "canceled") + self.assertIsNone(canceled.waiting) + session.harness.run.assert_not_called() + + def test_a_stalled_session_id_is_cleared_before_the_next_turn(self): + run_dir = self.open() + (run_dir / "session.id").write_text("stalled-thread\n") + session = RunSession("run-1", run_dir, HarnessAgent(self.agent("pass"))) + session.harness.run = Mock( + side_effect=[ + HarnessOutcome( + AgentResult(status="stalled", transcript_path="inert", error="silent"), + session="stalled-thread", + ), + HarnessOutcome( + AgentResult(status="completed", transcript_path="inert"), + session="fresh-thread", + ), + ] + ) + + stalled = session.turn() + self.assertEqual(stalled.error, "silent") + self.assertFalse((run_dir / "session.id").exists()) + + recovered = session.turn() + self.assertIsNone(recovered.error) + self.assertEqual(session.harness.run.call_args_list[0].kwargs["session"], "stalled-thread") + self.assertIsNone(session.harness.run.call_args_list[1].kwargs["session"]) + self.assertEqual((run_dir / "session.id").read_text(), "fresh-thread\n") + + def test_answer_arriving_during_outcome_read_remains_after_observed_cursor(self): + run_dir = self.open() + journal = Journal(run_dir) + session = RunSession("run-1", run_dir, HarnessAgent(self.agent("pass"))) + + def ask(*args, **kwargs): + journal.append("question", question="Use the retained clip?", step="clean") + return HarnessOutcome(AgentResult(status="completed", transcript_path="inert")) + + session.harness.run = Mock(side_effect=ask) + original_finished = Journal.finished + replied = False + + def finish_then_reply(current): + nonlocal replied + finished = original_finished(current) + if current.run_dir == run_dir and current.pending_question() and not replied: + replied = True + current.append("answer", text="Use the retained clip") + return finished + + with patch.object(Journal, "finished", autospec=True, side_effect=finish_then_reply): + outcome = session.turn() + + self.assertTrue(replied) + self.assertIsNone(outcome.waiting) + self.assertEqual(outcome.observed_entries, 1) + self.assertEqual(len(journal.entries()), 2) + self.assertTrue( + due(run_dir, {str(run_dir): (time.monotonic(), outcome.observed_entries)}, 3600) + ) + + def test_no_progress_note_advances_cursor_but_later_reply_wakes_run(self): + run_dir = self.open() + session = RunSession("run-1", run_dir, HarnessAgent(self.agent("pass"))) + session.harness.run = Mock( + return_value=HarnessOutcome(AgentResult(status="completed", transcript_path="inert")) + ) + + outcome = session.work(turns=1) + journal = Journal(run_dir) + self.assertEqual(outcome.observed_entries, len(journal.entries())) + pushed = {str(run_dir): (time.monotonic(), outcome.observed_entries)} + self.assertFalse(due(run_dir, pushed, 3600)) + + journal.append("answer", text="Continue with retained inputs") + self.assertTrue(due(run_dir, pushed, 3600)) + + def test_no_progress_note_leaves_cursor_behind_an_intervening_reply(self): + run_dir = self.open() + journal = Journal(run_dir) + session = RunSession("run-1", run_dir, HarnessAgent(self.agent("pass"))) + + def reply_after_observation(): + journal.append("answer", text="Continue with retained inputs") + return SessionOutcome(finished=None, waiting=None, steps_run=0, observed_entries=0) + + session.turn = Mock(side_effect=reply_after_observation) + outcome = session.work(turns=1) + self.assertEqual(outcome.observed_entries, 0) + self.assertEqual(len(journal.entries()), 2) + def test_a_turn_with_work_in_flight_is_not_a_dead_end(self): """Stopping while a ten-minute step runs is correct, and not the same as giving up.""" run_dir = self.open() @@ -300,7 +436,7 @@ def test_the_supervisor_reads_the_outcome_out_of_the_journal(self): delivered_source.parent.mkdir(parents=True) shutil.copyfile(self.clip, delivered_source) policy = self.agent('subprocess.run(["wander", "finish", "succeeded", "done"])') - self.assertIn("finished succeeded", work_on(self.runs / "demo", None, policy, 2)) + self.assertIn("finished succeeded", work_on(self.runs / "demo", None, policy, 2).message) self.assertEqual(runs_needing_work(self.runs), [], "a finished run is not picked up again") def test_a_run_holds_its_own_clip_and_its_own_plan(self): diff --git a/scripts/tests/test_run_controls.py b/scripts/tests/test_run_controls.py index 89482ea..e976544 100644 --- a/scripts/tests/test_run_controls.py +++ b/scripts/tests/test_run_controls.py @@ -66,6 +66,8 @@ def step_args(wait=False, *, attempt=None): def control(run, status): if status == "paused": (run.context.run_dir / "paused").write_text("paused by an operator\n") + elif status == "question": + run.journal.append("question", question="Which source should we use?", step="pi3x") else: run.journal.append("run.finished", status="canceled", summary="operator canceled") @@ -83,7 +85,7 @@ def session(run): ) -@pytest.mark.parametrize("status", ["paused", "canceled"]) +@pytest.mark.parametrize("status", ["paused", "canceled", "question"]) @pytest.mark.parametrize("wait", [False, True]) def test_operator_control_rejects_stage_before_preflight_or_subprocess( run, monkeypatch, status, wait @@ -103,7 +105,7 @@ def test_operator_control_rejects_stage_before_preflight_or_subprocess( assert run.repository.run_summary("run-1")["status"] == "canceled" -@pytest.mark.parametrize("status", ["paused", "canceled"]) +@pytest.mark.parametrize("status", ["paused", "canceled", "question"]) def test_detached_child_closes_pending_attempt_without_launch_or_receipt_change( run, monkeypatch, status ): @@ -271,3 +273,90 @@ def already_running(*args, **kwargs): summary = run.repository.run_summary("run-1") assert summary["status"] == "canceled" assert summary["attempts"][0]["status"] == "succeeded" + + +@pytest.mark.parametrize("wait", [False, True]) +def test_answer_releases_stage_hold_without_fabricating_work(run, monkeypatch, wait): + control(run, "question") + spawn = Mock(return_value=stage_process()) + monkeypatch.setattr(cli.subprocess, "Popen", spawn) + assert cli.command_step(run.context, step_args(wait)) == 2 + assert not run.repository.run_summary("run-1")["attempts"] + run.journal.append("answer", text="Use the saved source", author="operator") + assert cli.command_step(run.context, step_args(wait)) == 0 + spawn.assert_called_once() + + +def test_question_does_not_cancel_work_already_launched(run, monkeypatch): + def already_running(*args, **kwargs): + control(run, "question") + return stage_process() + + spawn = Mock(side_effect=already_running) + monkeypatch.setattr(cli.subprocess, "Popen", spawn) + assert cli.command_step(run.context, step_args(True)) == 0 + assert run.journal.last_status("pi3x") == "succeeded" + assert run.journal.unanswered() == "Which source should we use?" + + +@pytest.mark.parametrize("status", ["paused", "canceled", "question"]) +def test_hold_during_harness_prompt_preparation_prevents_actual_spawn(run, monkeypatch, status): + turn = session(run) + transcript = run.context.run_dir / "transcript.jsonl" + decision = run.context.run_dir / "decision.json" + transcript.write_text("previous transcript\n") + decision.write_text("previous decision\n") + open_file = Path.open + + def hold_before_reading_prompt(path, *args, **kwargs): + if path == run.context.run_dir / "prompt.txt" and args == ("rb",): + with run.journal.admission_lock(): + control(run, status) + return open_file(path, *args, **kwargs) + + monkeypatch.setattr(Path, "open", hold_before_reading_prompt) + spawn = Mock(side_effect=AssertionError("must not spawn a held harness")) + monkeypatch.setattr("orchestrator.agent.harness.subprocess.Popen", spawn) + outcome = turn.turn() + spawn.assert_not_called() + assert transcript.read_text() == "previous transcript\n" + assert decision.read_text() == "previous decision\n" + assert not outcome.wants_another_turn + if status == "question": + assert outcome.waiting == "Which source should we use?" + + +@pytest.mark.parametrize("text", ["", " \t\n"]) +def test_empty_question_is_rejected_without_changing_journal(run, text): + assert cli.command_ask(run.context, SimpleNamespace(question=text, step="pi3x")) == 2 + assert run.journal.entries() == [] + + +def test_zero_turn_budget_does_not_forget_existing_journal_cursor(run): + run.journal.append("note", text="Existing retained work") + outcome = session(run).work(turns=0) + assert outcome.observed_entries == len(run.journal.entries()) == 1 + + +def test_question_before_watchdog_nudge_preserves_new_session(run, monkeypatch): + from dataclasses import replace + from orchestrator.agent import harness + + turn = session(run) + turn.harness.policy = replace(turn.harness.policy, kind="codex", idle_nudges=1) + launches = [] + + def launch(*args, **kwargs): + launches.append(args) + kwargs["stdout"].write(b'{"type":"thread.started","thread_id":"asked-session"}\n') + kwargs["stdout"].flush() + control(run, "question") + process = Mock() + process.wait.side_effect = harness.Stalled() + return process + + monkeypatch.setattr(harness.subprocess, "Popen", launch) + outcome = turn.turn() + assert outcome.waiting == "Which source should we use?" + assert len(launches) == 1 + assert (run.context.run_dir / "session.id").read_text().strip() == "asked-session" diff --git a/scripts/tests/test_step_admission.py b/scripts/tests/test_step_admission.py index ec9193e..47bf9a8 100644 --- a/scripts/tests/test_step_admission.py +++ b/scripts/tests/test_step_admission.py @@ -213,7 +213,7 @@ def test_stale_completion_cannot_hide_older_running_attempt(run): assert cli.in_flight(run.journal) == {"pi3x": "older"} -@pytest.mark.parametrize("status", ["paused", "canceled"]) +@pytest.mark.parametrize("status", ["paused", "canceled", "question"]) def test_control_between_claim_and_launch_prevents_execution(run, monkeypatch, status): open_file = Path.open @@ -233,7 +233,10 @@ def control_before_log(path, *args, **kwargs): spawn.assert_not_called() -def test_launch_and_operator_control_are_serialized_without_waiting_for_execution(run, monkeypatch): +@pytest.mark.parametrize("status", ["paused", "question"]) +def test_launch_and_operator_control_are_serialized_without_waiting_for_execution( + run, monkeypatch, status +): at_launch, allow_launch, at_execution, finish = Event(), Event(), Event(), Event() control_requested, control_accepted = Event(), Event() @@ -249,8 +252,11 @@ def spawn(*args, **kwargs): def pause(): control_requested.set() - with run.journal.admission_lock(): - fixtures.control(run, "paused") + if status == "question": + cli.command_ask(run.context, SimpleNamespace(question="Review first", step="pi3x")) + else: + with run.journal.admission_lock(): + fixtures.control(run, status) control_accepted.set() monkeypatch.setattr(cli.subprocess, "Popen", spawn) @@ -270,7 +276,10 @@ def pause(): allow_launch.set() finish.set() assert owner.result(timeout=5) == 0 - assert run.journal.control_status() == "paused" + if status == "question": + assert run.journal.unanswered() == "Review first" + else: + assert run.journal.control_status() == "paused" def test_real_detached_processes_share_admission_and_reject_child_replay(run, tmp_path): diff --git a/scripts/tests/test_supervisor_wakeup.py b/scripts/tests/test_supervisor_wakeup.py new file mode 100644 index 0000000..00d7f99 --- /dev/null +++ b/scripts/tests/test_supervisor_wakeup.py @@ -0,0 +1,85 @@ +"""Journal events that arrive as a turn ends must wake the next supervisor pass.""" + +import time + +import pytest + +from orchestrator import supervisor +from orchestrator.agent.harness import HarnessPolicy +from orchestrator.journal import Journal + + +def test_answer_after_outcome_is_seen_on_next_pass_without_waiting_for_timer(tmp_path, monkeypatch): + run = supervisor.open_run(tmp_path, run_id="reply-run", name="clip", source=None) + supervisor.open_run(tmp_path, run_id="unrelated-run", name="another-clip", source=None) + journal = Journal(run) + calls = [] + + def work(*args): + calls.append(args[0]) + if len(calls) == 1: + journal.append("question", question="Continue?") + outcome = supervisor.WorkOutcome("waiting", len(journal.entries())) + # The operator answers after the session's last read but before the main + # loop records its cursor. The answer must still trigger another turn. + journal.append("answer", text="Continue with retained inputs only") + return outcome + assert journal.unanswered() is None + journal.append("question", question="Finished the bounded check; review outputs") + return supervisor.WorkOutcome("waiting for review", len(journal.entries())) + + passes = 0 + + class EndCheck(Exception): + pass + + def tick(_): + nonlocal passes + passes += 1 + if passes == 3: + raise EndCheck + + monkeypatch.setattr(supervisor, "repository_from_environment", lambda: None) + monkeypatch.setattr( + HarnessPolicy, + "from_environment", + lambda: HarnessPolicy(kind="command", command=("unused",)), + ) + monkeypatch.setattr(supervisor, "work_on", work) + monkeypatch.setattr(supervisor.time, "sleep", tick) + with pytest.raises(EndCheck): + supervisor.main( + ["--runs", str(tmp_path), "--push-after", "3600", "--run-id", "reply-run", "--watch"] + ) + assert calls == [run, run] + assert supervisor.runs_needing_work(tmp_path) == [tmp_path / "another-clip"] + + +def test_observed_journal_stays_quiet_until_a_new_event_or_timer(tmp_path): + journal = Journal(tmp_path) + journal.append("note", text="No progress in this turn") + pushed = {str(tmp_path): (time.monotonic(), len(journal.entries()))} + assert not supervisor.due(tmp_path, pushed, 3600) + journal.append("answer", text="Try the retained source") + assert supervisor.due(tmp_path, pushed, 3600) + + +@pytest.mark.parametrize("flags", [[], ["--once"]]) +def test_named_run_still_exits_after_one_pass_without_watch(tmp_path, monkeypatch, flags): + run = supervisor.open_run(tmp_path, run_id="reply-run", name="clip", source=None) + calls = [] + monkeypatch.setattr(supervisor, "repository_from_environment", lambda: None) + monkeypatch.setattr( + supervisor, + "work_on", + lambda path, *_: calls.append(path) or supervisor.WorkOutcome("quiet", 0), + ) + monkeypatch.setattr(supervisor.time, "sleep", lambda _: pytest.fail("one pass must not wait")) + assert supervisor.main(["--runs", str(tmp_path), "--run-id", "reply-run", *flags]) == 0 + assert calls == [run] + + +def test_once_and_watch_are_mutually_exclusive(): + with pytest.raises(SystemExit) as error: + supervisor.main(["--once", "--watch"]) + assert error.value.code == 2 From 6e6fcae922390045b2fc609cf98fb8a216bdc293 Mon Sep 17 00:00:00 2001 From: Austin Jian Date: Fri, 16 Oct 2026 17:35:27 -0400 Subject: [PATCH 2/2] fix: skip shot ranking during cut rejection checks --- docs/pipeline-setup.md | 2 ++ docs/pipeline-validation.md | 14 ++++++++++++++ scripts/run_clip.py | 4 ++-- scripts/tests/test_run_clip_cut_cache.py | 15 ++++++++++++++- 4 files changed, 32 insertions(+), 3 deletions(-) diff --git a/docs/pipeline-setup.md b/docs/pipeline-setup.md index 4d4ac34..de4e05d 100644 --- a/docs/pipeline-setup.md +++ b/docs/pipeline-setup.md @@ -137,6 +137,8 @@ New runs retain a source SHA-256 in `run.json`. Stages refuse missing or changed before launching; existing outputs stay on disk. Legacy manifests without a digest need a new submission instead of blindly reusing cached state. Orchestrated runs reject cuts instead of silently trimming into another run directory. The direct runner retains its explicit shot controls. +Reject-only cut checks skip shot ranking and its model loads; their detector arguments match +the cache key. Cut detection and rejection still run before any stage. `wander step` admits one unresolved attempt per step in a run directory. A filesystem lock serializes admission, then releases before execution so independent steps can proceed. Detached diff --git a/docs/pipeline-validation.md b/docs/pipeline-validation.md index 23d64c1..7a9ca93 100644 --- a/docs/pipeline-validation.md +++ b/docs/pipeline-validation.md @@ -104,6 +104,20 @@ sign-in; inherited unrelated MCP sign-in and duplicate-hook warnings did not pre turn. The broken system CLI was then updated to that version and its exec/resume startup checks passed. Evidence: `.context/evidence/worker-reply-resume/`. +A further bottle delivery check used the real Codex agent to inspect the retained inputs, +launch one detached `package_people` stage, end its turn, and resume when packaging finished. +The stage completed in 78.3 seconds; the agent read its result and requested viewer review. +The live admin showed one successful attempt and distinguished five saved-input stages from +fresh execution. All 240 animated person frames, room, object assets, original audio, and +presentation matched the prior delivery. Camera/source URLs correctly named the new run. +Browser checks loaded both people, enabled the original soundtrack, advanced playback, reached +frame 119 at 9.93 seconds, wrapped to frame zero, and moved beyond the room without an edge +fade. No console errors were observed. The supervisor stopped and the run was paused. +This was the explicitly requested fifth packaging execution after four successful delivery +checks, with no automatic retry, provider commands, paid-ledger change, or quality acceptance. +It joins actual agent control to retained packaging and viewer delivery; fresh reconstruction +and headset behavior remain unverified. Evidence: `.context/evidence/bottle-agent-delivery/`. + 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. diff --git a/scripts/run_clip.py b/scripts/run_clip.py index 18aea97..dfdb21f 100644 --- a/scripts/run_clip.py +++ b/scripts/run_clip.py @@ -3077,8 +3077,8 @@ def valid_report(value): "--threshold", str(a.cut_threshold), ] - if a.shot is not None or a.all_shots: - cmd.append("--no-score") # the choice is already made; do not pay to rank + if no_score: + cmd.append("--no-score") # chosen shots and reject-only checks do not need ranking say("cut check: looking for cuts before anything is spent") try: p = subprocess.run( diff --git a/scripts/tests/test_run_clip_cut_cache.py b/scripts/tests/test_run_clip_cut_cache.py index 2c639c5..0f79bbe 100644 --- a/scripts/tests/test_run_clip_cut_cache.py +++ b/scripts/tests/test_run_clip_cut_cache.py @@ -61,13 +61,14 @@ def restore(self): run_clip.ROOT = self.old_root self.temp.cleanup() - def args(self, threshold=0.1, shot=None, all_shots=False): + def args(self, threshold=0.1, shot=None, all_shots=False, reject_cuts=False): return SimpleNamespace( clip=str(self.clip), name="same-run", cut_threshold=threshold, shot=shot, all_shots=all_shots, + reject_cuts=reject_cuts, ) def cache_path(self): @@ -133,6 +134,18 @@ def test_score_mode_change_same_threshold_misses(self): self.assertEqual(len(self.calls), 2) self.assertIn("--no-score", self.calls[-1]) + def test_reject_cuts_skips_ranking_and_matches_its_cached_detector_args(self): + with patch.object(run_clip.subprocess, "run", side_effect=self.fake): + run_clip.cut_check(self.args()) + result = run_clip.cut_check(self.args(reject_cuts=True)) + cached = run_clip.cut_check(self.args(reject_cuts=True)) + self.assertEqual(len(self.calls), 2) + self.assertNotIn("--no-score", self.calls[0]) + self.assertIn("--no-score", self.calls[1]) + self.assertIn("--no-score", result["_cacheKey"]["detectorArgs"]) + self.assertFalse(result["continuous"], "skipping ranking must retain cut detection") + self.assertEqual(result, cached) + def test_detector_key_change_misses(self): with patch.object(run_clip.subprocess, "run", side_effect=self.fake): run_clip.cut_check(self.args())