Add althing seat monitor-behavior test artifacts

Disposable harnesses from the wake-RFI thread (01M2WT3F): exit1-behavior-test.py covers the monitor exit-code/gate path including a sustained-failure/no-backoff assertion; hash_gate_repro.py covers the detection-time hash persistence crash-before-read boundary with consumption/unchanged/changed controls. Both fail nonzero on mismatch. Committing so seat artifact provenance rides git rather than file mtimes.
2026-09-19 05:42:52 -07:00
parent 5fee7868b3
commit e43e2626f1
2 changed files with 181 additions and 0 deletions
@@ -0,0 +1,87 @@
+"""Behavioral repro: cron monitor hash gate persists at DETECTION time, so a
+run that dies before consuming mail is suppressed on the next tick if the
+observed output is unchanged. Disposable job dict + in-memory persist; the
+live job and store are untouched.
+
+Controls per forseti (2740):
+  - positive control: changed ID set DOES admit another run
+  - suppression control: unchanged output after a SUCCESSFUL run suppresses
+  - crash case: persist detection, simulate death before 'read', reload state
+    as after restart, rerun with SAME id set -> suppressed (the finding)
+Each condition repeated 3x.
+"""
+import sys, copy
+sys.path.insert(0, "/home/lkraven/.hermes/hermes-agent")
+from cron import monitor
+
+STATE = {}  # in-memory stand-in for jobs.json monitor_state
+
+def fake_persist(job_id, new_hash, output):
+    STATE[job_id] = {"last_output_hash": new_hash, "snapshot": output}
+
+monitor._persist_monitor_state = fake_persist
+monitor._read_last_output = lambda job_id: STATE.get(job_id, {}).get("snapshot", "")
+monitor._write_last_output = lambda job_id, output: STATE[job_id].__setitem__("snapshot", output)
+
+OUTPUT = [""]  # current observed monitor output, set per tick
+
+def fake_source(job):
+    return True, OUTPUT[0]
+
+monitor._run_monitor_source = fake_source
+
+def tick(ids):
+    """One monitor tick with the given unread id set. Returns (ok, changed)."""
+    OUTPUT[0] = "\n".join(f"[{i}]" for i in ids) + "\n"
+    job = {"id": "disposable-test-job", "monitor_script": "stub"}
+    if "disposable-test-job" in STATE:  # simulate fresh load from store
+        job["monitor_state"] = copy.deepcopy(STATE["disposable-test-job"])
+    out = monitor.check_monitor(job)
+    return out.ok, getattr(out, "changed", None)
+
+results = []
+for rep in range(1, 4):
+    # --- consumption control: a SUCCESSFUL run consumes -> output genuinely changes.
+    # (Was: same ids after success, which only proved unchanged-input suppression.
+    #  Fix: after the run consumes, the mailbox is empty, so the next observation
+    #  is the idle sentinel; that transition admits once, stable idle suppresses.)
+    STATE.clear()
+    r1 = tick(["100"])                      # first observation -> wake
+    r2 = tick([])                           # agent consumed -> monitor now says idle
+    r3 = tick([])                           # still idle -> suppressed
+    results.append((f"rep{rep} control-consume: first={r1} after-consume={r2} stable-idle={r3}",
+                    r1 == (True, True) and r2 == (True, True) and r3 == (True, False)))
+
+    # --- suppression control (renamed honestly): unchanged output suppresses ---
+    STATE.clear()
+    r1 = tick(["100"])
+    r2 = tick(["100"])                      # no consumption, same set next tick
+    results.append((f"rep{rep} control-unchanged-suppresses: first={r1} unchanged={r2}",
+                    r1 == (True, True) and r2 == (True, False)))
+
+    # --- positive control: changed ID set admits another run ---
+    STATE.clear()
+    r1 = tick(["100"])
+    r2 = tick(["100", "101"])               # new mail -> changed -> wake
+    results.append((f"rep{rep} control-change: first={r1} changed={r2}",
+                    r1 == (True, True) and r2 == (True, True)))
+
+    # --- crash case: detection persisted, agent died BEFORE consuming mail ---
+    STATE.clear()
+    r1 = tick(["200", "201"])               # detection: hash persisted pre-agent
+    # (agent session would run `postbox read` here; simulate it dying first)
+    r2 = tick(["200", "201"])               # next tick, same unread set, post-restart reload
+    r3 = tick(["200", "201"])               # and again: unbounded silence
+    results.append((f"rep{rep} crash-before-read: detect={r1} retry1={r2} retry2={r3}",
+                    r1 == (True, True) and r2 == (True, False) and r3 == (True, False)))
+
+    # --- recovery path: stranded mail + a NEW message eventually wakes ---
+    r4 = tick(["200", "201", "202"])        # a genuinely new message re-wakes,
+    results.append((f"rep{rep} recovery-on-new-mail: {r4}", r4 == (True, True)))
+
+ok = True
+for line, passed in results:
+    print(("PASS " if passed else "FAIL ") + line)
+    ok = ok and passed
+print("ALL PASS" if ok else "SOME FAILED")
+sys.exit(0 if ok else 1)