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/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..2a6fe1c --- /dev/null +++ b/docs/lanes/oy4-dead-holder-reclaim/evidence/failbefore_probe.py @@ -0,0 +1,64 @@ +"""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 +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 + +pytestmark = pytest.mark.integration +SILENCE = 400 + + +def _dead_pid() -> int: + p = subprocess.Popen([sys.executable, "-c", "pass"]) # noqa: S603 + 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/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/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_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(): 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