diff --git a/CHANGELOG.md b/CHANGELOG.md index 092ff47..70f2d7e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,47 @@ # Changelog +## Unreleased (breaking) + +**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 (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 + 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 - Improve README onboarding, install, initialization, configuration, and logging docs. diff --git a/README.md b/README.md index 6fa6046..591f43d 100644 --- a/README.md +++ b/README.md @@ -229,6 +229,12 @@ 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 (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 2afb850..2fbf966 100644 --- a/docs/http-contract.md +++ b/docs/http-contract.md @@ -4,7 +4,25 @@ loopy-loop exposes exactly two coordinator endpoints. ## POST /register -Request: `{}` (empty body) +Request (the worker's process identity is **required** — a breaking change in +0.3; pre-0.3 workers are rejected with HTTP 400): + +```json +{ + "worker": { + "hostname": "buildbox", + "pid": 4242, + "starttime": "lstart:Sun Jul 12 00:00:00 2026" + } +} +``` + +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: @@ -62,11 +80,40 @@ 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`; 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 @@ -80,17 +127,28 @@ 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: - 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 0c2acfa..90919ee 100644 --- a/docs/session-layout.md +++ b/docs/session-layout.md @@ -165,6 +165,24 @@ 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; 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, "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 `control.json` 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/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..81540bd 100644 --- a/src/loopy_loop/coordinator_app.py +++ b/src/loopy_loop/coordinator_app.py @@ -4,10 +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 @@ -24,10 +27,16 @@ 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 RecoveryIncompleteError +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,9 +48,19 @@ 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__) +# 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.""" + def create_coordinator_app( *, @@ -61,12 +80,40 @@ 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: + # 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: + 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 @@ -88,10 +135,53 @@ 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: + 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 _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]: + def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse | None]: current = _require_state(state=state) now = utc_now() @@ -101,10 +191,30 @@ 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. + self._raise_if_worker_alive(current_task=orphaned) recovered = self._read_recoverable_finished_request( current_task=orphaned ) - if recovered is None: + 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, @@ -112,7 +222,7 @@ def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: workflow_id=orphaned.workflow_id, session_id=orphaned.session_id, success=False, - error="abandoned", + error=error, started_at=orphaned.started_at, finished_at=now, ) @@ -120,22 +230,19 @@ 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) 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 +267,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, @@ -171,7 +279,21 @@ 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 + def mutator(state: LoopState | None) -> tuple[LoopState, TaskResponse]: current = _require_state(state=state) now = utc_now() @@ -184,7 +306,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 +328,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, @@ -223,6 +346,29 @@ 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 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 {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 ) @@ -244,7 +390,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 +416,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 +425,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 +481,38 @@ 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 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, + 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: @@ -405,7 +588,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", @@ -449,7 +632,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 @@ -482,7 +665,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( @@ -499,6 +682,7 @@ def _dispatch_child_session_if_requested( session_id=child_session_id, iteration=1, started_at=now, + worker=caller, ), ) child_store = StateStore( @@ -529,13 +713,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 +735,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 @@ -661,6 +847,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 c36cf07..0fa0113 100644 --- a/src/loopy_loop/models.py +++ b/src/loopy_loop/models.py @@ -45,12 +45,33 @@ class RootConfigSnapshot(BaseModel): team_harness_system_prompt_extension: str = Field(...) +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): workflow_set: str = Field(...) workflow_id: str = Field(...) session_id: str = Field(...) iteration: int = Field(...) started_at: datetime = Field(...) + worker: WorkerIdentity | None = Field(default=None) class TaskResponse(BaseModel): @@ -101,6 +122,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..4427135 --- /dev/null +++ b/src/loopy_loop/recovery.py @@ -0,0 +1,262 @@ +"""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 +import time +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", +} + +# 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. + + 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 + unsettled_workers: int = 0 + policy: str = "" + 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(policy=policy) + 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, + ) + # 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 + ) + 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(): + 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, + "unsettled_workers": outcome.unsettled_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..e239eb7 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,16 @@ class FinishedAssignment: def run_worker_loop(*, repo_root: Path, coordinator_url: str) -> None: base_url = coordinator_url.rstrip("/") + identity = current_worker_identity() with httpx.Client(timeout=30.0) as client: - task = _post_register(client=client, coordinator_url=base_url) + 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 +87,36 @@ 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) + # 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()) +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 +135,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 +212,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..e9829da --- /dev/null +++ b/src/loopy_loop/worker_identity.py @@ -0,0 +1,102 @@ +"""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 +import subprocess + +from loopy_loop.models import WorkerIdentity + + +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: + 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() + 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, 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) + if current != identity.starttime: + return False + return not _is_zombie(identity.pid) 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.py b/src/tests/test_worker.py index 5307081..714cc5e 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 @@ -67,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 new file mode 100644 index 0000000..e6fe678 --- /dev/null +++ b/src/tests/test_worker_liveness_recovery.py @@ -0,0 +1,728 @@ +"""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", +} + +REGISTER_BODY = {"worker": {"hostname": "test-host", "pid": 999983, "starttime": None}} + + +# --------------------------------------------------------------------------- +# 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, "_identity_module", lambda: object()) + monkeypatch.setattr(worker_identity_module, "_is_zombie", lambda pid: False) + 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 + + +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 +# --------------------------------------------------------------------------- + + +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=REGISTER_BODY) + 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: + # 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, + 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=REGISTER_BODY) + 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_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=REGISTER_BODY) + 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( + json.dumps( + {"run_id": run_id, "session_output_dir": str(output_root / run_id)} + ), + 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"] + # 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" + / "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], timeout: Any = None) -> _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 + + +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"] + # 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: + # 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=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/configuration/page.mdx b/website/src/app/docs/configuration/page.mdx index 65f3e0e..971f7bd 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` | 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 3863183..52db419 100644 --- a/website/src/app/docs/http-contract/page.mdx +++ b/website/src/app/docs/http-contract/page.mdx @@ -28,7 +28,21 @@ 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 — the worker's process identity is **required** (a breaking change in +0.3; pre-0.3 workers are rejected with HTTP 400): + +```json +{ + "worker": { "hostname": "buildbox", "pid": 4242, "starttime": "lstart:..." } +} +``` + +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: @@ -86,11 +100,20 @@ 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 - `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). + `/finished` — `/register` first verifies whether the recorded worker is *actually + 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"`. @@ -170,3 +193,23 @@ 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. 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 14a1973..a838df5 100644 --- a/website/src/app/docs/session-layout/page.mdx +++ b/website/src/app/docs/session-layout/page.mdx @@ -175,6 +175,17 @@ 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 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 `emits_goal_check: true`. Its schema in v1: