From d83e8115b3dae6f075c31a1859d96069e15d466f Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Thu, 3 Sep 2026 05:39:03 -0700 Subject: [PATCH 1/2] fix(custody): reclaim a hold whose holder PROCESS IS DEAD, without waiting out the TTL A custody record has carried `pid` and `host` since it was designed and nothing has ever read them for a decision. Liveness was inferred entirely from silence, so a lane whose process died sat unclaimable for the remainder of its 900s TTL plus up to a 300s sweep interval -- and its relaunched successor could do nothing at all: it could not work_claim (held by a dead agent), could not work_release (it does not hold it), and could not work_file (filing requires holding an item). FORENSICS (committed, docs/lanes/oy4-dead-holder-reclaim/evidence/): the measured incident on model_performance-h6v was NOT a broken TTL, a wrong field, or a dead sweep -- the item's three filed candidates are all falsified. Its holder renewed on a metronome-regular 120s cadence through 07:41:36Z and then stopped; all four refused successor claims (07:47/07:50/07:51/07:56Z) landed INSIDE the 900s TTL, the last by 45 seconds, and were refused correctly. The 'held_stale: 0' reading at 07:52Z was likewise correct. The defect is that the one fact which settled the matter -- the holder's process was gone -- sat unread. THE FIX. A third path to reclaim-eligible that OBSERVES the holder instead of inferring from its silence, fenced so it can only ever ACCELERATE the TTL and never take work from a live agent: the record must name this host, a real pid, and custody must already have been silent for two renewal intervals before any pid probe is consulted at all. Every unknowable case resolves to NOT eligible. Two further silent misses in the reaper, found while root-causing this: - reap_project read bd's DEFAULT 50-item list page, so a held item outside it was invisible to the reaper permanently, with nothing reporting the skip. - one item whose release raised aborted the reap of every remaining hold in that project, deterministically and forever, while the sweep still recorded itself completed. And the instrument that would have made this visible: doctor gains `sweeps.reclaiming`, which tells 'the reap loop is turning' apart from 'the reap loop is doing anything'. reap_loop discarded reap_sweep's per-project error results before stamping a completed heartbeat, so `sweeps.alive` (and work_tracker_status's running_healthy) read identically whether every project succeeded or every project failed. --- .../evidence/fail-before-pass-after.txt | 48 ++++ .../evidence/failbefore_probe.py | 50 ++++ .../evidence/h6v-forensic-timeline.txt | 38 +++ .../tests/test_dead_holder_successor_claim.py | 161 +++++++++++ src/amplifier_work_tracker/cli.py | 37 +++ src/amplifier_work_tracker/custody.py | 157 ++++++++++- src/amplifier_work_tracker/heartbeat.py | 108 +++++++- src/amplifier_work_tracker/supervisor.py | 89 +++++- tests/integration/test_dead_holder_reclaim.py | 262 ++++++++++++++++++ tests/unit/test_custody_dead_holder.py | 240 ++++++++++++++++ tests/unit/test_sweeps_reclaiming.py | 260 +++++++++++++++++ 11 files changed, 1433 insertions(+), 17 deletions(-) create mode 100644 docs/lanes/oy4-dead-holder-reclaim/evidence/fail-before-pass-after.txt create mode 100644 docs/lanes/oy4-dead-holder-reclaim/evidence/failbefore_probe.py create mode 100644 docs/lanes/oy4-dead-holder-reclaim/evidence/h6v-forensic-timeline.txt create mode 100644 modules/tool-work-tracker/tests/test_dead_holder_successor_claim.py create mode 100644 tests/integration/test_dead_holder_reclaim.py create mode 100644 tests/unit/test_custody_dead_holder.py create mode 100644 tests/unit/test_sweeps_reclaiming.py diff --git a/docs/lanes/oy4-dead-holder-reclaim/evidence/fail-before-pass-after.txt b/docs/lanes/oy4-dead-holder-reclaim/evidence/fail-before-pass-after.txt new file mode 100644 index 0000000..9f25f9b --- /dev/null +++ b/docs/lanes/oy4-dead-holder-reclaim/evidence/fail-before-pass-after.txt @@ -0,0 +1,48 @@ +FAIL-BEFORE / PASS-AFTER for model_performance-oy4 +================================================== + +Probe: `failbefore_probe.py` (in this directory), run as a pytest integration +test against this repo's isolated dolt server + real `bd`. IDENTICAL script on +both trees -- the only difference is whether `src/` carries the fix (captured +by `git stash push -- src/` and `git stash pop`), so what changes below is +BEHAVIOUR, not an API that did not exist yet. + +Scenario, reproducing the measured `model_performance-h6v` shape exactly: +an item claimed and given custody by an agent whose process is then genuinely +dead (a real subprocess, started and reaped, so `os.kill(pid, 0)` really does +report it gone), whose custody was last renewed 400 seconds ago -- i.e. WELL +INSIDE the documented 900s `CUSTODY_TTL_SECONDS`. Reaped with the default TTL: +exactly the call the background service makes every 300s. + + +-------------------------- FAIL-BEFORE (pre-fix tree) -------------------------- + + holder pid 3821490 running? False + custody last_seen 400s ago; CUSTODY_TTL_SECONDS=900 + work_stats view: held=1 held_stale=0 held_stale_oldest_age_seconds=None + reap_project(default ttl): reclaimed_count=0 reasons=[] + after sweep: status=held holder='dead-agent' + successor work_claim: REFUSED -- claim failbefore26743d9e270d-57i as + 'successor-agent' failed: Error claiming failbefore26743d9e270d-57i: + issue already claimed by dead-agent + +Note `held_stale=0` with `held_stale_oldest_age_seconds=None` beside it, and a +refusal naming a dead agent id. That is the incident's signature verbatim -- +the successor session measured exactly those three values on the live queue at +2026-09-03T07:52Z. + + +---------------------------- PASS-AFTER (with fix) ---------------------------- + + holder pid 3824385 running? False + custody last_seen 400s ago; CUSTODY_TTL_SECONDS=900 + work_stats view: held=1 held_stale=1 held_stale_oldest_age_seconds=401.256 + reap_project(default ttl): reclaimed_count=1 reasons=["holder process is dead + -- pid 3824385 on host 'spark-1' is not running, and custody has been silent + 401s (corroboration window 240s); ttl 900s not yet reached, but the holder + is gone"] + after sweep: status=open holder=None + successor work_claim: SUCCESS + +The reclaim reason states plainly that the TTL is NOT what fired, so a later +reader cannot mistake this for ordinary staleness arriving early. diff --git a/docs/lanes/oy4-dead-holder-reclaim/evidence/failbefore_probe.py b/docs/lanes/oy4-dead-holder-reclaim/evidence/failbefore_probe.py new file mode 100644 index 0000000..5d68582 --- /dev/null +++ b/docs/lanes/oy4-dead-holder-reclaim/evidence/failbefore_probe.py @@ -0,0 +1,50 @@ +"""FAIL-BEFORE / PASS-AFTER probe for model_performance-oy4. + +Reproduces the measured h6v shape with the SAME API on both trees, so the +before/after difference is behaviour, not a missing function signature: +a hold whose holder process is genuinely dead, last renewed 400s ago, +against the documented 900s CUSTODY_TTL_SECONDS. +""" +from __future__ import annotations +import json, os, socket, subprocess, sys, time +import pytest +from amplifier_work_tracker import adapter as A +from amplifier_work_tracker import custody as C +from amplifier_work_tracker import supervisor as SV + +pytestmark = pytest.mark.integration +SILENCE = 400 + + +def _dead_pid() -> int: + p = subprocess.Popen([sys.executable, "-c", "pass"]); p.wait() + return p.pid + + +def test_probe(workspace, project_factory): + name, bd = project_factory("failbefore") + item_id = bd.create("dead holder probe", priority=1) + bd.claim_item(item_id, actor="dead-agent") + pid = _dead_pid() + bd.take_custody(item_id, holder="dead-agent", pid=pid, host=socket.gethostname()) + rec = dict(bd.get(item_id).meta[C.CUSTODY_KEY]) + rec["last_seen"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - SILENCE)) + bd._run(["update", item_id, "--metadata", json.dumps({C.CUSTODY_KEY: rec})], actor="dead-agent") + + out = [] + out.append(f"holder pid {pid} running? {os.path.exists(f'/proc/{pid}')}") + out.append(f"custody last_seen {SILENCE}s ago; CUSTODY_TTL_SECONDS={C.CUSTODY_TTL_SECONDS}") + s = A.project_summary(workspace, name) + out.append(f"work_stats view: held={s.held} held_stale={s.held_stale} " + f"held_stale_oldest_age_seconds={s.held_stale_oldest_age_seconds}") + r = SV.reap_project(bd) # default TTL, exactly what the service runs + out.append(f"reap_project(default ttl): reclaimed_count={r['reclaimed_count']} " + f"reasons={[x['reason'] for x in r['reclaimed']]}") + after = bd.get(item_id) + out.append(f"after sweep: status={after.status} holder={after.holder!r}") + try: + bd.claim_item(item_id, actor="successor-agent") + out.append("successor work_claim: SUCCESS") + except Exception as e: + out.append(f"successor work_claim: REFUSED -- {str(e)[:150]}") + print("\n".join(" " + line for line in out)) diff --git a/docs/lanes/oy4-dead-holder-reclaim/evidence/h6v-forensic-timeline.txt b/docs/lanes/oy4-dead-holder-reclaim/evidence/h6v-forensic-timeline.txt new file mode 100644 index 0000000..976b557 --- /dev/null +++ b/docs/lanes/oy4-dead-holder-reclaim/evidence/h6v-forensic-timeline.txt @@ -0,0 +1,38 @@ +FORENSIC TIMELINE for model_performance-h6v, reconstructed from `events` on the +live shared dolt server (READ-ONLY). events.created_at is written in the dolt +server's LOCAL timezone (PDT, UTC-7) -- see adapter.py's TIMEZONE GOTCHA block -- +so every timestamp below is +7h to UTC. Captured by lane oy4. + +--- lifecycle events --- + 2026-09-03T06:52:14Z created agent-spark-1-2894760 + 2026-09-03T06:52:14Z label_added agent-spark-1-2894760 + 2026-09-03T07:03:16Z claimed agent-spark-1-563997 {"assignee":"agent-spark-1-563997","status":"in_progress"} + 2026-09-03T07:57:51Z status_changed Amplifier {"assignee":"","status":"open"} + 2026-09-03T07:59:09Z claimed agent-spark-1-2053775 {"assignee":"agent-spark-1-2053775","status":"in_progress"} + 2026-09-03T07:59:42Z status_changed agent-spark-1-2053775 {"assignee":"","status":"open"} + 2026-09-03T08:01:39Z claimed agent-spark-1-2015412 {"assignee":"agent-spark-1-2015412","status":"in_progress"} + 2026-09-03T08:01:59Z closed agent-spark-1-2015412 BLOCKED-AND-RECOVERED by the manager. The lane died markerless at ~00:33 mid str + +--- every custody RENEWAL write (n=22) --- + 2026-09-03T07:03:17Z agent-spark-1-563997 last_seen":"2026-09-03T07:03:16Z" + 2026-09-03T07:05:18Z agent-spark-1-563997 last_seen":"2026-09-03T07:05:17Z" + 2026-09-03T07:07:19Z agent-spark-1-563997 last_seen":"2026-09-03T07:07:18Z" + 2026-09-03T07:09:20Z agent-spark-1-563997 last_seen":"2026-09-03T07:09:19Z" + 2026-09-03T07:11:21Z agent-spark-1-563997 last_seen":"2026-09-03T07:11:20Z" + 2026-09-03T07:13:21Z agent-spark-1-563997 last_seen":"2026-09-03T07:13:21Z" + 2026-09-03T07:15:22Z agent-spark-1-563997 last_seen":"2026-09-03T07:15:22Z" + 2026-09-03T07:17:23Z agent-spark-1-563997 last_seen":"2026-09-03T07:17:23Z" + 2026-09-03T07:19:24Z agent-spark-1-563997 last_seen":"2026-09-03T07:19:24Z" + 2026-09-03T07:21:25Z agent-spark-1-563997 last_seen":"2026-09-03T07:21:25Z" + 2026-09-03T07:23:26Z agent-spark-1-563997 last_seen":"2026-09-03T07:23:25Z" + 2026-09-03T07:25:27Z agent-spark-1-563997 last_seen":"2026-09-03T07:25:26Z" + 2026-09-03T07:27:28Z agent-spark-1-563997 last_seen":"2026-09-03T07:27:28Z" + 2026-09-03T07:29:29Z agent-spark-1-563997 last_seen":"2026-09-03T07:29:29Z" + 2026-09-03T07:31:30Z agent-spark-1-563997 last_seen":"2026-09-03T07:31:30Z" + 2026-09-03T07:33:32Z agent-spark-1-563997 last_seen":"2026-09-03T07:33:31Z" + 2026-09-03T07:35:33Z agent-spark-1-563997 last_seen":"2026-09-03T07:35:32Z" + 2026-09-03T07:37:34Z agent-spark-1-563997 last_seen":"2026-09-03T07:37:33Z" + 2026-09-03T07:39:35Z agent-spark-1-563997 last_seen":"2026-09-03T07:39:35Z" + 2026-09-03T07:41:36Z agent-spark-1-563997 last_seen":"2026-09-03T07:41:36Z" + 2026-09-03T07:59:10Z agent-spark-1-2053775 last_seen":"2026-09-03T07:59:10Z" + 2026-09-03T08:01:40Z agent-spark-1-2015412 last_seen":"2026-09-03T08:01:39Z" diff --git a/modules/tool-work-tracker/tests/test_dead_holder_successor_claim.py b/modules/tool-work-tracker/tests/test_dead_holder_successor_claim.py new file mode 100644 index 0000000..2f45b3f --- /dev/null +++ b/modules/tool-work-tracker/tests/test_dead_holder_successor_claim.py @@ -0,0 +1,161 @@ +"""Tier 5 -- the AGENT SEAM half of `model_performance-oy4`: after a lane's +process dies holding a claim, its relaunched successor can `work_claim` the +same item again, automatically, with no human-equivalent intervention. + +WHY THIS LIVES HERE AND NOT ONLY IN tests/integration. The adapter tier +proves the sweep reclaims a dead holder's hold. This tier proves the thing a +STRANDED LANE actually experiences: the measured incident's successor session +could not `work_claim` (held by a dead agent), could not `work_release` (it +does not hold it) and could not `work_file` (filing requires holding an +item), so its own goal condition -- "write BLOCKED.md and release the item" +-- was literally unreachable. Those are all `WorkTrackerSession` verbs, so +that dead end is only reproducible at this seam. + +DELIBERATELY NOT `ttl_seconds=0`. Every other reap test in this suite forces +staleness that way (`_force_reap`), which is exactly the shortcut that lets a +dead-holder bug hide: with ttl 0 EVERY hold is stale, so nothing is being +asserted about a hold that is still comfortably inside its TTL. These reap +with the real default TTL and a real silence of 400s against 900s, so a +reclaim here can only have come from the holder-liveness path. +""" + +from __future__ import annotations + +import json +import shutil +import socket +import subprocess +import sys +import time +import uuid + +import pytest +from amplifier_module_tool_work_tracker import WorkTrackerSession + +import amplifier_work_tracker.adapter as A +import amplifier_work_tracker.custody as C +import amplifier_work_tracker.supervisor as SV + +pytestmark = pytest.mark.skipif( + shutil.which("bd") is None, reason="real `bd` binary not present in this environment" +) + +#: Consumed by the shared `project` fixture in conftest.py. +PROJECT_PREFIX = "deadholdproj" + +HOST = socket.gethostname() +SILENCE_INSIDE_TTL = 400 + + +def _unique(prefix: str) -> str: + return f"{prefix}{uuid.uuid4().hex[:10]}" + + +def _a_genuinely_dead_pid() -> int: + p = subprocess.Popen([sys.executable, "-c", "pass"]) # noqa: S603 + p.wait() + assert not C.pid_alive(p.pid) + return p.pid + + +def _rewind_custody(bd: A.Beads, item_id: str, *, seconds_ago: int) -> None: + rec = dict(bd.get(item_id).meta[C.CUSTODY_KEY]) + rec["last_seen"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - seconds_ago)) + bd._run( # noqa: SLF001 -- forging a past renewal is not a public verb + ["update", item_id, "--metadata", json.dumps({C.CUSTODY_KEY: rec})], actor=rec["holder"] + ) + + +async def _item_held_by_a_dead_lane(project: str) -> tuple[str, str, A.Beads]: + """Reproduce the incident's starting state: an item claimed and given + custody by a lane whose process then died, its last renewal + `SILENCE_INSIDE_TTL` seconds ago -- well inside the 900s TTL. + """ + dead_actor = _unique("dead-lane-") + adder = WorkTrackerSession({"actor": _unique("adder")}) + added = await adder.add(project, "work a dead lane was holding", acceptance="n/a") + assert added.success is True + item_id = added.output["added"] # type: ignore[index] + + session = WorkTrackerSession({"actor": dead_actor}) + bd = session._project(project) # noqa: SLF001 -- test-only reach, as elsewhere in this suite + bd.claim_item(item_id, actor=dead_actor) + bd.take_custody(item_id, holder=dead_actor, pid=_a_genuinely_dead_pid(), host=HOST) + _rewind_custody(bd, item_id, seconds_ago=SILENCE_INSIDE_TTL) + return item_id, dead_actor, bd + + +@pytest.mark.asyncio +async def test_successor_work_claim_is_refused_by_name_before_the_reclaim(project): + """The measured symptom, reproduced at the seam: four attempts, four + refusals, each naming an agent id whose process no longer exists. + """ + item_id, dead_actor, _bd = await _item_held_by_a_dead_lane(project) + + successor = WorkTrackerSession({"actor": _unique("successor")}) + for _attempt in range(2): + refused = await successor.claim(project, item_id=item_id) + assert refused.success is False + assert dead_actor in str(refused.output) + + +@pytest.mark.asyncio +async def test_successor_work_claim_succeeds_after_the_sweep_reclaims(project): + """THE deliverable. One real sweep at the real default TTL, and the + relaunched lane is working again -- no `unclaim`, no operator. + """ + item_id, _dead_actor, bd = await _item_held_by_a_dead_lane(project) + + reaped = SV.reap_project(bd) # default TTL -- no override + assert reaped["reclaimed_count"] == 1, reaped + assert "holder process is dead" in reaped["reclaimed"][0]["reason"] + + successor = WorkTrackerSession({"actor": _unique("successor")}) + claimed = await successor.claim(project, item_id=item_id) + assert claimed.success is True + assert claimed.output["claimed"] == item_id # type: ignore[index] + + # And it is a REAL claim, with its own live custody -- not a half state. + rec = bd.get(item_id).meta[C.CUSTODY_KEY] + assert rec["holder"] == successor._actor # noqa: SLF001 + assert C.reclaim_eligible(rec)[0] is False + + +@pytest.mark.asyncio +async def test_the_successor_can_then_resolve_the_item_it_reclaimed(project): + """The incident's lane could not reach ANY terminal state. Once the + reclaim happens the whole loop is available again, which is what makes a + relaunched lane able to finish (or honestly release) its own work. + """ + item_id, _dead_actor, bd = await _item_held_by_a_dead_lane(project) + assert SV.reap_project(bd)["reclaimed_count"] == 1 + + successor = WorkTrackerSession({"actor": _unique("successor")}) + assert (await successor.claim(project, item_id=item_id)).success is True + resolved = await successor.resolve(item_id, "finished the work the dead lane started") + assert resolved.success is True + assert bd.get(item_id).status == "resolved" + + +@pytest.mark.asyncio +async def test_a_live_sessions_hold_survives_the_same_sweep(project): + """The safety property at the seam: a session that is genuinely running + -- this test process -- holds an item with the SAME silence against the + SAME TTL, and the sweep must leave it strictly alone. Without this, the + fix above would be indistinguishable from "reap everything sooner". + """ + adder = WorkTrackerSession({"actor": _unique("adder")}) + added = await adder.add(project, "work a LIVE lane is holding", acceptance="n/a") + item_id = added.output["added"] # type: ignore[index] + + live = WorkTrackerSession({"actor": _unique("live-lane")}) + claimed = await live.claim(project, item_id=item_id) + assert claimed.success is True + + bd = live._project(project) # noqa: SLF001 + _rewind_custody(bd, item_id, seconds_ago=SILENCE_INSIDE_TTL) + + assert SV.reap_project(bd)["reclaimed_count"] == 0 + assert bd.get(item_id).status == "held" + # The live session is still able to close its own work. + assert (await live.resolve(item_id, "still mine, still working")).success is True diff --git a/src/amplifier_work_tracker/cli.py b/src/amplifier_work_tracker/cli.py index a6aced8..4aece94 100644 --- a/src/amplifier_work_tracker/cli.py +++ b/src/amplifier_work_tracker/cli.py @@ -198,6 +198,42 @@ def _check_sweeps_alive(root) -> contract.Result: return contract.Result("sweeps.alive", all_ok, "; ".join(details)) +def _check_sweeps_reclaiming(root) -> contract.Result: + """Is the reap sweep actually RECLAIMING -- not merely turning? + + `sweeps.alive` proves the loop completes sweeps. It cannot prove the + sweeps do anything, because `supervisor.reap_sweep` catches every + per-project exception into its return value and `reap_loop` used to + discard that value before stamping a completed heartbeat. So a sweep + that errored on every project recorded exactly the same heartbeat as a + perfectly healthy one, and `work_tracker_status` reported + `running_healthy` either way -- the gap named in + `model_performance-oy4`, where "healthy" and "actually reclaiming" were + measured to be different states with no instrument between them. + + The decision itself is `heartbeat.evaluate_reclaiming` (pure, so every + branch is unit-testable with no service). Same dependency-ordering + convention as `_check_sweeps_alive`: skipped, never failed, when the + service isn't installed and running -- a second red line on a root cause + already reported adds no information. + """ + info = S.describe_service() + if not info.supported or not info.installed: + return contract.Result( + "sweeps.reclaiming", + True, + "skipped (service not installed) -- no reap sweep runs until " + "`amplifier-work-tracker service install` (or a foregrounded `serve`) does", + ) + if not info.active: + return contract.Result( + "sweeps.reclaiming", True, "skipped (service.installed already failed)" + ) + record = HB.read_loop_heartbeat(HB.heartbeat_path(root), HB.REAP) + ok, detail = HB.evaluate_reclaiming(record, loop=HB.REAP) + return contract.Result("sweeps.reclaiming", ok, detail) + + def _check_restart_policy(service_check: contract.Result) -> contract.Result: """Does the INSTALLED systemd unit actually carry the self-healing restart policy this project depends on (`Restart=always`, not @@ -312,6 +348,7 @@ def cmd_doctor(a): results.append(_check_systemd_user_bus_reachable()) results.append(_check_dolt_reachable(service_check)) results.append(_check_sweeps_alive(_ws(a).root)) + results.append(_check_sweeps_reclaiming(_ws(a).root)) results.append(_check_restart_policy(service_check)) width = max(len(r.id) for r in results) failed = 0 diff --git a/src/amplifier_work_tracker/custody.py b/src/amplifier_work_tracker/custody.py index a8948fa..8a1da5a 100644 --- a/src/amplifier_work_tracker/custody.py +++ b/src/amplifier_work_tracker/custody.py @@ -25,12 +25,48 @@ and then dies must still go stale and be reclaimed like any other. Its only effect on the reclaim decision is the escalation ceiling below, which is a one-way door toward reclaim, never a hold against it. + +DEAD HOLDERS (model_performance-oy4). Silence is a *proxy* for death, and a +slow one: the TTL cannot fire until a full CUSTODY_TTL_SECONDS after the last +renewal, and the sweep that acts on it runs on its own interval on top of +that. MEASURED on the live queue for item `model_performance-h6v`: its holder +renewed on a perfectly regular 120s cadence up to 2026-09-03T07:41:36Z and +then stopped dead; four successor `work_claim` attempts (07:47, 07:50, 07:51, +07:56Z) were all refused, and all four were refused CORRECTLY -- the last one +landed 45s inside the 900s TTL. The stranding was not a broken TTL, a wrong +field, or a dead sweep: it was the TTL doing exactly what it says while the +one fact that mattered -- the holder's process was gone -- sat unread in the +custody record's own `pid`/`host` fields. + +So there is a THIRD path to reclaim-eligible, and it observes the holder +rather than inferring from its silence: a custody record naming a pid ON THIS +HOST that the kernel says is not running. It is fenced by three conditions so +it can only ever be an ACCELERATION of the TTL, never a way to take work from +a live agent: + + 1. `host` must equal this host. A pid on another machine is unknowable + from here -- never guessed. + 2. `pid` must be a real positive pid. + 3. The custody signal must already have been SILENT for at least + `DEAD_HOLDER_MIN_SILENCE_SECONDS` (default: two renewal intervals). A + live agent renews every RENEW_INTERVAL_SECONDS, so this is independent + corroboration that the holder has ALREADY missed a renewal before any + pid probe is allowed to decide anything -- which is what protects an + agent whose pid is not addressable from here (a container in its own + pid namespace that happens to report the same hostname): it keeps + renewing, so it never enters the window where the probe is consulted. + +Every failure of those conditions resolves to NOT eligible: unknowable is +never treated as dead. The probe is injectable for the same reason +`heartbeat.evaluate_freshness`'s is -- so every branch is testable with no +real processes and no real sleeps. """ from __future__ import annotations import calendar import os +import socket import time from dataclasses import asdict, dataclass @@ -48,6 +84,20 @@ # How often `amplifier-work-tracker custody` renews by default. RENEW_INTERVAL_SECONDS = int(os.environ.get("AMPLIFIER_WORK_TRACKER_RENEW_INTERVAL_SECONDS", "120")) +# How long a custody signal must ALREADY have been silent before a holder- +# liveness probe is allowed to decide anything (see the module docstring's +# condition 3). Two renewal intervals: a live agent renews every +# RENEW_INTERVAL_SECONDS, so crossing this window means it has already missed +# a renewal outright -- independent corroboration, gathered without a probe, +# before any probe is consulted. Always well under CUSTODY_TTL_SECONDS, or +# this path would never accelerate anything. +DEAD_HOLDER_MIN_SILENCE_SECONDS = int( + os.environ.get( + "AMPLIFIER_WORK_TRACKER_DEAD_HOLDER_MIN_SILENCE_SECONDS", + str(2 * RENEW_INTERVAL_SECONDS), + ) +) + STATE_WORKING = "working" STATE_AWAITING_HUMAN = "awaiting_human" VALID_STATES = (STATE_WORKING, STATE_AWAITING_HUMAN) @@ -136,27 +186,117 @@ def is_fresh( return age_seconds(c.last_seen, now=now) <= ttl +def local_host() -> str: + """This host's name, in the SAME form every writer stores in a custody + record's `host` field (`socket.gethostname()` -- see the tool module's + `take_custody` call site and `cli.cmd_custody`). Compared as an exact + string: a mismatch means "not knowable from here", never "dead". + """ + return socket.gethostname() + + +def pid_alive(pid: int) -> bool: + """Best-effort: is *pid* a live process on this host? Never raises. + + `os.kill(pid, 0)` sends no signal -- it only asks the kernel whether the + pid is addressable. A pid owned by another user raises PermissionError, + which PROVES it exists, so that answers True. + + Deliberately a mirror of `heartbeat.pid_alive` rather than an import of + it, for the reason `heartbeat._parse_iso` already states about its own + twin here: this module is the pure domain core and must stay importable + with no dependency on the supervisor/service plumbing stack. + + PID REUSE is the one imprecision, and it is imprecise in the SAFE + direction only: a recycled pid answers True, which merely falls back to + the TTL. It can never manufacture a False for a live holder. + """ + if pid <= 0: + return False + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True # exists, just owned by someone else + except OSError: + return False + return True + + +def holder_process_dead( + custody: dict | Custody | None, + *, + host: str | None = None, + min_silence: int = DEAD_HOLDER_MIN_SILENCE_SECONDS, + now: float | None = None, + is_pid_alive=pid_alive, +) -> tuple[bool, str]: + """Is this custody record's named holder PROVABLY not running? + + Returns (dead, reason). Every unknowable case returns False with an + empty reason -- see the module docstring for the three fences and why + each one resolves toward "not dead". This is the only place in the + system that reads `Custody.pid` / `Custody.host` for a decision. + """ + c = _coerce(custody) + if c is None: + return False, "" + if c.pid <= 0: + return False, "" + this_host = host if host is not None else local_host() + if not c.host or c.host != this_host: + # A pid on another machine says nothing to this one. Never guessed. + return False, "" + silence = age_seconds(c.last_seen, now=now) if c.last_seen else float("inf") + if silence < min_silence: + # Renewed too recently to corroborate death -- a probe here would be + # deciding on the probe alone. See module docstring, condition 3. + return False, "" + if is_pid_alive(c.pid): + return False, "" + return True, ( + f"holder process is dead -- pid {c.pid} on host {c.host!r} is not running, " + f"and custody has been silent {silence:.0f}s " + f"(corroboration window {min_silence}s)" + ) + + def reclaim_eligible( custody: dict | Custody | None, *, ttl: int = CUSTODY_TTL_SECONDS, escalation_hours: float = ESCALATION_HOURS, now: float | None = None, + host: str | None = None, + dead_holder_min_silence: int = DEAD_HOLDER_MIN_SILENCE_SECONDS, + is_pid_alive=pid_alive, ) -> tuple[bool, str]: """The whole reclaim decision, and nothing else decides it. - Two paths to eligible, and only two: + Three paths to eligible, and only three: 1. STALE -- custody was never renewed, or the renewal window lapsed. Total hold duration is irrelevant; only recency of the last renewal matters, so a healthily-renewed 12-hour hold is never touched. - 2. ESCALATION CEILING -- fresh, declaring awaiting_human, but has held + 2. DEAD HOLDER -- the record names a pid on THIS host that the kernel + says is not running, and custody has already been silent long + enough to corroborate it (`holder_process_dead`). Strictly an + ACCELERATION of path 1: it can only ever fire inside the TTL window + path 1 would eventually cover anyway, and only on positive evidence + of death. Added for `model_performance-oy4` -- see the module + docstring for the measured stranding that motivated it. + 3. ESCALATION CEILING -- fresh, declaring awaiting_human, but has held that declaration past `escalation_hours`. A terminal state, not a lock: one unresponsive human cannot immobilize an item forever. - `declared_state` affects ONLY path 2, and only as a ceiling stacked on + `declared_state` affects ONLY path 3, and only as a ceiling stacked on top of freshness -- never as a way to buy exemption from staleness. An item declaring awaiting_human with STALE custody is reclaimed via path 1, - same as any other stale item. + same as any other stale item; one whose process has died is reclaimed via + path 2, likewise regardless of what it declared. + + Path 1 is evaluated FIRST so an already-TTL-stale hold keeps its existing + reason string verbatim (and costs no pid probe at all). """ c = _coerce(custody) if c is None: @@ -164,6 +304,15 @@ def reclaim_eligible( if not is_fresh(c, ttl=ttl, now=now): age = age_seconds(c.last_seen, now=now) return True, f"custody stale -- last seen {age:.0f}s ago (ttl {ttl}s)" + dead, why = holder_process_dead( + c, + host=host, + min_silence=dead_holder_min_silence, + now=now, + is_pid_alive=is_pid_alive, + ) + if dead: + return True, f"{why}; ttl {ttl}s not yet reached, but the holder is gone" if c.declared_state == STATE_AWAITING_HUMAN and c.declared_since: held_hours = age_seconds(c.declared_since, now=now) / 3600.0 if held_hours >= escalation_hours: diff --git a/src/amplifier_work_tracker/heartbeat.py b/src/amplifier_work_tracker/heartbeat.py index 148c2c7..4619392 100644 --- a/src/amplifier_work_tracker/heartbeat.py +++ b/src/amplifier_work_tracker/heartbeat.py @@ -40,7 +40,7 @@ import os import tempfile import time -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, field from pathlib import Path REAP = "reap" @@ -100,11 +100,29 @@ class LoopHeartbeat: loop has started but has not yet finished a sweep -- distinct from "no record at all" (never started), and distinct from "completed, but long ago" (stale). See `evaluate_freshness` for how the three are told apart. + + The last three fields carry the OUTCOME of the most recent sweep, not + merely the fact that one finished (`model_performance-oy4`). Without + them, `reap_loop` recorded a completed sweep whether the sweep reclaimed + everything it should have or errored out on every single project -- + `reap_sweep` catches per-project exceptions into its return value and + that return value was discarded. So `doctor` could report the sweeps + healthy while nothing was being reclaimed at all, which is precisely the + "installed" vs "actually working" gap this whole module exists to close, + reappearing one level up. `evaluate_reclaiming` reads them. + + `failed_projects` absent (rather than empty) distinguishes "a supervisor + older than this change wrote this record" from "a sweep ran and nothing + failed" -- see `evaluate_reclaiming`, which must never report the first + as if it were the second. """ pid: int loop_started_at: str last_completed: str | None = None + projects: int | None = None + reclaimed: int | None = None + failed_projects: list[str] = field(default_factory=list) def _read_all(path: Path) -> dict: @@ -153,12 +171,29 @@ def record_loop_started(path: Path, loop: str, *, pid: int) -> None: _atomic_write(path, data) -def record_sweep_completed(path: Path, loop: str, *, pid: int) -> None: +def record_sweep_completed( + path: Path, + loop: str, + *, + pid: int, + projects: int | None = None, + reclaimed: int | None = None, + failed_projects: list[str] | None = None, +) -> None: """Stamp *loop* as having just completed a sweep. Called AFTER the sweep returns without raising -- this is what proves the sweep actually ran, not merely that the task exists and is sleeping. Preserves `loop_started_at` from `record_loop_started` if present; `pid` is refreshed defensively (should already match). + + `projects` / `reclaimed` / `failed_projects` carry the sweep's OUTCOME + (`model_performance-oy4`). A sweep in which every project raised still + "completes" -- `reap_sweep` catches per-project exceptions -- so without + these, a completed-sweep stamp proves only that the loop is turning, not + that it is doing anything. `failed_projects` is written as `[]` (empty, + present) by any caller that passes it, which is what lets + `evaluate_reclaiming` tell "nothing failed" from "an older supervisor + wrote this record and cannot tell you". """ data = _read_all(path) existing = data.get(loop) @@ -167,9 +202,23 @@ def record_sweep_completed(path: Path, loop: str, *, pid: int) -> None: prior_started = existing.get("loop_started_at") if isinstance(prior_started, str) and prior_started: loop_started_at = prior_started - data[loop] = asdict( - LoopHeartbeat(pid=pid, loop_started_at=loop_started_at, last_completed=now_iso()) + record = asdict( + LoopHeartbeat( + pid=pid, + loop_started_at=loop_started_at, + last_completed=now_iso(), + projects=projects, + reclaimed=reclaimed, + failed_projects=list(failed_projects) if failed_projects is not None else [], + ) ) + if failed_projects is None and projects is None and reclaimed is None: + # Caller reported no outcome at all (e.g. the notify loop, which has + # no reclaim semantics). Do not write an EMPTY `failed_projects`, + # which `evaluate_reclaiming` would read as a positive "nothing + # failed" claim this caller never made. + record.pop("failed_projects", None) + data[loop] = record _atomic_write(path, data) @@ -269,6 +318,56 @@ def evaluate_freshness( ) +def evaluate_reclaiming(record: dict | None, *, loop: str = REAP) -> tuple[bool, str]: + """Is *loop* actually DOING its work, not merely turning? + + `evaluate_freshness` answers "is the loop alive". This answers the + strictly stronger question `model_performance-oy4` was filed against: + `work_tracker_status` reported `running_healthy` while (it appeared) + nothing was being reclaimed, and nothing anywhere could tell those two + states apart. `reap_sweep` catches every per-project exception into its + return value, so a sweep that failed on EVERY project still returns + normally and still stamps a completed heartbeat. + + Returns (ok, detail). Pure -- a dict in, a verdict out. + + Three cases, and the third is the one that must not be fudged: + - failures recorded -> NOT ok, naming every failed project. + - `failed_projects` present and empty -> ok, with the counts. + - `failed_projects` ABSENT -> ok, but the detail says plainly that the + running supervisor predates outcome reporting and cannot answer. + Reported rather than assumed: claiming "0 failed" from a record that + never carried the field would be inventing the very reassurance this + function exists to stop being invented. + """ + if record is None: + return True, ( + f"skipped -- no {loop} heartbeat recorded yet (see the {loop} sweep " + f"liveness check, which reports that directly)" + ) + if "failed_projects" not in record: + return True, ( + f"unknown -- the running supervisor predates per-sweep outcome reporting, " + f"so its {loop} heartbeat cannot say whether any project failed; restart the " + f"service (`amplifier-work-tracker service restart`) to start recording it" + ) + failed = record.get("failed_projects") or [] + projects = record.get("projects") + reclaimed = record.get("reclaimed") + scope = f"{projects} project(s)" if isinstance(projects, int) else "an unknown project count" + got = f"{reclaimed} reclaimed" if isinstance(reclaimed, int) else "reclaim count unknown" + if failed: + names = ", ".join(str(f) for f in failed[:10]) + more = f" (+{len(failed) - 10} more)" if len(failed) > 10 else "" + return False, ( + f"the last {loop} sweep swept {scope} and FAILED on {len(failed)}: {names}{more} " + f"-- the loop is alive but is not reclaiming in those projects, so a stale hold " + f"there will never be released; check the service log " + f"(`journalctl --user -u amplifier-work-tracker`) for the per-project error" + ) + return True, f"last {loop} sweep: {scope}, 0 failed, {got}" + + __all__ = [ "DEFAULT_STALE_MULTIPLE", "HEARTBEAT_FILENAME", @@ -277,6 +376,7 @@ def evaluate_freshness( "NOTIFY", "REAP", "evaluate_freshness", + "evaluate_reclaiming", "heartbeat_path", "now_iso", "pid_alive", diff --git a/src/amplifier_work_tracker/supervisor.py b/src/amplifier_work_tracker/supervisor.py index af29700..c23bd69 100644 --- a/src/amplifier_work_tracker/supervisor.py +++ b/src/amplifier_work_tracker/supervisor.py @@ -108,21 +108,47 @@ def reap_project( ttl_seconds: int | None = None, escalation_hours: float | None = None, ) -> dict[str, Any]: - """Release items in *one* project whose custody has gone stale or hit the - escalation ceiling. Identical logic to `cli.cmd_reap` -- extracted here so - the sweep (below) and the single-project CLI command can never drift - apart on what "reap" means. + """Release items in *one* project whose custody has gone stale, whose + holder process is provably dead, or which has hit the escalation ceiling. + Identical logic to `cli.cmd_reap` -- extracted here so the sweep (below) + and the single-project CLI command can never drift apart on what "reap" + means. + + Reads with an EXPLICIT `status="held"` and `limit=0` (unlimited). It used + to read `bd.list(include_resolved=False)` and filter in Python -- but + `Beads.list()` with no `limit` applies bd's own default cap of + `LIST_DEFAULT_LIMIT` (50), ordered `priority ASC, created_at DESC, id + ASC`. In a project with more than 50 non-closed items, a held item + outside that first page was invisible to the reaper *permanently*, with + nothing anywhere reporting that it had been skipped. Not the cause of + `model_performance-oy4` (that project held 20-22 non-closed items at the + time, measured -- the stale hold ranked 2nd-4th), but a live silent-miss + on any busier queue, found while root-causing it. + + Per-item isolation: a single item whose `release` raises must not abort + the reap of every OTHER stale hold in the project. Before this, one + wedged item deterministically shadowed the rest of the queue on every + sweep, forever, while the sweep still reported itself completed. Failures + are returned in `failed` -- named, never swallowed -- and propagate up + through `reap_sweep` into the heartbeat that `doctor`'s + `sweeps.reclaiming` check reads. """ ttl = ttl_seconds if ttl_seconds is not None else C.CUSTODY_TTL_SECONDS esc = escalation_hours if escalation_hours is not None else C.ESCALATION_HOURS - held = [i for i in bd.list(include_resolved=False) if i.status == "held"] + held = [i for i in bd.list(status="held", limit=0) if i.status == "held"] reclaimed: list[dict[str, Any]] = [] kept: list[dict[str, Any]] = [] + failed: list[dict[str, Any]] = [] for item in held: rec = item.meta.get(C.CUSTODY_KEY) eligible, reason = C.reclaim_eligible(rec, ttl=ttl, escalation_hours=esc) if eligible: - bd.release(item.id) + try: + bd.release(item.id) + except Exception as e: # noqa: BLE001 -- one wedged item must never shadow the queue + logger.exception("reap could not release %s -- continuing with the rest", item.id) + failed.append({"id": item.id, "holder": item.holder, "error": str(e)}) + continue reclaimed.append({"id": item.id, "was_holder": item.holder, "reason": reason}) # ALARM: custody-TTL breach is a real alarm condition. Sync, never raises # (a push failure must never prevent/undo the reclaim above); any failure @@ -131,7 +157,13 @@ def reap_project( else: note = "quiet (awaiting_human)" if not C.should_notify(rec) else "ok" kept.append({"id": item.id, "holder": item.holder, "note": note}) - return {"reclaimed": reclaimed, "reclaimed_count": len(reclaimed), "kept": kept} + return { + "reclaimed": reclaimed, + "reclaimed_count": len(reclaimed), + "kept": kept, + "failed": failed, + "failed_count": len(failed), + } def notify_project(bd: A.Beads) -> dict[str, Any]: @@ -178,6 +210,26 @@ def reap_sweep( return out +def sweep_failures(result: dict[str, dict[str, Any]]) -> list[str]: + """Every project name in a `reap_sweep` result that did NOT fully do its + job -- either the whole project raised (`{"error": ...}`, caught per + project so one broken project cannot abort the sweep) or an individual + item's release failed (`failed_count`, see `reap_project`). + + Pure, so it is testable without a sweep. This is the one place that + decides what "the sweep failed here" means, shared by the loop's log line + and the heartbeat the `sweeps.reclaiming` doctor check reads -- the two + can never disagree about what counts (`model_performance-oy4`). + """ + names: list[str] = [] + for name, r in result.items(): + if not isinstance(r, dict): + names.append(name) + elif r.get("error") is not None or int(r.get("failed_count", 0) or 0) > 0: + names.append(name) + return names + + def notify_sweep(ws: A.Workspace) -> dict[str, dict[str, Any]]: """Sweep `notify_project` across every known project. See `reap_sweep`'s docstring -- same per-project isolation.""" @@ -222,10 +274,28 @@ async def reap_loop( if stop_event.is_set(): return try: - await asyncio.to_thread( + result = await asyncio.to_thread( reap_sweep, ws, ttl_seconds=ttl_seconds, escalation_hours=escalation_hours ) - HB.record_sweep_completed(hb_path, HB.REAP, pid=os.getpid()) + failed = sorted(sweep_failures(result)) + if failed: + logger.error( + "reap sweep completed but FAILED on %d project(s): %s", + len(failed), + ", ".join(failed), + ) + HB.record_sweep_completed( + hb_path, + HB.REAP, + pid=os.getpid(), + projects=len(result), + reclaimed=sum( + int(r.get("reclaimed_count", 0) or 0) + for r in result.values() + if isinstance(r, dict) + ), + failed_projects=failed, + ) except Exception: # noqa: BLE001 -- see docstring logger.exception("reap sweep crashed -- continuing on the next interval") @@ -1005,4 +1075,5 @@ def serve( "read_owned_pid", "serve", "spawn_dolt", + "sweep_failures", ] diff --git a/tests/integration/test_dead_holder_reclaim.py b/tests/integration/test_dead_holder_reclaim.py new file mode 100644 index 0000000..4115a19 --- /dev/null +++ b/tests/integration/test_dead_holder_reclaim.py @@ -0,0 +1,262 @@ +"""Tier 2 -- a hold whose holder PROCESS IS DEAD is reclaimed automatically, +end to end, against real `bd` and a real dolt server (`model_performance-oy4`). + +The unit tier (`tests/unit/test_custody_dead_holder.py`) pins the decision +and its three fences with an injected probe. This file pins the whole chain +with the REAL probe against a REAL exited process: custody record -> the +`held_stale` a reporting caller sees -> the sweep that acts on it -> and the +property that actually unblocks a relaunched lane, which is that its +successor can then claim the item instead of being refused indefinitely with +"already claimed by ". + +Every "dead" pid here belongs to a subprocess this test started and reaped, +so the death is real rather than a number chosen for looking implausible. + +Also pinned here, both found while root-causing oy4 and both silent misses +in the reaper itself: + + - `reap_project` used bd's DEFAULT list page (50 items, ordered + `priority ASC, created_at DESC, id ASC`). A held item outside that page + was invisible to the reaper permanently, and nothing reported the skip. + - one item whose `release` raised aborted the reap of every remaining + held item in that project -- deterministically, on every sweep, forever, + while the sweep still reported itself completed. +""" + +from __future__ import annotations + +import json +import socket +import subprocess +import sys +import time + +import pytest + +from amplifier_work_tracker import adapter as A +from amplifier_work_tracker import custody as C +from amplifier_work_tracker import supervisor as SV + +pytestmark = pytest.mark.integration + +HOST = socket.gethostname() + +#: Silence used for every "dead holder" case below: comfortably past the +#: corroboration window (240s) and comfortably INSIDE the 900s TTL, so a +#: reclaim here can only have come from the liveness path -- never from +#: ordinary staleness. This gap is the whole point: before this change the +#: item sat unclaimable for the remaining ~10 minutes of its TTL plus up to +#: a full sweep interval on top. +SILENCE_INSIDE_TTL = 400 + + +def _a_genuinely_dead_pid() -> int: + """A pid that really did exist and really has exited. `wait()` reaps it, + so it is not a zombie still addressable by `os.kill(pid, 0)`. + """ + p = subprocess.Popen([sys.executable, "-c", "pass"]) # noqa: S603 + p.wait() + assert not C.pid_alive(p.pid), ( + f"pid {p.pid} is still addressable after wait() -- the OS recycled it " + f"mid-test; rerun (this is the documented pid-reuse imprecision, and it " + f"errs toward NOT reclaiming)" + ) + return p.pid + + +def _rewind_custody(bd: A.Beads, item_id: str, *, seconds_ago: int) -> dict: + """Age an existing custody record's `last_seen` without sleeping. Writes + through bd's own metadata merge, so the record read back is the one a + real renewal would have left behind `seconds_ago` seconds ago. + """ + rec = dict(bd.get(item_id).meta[C.CUSTODY_KEY]) + rec["last_seen"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - seconds_ago)) + bd._run( # noqa: SLF001 -- deliberate: forging a past renewal is not a public verb + ["update", item_id, "--metadata", json.dumps({C.CUSTODY_KEY: rec})], + actor=rec["holder"], + ) + return bd.get(item_id).meta[C.CUSTODY_KEY] + + +def _held_by_a_dead_holder(bd: A.Beads, *, title: str, actor: str, priority: int = 1) -> str: + item_id = bd.create(title, priority=priority) + bd.claim_item(item_id, actor=actor) + bd.take_custody(item_id, holder=actor, pid=_a_genuinely_dead_pid(), host=HOST) + _rewind_custody(bd, item_id, seconds_ago=SILENCE_INSIDE_TTL) + return item_id + + +# -------------------------------------------------------------------------- +# The flagship: dead holder -> stale -> reclaimed -> successor claims. +# -------------------------------------------------------------------------- + + +def test_dead_holders_hold_is_reported_stale_inside_the_ttl(workspace, project_factory): + """`work_stats`/`work_status` must SEE it. During the oy4 incident this + field read `held_stale: 0` with a `held_stale_oldest_age_seconds: null` + beside it, which is what told the successor session there was nothing to + wait for. + """ + name, bd = project_factory("deadhold") + _held_by_a_dead_holder(bd, title="dead holder: reported stale", actor="dead-agent") + + summary = A.project_summary(workspace, name) + assert summary.held == 1 + assert summary.held_stale == 1, ( + "a hold whose holder process is gone must be reported stale before its " + "TTL expires -- otherwise a successor session has no signal at all" + ) + assert summary.held_stale_oldest_age_seconds is not None + assert summary.held_stale_oldest_age_seconds >= SILENCE_INSIDE_TTL - 60 + # And it is genuinely INSIDE the TTL -- so this is the liveness path, + # not staleness arriving early. + assert summary.held_stale_oldest_age_seconds < C.CUSTODY_TTL_SECONDS + + +def test_the_sweep_reclaims_a_dead_holders_hold_with_the_default_ttl(project_factory): + """The real sweep entry point, with NO ttl override -- exactly what the + background service calls every `DEFAULT_REAP_INTERVAL_SECONDS`. + """ + _name, bd = project_factory("deadhold") + item_id = _held_by_a_dead_holder(bd, title="dead holder: reaped", actor="dead-agent") + + result = SV.reap_project(bd) + + assert result["reclaimed_count"] == 1, result + assert result["reclaimed"][0]["id"] == item_id + assert result["reclaimed"][0]["was_holder"] == "dead-agent" + assert "holder process is dead" in result["reclaimed"][0]["reason"] + assert result["failed_count"] == 0 + + after = bd.get(item_id) + assert after.status == "open" + assert after.holder is None + + +def test_a_successor_session_can_claim_the_item_after_the_reclaim(project_factory): + """THE property that actually unblocks a relaunched lane. In the measured + incident the successor could not `work_claim` (held by a dead agent), + could not `work_release` (it did not hold it) and could not `work_file` + (filing requires holding an item) -- so it could do nothing but wait for + a human-equivalent intervention. + """ + _name, bd = project_factory("deadhold") + item_id = _held_by_a_dead_holder(bd, title="dead holder: successor", actor="dead-agent") + + # Before the reclaim, the successor is refused -- and refused by NAME, + # which is the message the incident reported four times. + with pytest.raises(A.BeadsError) as refused: + bd.claim_item(item_id, actor="successor-agent") + assert "dead-agent" in str(refused.value) + + assert SV.reap_project(bd)["reclaimed_count"] == 1 + + back = bd.claim_item(item_id, actor="successor-agent") + assert back.status == "held" + assert back.holder == "successor-agent" + + +def test_a_live_holder_inside_the_ttl_is_never_reclaimed(project_factory): + """The safety property, with the REAL probe: this test's own process is + alive, so an identically-silent hold naming it must survive the sweep. + Same silence, same TTL, same code path -- only liveness differs. + """ + _name, bd = project_factory("livehold") + item_id = bd.create("live holder must survive", priority=1) + bd.claim_item(item_id, actor="live-agent") + bd.take_custody(item_id, holder="live-agent", pid=__import__("os").getpid(), host=HOST) + _rewind_custody(bd, item_id, seconds_ago=SILENCE_INSIDE_TTL) + + result = SV.reap_project(bd) + + assert result["reclaimed_count"] == 0, result + assert bd.get(item_id).status == "held" + assert bd.get(item_id).holder == "live-agent" + + +def test_a_holder_recorded_on_another_host_is_never_reclaimed(project_factory): + """A dead pid number means nothing across machines: the same number may + well be a live process here. Unknowable resolves to "leave it alone", + and the TTL still covers it in the end. + """ + _name, bd = project_factory("remotehold") + item_id = bd.create("remote holder must survive", priority=1) + bd.claim_item(item_id, actor="remote-agent") + bd.take_custody( + item_id, holder="remote-agent", pid=_a_genuinely_dead_pid(), host="some-other-box" + ) + _rewind_custody(bd, item_id, seconds_ago=SILENCE_INSIDE_TTL) + + assert SV.reap_project(bd)["reclaimed_count"] == 0 + assert bd.get(item_id).status == "held" + + +# -------------------------------------------------------------------------- +# Two silent misses in the reaper itself, found while root-causing oy4. +# -------------------------------------------------------------------------- + + +def test_the_reaper_does_not_depend_on_bds_default_list_page(project_factory, monkeypatch): + """`reap_project` used to read `bd.list(include_resolved=False)`, which + applies bd's own default cap (`LIST_DEFAULT_LIMIT`, 50) ordered + `priority ASC, created_at DESC, id ASC`. A held item outside that first + page was invisible to the reaper permanently -- silently, with nothing + anywhere reporting it had been skipped. + + The page size is shrunk here rather than creating 51 real items: it is + the same constant, read by the same call, so this exercises the actual + mechanism at a fraction of the cost. The stale hold is given the WORST + priority so it sorts off the end of the page. + """ + _name, bd = project_factory("pagedreap") + item_id = _held_by_a_dead_holder( + bd, title="stale hold, worst priority", actor="dead-agent", priority=4 + ) + for n in range(3): + bd.create(f"filler {n}", priority=0) + + monkeypatch.setattr(A, "LIST_DEFAULT_LIMIT", 2) + + # FAIL-BEFORE, made explicit: the old read genuinely cannot see it. + off_page = bd.list(include_resolved=False) + assert item_id not in {i.id for i in off_page}, ( + "test setup is wrong -- the stale hold must be off the first page for " + "this regression to mean anything" + ) + + assert SV.reap_project(bd)["reclaimed_count"] == 1 + assert bd.get(item_id).status == "open" + + +def test_one_unreleasable_item_does_not_shadow_the_rest_of_the_queue(project_factory, monkeypatch): + """Before this, the loop in `reap_project` had no per-item guard: the + first `release` that raised propagated out, `reap_sweep` caught it per + project, and every remaining stale hold in that project went unreaped -- + on that sweep and on every sweep after it, since the failure is + deterministic. The sweep still recorded itself completed. + """ + _name, bd = project_factory("wedgedreap") + wedged = _held_by_a_dead_holder(bd, title="wedged hold", actor="dead-agent-a") + other = _held_by_a_dead_holder(bd, title="second stale hold", actor="dead-agent-b") + + real_release = A.Beads.release + + def _release(self, item_id): + if item_id == wedged: + raise A.BeadsError("simulated wedged release") + return real_release(self, item_id) + + monkeypatch.setattr(A.Beads, "release", _release) + + result = SV.reap_project(bd) + + assert result["reclaimed_count"] == 1 + assert result["reclaimed"][0]["id"] == other + assert result["failed_count"] == 1 + assert result["failed"][0]["id"] == wedged + assert "simulated wedged release" in result["failed"][0]["error"] + # The healthy item really was freed, not merely reported. + assert bd.get(other).status == "open" + # And the failure is visible to the sweep-level reporting the + # `sweeps.reclaiming` doctor check reads -- never swallowed. + assert SV.sweep_failures({"wedgedreap": result}) == ["wedgedreap"] diff --git a/tests/unit/test_custody_dead_holder.py b/tests/unit/test_custody_dead_holder.py new file mode 100644 index 0000000..afcc83c --- /dev/null +++ b/tests/unit/test_custody_dead_holder.py @@ -0,0 +1,240 @@ +"""Tier 1 -- the DEAD-HOLDER reclaim path (`model_performance-oy4`). + +WHAT THIS PINS, and why it is not just "another staleness test". Custody's +freshness rule infers liveness from silence: a hold is reclaim-eligible +`CUSTODY_TTL_SECONDS` (900s) after its last renewal, and the sweep that acts +on it runs on its own interval on top of that -- up to 20 minutes before a +dead lane's item returns to the queue. Meanwhile the custody record has +carried `pid` and `host` since it was designed, and nothing has ever read +them for a decision. + +MEASURED, on the live queue, item `model_performance-h6v` (the forensic +timeline is committed at +`docs/lanes/oy4-dead-holder-reclaim/evidence/h6v-forensic-timeline.txt`): +its holder renewed on a metronome-regular 120s cadence through +2026-09-03T07:41:36Z and then stopped. Four successor `work_claim` attempts +-- 07:47, 07:50, 07:51 and 07:56Z -- were all refused, and every one of them +was refused CORRECTLY: the last landed 45 seconds inside the 900s TTL. The +item was finally freed by a hand-run `unclaim` at 07:57:51Z. Nothing was +broken; the TTL was doing exactly what it says, blind to the one fact that +settled the matter. + +So these tests fix the ACCELERATION and, just as importantly, its three +fences. The fences are the whole safety argument: this path may only ever +fire on positive evidence of death, and every unknowable case must resolve +to NOT eligible, because a false positive here takes work away from a live +agent. +""" + +from __future__ import annotations + +import time + +from amplifier_work_tracker import custody as C + +HOST = "test-host" + + +def _ts(seconds_ago: float) -> str: + return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - seconds_ago)) + + +def _record(*, last_seen_ago: float, pid: int = 4242, host: str = HOST) -> dict: + return { + "holder": "agent-test-4242", + "pid": pid, + "host": host, + "generation": 1, + "started_at": _ts(last_seen_ago + 600), + "last_seen": _ts(last_seen_ago), + "declared_state": C.STATE_WORKING, + "declared_since": _ts(last_seen_ago + 600), + } + + +def _dead(_pid: int) -> bool: + return False + + +def _alive(_pid: int) -> bool: + return True + + +# -------------------------------------------------------------------------- +# The acceleration itself. +# -------------------------------------------------------------------------- + + +def test_dead_holder_is_reclaim_eligible_well_inside_the_ttl(): + """THE FAIL-BEFORE. 300s of silence against a 900s TTL: two thirds of the + TTL still to run, so path 1 cannot fire and (before this change) nothing + else could either. The holder's pid is not running on this very host -- + that is enough. + """ + rec = _record(last_seen_ago=300) + eligible, reason = C.reclaim_eligible(rec, ttl=900, host=HOST, is_pid_alive=_dead) + assert eligible is True + assert "holder process is dead" in reason + assert "pid 4242" in reason + # The reason must say the TTL was NOT what fired, or a reader will + # mis-attribute the reclaim to ordinary staleness. + assert "ttl 900s not yet reached" in reason + + +def test_dead_holder_reason_is_distinguishable_from_ordinary_staleness(): + """A reclaim's reason is read by humans triaging a stranded lane, and is + what `reap_project` records. "Died" and "went quiet" are different + diagnoses and must never print the same sentence. + """ + dead = C.reclaim_eligible(_record(last_seen_ago=300), ttl=900, host=HOST, is_pid_alive=_dead)[1] + stale = C.reclaim_eligible(_record(last_seen_ago=1200), ttl=900, host=HOST)[1] + assert dead != stale + assert stale.startswith("custody stale") + assert dead.startswith("holder process is dead") + + +def test_ttl_staleness_still_wins_and_costs_no_pid_probe(): + """Path 1 is evaluated first, so an already-stale hold keeps its exact + prior reason string -- and a probe that would explode is never called, + proving the ordering rather than asserting it. + """ + + def _explode(_pid: int) -> bool: # pragma: no cover - must never run + raise AssertionError("pid probe consulted for an already-TTL-stale hold") + + eligible, reason = C.reclaim_eligible( + _record(last_seen_ago=1200), ttl=900, host=HOST, is_pid_alive=_explode + ) + assert eligible is True + assert reason.startswith("custody stale -- last seen") + + +# -------------------------------------------------------------------------- +# Fence 1: another host is unknowable, never dead. +# -------------------------------------------------------------------------- + + +def test_holder_on_another_host_is_never_called_dead(): + """A pid means nothing across machines. The probe would answer for a + LOCAL pid of the same number -- a coincidence that must never reclaim a + live remote agent's work -- so the host check has to come first. + """ + rec = _record(last_seen_ago=300, host="some-other-box") + eligible, _ = C.reclaim_eligible(rec, ttl=900, host=HOST, is_pid_alive=_dead) + assert eligible is False + + +def test_holder_with_no_host_recorded_is_never_called_dead(): + rec = _record(last_seen_ago=300, host="") + assert C.reclaim_eligible(rec, ttl=900, host=HOST, is_pid_alive=_dead)[0] is False + + +# -------------------------------------------------------------------------- +# Fence 2: a pid that is not a pid. +# -------------------------------------------------------------------------- + + +def test_missing_or_zero_pid_is_never_called_dead(): + for pid in (0, -1): + rec = _record(last_seen_ago=300, pid=pid) + assert C.reclaim_eligible(rec, ttl=900, host=HOST, is_pid_alive=_dead)[0] is False + + +# -------------------------------------------------------------------------- +# Fence 3: the corroboration window. This is the fence that protects an +# agent whose pid simply is not addressable from here (a container with its +# own pid namespace reporting the same hostname): such an agent keeps +# renewing, so it never enters the window where the probe is consulted at +# all. +# -------------------------------------------------------------------------- + + +def test_a_recently_renewed_hold_is_never_probed_into_death(): + """One renewal interval of silence is normal operation, not death.""" + rec = _record(last_seen_ago=60) + eligible, _ = C.reclaim_eligible( + rec, ttl=900, host=HOST, dead_holder_min_silence=240, is_pid_alive=_dead + ) + assert eligible is False + + +def test_the_corroboration_window_boundary_is_inclusive_from_min_silence(): + rec_just_under = _record(last_seen_ago=239) + rec_just_over = _record(last_seen_ago=241) + kw = {"ttl": 900, "host": HOST, "dead_holder_min_silence": 240, "is_pid_alive": _dead} + assert C.reclaim_eligible(rec_just_under, **kw)[0] is False + assert C.reclaim_eligible(rec_just_over, **kw)[0] is True + + +def test_default_corroboration_window_is_two_renewal_intervals(): + """The default must stay tied to the renewal cadence, not a magic number + -- otherwise a deployment that changes the renewal interval silently + changes how aggressive this path is. + """ + assert C.DEAD_HOLDER_MIN_SILENCE_SECONDS == 2 * C.RENEW_INTERVAL_SECONDS + assert C.DEAD_HOLDER_MIN_SILENCE_SECONDS < C.CUSTODY_TTL_SECONDS + + +# -------------------------------------------------------------------------- +# A live holder is never touched -- the property everything above exists to +# protect. +# -------------------------------------------------------------------------- + + +def test_a_live_holder_deep_inside_the_ttl_is_left_alone(): + rec = _record(last_seen_ago=300) + assert C.reclaim_eligible(rec, ttl=900, host=HOST, is_pid_alive=_alive)[0] is False + + +def test_pid_reuse_errs_toward_leaving_the_hold_alone(): + """A recycled pid answers "alive", which merely falls back to the TTL. + Imprecision only ever in the safe direction. + """ + rec = _record(last_seen_ago=800) + assert C.reclaim_eligible(rec, ttl=900, host=HOST, is_pid_alive=_alive)[0] is False + + +def test_awaiting_human_does_not_shield_a_dead_holder(): + """`declared_state` is reporting only -- it never buys exemption from + staleness, and it must not buy exemption from death either. + """ + rec = _record(last_seen_ago=300) + rec["declared_state"] = C.STATE_AWAITING_HUMAN + rec["declared_since"] = _ts(400) + assert C.reclaim_eligible(rec, ttl=900, host=HOST, is_pid_alive=_dead)[0] is True + + +# -------------------------------------------------------------------------- +# The probe itself. +# -------------------------------------------------------------------------- + + +def test_pid_alive_reports_this_process_alive_and_a_nonexistent_pid_dead(): + import os + + assert C.pid_alive(os.getpid()) is True + assert C.pid_alive(0) is False + assert C.pid_alive(-5) is False + + +def test_local_host_matches_what_writers_store(): + import socket + + assert C.local_host() == socket.gethostname() + + +def test_holder_process_dead_returns_empty_reason_when_not_dead(): + """Every "not dead" answer must carry an empty reason, so a caller can + never print a half-built explanation for a decision that did not fire. + """ + dead, reason = C.holder_process_dead( + _record(last_seen_ago=60), host=HOST, min_silence=240, is_pid_alive=_dead + ) + assert (dead, reason) == (False, "") + + +def test_no_custody_record_is_not_a_dead_holder_claim(): + """`reclaim_eligible(None)` already reports "claimed but never renewed"; + the dead-holder helper must not also claim a process it never saw. + """ + assert C.holder_process_dead(None) == (False, "") diff --git a/tests/unit/test_sweeps_reclaiming.py b/tests/unit/test_sweeps_reclaiming.py new file mode 100644 index 0000000..ba2a843 --- /dev/null +++ b/tests/unit/test_sweeps_reclaiming.py @@ -0,0 +1,260 @@ +"""Tier 1 -- `sweeps.reclaiming`: the instrument that tells "the reap loop +is turning" apart from "the reap loop is doing anything" (`model_performance-oy4`). + +THE GAP THIS CLOSES. `supervisor.reap_sweep` catches every per-project +exception into its return value so one broken project cannot abort the sweep +-- correct, and the reason a sweep in which EVERY project raised still +returns normally. `reap_loop` then discarded that return value and stamped a +completed heartbeat regardless. `sweeps.alive` reads that stamp, so it +reported healthy either way, and `work_tracker_status` reported +`running_healthy` on top of it. During the `model_performance-oy4` incident +that pairing was observed for 23 minutes while (it appeared) nothing was +being reclaimed, and no instrument anywhere could separate the two states. + +Three layers are pinned here, all pure -- no service, no sleeps, no sweeps: + - `supervisor.sweep_failures`: what counts as "the sweep failed here". + - `heartbeat.evaluate_reclaiming`: the verdict, including the case that + must not be fudged -- a record from a supervisor too old to know. + - `cli._check_sweeps_reclaiming`: the doctor wiring and its skip rules. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass + +from amplifier_work_tracker import cli +from amplifier_work_tracker import heartbeat as HB +from amplifier_work_tracker import service as S +from amplifier_work_tracker import supervisor as SV + + +def _ts(seconds_ago: float) -> str: + return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - seconds_ago)) + + +@dataclass +class _Info: + supported: bool = True + installed: bool = True + active: bool | None = True + unit_path = None + detail: str = "installed and active" + platform: str = "linux" + + +# -------------------------------------------------------------------------- +# supervisor.sweep_failures -- one definition of "failed here". +# -------------------------------------------------------------------------- + + +def test_sweep_failures_names_a_project_that_raised_wholesale(): + result = {"good": {"reclaimed_count": 1, "failed_count": 0}, "broken": {"error": "boom"}} + assert SV.sweep_failures(result) == ["broken"] + + +def test_sweep_failures_names_a_project_whose_individual_release_failed(): + """A project that swept fine except for one item whose release raised is + still not doing its job for that item -- and that item is exactly the + stranded hold this whole feature exists to free. + """ + result = { + "ok": {"reclaimed_count": 2, "failed_count": 0}, + "partly": {"reclaimed_count": 1, "failed_count": 1, "failed": [{"id": "x-1"}]}, + } + assert SV.sweep_failures(result) == ["partly"] + + +def test_sweep_failures_is_empty_for_a_clean_sweep(): + assert SV.sweep_failures({"a": {"reclaimed_count": 0, "failed_count": 0}}) == [] + + +def test_sweep_failures_treats_a_non_dict_result_as_a_failure(): + """Defensive: an unexpected shape is reported, never silently read as + success -- the failure mode this whole item is about. + """ + assert SV.sweep_failures({"weird": None}) == ["weird"] # type: ignore[dict-item] + + +# -------------------------------------------------------------------------- +# heartbeat.evaluate_reclaiming -- the verdict. +# -------------------------------------------------------------------------- + + +def test_clean_sweep_is_ok_and_reports_the_counts(): + ok, detail = HB.evaluate_reclaiming( + { + "pid": 1, + "loop_started_at": _ts(600), + "last_completed": _ts(10), + "projects": 32, + "reclaimed": 0, + "failed_projects": [], + } + ) + assert ok is True + assert "32 project(s)" in detail + assert "0 failed" in detail + + +def test_a_sweep_with_failed_projects_fails_loudly_and_names_them(): + """THE FAIL-BEFORE for the doctor half: before this, exactly this record + was indistinguishable from the clean one above. + """ + ok, detail = HB.evaluate_reclaiming( + { + "pid": 1, + "loop_started_at": _ts(600), + "last_completed": _ts(10), + "projects": 32, + "reclaimed": 0, + "failed_projects": ["model_performance", "cortex"], + } + ) + assert ok is False + assert "model_performance" in detail + assert "cortex" in detail + assert "not reclaiming" in detail + # A failure detail must carry its own next step, per this repo's + # error-visibility convention. + assert "journalctl" in detail + + +def test_many_failed_projects_are_truncated_but_counted_honestly(): + names = [f"p{i}" for i in range(14)] + ok, detail = HB.evaluate_reclaiming( + { + "pid": 1, + "loop_started_at": _ts(600), + "last_completed": _ts(10), + "projects": 20, + "reclaimed": 0, + "failed_projects": names, + } + ) + assert ok is False + assert "FAILED on 14" in detail + assert "+4 more" in detail + + +def test_a_record_from_an_older_supervisor_says_unknown_not_zero(): + """The case that must not be fudged. A heartbeat with no + `failed_projects` key was written by a supervisor that never recorded + one -- reporting it as "0 failed" would invent precisely the reassurance + this check exists to stop being invented. + """ + ok, detail = HB.evaluate_reclaiming( + {"pid": 1, "loop_started_at": _ts(600), "last_completed": _ts(10)} + ) + assert ok is True + assert "unknown" in detail + assert "predates" in detail + assert "restart the service" in detail + + +def test_no_record_defers_to_the_liveness_check(): + ok, detail = HB.evaluate_reclaiming(None) + assert ok is True + assert "skipped" in detail + + +def test_notify_loop_completion_does_not_claim_zero_failures(tmp_path): + """`record_sweep_completed` with no outcome arguments (the notify loop, + which has no reclaim semantics) must NOT write an empty + `failed_projects`, or the reap check would read a claim nobody made. + """ + path = HB.heartbeat_path(tmp_path) + HB.record_loop_started(path, HB.NOTIFY, pid=1) + HB.record_sweep_completed(path, HB.NOTIFY, pid=1) + rec = HB.read_loop_heartbeat(path, HB.NOTIFY) + assert rec is not None + assert "failed_projects" not in rec + + +def test_reap_loop_outcome_round_trips_through_the_heartbeat_file(tmp_path): + path = HB.heartbeat_path(tmp_path) + HB.record_loop_started(path, HB.REAP, pid=1) + HB.record_sweep_completed( + path, HB.REAP, pid=1, projects=3, reclaimed=2, failed_projects=["bad"] + ) + rec = HB.read_loop_heartbeat(path, HB.REAP) + assert rec is not None + assert rec["projects"] == 3 + assert rec["reclaimed"] == 2 + assert rec["failed_projects"] == ["bad"] + assert HB.evaluate_reclaiming(rec)[0] is False + + +def test_recording_an_outcome_does_not_break_the_liveness_check(tmp_path): + """The two checks read the same record; adding fields to it must not + disturb `evaluate_freshness`, which existing behaviour depends on. + """ + path = HB.heartbeat_path(tmp_path) + HB.record_loop_started(path, HB.REAP, pid=1) + HB.record_sweep_completed(path, HB.REAP, pid=1, projects=1, reclaimed=0, failed_projects=[]) + rec = HB.read_loop_heartbeat(path, HB.REAP) + ok, _ = HB.evaluate_freshness( + rec, loop=HB.REAP, interval=SV.DEFAULT_REAP_INTERVAL_SECONDS, is_pid_alive=lambda _p: True + ) + assert ok is True + + +# -------------------------------------------------------------------------- +# cli._check_sweeps_reclaiming -- the doctor wiring. +# -------------------------------------------------------------------------- + + +def test_check_is_skipped_when_the_service_is_not_installed(monkeypatch, tmp_path): + monkeypatch.setattr(S, "describe_service", lambda: _Info(installed=False, active=None)) + result = cli._check_sweeps_reclaiming(tmp_path) + assert result.id == "sweeps.reclaiming" + assert result.ok is True + assert "skipped" in result.detail + + +def test_check_is_skipped_when_the_service_is_installed_but_inactive(monkeypatch, tmp_path): + monkeypatch.setattr(S, "describe_service", lambda: _Info(installed=True, active=False)) + result = cli._check_sweeps_reclaiming(tmp_path) + assert result.ok is True + assert "service.installed already failed" in result.detail + + +def test_check_fails_when_the_last_sweep_failed_on_a_project(monkeypatch, tmp_path): + monkeypatch.setattr(S, "describe_service", lambda: _Info(installed=True, active=True)) + path = HB.heartbeat_path(tmp_path) + HB.record_loop_started(path, HB.REAP, pid=1) + HB.record_sweep_completed( + path, HB.REAP, pid=1, projects=4, reclaimed=0, failed_projects=["model_performance"] + ) + result = cli._check_sweeps_reclaiming(tmp_path) + assert result.ok is False + assert "model_performance" in result.detail + + +def test_check_passes_on_a_clean_sweep(monkeypatch, tmp_path): + monkeypatch.setattr(S, "describe_service", lambda: _Info(installed=True, active=True)) + path = HB.heartbeat_path(tmp_path) + HB.record_loop_started(path, HB.REAP, pid=1) + HB.record_sweep_completed(path, HB.REAP, pid=1, projects=4, reclaimed=1, failed_projects=[]) + result = cli._check_sweeps_reclaiming(tmp_path) + assert result.ok is True + + +def test_sweeps_alive_and_sweeps_reclaiming_are_two_distinct_assumptions(monkeypatch, tmp_path): + """The whole point: a heartbeat can be FRESH (loop alive) and still prove + the sweep is not reclaiming. Both verdicts are computed from the same + record and must disagree here. + """ + monkeypatch.setattr(S, "describe_service", lambda: _Info(installed=True, active=True)) + path = HB.heartbeat_path(tmp_path) + for loop in (HB.REAP, HB.NOTIFY): + HB.record_loop_started(path, loop, pid=1) + HB.record_sweep_completed(path, HB.NOTIFY, pid=1) + HB.record_sweep_completed( + path, HB.REAP, pid=1, projects=2, reclaimed=0, failed_projects=["stuck"] + ) + monkeypatch.setattr(HB, "pid_alive", lambda _p: True) + alive = cli._check_sweeps_alive(tmp_path) + reclaiming = cli._check_sweeps_reclaiming(tmp_path) + assert alive.ok is True, alive.detail + assert reclaiming.ok is False From f8b971be78b998063c794b361107c97b1a7382ba Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Thu, 3 Sep 2026 06:26:10 -0700 Subject: [PATCH 2/2] test+docs(oy4): unit pins for the reaper's list-page and per-item isolation; doctor 37->38 (MEASURED); lane note + evidence _FakeBeads' list() now mirrors A.Beads.list's real signature (status/limit) -- a double that silently ignored either would let a caller pass a filter that never took effect and still look correct. Two new unit pins on the call SHAPE (the outcome looks identical on any project small enough to fit inside bd's default page) and on per-item release isolation. AGENTS.md's two doctor-count sites go 37 -> 38, read off the tool on this branch, not computed -- the run is committed under the lane's evidence dir. --- AGENTS.md | 12 +- .../oy4-dead-holder-reclaim/DONE-NOTE.md | 287 ++++++++++++++++++ .../evidence/cli-tier-jyg-failure.txt | 67 ++++ .../evidence/doctor-measured.txt | 41 +++ .../evidence/failbefore_probe.py | 28 +- .../evidence/integration-tier.log | 21 ++ tests/unit/test_supervisor.py | 85 +++++- 7 files changed, 527 insertions(+), 14 deletions(-) create mode 100644 docs/lanes/oy4-dead-holder-reclaim/DONE-NOTE.md create mode 100644 docs/lanes/oy4-dead-holder-reclaim/evidence/cli-tier-jyg-failure.txt create mode 100644 docs/lanes/oy4-dead-holder-reclaim/evidence/doctor-measured.txt create mode 100644 docs/lanes/oy4-dead-holder-reclaim/evidence/integration-tier.log diff --git a/AGENTS.md b/AGENTS.md index e4303c0..5a294ee 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -26,10 +26,13 @@ Nothing above the seam should ever need to change for a Beads upgrade. ## `doctor` is the gate, not a suggestion Run `amplifier-work-tracker doctor` after any `bd` upgrade and before -trusting parallel agents against a queue. It must report **37/37 +trusting parallel agents against a queue. It must report **38/38 assumptions hold**; anything less means Beads' behavior moved out from under an assumption we depend on (or, for `sweeps.alive`, that the reap/notify sweep loops have stopped completing sweeps, or, for +`sweeps.reclaiming`, that the reap loop is still turning but its last +sweep FAILED on one or more projects -- alive and not reclaiming are +different states, and `sweeps.alive` alone cannot tell them apart, or, for `project.removal`, that `remove`/`new` no longer honestly handle a database that outlives its project directory, or, for `service.restart_policy`, that the installed unit's Restart= line has @@ -50,7 +53,10 @@ not 33+2 arithmetic, which would have said 35. The two destructive-reopen defect, `model_performance-2nx` -- then take it to **36**, again MEASURED from `doctor`, not computed. `read.unavailable_not_absent` (model_performance-8zv) makes it **37** -- measured on the rebased branch, not -computed from 36+1.) +computed from 36+1. `sweeps.reclaiming` (model_performance-oy4) makes it +**38** -- again MEASURED by running `doctor` on this branch, not computed; +the run is committed at +`docs/lanes/oy4-dead-holder-reclaim/evidence/doctor-measured.txt`.) ## Test scope @@ -136,7 +142,7 @@ runs itself is how you lose data you meant to keep. ## What "done" looks like -Full suite green, `doctor` 37/37, `ruff check` / `ruff format --check` / +Full suite green, `doctor` 38/38, `ruff check` / `ruff format --check` / `pyright` clean. For any change to the bundle's zero-state install path (service bootstrap, `work_tracker_install`, prereqs), the acceptance gate is a fresh Digital Twin Universe run from a genuinely empty machine (no `bd`, diff --git a/docs/lanes/oy4-dead-holder-reclaim/DONE-NOTE.md b/docs/lanes/oy4-dead-holder-reclaim/DONE-NOTE.md new file mode 100644 index 0000000..c4043c6 --- /dev/null +++ b/docs/lanes/oy4-dead-holder-reclaim/DONE-NOTE.md @@ -0,0 +1,287 @@ +# DONE-NOTE — lane `oy4-dead-holder-reclaim` + +Item: `model_performance-oy4` — *"Custody hold whose holder PROCESS IS DEAD did +not become reclaim-eligible in 23+ min against a 900s TTL — a dead lane strands +its own item."* + +Repo: `microsoft/amplifier-work-tracker`, branch `lane/oy4-dead-holder-reclaim`. + +**OUTCOME: branch A — RESOLVED.** Every deliverable is DONE. Spend **$0.00** +against a **$0.00** authority; nothing needed buying, and nothing was bought. + +--- + +## 1. The headline, and it corrects the item's own premise + +**The item's three candidates are all falsified.** The reap sweep was running; +staleness *was* computed from last renewal; the TTL *was* the documented 900s. +The stranding was real, but not for any of those reasons — and the "23+ minutes +past death" in the item's title is arithmetic built on a wrong death time. + +Forensics, read out of the live `events` table for `model_performance-h6v` +(committed at `evidence/h6v-forensic-timeline.txt`; `events.created_at` is +written in the dolt server's local timezone, +7h to UTC — see `adapter.py`'s +own TIMEZONE GOTCHA block): + +| time (UTC) | what the record shows | +|---|---| +| 07:03:16 | `agent-spark-1-563997` claims, takes custody | +| 07:03:17 → **07:41:36** | **20 custody renewals, on a metronome-regular 120s cadence** | +| — | *then nothing. The next renewal (07:43:36) never happened.* | +| 07:47 / 07:50 / 07:51 / 07:56 | successor `work_claim` refused ×4 | +| 07:52 | `work_stats` reports `held_stale: 0`, `oldest: null` | +| 07:57:51 | hand-run `unclaim` (`actor: Amplifier`) releases it | +| 07:59:09 | successor finally claims | + +Last renewal **07:41:36Z** + TTL **900s** ⇒ first reclaim-eligible at +**07:56:36Z**. So: + +- **All four refusals were CORRECT.** The last one landed **45 seconds inside** + the TTL. Nothing refused a claim it should have granted. +- **`held_stale: 0` at 07:52Z was CORRECT.** Silence was 623s against a 900s + TTL. +- The holder actually died between 07:41:36 and 07:43:36 — **~10 minutes later + than the "~07:33Z" the item records.** The 19- and 23-minute figures are + measured from that wrong start. +- The manual `unclaim` at 07:57:51Z beat the next sweep by ~3 minutes. The + machinery would have freed the item on its own, just later. + +**So what IS the defect?** The one fact that settled the matter — *the holder's +process was gone* — was sitting unread in the custody record. `Custody` has +carried `pid` and `host` since it was designed +(`src/amplifier_work_tracker/custody.py:89-96`) and **nothing has ever read +them for a decision**. `reclaim_eligible` (`custody.py:139-175` pre-fix) decides +on `last_seen` recency and the escalation ceiling, and nothing else. Liveness is +*inferred from silence*, never *observed from the holder*. + +Consequence: a lane whose process dies is stranded for the remainder of its +900s TTL **plus up to a 300s sweep interval — up to 20 minutes** — on a fact +knowable in microseconds on the same host. And during that window a relaunched +successor can do nothing at all: it cannot `work_claim` (held), cannot +`work_release` (it does not hold it), cannot `work_file` (filing requires +holding an item). That dead end is exactly what forced the hand-run `unclaim`, +twice today (`h6v`, `2nx`). + +**Mechanism named at file:line** (post-fix line numbers): + +| site | what it does / did | +|---|---| +| `custody.py:139-175` (pre-fix `reclaim_eligible`) | two paths only: TTL staleness, escalation ceiling. Neither consults the holder. | +| `custody.py:89-96` (`Custody.pid` / `.host`) | recorded on every `take_custody`, read by nothing | +| `supervisor.py:105-134` (`reap_project`) | correct — it reclaims exactly what `reclaim_eligible` says, and no more | +| `adapter.py:5362-5382` (`_held_stale_count`) | correct — calls the same function verbatim, which is why `held_stale` agreed with the reaper | + +--- + +## 2. Deliverables + +### D1 — Mechanism named at file:line, settling the three candidates — **DONE** + +Settled above, from the events table rather than from reading code and +guessing. (a) sweep not running: **NO** — the reap heartbeat shows sweeps +completing continuously, and the machinery released `h6v` correctly once asked. +(b) staleness from the wrong field: **NO** — it is computed from `last_seen`, +exactly as documented, and every value measured during the incident was +arithmetically right. (c) a different TTL: **NO** — 900s, as documented and as +`doctor` reports. + +### D2 — A dead holder's hold becomes `held_stale` and is RECLAIMED, within the documented TTL, with a fail-before — **DONE** + +A **third path** to reclaim-eligible that observes the holder instead of +inferring from its silence, fenced so it can only ever *accelerate* the TTL and +never take work from a live agent: + +1. the record's `host` must equal this host (a pid on another machine is + unknowable — never guessed); +2. `pid` must be a real positive pid; +3. custody must **already** have been silent for `DEAD_HOLDER_MIN_SILENCE_SECONDS` + (default `2 × RENEW_INTERVAL_SECONDS` = 240s) before any pid probe is + consulted at all. + +Every unknowable case resolves to **not** eligible. Fence 3 is what protects an +agent whose pid is simply not addressable from here (a container with its own +pid namespace reporting the same hostname): such an agent keeps renewing, so it +never enters the window where the probe runs. PID reuse is imprecise only in the +safe direction — a recycled pid answers "alive", which merely falls back to the +TTL. + +Reclaim latency for a dead holder: **up to 1200s → up to 540s**, and the reason +string says plainly that the TTL is *not* what fired. + +**Fail-before / pass-after** (`evidence/fail-before-pass-after.txt`, identical +probe on both trees via `git stash push -- src/`), reproducing the incident +signature verbatim — dead holder, 400s silence, 900s TTL: + +``` +BEFORE work_stats: held=1 held_stale=0 held_stale_oldest_age_seconds=None + reap_project(default ttl): reclaimed_count=0 + successor work_claim: REFUSED -- issue already claimed by dead-agent + +AFTER work_stats: held=1 held_stale=1 held_stale_oldest_age_seconds=401.256 + reap_project(default ttl): reclaimed_count=1 + "holder process is dead -- pid 3824385 on host 'spark-1' is not + running, and custody has been silent 401s (corroboration window + 240s); ttl 900s not yet reached, but the holder is gone" + successor work_claim: SUCCESS +``` + +Tests: `tests/unit/test_custody_dead_holder.py` (19, injected probe — the +acceleration plus all three fences plus the live-holder safety property); +`tests/integration/test_dead_holder_reclaim.py` (7, real `bd`, real dolt, and a +**real** dead pid — a subprocess started and reaped, not a number chosen for +looking implausible). + +### D3 — A successor session can `work_claim` after the reclaim, end to end — **DONE** + +`modules/tool-work-tracker/tests/test_dead_holder_successor_claim.py` (4), at +the **agent seam** — `WorkTrackerSession.claim/resolve`, the verbs the stranded +lane actually had. Refused by name before the sweep; succeeds after; the +successor can then resolve. Deliberately **not** `ttl_seconds=0`, which is how +every other reap test in that suite forces staleness — with ttl 0 every hold is +stale and a dead-holder bug hides completely. These use the real default TTL and +a real 400s silence, so a reclaim can only have come from the liveness path. A +live session's identically-silent hold survives the same sweep. + +### D4 — `doctor` distinguishes "sweeps installed" from "sweeps actually reclaiming" — **DONE** + +The gap was real even though it is not what bit `h6v`. `reap_sweep` catches +every per-project exception into its return value (correct — one broken project +must not abort the sweep), and `reap_loop` **discarded that return value** before +stamping a completed heartbeat. A sweep that failed on *every* project recorded +the same heartbeat as a perfectly healthy one, so `sweeps.alive` — and +`work_tracker_status`'s `running_healthy` on top of it — read identically either +way. + +Now: the reap heartbeat carries the sweep's **outcome** (`projects`, +`reclaimed`, `failed_projects`), `reap_loop` `logger.error`s any failed +projects by name, and `doctor` gains **`sweeps.reclaiming`** — +`heartbeat.evaluate_reclaiming`, pure and fully unit-tested — which FAILS naming +every project the last sweep failed on. + +One honesty note, and it is deliberate: a heartbeat written by a supervisor +older than this change carries no `failed_projects` key, and that case reports +**`unknown`**, never `0 failed`. Claiming "nothing failed" from a record that +never carried the field would invent exactly the reassurance this check exists +to stop being invented. That is what the live box reports right now (see +`evidence/doctor-measured.txt`) and it clears on the next service restart. + +Tests: `tests/unit/test_sweeps_reclaiming.py` (17). + +### D5 — Verdict on `model_performance-c0e` — **DONE: genuinely distinct, and its primary half is ALREADY FIXED** + +Argued from the code, and then measured. + +| | `oy4` | `c0e` | +|---|---|---| +| site | `custody.reclaim_eligible` | `Beads.resolve`'s custody fence | +| proxy relied on | last-renewal **timestamp** | item **status** (`if current.status == "held"`) | +| direction | a live successor wrongly **prevented** from acting | a stale holder wrongly **permitted** to act | +| when | reclaim **never happens** | **after** a reclaim happens | + +They rhyme — both are "custody state read from a proxy rather than from the +custody record" — but they are different proxies at different call sites failing +in opposite directions, and neither fix touches the other's code. `oy4`'s fix +does not go near the resolve fence; `c0e`'s does not go near reclaim +eligibility. **Two defects. Fixed only `oy4`, as the goal directs.** + +And `c0e`'s named failing test **passes on this tree**: the full modules suite +is **119 passed, 0 failed**, including +`test_reap_recovery.py::test_explicit_resolve_refusal_after_reap_clears_held_and_allows_new_claim`, +the exact test `c0e` reports as failing. `c0e` was filed 03:53Z; PR #68 (the +post-reclaim fence, re-keyed on custody **identity** rather than item status) +landed afterwards and closed it. `c0e`'s **second** finding — that ledger row +CCV1-023's note claims the fixture "passes via the session latch" — is a ledger +note, a different subject, and I did **not** verify or touch it. **I have not +resolved `c0e`**: I do not hold it, its second half is unverified by me, and +resolving another lane's item on a partial reading is exactly the kind of quiet +overreach this program keeps paying for. Recommendation for whoever holds it: +re-run the named test, confirm it passes, and close on the ledger-note half +alone. + +### D6 — Two further silent misses in the reaper, found while root-causing — **DONE** + +Neither caused this incident; both are live silent-miss bugs in the same +function, and both are one-line-shaped fixes with tests. + +1. **`reap_project` read bd's DEFAULT list page.** `bd.list(include_resolved=False)` + with no `limit` applies `LIST_DEFAULT_LIMIT` (50), ordered `priority ASC, + created_at DESC, id ASC`. **A held item outside that first page was invisible + to the reaper permanently, with nothing anywhere reporting the skip.** Now + `bd.list(status="held", limit=0)`. Measured and ruled out as this incident's + cause: `model_performance` held 20–22 non-closed items across the window and + `h6v` ranked 2nd–4th. Pinned twice — on the *call shape* + (`test_supervisor.py`, since the outcome looks identical on any project small + enough to fit) and end-to-end with a shrunken page size. +2. **One unreleasable item shadowed the whole queue.** The loop had no per-item + guard, so the first `release` that raised propagated out, `reap_sweep` caught + it per project, and every remaining stale hold went unreaped — deterministically, + on every sweep after it, while the sweep still recorded itself completed. + Now isolated per item and reported in `failed`, which flows into D4's + heartbeat. + +--- + +## 3. Tiers — reported BY NAME + +| tier | command | result | +|---|---|---| +| unit | `pytest tests/unit` | **881 passed, 0 failed** | +| integration | `pytest -m integration tests/integration` | **374 passed, 3 skipped, 0 failed** (`evidence/integration-tier.log`) | +| cli | `pytest -m cli tests/cli` | **88 passed, 1 failed** — `test_doctor_quick_succeeds_against_the_real_installed_bd`, which is `model_performance-jyg`, **not mine** (see below) | +| ledger | `pytest ledger/checks` | **26 passed** | +| ledger mutation | `python -m ledger.checks.mutation_harness` | **proven 15 / 15**, none unproven | +| modules (tier 5) | `pytest modules/tool-work-tracker/tests` | **119 passed, 0 failed** | +| lint | `ruff check .` / `ruff format --check .` | clean, 163 files | +| types | `pyright src tests` | **0 errors, 0 warnings** | +| **doctor** | `python -m amplifier_work_tracker.cli doctor` | **All 38 assumptions hold** — **MEASURED**, not computed (`evidence/doctor-measured.txt`) | + +**The one cli failure is `jyg`, and I verified it rather than assuming it.** The +full doctor output inside that failure shows a single `[FAIL]`, on +`sweeps.alive`: *"no heartbeat ever recorded for the reap sweep loop"* — the +isolated test root has no sweep heartbeat, exactly as `jyg` describes. My own new +check appears one line below it as `[PASS] sweeps.reclaiming skipped -- no reap +heartbeat recorded yet`, correctly declining to pile a second red line on a root +cause already reported. Captured in `evidence/cli-tier-jyg-failure.txt`. + +`doctor` **37 → 38**: the count was read off the tool on this branch, never +computed. `AGENTS.md`'s two sites are updated. + +## 4. Spend + +**$0.00** of a **$0.00** authority. The goal's arithmetic (`0 runs × 0 arms × +$0 / 1.00 = $0.00`) closes: this is a pure local code change, no API calls, no +DTU, no infrastructure created — so nothing was registered in the infra ledger +and nothing needed tearing down. No residue was left: the reads against the live +`model_performance` database were pure `SELECT`s, and every test project came +from the repo's own isolated-server fixtures. + +## 5. Deviations and judgement calls, all deliberate + +1. **I corrected the item's premise rather than confirming it.** The item asks + which of three candidates is true; the honest answer is *none*, and the + supporting arithmetic is built on a death time ~10 minutes early. Reporting + any of the three would have been a fabricated root cause. +2. **A corroboration window instead of an instant probe.** A bare "pid gone ⇒ + reclaim" would fire on a container whose pid is unaddressable from here. Two + independent signals — a missed renewal *and* a dead pid — cost ~4 extra + minutes of latency and remove the only way this fix could steal live work. +3. **`sweeps.reclaiming` reports `unknown` for an older supervisor's record.** + Weaker than failing, and named as such above. Failing would red-line every + box until its service restarts; claiming `0 failed` would be a lie. Reporting + the gap in the assumption's own text is the honest third option. +4. **I did not resolve `c0e`,** for the reasons in D5. +5. **I did not touch `jyg`** (the goal scopes it out) and did not run + `reap --project` against `model_performance` at any point — live lanes hold + items in it right now. Every live-queue interaction in this lane was a + read-only `SELECT`. + +## 6. Evidence in this directory + +| file | what it is | +|---|---| +| `evidence/h6v-forensic-timeline.txt` | the `events`-table reconstruction that falsified all three candidates | +| `evidence/fail-before-pass-after.txt` | identical probe, pre-fix vs post-fix, reproducing the incident signature | +| `evidence/failbefore_probe.py` | that probe, runnable | +| `evidence/doctor-measured.txt` | `doctor` 38/38, measured on this branch | +| `evidence/cli-tier-jyg-failure.txt` | proof the one cli failure is `jyg`, with my check passing beside it | +| `evidence/integration-tier.log` | integration tier output | diff --git a/docs/lanes/oy4-dead-holder-reclaim/evidence/cli-tier-jyg-failure.txt b/docs/lanes/oy4-dead-holder-reclaim/evidence/cli-tier-jyg-failure.txt new file mode 100644 index 0000000..7a3947b --- /dev/null +++ b/docs/lanes/oy4-dead-holder-reclaim/evidence/cli-tier-jyg-failure.txt @@ -0,0 +1,67 @@ +F [100%] +=================================== FAILURES =================================== +___________ test_doctor_quick_succeeds_against_the_real_installed_bd ___________ + +run_cli = ._run at 0xeb8d1098ec00> + + def test_doctor_quick_succeeds_against_the_real_installed_bd(run_cli): + result = run_cli(["doctor", "--quick"]) +> assert result.returncode == 0, result.stdout + result.stderr +E AssertionError: [PASS] version bd 1.1.2 +E [PASS] capabilities all required bd commands present +E [PASS] read.unavailable_not_absent an infrastructure read failure raises BeadsUnavailableError with its cause intact on read/claim and reports UNAVAILABLE (not ERROR) per project, while genuine absence on a healthy database still reports plain 'not found' +E [PASS] resolve.fenced stale holder refused, as required +E [PASS] resolve.divergent_text_refused resolving a closed item with different text refuses and writes nothing +E [PASS] resolve.identical_text_idempotent re-sending identical resolution text is an idempotent success +E [PASS] reopen.reopens a resolved item reopens unassigned and is directly claimable again +E [PASS] reopen.clears_closed_at reopen clears closed_at (the documented, surfaced accounting cost) +E [PASS] reopen.close_reason_disposition reopen clears close_reason (measured), and the wrapper's archive comment preserves the previous resolution regardless +E [PASS] reopen.emits_event bd records a `reopened` events row, attributed +E [PASS] defer.refuses_resolved defer on a resolved item refuses, writes nothing, and names `reopen` +E [PASS] block.refuses_resolved block on a resolved item refuses, writes nothing, and names `reopen` +E [PASS] release.reopens_unresolved release reopens a held item with no resolution, and it is re-claimable +E [PASS] claim.subcommand --claim present, rejects --assignee as expected +E [PASS] claim.atomic skipped (--quick); run full doctor before trusting parallel agents +E [PASS] claim.directed_atomic skipped (--quick); run full doctor before trusting parallel agents +E [PASS] link.nonblocking discovered-from is non-blocking +E [PASS] list.includes_closed all-flag required and working +E [PASS] list.status_filter_includes_closed an explicit --status filter shows closed items without --all +E [PASS] show.dependents reverse link visible (1 links) +E [PASS] read.no_mutation repeated reads (including not-found/wrong-project misses) leave status, holder, and metadata unchanged +E [PASS] resolution.readable resolution text round-trips +E [PASS] timestamps.readable created_at/updated_at/closed_at all round-trip as real datetimes +E [PASS] metadata.roundtrip arbitrary JSON metadata round-trips +E [PASS] project.name_rules dotted names appear usable now; validator may be relaxed +E [PASS] custody.fresh_survives a fresh renewal survives regardless of total hold duration +E [PASS] custody.stale_reclaimed stale custody is reclaimed: custody stale -- last seen 3600s ago (ttl 900s) +E [PASS] custody.idle_not_exempt awaiting_human with stale custody is still reclaimed: custody stale -- last seen 3600s ago (ttl 900s) +E [PASS] custody.fenced old holder's renew and resolve are both refused after takeover +E [PASS] project.removal remove() drops both the directory and database; re-create afterward is genuinely empty +E [PASS] project.create_atomic an abandoned creation lock (dead pid) is healed automatically and create() completes fresh in the same call; path=/tmp/awtcontract_zf0anue7/projects/contract178844191296atomic +E [PASS] project.creation_state_reporting creation_state distinguishes none/creating/abandoned correctly +E [PASS] service.installed installed and active (unit: /home/bkrabach/.config/systemd/user/amplifier-work-tracker.service) +E [PASS] systemd.user_bus_reachable systemctl --user show-environment succeeded +E [PASS] dolt.reachable dolt sql-server responds on 127.0.0.1:35411 +E [FAIL] sweeps.alive no heartbeat ever recorded for the reap sweep loop -- it may never have started, or the heartbeat file was removed; restart the service (`amplifier-work-tracker service restart`); no heartbeat ever recorded for the notify sweep loop -- it may never have started, or the heartbeat file was removed; restart the service (`amplifier-work-tracker service restart`) +E [PASS] sweeps.reclaiming skipped -- no reap heartbeat recorded yet (see the reap sweep liveness check, which reports that directly) +E [PASS] service.restart_policy installed unit has Restart=always -- survives a clean/unintended exit +E +E 1 assumption(s) VIOLATED. Beads has changed underneath us. +E Fix scope: amplifier_work_tracker/adapter.py only -- nothing above the seam encodes Beads behaviour. +E project 'contract178844191296atomic': healing an abandoned creation attempt (lock /tmp/awtcontract_zf0anue7/projects/contract178844191296atomic/.create.lock named a dead pid) before retrying +E +E assert 1 == 0 +E + where 1 = CompletedProcess(args=['/home/bkrabach/dev/hw-model-performance/lanes/oy4-dead-holder-reclaim/amplifier-work-tracker/.... (lock /tmp/awtcontract_zf0anue7/projects/contract178844191296atomic/.create.lock named a dead pid) before retrying\n").returncode + +tests/cli/test_cli_surface.py:831: AssertionError +---------------------------- Captured stdout setup ----------------------------- +Starting server with Config HP="127.0.0.1:35411"|T="28800000"|R="false"|L="info" +---------------------------- Captured stderr setup ----------------------------- +----------------------------- Captured stdout call ----------------------------- + + +----------------------------- Captured stderr call ----------------------------- +--------------------------- Captured stderr teardown --------------------------- +=========================== short test summary info ============================ +FAILED tests/cli/test_cli_surface.py::test_doctor_quick_succeeds_against_the_real_installed_bd +1 failed in 46.41s diff --git a/docs/lanes/oy4-dead-holder-reclaim/evidence/doctor-measured.txt b/docs/lanes/oy4-dead-holder-reclaim/evidence/doctor-measured.txt new file mode 100644 index 0000000..49cd777 --- /dev/null +++ b/docs/lanes/oy4-dead-holder-reclaim/evidence/doctor-measured.txt @@ -0,0 +1,41 @@ +project 'contract1788440121343atomic': healing an abandoned creation attempt (lock /tmp/awtcontract_0m323x7t/projects/contract1788440121343atomic/.create.lock named a dead pid) before retrying + [PASS] version bd 1.1.2 + [PASS] capabilities all required bd commands present + [PASS] read.unavailable_not_absent an infrastructure read failure raises BeadsUnavailableError with its cause intact on read/claim and reports UNAVAILABLE (not ERROR) per project, while genuine absence on a healthy database still reports plain 'not found' + [PASS] resolve.fenced stale holder refused, as required + [PASS] resolve.divergent_text_refused resolving a closed item with different text refuses and writes nothing + [PASS] resolve.identical_text_idempotent re-sending identical resolution text is an idempotent success + [PASS] reopen.reopens a resolved item reopens unassigned and is directly claimable again + [PASS] reopen.clears_closed_at reopen clears closed_at (the documented, surfaced accounting cost) + [PASS] reopen.close_reason_disposition reopen clears close_reason (measured), and the wrapper's archive comment preserves the previous resolution regardless + [PASS] reopen.emits_event bd records a `reopened` events row, attributed + [PASS] defer.refuses_resolved defer on a resolved item refuses, writes nothing, and names `reopen` + [PASS] block.refuses_resolved block on a resolved item refuses, writes nothing, and names `reopen` + [PASS] release.reopens_unresolved release reopens a held item with no resolution, and it is re-claimable + [PASS] claim.subcommand --claim present, rejects --assignee as expected + [PASS] claim.atomic 5 trials x 12 concurrent claimers, no double-claims + [PASS] claim.directed_atomic 5 trials x 12 concurrent directed claimers on the SAME item, exactly one winner each time + [PASS] link.nonblocking discovered-from is non-blocking + [PASS] list.includes_closed all-flag required and working + [PASS] list.status_filter_includes_closed an explicit --status filter shows closed items without --all + [PASS] show.dependents reverse link visible (1 links) + [PASS] read.no_mutation repeated reads (including not-found/wrong-project misses) leave status, holder, and metadata unchanged + [PASS] resolution.readable resolution text round-trips + [PASS] timestamps.readable created_at/updated_at/closed_at all round-trip as real datetimes + [PASS] metadata.roundtrip arbitrary JSON metadata round-trips + [PASS] project.name_rules dotted names appear usable now; validator may be relaxed + [PASS] custody.fresh_survives a fresh renewal survives regardless of total hold duration + [PASS] custody.stale_reclaimed stale custody is reclaimed: custody stale -- last seen 3600s ago (ttl 900s) + [PASS] custody.idle_not_exempt awaiting_human with stale custody is still reclaimed: custody stale -- last seen 3600s ago (ttl 900s) + [PASS] custody.fenced old holder's renew and resolve are both refused after takeover + [PASS] project.removal remove() drops both the directory and database; re-create afterward is genuinely empty + [PASS] project.create_atomic an abandoned creation lock (dead pid) is healed automatically and create() completes fresh in the same call; path=/tmp/awtcontract_0m323x7t/projects/contract1788440121343atomic + [PASS] project.creation_state_reporting creation_state distinguishes none/creating/abandoned correctly + [PASS] service.installed installed and active (unit: /home/bkrabach/.config/systemd/user/amplifier-work-tracker.service) + [PASS] systemd.user_bus_reachable systemctl --user show-environment succeeded + [PASS] dolt.reachable dolt sql-server responds on 127.0.0.1:3308 + [PASS] sweeps.alive reap sweep completed 159s ago (threshold 900s); notify sweep completed 45s ago (threshold 900s) + [PASS] sweeps.reclaiming unknown -- the running supervisor predates per-sweep outcome reporting, so its reap heartbeat cannot say whether any project failed; restart the service (`amplifier-work-tracker service restart`) to start recording it + [PASS] service.restart_policy installed unit has Restart=always -- survives a clean/unintended exit + +All 38 assumptions hold. Safe to run parallel agents. diff --git a/docs/lanes/oy4-dead-holder-reclaim/evidence/failbefore_probe.py b/docs/lanes/oy4-dead-holder-reclaim/evidence/failbefore_probe.py index 5d68582..2a6fe1c 100644 --- a/docs/lanes/oy4-dead-holder-reclaim/evidence/failbefore_probe.py +++ b/docs/lanes/oy4-dead-holder-reclaim/evidence/failbefore_probe.py @@ -5,9 +5,18 @@ a hold whose holder process is genuinely dead, last renewed 400s ago, against the documented 900s CUSTODY_TTL_SECONDS. """ + from __future__ import annotations -import json, os, socket, subprocess, sys, time + +import json +import os +import socket +import subprocess +import sys +import time + import pytest + from amplifier_work_tracker import adapter as A from amplifier_work_tracker import custody as C from amplifier_work_tracker import supervisor as SV @@ -17,7 +26,8 @@ def _dead_pid() -> int: - p = subprocess.Popen([sys.executable, "-c", "pass"]); p.wait() + p = subprocess.Popen([sys.executable, "-c", "pass"]) # noqa: S603 + p.wait() return p.pid @@ -35,11 +45,15 @@ def test_probe(workspace, project_factory): out.append(f"holder pid {pid} running? {os.path.exists(f'/proc/{pid}')}") out.append(f"custody last_seen {SILENCE}s ago; CUSTODY_TTL_SECONDS={C.CUSTODY_TTL_SECONDS}") s = A.project_summary(workspace, name) - out.append(f"work_stats view: held={s.held} held_stale={s.held_stale} " - f"held_stale_oldest_age_seconds={s.held_stale_oldest_age_seconds}") - r = SV.reap_project(bd) # default TTL, exactly what the service runs - out.append(f"reap_project(default ttl): reclaimed_count={r['reclaimed_count']} " - f"reasons={[x['reason'] for x in r['reclaimed']]}") + out.append( + f"work_stats view: held={s.held} held_stale={s.held_stale} " + f"held_stale_oldest_age_seconds={s.held_stale_oldest_age_seconds}" + ) + r = SV.reap_project(bd) # default TTL, exactly what the service runs + out.append( + f"reap_project(default ttl): reclaimed_count={r['reclaimed_count']} " + f"reasons={[x['reason'] for x in r['reclaimed']]}" + ) after = bd.get(item_id) out.append(f"after sweep: status={after.status} holder={after.holder!r}") try: diff --git a/docs/lanes/oy4-dead-holder-reclaim/evidence/integration-tier.log b/docs/lanes/oy4-dead-holder-reclaim/evidence/integration-tier.log new file mode 100644 index 0000000..3ba8165 --- /dev/null +++ b/docs/lanes/oy4-dead-holder-reclaim/evidence/integration-tier.log @@ -0,0 +1,21 @@ +........................................................................ [ 19%] +........................................................................ [ 38%] +........................................................................ [ 57%] +..................................................s..................... [ 76%] +..............................ss........................................ [ 95%] +................. [100%] +=============================== warnings summary =============================== +tests/integration/test_observatory_web.py:20 + /home/bkrabach/dev/hw-model-performance/lanes/oy4-dead-holder-reclaim/amplifier-work-tracker/tests/integration/test_observatory_web.py:20: StarletteDeprecationWarning: Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead. + from starlette.testclient import TestClient # noqa: E402 + +.venv/lib/python3.12/site-packages/starlette/testclient.py:53 + /home/bkrabach/dev/hw-model-performance/lanes/oy4-dead-holder-reclaim/amplifier-work-tracker/.venv/lib/python3.12/site-packages/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead. + _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]] + +-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html +=========================== short test summary info ============================ +SKIPPED [1] tests/integration/test_web.py:699: bd set no `owner` on this item -- happens where the environment has no git identity (e.g. a bare CI container with no `git config user.email`). `owner` is an environment-provided value, not a code invariant; the environment-independent humanization guarantees are covered by test_humanize_identity_* in tests/unit. +SKIPPED [1] tests/integration/test_web_pwa.py:234: Pillow not installed (dev/build-only tool) +SKIPPED [1] tests/integration/test_web_pwa.py:249: Pillow not installed (dev/build-only tool) +374 passed, 3 skipped, 2 warnings in 1125.40s (0:18:45) diff --git a/tests/unit/test_supervisor.py b/tests/unit/test_supervisor.py index 1ca9207..8685539 100644 --- a/tests/unit/test_supervisor.py +++ b/tests/unit/test_supervisor.py @@ -326,11 +326,33 @@ def __init__(self, items: dict[str, _FakeItem]): self.items = items self.released: list[str] = [] self.resolved: list[tuple[str, str, str]] = [] - - def list(self, *, lane: str | None = None, include_resolved: bool = False): + #: Every `list()` call's kwargs, so a test can assert HOW the reaper + #: asked -- `model_performance-oy4`: asking with bd's default page + #: size silently hid held items past the 50th from the reaper. + self.list_calls: list[dict] = [] + + def list( + self, + *, + lane: str | None = None, + include_resolved: bool = False, + status: str | None = None, + limit: int | None = None, + ): + """Mirrors `A.Beads.list`'s real signature, including `status` and + `limit` -- a double that silently ignored either would let a caller + pass a filter that never took effect and still look correct here. + """ + self.list_calls.append( + {"lane": lane, "include_resolved": include_resolved, "status": status, "limit": limit} + ) out = list(self.items.values()) - if not include_resolved: + if status is not None: + out = [i for i in out if i.status == status] + elif not include_resolved: out = [i for i in out if i.status != "resolved"] + if limit: # 0/None both mean unlimited, matching bd's own convention + out = out[:limit] return out def get(self, item_id: str, *, with_links: bool = False): @@ -406,13 +428,68 @@ def test_reap_sweep_isolates_a_broken_project_from_the_others(): ) class _ExplodingBeads(_FakeBeads): - def list(self, *, lane: str | None = None, include_resolved: bool = False): + def list(self, **_kwargs): raise RuntimeError("simulated bd outage") ws = _FakeWorkspace({"broken": _ExplodingBeads({}), "good": good_bd}) results = SV.reap_sweep(ws, ttl_seconds=900) # type: ignore[arg-type] assert "error" in results["broken"] assert results["good"]["reclaimed_count"] == 1 + # The isolation must not also be a silence: the broken project has to be + # nameable by the sweep-level reporting the `sweeps.reclaiming` doctor + # check reads (`model_performance-oy4`). + assert SV.sweep_failures(results) == ["broken"] + + +def test_reap_project_asks_for_every_held_item_not_bds_default_page(): + """`model_performance-oy4`. `Beads.list()` with no `limit` applies bd's + own default cap (`LIST_DEFAULT_LIMIT`, 50) ordered `priority ASC, + created_at DESC, id ASC`. The reaper used to read that default page, so + in a project with more than 50 non-closed items a held item outside it + was invisible to the reaper permanently -- and nothing anywhere reported + that it had been skipped. Asserted on the CALL, not just the outcome, + because the outcome looks identical on any project small enough to fit. + """ + bd = _FakeBeads({}) + SV.reap_project(bd, ttl_seconds=900) # type: ignore[arg-type] + assert bd.list_calls == [ + {"lane": None, "include_resolved": False, "status": "held", "limit": 0} + ] + + +def test_reap_project_reports_an_item_it_could_not_release_instead_of_aborting(): + """One wedged item must not shadow the rest of the project's stale holds. + Before this, the first `release` that raised propagated out of + `reap_project`, `reap_sweep` caught it per project, and every remaining + hold went unreaped -- on that sweep and, since the failure is + deterministic, on every sweep after it. + """ + + def _stale(holder: str) -> dict: + return {C.CUSTODY_KEY: {"holder": holder, "last_seen": _ts(3600)}} + + bd = _FakeBeads( + { + "w-1": _FakeItem("w-1", status="held", holder="agent-a", meta=_stale("agent-a")), + "w-2": _FakeItem("w-2", status="held", holder="agent-b", meta=_stale("agent-b")), + } + ) + real_release = bd.release + + def _release(item_id: str) -> None: + if item_id == "w-1": + raise RuntimeError("simulated wedged release") + real_release(item_id) + + bd.release = _release # type: ignore[method-assign] + + result = SV.reap_project(bd, ttl_seconds=900) # type: ignore[arg-type] + + assert result["reclaimed_count"] == 1 + assert result["reclaimed"][0]["id"] == "w-2" + assert result["failed_count"] == 1 + assert result["failed"][0]["id"] == "w-1" + assert "simulated wedged release" in result["failed"][0]["error"] def test_notify_project_flips_only_linked_unresolved_reports():