From 70993ec57b7edcb873f9cacfed48e77b3f23a1ef Mon Sep 17 00:00:00 2001 From: keeganmccallum Date: Tue, 11 Aug 2026 17:12:52 -0700 Subject: [PATCH] =?UTF-8?q?fix(orchestrator):=20bounded=20dispatch=20wait?= =?UTF-8?q?=20=E2=80=94=20advance=5Fproject=20never=20holds=20the=20MCP=20?= =?UTF-8?q?call=20for=20a=20whole=20worker=20run?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ACP dispatcher blocks on the worker's entire session, so advance_project held the orchestrator's MCP request for hours. MCP clients abort held requests (Prime Agent: 'Request was aborted' — the abort killed the caller's kernel cell), making a healthy dispatch look like 'nothing is running'. Fix: - StepResult.in_progress: advance returns promptly when workers are in flight; the next call reconciles their handoff files (workers write them directly). - .dispatched markers distinguish in-flight attempts from lost ones; the legacy 'running but no attempt file' crash-recovery only fires on missing or stale markers (attempt_stale_s, default 6h). - dispatch_wait_s (default 50s, ZENITH_DISPATCH_WAIT_S) bounds the wait — safely under MCP client timeouts. Tests: tests/test_bounded_dispatch.py (4) — prompt return under a slow worker, in_progress reconcile without fake failures, late-handoff application, lost and stale-marker paths. Full suite 217 passed, 7 skipped. --- zenith/src/zenith_harness/config.py | 21 +++ zenith/src/zenith_harness/controller.py | 2 +- zenith/src/zenith_harness/coordinator.py | 110 ++++++++++++++- zenith/src/zenith_harness/server.py | 18 ++- zenith/tests/test_bounded_dispatch.py | 164 +++++++++++++++++++++++ 5 files changed, 304 insertions(+), 11 deletions(-) create mode 100644 zenith/tests/test_bounded_dispatch.py diff --git a/zenith/src/zenith_harness/config.py b/zenith/src/zenith_harness/config.py index 22adcc0..d743e58 100644 --- a/zenith/src/zenith_harness/config.py +++ b/zenith/src/zenith_harness/config.py @@ -33,6 +33,15 @@ def _resolve_optional_path(value: str | None) -> Path | None: return Path(value).expanduser().resolve() +def _resolve_float(raw: str | None, *, default: float) -> float: + if raw is None or not raw.strip(): + return default + try: + return float(raw) + except ValueError: + return default + + def _resolve_max_parallel(value: str | None) -> int: if not value: return DEFAULT_MAX_PARALLEL_NODES @@ -72,6 +81,12 @@ class HarnessConfig: terminal_reviewer_provider_name: str | None terminal_reviewer_acp_command: str | None max_parallel_nodes: int = DEFAULT_MAX_PARALLEL_NODES + # Bounded dispatch wait: how long advance_project waits for a worker + # handoff before returning in_progress. MUST stay under orchestrator MCP + # client timeouts (Prime Agent aborts held requests). + dispatch_wait_s: float = 50.0 + # A .dispatched marker older than this with no handoff file = lost attempt. + attempt_stale_s: float = 6 * 3600.0 # Per-role reasoning effort for providers whose ACP command accepts one # (codex today). None means the provider default ("xhigh" for codex). worker_reasoning_effort: str | None = None @@ -123,6 +138,12 @@ def discover(cls) -> HarnessConfig: max_parallel_nodes=_resolve_max_parallel( os.environ.get("ZENITH_MAX_PARALLEL_NODES") ), + dispatch_wait_s=_resolve_float( + os.environ.get("ZENITH_DISPATCH_WAIT_S"), default=50.0 + ), + attempt_stale_s=_resolve_float( + os.environ.get("ZENITH_ATTEMPT_STALE_S"), default=6 * 3600.0 + ), worker_reasoning_effort=_resolve_reasoning_effort( os.environ.get("ZENITH_WORKER_REASONING_EFFORT"), env_var="ZENITH_WORKER_REASONING_EFFORT", diff --git a/zenith/src/zenith_harness/controller.py b/zenith/src/zenith_harness/controller.py index f7aeeeb..b40092c 100644 --- a/zenith/src/zenith_harness/controller.py +++ b/zenith/src/zenith_harness/controller.py @@ -129,7 +129,7 @@ def advance_project( steps = 0 while True: result = coordinator.step() - if result.kind in ("attention_needed", "terminal", "idle"): + if result.kind in ("attention_needed", "terminal", "idle", "in_progress"): break steps += 1 if max_steps is not None and steps >= max_steps: diff --git a/zenith/src/zenith_harness/coordinator.py b/zenith/src/zenith_harness/coordinator.py index 3fa17bb..99990b8 100644 --- a/zenith/src/zenith_harness/coordinator.py +++ b/zenith/src/zenith_harness/coordinator.py @@ -7,6 +7,8 @@ from __future__ import annotations import concurrent.futures +import json +import time from dataclasses import dataclass, field from typing import Literal @@ -42,7 +44,7 @@ # --------------------------------------------------------------------------- -StepKind = Literal["idle", "advanced", "attention_needed", "terminal"] +StepKind = Literal["idle", "advanced", "attention_needed", "terminal", "in_progress"] @dataclass(frozen=True) @@ -66,6 +68,14 @@ def attention_needed(cls, detail: str = "") -> "StepResult": def terminal(cls, detail: str = "") -> "StepResult": return cls("terminal", detail) + @classmethod + def in_progress(cls, detail: str = "") -> "StepResult": + """Workers are dispatched and still running; the caller should return + now and re-enter advance_project later to reconcile completed handoffs. + Exists so a bounded orchestrator call never blocks a whole worker run + (MCP clients abort long-held requests).""" + return cls("in_progress", detail) + # --------------------------------------------------------------------------- # MissionCoordinator @@ -153,6 +163,7 @@ def _dispatch_one(self, mid: str, task: Task) -> StepResult: task_state.set_last_attempt(task.id, spawn_ts) self.store.save_task_state(self.project_id, mid, task_state) + self._write_dispatch_marker(mid, task.id, spawn_ts) request = DispatchRequest( project_id=self.project_id, mission_id=mid, @@ -160,8 +171,11 @@ def _dispatch_one(self, mid: str, task: Task) -> StepResult: spawn_ts=spawn_ts, ) try: - handoff = self.dispatcher.dispatch(request) + handoff, completed = self._call_with_timeout( + lambda: self.dispatcher.dispatch(request) + ) except Exception as exc: # noqa: BLE001 + self._clear_dispatch_marker(mid, task.id, spawn_ts) synthetic = self._synthesize_handoff(task, f"Dispatcher crashed: {exc}") self.store.save_attempt( self.project_id, @@ -171,7 +185,13 @@ def _dispatch_one(self, mid: str, task: Task) -> StepResult: synthetic, ) return self._apply_handoff(mid, task, synthetic, spawn_ts) + if not completed: + # Bounded-wait law: never hold the orchestrator call for a whole + # worker run. The worker writes its own handoff file; the next + # advance reconciles it (see _reconcile_pending_attempts). + return StepResult.in_progress(f"dispatched {task.id}; worker still running") + self._clear_dispatch_marker(mid, task.id, spawn_ts) self.store.save_attempt(self.project_id, mid, spawn_ts, task.id, handoff) return self._apply_handoff(mid, task, handoff, spawn_ts) @@ -271,6 +291,8 @@ def _dispatch_batch( _BatchAttempt(task=task, spawn_ts=spawn_ts) ) self.store.save_task_state(self.project_id, mid, task_state) + for attempt in batch_attempts: + self._write_dispatch_marker(mid, attempt.task.id, attempt.spawn_ts) requests = [ DispatchRequest( @@ -281,11 +303,19 @@ def _dispatch_batch( ) for attempt in batch_attempts ] - handoffs = self._dispatch_requests(requests) + handoffs, completed = self._call_with_timeout( + lambda: self._dispatch_requests(requests) + ) + if not completed: + # Bounded-wait law: workers write their own handoff files; the next + # advance reconciles them (see _reconcile_pending_attempts). + ids = ", ".join(attempt.task.id for attempt in batch_attempts) + return StepResult.in_progress(f"dispatched {ids}; workers still running") attention: list[AttentionItemInternal] = [] for attempt in sorted(batch_attempts, key=lambda item: item.task.id): handoff = handoffs[attempt.task.id] + self._clear_dispatch_marker(mid, attempt.task.id, attempt.spawn_ts) self.store.save_attempt( self.project_id, mid, @@ -352,6 +382,50 @@ def _run(request: DispatchRequest) -> tuple[str, NodeHandoff]: def _batch_spawn_ts(index: int) -> str: return f"{utc_now_filesafe()}-{index:04d}" + # ------------------------------------------------------------------ + # Bounded dispatch wait + in-flight markers + # + # The ACP runner's dispatch blocks for the worker's WHOLE run (the ACP + # session/prompt request returns only when the worker's turn ends). An + # orchestrator MCP call held that long gets aborted by clients (Prime + # Agent: "Request was aborted"; the abort can kill the caller's session + # cell), which looked exactly like "nothing is running". The worker + # process writes its own handoff file independently of this wait, so the + # wait is bounded and in-flight attempts are marked; the next + # advance_project reconciles completed handoffs from the store. + # ------------------------------------------------------------------ + + def _dispatch_marker_path(self, mid: str, node_id: str, spawn_ts: str): + d = self.store.attempts_runtime_dir(self.project_id, mid) + return d / f"{spawn_ts}__{node_id}.dispatched" + + def _write_dispatch_marker(self, mid: str, node_id: str, spawn_ts: str) -> None: + path = self._dispatch_marker_path(mid, node_id, spawn_ts) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"spawn_ts": spawn_ts, "node_id": node_id})) + + def _clear_dispatch_marker(self, mid: str, node_id: str, spawn_ts: str) -> None: + try: + self._dispatch_marker_path(mid, node_id, spawn_ts).unlink() + except FileNotFoundError: + pass + + def _call_with_timeout(self, fn): + """Run fn on a helper thread, bounded by config.dispatch_wait_s. + + Returns (result, True) on completion, (None, False) on timeout. A + timed-out thread is left to finish in the background — its result is + discarded because the worker's own handoff file is the durable record. + """ + wait_s = getattr(self.store.config, "dispatch_wait_s", 50.0) + pool = concurrent.futures.ThreadPoolExecutor(max_workers=1) + fut = pool.submit(fn) + try: + return fut.result(timeout=wait_s), True + except concurrent.futures.TimeoutError: + pool.shutdown(wait=False) + return None, False + def _apply_handoff( self, mid: str, @@ -760,6 +834,9 @@ def _reconcile_pending_attempts( ) -> StepResult | None: attention: list[AttentionItemInternal] = [] saw_running = False + applied = 0 + in_flight: list[str] = [] + stale_s = getattr(self.store.config, "attempt_stale_s", 6 * 3600.0) for task in tl.tasks: if task_state.status_of(task.id) != "running": continue @@ -768,8 +845,23 @@ def _reconcile_pending_attempts( if not attempts: entry = task_state.tasks.get(task.id) spawn_ts = entry.last_attempt if entry is not None else None + # In-flight vs lost: a fresh .dispatched marker means a worker + # is legitimately still running (bounded-wait dispatch); only + # a missing or stale marker means the attempt was lost. + marker = ( + self._dispatch_marker_path(mid, task.id, spawn_ts) + if spawn_ts is not None + else None + ) + if marker is not None and marker.exists(): + age_s = time.time() - marker.stat().st_mtime + if age_s < stale_s: + in_flight.append(task.id) + continue if spawn_ts is None: spawn_ts = utc_now_filesafe() + if marker is not None: + self._clear_dispatch_marker(mid, task.id, spawn_ts) handoff = self._synthesize_handoff( task, "Coordinator resumed with task marked running but no attempt file was present.", @@ -784,21 +876,33 @@ def _reconcile_pending_attempts( attention.extend( self._apply_handoff_collect(mid, task, handoff, spawn_ts) ) + applied += 1 continue last = attempts[-1] read_handoff = self.store.read_attempt( self.project_id, mid, last.spawn_ts, task.id ) if read_handoff is None: + # Attempt dir entry exists but no parseable json yet; a fresh + # marker means the worker is still writing. + marker = self._dispatch_marker_path(mid, task.id, last.spawn_ts) + if marker.exists() and (time.time() - marker.stat().st_mtime) < stale_s: + in_flight.append(task.id) continue + self._clear_dispatch_marker(mid, task.id, last.spawn_ts) attention.extend( self._apply_handoff_collect(mid, task, read_handoff, last.spawn_ts) ) + applied += 1 if not saw_running: return None if attention: self._raise_attention(attention) return StepResult.attention_needed("resume_attention") + if applied: + return StepResult.advanced(f"reconciled {applied} completed attempt(s)") + if in_flight: + return StepResult.in_progress("workers still running: " + ", ".join(in_flight)) return StepResult.advanced("reconciled pending attempts") # ------------------------------------------------------------------ diff --git a/zenith/src/zenith_harness/server.py b/zenith/src/zenith_harness/server.py index 19c37eb..e0e4111 100644 --- a/zenith/src/zenith_harness/server.py +++ b/zenith/src/zenith_harness/server.py @@ -174,13 +174,17 @@ async def submit_plan( @mcp.tool( name="advance_project", description=( - "Drive the runtime forward. BLOCKING — may run for many minutes while " - "workers dispatch according to runtime scheduling. " - "Call whenever state is mission_running. " - "Returns when attention is needed, no runnable task work remains, or " - "`max_steps` exhausts. It does not request mission closure; call " - "end_mission when you intend to close after task work is quiescent. " - "If it returns still mission_running with runnable work, call it again." + "Drive the runtime forward. BOUNDED — dispatches runnable work, then " + "waits at most `dispatch_wait_s` (default 50s, ZENITH_DISPATCH_WAIT_S) " + "for handoffs before returning; it NEVER holds the call for a whole " + "worker run (workers write their own handoff files; the next call " + "reconciles them). Call whenever state is mission_running. " + "Returns when attention is needed, workers are in flight " + "(in_progress), no runnable task work remains, or `max_steps` " + "exhausts. It does not request mission closure; call end_mission " + "when you intend to close after task work is quiescent. " + "If it returns still mission_running with runnable or running work, " + "call it again." ), ) async def advance_project( diff --git a/zenith/tests/test_bounded_dispatch.py b/zenith/tests/test_bounded_dispatch.py new file mode 100644 index 0000000..3adf7b8 --- /dev/null +++ b/zenith/tests/test_bounded_dispatch.py @@ -0,0 +1,164 @@ +"""Bounded dispatch wait — advance_project must never hold the orchestrator +call for a whole worker run. + +The ACP dispatcher blocks on the worker's entire session; an orchestrator MCP +call held that long gets aborted by clients (Prime Agent: "Request was +aborted"). The bounded wait + .dispatched markers make advance return +in_progress while workers run, and reconcile applies their handoff files when +they land. +""" +from __future__ import annotations + +import os +import time +from pathlib import Path + +import pytest + +from zenith_harness.config import HarnessConfig +from zenith_harness.controller import ProjectController +from zenith_harness.dispatcher import DispatchRequest, MockDispatcher, MockTerminalReviewer +from zenith_harness.models import ( + AttentionNeeded, + Task, + TaskList, + TerminalReviewHandoff, + WorkHandoff, +) +from zenith_harness.storage import ProjectStore + + +@pytest.fixture +def config(harness_home: Path) -> HarnessConfig: + bundled = Path(__file__).resolve().parents[1] / "src" / "zenith_harness" / "bundled" + cfg = HarnessConfig( + bundled_dir=bundled, + harness_home=harness_home, + projects_dir=harness_home / "projects", + orchestrator_provider_name="claude", + worker_provider_name="claude", + worker_acp_command=None, + validator_provider_name=None, + validator_acp_command=None, + terminal_reviewer_provider_name=None, + terminal_reviewer_acp_command=None, + ) + object.__setattr__(cfg, "dispatch_wait_s", 0.2) + return cfg + + +def _task(tid: str, target: str) -> Task: + return Task(id=tid, type="work", body="b", targets=[target], skill="s") + + +def _write_contract(store: ProjectStore, pid: str, mission_id: str, assertion: str) -> None: + d = store.ensure_contract_dir(pid, mission_id) + (d / f"{assertion}.md").write_text(f"# {assertion}\n\nStatement body.\n") + + +def _started_controller(config, workspace, responder): + controller = ProjectController( + config, + MockDispatcher(responder), + MockTerminalReviewer(TerminalReviewHandoff(done=True, report="")), + ) + controller.start_project("Brief.", str(workspace)) + pid = controller.store.list_projects()[0].id + _write_contract(controller.store, pid, "mission-001", "VAL-A") + controller.submit_plan(pid, TaskList(tasks=[_task("a", "VAL-A")])) + return controller, pid + + +def test_advance_returns_promptly_while_worker_runs(config, workspace) -> None: + """A slow worker must not hold advance_project: returns fast, task running.""" + release = [] + + def slow(req: DispatchRequest) -> WorkHandoff: + release.append(req.task.id) + time.sleep(5) # far beyond dispatch_wait_s=0.2 + return WorkHandoff(node_id=req.task.id, done=True, report="late") + + controller, pid = _started_controller(config, workspace, slow) + t0 = time.monotonic() + env = controller.advance_project(pid) + elapsed = time.monotonic() - t0 + assert elapsed < 3.0, f"advance held the call for {elapsed:.1f}s" + assert env.state.state == "mission_running" + ts = controller.store.load_task_state(pid, "mission-001") + assert ts.status_of("a") == "running" + # in-flight marker present (worker still writing its handoff) + spawn_ts = ts.tasks["a"].last_attempt + marker = controller.store.attempts_runtime_dir(pid, "mission-001") / f"{spawn_ts}__a.dispatched" + assert marker.exists() + + +def test_reconcile_reports_in_progress_then_applies_late_handoff(config, workspace) -> None: + """Second advance while the worker runs: in_progress, no fake failure. + After the handoff file lands: applied, task cleared.""" + held: list[DispatchRequest] = [] + + def slow(req: DispatchRequest) -> WorkHandoff: + held.append(req) + time.sleep(5) + return WorkHandoff(node_id=req.task.id, done=True, report="late") + + controller, pid = _started_controller(config, workspace, slow) + controller.advance_project(pid) + + # Worker still in flight: reconcile must NOT synthesize a failure. + env = controller.advance_project(pid) + assert env.state.state == "mission_running" # not attention_needed + ts = controller.store.load_task_state(pid, "mission-001") + assert ts.status_of("a") == "running" + + # Worker finishes: its handoff file lands (what the worker MCP server writes). + spawn_ts = ts.tasks["a"].last_attempt + controller.store.save_attempt( + pid, "mission-001", spawn_ts, "a", WorkHandoff(node_id="a", done=True, report="done") + ) + env = controller.advance_project(pid) + ts = controller.store.load_task_state(pid, "mission-001") + assert ts.status_of("a") == "cleared" + # marker cleared on application + marker = controller.store.attempts_runtime_dir(pid, "mission-001") / f"{spawn_ts}__a.dispatched" + assert not marker.exists() + + +def test_lost_attempt_without_marker_still_synthesizes_failure(config, workspace) -> None: + """Legacy crash-recovery: running + no marker + no attempt = attention.""" + def instant(req: DispatchRequest) -> WorkHandoff: + return WorkHandoff(node_id=req.task.id, done=True, report="ok") + + controller, pid = _started_controller(config, workspace, instant) + # Simulate a pre-fix/crashed dispatch: running, last_attempt set, no marker. + ts = controller.store.load_task_state(pid, "mission-001") + ts.set_status("a", "running") + ts.set_last_attempt("a", "2000-01-01T00-00-00Z-0000") + controller.store.save_task_state(pid, "mission-001", ts) + + env = controller.advance_project(pid) + assert env.state.state == "attention_needed" + + +def test_stale_marker_counts_as_lost(config, workspace) -> None: + """A marker older than attempt_stale_s with no handoff = lost attempt.""" + object.__setattr__(config, "attempt_stale_s", 0.05) + + def instant(req: DispatchRequest) -> WorkHandoff: + return WorkHandoff(node_id=req.task.id, done=True, report="ok") + + controller, pid = _started_controller(config, workspace, instant) + ts = controller.store.load_task_state(pid, "mission-001") + ts.set_status("a", "running") + spawn_ts = "2000-01-01T00-00-00Z-0000" + ts.set_last_attempt("a", spawn_ts) + controller.store.save_task_state(pid, "mission-001", ts) + d = controller.store.attempts_runtime_dir(pid, "mission-001") + d.mkdir(parents=True, exist_ok=True) + marker = d / f"{spawn_ts}__a.dispatched" + marker.write_text("{}") + old = time.time() - 3600 + os.utime(marker, (old, old)) + + env = controller.advance_project(pid) + assert env.state.state == "attention_needed"