fix(orchestrator): bounded dispatch wait — advance_project never holds the MCP call for a whole worker run

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",

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:

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,15 +163,19 @@ 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,
             task=task,
             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")
 
     # ------------------------------------------------------------------

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(