From 45bba3fa7f7c5a48a480f691316d210c2d655868 Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Thu, 9 Jul 2026 22:33:41 -0400 Subject: [PATCH 1/3] feat(routing): Phase-2 reship-monitor (slice 4, the Phase-3 trigger) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reship_stats(summary): pure summary of what a routed run re-shipped. The load-bearing signal is the SHARD reship ratio (shards shipped / shards present) — appending new big files ships new SOLOS and leaves shards untouched (~0), so healthy GROWTH never trips it; re-shipping many whole SHARDS does, which is small-file churn = the uniform-medium-churn signature Phase 3 (leveling) is for. byte_ratio reported for context only (conflates growth w/ churn, not the trigger). reship_summary_line for the print+email. __main__ size-routing completion: prints the reship line; when shard_ratio > 25% (RESHIP_WATCH_RATIO) flags [RESHIP-WATCH] and the completion email subject/body carry it (warn-only; the operator decides whether to consider Phase 3 for that target). Tests test_reship_monitor.py (7): append -> no watch, wide shard churn -> watch, strict 25% threshold, all-solos never watches, byte-ratio math, empty target, summary line. Suite 238 -> 245. Based on main, not stacked. Co-Authored-By: Claude Opus 4.8 --- archive.py | 72 ++++++++++++++++++++++++++++++- tests/test_reship_monitor.py | 82 ++++++++++++++++++++++++++++++++++++ 2 files changed, 152 insertions(+), 2 deletions(-) create mode 100644 tests/test_reship_monitor.py diff --git a/archive.py b/archive.py index 8e460f6..f7d93d7 100644 --- a/archive.py +++ b/archive.py @@ -1060,6 +1060,58 @@ def route_and_ship(project_name, source_folder, file_pattern, survivors, exclusi } +# The reship-monitor's watch threshold (RFC §2.7): if more than this fraction of a +# target's SHARD objects re-ship in a run, that is the uniform-medium-churn signature +# (a large small-file tier changing repeatedly, which routing re-ships as whole shards +# and exclusion can't drop) — the one case Phase 3 (leveling) is for. Warn-only. +RESHIP_WATCH_RATIO = 0.25 + + +def reship_stats(summary, watch_ratio=RESHIP_WATCH_RATIO): + """ + Reship-monitor (Phase 2, RFC §2.7) — pure summary of what a routed run re-shipped, + from route_and_ship's return dict. The load-bearing signal is the SHARD reship ratio + (shards shipped / shards present): appending new big files ships new SOLOS and leaves + shards untouched (~0), so healthy GROWTH does not trip it; re-shipping many whole + SHARDS does — that is small-file churn, the Phase-3 trigger. byte_ratio is reported + for context only (it conflates growth with churn, so it is NOT the trigger). + + Returns a dict incl. `watch` (True iff shards exist and shard_ratio > watch_ratio). + """ + objects = (summary.get("manifest") or {}).get("objects") or [] + shipped = set(summary.get("shipped") or []) + by_slot = {o.get("slot"): o for o in objects} + + shard_slots = [o["slot"] for o in objects if o.get("kind") == "shard"] + solo_slots = [o["slot"] for o in objects if o.get("kind") == "solo"] + shards_total = len(shard_slots) + shards_shipped = sum(1 for s in shard_slots if s in shipped) + shard_ratio = (shards_shipped / shards_total) if shards_total else 0.0 + + total_bytes = sum((o.get("source_bytes") or 0) for o in objects) + shipped_bytes = sum((by_slot[s].get("source_bytes") or 0) + for s in shipped if s in by_slot) + byte_ratio = (shipped_bytes / total_bytes) if total_bytes else 0.0 + + return { + "shards_total": shards_total, "shards_shipped": shards_shipped, + "shard_ratio": shard_ratio, + "solos_total": len(solo_slots), + "solos_shipped": sum(1 for s in solo_slots if s in shipped), + "total_bytes": total_bytes, "shipped_bytes": shipped_bytes, + "byte_ratio": byte_ratio, + "watch_ratio": watch_ratio, + "watch": shards_total > 0 and shard_ratio > watch_ratio, + } + + +def reship_summary_line(stats): + """One-line human summary of reship_stats, for the run print + completion email.""" + return ("reship: {}/{} shards ({:.0%}), {}/{} solos, {:.0%} of bytes" + .format(stats["shards_shipped"], stats["shards_total"], stats["shard_ratio"], + stats["solos_shipped"], stats["solos_total"], stats["byte_ratio"])) + + # --------------------------------------------------------------------------- # Exclusion primitives (see docs/EXCLUSION_SPEC.md). # @@ -2722,18 +2774,34 @@ def ship_one(obj, stem): f"{n_skip} skipped (unchanged), {len(summary['dropped'])} dropped") for w in summary["warnings"]: print(f" WARNING: {w}") + # Reship-monitor (RFC §2.7): the Phase-3 trigger. Warn-only. + stats = reship_stats(summary) + print(f" Reship: {reship_summary_line(stats)}") + if stats["watch"]: + print(f" [RESHIP-WATCH] shard reship {stats['shard_ratio']:.0%} > " + f"{stats['watch_ratio']:.0%} — uniform-medium-churn signature " + f"(consider Phase 3 for this target)") print(f" Manifest: {manifest_path(vault_dir, summary['badge'])}") + watch = stats["watch"] body = (f"Size-routing archive complete for {project_name}.\n\n" f"Badge {summary['badge']} (K={summary['shard_count']}, " f"T_small={summary['t_small']} bytes)\n" f"Objects: {summary['n_objects']} total — {n_ship} shipped, " f"{n_skip} skipped (unchanged), {len(summary['dropped'])} dropped.\n" + f"{reship_summary_line(stats)}\n" + ("\nBudget warnings:\n " + "\n ".join(summary["warnings"]) if summary["warnings"] else "") - + f"\n\nShipped slots: {', '.join(summary['shipped']) or '(none)'}\n" + + (f"\n[RESHIP-WATCH] {stats['shards_shipped']}/{stats['shards_total']} shards " + f"({stats['shard_ratio']:.0%}) re-shipped, over the {stats['watch_ratio']:.0%} " + f"threshold — this is the uniform-medium-churn signature (a small-file tier " + f"changing repeatedly). Routing re-ships whole shards for this; it is the " + f"signal to consider Phase 3 (leveling) for this target.\n" if watch else "") + + f"\nShipped slots: {', '.join(summary['shipped']) or '(none)'}\n" f"Manifest: {manifest_path(vault_dir, summary['badge'])}\n") - send_alert_email(emails, f"[SUCCESS] Fortress size-routing: {project_name}", body) + subject = (f"[RESHIP-WATCH] Fortress size-routing: {project_name}" if watch + else f"[SUCCESS] Fortress size-routing: {project_name}") + send_alert_email(emails, subject, body) sys.exit(0) # Step 1: Find files, zip, and verify on the compute system diff --git a/tests/test_reship_monitor.py b/tests/test_reship_monitor.py new file mode 100644 index 0000000..7908536 --- /dev/null +++ b/tests/test_reship_monitor.py @@ -0,0 +1,82 @@ +""" +Unit tests for the Phase-2 reship-monitor (docs/RFC_incremental_v2.md §2.7, slice 4). + +The load-bearing signal is the SHARD reship ratio: appending new big files ships new +SOLOS and leaves shards untouched (healthy growth, ~0), while re-shipping many whole +SHARDS is small-file churn (the Phase-3 trigger). Pure function over route_and_ship's +summary dict — no tape/Slurm/Globus. +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import archive # noqa: E402 + + +def obj(slot, kind, source_bytes): + return {"slot": slot, "kind": kind, "source_bytes": source_bytes} + + +def summ(objects, shipped): + return {"manifest": {"objects": objects}, "shipped": list(shipped), + "shard_count": 4, "badge": "b", "t_small": 1, + "n_objects": len(objects), "skipped": [], "dropped": [], "warnings": []} + + +BIG = 300 * 1024 * 1024 + + +class TestReshipStats(unittest.TestCase): + def test_append_growth_does_not_watch(self): + # New solo shipped; both shards untouched -> shard_ratio 0 -> no watch. + s = summ([obj("solo:new", "solo", BIG), + obj("shard:0", "shard", 100), obj("shard:1", "shard", 100)], + shipped=["solo:new"]) + st = archive.reship_stats(s) + self.assertEqual(st["shards_shipped"], 0) + self.assertEqual(st["shard_ratio"], 0.0) + self.assertFalse(st["watch"]) + self.assertEqual(st["solos_shipped"], 1) + + def test_wide_shard_churn_watches(self): + objs = [obj("shard:%d" % k, "shard", 100) for k in range(4)] + s = summ(objs, shipped=["shard:0", "shard:1", "shard:2"]) # 3/4 = 75% + st = archive.reship_stats(s) + self.assertAlmostEqual(st["shard_ratio"], 0.75) + self.assertTrue(st["watch"]) + + def test_threshold_is_strict(self): + objs = [obj("shard:%d" % k, "shard", 100) for k in range(4)] + self.assertFalse(archive.reship_stats(summ(objs, ["shard:0"]))["watch"]) # 25% not > 25% + self.assertTrue(archive.reship_stats(summ(objs, ["shard:0", "shard:1"]))["watch"]) # 50% + + def test_all_solos_never_watches(self): + objs = [obj("solo:%d" % i, "solo", BIG) for i in range(3)] + s = summ(objs, shipped=[o["slot"] for o in objs]) # every solo shipped (growth) + st = archive.reship_stats(s) + self.assertEqual(st["shards_total"], 0) + self.assertFalse(st["watch"]) + + def test_byte_ratio_math(self): + objs = [obj("shard:0", "shard", 30), obj("shard:1", "shard", 70)] + st = archive.reship_stats(summ(objs, shipped=["shard:0"])) + self.assertEqual(st["total_bytes"], 100) + self.assertEqual(st["shipped_bytes"], 30) + self.assertAlmostEqual(st["byte_ratio"], 0.30) + + def test_empty_target(self): + st = archive.reship_stats(summ([], shipped=[])) + self.assertFalse(st["watch"]) + self.assertEqual(st["shard_ratio"], 0.0) + self.assertEqual(st["byte_ratio"], 0.0) + + def test_summary_line(self): + st = archive.reship_stats(summ( + [obj("shard:0", "shard", 100), obj("shard:1", "shard", 100)], ["shard:0"])) + line = archive.reship_summary_line(st) + self.assertIn("1/2 shards", line) + + +if __name__ == "__main__": + unittest.main() From f16b64fbb44dbfff10e1fd32584febee3088ca08 Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Thu, 9 Jul 2026 22:54:44 -0400 Subject: [PATCH 2/3] feat(routing): durable reship-watch streak in the manifest (dependable Phase-3 trigger) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Slice 4's [RESHIP-WATCH] email line is a fleeting per-run signal: it fires once routing is enabled, cries wolf on a one-off re-processing pass, and rides in a routine email the operator tunes out. Make it a DURABLE, de-noised, queryable trigger instead. - update_reship_watch(prior_watch, stats, run_epoch, sustain=3): pure fold of a run's reship_stats into a streak block persisted in the target's manifest — consecutive_trips (+1 on a tripped run, reset to 0 on a quiet one), sustained (>= sustain runs), sustained_since, last_ratio/last_run/shards_*, and an `alerted` idempotency latch OWNED BY the deployment alerter (engine only preserves it across trips, clears it on reset so a re-crossing re-alerts once). - route_and_ship computes the watch over the final post-prune object set and always writes the manifest once at the end (records a streak RESET even on an all-skipped run). Adds reship_watch to the return dict. - __main__ surfaces the streak (N/sustain, SUSTAINED?) in the completion print + email so a single trip reads as "not yet sustained", not an action item. The Phase-3 alert (a Teams @mention, fired once on crossing into sustained) is deployment glue that reads this manifest field — kept out of the engine. Tests: 9 pure-function cases (mock summary dicts + prior blocks: increment, reset, sustain-after-N, alerted latch preserve/clear, custom sustain, end-to-end from reship_stats) + 3 persistence cases in route_and_ship (persist+reset, append-does-not-accumulate, repeated-churn-sustains). Suite 245 -> 257. Co-Authored-By: Claude Opus 4.8 --- archive.py | 97 ++++++++++++++++++++++++++++++++---- tests/test_reship_monitor.py | 90 +++++++++++++++++++++++++++++++++ tests/test_routing_wiring.py | 47 +++++++++++++++++ 3 files changed, 224 insertions(+), 10 deletions(-) diff --git a/archive.py b/archive.py index f7d93d7..29294ea 100644 --- a/archive.py +++ b/archive.py @@ -1046,17 +1046,26 @@ def route_and_ship(project_name, source_folder, file_pattern, survivors, exclusi if dropped: manifest["objects"] = [o for o in manifest["objects"] if o["slot"] in current_slots] - write_manifest(vault_dir, badge, manifest) - elif not manifest["objects"] and not prior: - # Nothing shipped and no prior manifest (e.g. everything skipped is impossible - # at baseline) — still write so reclaim can see the (empty) target. - write_manifest(vault_dir, badge, manifest) + + # Reship-watch (RFC §2.7): fold this run's SHARD reship ratio into a durable + # streak carried in the manifest — the dependable Phase-3 trigger. Computed over + # the FINAL object set + shipped list (post-prune) so append (new solos, shards + # untouched) reads ~0 and a wide shard re-ship reads high. The alerter (deployment + # glue) reads `reship_watch` and fires once on crossing into a sustained watch. + stats = reship_stats({"manifest": manifest, "shipped": shipped}) + manifest["reship_watch"] = update_reship_watch( + (prior or {}).get("reship_watch"), stats, now_epoch) + + # Always persist once at the end: this records the reship-watch streak even on an + # all-skipped run (which must RESET the streak) and lets reclaim see an empty + # baseline target. Per-object writes above keep resume crash-safe. + write_manifest(vault_dir, badge, manifest) return { "badge": badge, "shard_count": shard_count, "t_small": t_small, "n_objects": len(objects), "shipped": shipped, "skipped": skipped, "dropped": dropped, "warnings": warnings, "exclusion": exclusion, - "manifest": manifest, + "reship_watch": manifest["reship_watch"], "manifest": manifest, } @@ -1112,6 +1121,61 @@ def reship_summary_line(stats): stats["solos_shipped"], stats["solos_total"], stats["byte_ratio"])) +# Consecutive tripped runs before a reship-watch is treated as SUSTAINED. A single +# tripped run is a one-off (a manual re-processing pass, a bulk mtime touch); only a +# streak is the uniform-medium-churn signature Phase 3 exists for. De-noising the +# per-run signal here is what makes the Phase-3 trigger DEPENDABLE rather than a +# cry-wolf email line (see docs/RFC_incremental_v2.md §2.7). +RESHIP_WATCH_SUSTAIN_RUNS = 3 + + +def update_reship_watch(prior_watch, stats, run_epoch, + sustain=RESHIP_WATCH_SUSTAIN_RUNS): + """ + Fold this run's reship_stats into the target's DURABLE watch streak (RFC §2.7). + Pure: takes the prior `reship_watch` block (or None, at baseline) plus this run's + stats, returns the new block. It is persisted in the target's manifest so the + Phase-3 trigger is durable + queryable state — NOT a fleeting per-run email line + (a) it survives across runs, (b) it de-noises one-offs, and (c) the deployment + alerter reads it to fire ONE exception alert on crossing into a sustained watch. + + Streak semantics: + - `consecutive_trips`: +1 when this run tripped (stats["watch"], i.e. shard + reship ratio > RESHIP_WATCH_RATIO); RESET to 0 on any quiet run. So one + re-processing pass cannot raise the signal on its own. + - `sustained`: consecutive_trips >= `sustain` (default 3) — the de-noised + signal the alerter and RUNBOOK key off. + - `sustained_since`: run_epoch at which the streak first reached `sustain`, + preserved while it holds (lets the alerter re-nag on a low cadence); None + while not sustained. + - `alerted`: an idempotency latch OWNED BY THE DEPLOYMENT ALERTER, never set + here. This function only PRESERVES it across tripped runs and CLEARS it + (-> False) whenever the streak resets, so the next crossing re-alerts once. + """ + prior = prior_watch or {} + tripped = bool(stats.get("watch")) + trips = (int(prior.get("consecutive_trips") or 0) + 1) if tripped else 0 + sustained = trips >= sustain + if tripped: + alerted = bool(prior.get("alerted")) + sustained_since = (prior.get("sustained_since") or run_epoch) if sustained else None + else: + # Streak broke: drop the alert latch so a fresh crossing alerts again. + alerted = False + sustained_since = None + return { + "consecutive_trips": trips, + "sustained": sustained, + "sustain_runs": sustain, + "last_ratio": float(stats.get("shard_ratio") or 0.0), + "last_run": run_epoch, + "shards_total": int(stats.get("shards_total") or 0), + "shards_shipped": int(stats.get("shards_shipped") or 0), + "sustained_since": sustained_since, + "alerted": alerted, + } + + # --------------------------------------------------------------------------- # Exclusion primitives (see docs/EXCLUSION_SPEC.md). # @@ -2774,13 +2838,20 @@ def ship_one(obj, stem): f"{n_skip} skipped (unchanged), {len(summary['dropped'])} dropped") for w in summary["warnings"]: print(f" WARNING: {w}") - # Reship-monitor (RFC §2.7): the Phase-3 trigger. Warn-only. + # Reship-monitor (RFC §2.7): the Phase-3 trigger. Warn-only. The durable + # streak (reship_watch) lives in the manifest; a single tripped run is NOT + # the actionable signal — the deployment alerter only fires when it has been + # SUSTAINED for reship_watch.sustain_runs consecutive runs. stats = reship_stats(summary) + rw = summary.get("reship_watch") or {} print(f" Reship: {reship_summary_line(stats)}") if stats["watch"]: print(f" [RESHIP-WATCH] shard reship {stats['shard_ratio']:.0%} > " - f"{stats['watch_ratio']:.0%} — uniform-medium-churn signature " - f"(consider Phase 3 for this target)") + f"{stats['watch_ratio']:.0%} — uniform-medium-churn signature; " + f"tripped {rw.get('consecutive_trips')}/{rw.get('sustain_runs')} " + f"consecutive runs" + + (" — SUSTAINED (Phase-3 candidate)" if rw.get("sustained") + else " (not yet sustained)")) print(f" Manifest: {manifest_path(vault_dir, summary['badge'])}") watch = stats["watch"] @@ -2796,7 +2867,13 @@ def ship_one(obj, stem): f"({stats['shard_ratio']:.0%}) re-shipped, over the {stats['watch_ratio']:.0%} " f"threshold — this is the uniform-medium-churn signature (a small-file tier " f"changing repeatedly). Routing re-ships whole shards for this; it is the " - f"signal to consider Phase 3 (leveling) for this target.\n" if watch else "") + f"signal to consider Phase 3 (leveling) for this target.\n" + f"Sustained streak: {rw.get('consecutive_trips')}/{rw.get('sustain_runs')} " + f"consecutive tripped runs" + + (" — SUSTAINED; the exception alerter will @mention on crossing.\n" + if rw.get("sustained") + else "; not yet sustained (one-off runs do not trigger).\n") + if watch else "") + f"\nShipped slots: {', '.join(summary['shipped']) or '(none)'}\n" f"Manifest: {manifest_path(vault_dir, summary['badge'])}\n") subject = (f"[RESHIP-WATCH] Fortress size-routing: {project_name}" if watch diff --git a/tests/test_reship_monitor.py b/tests/test_reship_monitor.py index 7908536..646618b 100644 --- a/tests/test_reship_monitor.py +++ b/tests/test_reship_monitor.py @@ -78,5 +78,95 @@ def test_summary_line(self): self.assertIn("1/2 shards", line) +def stats_watch(watch, ratio=0.75, shards_total=4, shards_shipped=3): + """Minimal reship_stats-shaped dict for driving update_reship_watch directly.""" + return {"watch": watch, "shard_ratio": ratio, + "shards_total": shards_total, "shards_shipped": shards_shipped} + + +class TestUpdateReshipWatch(unittest.TestCase): + """ + The DURABLE, de-noised streak (the dependable Phase-3 trigger). A single tripped + run is a one-off; only `sustain` consecutive trips is the actionable signal. The + `alerted` latch is owned by the deployment alerter — the engine only preserves it + across trips and clears it on reset. + """ + + def test_baseline_quiet(self): + w = archive.update_reship_watch(None, stats_watch(False, 0.0, 4, 0), 1000) + self.assertEqual(w["consecutive_trips"], 0) + self.assertFalse(w["sustained"]) + self.assertFalse(w["alerted"]) + self.assertIsNone(w["sustained_since"]) + self.assertEqual(w["last_run"], 1000) + self.assertEqual(w["sustain_runs"], archive.RESHIP_WATCH_SUSTAIN_RUNS) + + def test_trip_increments_but_not_yet_sustained(self): + w = archive.update_reship_watch(None, stats_watch(True), 1) + self.assertEqual(w["consecutive_trips"], 1) + self.assertFalse(w["sustained"]) + w = archive.update_reship_watch(w, stats_watch(True), 2) + self.assertEqual(w["consecutive_trips"], 2) + self.assertFalse(w["sustained"]) # default sustain = 3 + + def test_reset_on_quiet_run(self): + w = archive.update_reship_watch(None, stats_watch(True), 1) + w = archive.update_reship_watch(w, stats_watch(True), 2) + w = archive.update_reship_watch(w, stats_watch(False, 0.0, 4, 0), 3) + self.assertEqual(w["consecutive_trips"], 0) + self.assertFalse(w["sustained"]) + self.assertIsNone(w["sustained_since"]) + + def test_sustained_after_N_consecutive(self): + w = None + for run in (1, 2, 3): + w = archive.update_reship_watch(w, stats_watch(True), run) + self.assertEqual(w["consecutive_trips"], 3) + self.assertTrue(w["sustained"]) + self.assertEqual(w["sustained_since"], 3) # crossed at run 3 + self.assertFalse(w["alerted"]) # engine never sets alerted + + def test_alerted_latch_preserved_across_trips(self): + w = None + for run in (1, 2, 3): + w = archive.update_reship_watch(w, stats_watch(True), run) + # deployment alerter fires once + latches + w["alerted"] = True + w2 = archive.update_reship_watch(w, stats_watch(True), 4) + self.assertTrue(w2["sustained"]) + self.assertTrue(w2["alerted"]) # preserved -> no re-alert + self.assertEqual(w2["sustained_since"], 3) # crossing epoch unchanged + + def test_alerted_cleared_on_reset_so_recrossing_realerts(self): + w = None + for run in (1, 2, 3): + w = archive.update_reship_watch(w, stats_watch(True), run) + w["alerted"] = True + w = archive.update_reship_watch(w, stats_watch(False, 0.0, 4, 0), 4) # quiet + self.assertFalse(w["alerted"]) + self.assertEqual(w["consecutive_trips"], 0) + + def test_custom_sustain_runs(self): + w = archive.update_reship_watch(None, stats_watch(True), 1, sustain=1) + self.assertTrue(w["sustained"]) + self.assertEqual(w["sustain_runs"], 1) + self.assertEqual(w["sustained_since"], 1) + + def test_records_ratio_and_counts(self): + w = archive.update_reship_watch(None, stats_watch(True, 0.5, 8, 4), 42) + self.assertAlmostEqual(w["last_ratio"], 0.5) + self.assertEqual(w["shards_total"], 8) + self.assertEqual(w["shards_shipped"], 4) + self.assertEqual(w["last_run"], 42) + + def test_driven_end_to_end_from_reship_stats(self): + # mock summary dict -> reship_stats -> update_reship_watch (fully pure path). + objs = [obj("shard:%d" % k, "shard", 100) for k in range(4)] + st = archive.reship_stats(summ(objs, ["shard:0", "shard:1", "shard:2"])) # 75% + w = archive.update_reship_watch(None, st, 100) + self.assertEqual(w["consecutive_trips"], 1) + self.assertAlmostEqual(w["last_ratio"], 0.75) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_routing_wiring.py b/tests/test_routing_wiring.py index 3e66365..172eba4 100644 --- a/tests/test_routing_wiring.py +++ b/tests/test_routing_wiring.py @@ -295,5 +295,52 @@ def test_resume_after_crash_persists_shipped_objects(self): self.assertEqual(len(summ["shipped"]), 2) +class TestReshipWatchPersistence(unittest.TestCase): + """route_and_ship persists the durable reship-watch streak in the manifest and + advances/resets it across real runs (the durable Phase-3 trigger, RFC §2.7).""" + + def test_persisted_and_resets_on_quiet_run(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c = cat(a=(1, SMALL), b=(1, SMALL), e=(1, SMALL), f=(1, SMALL)) + s1 = ship(c, vault, Recorder(c), shard_count_cfg=4) + self.assertIn("reship_watch", s1) + self.assertEqual(s1["reship_watch"]["consecutive_trips"], 1) # baseline ships all shards + m = archive.load_manifest(vault, s1["badge"]) + self.assertEqual(m["reship_watch"]["consecutive_trips"], 1) # persisted to disk + # unchanged re-run: nothing ships -> shard_ratio 0 -> streak resets. + s2 = ship(c, vault, Recorder(c), shard_count_cfg=4) + self.assertEqual(s2["shipped"], []) + self.assertEqual(s2["reship_watch"]["consecutive_trips"], 0) + self.assertFalse(s2["reship_watch"]["sustained"]) + + def test_append_does_not_accumulate_streak(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c1 = cat(big1=(1, BIG), a=(1, SMALL), b=(1, SMALL)) + ship(c1, vault, Recorder(c1), shard_count_cfg=4) + # append a NEW big file (new solo); shards untouched -> shard_ratio 0. + c2 = cat(big1=(1, BIG), a=(1, SMALL), b=(1, SMALL), big2=(2, BIG)) + s2 = ship(c2, vault, Recorder(c2), shard_count_cfg=4) + self.assertEqual(s2["reship_watch"]["consecutive_trips"], 0) + self.assertFalse(s2["reship_watch"]["sustained"]) + + def test_repeated_shard_churn_sustains(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + badge = None + # baseline + 3 runs that bump every shard member's mtime -> every shard + # re-ships each run (ratio 1.0) -> the sustained streak the alerter wants. + for run in (1, 2, 3, 4): + c = cat(a=(run, SMALL), b=(run, SMALL), e=(run, SMALL), f=(run, SMALL)) + s = ship(c, vault, Recorder(c), shard_count_cfg=4) + badge = s["badge"] + rw = archive.load_manifest(vault, badge)["reship_watch"] + self.assertGreaterEqual(rw["consecutive_trips"], 3) + self.assertTrue(rw["sustained"]) + self.assertIsNotNone(rw["sustained_since"]) + self.assertFalse(rw["alerted"]) # engine leaves alerting to the deployment glue + + if __name__ == "__main__": unittest.main() From 056624c473060f44a1f12586229bc889266d44d3 Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Thu, 9 Jul 2026 22:58:35 -0400 Subject: [PATCH 3/3] feat(routing): preserve alerter-owned alerted_at in reship-watch (re-nag cadence) The durable watch carries two alerter-owned latch fields, alerted + alerted_at (when the deployment alerter last @mentioned). The engine treats both as opaque: preserve across tripped runs, clear on streak reset. This lets the alerter re-nag on a low cadence while a watch stays sustained without the timer being erased by an intervening routed run. Suite 257 -> 258. Co-Authored-By: Claude Opus 4.8 --- archive.py | 11 ++++++++--- tests/test_reship_monitor.py | 13 +++++++++++++ 2 files changed, 21 insertions(+), 3 deletions(-) diff --git a/archive.py b/archive.py index 29294ea..b2f355e 100644 --- a/archive.py +++ b/archive.py @@ -1148,9 +1148,11 @@ def update_reship_watch(prior_watch, stats, run_epoch, - `sustained_since`: run_epoch at which the streak first reached `sustain`, preserved while it holds (lets the alerter re-nag on a low cadence); None while not sustained. - - `alerted`: an idempotency latch OWNED BY THE DEPLOYMENT ALERTER, never set - here. This function only PRESERVES it across tripped runs and CLEARS it - (-> False) whenever the streak resets, so the next crossing re-alerts once. + - `alerted` / `alerted_at`: an idempotency latch OWNED BY THE DEPLOYMENT + ALERTER (never set here — `alerted_at` is when it last @mentioned, for the + re-nag cadence). This function treats them as opaque: it PRESERVES them + across tripped runs and CLEARS them whenever the streak resets, so the next + crossing re-alerts once. """ prior = prior_watch or {} tripped = bool(stats.get("watch")) @@ -1158,10 +1160,12 @@ def update_reship_watch(prior_watch, stats, run_epoch, sustained = trips >= sustain if tripped: alerted = bool(prior.get("alerted")) + alerted_at = prior.get("alerted_at") sustained_since = (prior.get("sustained_since") or run_epoch) if sustained else None else: # Streak broke: drop the alert latch so a fresh crossing alerts again. alerted = False + alerted_at = None sustained_since = None return { "consecutive_trips": trips, @@ -1173,6 +1177,7 @@ def update_reship_watch(prior_watch, stats, run_epoch, "shards_shipped": int(stats.get("shards_shipped") or 0), "sustained_since": sustained_since, "alerted": alerted, + "alerted_at": alerted_at, } diff --git a/tests/test_reship_monitor.py b/tests/test_reship_monitor.py index 646618b..b9d0547 100644 --- a/tests/test_reship_monitor.py +++ b/tests/test_reship_monitor.py @@ -146,6 +146,19 @@ def test_alerted_cleared_on_reset_so_recrossing_realerts(self): self.assertFalse(w["alerted"]) self.assertEqual(w["consecutive_trips"], 0) + def test_alerted_at_preserved_across_trips_and_cleared_on_reset(self): + # alerted_at is alerter-owned (re-nag cadence); the engine must carry it + # across tripped runs and drop it when the streak resets. + w = None + for run in (1, 2, 3): + w = archive.update_reship_watch(w, stats_watch(True), run) + w["alerted"] = True + w["alerted_at"] = 12345 # deployment alerter stamps its send + w2 = archive.update_reship_watch(w, stats_watch(True), 4) + self.assertEqual(w2["alerted_at"], 12345) # preserved -> re-nag timer intact + w3 = archive.update_reship_watch(w2, stats_watch(False, 0.0, 4, 0), 5) + self.assertIsNone(w3["alerted_at"]) # cleared on reset + def test_custom_sustain_runs(self): w = archive.update_reship_watch(None, stats_watch(True), 1, sustain=1) self.assertTrue(w["sustained"])