From c6316cc3742c1e9f68f76e5d6e173c3518ff2932 Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Mon, 13 Jul 2026 09:34:12 +0200 Subject: [PATCH 1/3] feat: worker liveness verification + orphaned-agent recovery (D7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Loopy consumer side of team-harness TH-D5 (proposals P0.1 liveness slice + P2.5), per decision D7's ownership split. Worker liveness (makes reclaim safe, not optimistic): - The worker sends its process identity (hostname + pid + team-harness's pid-reuse-proof starttime token) with /register and /finished; the coordinator stamps it onto every dispatched CurrentTask (register, finished, child dispatch, parent resume). - A /register while the recorded worker is VERIFIABLY alive returns HTTP 409 instead of abandoning live work — closing the duplicate-work window. Verification is same-host + token match; unknown identities (old workers, remote hosts, no token) preserve the pre-existing behavior. The wire change is backward compatible ({} register still works). - The bundled worker exits with code 3 on 409 and uses an unbounded read timeout (recovery can legitimately block /register up to the drain deadline). Orphaned-agent recovery (bounded drain by default): - On a confirmed-dead worker with nothing recoverable, the coordinator discovers the interrupted harness run(s) from the iteration's harness_outputs directory and applies recovery_policy (new config: drain|reap, default drain bounded by recovery_drain_timeout_s, default 600s) via team-harness's reaper. Drained agents finish; their repo edits survive (D1); the iteration is still re-run and its result is never fabricated (D3). - salvage.json in the interrupted iteration directory records the reap reports; the history entry is abandoned_after_drain instead of abandoned, so "crashed but salvaged" and "crashed, lost" are distinguishable. - team-harness's parent-liveness guard (ReapRefusedError) surfaces as 409. - team-harness is resolved at call time (importlib): versions without the reaper (<= 0.2.10) degrade gracefully to plain abandonment. New modules: worker_identity.py (capture + verified liveness), recovery.py (discovery + reap + salvage record). Config: recovery_policy, recovery_drain_timeout_s (also in the snapshot). Docs: http-contract, session-layout, README, website MDX pages (http-contract, session-layout, configuration), CHANGELOG. Tests: 19 new (135 total) — verified-alive 409 (state untouched), verified-dead + unknown-identity recovery, abandoned_after_drain, RecoveryRefused 409, identity stamping via register and finished, empty-body compatibility, recovery module (fake reaper: discovery, salvage schema, refusal, degraded mode, no-runs no-salvage), worker payload identity, 409 exit code, config defaults/overrides. Integration-verified against the real team-harness feat/th-d5-process-lifecycle branch in a separate venv: a real orphan was drained through the real reaper (salvage.json written, run.json status drained); the pinned 0.2.10 suite run verifies the degraded path. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01PN8aFwwxA8FzMy9LgpbQFt --- CHANGELOG.md | 22 + README.md | 4 + docs/http-contract.md | 58 +- docs/session-layout.md | 14 + src/loopy_loop/config.py | 18 + src/loopy_loop/coordinator_app.py | 92 +++- src/loopy_loop/models.py | 26 + src/loopy_loop/recovery.py | 191 +++++++ src/loopy_loop/worker.py | 43 +- src/loopy_loop/worker_identity.py | 64 +++ src/tests/test_worker.py | 6 +- src/tests/test_worker_liveness_recovery.py | 540 +++++++++++++++++++ website/src/app/docs/configuration/page.mdx | 2 + website/src/app/docs/http-contract/page.mdx | 37 +- website/src/app/docs/session-layout/page.mdx | 8 + 15 files changed, 1093 insertions(+), 32 deletions(-) create mode 100644 src/loopy_loop/recovery.py create mode 100644 src/loopy_loop/worker_identity.py create mode 100644 src/tests/test_worker_liveness_recovery.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 092ff47..e1e6d58 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,27 @@ # Changelog +## Unreleased + +- **Worker liveness verification (D7).** The worker now sends its process + identity (hostname + pid + a pid-reuse-proof start-time token) with + `/register` and `/finished`; the coordinator stamps it onto the dispatched + task. A `/register` while the recorded worker is *verifiably still alive* + returns HTTP 409 instead of abandoning live work — closing the + duplicate-work window. Unverifiable identities (old workers, remote hosts) + keep the pre-existing behavior; the wire change is backward compatible. +- **Orphaned-agent recovery (P2.5 / TH-D5 consumer side).** When a worker is + confirmed dead with nothing recoverable, the coordinator applies + `recovery_policy` (new config; default `drain`, bounded by + `recovery_drain_timeout_s`, default 600s) to agent processes the dead + worker's harness run left behind, via team-harness's process reaper: drained + agents finish and their repo edits survive; a `salvage.json` in the + interrupted iteration directory records what was handled; the history entry + is `abandoned_after_drain` instead of `abandoned`. Requires team-harness + with the process reaper (> 0.2.10); older versions skip orphan recovery + gracefully. team-harness's own parent-liveness guard surfaces as HTTP 409. +- The bundled worker uses an unbounded read timeout (recovery can legitimately + block `/register` up to the drain deadline) and exits with code 3 on a 409. + ## 0.2.1 - Improve README onboarding, install, initialization, configuration, and logging docs. diff --git a/README.md b/README.md index 6fa6046..10f0832 100644 --- a/README.md +++ b/README.md @@ -229,6 +229,10 @@ Important rules: - `team_harness_max_retries`, `team_harness_retry_base_delay_s`, and `team_harness_retry_max_delay_s` are optional retry controls for transient team-harness API/network errors. +- `recovery_policy` (`drain` by default, or `reap`) and + `recovery_drain_timeout_s` control what crash recovery does with agent + processes a dead worker left running: drain lets them finish (bounded) + before the iteration is re-run; reap kills them immediately. Workflow config lives beside each workflow prompt: diff --git a/docs/http-contract.md b/docs/http-contract.md index 2afb850..626b523 100644 --- a/docs/http-contract.md +++ b/docs/http-contract.md @@ -4,7 +4,23 @@ loopy-loop exposes exactly two coordinator endpoints. ## POST /register -Request: `{}` (empty body) +Request: `{}` (empty body), or with the worker's process identity: + +```json +{ + "worker": { + "hostname": "buildbox", + "pid": 4242, + "starttime": "lstart:Sun Jul 12 00:00:00 2026" + } +} +``` + +`worker` is optional and backward compatible. When present, the coordinator +stamps it onto the dispatched task so a later `/register` can *verify* whether +that worker is still alive before reclaiming its task. `starttime` is +team-harness's pid-reuse-proof process-identity token (requires team-harness +with process identity support; omitted otherwise). Run response: @@ -33,7 +49,9 @@ Run response: "team_harness_retry_max_delay_s": null, "team_harness_api_base": "https://openrouter.ai/api/v1", "team_harness_api_key_env": "OPENROUTER_API_KEY", - "team_harness_system_prompt_extension": "" + "team_harness_system_prompt_extension": "", + "recovery_policy": "drain", + "recovery_drain_timeout_s": 600.0 }, "stop_reason": null } @@ -62,11 +80,27 @@ Rules: `.loopy_loop/workflow_sets//workflows//` directory to load. - If `current_task` is already set (previous worker crashed without calling - `/finished`), `/register` first checks the current iteration directory for - `pending_finished_request.json` or `result.json`. If either file proves the - task completed, the coordinator records the completed task in history before - checking stop conditions. Only tasks with no recoverable local completion are - recorded as failed with `error="abandoned"`. + `/finished`), `/register` proceeds in three steps: + 1. **Liveness check.** If the recorded worker identity is *verifiably still + alive* (same host, matching pid + starttime), the call is refused with + **HTTP 409** and no state is mutated — the task is not abandoned and no + duplicate work is dispatched. Unverifiable identities (no identity + recorded, remote host, no starttime token) fall through. + 2. **Result recovery.** The coordinator checks the current iteration + directory for `pending_finished_request.json` or `result.json`. If either + file proves the task completed, the completed task is recorded in history + before checking stop conditions. + 3. **Orphan recovery.** With nothing recoverable, the coordinator applies the + configured recovery policy (`recovery_policy`, default `drain` bounded by + `recovery_drain_timeout_s`) to any agent processes the dead worker's + harness run left behind, writes a `salvage.json` into the interrupted + iteration directory when something was handled, and records the iteration + as failed with `error="abandoned_after_drain"` (or plain `"abandoned"` + when nothing was reaped). Requires team-harness with the process reaper; + older versions skip this step. If team-harness's own guard reports the + run's owning process still alive, `/register` returns **HTTP 409**. + Note: draining can block `/register` for up to `recovery_drain_timeout_s`; + the bundled worker uses an unbounded read timeout for this reason. - If the loop is in a terminal state, `/register` immediately returns `action=stop`. ## POST /finished @@ -80,10 +114,18 @@ Request: "iteration": 3, "success": true, "text": "done", - "error": null + "error": null, + "worker": { + "hostname": "buildbox", + "pid": 4242, + "starttime": "lstart:Sun Jul 12 00:00:00 2026" + } } ``` +`worker` is optional (same semantics as `/register`): the calling worker will +run the next dispatched task, so its identity is stamped onto that task. + Response: same shape as `/register` response (`action` is either `"run"` or `"stop"`). Rules: diff --git a/docs/session-layout.md b/docs/session-layout.md index 0c2acfa..73427e6 100644 --- a/docs/session-layout.md +++ b/docs/session-layout.md @@ -165,6 +165,20 @@ eval-banana run \ - If the file is missing but `result.json` exists for the active task, the coordinator can reconstruct the finished request from `result.json` +`salvage.json` + +- Written into the interrupted iteration's directory during crash recovery, + when the coordinator applied the recovery policy (`recovery_policy`, default + bounded drain) to agent processes a dead worker's harness run left behind +- Records the reap reports: which orphaned agents were drained (allowed to + finish), reaped (killed), or skipped, so the provenance of any surviving + working-tree edits is auditable rather than a mystery diff +- The iteration is still re-run — its `result.json` never existed and is never + fabricated; the corresponding history entry carries + `error="abandoned_after_drain"` instead of plain `"abandoned"` +- Schema: `{"schema_version": 1, "recorded_at": ..., "policy": ..., + "reaped_runs": N, "settled_workers": N, "reports": [...]}` + ## Control Contracts `control.json` diff --git a/src/loopy_loop/config.py b/src/loopy_loop/config.py index 07d55ac..8c4a28e 100644 --- a/src/loopy_loop/config.py +++ b/src/loopy_loop/config.py @@ -4,6 +4,7 @@ import os from pathlib import Path from typing import Any +from typing import Literal from pydantic import BaseModel from pydantic import computed_field @@ -131,6 +132,23 @@ class RootConfig(BaseModel): default=DEFAULT_SYSTEM_PROMPT_EXTENSION, description="Additional system prompt text appended for every harness run.", ) + recovery_policy: Literal["drain", "reap"] = Field( + default="drain", + description=( + "What to do with agent processes orphaned by a crashed worker: " + "'drain' lets in-flight agents finish (bounded by " + "recovery_drain_timeout_s) before re-running the iteration; " + "'reap' kills them immediately." + ), + ) + recovery_drain_timeout_s: float = Field( + default=600.0, + ge=0, + description=( + "Shared deadline (seconds) for draining orphaned agents during " + "crash recovery before they are killed." + ), + ) @computed_field @property diff --git a/src/loopy_loop/coordinator_app.py b/src/loopy_loop/coordinator_app.py index c03e325..9bde214 100644 --- a/src/loopy_loop/coordinator_app.py +++ b/src/loopy_loop/coordinator_app.py @@ -8,6 +8,7 @@ from typing import TypeVar from fastapi import FastAPI +from fastapi import HTTPException from pydantic import BaseModel from loopy_loop.config import ConfigError @@ -24,10 +25,15 @@ from loopy_loop.models import HistoryEntry from loopy_loop.models import IterationResult from loopy_loop.models import LoopState +from loopy_loop.models import RegisterRequest from loopy_loop.models import RootConfigSnapshot from loopy_loop.models import STOP_ACTION from loopy_loop.models import TaskResponse from loopy_loop.models import utc_now +from loopy_loop.models import WorkerIdentity +from loopy_loop.recovery import recover_interrupted_iteration +from loopy_loop.recovery import RecoveryOutcome +from loopy_loop.recovery import RecoveryRefusedError from loopy_loop.scheduler import choose_next_workflow from loopy_loop.sessions import child_requests_dir_path from loopy_loop.sessions import children_path @@ -39,10 +45,15 @@ from loopy_loop.sessions import result_path from loopy_loop.sessions import state_path from loopy_loop.state_store import StateStore +from loopy_loop.worker_identity import is_worker_alive logger = logging.getLogger(__name__) +class WorkerBusyError(RuntimeError): + """The current task's worker is verifiably still alive — do not reclaim.""" + + def create_coordinator_app( *, repo_root: Path, @@ -61,8 +72,11 @@ def create_coordinator_app( app.state.service = service @app.post("/register", response_model=TaskResponse) - def register_worker() -> TaskResponse: - return service.register_worker() + def register_worker(request: RegisterRequest | None = None) -> TaskResponse: + try: + return service.register_worker(request=request) + except (WorkerBusyError, RecoveryRefusedError) as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc @app.post("/finished", response_model=TaskResponse) def finish_assignment(request: FinishedRequest) -> TaskResponse: @@ -88,8 +102,11 @@ def __init__( self.state_store = state_store self._prepare_state(resume=resume) - def register_worker(self) -> TaskResponse: + def register_worker( + self, *, request: RegisterRequest | None = None + ) -> TaskResponse: recovered_pending_paths: list[Path] = [] + caller = request.worker if request is not None else None def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: current = _require_state(state=state) @@ -101,10 +118,26 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: # behaviour. The stop check (step 4) will catch it immediately after. if current.current_task is not None: orphaned = current.current_task + # Verified-alive worker: its task is NOT abandoned. Refuse this + # register instead of dispatching duplicate work (D7). Unknown + # (None) falls through to the pre-existing recovery behavior. + if is_worker_alive(orphaned.worker) is True: + raise WorkerBusyError( + f"worker pid={orphaned.worker.pid} on " # type: ignore[union-attr] + f"{orphaned.worker.hostname} is still running iteration " # type: ignore[union-attr] + f"{orphaned.iteration} ({orphaned.workflow_id}); " + "refusing to dispatch duplicate work" + ) recovered = self._read_recoverable_finished_request( current_task=orphaned ) if recovered is None: + # The worker is dead with nothing recoverable: apply the + # recovery policy to any agent processes it orphaned + # (default bounded drain — TH-D5/P2.5), then re-run. + # May raise RecoveryRefusedError -> 409 if team-harness's + # parent-liveness guard says the run's owner still lives. + recovery = self._recover_orphaned_agents(current_task=orphaned) current.history.append( HistoryEntry( iteration=orphaned.iteration, @@ -112,7 +145,9 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: workflow_id=orphaned.workflow_id, session_id=orphaned.session_id, success=False, - error="abandoned", + error="abandoned_after_drain" + if recovery.salvaged + else "abandoned", started_at=orphaned.started_at, finished_at=now, ) @@ -135,7 +170,9 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: if stop_response is not None: return current, stop_response - child_response = self._dispatch_child_session_after_success(state=current) + child_response = self._dispatch_child_session_after_success( + state=current, caller=caller + ) if child_response is not None: return current, child_response @@ -160,6 +197,7 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: session_id=current.active_session_id, iteration=current.iteration_count + 1, started_at=now, + worker=caller, ) return current, _build_run_response( current_task=current.current_task, @@ -172,6 +210,8 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: return response def finish_assignment(self, *, request: FinishedRequest) -> TaskResponse: + caller = request.worker + def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: current = _require_state(state=state) now = utc_now() @@ -184,7 +224,7 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: if stop_response is not None: return current, stop_response child_response = self._dispatch_child_session_after_success( - state=current + state=current, caller=caller ) if child_response is not None: return current, child_response @@ -206,6 +246,7 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: session_id=current.active_session_id, iteration=current.iteration_count + 1, started_at=now, + worker=caller, ) return current, _build_run_response( current_task=current.current_task, @@ -244,7 +285,9 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: return current, stop_response # Step 8: Dispatch next workflow. - child_response = self._dispatch_child_session_after_success(state=current) + child_response = self._dispatch_child_session_after_success( + state=current, caller=caller + ) if child_response is not None: return current, child_response @@ -268,6 +311,7 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: session_id=current.active_session_id, iteration=current.iteration_count + 1, started_at=now, + worker=caller, ) return current, _build_run_response( current_task=current.current_task, @@ -276,7 +320,9 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: response = self.state_store.mutate(mutator) if response.action == STOP_ACTION: - parent_response = self._resume_parent_if_active_child_completed() + parent_response = self._resume_parent_if_active_child_completed( + caller=caller + ) if parent_response is not None: return parent_response return response @@ -330,6 +376,23 @@ def _record_finished_task( state.iteration_count += 1 state.current_task = None + def _recover_orphaned_agents(self, *, current_task: CurrentTask) -> RecoveryOutcome: + """Apply the configured recovery policy to a dead worker's orphans. + + Runs while the state lock is held: nothing else may be dispatched + until recovery settles (draining can take up to the configured + timeout — the single-worker model makes blocking here correct). + """ + config = self.preflight.root_config + return recover_interrupted_iteration( + repo_root=self.repo_root, + session_id=current_task.session_id, + iteration=current_task.iteration, + workflow_id=current_task.workflow_id, + policy=config.recovery_policy, + drain_timeout_s=config.recovery_drain_timeout_s, + ) + def _read_recoverable_finished_request( self, *, current_task: CurrentTask ) -> tuple[FinishedRequest, Path | None] | None: @@ -449,7 +512,7 @@ def _workflows_by_id_for( } def _dispatch_child_session_if_requested( - self, *, state: LoopState + self, *, state: LoopState, caller: WorkerIdentity | None = None ) -> TaskResponse | None: if state.parent_session_id is not None: return None @@ -499,6 +562,7 @@ def _dispatch_child_session_if_requested( session_id=child_session_id, iteration=1, started_at=now, + worker=caller, ), ) child_store = StateStore( @@ -529,13 +593,15 @@ def _dispatch_child_session_if_requested( return None def _dispatch_child_session_after_success( - self, *, state: LoopState + self, *, state: LoopState, caller: WorkerIdentity | None = None ) -> TaskResponse | None: if not state.history or not state.history[-1].success: return None - return self._dispatch_child_session_if_requested(state=state) + return self._dispatch_child_session_if_requested(state=state, caller=caller) - def _resume_parent_if_active_child_completed(self) -> TaskResponse | None: + def _resume_parent_if_active_child_completed( + self, *, caller: WorkerIdentity | None = None + ) -> TaskResponse | None: child_state = self.state_store.read_state() if child_state is None or child_state.parent_session_id is None: return None @@ -549,7 +615,7 @@ def _resume_parent_if_active_child_completed(self) -> TaskResponse | None: ), ) self.state_store = parent_store - return self.register_worker() + return self.register_worker(request=RegisterRequest(worker=caller)) def _append_child_record( self, *, parent_session_id: str, record: ChildSessionRecord diff --git a/src/loopy_loop/models.py b/src/loopy_loop/models.py index c36cf07..356c2ae 100644 --- a/src/loopy_loop/models.py +++ b/src/loopy_loop/models.py @@ -43,6 +43,28 @@ class RootConfigSnapshot(BaseModel): team_harness_api_base: str = Field(...) team_harness_api_key_env: str = Field(...) team_harness_system_prompt_extension: str = Field(...) + recovery_policy: Literal["drain", "reap"] = Field(default="drain") + recovery_drain_timeout_s: float = Field(default=600.0) + + +class WorkerIdentity(BaseModel): + """Durable identity of the worker process holding an assignment. + + Lets the coordinator *verify* whether that worker is still alive before + reclaiming its task (instead of assuming abandonment), closing the + duplicate-work window on a second /register. ``starttime`` is the + team-harness process-identity token (pid-reuse-proof); verification is + only possible on the coordinator's own host — remote workers fall back + to the old assume-abandoned behavior. + """ + + hostname: str = Field(...) + pid: int = Field(...) + starttime: str | None = Field(default=None) + + +class RegisterRequest(BaseModel): + worker: WorkerIdentity | None = Field(default=None) class CurrentTask(BaseModel): @@ -51,6 +73,7 @@ class CurrentTask(BaseModel): session_id: str = Field(...) iteration: int = Field(...) started_at: datetime = Field(...) + worker: WorkerIdentity | None = Field(default=None) class TaskResponse(BaseModel): @@ -101,6 +124,9 @@ class FinishedRequest(BaseModel): success: bool = Field(...) text: str | None = Field(default=None) error: str | None = Field(default=None) + # Identity of the calling worker — the same worker will run the NEXT task + # this response dispatches, so it is stamped onto that CurrentTask. + worker: WorkerIdentity | None = Field(default=None) class ControlSignal(BaseModel): diff --git a/src/loopy_loop/recovery.py b/src/loopy_loop/recovery.py new file mode 100644 index 0000000..a5de578 --- /dev/null +++ b/src/loopy_loop/recovery.py @@ -0,0 +1,191 @@ +"""Crash recovery for a dead worker's orphaned agent processes (D7 / P2.5). + +When the coordinator confirms a worker is dead and its iteration produced no +recoverable result, the agent CLIs that worker's harness spawned may still be +running — orphaned, spending money, writing to the checkout. team-harness owns +the mechanism (persisted process identity + drain/reap policies, TH-D5); this +module is the loopy-side trigger: + +1. **Discover** the interrupted harness run(s): team-harness routes each run's + session output under the iteration's ``harness_outputs`` directory, named by + run id, so the run ids are the directory names — and each run's crash-durable + ``run.json`` lives under team-harness's runs dir. +2. **Apply the recovery policy** via ``team_harness.tracking.reaper.reap_run``: + ``drain`` (default — let in-flight agents finish within a shared bounded + timeout, preserving near-complete work and a clean tree) or ``reap`` (kill). +3. **Record the salvage**: a ``salvage.json`` in the interrupted iteration's + directory carrying the reap reports, so the provenance of any surviving + working-tree edits is auditable (the iteration itself is still re-run — its + ``result.json`` never existed and is never fabricated; loopy `decisions.md` + D3/D7). + +team-harness versions without the reaper are tolerated: recovery degrades to +the pre-existing behavior (mark abandoned, re-run) with nothing reaped. +``ReapRefusedError`` — team-harness's own parent-liveness guard — bubbles up so +the coordinator can treat "the run's owner is still alive" as a busy signal. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from dataclasses import field +import importlib +import json +import logging +from pathlib import Path +from typing import Any + +from loopy_loop.models import utc_now +from loopy_loop.sessions import iteration_dir_path +from loopy_loop.sessions import iteration_harness_output_root + +logger = logging.getLogger(__name__) + +SALVAGE_FILENAME = "salvage.json" +SALVAGE_SCHEMA_VERSION = 1 + +# Outcomes that mean the orphaned work reached a settled end state (finished on +# its own, or was killed) — i.e. something real was handled during recovery. +_SETTLED_OUTCOMES = { + "drained", + "already_exited", + "reaped", + "drain_timed_out_then_reaped", +} + + +class RecoveryRefusedError(RuntimeError): + """team-harness refused to reap: the run's owning process is still alive. + + Loopy-owned wrapper for team-harness's ``ReapRefusedError`` so callers + have a stable type to catch regardless of the installed harness version. + """ + + +def _load_reaper() -> tuple[Any, Any] | None: + """Resolve team-harness's reaper at call time (needs team-harness >= 0.2.11). + + Runtime resolution (not a static import) so loopy-loop keeps working — + minus orphan recovery — against older team-harness versions. + """ + try: + reaper = importlib.import_module("team_harness.tracking.reaper") + th_config = importlib.import_module("team_harness.config") + except ImportError: + return None + return reaper, th_config + + +@dataclass +class RecoveryOutcome: + """What crash recovery did about a dead worker's orphaned agents.""" + + reaped_runs: int = 0 + settled_workers: int = 0 + reports: list[dict[str, Any]] = field(default_factory=list) + + @property + def salvaged(self) -> bool: + return self.settled_workers > 0 + + +def recover_interrupted_iteration( + *, + repo_root: Path, + session_id: str, + iteration: int, + workflow_id: str, + policy: str, + drain_timeout_s: float, +) -> RecoveryOutcome: + """Drain/reap the interrupted iteration's orphaned agents; write salvage.json. + + Raises ``RecoveryRefusedError`` when team-harness's parent-liveness guard + finds the run's owning process still alive — the caller should treat that + as "the previous worker is still running". + """ + outcome = RecoveryOutcome() + loaded = _load_reaper() + if loaded is None: + logger.warning( + "team-harness has no process reaper (needs >= 0.2.11); " + "skipping orphan recovery for iteration %04d_%s", + iteration, + workflow_id, + ) + return outcome + reaper, th_config = loaded + output_root = iteration_harness_output_root( + repo_root=repo_root, + session_id=session_id, + iteration=iteration, + workflow_id=workflow_id, + ) + for run_id in _discover_run_ids(output_root): + run_json = Path(th_config.RUNS_DIR) / run_id / "run.json" + if not run_json.exists(): + logger.warning("no run.json for interrupted harness run %s", run_id) + continue + try: + report = reaper.reap_run( + run_json, policy=policy, drain_timeout_s=drain_timeout_s + ) + except reaper.ReapRefusedError as exc: + raise RecoveryRefusedError(str(exc)) from exc + outcome.reaped_runs += 1 + outcome.settled_workers += sum( + 1 for worker in report.workers if worker.outcome in _SETTLED_OUTCOMES + ) + outcome.reports.append(report.model_dump(mode="json")) + if outcome.reaped_runs: + _write_salvage_record( + repo_root=repo_root, + session_id=session_id, + iteration=iteration, + workflow_id=workflow_id, + outcome=outcome, + policy=policy, + ) + return outcome + + +def _discover_run_ids(output_root: Path) -> list[str]: + """The iteration's harness output root contains one directory per run id.""" + if not output_root.is_dir(): + return [] + return sorted(entry.name for entry in output_root.iterdir() if entry.is_dir()) + + +def _write_salvage_record( + *, + repo_root: Path, + session_id: str, + iteration: int, + workflow_id: str, + outcome: RecoveryOutcome, + policy: str, +) -> None: + """Make the salvage auditable: which orphans were handled, and how. + + The iteration is still re-run (D3: its result.json never existed and is + never fabricated); surviving working-tree edits are explained by this + record instead of appearing as a mystery diff. + """ + iteration_dir = iteration_dir_path( + repo_root=repo_root, + session_id=session_id, + iteration=iteration, + workflow_id=workflow_id, + ) + iteration_dir.mkdir(parents=True, exist_ok=True) + payload = { + "schema_version": SALVAGE_SCHEMA_VERSION, + "recorded_at": utc_now().isoformat().replace("+00:00", "Z"), + "policy": policy, + "reaped_runs": outcome.reaped_runs, + "settled_workers": outcome.settled_workers, + "reports": outcome.reports, + } + (iteration_dir / SALVAGE_FILENAME).write_text( + json.dumps(payload, indent=2), encoding="utf-8" + ) diff --git a/src/loopy_loop/worker.py b/src/loopy_loop/worker.py index 50bf832..e55a1d7 100644 --- a/src/loopy_loop/worker.py +++ b/src/loopy_loop/worker.py @@ -16,8 +16,10 @@ from loopy_loop.harness_runner import write_iteration_artifacts from loopy_loop.models import FinishedRequest from loopy_loop.models import IterationResult +from loopy_loop.models import RegisterRequest from loopy_loop.models import RootConfigSnapshot from loopy_loop.models import TaskResponse +from loopy_loop.models import WorkerIdentity from loopy_loop.sessions import child_requests_dir_path from loopy_loop.sessions import control_path from loopy_loop.sessions import ensure_iteration_dir @@ -31,6 +33,7 @@ from loopy_loop.sessions import session_dir_path from loopy_loop.sessions import session_goal_path from loopy_loop.sessions import updates_from_user_path +from loopy_loop.worker_identity import current_worker_identity # Internal retry constants for /finished — not configurable externally. # If all retries fail, the exception propagates and the process exits; @@ -55,11 +58,20 @@ class FinishedAssignment: def run_worker_loop(*, repo_root: Path, coordinator_url: str) -> None: base_url = coordinator_url.rstrip("/") - with httpx.Client(timeout=30.0) as client: - task = _post_register(client=client, coordinator_url=base_url) + # The read timeout is unbounded because /register may legitimately block + # while the coordinator drains a crashed predecessor's orphaned agents + # (bounded by the coordinator's recovery_drain_timeout_s, not by us). + timeout = httpx.Timeout(30.0, read=None) + identity = current_worker_identity() + with httpx.Client(timeout=timeout) as client: + task = _post_register( + client=client, coordinator_url=base_url, identity=identity + ) while task.action == "run": try: - finished_assignment = _run_task(repo_root=repo_root, task=task) + finished_assignment = _run_task( + repo_root=repo_root, task=task, identity=identity + ) except FatalAssignmentError as exc: print(str(exc), file=sys.stderr) _post_finished( @@ -79,12 +91,28 @@ def run_worker_loop(*, repo_root: Path, coordinator_url: str) -> None: _clear_pending_finished_request(path=finished_assignment.pending_path) -def _post_register(*, client: httpx.Client, coordinator_url: str) -> TaskResponse: - response = client.post(f"{coordinator_url}/register", json={}) +def _post_register( + *, client: httpx.Client, coordinator_url: str, identity: WorkerIdentity +) -> TaskResponse: + request = RegisterRequest(worker=identity) + response = client.post(f"{coordinator_url}/register", json=request.model_dump()) + _exit_if_busy(response) response.raise_for_status() return TaskResponse.model_validate(response.json()) +def _exit_if_busy(response: httpx.Response) -> None: + """409 means another worker verifiably holds the current task (D7).""" + if response.status_code != 409: + return + try: + detail = response.json().get("detail", response.text) + except ValueError: + detail = response.text + print(f"Coordinator refused this worker: {detail}", file=sys.stderr) + sys.exit(3) + + def _post_finished( *, client: httpx.Client, coordinator_url: str, request: FinishedRequest ) -> TaskResponse: @@ -103,7 +131,9 @@ def _post_finished( raise RuntimeError("unreachable") -def _run_task(*, repo_root: Path, task: TaskResponse) -> FinishedAssignment: +def _run_task( + *, repo_root: Path, task: TaskResponse, identity: WorkerIdentity | None = None +) -> FinishedAssignment: if ( task.session_id is None or task.workflow_set is None @@ -178,6 +208,7 @@ def _run_task(*, repo_root: Path, task: TaskResponse) -> FinishedAssignment: success=iteration_result.success, text=iteration_result.text, error=iteration_result.error, + worker=identity, ) pending_path = _write_pending_finished_request( repo_root=repo_root, request=finished_request diff --git a/src/loopy_loop/worker_identity.py b/src/loopy_loop/worker_identity.py new file mode 100644 index 0000000..99b4774 --- /dev/null +++ b/src/loopy_loop/worker_identity.py @@ -0,0 +1,64 @@ +"""Worker process identity: capture and verification (loopy side of TH-D5/D7). + +The worker sends its identity (hostname + pid + a pid-reuse-proof start-time +token) with every /register and /finished call; the coordinator stamps it onto +the dispatched ``CurrentTask``. On a later /register while that task is still +current, the coordinator can then *verify* whether the recorded worker is +actually dead before reclaiming the task — instead of assuming abandonment — +closing the duplicate-work window (design: loopy `decisions.md` D7, proposals +P0.1). + +The start-time token comes from team-harness's ``process_identity`` module +(the same mechanism its orphan reaper uses). team-harness versions without it +are tolerated: identity then carries no starttime and verification degrades to +"unknown", which preserves the pre-existing assume-abandoned behavior. +Verification is only meaningful on the coordinator's own host; identities from +another hostname are likewise "unknown". +""" + +from __future__ import annotations + +import importlib +import os +import socket + +from loopy_loop.models import WorkerIdentity + + +def capture_starttime(pid: int) -> str | None: + """team-harness's pid-reuse-proof start-time token; None if unavailable. + + Resolved at call time so loopy-loop keeps working with team-harness + versions that predate process identity (< 0.2.11). + """ + try: + module = importlib.import_module("team_harness.agents.process_identity") + except ImportError: + return None + return module.capture_starttime(pid) + + +def current_worker_identity() -> WorkerIdentity: + """Identity of THIS process, captured for /register and /finished calls.""" + pid = os.getpid() + return WorkerIdentity( + hostname=socket.gethostname(), pid=pid, starttime=capture_starttime(pid) + ) + + +def is_worker_alive(identity: WorkerIdentity | None) -> bool | None: + """Verified liveness of a recorded worker identity. + + Returns True/False only when the answer is *verified*: same host, a + recorded start-time token, and a current capture that matches (alive) or + differs/is absent (dead — the pid is gone or was recycled). Returns None + when verification is impossible (no identity recorded, remote host, or no + token) — callers must treat None as "unknown" and fall back to the + pre-existing recovery behavior, never as "alive". + """ + if identity is None or identity.starttime is None: + return None + if identity.hostname != socket.gethostname(): + return None + current = capture_starttime(identity.pid) + return current == identity.starttime diff --git a/src/tests/test_worker.py b/src/tests/test_worker.py index 5307081..7c276d1 100644 --- a/src/tests/test_worker.py +++ b/src/tests/test_worker.py @@ -46,8 +46,12 @@ def _make_stop_response(*, stop_reason: str = "goal_met") -> dict[str, object]: class FakeResponse: - def __init__(self, payload: dict[str, object]) -> None: + def __init__( + self, payload: dict[str, object], status_code: int = 200, text: str = "" + ) -> None: self._payload = payload + self.status_code = status_code + self.text = text def raise_for_status(self) -> None: return None diff --git a/src/tests/test_worker_liveness_recovery.py b/src/tests/test_worker_liveness_recovery.py new file mode 100644 index 0000000..d0768d8 --- /dev/null +++ b/src/tests/test_worker_liveness_recovery.py @@ -0,0 +1,540 @@ +"""Tests for worker liveness verification and orphan recovery (D7 / P0.1 / P2.5). + +The worker sends its process identity with /register and /finished; the +coordinator stamps it onto the dispatched CurrentTask, refuses (409) to reclaim +a task whose worker is verifiably alive, and — on a confirmed-dead worker with +nothing recoverable — applies the recovery policy to orphaned agent processes +via team-harness before re-running the iteration. +""" + +from __future__ import annotations + +import json +from pathlib import Path +import socket +from typing import Any + +from fastapi.testclient import TestClient +import pytest + +from loopy_loop import recovery as recovery_module +from loopy_loop import worker_identity as worker_identity_module +from loopy_loop.coordinator_app import create_coordinator_app +from loopy_loop.models import RegisterRequest +from loopy_loop.models import WorkerIdentity +from loopy_loop.recovery import recover_interrupted_iteration +from loopy_loop.recovery import RecoveryOutcome +from loopy_loop.recovery import RecoveryRefusedError +from loopy_loop.recovery import SALVAGE_FILENAME +from loopy_loop.state_store import StateStore +from loopy_loop.worker import run_worker_loop +from loopy_loop.worker_identity import current_worker_identity +from loopy_loop.worker_identity import is_worker_alive + +LOCAL_IDENTITY = { + "hostname": socket.gethostname(), + "pid": 4242, + "starttime": "lstart:Sun Jul 12 00:00:00 2026", +} + + +# --------------------------------------------------------------------------- +# worker_identity unit behavior +# --------------------------------------------------------------------------- + + +def test_current_worker_identity_carries_pid_and_hostname() -> None: + identity = current_worker_identity() + assert identity.pid > 0 + assert identity.hostname == socket.gethostname() + # starttime may be None when team-harness predates process identity. + + +def test_is_worker_alive_unknown_cases() -> None: + assert is_worker_alive(None) is None + no_token = WorkerIdentity(hostname=socket.gethostname(), pid=1, starttime=None) + assert is_worker_alive(no_token) is None + remote = WorkerIdentity(hostname="another-host", pid=1, starttime="lstart:x") + assert is_worker_alive(remote) is None + + +def test_is_worker_alive_verified_true_and_false(monkeypatch: Any) -> None: + identity = WorkerIdentity(**LOCAL_IDENTITY) + monkeypatch.setattr( + worker_identity_module, + "capture_starttime", + lambda pid: LOCAL_IDENTITY["starttime"], + ) + assert is_worker_alive(identity) is True + monkeypatch.setattr(worker_identity_module, "capture_starttime", lambda pid: None) + assert is_worker_alive(identity) is False # pid gone + monkeypatch.setattr( + worker_identity_module, "capture_starttime", lambda pid: "lstart:other" + ) + assert is_worker_alive(identity) is False # pid recycled + + +# --------------------------------------------------------------------------- +# Coordinator: verified-alive worker -> 409, no state mutation +# --------------------------------------------------------------------------- + + +def _register_with_orphan( + repo_builder: Any, + current_task_factory: Any, + monkeypatch: Any, + *, + orphan_worker: dict[str, Any] | None, + liveness: bool | None, + recovery_outcome: RecoveryOutcome | None = None, +) -> tuple[Any, StateStore, Any]: + """Build a repo with an orphaned current_task and post /register.""" + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder( + workflows={ + "planner": { + "prompt": "Plan work.", + "config": { + "enabled": True, + "run_every": 1, + "must_follow": None, + "not_before_iteration": 0, + "description": "Plan", + }, + }, + "implement": { + "prompt": "Implement.", + "config": { + "enabled": True, + "run_every": 1, + "must_follow": None, + "not_before_iteration": 0, + "description": "Implement", + }, + }, + } + ) + client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) + store = StateStore(repo_root=repo_root) + state = store.read_state() + assert state is not None + overrides: dict[str, Any] = { + "workflow_id": "implement", + "session_id": state.active_session_id, + "iteration": 1, + } + if orphan_worker is not None: + overrides["worker"] = orphan_worker + state.current_task = current_task_factory(**overrides) + store.write_state(state=state) + monkeypatch.setattr( + "loopy_loop.coordinator_app.is_worker_alive", lambda identity: liveness + ) + monkeypatch.setattr( + "loopy_loop.coordinator_app.recover_interrupted_iteration", + lambda **kwargs: recovery_outcome or RecoveryOutcome(), + ) + response = client.post("/register", json={}) + return response, store, state + + +def test_register_refuses_when_worker_verifiably_alive( + repo_builder: Any, monkeypatch: Any, current_task_factory: Any +) -> None: + response, store, _ = _register_with_orphan( + repo_builder, + current_task_factory, + monkeypatch, + orphan_worker=LOCAL_IDENTITY, + liveness=True, + ) + assert response.status_code == 409 + assert "still running" in response.json()["detail"] + updated = store.read_state() + assert updated is not None + assert updated.current_task is not None # untouched + assert updated.history == [] # nothing abandoned + + +def test_register_recovers_when_worker_verifiably_dead( + repo_builder: Any, monkeypatch: Any, current_task_factory: Any +) -> None: + response, store, _ = _register_with_orphan( + repo_builder, + current_task_factory, + monkeypatch, + orphan_worker=LOCAL_IDENTITY, + liveness=False, + ) + assert response.status_code == 200 + assert response.json()["action"] == "run" + updated = store.read_state() + assert updated is not None + assert updated.history[0].error == "abandoned" + + +def test_register_recovers_when_liveness_unknown( + repo_builder: Any, monkeypatch: Any, current_task_factory: Any +) -> None: + # Old workers / remote hosts: no identity -> pre-existing behavior. + response, store, _ = _register_with_orphan( + repo_builder, + current_task_factory, + monkeypatch, + orphan_worker=None, + liveness=None, + ) + assert response.status_code == 200 + updated = store.read_state() + assert updated is not None + assert updated.history[0].error == "abandoned" + + +def test_register_records_abandoned_after_drain_when_salvaged( + repo_builder: Any, monkeypatch: Any, current_task_factory: Any +) -> None: + response, store, _ = _register_with_orphan( + repo_builder, + current_task_factory, + monkeypatch, + orphan_worker=LOCAL_IDENTITY, + liveness=False, + recovery_outcome=RecoveryOutcome(reaped_runs=1, settled_workers=2), + ) + assert response.status_code == 200 + updated = store.read_state() + assert updated is not None + assert updated.history[0].error == "abandoned_after_drain" + + +def test_register_surfaces_recovery_refused_as_409( + repo_builder: Any, monkeypatch: Any, current_task_factory: Any +) -> None: + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder() + client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) + store = StateStore(repo_root=repo_root) + state = store.read_state() + assert state is not None + state.current_task = current_task_factory( + workflow_id="planner", session_id=state.active_session_id, iteration=1 + ) + store.write_state(state=state) + + def refuse(**kwargs: Any) -> RecoveryOutcome: + raise RecoveryRefusedError("run owner still alive") + + monkeypatch.setattr( + "loopy_loop.coordinator_app.recover_interrupted_iteration", refuse + ) + response = client.post("/register", json={}) + assert response.status_code == 409 + assert "still alive" in response.json()["detail"] + updated = store.read_state() + assert updated is not None + assert updated.current_task is not None # state untouched + + +def test_register_stamps_worker_identity_on_dispatched_task( + repo_builder: Any, monkeypatch: Any +) -> None: + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder() + client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) + body = RegisterRequest(worker=WorkerIdentity(**LOCAL_IDENTITY)).model_dump() + response = client.post("/register", json=body) + assert response.status_code == 200 + state = StateStore(repo_root=repo_root).read_state() + assert state is not None + assert state.current_task is not None + assert state.current_task.worker is not None + assert state.current_task.worker.pid == LOCAL_IDENTITY["pid"] + + +def test_finished_stamps_callers_identity_on_next_task( + repo_builder: Any, monkeypatch: Any +) -> None: + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder() + client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) + body = RegisterRequest(worker=WorkerIdentity(**LOCAL_IDENTITY)).model_dump() + task = client.post("/register", json=body).json() + finished = { + "workflow_id": task["workflow_id"], + "session_id": task["session_id"], + "iteration": task["iteration"], + "success": True, + "text": "done", + "worker": LOCAL_IDENTITY, + } + next_task = client.post("/finished", json=finished).json() + assert next_task["action"] == "run" + state = StateStore(repo_root=repo_root).read_state() + assert state is not None + assert state.current_task is not None + assert state.current_task.worker is not None + assert state.current_task.worker.pid == LOCAL_IDENTITY["pid"] + + +def test_register_without_body_still_works(repo_builder: Any, monkeypatch: Any) -> None: + # Pre-identity workers post an empty body; the contract stays compatible. + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder() + client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) + response = client.post("/register", json={}) + assert response.status_code == 200 + assert response.json()["action"] == "run" + + +# --------------------------------------------------------------------------- +# recovery module +# --------------------------------------------------------------------------- + + +class _FakeWorkerOutcome: + def __init__(self, outcome: str) -> None: + self.outcome = outcome + + +class _FakeReport: + def __init__(self, outcomes: list[str]) -> None: + self.workers = [_FakeWorkerOutcome(outcome) for outcome in outcomes] + + def model_dump(self, mode: str = "json") -> dict[str, Any]: + return {"workers": [worker.outcome for worker in self.workers]} + + +class _FakeReaperModule: + class ReapRefusedError(RuntimeError): + pass + + def __init__(self, outcomes: list[str], *, refuse: bool = False) -> None: + self.calls: list[dict[str, Any]] = [] + self._outcomes = outcomes + self._refuse = refuse + + def reap_run(self, run_json: Path, **kwargs: Any) -> _FakeReport: + self.calls.append({"run_json": run_json, **kwargs}) + if self._refuse: + raise self.ReapRefusedError("owner alive") + return _FakeReport(self._outcomes) + + +class _FakeThConfig: + def __init__(self, runs_dir: Path) -> None: + self.RUNS_DIR = runs_dir + + +def _build_interrupted_iteration( + tmp_path: Path, *, run_ids: list[str] +) -> tuple[Path, str, Path]: + session_id = "20260712_000000_deadbeef_ab12cd34" + output_root = ( + tmp_path + / ".loopy_loop" + / "sessions" + / session_id + / "harness_outputs" + / "0001_implement" + ) + for run_id in run_ids: + (output_root / run_id).mkdir(parents=True, exist_ok=True) + runs_dir = tmp_path / "th_runs" + for run_id in run_ids: + run_dir = runs_dir / run_id + run_dir.mkdir(parents=True, exist_ok=True) + (run_dir / "run.json").write_text("{}", encoding="utf-8") + return tmp_path, session_id, runs_dir + + +def test_recover_interrupted_iteration_reaps_and_writes_salvage( + tmp_path: Path, monkeypatch: Any +) -> None: + repo_root, session_id, runs_dir = _build_interrupted_iteration( + tmp_path, run_ids=["run_a", "run_b"] + ) + reaper = _FakeReaperModule(["drained", "identity_mismatch_skipped"]) + monkeypatch.setattr( + recovery_module, "_load_reaper", lambda: (reaper, _FakeThConfig(runs_dir)) + ) + outcome = recover_interrupted_iteration( + repo_root=repo_root, + session_id=session_id, + iteration=1, + workflow_id="implement", + policy="drain", + drain_timeout_s=5, + ) + assert outcome.reaped_runs == 2 + assert outcome.settled_workers == 2 # one drained per fake report + assert outcome.salvaged + assert [call["policy"] for call in reaper.calls] == ["drain", "drain"] + assert [call["drain_timeout_s"] for call in reaper.calls] == [5, 5] + salvage_path = ( + repo_root + / ".loopy_loop" + / "sessions" + / session_id + / "iterations" + / "0001_implement" + / SALVAGE_FILENAME + ) + payload = json.loads(salvage_path.read_text()) + assert payload["schema_version"] == 1 + assert payload["reaped_runs"] == 2 + assert payload["settled_workers"] == 2 + assert len(payload["reports"]) == 2 + + +def test_recover_interrupted_iteration_refusal_propagates( + tmp_path: Path, monkeypatch: Any +) -> None: + repo_root, session_id, runs_dir = _build_interrupted_iteration( + tmp_path, run_ids=["run_a"] + ) + reaper = _FakeReaperModule([], refuse=True) + monkeypatch.setattr( + recovery_module, "_load_reaper", lambda: (reaper, _FakeThConfig(runs_dir)) + ) + with pytest.raises(RecoveryRefusedError): + recover_interrupted_iteration( + repo_root=repo_root, + session_id=session_id, + iteration=1, + workflow_id="implement", + policy="drain", + drain_timeout_s=5, + ) + + +def test_recover_interrupted_iteration_without_reaper_degrades( + tmp_path: Path, monkeypatch: Any +) -> None: + repo_root, session_id, _ = _build_interrupted_iteration(tmp_path, run_ids=["run_a"]) + monkeypatch.setattr(recovery_module, "_load_reaper", lambda: None) + outcome = recover_interrupted_iteration( + repo_root=repo_root, + session_id=session_id, + iteration=1, + workflow_id="implement", + policy="drain", + drain_timeout_s=5, + ) + assert outcome.reaped_runs == 0 + assert not outcome.salvaged + + +def test_recover_interrupted_iteration_no_runs_no_salvage( + tmp_path: Path, monkeypatch: Any +) -> None: + repo_root, session_id, runs_dir = _build_interrupted_iteration(tmp_path, run_ids=[]) + reaper = _FakeReaperModule(["drained"]) + monkeypatch.setattr( + recovery_module, "_load_reaper", lambda: (reaper, _FakeThConfig(runs_dir)) + ) + outcome = recover_interrupted_iteration( + repo_root=repo_root, + session_id=session_id, + iteration=1, + workflow_id="implement", + policy="reap", + drain_timeout_s=5, + ) + assert outcome.reaped_runs == 0 + assert reaper.calls == [] + iteration_dir = ( + repo_root + / ".loopy_loop" + / "sessions" + / session_id + / "iterations" + / "0001_implement" + ) + assert not (iteration_dir / SALVAGE_FILENAME).exists() + + +# --------------------------------------------------------------------------- +# Worker client: identity in payloads, 409 handling +# --------------------------------------------------------------------------- + + +class _Response: + def __init__( + self, payload: dict[str, Any], status_code: int = 200, text: str = "" + ) -> None: + self._payload = payload + self.status_code = status_code + self.text = text + + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, Any]: + return self._payload + + +class _Client: + def __init__(self, responses: list[_Response]) -> None: + self._responses = responses + self.posted: list[tuple[str, dict[str, Any]]] = [] + + def __enter__(self) -> _Client: + return self + + def __exit__(self, *args: Any) -> None: + return None + + def post(self, url: str, json: dict[str, Any]) -> _Response: + self.posted.append((url, json)) + return self._responses.pop(0) + + +def test_worker_register_payload_carries_identity( + tmp_path: Path, monkeypatch: Any +) -> None: + client = _Client([_Response({"action": "stop", "stop_reason": "goal_met"})]) + monkeypatch.setattr("loopy_loop.worker.httpx.Client", lambda timeout: client) + run_worker_loop(repo_root=tmp_path, coordinator_url="http://coordinator") + url, payload = client.posted[0] + assert url.endswith("/register") + assert payload["worker"]["pid"] > 0 + assert payload["worker"]["hostname"] == socket.gethostname() + + +def test_worker_exits_with_code_3_on_busy(tmp_path: Path, monkeypatch: Any) -> None: + client = _Client( + [_Response({"detail": "worker pid=1 is still running"}, status_code=409)] + ) + monkeypatch.setattr("loopy_loop.worker.httpx.Client", lambda timeout: client) + with pytest.raises(SystemExit) as excinfo: + run_worker_loop(repo_root=tmp_path, coordinator_url="http://coordinator") + assert excinfo.value.code == 3 + + +# --------------------------------------------------------------------------- +# Config surface +# --------------------------------------------------------------------------- + + +def test_recovery_config_defaults(repo_builder: Any, monkeypatch: Any) -> None: + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder() + from loopy_loop.config import run_preflight + + preflight = run_preflight(repo_root=repo_root, workflow_set=None, goal_file=None) + assert preflight.root_config.recovery_policy == "drain" + assert preflight.root_config.recovery_drain_timeout_s == 600.0 + + +def test_recovery_config_overrides_and_validation( + repo_builder: Any, monkeypatch: Any +) -> None: + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder( + root_config={"recovery_policy": "reap", "recovery_drain_timeout_s": 30} + ) + from loopy_loop.config import run_preflight + + preflight = run_preflight(repo_root=repo_root, workflow_set=None, goal_file=None) + assert preflight.root_config.recovery_policy == "reap" + assert preflight.root_config.recovery_drain_timeout_s == 30.0 diff --git a/website/src/app/docs/configuration/page.mdx b/website/src/app/docs/configuration/page.mdx index 65f3e0e..52e0e58 100644 --- a/website/src/app/docs/configuration/page.mdx +++ b/website/src/app/docs/configuration/page.mdx @@ -72,6 +72,8 @@ These optional fields tune how team-harness retries transient API and network er | `team_harness_max_retries` | integer | unset | Coordinator retry budget. Must be `0` or greater. | | `team_harness_retry_base_delay_s` | number | unset | Base delay in seconds for retry backoff. Must be greater than `0`. | | `team_harness_retry_max_delay_s` | number | unset | Maximum delay in seconds for retry backoff. Must be greater than `0`, and no less than `team_harness_retry_base_delay_s`. | +| `recovery_policy` | string | `drain` | What crash recovery does with agent processes a dead worker left running: `drain` lets them finish (bounded by `recovery_drain_timeout_s`) before the iteration is re-run; `reap` kills them immediately. | +| `recovery_drain_timeout_s` | number | `600` | Shared deadline in seconds for draining orphaned agents during crash recovery. Must be `0` or greater. | The `inner_outer_eval` template ships these as commented-out examples so you can see the shape before enabling them. diff --git a/website/src/app/docs/http-contract/page.mdx b/website/src/app/docs/http-contract/page.mdx index 3863183..956c498 100644 --- a/website/src/app/docs/http-contract/page.mdx +++ b/website/src/app/docs/http-contract/page.mdx @@ -28,7 +28,18 @@ Every response from either endpoint is a `TaskResponse` whose `action` is either A worker calls `/register` to receive its first assignment. The request body is empty. -Request: `{}` +Request: `{}` — or, optionally, the worker's process identity: + +```json +{ + "worker": { "hostname": "buildbox", "pid": 4242, "starttime": "lstart:..." } +} +``` + +When present, the coordinator stamps the identity onto the dispatched task so a +later `/register` can verify whether that worker is still alive before reclaiming +its task. The same optional `worker` field exists on `/finished` (the caller runs +the next dispatched task). Old workers that post `{}` keep working unchanged. Run response: @@ -86,11 +97,17 @@ Rules for `/register`: `.loopy_loop/workflow_sets//workflows//` directory to load. - If a task is already current — meaning the previous worker crashed without calling - `/finished` — `/register` first checks that iteration directory for + `/finished` — `/register` first verifies whether the recorded worker is *actually + dead*: a verifiably-alive worker (same host, matching pid + start-time token) gets + **HTTP 409** instead of having its task reclaimed. For a confirmed-dead or + unverifiable worker, the coordinator checks that iteration directory for `pending_finished_request.json` or `result.json`. If either proves the task completed, the coordinator records the completed task in history before checking - stop conditions. Only a task with no recoverable local completion is recorded as - failed with `error="abandoned"`. See [Crash recovery](#crash-recovery-and-stale-retries). + stop conditions. A task with no recoverable local completion has its orphaned + agent processes drained or reaped per `recovery_policy` (a `salvage.json` records + what was handled) and is recorded as failed with `error="abandoned_after_drain"` + (or plain `"abandoned"` when nothing was reaped). See + [Crash recovery](#crash-recovery-and-stale-retries). - If the loop is already in a terminal state, `/register` immediately returns `action: "stop"`. @@ -170,3 +187,15 @@ window, the next `/register` recovers the completed result from pending file is missing) and records the task as completed — rather than marking it `abandoned`. A locally written result is treated as authoritative. The per-iteration files behind this are described in [Session Layout](/docs/session-layout). + +**Worker crash mid-run.** With nothing recoverable, `/register` applies the +configured recovery policy to any agent processes the dead worker's harness run +left behind — by default **bounded drain**: let in-flight agents finish (their +completed repo edits survive in the working tree; git is the source of truth), +then re-run the iteration from that better starting point. A `salvage.json` in the +interrupted iteration directory records what was drained or reaped, so surviving +edits are auditable rather than a mystery diff. The iteration's own result is never +fabricated. Draining can block `/register` for up to `recovery_drain_timeout_s`; +the bundled worker uses an unbounded read timeout for exactly this reason. This +requires a team-harness version with the process reaper — older versions skip +orphan recovery and fall back to plain abandonment. diff --git a/website/src/app/docs/session-layout/page.mdx b/website/src/app/docs/session-layout/page.mdx index 14a1973..a161d08 100644 --- a/website/src/app/docs/session-layout/page.mdx +++ b/website/src/app/docs/session-layout/page.mdx @@ -175,6 +175,14 @@ task instead of marking it `abandoned`; if it is missing but `result.json` exist for the active task, the coordinator reconstructs the finished request from `result.json`. See [HTTP Contract](/docs/http-contract) for the recovery rules. +**`salvage.json`** — Written into the interrupted iteration's directory during +crash recovery, when the coordinator drained or reaped agent processes a dead +worker's harness run left behind (per `recovery_policy`). It records the reap +reports so the provenance of any surviving working-tree edits is auditable rather +than a mystery diff. The iteration is still re-run — its `result.json` never +existed and is never fabricated — and its history entry carries +`error="abandoned_after_drain"` instead of plain `"abandoned"`. + **`goal_check.json`** — A per-iteration eval artifact, present only for the reserved `goal_check` workflow or a workflow configured with `emits_goal_check: true`. Its schema in v1: From 0d0b59c21adfd67ba1a34f689678bac2408c999f Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Mon, 13 Jul 2026 09:55:28 +0200 Subject: [PATCH 2/3] fix: harden worker liveness + recovery per adversarial reviews MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both reviewers (Codex gpt-5.6-sol xhigh, Antigravity) returned request-changes on c6316cc; this addresses their reconciled findings. Protocol compatibility (Codex M7 — the compat claim was false): - recovery_policy / recovery_drain_timeout_s removed from the wire RootConfigSnapshot: released workers validate the snapshot with extra="forbid", so the new response fields would have crashed every old worker at parse time. The settings are coordinator-side only and excluded when snapshots are built. Liveness correctness (both reviews' top finding + Codex M6): - is_worker_alive distinguishes "pid gone" from "cannot capture": a coordinator whose team-harness lacks process identity (version skew) now returns unknown -> legacy behavior, never "verified dead" -> duplicate dispatch. Zombies count as dead (an exited-but-unreaped worker cannot hold its task hostage via 409). - The unbounded read timeout applies to /register ONLY; /finished keeps the bounded default so a wedged response cannot leave a worker alive-but-stuck forever. 409 messages name the pid and the kill-and-reregister escape hatch for hung workers. Recovery safety (Codex M2/M3/M5/m1, Antigravity 3/4/5): - Two-phase /register: the potentially long drain runs OUTSIDE the state lock (phase A), with a locked re-validating commit (phase B) that replans when state moved in between — loopy status/stop and /finished stay usable during recovery; lock contention surfaces as HTTP 503 and friendly CLI errors instead of raw tracebacks. - Unsettled outcomes block dispatch: if any orphan may still be running after recovery (unverifiable identity, probe failure, kill that did not land), /register returns 409 instead of starting a second writer; the salvage record (now with unsettled_workers, written even when a mid-loop refusal aborts recovery) names the unresolved processes. - Same-host gate: a worker identity from another hostname skips local reaping (signals cannot reach it; shared-filesystem deployments would otherwise probe the wrong host's pids and report false success). - Run-record validation: a discovered directory name is only reaped when its run.json's run_id and session_output_dir point back at this iteration (stray names cannot select an unrelated run). - ONE recovery_drain_timeout_s deadline shared across all of the iteration's runs (was per-run). Ownership (Codex M8): - A stale /finished from a DIFFERENT identified worker is refused (409) instead of being handed the live task (second executor); unknown identities keep the pre-existing stale-retry behavior. Audit honesty (Codex m2, Antigravity 7): - History error is abandoned_after_ (drain/reap), only when something actually settled. Docs: http-contract (canonical + website), session-layout (both), configuration, README, CHANGELOG all realigned with the implemented behavior (same-host-only recovery, hung-worker escape hatch, shared deadline, coordinator-only settings, stale-finished 409, 503 semantics). Tests: 6 added (141 total) — version-skew unknown, zombie-as-dead, stray-directory skip, unsettled -> RecoveryIncompleteError with salvage, stale-finished 409 + legacy passthrough, no-recovery-fields-on-the-wire. Integration re-verified against the real team-harness branch: a real orphan drained through the hardened flow (record-match guard passing, salvage with settled/unsettled counts). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01PN8aFwwxA8FzMy9LgpbQFt --- CHANGELOG.md | 31 +++- README.md | 6 +- docs/http-contract.md | 44 +++-- docs/session-layout.md | 10 +- src/loopy_loop/cli.py | 15 +- src/loopy_loop/coordinator_app.py | 184 +++++++++++++++---- src/loopy_loop/models.py | 2 - src/loopy_loop/recovery.py | 119 +++++++++--- src/loopy_loop/worker.py | 16 +- src/loopy_loop/worker_identity.py | 56 +++++- src/tests/test_worker.py | 4 +- src/tests/test_worker_liveness_recovery.py | 146 ++++++++++++++- website/src/app/docs/configuration/page.mdx | 4 +- website/src/app/docs/http-contract/page.mdx | 43 +++-- website/src/app/docs/session-layout/page.mdx | 7 +- 15 files changed, 555 insertions(+), 132 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e1e6d58..92f0a33 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,16 +11,27 @@ keep the pre-existing behavior; the wire change is backward compatible. - **Orphaned-agent recovery (P2.5 / TH-D5 consumer side).** When a worker is confirmed dead with nothing recoverable, the coordinator applies - `recovery_policy` (new config; default `drain`, bounded by - `recovery_drain_timeout_s`, default 600s) to agent processes the dead - worker's harness run left behind, via team-harness's process reaper: drained - agents finish and their repo edits survive; a `salvage.json` in the - interrupted iteration directory records what was handled; the history entry - is `abandoned_after_drain` instead of `abandoned`. Requires team-harness - with the process reaper (> 0.2.10); older versions skip orphan recovery - gracefully. team-harness's own parent-liveness guard surfaces as HTTP 409. -- The bundled worker uses an unbounded read timeout (recovery can legitimately - block `/register` up to the drain deadline) and exits with code 3 on a 409. + `recovery_policy` (new coordinator-side config, NOT part of the wire + snapshot; default `drain`, one shared `recovery_drain_timeout_s` deadline, + default 600s) to agent processes the dead worker's harness run left behind, + via team-harness's process reaper: drained agents finish and their repo + edits survive; a `salvage.json` in the interrupted iteration directory + records what was handled; the history entry is `abandoned_after_` + when anything settled. Recovery runs OUTSIDE the state lock (`loopy + status`/`stop` stay usable while draining), validates that each discovered + run record actually belongs to the iteration, is same-host-only (a worker + identity from another hostname skips reaping), and refuses to dispatch + replacement work (HTTP 409) when any orphan may still be running or when + team-harness's parent-liveness guard reports the run's owner alive. + Requires team-harness with the process reaper (> 0.2.10); older versions + skip orphan recovery gracefully. +- A stale `/finished` from a **different identified worker** now gets HTTP 409 + instead of a second copy of the live task; unknown identities keep the + pre-existing stale-retry behavior. State-lock contention surfaces as a clean + HTTP 503 (and friendly CLI errors) instead of raw tracebacks. +- The bundled worker uses an unbounded read timeout on `/register` only + (recovery can legitimately block registration up to the drain deadline), + keeps the bounded timeout on `/finished`, and exits with code 3 on a 409. ## 0.2.1 diff --git a/README.md b/README.md index 10f0832..591f43d 100644 --- a/README.md +++ b/README.md @@ -231,8 +231,10 @@ Important rules: team-harness API/network errors. - `recovery_policy` (`drain` by default, or `reap`) and `recovery_drain_timeout_s` control what crash recovery does with agent - processes a dead worker left running: drain lets them finish (bounded) - before the iteration is re-run; reap kills them immediately. + processes a dead worker left running: drain lets them finish (one shared + bounded deadline) before the iteration is re-run; reap kills them + immediately. These are coordinator-side settings and are not part of the + config snapshot sent to workers. Workflow config lives beside each workflow prompt: diff --git a/docs/http-contract.md b/docs/http-contract.md index 626b523..1040fce 100644 --- a/docs/http-contract.md +++ b/docs/http-contract.md @@ -49,9 +49,7 @@ Run response: "team_harness_retry_max_delay_s": null, "team_harness_api_base": "https://openrouter.ai/api/v1", "team_harness_api_key_env": "OPENROUTER_API_KEY", - "team_harness_system_prompt_extension": "", - "recovery_policy": "drain", - "recovery_drain_timeout_s": 600.0 + "team_harness_system_prompt_extension": "" }, "stop_reason": null } @@ -91,16 +89,29 @@ Rules: file proves the task completed, the completed task is recorded in history before checking stop conditions. 3. **Orphan recovery.** With nothing recoverable, the coordinator applies the - configured recovery policy (`recovery_policy`, default `drain` bounded by - `recovery_drain_timeout_s`) to any agent processes the dead worker's - harness run left behind, writes a `salvage.json` into the interrupted - iteration directory when something was handled, and records the iteration - as failed with `error="abandoned_after_drain"` (or plain `"abandoned"` - when nothing was reaped). Requires team-harness with the process reaper; - older versions skip this step. If team-harness's own guard reports the - run's owning process still alive, `/register` returns **HTTP 409**. - Note: draining can block `/register` for up to `recovery_drain_timeout_s`; - the bundled worker uses an unbounded read timeout for this reason. + configured recovery policy (`recovery_policy`, default `drain`; ONE + `recovery_drain_timeout_s` deadline shared across all of the iteration's + interrupted runs) to any agent processes the dead worker's harness run + left behind, writes a `salvage.json` into the interrupted iteration + directory when something was handled, and records the iteration as failed + with `error="abandoned_after_"` (or plain `"abandoned"` when + nothing settled). Requires team-harness with the process reaper; older + versions skip this step. Recovery refuses to dispatch replacement work — + **HTTP 409** — when team-harness's guard reports the run's owner still + alive, or when any orphan's state after recovery is "may still be + running" (unverifiable identity, probe failure, or a kill that did not + land); the salvage record documents the unresolved processes. + The recovery settings are coordinator-side configuration only — they are + **not** part of the wire `config_snapshot` (released workers reject unknown + snapshot fields). + Notes: recovery runs outside the state lock, so `loopy status`/`stop` stay + usable while it drains; `/register` can still block roughly up to the drain + deadline (plus kill grace periods), and the bundled worker uses an unbounded + read timeout on `/register` only. Process recovery is same-host: a worker + identity from another hostname skips reaping (its processes cannot be + reached from here). A **hung-but-alive** worker keeps its task (409); the + escape hatch is to kill that process and register again — the 409 message + names its pid. - If the loop is in a terminal state, `/register` immediately returns `action=stop`. ## POST /finished @@ -131,8 +142,11 @@ Response: same shape as `/register` response (`action` is either `"run"` or `"st Rules: - If `session_id` + `workflow_id` + `iteration` does not match `current_task`, - the call is treated as stale: state is not mutated, `current_task` is not - changed, and the current task's run response is returned to the caller. + the call is treated as stale: state is not mutated and `current_task` is not + changed. The current task's run response is returned only when the caller's + identity matches the task's recorded owner (or either identity is unknown — + the pre-identity behavior); a stale call from a **different identified + worker** gets **HTTP 409** instead of a second copy of the live task. - If `current_task` is `None` (no task is active), the coordinator dispatches the next available task as if `/register` had been called. If the state is terminal, it returns `action=stop`. diff --git a/docs/session-layout.md b/docs/session-layout.md index 73427e6..90919ee 100644 --- a/docs/session-layout.md +++ b/docs/session-layout.md @@ -174,10 +174,14 @@ eval-banana run \ finish), reaped (killed), or skipped, so the provenance of any surviving working-tree edits is auditable rather than a mystery diff - The iteration is still re-run — its `result.json` never existed and is never - fabricated; the corresponding history entry carries - `error="abandoned_after_drain"` instead of plain `"abandoned"` + fabricated; when any orphan actually settled, the corresponding history entry + carries `error="abandoned_after_"` (e.g. `abandoned_after_drain`) + instead of plain `"abandoned"` - Schema: `{"schema_version": 1, "recorded_at": ..., "policy": ..., - "reaped_runs": N, "settled_workers": N, "reports": [...]}` + "reaped_runs": N, "settled_workers": N, "unsettled_workers": N, + "reports": [...]}`. A non-zero `unsettled_workers` means some orphan may + still be running — the coordinator refuses to dispatch replacement work + (HTTP 409) until it is resolved ## Control Contracts diff --git a/src/loopy_loop/cli.py b/src/loopy_loop/cli.py index 746179f..71658a6 100644 --- a/src/loopy_loop/cli.py +++ b/src/loopy_loop/cli.py @@ -5,6 +5,7 @@ from pathlib import Path import click +from filelock import Timeout as FileLockTimeout import uvicorn from loopy_loop.config import ConfigError @@ -235,7 +236,12 @@ def worker(coordinator_url: str) -> None: def status() -> None: """Show loop status.""" repo_root = Path.cwd() - state = StateStore(repo_root=repo_root).read_state() + try: + state = StateStore(repo_root=repo_root).read_state() + except FileLockTimeout: + raise click.ClickException( + "coordinator state is locked (likely mid-request); retry shortly" + ) from None if state is None: click.echo("No loopy-loop state found.") return @@ -266,7 +272,12 @@ def mutator(state: LoopState | None) -> tuple[LoopState, None]: state.stop_requested = True return state, None - store.mutate(mutator) + try: + store.mutate(mutator) + except FileLockTimeout: + raise click.ClickException( + "coordinator state is locked (likely mid-request); retry shortly" + ) from None click.echo("stop requested") diff --git a/src/loopy_loop/coordinator_app.py b/src/loopy_loop/coordinator_app.py index 9bde214..e74d056 100644 --- a/src/loopy_loop/coordinator_app.py +++ b/src/loopy_loop/coordinator_app.py @@ -4,11 +4,13 @@ import json import logging from pathlib import Path +import socket from typing import Any from typing import TypeVar from fastapi import FastAPI from fastapi import HTTPException +from filelock import Timeout as FileLockTimeout from pydantic import BaseModel from loopy_loop.config import ConfigError @@ -32,6 +34,7 @@ from loopy_loop.models import utc_now from loopy_loop.models import WorkerIdentity from loopy_loop.recovery import recover_interrupted_iteration +from loopy_loop.recovery import RecoveryIncompleteError from loopy_loop.recovery import RecoveryOutcome from loopy_loop.recovery import RecoveryRefusedError from loopy_loop.scheduler import choose_next_workflow @@ -49,6 +52,11 @@ logger = logging.getLogger(__name__) +# Coordinator-operational settings that must never enter the wire config +# snapshot: released workers validate the snapshot with extra="forbid", so any +# new response field would crash them at parse time (protocol compatibility). +_COORDINATOR_ONLY_FIELDS = {"recovery_policy", "recovery_drain_timeout_s"} + class WorkerBusyError(RuntimeError): """The current task's worker is verifiably still alive — do not reclaim.""" @@ -75,12 +83,27 @@ def create_coordinator_app( def register_worker(request: RegisterRequest | None = None) -> TaskResponse: try: return service.register_worker(request=request) - except (WorkerBusyError, RecoveryRefusedError) as exc: + except (WorkerBusyError, RecoveryRefusedError, RecoveryIncompleteError) as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc + except FileLockTimeout as exc: + raise HTTPException( + status_code=503, + detail="coordinator state is briefly locked (crash recovery " + "or a concurrent request); retry shortly", + ) from exc @app.post("/finished", response_model=TaskResponse) def finish_assignment(request: FinishedRequest) -> TaskResponse: - return service.finish_assignment(request=request) + try: + return service.finish_assignment(request=request) + except WorkerBusyError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + except FileLockTimeout as exc: + raise HTTPException( + status_code=503, + detail="coordinator state is briefly locked (crash recovery " + "or a concurrent request); retry shortly", + ) from exc return app @@ -105,10 +128,50 @@ def __init__( def register_worker( self, *, request: RegisterRequest | None = None ) -> TaskResponse: - recovered_pending_paths: list[Path] = [] caller = request.worker if request is not None else None + # Two-phase recovery: the potentially long drain of a dead worker's + # orphaned agents (up to recovery_drain_timeout_s) runs in phase A, + # OUTSIDE the state lock, so `loopy status`/`stop` and /finished stay + # responsive. Phase B re-validates under the lock and retries from + # phase A when the state moved in between. + for _ in range(3): + recovery = self._plan_orphan_recovery() + response = self._register_attempt(caller=caller, recovery=recovery) + if response is not None: + return response + raise WorkerBusyError( + "crash recovery is contended (state changed repeatedly while " + "recovering); retry shortly" + ) - def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: + def _plan_orphan_recovery(self) -> tuple[CurrentTask, RecoveryOutcome] | None: + """Phase A: inspect state and, if needed, drain/reap OUTSIDE the lock. + + Raises WorkerBusyError when the current task's worker is verifiably + alive, and RecoveryRefusedError when team-harness's parent guard finds + the interrupted run's owner still living (both -> HTTP 409). + """ + state = self.state_store.read_state() + if state is None or state.current_task is None: + return None + orphaned = state.current_task + self._raise_if_worker_alive(current_task=orphaned) + if self._read_recoverable_finished_request(current_task=orphaned) is not None: + # A completed result exists: phase B recovers it under the lock; + # nothing to reap (the worker finished its harness run). + return None + return orphaned, self._recover_orphaned_agents(current_task=orphaned) + + def _register_attempt( + self, + *, + caller: WorkerIdentity | None, + recovery: tuple[CurrentTask, RecoveryOutcome] | None, + ) -> TaskResponse | None: + """Phase B: commit under the state lock; None means retry from phase A.""" + recovered_pending_paths: list[Path] = [] + + def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse | None]: current = _require_state(state=state) now = utc_now() @@ -121,23 +184,27 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: # Verified-alive worker: its task is NOT abandoned. Refuse this # register instead of dispatching duplicate work (D7). Unknown # (None) falls through to the pre-existing recovery behavior. - if is_worker_alive(orphaned.worker) is True: - raise WorkerBusyError( - f"worker pid={orphaned.worker.pid} on " # type: ignore[union-attr] - f"{orphaned.worker.hostname} is still running iteration " # type: ignore[union-attr] - f"{orphaned.iteration} ({orphaned.workflow_id}); " - "refusing to dispatch duplicate work" - ) + self._raise_if_worker_alive(current_task=orphaned) recovered = self._read_recoverable_finished_request( current_task=orphaned ) - if recovered is None: - # The worker is dead with nothing recoverable: apply the - # recovery policy to any agent processes it orphaned - # (default bounded drain — TH-D5/P2.5), then re-run. - # May raise RecoveryRefusedError -> 409 if team-harness's - # parent-liveness guard says the run's owner still lives. - recovery = self._recover_orphaned_agents(current_task=orphaned) + if recovered is not None: + recovered_request, pending_path = recovered + if pending_path is not None: + recovered_pending_paths.append(pending_path) + self._record_finished_task( + state=current, + active=orphaned, + request=recovered_request, + now=now, + ) + elif recovery is not None and _same_task(orphaned, recovery[0]): + # The dead worker's orphans were handled in phase A + # (outside the lock); commit the abandonment. + outcome = recovery[1] + error = "abandoned" + if outcome.salvaged: + error = f"abandoned_after_{outcome.policy or 'drain'}" current.history.append( HistoryEntry( iteration=orphaned.iteration, @@ -145,9 +212,7 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: workflow_id=orphaned.workflow_id, session_id=orphaned.session_id, success=False, - error="abandoned_after_drain" - if recovery.salvaged - else "abandoned", + error=error, started_at=orphaned.started_at, finished_at=now, ) @@ -155,15 +220,10 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: current.iteration_count += 1 current.current_task = None else: - recovered_request, pending_path = recovered - if pending_path is not None: - recovered_pending_paths.append(pending_path) - self._record_finished_task( - state=current, - active=orphaned, - request=recovered_request, - now=now, - ) + # The state moved between phase A and phase B (a stale + # /finished landed, or another register recovered first): + # do not act on a stale plan — replan. + return current, None # Step 4: Check stop conditions after abandoned-task cleanup. stop_response = self._stop_response_if_needed(state=current) @@ -209,6 +269,18 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: path.unlink(missing_ok=True) return response + def _raise_if_worker_alive(self, *, current_task: CurrentTask) -> None: + if is_worker_alive(current_task.worker) is not True: + return + worker = current_task.worker + assert worker is not None + raise WorkerBusyError( + f"worker pid={worker.pid} on {worker.hostname} is still running " + f"iteration {current_task.iteration} ({current_task.workflow_id}); " + "refusing to dispatch duplicate work. If that worker is hung, " + "kill the process and register again." + ) + def finish_assignment(self, *, request: FinishedRequest) -> TaskResponse: caller = request.worker @@ -264,6 +336,25 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: or request.workflow_id != active.workflow_id or request.iteration != active.iteration ): + # Replaying the live task is only safe to its recorded owner: + # handing it to a DIFFERENT identified worker would start a + # second executor of the same task. Unknown identities (old + # workers) keep the pre-existing stale-retry behavior. + if ( + caller is not None + and active.worker is not None + and ( + caller.hostname != active.worker.hostname + or caller.pid != active.worker.pid + or caller.starttime != active.worker.starttime + ) + ): + raise WorkerBusyError( + f"stale /finished from pid={caller.pid} on " + f"{caller.hostname}: iteration {active.iteration} " + f"({active.workflow_id}) belongs to worker " + f"pid={active.worker.pid} on {active.worker.hostname}" + ) return current, _build_run_response( current_task=active, config_snapshot=current.config_snapshot ) @@ -379,11 +470,26 @@ def _record_finished_task( def _recover_orphaned_agents(self, *, current_task: CurrentTask) -> RecoveryOutcome: """Apply the configured recovery policy to a dead worker's orphans. - Runs while the state lock is held: nothing else may be dispatched - until recovery settles (draining can take up to the configured - timeout — the single-worker model makes blocking here correct). + Runs in phase A, OUTSIDE the state lock — draining can take up to the + configured timeout without blocking status/stop or /finished. + + Host gate: process signals only reach this host. When the recorded + worker ran elsewhere (shared-filesystem deployments), reaping here + would probe/kill the wrong host's pids and could falsely report the + orphans handled — skip instead and leave a plain abandonment. """ config = self.preflight.root_config + worker = current_task.worker + if worker is not None and worker.hostname != socket.gethostname(): + logger.warning( + "worker for iteration %04d_%s ran on %s (not this host); " + "skipping orphan recovery — its agent processes cannot be " + "reached from here", + current_task.iteration, + current_task.workflow_id, + worker.hostname, + ) + return RecoveryOutcome(policy=config.recovery_policy) return recover_interrupted_iteration( repo_root=self.repo_root, session_id=current_task.session_id, @@ -468,7 +574,7 @@ def _write_fresh_state(self) -> None: state_path=state_path(repo_root=self.repo_root, session_id=session_id), ) snapshot = RootConfigSnapshot.model_validate( - self.preflight.root_config.model_dump() + self.preflight.root_config.model_dump(exclude=_COORDINATOR_ONLY_FIELDS) ) state = LoopState( status="running", @@ -545,7 +651,7 @@ def _dispatch_child_session_if_requested( parent_session_id=state.active_session_id, ) snapshot = RootConfigSnapshot.model_validate( - preflight.root_config.model_dump() + preflight.root_config.model_dump(exclude=_COORDINATOR_ONLY_FIELDS) ) now = utc_now() child_state = LoopState( @@ -727,6 +833,14 @@ def _require_state(*, state: LoopState | None) -> LoopState: return state +def _same_task(a: CurrentTask, b: CurrentTask) -> bool: + return ( + a.session_id == b.session_id + and a.workflow_id == b.workflow_id + and a.iteration == b.iteration + ) + + def _matches_current_task( *, request: FinishedRequest, current_task: CurrentTask ) -> bool: diff --git a/src/loopy_loop/models.py b/src/loopy_loop/models.py index 356c2ae..0fa0113 100644 --- a/src/loopy_loop/models.py +++ b/src/loopy_loop/models.py @@ -43,8 +43,6 @@ class RootConfigSnapshot(BaseModel): team_harness_api_base: str = Field(...) team_harness_api_key_env: str = Field(...) team_harness_system_prompt_extension: str = Field(...) - recovery_policy: Literal["drain", "reap"] = Field(default="drain") - recovery_drain_timeout_s: float = Field(default=600.0) class WorkerIdentity(BaseModel): diff --git a/src/loopy_loop/recovery.py b/src/loopy_loop/recovery.py index a5de578..4427135 100644 --- a/src/loopy_loop/recovery.py +++ b/src/loopy_loop/recovery.py @@ -33,6 +33,7 @@ import json import logging from pathlib import Path +import time from typing import Any from loopy_loop.models import utc_now @@ -53,6 +54,26 @@ "drain_timed_out_then_reaped", } +# Outcomes that mean an orphan MAY STILL BE RUNNING after recovery: the +# coordinator must not dispatch replacement work on top of a possibly-live +# writer. (identity_mismatch_skipped means OUR group is gone — settled-safe; +# no_process_identity is a pre-identity record and keeps the legacy +# redispatch behavior; left_running only occurs under the ignore policy.) +_UNSETTLED_OUTCOMES = { + "identity_unverifiable_skipped", + "probe_failed", + "kill_failed_still_running", +} + + +class RecoveryIncompleteError(RuntimeError): + """Recovery ran but orphaned agents may still be running. + + Dispatching replacement work now could put two writers on the checkout. + The salvage record documents exactly which orphans are unresolved; an + operator can kill them (or wait for them to exit) and register again. + """ + class RecoveryRefusedError(RuntimeError): """team-harness refused to reap: the run's owning process is still alive. @@ -82,6 +103,8 @@ class RecoveryOutcome: reaped_runs: int = 0 settled_workers: int = 0 + unsettled_workers: int = 0 + policy: str = "" reports: list[dict[str, Any]] = field(default_factory=list) @property @@ -104,7 +127,7 @@ def recover_interrupted_iteration( finds the run's owning process still alive — the caller should treat that as "the previous worker is still running". """ - outcome = RecoveryOutcome() + outcome = RecoveryOutcome(policy=policy) loaded = _load_reaper() if loaded is None: logger.warning( @@ -121,34 +144,81 @@ def recover_interrupted_iteration( iteration=iteration, workflow_id=workflow_id, ) - for run_id in _discover_run_ids(output_root): - run_json = Path(th_config.RUNS_DIR) / run_id / "run.json" - if not run_json.exists(): - logger.warning("no run.json for interrupted harness run %s", run_id) - continue - try: - report = reaper.reap_run( - run_json, policy=policy, drain_timeout_s=drain_timeout_s + # ONE deadline shared across every discovered run — the advertised + # timeout bounds the whole recovery, not each run separately. + deadline = time.monotonic() + drain_timeout_s + try: + for run_id in _discover_run_ids(output_root): + run_json = Path(th_config.RUNS_DIR) / run_id / "run.json" + if not run_json.exists(): + logger.warning("no run.json for interrupted harness run %s", run_id) + continue + if not _run_record_matches( + run_json=run_json, output_root=output_root, run_id=run_id + ): + logger.warning( + "run.json for %s does not reference this iteration's " + "output directory; skipping (stray directory name?)", + run_id, + ) + continue + try: + report = reaper.reap_run( + run_json, + policy=policy, + drain_timeout_s=max(0.0, deadline - time.monotonic()), + ) + except reaper.ReapRefusedError as exc: + raise RecoveryRefusedError(str(exc)) from exc + outcome.reaped_runs += 1 + outcome.settled_workers += sum( + 1 for worker in report.workers if worker.outcome in _SETTLED_OUTCOMES ) - except reaper.ReapRefusedError as exc: - raise RecoveryRefusedError(str(exc)) from exc - outcome.reaped_runs += 1 - outcome.settled_workers += sum( - 1 for worker in report.workers if worker.outcome in _SETTLED_OUTCOMES - ) - outcome.reports.append(report.model_dump(mode="json")) - if outcome.reaped_runs: - _write_salvage_record( - repo_root=repo_root, - session_id=session_id, - iteration=iteration, - workflow_id=workflow_id, - outcome=outcome, - policy=policy, + outcome.unsettled_workers += sum( + 1 for worker in report.workers if worker.outcome in _UNSETTLED_OUTCOMES + ) + outcome.reports.append(report.model_dump(mode="json")) + finally: + # The salvage record survives a mid-loop refusal: whatever was already + # reaped/drained stays auditable even when this call raises. + if outcome.reaped_runs: + _write_salvage_record( + repo_root=repo_root, + session_id=session_id, + iteration=iteration, + workflow_id=workflow_id, + outcome=outcome, + policy=policy, + ) + if outcome.unsettled_workers: + raise RecoveryIncompleteError( + f"{outcome.unsettled_workers} orphaned agent(s) of iteration " + f"{iteration:04d}_{workflow_id} may still be running " + "(unverifiable identity, probe failure, or a kill that did not " + "land); refusing to dispatch replacement work. See salvage.json " + "in the iteration directory, resolve the leftover processes, " + "and register again." ) return outcome +def _run_record_matches(*, run_json: Path, output_root: Path, run_id: str) -> bool: + """Guard against stray directory names: the run record must point back at + this iteration's output directory before we act on it.""" + try: + payload = json.loads(run_json.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return False + if payload.get("run_id") != run_id: + return False + recorded = payload.get("session_output_dir") + if recorded is None: + # Older team-harness run records don't carry it; fall back to the + # directory-name match already established. + return True + return Path(recorded).resolve() == (output_root / run_id).resolve() + + def _discover_run_ids(output_root: Path) -> list[str]: """The iteration's harness output root contains one directory per run id.""" if not output_root.is_dir(): @@ -184,6 +254,7 @@ def _write_salvage_record( "policy": policy, "reaped_runs": outcome.reaped_runs, "settled_workers": outcome.settled_workers, + "unsettled_workers": outcome.unsettled_workers, "reports": outcome.reports, } (iteration_dir / SALVAGE_FILENAME).write_text( diff --git a/src/loopy_loop/worker.py b/src/loopy_loop/worker.py index e55a1d7..e239eb7 100644 --- a/src/loopy_loop/worker.py +++ b/src/loopy_loop/worker.py @@ -58,12 +58,8 @@ class FinishedAssignment: def run_worker_loop(*, repo_root: Path, coordinator_url: str) -> None: base_url = coordinator_url.rstrip("/") - # The read timeout is unbounded because /register may legitimately block - # while the coordinator drains a crashed predecessor's orphaned agents - # (bounded by the coordinator's recovery_drain_timeout_s, not by us). - timeout = httpx.Timeout(30.0, read=None) identity = current_worker_identity() - with httpx.Client(timeout=timeout) as client: + with httpx.Client(timeout=30.0) as client: task = _post_register( client=client, coordinator_url=base_url, identity=identity ) @@ -95,7 +91,15 @@ def _post_register( *, client: httpx.Client, coordinator_url: str, identity: WorkerIdentity ) -> TaskResponse: request = RegisterRequest(worker=identity) - response = client.post(f"{coordinator_url}/register", json=request.model_dump()) + # Unbounded read for /register ONLY: registration may legitimately block + # while the coordinator drains a crashed predecessor's orphaned agents. + # /finished keeps the bounded default so a wedged response cannot leave + # this worker alive-but-stuck forever (which would 409 all reclaims). + response = client.post( + f"{coordinator_url}/register", + json=request.model_dump(), + timeout=httpx.Timeout(30.0, read=None), + ) _exit_if_busy(response) response.raise_for_status() return TaskResponse.model_validate(response.json()) diff --git a/src/loopy_loop/worker_identity.py b/src/loopy_loop/worker_identity.py index 99b4774..e9829da 100644 --- a/src/loopy_loop/worker_identity.py +++ b/src/loopy_loop/worker_identity.py @@ -21,23 +21,50 @@ import importlib import os import socket +import subprocess from loopy_loop.models import WorkerIdentity -def capture_starttime(pid: int) -> str | None: - """team-harness's pid-reuse-proof start-time token; None if unavailable. +def _identity_module(): + """team-harness's process-identity module, or None on older versions. Resolved at call time so loopy-loop keeps working with team-harness versions that predate process identity (< 0.2.11). """ try: - module = importlib.import_module("team_harness.agents.process_identity") + return importlib.import_module("team_harness.agents.process_identity") except ImportError: return None + + +def capture_starttime(pid: int) -> str | None: + """team-harness's pid-reuse-proof start-time token; None if unavailable.""" + module = _identity_module() + if module is None: + return None return module.capture_starttime(pid) +def _is_zombie(pid: int) -> bool: + """A zombie has exited: it holds no resources and will never call back. + + An exited worker lingers as a zombie only until its parent (usually the + operator's shell) reaps it — but while it lingers, its start time still + matches, so liveness must not mistake it for a running worker. + """ + try: + output = subprocess.run( + ["ps", "-p", str(pid), "-o", "stat="], + capture_output=True, + text=True, + timeout=10, + ).stdout.strip() + except (OSError, subprocess.SubprocessError): + return False + return output.startswith("Z") + + def current_worker_identity() -> WorkerIdentity: """Identity of THIS process, captured for /register and /finished calls.""" pid = os.getpid() @@ -50,15 +77,26 @@ def is_worker_alive(identity: WorkerIdentity | None) -> bool | None: """Verified liveness of a recorded worker identity. Returns True/False only when the answer is *verified*: same host, a - recorded start-time token, and a current capture that matches (alive) or - differs/is absent (dead — the pid is gone or was recycled). Returns None - when verification is impossible (no identity recorded, remote host, or no - token) — callers must treat None as "unknown" and fall back to the - pre-existing recovery behavior, never as "alive". + recorded start-time token, capture capability present, and a current + capture that matches (alive, and not a zombie) or differs/is absent + (dead — the pid is gone or was recycled). Returns None when verification + is impossible (no identity recorded, remote host, no token, or a + team-harness without process identity) — callers must treat None as + "unknown" and fall back to the pre-existing recovery behavior, never as + "alive". + + The capability check matters for version skew: a coordinator whose + team-harness cannot capture start times must NOT interpret its own + None-capture as "the pid is gone" — that would declare a live worker + verifiably dead and dispatch duplicate work. """ if identity is None or identity.starttime is None: return None if identity.hostname != socket.gethostname(): return None + if _identity_module() is None: + return None current = capture_starttime(identity.pid) - return current == identity.starttime + if current != identity.starttime: + return False + return not _is_zombie(identity.pid) diff --git a/src/tests/test_worker.py b/src/tests/test_worker.py index 7c276d1..714cc5e 100644 --- a/src/tests/test_worker.py +++ b/src/tests/test_worker.py @@ -71,7 +71,9 @@ def __enter__(self) -> FakeClient: def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None: return None - def post(self, url: str, json: dict[str, object]) -> FakeResponse: + def post( + self, url: str, json: dict[str, object], timeout: object = None + ) -> FakeResponse: self._posted_payloads.append({"url": url, "json": json}) item = self._responses.pop(0) if isinstance(item, Exception): diff --git a/src/tests/test_worker_liveness_recovery.py b/src/tests/test_worker_liveness_recovery.py index d0768d8..ed5bebc 100644 --- a/src/tests/test_worker_liveness_recovery.py +++ b/src/tests/test_worker_liveness_recovery.py @@ -60,6 +60,8 @@ def test_is_worker_alive_unknown_cases() -> None: def test_is_worker_alive_verified_true_and_false(monkeypatch: Any) -> None: identity = WorkerIdentity(**LOCAL_IDENTITY) + monkeypatch.setattr(worker_identity_module, "_identity_module", lambda: object()) + monkeypatch.setattr(worker_identity_module, "_is_zombie", lambda pid: False) monkeypatch.setattr( worker_identity_module, "capture_starttime", @@ -74,6 +76,29 @@ def test_is_worker_alive_verified_true_and_false(monkeypatch: Any) -> None: assert is_worker_alive(identity) is False # pid recycled +def test_is_worker_alive_unknown_when_capture_unavailable(monkeypatch: Any) -> None: + # Version skew: a coordinator whose team-harness cannot capture start + # times must NOT interpret its own None-capture as "the pid is gone" — + # that would declare a live worker verifiably dead (duplicate work). + identity = WorkerIdentity(**LOCAL_IDENTITY) + monkeypatch.setattr(worker_identity_module, "_identity_module", lambda: None) + assert is_worker_alive(identity) is None + + +def test_is_worker_alive_zombie_counts_as_dead(monkeypatch: Any) -> None: + # An exited-but-unreaped worker still matches its start time, but it will + # never call /finished: it must not hold the task hostage via 409. + identity = WorkerIdentity(**LOCAL_IDENTITY) + monkeypatch.setattr(worker_identity_module, "_identity_module", lambda: object()) + monkeypatch.setattr( + worker_identity_module, + "capture_starttime", + lambda pid: LOCAL_IDENTITY["starttime"], + ) + monkeypatch.setattr(worker_identity_module, "_is_zombie", lambda pid: True) + assert is_worker_alive(identity) is False + + # --------------------------------------------------------------------------- # Coordinator: verified-alive worker -> 409, no state mutation # --------------------------------------------------------------------------- @@ -343,7 +368,12 @@ def _build_interrupted_iteration( for run_id in run_ids: run_dir = runs_dir / run_id run_dir.mkdir(parents=True, exist_ok=True) - (run_dir / "run.json").write_text("{}", encoding="utf-8") + (run_dir / "run.json").write_text( + json.dumps( + {"run_id": run_id, "session_output_dir": str(output_root / run_id)} + ), + encoding="utf-8", + ) return tmp_path, session_id, runs_dir @@ -369,7 +399,10 @@ def test_recover_interrupted_iteration_reaps_and_writes_salvage( assert outcome.settled_workers == 2 # one drained per fake report assert outcome.salvaged assert [call["policy"] for call in reaper.calls] == ["drain", "drain"] - assert [call["drain_timeout_s"] for call in reaper.calls] == [5, 5] + # One SHARED deadline across runs: each call gets the remaining budget. + budgets = [call["drain_timeout_s"] for call in reaper.calls] + assert budgets[0] == pytest.approx(5, abs=1.0) + assert budgets[1] <= budgets[0] salvage_path = ( repo_root / ".loopy_loop" @@ -484,7 +517,7 @@ def __enter__(self) -> _Client: def __exit__(self, *args: Any) -> None: return None - def post(self, url: str, json: dict[str, Any]) -> _Response: + def post(self, url: str, json: dict[str, Any], timeout: Any = None) -> _Response: self.posted.append((url, json)) return self._responses.pop(0) @@ -538,3 +571,110 @@ def test_recovery_config_overrides_and_validation( preflight = run_preflight(repo_root=repo_root, workflow_set=None, goal_file=None) assert preflight.root_config.recovery_policy == "reap" assert preflight.root_config.recovery_drain_timeout_s == 30.0 + + +def test_recover_skips_stray_directory_names(tmp_path: Path, monkeypatch: Any) -> None: + # A directory under harness_outputs whose run.json does not point back at + # this iteration must not be reaped (it may belong to another run). + repo_root, session_id, runs_dir = _build_interrupted_iteration( + tmp_path, run_ids=["run_real"] + ) + stray = runs_dir / "run_real" / "run.json" + stray.write_text( + json.dumps({"run_id": "run_real", "session_output_dir": "/elsewhere"}), + encoding="utf-8", + ) + reaper = _FakeReaperModule(["drained"]) + monkeypatch.setattr( + recovery_module, "_load_reaper", lambda: (reaper, _FakeThConfig(runs_dir)) + ) + outcome = recover_interrupted_iteration( + repo_root=repo_root, + session_id=session_id, + iteration=1, + workflow_id="implement", + policy="drain", + drain_timeout_s=5, + ) + assert outcome.reaped_runs == 0 + assert reaper.calls == [] + + +def test_recover_raises_when_orphans_may_still_run( + tmp_path: Path, monkeypatch: Any +) -> None: + # An unsettled outcome (agent may still be running) must block dispatch: + # two writers on one checkout is exactly what recovery exists to prevent. + from loopy_loop.recovery import RecoveryIncompleteError + + repo_root, session_id, runs_dir = _build_interrupted_iteration( + tmp_path, run_ids=["run_a"] + ) + reaper = _FakeReaperModule(["kill_failed_still_running", "drained"]) + monkeypatch.setattr( + recovery_module, "_load_reaper", lambda: (reaper, _FakeThConfig(runs_dir)) + ) + with pytest.raises(RecoveryIncompleteError, match="may still be running"): + recover_interrupted_iteration( + repo_root=repo_root, + session_id=session_id, + iteration=1, + workflow_id="implement", + policy="drain", + drain_timeout_s=5, + ) + # The salvage record still documents what WAS handled before the refusal. + salvage_path = ( + repo_root + / ".loopy_loop" + / "sessions" + / session_id + / "iterations" + / "0001_implement" + / SALVAGE_FILENAME + ) + payload = json.loads(salvage_path.read_text()) + assert payload["settled_workers"] == 1 + assert payload["unsettled_workers"] == 1 + + +def test_stale_finished_from_other_identified_worker_is_refused( + repo_builder: Any, monkeypatch: Any +) -> None: + # M8: a stale /finished from a DIFFERENT identified worker must not be + # handed the live task (that would start a second executor). + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder() + client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) + body = RegisterRequest(worker=WorkerIdentity(**LOCAL_IDENTITY)).model_dump() + task = client.post("/register", json=body).json() + other = {"hostname": socket.gethostname(), "pid": 777, "starttime": "lstart:x"} + stale = { + "workflow_id": task["workflow_id"], + "session_id": task["session_id"], + "iteration": 999, # mismatch -> stale path + "success": True, + "worker": other, + } + response = client.post("/finished", json=stale) + assert response.status_code == 409 + assert "belongs to worker" in response.json()["detail"] + # Unknown identity keeps the legacy stale-retry behavior: + stale_legacy = {**stale} + del stale_legacy["worker"] + response = client.post("/finished", json=stale_legacy) + assert response.status_code == 200 + assert response.json()["iteration"] == task["iteration"] + + +def test_config_snapshot_on_the_wire_has_no_recovery_fields( + repo_builder: Any, monkeypatch: Any +) -> None: + # M7: released workers validate the snapshot with extra="forbid"; new + # response fields would crash them. Recovery settings stay coordinator-only. + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder(root_config={"recovery_policy": "reap"}) + client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) + task = client.post("/register", json={}).json() + assert "recovery_policy" not in task["config_snapshot"] + assert "recovery_drain_timeout_s" not in task["config_snapshot"] diff --git a/website/src/app/docs/configuration/page.mdx b/website/src/app/docs/configuration/page.mdx index 52e0e58..971f7bd 100644 --- a/website/src/app/docs/configuration/page.mdx +++ b/website/src/app/docs/configuration/page.mdx @@ -72,8 +72,8 @@ These optional fields tune how team-harness retries transient API and network er | `team_harness_max_retries` | integer | unset | Coordinator retry budget. Must be `0` or greater. | | `team_harness_retry_base_delay_s` | number | unset | Base delay in seconds for retry backoff. Must be greater than `0`. | | `team_harness_retry_max_delay_s` | number | unset | Maximum delay in seconds for retry backoff. Must be greater than `0`, and no less than `team_harness_retry_base_delay_s`. | -| `recovery_policy` | string | `drain` | What crash recovery does with agent processes a dead worker left running: `drain` lets them finish (bounded by `recovery_drain_timeout_s`) before the iteration is re-run; `reap` kills them immediately. | -| `recovery_drain_timeout_s` | number | `600` | Shared deadline in seconds for draining orphaned agents during crash recovery. Must be `0` or greater. | +| `recovery_policy` | string | `drain` | Coordinator-side (not sent to workers). What crash recovery does with agent processes a dead worker left running: `drain` lets them finish (one shared `recovery_drain_timeout_s` deadline) before the iteration is re-run; `reap` kills them immediately. | +| `recovery_drain_timeout_s` | number | `600` | Coordinator-side. Shared deadline in seconds for draining ALL of an iteration's orphaned agents during crash recovery. Must be `0` or greater. | The `inner_outer_eval` template ships these as commented-out examples so you can see the shape before enabling them. diff --git a/website/src/app/docs/http-contract/page.mdx b/website/src/app/docs/http-contract/page.mdx index 956c498..a2c448d 100644 --- a/website/src/app/docs/http-contract/page.mdx +++ b/website/src/app/docs/http-contract/page.mdx @@ -98,15 +98,18 @@ Rules for `/register`: load. - If a task is already current — meaning the previous worker crashed without calling `/finished` — `/register` first verifies whether the recorded worker is *actually - dead*: a verifiably-alive worker (same host, matching pid + start-time token) gets - **HTTP 409** instead of having its task reclaimed. For a confirmed-dead or - unverifiable worker, the coordinator checks that iteration directory for - `pending_finished_request.json` or `result.json`. If either proves the task - completed, the coordinator records the completed task in history before checking - stop conditions. A task with no recoverable local completion has its orphaned - agent processes drained or reaped per `recovery_policy` (a `salvage.json` records - what was handled) and is recorded as failed with `error="abandoned_after_drain"` - (or plain `"abandoned"` when nothing was reaped). See + dead*: a verifiably-alive worker (same host, matching pid + start-time token, and + not a zombie) gets **HTTP 409** instead of having its task reclaimed. If that + worker is hung, kill its process (the 409 names the pid) and register again. For a + confirmed-dead or unverifiable worker, the coordinator checks that iteration + directory for `pending_finished_request.json` or `result.json`. If either proves + the task completed, the coordinator records the completed task in history before + checking stop conditions. A task with no recoverable local completion has its + orphaned agent processes drained or reaped per `recovery_policy` (a `salvage.json` + records what was handled) and is recorded as failed with + `error="abandoned_after_"` (or plain `"abandoned"` when nothing settled). + If any orphan may still be running after recovery, `/register` returns **HTTP + 409** rather than dispatching a second writer. See [Crash recovery](#crash-recovery-and-stale-retries). - If the loop is already in a terminal state, `/register` immediately returns `action: "stop"`. @@ -192,10 +195,18 @@ files behind this are described in [Session Layout](/docs/session-layout). configured recovery policy to any agent processes the dead worker's harness run left behind — by default **bounded drain**: let in-flight agents finish (their completed repo edits survive in the working tree; git is the source of truth), -then re-run the iteration from that better starting point. A `salvage.json` in the -interrupted iteration directory records what was drained or reaped, so surviving -edits are auditable rather than a mystery diff. The iteration's own result is never -fabricated. Draining can block `/register` for up to `recovery_drain_timeout_s`; -the bundled worker uses an unbounded read timeout for exactly this reason. This -requires a team-harness version with the process reaper — older versions skip -orphan recovery and fall back to plain abandonment. +then re-run the iteration from that better starting point. One +`recovery_drain_timeout_s` deadline is shared across all of the iteration's +interrupted runs. A `salvage.json` in the interrupted iteration directory records +what was drained or reaped, so surviving edits are auditable rather than a mystery +diff. The iteration's own result is never fabricated. Recovery runs outside the +state lock (`loopy status`/`stop` stay usable), but `/register` itself can block +roughly up to the drain deadline plus kill grace periods — the bundled worker uses +an unbounded read timeout on `/register` only. Process recovery is same-host-only: +a worker identity from another hostname skips reaping. If any orphan may still be +running afterwards (unverifiable identity, probe failure, or a kill that did not +land), the coordinator refuses to dispatch replacement work (HTTP 409) and the +salvage record names the unresolved processes. This requires a team-harness version +with the process reaper — older versions skip orphan recovery and fall back to +plain abandonment. The recovery settings are coordinator-side only; they are not +part of the wire `config_snapshot`. diff --git a/website/src/app/docs/session-layout/page.mdx b/website/src/app/docs/session-layout/page.mdx index a161d08..a838df5 100644 --- a/website/src/app/docs/session-layout/page.mdx +++ b/website/src/app/docs/session-layout/page.mdx @@ -180,8 +180,11 @@ crash recovery, when the coordinator drained or reaped agent processes a dead worker's harness run left behind (per `recovery_policy`). It records the reap reports so the provenance of any surviving working-tree edits is auditable rather than a mystery diff. The iteration is still re-run — its `result.json` never -existed and is never fabricated — and its history entry carries -`error="abandoned_after_drain"` instead of plain `"abandoned"`. +existed and is never fabricated — and when any orphan actually settled, its +history entry carries `error="abandoned_after_"` (e.g. +`abandoned_after_drain`) instead of plain `"abandoned"`. A non-zero +`unsettled_workers` in the record means some orphan may still be running; the +coordinator refuses to dispatch replacement work (HTTP 409) until it is resolved. **`goal_check.json`** — A per-iteration eval artifact, present only for the reserved `goal_check` workflow or a workflow configured with From 8277b2dab9b16f250a3b971260a9104fe4178956 Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Mon, 13 Jul 2026 10:13:40 +0200 Subject: [PATCH 3/3] feat!: require worker identity at /register; stale replay only to the owner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The maintainer confirmed backward compatibility is not required, which unlocks closing the one residual duplicate-execution window the compat compromise had left open (Codex M8's weak branch): - /register without a worker identity is rejected with HTTP 400 (breaking; pre-0.3 workers cannot register — upgrade workers and coordinator together). Every dispatched task therefore has a recorded owner and liveness verification is always at least attemptable. - A stale /finished is replayed ONLY to the task's recorded owner; any other caller — identified or anonymous — gets HTTP 409. A task persisted by a pre-identity version (owner None) keeps the legacy replay for that one resume. Deliberately KEPT despite permission to break: - recovery settings stay off the wire snapshot (coordinator-only concern — better design regardless of compat); - the "unknown" liveness verdict and importlib soft-resolution remain: they guard against runtime environment facts (remote hosts, a team-harness without process identity), not old clients. Docs (http-contract both copies, CHANGELOG breaking header per the 0.2.0 house style) and tests updated: register-requires-identity, anonymous stale 409, owner replay, pre-upgrade ownerless replay. 143 tests green. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01PN8aFwwxA8FzMy9LgpbQFt --- CHANGELOG.md | 17 +++-- docs/http-contract.md | 14 +++-- src/loopy_loop/coordinator_app.py | 44 ++++++++----- src/tests/test_coordinator_app.py | 46 +++++++------- src/tests/test_idempotent_finished.py | 10 ++- src/tests/test_worker_liveness_recovery.py | 70 +++++++++++++++++---- website/src/app/docs/http-contract/page.mdx | 13 ++-- 7 files changed, 149 insertions(+), 65 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 92f0a33..70f2d7e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,14 +1,23 @@ # Changelog -## Unreleased +## Unreleased (breaking) -- **Worker liveness verification (D7).** The worker now sends its process +**Breaking API change — `/register` requires the worker's process identity.** +A register without a `worker` object is rejected with HTTP 400. Pre-0.3 +workers cannot register against a 0.3 coordinator; upgrade workers and +coordinator together (they normally ship in the same install). + +- **Worker liveness verification (D7).** The worker sends its process identity (hostname + pid + a pid-reuse-proof start-time token) with `/register` and `/finished`; the coordinator stamps it onto the dispatched task. A `/register` while the recorded worker is *verifiably still alive* returns HTTP 409 instead of abandoning live work — closing the - duplicate-work window. Unverifiable identities (old workers, remote hosts) - keep the pre-existing behavior; the wire change is backward compatible. + duplicate-work window. Unverifiable identities (remote hosts, or a + team-harness without process identity) keep the pre-existing + assume-abandoned recovery behavior. Because every dispatched task now has + a recorded owner, a stale `/finished` is replayed **only to that owner** + (anyone else gets HTTP 409) — a task persisted by a pre-identity version + keeps the legacy replay for that one resume. - **Orphaned-agent recovery (P2.5 / TH-D5 consumer side).** When a worker is confirmed dead with nothing recoverable, the coordinator applies `recovery_policy` (new coordinator-side config, NOT part of the wire diff --git a/docs/http-contract.md b/docs/http-contract.md index 1040fce..2fbf966 100644 --- a/docs/http-contract.md +++ b/docs/http-contract.md @@ -4,7 +4,8 @@ loopy-loop exposes exactly two coordinator endpoints. ## POST /register -Request: `{}` (empty body), or with the worker's process identity: +Request (the worker's process identity is **required** — a breaking change in +0.3; pre-0.3 workers are rejected with HTTP 400): ```json { @@ -16,11 +17,12 @@ Request: `{}` (empty body), or with the worker's process identity: } ``` -`worker` is optional and backward compatible. When present, the coordinator -stamps it onto the dispatched task so a later `/register` can *verify* whether -that worker is still alive before reclaiming its task. `starttime` is -team-harness's pid-reuse-proof process-identity token (requires team-harness -with process identity support; omitted otherwise). +The coordinator stamps the identity onto the dispatched task, which is what +makes two guarantees possible: a later `/register` can *verify* whether that +worker is still alive before reclaiming its task, and a stale `/finished` is +only ever replayed to the task's recorded owner. `starttime` is team-harness's +pid-reuse-proof process-identity token (null when the worker's team-harness +predates process identity — verification then degrades to "unknown"). Run response: diff --git a/src/loopy_loop/coordinator_app.py b/src/loopy_loop/coordinator_app.py index e74d056..81540bd 100644 --- a/src/loopy_loop/coordinator_app.py +++ b/src/loopy_loop/coordinator_app.py @@ -81,6 +81,16 @@ def create_coordinator_app( @app.post("/register", response_model=TaskResponse) def register_worker(request: RegisterRequest | None = None) -> TaskResponse: + # Breaking change (0.3): identity is required. It guarantees every + # dispatched task has a recorded owner, so liveness verification and + # the stale-/finished owner check are always possible. Old workers + # fail fast here with a clear message instead of degrading silently. + if request is None or request.worker is None: + raise HTTPException( + status_code=400, + detail="worker identity is required; upgrade the worker CLI " + "(loopy-loop >= 0.3)", + ) try: return service.register_worker(request=request) except (WorkerBusyError, RecoveryRefusedError, RecoveryIncompleteError) as exc: @@ -337,23 +347,27 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: or request.iteration != active.iteration ): # Replaying the live task is only safe to its recorded owner: - # handing it to a DIFFERENT identified worker would start a - # second executor of the same task. Unknown identities (old - # workers) keep the pre-existing stale-retry behavior. - if ( - caller is not None - and active.worker is not None - and ( - caller.hostname != active.worker.hostname - or caller.pid != active.worker.pid - or caller.starttime != active.worker.starttime - ) + # handing it to anyone else would start a second executor of + # the same task. Identity is required at /register, so every + # task dispatched by this version has an owner; a None owner + # can only come from pre-upgrade persisted state and keeps + # the legacy replay behavior for that one resume. + if active.worker is not None and ( + caller is None + or caller.hostname != active.worker.hostname + or caller.pid != active.worker.pid + or caller.starttime != active.worker.starttime ): + caller_desc = ( + f"pid={caller.pid} on {caller.hostname}" + if caller is not None + else "an unidentified caller" + ) raise WorkerBusyError( - f"stale /finished from pid={caller.pid} on " - f"{caller.hostname}: iteration {active.iteration} " - f"({active.workflow_id}) belongs to worker " - f"pid={active.worker.pid} on {active.worker.hostname}" + f"stale /finished from {caller_desc}: iteration " + f"{active.iteration} ({active.workflow_id}) belongs " + f"to worker pid={active.worker.pid} on " + f"{active.worker.hostname}" ) return current, _build_run_response( current_task=active, config_snapshot=current.config_snapshot diff --git a/src/tests/test_coordinator_app.py b/src/tests/test_coordinator_app.py index f5c0889..e06613d 100644 --- a/src/tests/test_coordinator_app.py +++ b/src/tests/test_coordinator_app.py @@ -15,13 +15,15 @@ from loopy_loop.sessions import session_dir_path from loopy_loop.state_store import StateStore +REGISTER_BODY = {"worker": {"hostname": "test-host", "pid": 999983, "starttime": None}} + def test_register_returns_run_response(repo_builder: Any, monkeypatch: Any) -> None: monkeypatch.setenv("OPENROUTER_API_KEY", "secret") repo_root = repo_builder() client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() assert response["action"] == "run" @@ -33,7 +35,7 @@ def test_register_response_has_correct_fields( repo_root = repo_builder() client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() assert response["action"] == "run" assert response["workflow_id"] == "planner" @@ -51,7 +53,7 @@ def test_register_sets_current_task(repo_builder: Any, monkeypatch: Any) -> None client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) store = StateStore(repo_root=repo_root) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() state = store.read_state() assert state is not None @@ -67,7 +69,7 @@ def test_finished_records_history(repo_builder: Any, monkeypatch: Any) -> None: client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) store = StateStore(repo_root=repo_root) - reg = client.post("/register", json={}).json() + reg = client.post("/register", json=REGISTER_BODY).json() client.post( "/finished", json={ @@ -97,7 +99,7 @@ def test_finished_returns_next_run(repo_builder: Any, monkeypatch: Any) -> None: repo_root = repo_builder() client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) - reg = client.post("/register", json={}).json() + reg = client.post("/register", json=REGISTER_BODY).json() finished = client.post( "/finished", json={ @@ -147,7 +149,7 @@ def test_child_session_runs_inside_parent_and_resumes_parent( client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) store = StateStore(repo_root=repo_root) - parent_task = client.post("/register", json={}).json() + parent_task = client.post("/register", json=REGISTER_BODY).json() request_dir = child_requests_dir_path( repo_root=repo_root, session_id=parent_task["session_id"] ) @@ -250,7 +252,7 @@ def test_failed_parent_iteration_does_not_dispatch_child_request( ) client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) - parent_task = client.post("/register", json={}).json() + parent_task = client.post("/register", json=REGISTER_BODY).json() request_path = ( child_requests_dir_path( repo_root=repo_root, session_id=parent_task["session_id"] @@ -294,11 +296,13 @@ def test_finished_stale_mismatch_does_not_mutate( client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) store = StateStore(repo_root=repo_root) - reg = client.post("/register", json={}).json() - # Send /finished with wrong session_id — stale mismatch. + reg = client.post("/register", json=REGISTER_BODY).json() + # Send /finished with wrong session_id — stale mismatch FROM THE OWNER + # (the live-task replay is only served to the task's recorded worker). stale = client.post( "/finished", json={ + "worker": REGISTER_BODY["worker"], "workflow_id": reg["workflow_id"], "session_id": "wrong-session-id", "iteration": reg["iteration"], @@ -422,7 +426,7 @@ def test_register_recovers_abandoned_task( state.current_task = orphaned store.write_state(state=state) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() updated = store.read_state() # Abandoned task should be recorded in history. @@ -472,7 +476,7 @@ def test_register_recovers_completed_task_from_pending_finished_request( encoding="utf-8", ) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() updated = store.read_state() assert response["action"] == "stop" @@ -523,7 +527,7 @@ def test_register_recovers_completed_task_from_result_json( encoding="utf-8", ) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() updated = store.read_state() assert response["action"] == "stop" @@ -560,7 +564,7 @@ def test_register_terminal_plus_abandoned_task_cleanup_first( state.current_task = orphaned store.write_state(state=state) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() updated = store.read_state() # Abandoned entry must be recorded even though state was terminal. @@ -582,7 +586,7 @@ def test_register_stop_when_terminal(repo_builder: Any, monkeypatch: Any) -> Non state.goal_met = True store.write_state(state=state) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() assert response["action"] == "stop" assert response["stop_reason"] == "goal_met" @@ -594,7 +598,7 @@ def test_finished_stop_after_max_turns(repo_builder: Any, monkeypatch: Any) -> N client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) store = StateStore(repo_root=repo_root) - reg = client.post("/register", json={}).json() + reg = client.post("/register", json=REGISTER_BODY).json() response = client.post( "/finished", json={ @@ -625,7 +629,7 @@ def test_stop_precedence_goal_met(repo_builder: Any, monkeypatch: Any) -> None: state.stop_requested = True store.write_state(state=state) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() assert response["action"] == "stop" assert response["stop_reason"] == "goal_met" @@ -698,7 +702,7 @@ def test_session_control_signal_sets_goal_met( encoding="utf-8", ) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() updated = store.read_state() assert response["action"] == "stop" @@ -723,7 +727,7 @@ def test_invalid_session_control_signal_stops( encoding="utf-8", ) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() updated = store.read_state() assert response["action"] == "stop" @@ -966,7 +970,7 @@ def test_no_eligible_workflow_stops(repo_builder: Any, monkeypatch: Any) -> None ) client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() assert response["action"] == "stop" assert response["stop_reason"] == "no_eligible_workflow" @@ -1018,7 +1022,7 @@ def test_stop_precedence_matrix( setattr(state, key, value) store.write_state(state=state) - response = client.post("/register", json={}).json() + response = client.post("/register", json=REGISTER_BODY).json() assert response["action"] == "stop" assert response["stop_reason"] == expected_stop_reason @@ -1037,7 +1041,7 @@ def test_resume_reuses_in_progress_session(repo_builder: Any, monkeypatch: Any) resumed_client = TestClient(resumed_app) resumed_store = StateStore(repo_root=repo_root) resumed_state = resumed_store.read_state() - register_response = resumed_client.post("/register", json={}).json() + register_response = resumed_client.post("/register", json=REGISTER_BODY).json() assert first_app is not None assert resumed_state is not None diff --git a/src/tests/test_idempotent_finished.py b/src/tests/test_idempotent_finished.py index 5fc7f49..02701bc 100644 --- a/src/tests/test_idempotent_finished.py +++ b/src/tests/test_idempotent_finished.py @@ -7,6 +7,8 @@ from loopy_loop.coordinator_app import create_coordinator_app from loopy_loop.state_store import StateStore +REGISTER_BODY = {"worker": {"hostname": "test-host", "pid": 999983, "starttime": None}} + def test_stale_finished_mismatch_does_not_record_history_twice( repo_builder: Any, monkeypatch: Any @@ -17,7 +19,7 @@ def test_stale_finished_mismatch_does_not_record_history_twice( client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) store = StateStore(repo_root=repo_root) - reg = client.post("/register", json={}).json() + reg = client.post("/register", json=REGISTER_BODY).json() # First /finished — legitimate call, processes the task. first = client.post( "/finished", @@ -65,11 +67,13 @@ def test_stale_finished_returns_current_task_run_response( client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) store = StateStore(repo_root=repo_root) - reg = client.post("/register", json={}).json() - # Send /finished with wrong session_id — stale mismatch. + reg = client.post("/register", json=REGISTER_BODY).json() + # Send /finished with wrong session_id — stale mismatch FROM THE OWNER + # (the live-task replay is only served to the task's recorded worker). stale = client.post( "/finished", json={ + "worker": REGISTER_BODY["worker"], "workflow_id": reg["workflow_id"], "session_id": "stale-session-id", "iteration": reg["iteration"], diff --git a/src/tests/test_worker_liveness_recovery.py b/src/tests/test_worker_liveness_recovery.py index ed5bebc..e6fe678 100644 --- a/src/tests/test_worker_liveness_recovery.py +++ b/src/tests/test_worker_liveness_recovery.py @@ -37,6 +37,8 @@ "starttime": "lstart:Sun Jul 12 00:00:00 2026", } +REGISTER_BODY = {"worker": {"hostname": "test-host", "pid": 999983, "starttime": None}} + # --------------------------------------------------------------------------- # worker_identity unit behavior @@ -159,7 +161,7 @@ def _register_with_orphan( "loopy_loop.coordinator_app.recover_interrupted_iteration", lambda **kwargs: recovery_outcome or RecoveryOutcome(), ) - response = client.post("/register", json={}) + response = client.post("/register", json=REGISTER_BODY) return response, store, state @@ -201,7 +203,8 @@ def test_register_recovers_when_worker_verifiably_dead( def test_register_recovers_when_liveness_unknown( repo_builder: Any, monkeypatch: Any, current_task_factory: Any ) -> None: - # Old workers / remote hosts: no identity -> pre-existing behavior. + # Pre-upgrade persisted tasks / remote hosts: no verifiable identity + # -> the pre-existing assume-abandoned recovery behavior. response, store, _ = _register_with_orphan( repo_builder, current_task_factory, @@ -252,7 +255,7 @@ def refuse(**kwargs: Any) -> RecoveryOutcome: monkeypatch.setattr( "loopy_loop.coordinator_app.recover_interrupted_iteration", refuse ) - response = client.post("/register", json={}) + response = client.post("/register", json=REGISTER_BODY) assert response.status_code == 409 assert "still alive" in response.json()["detail"] updated = store.read_state() @@ -301,12 +304,12 @@ def test_finished_stamps_callers_identity_on_next_task( assert state.current_task.worker.pid == LOCAL_IDENTITY["pid"] -def test_register_without_body_still_works(repo_builder: Any, monkeypatch: Any) -> None: - # Pre-identity workers post an empty body; the contract stays compatible. +def test_register_with_identity_dispatches(repo_builder: Any, monkeypatch: Any) -> None: + # The canonical register: identity present, task dispatched and stamped. monkeypatch.setenv("OPENROUTER_API_KEY", "secret") repo_root = repo_builder() client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) - response = client.post("/register", json={}) + response = client.post("/register", json=REGISTER_BODY) assert response.status_code == 200 assert response.json()["action"] == "run" @@ -659,14 +662,59 @@ def test_stale_finished_from_other_identified_worker_is_refused( response = client.post("/finished", json=stale) assert response.status_code == 409 assert "belongs to worker" in response.json()["detail"] - # Unknown identity keeps the legacy stale-retry behavior: - stale_legacy = {**stale} - del stale_legacy["worker"] - response = client.post("/finished", json=stale_legacy) + # An unidentified stale caller is refused too (identity is required at + # /register, so every live task has a recorded owner): + stale_anonymous = {**stale} + del stale_anonymous["worker"] + response = client.post("/finished", json=stale_anonymous) + assert response.status_code == 409 + # The OWNER's stale call still gets the live-task replay: + stale_owner = {**stale, "worker": LOCAL_IDENTITY} + response = client.post("/finished", json=stale_owner) assert response.status_code == 200 assert response.json()["iteration"] == task["iteration"] +def test_stale_finished_replays_for_pre_upgrade_ownerless_task( + repo_builder: Any, monkeypatch: Any +) -> None: + # A task persisted by a pre-identity version has no recorded owner; the + # legacy stale-replay behavior is preserved for that one resume. + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder() + client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) + body = RegisterRequest(worker=WorkerIdentity(**LOCAL_IDENTITY)).model_dump() + task = client.post("/register", json=body).json() + store = StateStore(repo_root=repo_root) + state = store.read_state() + assert state is not None and state.current_task is not None + state.current_task = state.current_task.model_copy(update={"worker": None}) + store.write_state(state=state) + stale = { + "workflow_id": task["workflow_id"], + "session_id": task["session_id"], + "iteration": 999, + "success": True, + "worker": {"hostname": "someone-else", "pid": 1, "starttime": None}, + } + response = client.post("/finished", json=stale) + assert response.status_code == 200 + assert response.json()["iteration"] == task["iteration"] + + +def test_register_requires_worker_identity(repo_builder: Any, monkeypatch: Any) -> None: + # Breaking change (0.3): a register without identity fails fast with a + # clear message instead of silently degrading recovery guarantees. + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder() + client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) + response = client.post("/register", json={}) + assert response.status_code == 400 + assert "worker identity is required" in response.json()["detail"] + response = client.post("/register") + assert response.status_code == 400 + + def test_config_snapshot_on_the_wire_has_no_recovery_fields( repo_builder: Any, monkeypatch: Any ) -> None: @@ -675,6 +723,6 @@ def test_config_snapshot_on_the_wire_has_no_recovery_fields( monkeypatch.setenv("OPENROUTER_API_KEY", "secret") repo_root = repo_builder(root_config={"recovery_policy": "reap"}) client = TestClient(create_coordinator_app(repo_root=repo_root, resume=False)) - task = client.post("/register", json={}).json() + task = client.post("/register", json=REGISTER_BODY).json() assert "recovery_policy" not in task["config_snapshot"] assert "recovery_drain_timeout_s" not in task["config_snapshot"] diff --git a/website/src/app/docs/http-contract/page.mdx b/website/src/app/docs/http-contract/page.mdx index a2c448d..52db419 100644 --- a/website/src/app/docs/http-contract/page.mdx +++ b/website/src/app/docs/http-contract/page.mdx @@ -28,7 +28,8 @@ Every response from either endpoint is a `TaskResponse` whose `action` is either A worker calls `/register` to receive its first assignment. The request body is empty. -Request: `{}` — or, optionally, the worker's process identity: +Request — the worker's process identity is **required** (a breaking change in +0.3; pre-0.3 workers are rejected with HTTP 400): ```json { @@ -36,10 +37,12 @@ Request: `{}` — or, optionally, the worker's process identity: } ``` -When present, the coordinator stamps the identity onto the dispatched task so a -later `/register` can verify whether that worker is still alive before reclaiming -its task. The same optional `worker` field exists on `/finished` (the caller runs -the next dispatched task). Old workers that post `{}` keep working unchanged. +The coordinator stamps the identity onto the dispatched task, so a later +`/register` can verify whether that worker is still alive before reclaiming its +task, and a stale `/finished` is only ever replayed to the task's recorded owner. +The same `worker` field rides on `/finished` (the caller runs the next dispatched +task); `starttime` is null when the worker's team-harness predates process +identity, and verification then degrades to "unknown". Run response: