diff --git a/scripts/coord/seat_clock_alarm.py b/scripts/coord/seat_clock_alarm.py new file mode 100644 index 00000000..95754b3e --- /dev/null +++ b/scripts/coord/seat_clock_alarm.py @@ -0,0 +1,277 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Organization and contributors +"""A watchdog cannot watch its own death, so something outside the chain has to (BACKLOG #1269). + +The seat clock (``MEFOR-Seat-Clock``, ``PT10M``) is what makes a session's "keep going" duty +physically possible. The chain is SERIAL -- each tick re-arms the next watcher -- so it has ONE LIFE. +One missed re-arm and every seat whose only wake source is the clock goes quiet, silently. + +WHICH FILE ANSWERS WHICH QUESTION, AND WHY THE OBVIOUS ONE IS WRONG +-------------------------------------------------------------------- +The obvious build reads ``seat-tick.last`` and asks whether the watched seat appears in it. That file +cannot answer the question, and its own author says so at the point where the second file is declared: + + "seat-tick.last is a human-readable one-liner describing the LAST RUN; it cannot answer 'when did + THIS seat last actually get a tick', because a run that reported COLD for a seat OVERWRITES the + run in which that seat was SENT." + +An alarm reading only that file GOES HEALTHY ON THE OVERWRITE -- precisely the defect this item was +filed to prevent. **The throttle already hit this exact wall and was given its own file for it, so +this reuses the mechanism the same author built for the same problem rather than inventing one.** + + seat-tick.state.json WHEN and WHO -- keyed by ABSOLUTE WORKTREE PATH, values unix seconds + seat-tick.last WHY NOT -- consulted only to decide whether a gap is deliberate + +THE SEAT NAME IS NOT A KEY. Measured 2026-08-23: the live one-liner carried ``steward``, ``lander`` +and ``dispatcher`` TWICE EACH -- once ``STALE(no-live-session)``, once ``SENT:``. A first-match +scan for a seat name reads STALE for all three while the clock ticks normally. That is the same +first-match trap that cost four seats a wrong answer the same morning, sitting inside the file the +alarm was told to parse. The worktree PATH is unique; the seat name is not. + +PIN THE PATH. DO NOT GLOB THE FILENAME +---------------------------------------- +At least four files are named ``seat-tick.last``. The two decoys the item recorded are already gone +and three different ones exist today, all under a live lane's scratchpad, one in a directory named +``ticktest``. **Stale evidence for a correct rule is the strongest kind**: "search, then take the +newest" works until any scratchpad copy is written after the real one, and today it would land in +another lane's test fixture. The decoys also fail in OPPOSITE directions -- one is a permanent +``FATAL`` record that makes the alarm scream on a healthy clock; the other reads ``THROTTLED`` for +every seat and routes into the exclusion rule below, so the alarm goes SILENT on a ten-hour-old file. + +BOTH CONSTRUCTION FAULTS REPORT A HEALTHY CLOCK AS BROKEN, AND THAT DIRECTION MATTERS +-------------------------------------------------------------------------------------- +An alarm that fires on healthy cases gets discounted, and is therefore absent on the day it matters. +A false-positive watchdog is not a safe failure mode; it is a slow-acting off switch. + +* **Dedupe by tick IDENTITY, not timestamp proximity.** A raw scan once produced ten 0.0-minute + intervals, because 22 records were about 12 ticks differing in milliseconds. Here the identity is + the stamp VALUE: an unchanged value is the SAME tick observed twice, never a zero-length interval. +* **Do not read a suppressed seat as a dead one.** An 88-minute gap that read as "chain broken" was + deliberate suppression on a seat taking continuous turns. + +VOCABULARY IS AS MEASURED TODAY, NOT AS THE ITEM LISTED IT. The item names +COLD/BACKLOG/THROTTLED. The emitter also produces ``STALE(no-live-session)`` and can suffix a +send with ``(roster-blind)``, both of which postdate the item. + +THE EMITTER IS NOT IN THIS REPOSITORY. ``seat-tick.ps1`` is a machine-global install: +``git ls-tree -r origin/main`` returns zero for it, against a positive control of one for +``scripts/coord/seat.ps1``. So no line number could be resolved by a reader of this repo, and +any quoted here would drift silently. Grep that file for its ``$results.Add`` sites instead -- +each token is minted in exactly one place. Treat this list as a snapshot of a file this +repository cannot pin, and re-derive it rather than trusting it. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +import time +from pathlib import Path +from typing import NamedTuple + +#: THE ONE AUTHORITATIVE DIRECTORY. Never search for these basenames -- see the module docstring. +_MEFOR = Path(os.environ.get("USERPROFILE", str(Path.home()))) / ".claude" / "mefor-usage" + +#: A gap longer than this, with no deliberate suppression, means the chain is gone. The clock is +#: ``PT10M``, so this tolerates one missed tick rather than firing on ordinary jitter. +DEAD_AFTER_SECONDS = 25 * 60 + +#: Ticks closer together than this are OVER-firing, which is the EXPENSIVE fault: every tick wakes a +#: seat and spends a turn, so a runaway clock burns the budget the alarm exists to protect. +OVERFIRE_UNDER_SECONDS = 6 * 60 + +#: Statuses that make a gap DELIBERATE rather than evidence of death. +_SUPPRESSED = ("THROTTLED", "COLD", "BACKLOG", "STALE") + +#: ``steward=THROTTLED(last-send--35997s-ago,floor-360s)`` +_THROTTLE = re.compile(r"THROTTLED\(last-send-(-?\d+)s-ago,floor-(\d+)s\)") + +#: A THROTTLED record whose own age exceeds its own floor by more than this is self-contradictory: +#: throttling means "suppressed because a send was too RECENT". The known decoy claims ``floor-360s`` +#: at an age of ~36000s -- a hundred-fold contradiction of the state it declares. +_THROTTLE_SANITY_FACTOR = 10 + + +class Verdict(NamedTuple): + """``alarm`` is the only field a caller must act on; the rest explain it.""" + + alarm: bool + code: str + detail: str + + +class Unreadable(Exception): + """The state file EXISTS but cannot be understood. NOT an alarm condition. + + A watchdog that reports DEAD because its own instrument broke is a false positive that fires for + EVERY watched seat at once, and the first false alarm is the loudest -- the worst possible + introduction for a tool whose only value is being believed. The item calls that a slow-acting off + switch: it does not fail on the day it fires, it fails weeks later, having been discounted. + + This is the SAME QUESTION the missing-file check already answers, in its second form. The shape + was never missing; only its second application was. + """ + + +def read_state(path: Path) -> dict[str, int]: + """Worktree path (normalised) -> unix seconds of that worktree's last real tick. + + Keys are lowercased because a Windows path-casing collision has already killed this clock once: + ``seats.json`` carried the same directory under two casings and the tick script died on the parse. + """ + try: + doc = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + # NOT HYPOTHETICAL. The item records that a Windows path-casing collision killed this clock + # once already BY DYING ON A JSON PARSE. Uncaught, that arrived here as a traceback and a + # nonzero exit -- which reads as ALARM. + raise Unreadable(f"{path} could not be parsed: {exc}") from exc + if not isinstance(doc, dict): + raise Unreadable( + f"{path} holds a {type(doc).__name__} at the top level, expected an object" + ) + + out: dict[str, int] = {} + for key, value in doc.items(): + if not isinstance(key, str) or not isinstance(value, int): + continue + out[key.replace("/", "\\").lower()] = value + + # EVERY RECORD SKIPPED IS A SCHEMA SIGNAL, NOT AN EMPTY REGISTRY. Dropping them silently left the + # alarm reporting ABSENT for every seat, confidently, on data it had not understood. + if doc and not out: + raise Unreadable( + f"{path} holds {len(doc)} record(s) and NONE match the expected path -> unix-seconds " + "shape. That is a schema this reader does not know, not an empty registry." + ) + return out + + +def suppression_for(last_line: str, seat: str) -> str | None: + """The suppression token for ``seat``, or ``None`` if it was sent or is absent. + + A SEND ANYWHERE IN THE LINE WINS. The seat name repeats -- measured, three seats twice each in one + line -- so returning the first match is exactly how a normally-ticking seat reads as STALE. + """ + found: list[str] = [] + for token in last_line.split(): + name, sep, status = token.partition("=") + if sep and name == seat: + found.append(status) + if not found: + return None + if any(s.startswith("SENT:") for s in found): + return None + for status in found: + if status.startswith(_SUPPRESSED): + return status + return None + + +def throttle_is_credible(status: str) -> bool: + """False when a THROTTLED record contradicts itself, so its exclusion must not be honoured. + + One comparison. It rejects the ten-hour-old decoy, and it catches genuine corruption in the + authoritative file -- which is the better reason to have it. + """ + m = _THROTTLE.search(status) + if m is None: + return True + age, floor = int(m.group(1)), int(m.group(2)) + if age < 0: + return False # a send in the FUTURE; self-contradictory either way it is read + return age <= floor * _THROTTLE_SANITY_FACTOR + + +def evaluate( + watched: str, + seat: str, + state: dict[str, int], + last_line: str, + now: float, + previous_tick: int | None = None, +) -> Verdict: + """Decide whether to alarm for one watched worktree.""" + key = watched.replace("/", "\\").lower() + tick = state.get(key) + + suppression = suppression_for(last_line, seat) + if suppression is not None and not throttle_is_credible(suppression): + suppression = None # the record contradicts itself; it must not silence the alarm + + if tick is None: + # THE DISCRIMINATING CASE. A freshness-only alarm is GREEN here, because other seats keep the + # file fresh -- and this is exactly when the watched seat has stopped being woken. + if suppression is not None: + return Verdict( + False, + "SUPPRESSED-ABSENT", + f"{seat} absent from state but {suppression}", + ) + return Verdict(True, "ABSENT", f"{seat} ({watched}) has no entry in the state file at all") + + age = int(now - tick) + + # DEDUPE BY IDENTITY: an unchanged stamp is the SAME tick observed twice, not a zero-length gap. + if previous_tick is not None and tick != previous_tick: + interval = tick - previous_tick + if 0 < interval < OVERFIRE_UNDER_SECONDS: + return Verdict( + True, + "OVERFIRING", + f"two ticks {interval}s apart, under the {OVERFIRE_UNDER_SECONDS}s floor", + ) + + if age > DEAD_AFTER_SECONDS: + if suppression is not None: + return Verdict( + False, + "SUPPRESSED", + f"{age}s since last tick, but {seat} is {suppression}", + ) + return Verdict( + True, + "DEAD", + f"{age}s since {seat} last ticked (dead after {DEAD_AFTER_SECONDS}s)", + ) + + return Verdict(False, "OK", f"{age}s since last tick") + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + ap.add_argument("--worktree", required=True, help="absolute path of the worktree to watch") + ap.add_argument("--seat", required=True, help="that worktree's seat name in seat-tick.last") + ap.add_argument("--state", type=Path, default=_MEFOR / "seat-tick.state.json") + ap.add_argument("--last", type=Path, default=_MEFOR / "seat-tick.last") + ap.add_argument("--previous-tick", type=int, default=None) + args = ap.parse_args(argv) + + # A MISSING INSTRUMENT IS NOT A CLEAN RESULT. Reporting OK here would be the same false green the + # alarm exists to catch, one level up. + for path in (args.state, args.last): + if not path.is_file(): + print(f"seat-clock-alarm: CANNOT MEASURE -- {path} is absent", file=sys.stderr) + return 2 + + try: + state = read_state(args.state) + except Unreadable as exc: + print(f"seat-clock-alarm: CANNOT MEASURE -- {exc}", file=sys.stderr) + return 2 + last_line = args.last.read_text(encoding="utf-8", errors="replace").strip() + + verdict = evaluate(args.worktree, args.seat, state, last_line, time.time(), args.previous_tick) + # THE DENOMINATOR IS PART OF THE RESULT: a state file holding 3 worktrees and one holding 60 must + # not print the same reassuring line. + print(f"seat-clock-alarm: {verdict.code} -- {verdict.detail} ({len(state)} worktrees in state)") + return 1 if verdict.alarm else 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/tests/test_coord_seat_clock_alarm.py b/tests/test_coord_seat_clock_alarm.py new file mode 100644 index 00000000..13ccd073 --- /dev/null +++ b/tests/test_coord_seat_clock_alarm.py @@ -0,0 +1,345 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Organization and contributors +"""The seat-clock alarm must fire on a dead chain and stay quiet on a healthy one (#1269). + +Every test is one half of a PAIR, because the item's whole thesis is that the OBVIOUS +implementation reads healthy at the moment it should fire. A suite of must-fire arms alone would be +satisfied by the freshness-only version this build exists to replace. + +``test_it_fires_when_the_watched_worktree_is_absent_from_a_FRESH_state_file`` is the discriminating +one: it is green for the broken implementation and red for a correct one, so it is the only arm that +separates them. Every other arm guards a fault that would make a HEALTHY clock report as broken -- +and a false-positive watchdog is not a safe failure mode, it is a slow-acting off switch. +""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from types import ModuleType + +import pytest + +_ROOT = Path(__file__).resolve().parents[1] +_SCRIPT = _ROOT / "scripts" / "coord" / "seat_clock_alarm.py" + + +def _load() -> ModuleType: + spec = importlib.util.spec_from_file_location("seat_clock_alarm", _SCRIPT) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = mod + spec.loader.exec_module(mod) + return mod + + +alarm = _load() + +WATCHED = r"C:\work\lane-a" +NOW = 1_787_000_000.0 +FRESH = int(NOW) - 60 +STALE_TICK = int(NOW) - (60 * 60) + + +def _state(**extra: int) -> dict[str, int]: + base = {r"c:\work\other-lane": FRESH} + base.update(extra) + return base + + +# -------------------------------------------------------------------------------------------- +# THE DISCRIMINATING PAIR +# -------------------------------------------------------------------------------------------- + + +def test_it_fires_when_the_watched_worktree_is_absent_from_a_FRESH_state_file() -> None: + """MUST FIRE, AND THIS IS THE ONLY ARM THAT SEPARATES THE TWO IMPLEMENTATIONS. + + Other seats keep the heartbeat fresh, so a freshness-only alarm is GREEN here -- at exactly the + moment the watched seat has stopped being woken. Measured precedent: ticks 59m58s apart while the + file stayed under two minutes old for the whole hour.""" + v = alarm.evaluate(WATCHED, "lane-a", _state(), "other-lane=SENT:abc", NOW) + assert v.alarm is True + assert v.code == "ABSENT" + assert "no entry" in v.detail + + +def test_it_is_silent_when_the_watched_worktree_is_present_and_recent() -> None: + """MUST NOT FIRE -- the twin. Without it, an alarm that fired unconditionally would pass above.""" + v = alarm.evaluate( + WATCHED, "lane-a", _state(**{r"c:\work\lane-a": FRESH}), "lane-a=SENT:abc", NOW + ) + assert v.alarm is False + assert v.code == "OK" + + +# -------------------------------------------------------------------------------------------- +# THE FIRST-MATCH TRAP -- measured on the live file, three seats twice each +# -------------------------------------------------------------------------------------------- + + +def test_a_seat_named_TWICE_reads_as_SENT_not_as_the_first_token() -> None: + """MUST NOT FIRE, AND THIS IS A LIVE SHAPE RATHER THAN A HYPOTHETICAL. + + Measured 2026-08-23: the real one-liner carried steward, lander and dispatcher TWICE EACH -- + once STALE(no-live-session), once SENT:. Taking the first match reads STALE for all three + while the clock ticks normally, which is the same first-match trap that cost four seats a wrong + answer that morning.""" + line = "lane-a=STALE(no-live-session) other=SENT:zzz lane-a=SENT:20260823T184037416-texbqx" + assert alarm.suppression_for(line, "lane-a") is None, "a send anywhere in the line must win" + + +def test_a_seat_named_only_as_STALE_is_read_as_suppressed() -> None: + """MUST NOT FIRE AS A DEATH -- the twin of the arm above, differing only by the SENT token. + + Without this pair, 'a send wins' could be implemented as 'never suppress' and both would pass.""" + line = "lane-a=STALE(no-live-session) other=SENT:zzz" + assert alarm.suppression_for(line, "lane-a") == "STALE(no-live-session)" + + +# -------------------------------------------------------------------------------------------- +# SUPPRESSION -- a deliberate gap is not a dead chain +# -------------------------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "status", + [ + "THROTTLED(last-send-120s-ago,floor-360s)", + "COLD(3-ticks,2-peer-mail,0-unreadable,oldest=20m,needs-send_message)", + "BACKLOG(4-pending,oldest=30m,suppressed)", + "STALE(no-live-session)", + ], +) +def test_a_long_gap_under_a_suppression_token_does_not_alarm(status: str) -> None: + """MUST NOT FIRE. An 88-minute gap once read as 'chain broken' and was deliberate suppression on + a seat taking continuous turns. All four tokens are read from the emitter, not from the item -- + STALE and the (roster-blind) suffix postdate the item's list.""" + v = alarm.evaluate( + WATCHED, + "lane-a", + _state(**{r"c:\work\lane-a": STALE_TICK}), + f"lane-a={status}", + NOW, + ) + assert v.alarm is False + assert v.code == "SUPPRESSED" + + +def test_the_same_long_gap_with_NO_suppression_token_does_alarm() -> None: + """MUST FIRE -- the twin. Otherwise 'honour suppression' could be implemented as 'never alarm on + age' and every suppression arm above would still pass.""" + v = alarm.evaluate( + WATCHED, + "lane-a", + _state(**{r"c:\work\lane-a": STALE_TICK}), + "lane-a=SENT:abc", + NOW, + ) + assert v.alarm is True + assert v.code == "DEAD" + + +# -------------------------------------------------------------------------------------------- +# THROTTLE-AGE CONSISTENCY -- one comparison that rejects the decoy and catches corruption +# -------------------------------------------------------------------------------------------- + + +def test_a_throttle_record_that_contradicts_itself_does_NOT_silence_the_alarm() -> None: + """MUST FIRE. The known decoy claims THROTTLED with floor-360s while its own stated age is + ~36000s -- a hundred-fold contradiction, since throttling means suppressed because a send was too + RECENT. Honouring it is how an alarm goes quiet over a ten-hour-old file and reports nothing + wrong, because from its point of view nothing IS wrong.""" + line = "lane-a=THROTTLED(last-send-35997s-ago,floor-360s)" + assert alarm.throttle_is_credible("THROTTLED(last-send-35997s-ago,floor-360s)") is False + v = alarm.evaluate(WATCHED, "lane-a", _state(**{r"c:\work\lane-a": STALE_TICK}), line, NOW) + assert v.alarm is True and v.code == "DEAD" + + +def test_a_CREDIBLE_throttle_record_is_still_honoured() -> None: + """MUST NOT FIRE -- the twin. A sanity check that rejected every throttle would turn ordinary + suppression into a permanent alarm, which is the false-positive direction that gets it ignored.""" + assert alarm.throttle_is_credible("THROTTLED(last-send-120s-ago,floor-360s)") is True + + +def test_a_throttle_claiming_a_send_in_the_FUTURE_is_rejected() -> None: + """MUST FIRE. The decoy's literal text is `last-send--35997s-ago` -- a NEGATIVE age, i.e. a send + ten hours from now. Both readings of that string are self-contradictory and either must trip.""" + assert alarm.throttle_is_credible("THROTTLED(last-send--35997s-ago,floor-360s)") is False + + +# -------------------------------------------------------------------------------------------- +# DEDUPE BY TICK IDENTITY, and the over-firing bound +# -------------------------------------------------------------------------------------------- + + +def test_an_UNCHANGED_stamp_is_the_same_tick_not_a_zero_length_interval() -> None: + """MUST NOT FIRE, AND THIS IS THE MEASURED FAULT. A raw scan produced ten 0.0-minute intervals + and flagged over-firing ten times; 22 records were about 12 ticks differing in milliseconds. + Identity is the stamp VALUE -- seeing it twice is one tick observed twice.""" + v = alarm.evaluate( + WATCHED, + "lane-a", + _state(**{r"c:\work\lane-a": FRESH}), + "lane-a=SENT:abc", + NOW, + previous_tick=FRESH, + ) + assert v.alarm is False, "the same stamp seen twice is not a zero-second interval" + assert v.code == "OK" + + +def test_two_GENUINELY_close_ticks_do_alarm_as_over_firing() -> None: + """MUST FIRE -- the twin of the dedupe arm. Over-firing is the EXPENSIVE fault: every tick wakes a + seat and spends a turn, so a runaway clock burns the budget the alarm exists to protect. Six + manual runs once cost ~9 points of a shared pool in nine minutes.""" + v = alarm.evaluate( + WATCHED, + "lane-a", + _state(**{r"c:\work\lane-a": FRESH}), + "lane-a=SENT:abc", + NOW, + previous_tick=FRESH - 30, + ) + assert v.alarm is True + assert v.code == "OVERFIRING" + + +# -------------------------------------------------------------------------------------------- +# PATH DISCIPLINE AND THE MISSING-INSTRUMENT REFUSAL +# -------------------------------------------------------------------------------------------- + + +def test_the_watched_path_matches_regardless_of_case_and_separator() -> None: + """Windows path casing has already killed this clock once -- seats.json carried one directory + under two casings and the tick script died on the parse. The state file's own keys are + lowercased, so the lookup must normalise both sides.""" + state = {r"c:\work\lane-a": FRESH} + v = alarm.evaluate("C:/Work/Lane-A", "lane-a", state, "lane-a=SENT:abc", NOW) + assert v.code == "OK", "a case- or separator-differing path must still find its entry" + + +def test_a_missing_instrument_returns_2_rather_than_reporting_OK( + tmp_path: Path, +) -> None: + """A MISSING INSTRUMENT IS NOT A CLEAN RESULT. Returning 0 here would be the same false green the + alarm exists to catch, one level up -- and it is how a glob that lands nowhere reports health.""" + rc = alarm.main( + [ + "--worktree", + WATCHED, + "--seat", + "lane-a", + "--state", + str(tmp_path / "nope.json"), + "--last", + str(tmp_path / "nope.last"), + ] + ) + assert rc == 2, "absent instruments must be distinguishable from a healthy clock" + + +def test_the_cli_reports_its_denominator( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """A state file holding 3 worktrees and one holding 60 must not print the same reassuring line.""" + state = tmp_path / "s.json" + state.write_text(json.dumps({r"c:\work\lane-a": FRESH, r"c:\work\b": FRESH}), encoding="utf-8") + last = tmp_path / "s.last" + last.write_text("2026-01-01T00:00:00Z\tlane-a=SENT:abc", encoding="utf-8") + alarm.main( + [ + "--worktree", + WATCHED, + "--seat", + "lane-a", + "--state", + str(state), + "--last", + str(last), + ] + ) + assert "2 worktrees in state" in capsys.readouterr().out + + +# --------------------------------------------------------------------------------------------- +# A BROKEN INSTRUMENT IS NOT AN ALARM CONDITION. +# +# The first cut reported DEAD/ABSENT when it could not READ the state file -- a false positive that +# fires for EVERY watched seat at once, so the first false alarm is also the loudest. That is the +# worst possible introduction for a tool whose only value is being believed, and the module docstring +# already names the consequence: a false-positive watchdog is a slow-acting off switch. +# +# exit 2 CANNOT MEASURE already existed for the MISSING file. These arms are its second application. +# --------------------------------------------------------------------------------------------- + + +def _files(tmp_path: Path, state_text: str) -> tuple[Path, Path]: + s = tmp_path / "state.json" + s.write_text(state_text, encoding="utf-8") + last = tmp_path / "state.last" + last.write_text("2026-01-01T00:00:00Z\tlane-a=SENT:abc", encoding="utf-8") + return s, last + + +def _rc(tmp_path: Path, state_text: str) -> int: + s, last = _files(tmp_path, state_text) + return alarm.main( + [ + "--worktree", + WATCHED, + "--seat", + "lane-a", + "--state", + str(s), + "--last", + str(last), + ] + ) + + +def test_a_CORRUPT_state_file_cannot_measure_rather_than_alarming( + tmp_path: Path, +) -> None: + """MUST NOT ALARM. This is an OBSERVED event here, not a hypothetical: the item records that a + Windows path-casing collision killed this clock once already by dying on a JSON parse.""" + assert _rc(tmp_path, '{"c:\\work\\lane-a": 178700') == 2 + + +def test_a_state_SCHEMA_this_reader_does_not_know_cannot_measure( + tmp_path: Path, +) -> None: + """MUST NOT ALARM. Every record skipped is a signal the schema changed, never an empty registry. + + Before this, the reader dropped them silently and the alarm reported ABSENT for every seat -- + confidently, on data it had never understood. The Liaison's form of the same rule: when every + field you asked for comes back empty, you are querying a schema you did not read.""" + assert _rc(tmp_path, '{"c:\\work\\lane-a": {"last": 1787000000}}') == 2 + + +def test_a_GENUINELY_EMPTY_registry_still_alarms(tmp_path: Path) -> None: + """MUST ALARM -- THE TWIN, and the arm that stops the fix swallowing the real case. + + An empty object is not an unreadable one. No records means the watched seat really has no tick, + which is the discriminating condition this whole tool exists for. A schema check that treated + {} as unreadable would silence the alarm exactly when it should fire.""" + assert _rc(tmp_path, "{}") == 1 + + +def test_a_MISSING_state_file_still_cannot_measure(tmp_path: Path) -> None: + """REGRESSION. The original exit-2 arm must survive the new ones.""" + rc = alarm.main( + [ + "--worktree", + WATCHED, + "--seat", + "lane-a", + "--state", + str(tmp_path / "absent.json"), + "--last", + str(tmp_path / "absent.last"), + ] + ) + assert rc == 2 diff --git a/tests/tooling_manifest.txt b/tests/tooling_manifest.txt index 8c40b8fd..c74a3795 100644 --- a/tests/tooling_manifest.txt +++ b/tests/tooling_manifest.txt @@ -62,6 +62,7 @@ tests/test_coord_overlap_cache.py tests/test_coord_overlap_signals.py tests/test_coord_presence.py tests/test_coord_dispatch_gate.py +tests/test_coord_seat_clock_alarm.py tests/test_coord_throughput.py tests/test_doc_guards_lane.py tests/test_quality_expiry_audit.py