feat(routing): Phase-2 reship-monitor + durable Phase-3 trigger (slice 4)

archive.py

@@ -1046,17 +1046,138 @@ 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,
+    }
+
+
+# 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"]))
+
+
+# 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` / `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"))
+    trips = (int(prior.get("consecutive_trips") or 0) + 1) if tripped else 0
+    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,
+        "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,
+        "alerted_at": alerted_at,
     }
 
 
@@ -2722,18 +2843,47 @@ 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. 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"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"]
         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"
+                   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")
-        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

tests/test_reship_monitor.py

@@ -0,0 +1,185 @@
+"""
+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)
+
+
+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_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"])
+        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()