feat: worker liveness verification + orphaned-agent recovery (D7)

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_<policy>`
+  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.

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:
 

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/<workflow_set>/workflows/<workflow_id>/` 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_<policy>"` (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`.

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_<policy>"` (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`

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

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