Snapshot of covenant main (full git history on GitHub)

.gitignore

@@ -274,6 +274,8 @@ ops/distill_rejected_live.jsonl
 ops/heal.jsonl
 ops/immune.jsonl
 ops/own_work_state.json
+# laptop auto-push pass log (tools/repo_autopush.py): a local operational record, not corpus state
+ops/repo_autopush.jsonl
 # grid guard runtime state (config files grid_signers/interlocks/policy ARE tracked)
 ops/grid_commands.jsonl
 ops/grid_commands.chain.jsonl
@@ -285,6 +287,13 @@ ops/pause/
 # Node A's off-box peers (the phone's tailnet address), read by covenant_watchdog.py
 # and covenant_prod.bat alike, so neither public file carries the address.
 ops/local_peers.txt
+# the AUTO_GATED auto-onboard lane's LOCAL state (covenant_onboard.py): the operator's
+# arm/trust-root switch and the tasks the lane has armed. Both machine-local decisions,
+# not repo content -- the lane ships DARK (no arm file) and writes armed_tasks only once
+# armed. NODE_ROSTER.json and node_refusals.json ARE tracked (the whois trust root needs
+# the roster committed; a refusal only tightens).
+ops/AUTO_ONBOARD_ARMED
+ops/armed_tasks.json
 # the laptop site auto-deploy job's per-pass record (tools/site_autodeploy.py), machine-local
 ops/site_autodeploy.jsonl
 ops/local_unison_feed.txt
@@ -435,6 +444,8 @@ ops/model_asks.jsonl
 ops/model_asks.jsonl.prev
 # A294 (2026-10-07): the threefold witness's readings and its first-seen gaps
 ops/threefold_witness.json
+# A405 (2026-11-02): the highway's per-copy mirror-lag cycle counts (first-seen lag, cycles behind the unison target)
+ops/unison_lag_witness.json
 # A375 (2026-10-27): the spiking witness's negatives-accrual pipeline -- runtime state, accrues over months
 ops/witness_state.json
 ops/witness_negatives.jsonl

covenant_failover.py

@@ -84,6 +84,7 @@ SPOF_REMIND_MARKER = os.path.join(HERE, "ops", "REDUNDANCY_SPOF_REMIND")
 BEACON = os.path.join(HERE, "covenant_beacon.py")
 BEACON_WIN_PORT = 8771
 BEACON_WSL_PORT = 8772
+BEACON_PHONE_PORT = 8773
 GITHOST_STANDBY = os.path.join(HERE, "covenant_githost_standby.py")
 GITHOST_STANDBY_PORT = 8089
 
@@ -188,6 +189,9 @@ def _probe_instance(inst, probe=None):
     if p.get("type") == "feed":
         url = _feed_url()
         return _http_probe(url) if url else None      # no private feed URL configured here -> UNKNOWN
+    if p.get("type") == "phone":
+        url = _phone_beacon_url()
+        return _http_probe(url) if url else None      # no phone address configured here -> UNKNOWN (rule 9)
     return None                           # cannot probe here -> UNKNOWN for this instance
 
 
@@ -205,6 +209,22 @@ def _feed_url():
         return None
 
 
+def _phone_beacon_url():
+    """The phone beacon's real URL -- env COVENANT_PHONE_BEACON, else the first line of the
+    gitignored ops/local_phone_beacon.txt (the phone's tailnet address lives in a private file,
+    never in this public repo, exactly like the unison feed). None where neither is set, so an
+    intermittent phone that is asleep -- or has never been provisioned -- reads UNKNOWN here,
+    never 'down' (rule 9): its absence must not masquerade as a failure OR as redundancy."""
+    url = os.environ.get("COVENANT_PHONE_BEACON", "").strip()
+    if url:
+        return url
+    try:
+        with open(os.path.join(HERE, "ops", "local_phone_beacon.txt"), encoding="utf-8") as fh:
+            return fh.readline().strip() or None
+    except OSError:
+        return None
+
+
 def _inst_id(inst):
     return "%s/%s@%s" % (inst.get("name") or "?", inst.get("env") or "?", inst.get("host") or "?")
 
@@ -278,7 +298,7 @@ SERVICES = {
     # The one service this module actually provisions as a live cross-OS pair.
     # DARK until ops/REDUNDANCY_ARMED exists (RAM budget).
     "beacon": {
-        "desc": "the redundancy read-API itself -- a provisioned cross-OS pair",
+        "desc": "the redundancy read-API itself -- a provisioned cross-OS pair (+ an intermittent phone third env)",
         "require_redundant": True, "min_environments": 2, "redundancy_model": "hot",
         "failover_class": AUTO_REVERSIBLE, "arm_gated": True,
         "instances": [
@@ -288,6 +308,19 @@ SERVICES = {
             {"name": "beacon-wsl", "kind": "redundant", "env": "wsl-ubuntu", "host": "laptop",
              "provisioned": True, "hot": True, "ram_mb": 15, "launch": "wsl-ubuntu", "port": BEACON_WSL_PORT,
              "probe": {"type": "http", "url": "http://127.0.0.1:%d/beacon" % BEACON_WSL_PORT}},
+            # A THIRD environment, on the phone's own device (Galaxy S25+) -- survives the loss of BOTH
+            # laptop OSes (the operator, 2026-11-03: "make redundancies on the phone too"). Witness-class:
+            # the phone is INTERMITTENT (it sleeps) and is provisioned ONLY by an APP INSTALL on the phone,
+            # never started from the laptop (nothing pushes to the phone -- launch "phone" is declined).
+            # provisioned=False until the phone node actually runs the beacon AND its tailnet URL is in
+            # COVENANT_PHONE_BEACON / ops/local_phone_beacon.txt; until then it is DECLARED, not counted
+            # (so it is neither a false SPOF nor a phantom standby). It rides on the headscale join
+            # (phone-app-join). min_environments stays 2 -- the two laptop OSes carry "healthy"; the phone
+            # is a bonus third env when awake, and when it is provisioned+awake a laptop env dying while the
+            # phone is asleep correctly reads degraded.
+            {"name": "beacon-phone", "kind": "redundant", "env": "phone", "host": "phone",
+             "provisioned": False, "hot": True, "ram_mb": 15, "intermittent": True,
+             "launch": "phone", "port": BEACON_PHONE_PORT, "probe": {"type": "phone"}},
         ],
     },
 }
@@ -530,6 +563,9 @@ def _start_instance(inst, dry_run):
         return _launch_windows_beacon(inst, dry_run)
     if launch == "wsl-ubuntu":
         return _launch_wsl_beacon(inst, dry_run)
+    if launch == "phone":
+        return False, ("the phone beacon is provisioned by an APP INSTALL on the phone and never started "
+                       "from the laptop (nothing pushes to the phone) -- coordinate with the phone node")
     return False, "no known launcher for %s (provision it first -- PROPOSE_ONLY)" % inst.get("name")
 
 

covenant_highway.py

@@ -2378,6 +2378,373 @@ def detect_git_host_down(health=None, url=None, marker=None, **kwargs):
             "url": url, "error": "%s: %s" % (type(e).__name__, str(e)[:120])}}
 
 
+# ----------------------------------------------------------------- the sync fabric (A405, 2026-11-02)
+#
+# His words, 2026-11-02: "continue growing the mycelial highway. Make it pave itself." The highway already
+# self-paves what it has always owned; today a new SYNC FABRIC was built around it and nothing in this file
+# could see it. The fabric is: a UNISON FEED served on TETSU (:8800/unison), one reading every `interval_s`,
+# that says for each repo whether its github / hf / website copies have reached a common `target`; and two
+# operator-armed scheduled tasks that act on that feed -- CovenantHFMirror (repo history -> Hugging Face,
+# fast-forward-only, opsec-gated, visibility-preserving) and CovenantSiteAutoDeploy (GitHub -> the website's
+# repo mirror, opsec-gated, never a half-sync). A GitHub pusher, CovenantRepoAutopush (open-PR / opsec / ff
+# gated), is being built in a sibling session and may not be armed yet.
+#
+# These detectors let the loop SENSE that fabric, and -- only in the one narrow case that is strictly
+# reversible: re-running an already-armed, guarded, idempotent task -- reversibly HEAL it. The line the
+# operator is holding does NOT move here. Nothing auto-ONBOARDS, auto-INSTALLS or auto-ARMS a node or a task
+# (arming a task is config, PROPOSE_ONLY by construction). A public GitHub push is NEVER AUTO_REVERSIBLE here;
+# it belongs to repo_autopush, itself gated. Keys, private holdings and the portfolio never ride the fabric
+# and are not read by anything below.
+#
+# SAFE WHEN THE LOOP RUNS DRY (A244, highway_may_act). The detectors only READ (a feed GET, a scheduler read);
+# reading is safe in a staged/worktree copy. The remedies execute nothing when dry_run is True (the
+# remedy_restart_nodes pattern), so a staged pass that senses the live fabric still touches nothing.
+#
+# WHERE THE FEED'S ADDRESS LIVES. His network addresses never go in this public repo (the rule ops/local_peers.txt
+# and tools/site_autodeploy.py already follow). The feed URL is read from COVENANT_UNISON_FEED or the first line
+# of the gitignored ops/local_unison_feed.txt; with neither set the feed is "not configured here" and the feed
+# detectors read UNKNOWN -- never OK (rule 9, P20).
+
+UNISON_FEED_LOCAL = os.path.join(HERE, "ops", "local_unison_feed.txt")
+UNISON_FEED_CACHE_S = 120           # the three feed detectors in one sense() pass share one GET; the feed moves every interval_s
+UNISON_COPIES = ("github", "hf", "website")
+# The tasks that act on the fabric, and the copy each one keeps in step (so a lagging copy is re-synced by the
+# task that owns it). RepoAutopush owns the github copy and is OPTIONAL -- checked only if it is armed.
+SYNC_TASKS_EXPECTED = ("CovenantHFMirror", "CovenantSiteAutoDeploy")
+SYNC_TASKS_OPTIONAL = ("CovenantRepoAutopush",)
+COPY_TASK = {"hf": "CovenantHFMirror", "website": "CovenantSiteAutoDeploy", "github": "CovenantRepoAutopush"}
+STALE_RUN_INTERVALS = 3             # a task whose last run is older than this many of its OWN intervals is not firing
+MIRROR_LAG_CYCLES = 3               # a copy behind the target across more than this many DISTINCT feed readings
+UNISON_LAG_CACHE = os.path.join(HERE, "ops", "unison_lag_witness.json")   # gitignored: first-seen lag and cycle counts
+# LastTaskResult codes that are NOT a failed run: 0 is success; 0x4130x are Windows task-STATE codes (ready,
+# running, not-yet-run, terminated, ...) that a scheduler can report in place of an exit code. Treating them as
+# non-failures keeps a task mid-run or freshly armed from reading as a failure (a false positive that would
+# re-run a task that is already running).
+_TASK_OK_RESULTS = frozenset({0, 0x41300, 0x41301, 0x41302, 0x41303, 0x41304, 0x41305, 0x41306})
+
+_sync_task_cache = {"at": 0.0, "tasks": None}
+_unison_feed_cache = {"at": 0.0, "feed": None, "note": ""}
+
+
+def _naive_local_ts(s):
+    """Epoch for a naive ISO stamp like '2026-11-02T13:38:15', read as local time (the feed writes TETSU-local)."""
+    import datetime
+    return datetime.datetime.strptime(str(s)[:19], "%Y-%m-%dT%H:%M:%S").timestamp()
+
+
+def _iso8601_dur_s(s):
+    """Seconds in an ISO-8601 duration's TIME part, e.g. 'PT2H'->7200, 'PT15M'->900, 'PT1H30M'->5400; None when
+    there is no readable time part. Discovered from a task's own trigger so a cadence the operator changes is
+    judged by the new cadence, not a number hardcoded here (rule 2)."""
+    import re as _re
+    if not s or not isinstance(s, str):
+        return None
+    m = _re.match(r"^P(?:\d+W)?(?:\d+D)?(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?)?$", s.strip())
+    if not m or not any(m.groups()):
+        return None
+    h, mi, se = (int(x) if x else 0 for x in m.groups())
+    return (h * 3600 + mi * 60 + se) or None
+
+
+def _unison_feed_url():
+    """The feed's URL -- env COVENANT_UNISON_FEED, else the first line of gitignored ops/local_unison_feed.txt,
+    else None. His addresses never sit in this public repo (the ops/local_peers.txt rule)."""
+    url = (os.environ.get("COVENANT_UNISON_FEED") or "").strip()
+    if url:
+        return url
+    try:
+        with io.open(UNISON_FEED_LOCAL, encoding="utf-8") as fh:
+            return fh.readline().strip() or None
+    except OSError:
+        return None
+
+
+def _unison_feed(url=None, get=None):
+    """(feed|None, note). `get`, when given, is a callable(url)->dict the suite uses in place of the network.
+    note is '' on a clean read, else WHY it could not be read; the 'no feed URL configured' note is how the
+    caller tells 'unconfigured here' (UNKNOWN) apart from 'configured and down' (a finding)."""
+    if get is not None:
+        try:
+            return get(url), ""
+        except Exception as e:                                    # noqa: BLE001
+            return None, "%s: %s" % (type(e).__name__, str(e)[:80])
+    u = url or _unison_feed_url()
+    if not u:
+        return None, "no feed URL configured here (COVENANT_UNISON_FEED / ops/local_unison_feed.txt)"
+    try:
+        return _get(u, timeout=10), ""
+    except Exception as e:                                        # noqa: BLE001
+        return None, "%s: %s" % (type(e).__name__, str(e)[:80])
+
+
+def _cached_feed():
+    """The feed, cached for UNISON_FEED_CACHE_S so the three feed detectors in one pass make one GET, not three."""
+    now = time.time()
+    if _unison_feed_cache["feed"] is None or now - _unison_feed_cache["at"] > UNISON_FEED_CACHE_S:
+        feed, note = _unison_feed()
+        _unison_feed_cache.update(at=now, feed=feed, note=note)
+    return _unison_feed_cache["feed"], _unison_feed_cache["note"]
+
+
+def _resolve_feed(feed, url, get):
+    """A detector's feed: an explicit `feed` (the suite), else a `url`/`get` read (the suite), else the cache."""
+    if feed is not None:
+        return feed, ""
+    if url is not None or get is not None:
+        return _unison_feed(url, get)
+    return _cached_feed()
+
+
+def detect_unison_feed_stale(health=None, feed=None, url=None, get=None, now=None):
+    """The unison feed -- the one reading the whole sync fabric agrees on -- is unreachable, not OK, or old.
+
+    THE GAP THIS CLOSES (2026-11-02). CovenantHFMirror and CovenantSiteAutoDeploy act ONLY on what the feed says
+    is due; site_autodeploy 'never deploys blind' and does nothing when the feed is unready. So a feed that has
+    stopped answering, or is stuck on an old reading, silently freezes the whole fabric: every downstream task
+    reads 'nothing due' and the road reads quiet. This asks the one question none of the others can -- is the
+    feed itself alive and fresh?
+
+    PRESENT when the feed is unreachable, its `status` is not 'ok', or its own `checked` stamp is older than
+    ~2x its `interval_s` (a reading it owed and did not refresh). UNKNOWN -- never OK -- when no feed URL is
+    configured here or the read failed in a way that is the reader's blindness, not the feed's fault.
+
+    NO AUTO REMEDY. The feed runs on TETSU; nothing on this machine can restart it, and reaching across to his
+    machine to restart a service is not reversible from here. The remedy is PROPOSE_ONLY (propose_unison_feed):
+    the condition, the covenant's reading, and the cost beside the gain, for a person to act on."""
+    now = time.time() if now is None else now
+    feed, note = _resolve_feed(feed, url, get)
+    if feed is None:
+        # Unconfigured here is the reader's blindness (UNKNOWN); a configured feed that will not answer is the
+        # finding (PRESENT). The two are told apart by whether a URL was set.
+        if "no feed URL configured" in note:
+            return {"state": UNKNOWN, "measured": {"why": note}}
+        return {"state": PRESENT, "measured": {"reachable": False, "why": note or "unreachable"}}
+    if not isinstance(feed, dict):
+        return {"state": PRESENT, "measured": {"reachable": True, "why": "feed is not a JSON object"}}
+    status, interval, checked = feed.get("status"), feed.get("interval_s"), feed.get("checked")
+    age_s = None
+    try:
+        if checked:
+            age_s = now - _naive_local_ts(checked)
+    except Exception:                                            # noqa: BLE001
+        age_s = None
+    stale_after = (2.0 * float(interval)) if isinstance(interval, (int, float)) and interval else None
+    reasons = []
+    if status != "ok":
+        reasons.append("status is %r, not 'ok'" % status)
+    if stale_after is not None and age_s is not None and age_s > stale_after:
+        reasons.append("last checked %.0fs ago, over 2x interval (%.0fs)" % (age_s, stale_after))
+    measured = {"reachable": True, "status": status, "interval_s": interval, "checked": checked,
+                "age_s": None if age_s is None else round(age_s, 1), "reasons": reasons}
+    if stale_after is None:
+        measured["note"] = "no usable interval_s on the feed; staleness not judged, status only"
+    return {"state": PRESENT if reasons else ABSENT, "measured": measured}
+
+
+def _read_sync_tasks(names, run=None):
+    """{name: {present, state, last_run(epoch|None), result(int|None), interval_s(|None), next_run(epoch|None)}} for
+    each requested task. None (the whole dict) when the scheduler cannot be read here (not Windows / read failed).
+    `run` is the seam the suite replaces: a callable()->the parsed JSON (list|dict|None), so no PowerShell runs
+    in a test."""
+    if run is None and os.name != "nt":
+        return None
+    if run is not None:
+        data = run()
+    else:
+        quoted = ",".join("'%s'" % n.replace("'", "''") for n in names)
+        ps = ("Get-ScheduledTask -TaskName %s -ErrorAction SilentlyContinue | ForEach-Object { "
+              "$i = $_ | Get-ScheduledTaskInfo; [pscustomobject]@{name=$_.TaskName; state=[string]$_.State; "
+              "last=$(if ($i.LastRunTime -and $i.LastRunTime.Year -gt 2000) {([DateTimeOffset]$i.LastRunTime).ToUnixTimeSeconds()} else {$null}); "
+              "result=$i.LastTaskResult; "
+              "next=$(if ($i.NextRunTime) {([DateTimeOffset]$i.NextRunTime).ToUnixTimeSeconds()} else {$null}); "
+              "interval=[string]($_.Triggers | ForEach-Object { $_.Repetition.Interval } | Where-Object { $_ } | Select-Object -First 1)} } "
+              "| ConvertTo-Json -Compress") % quoted
+        import subprocess
+        try:
+            r = subprocess.run(["powershell", "-NoProfile", "-Command", ps], capture_output=True, text=True,
+                               timeout=60, creationflags=_NOWIN)
+            data = json.loads(r.stdout or "null")
+        except Exception:                                        # noqa: BLE001
+            return None
+    data = [] if data is None else (data if isinstance(data, list) else [data])
+    out = {}
+    for d in data:
+        nm = d.get("name")
+        if not nm:
+            continue
+        out[nm] = {"present": True, "state": d.get("state"),
+                   "last_run": (float(d["last"]) if d.get("last") is not None else None),
+                   "result": (int(d["result"]) if d.get("result") is not None else None),
+                   "interval_s": _iso8601_dur_s(d.get("interval")),
+                   "next_run": (float(d["next"]) if d.get("next") is not None else None)}
+    for nm in names:
+        out.setdefault(nm, {"present": False})
+    return out
+
+
+def _cached_sync_tasks():
+    """The sync tasks' scheduler state, cached for SCHEDULE_EVERY_S (as detect_schedule_stalled caches its read)."""
+    now = time.time()
+    if _sync_task_cache["tasks"] is None or now - _sync_task_cache["at"] > SCHEDULE_EVERY_S:
+        _sync_task_cache.update(at=now, tasks=_read_sync_tasks(SYNC_TASKS_EXPECTED + SYNC_TASKS_OPTIONAL))
+    return _sync_task_cache["tasks"]
+
+
+def detect_sync_task_not_firing(health=None, tasks=None, now=None):
+    """An armed sync task that is present and NOT doing its job: its last run FAILED (a nonzero exit that is not a
+    task-state code), or it has not run in ~3x its OWN interval. The one condition the loop may reversibly heal
+    here -- by re-running the same already-armed, guarded, idempotent task (rerun_sync_task, AUTO_REVERSIBLE).
+
+    SEPARATE FROM detect_sync_task_missing, the split detect_build_stale_on_pc was carved out for: a remedy must
+    be graded against the condition it can CLEAR. Re-running a task cannot conjure a task that is not armed;
+    pairing the re-run with 'a task is missing' would grade it 'did not fix' every pass a task stayed unarmed and
+    quarantine the one remedy that heals the tasks that DO exist. Missing-ness is a config gap, the operator's
+    call (PROPOSE_ONLY); detect_sync_task_missing carries it, with no remedy of its own.
+
+    SEPARATE FROM detect_schedule_stalled too: that reads next-run/missed for EVERY Covenant* task and never
+    reads the LAST RESULT, so a task that fires on time and exits nonzero every time reads healthy to it. This
+    reads the exit code, for the sync tasks only, and pairs a remedy. A disabled task is a person's choice and is
+    skipped (its rule). A never-run task with a known interval is not-firing; with no readable interval it is
+    judged on its exit code alone."""
+    now = time.time() if now is None else now
+    tasks = _cached_sync_tasks() if tasks is None else tasks
+    if tasks is None:
+        return {"state": UNKNOWN, "measured": {"why": "the scheduler could not be read here (not Windows, or the read failed)"}}
+    not_firing, why = [], {}
+    for name in SYNC_TASKS_EXPECTED + SYNC_TASKS_OPTIONAL:
+        t = tasks.get(name) or {}
+        if not t.get("present") or str(t.get("state")) == "Disabled":
+            continue
+        res = t.get("result")
+        reason = None
+        if res is not None and res not in _TASK_OK_RESULTS:
+            reason = "last run exited %s (0x%X)" % (res, res & 0xFFFFFFFF)
+        else:
+            interval, last = t.get("interval_s"), t.get("last_run")
+            if interval:
+                if last is None:
+                    reason = "has a %.0fs interval but has never run" % interval
+                elif now - last > STALE_RUN_INTERVALS * interval:
+                    reason = "last ran %.0fs ago, over %dx its %.0fs interval" % (now - last, STALE_RUN_INTERVALS, interval)
+        if reason:
+            not_firing.append(name)
+            why[name] = reason
+    return {"state": PRESENT if not_firing else ABSENT,
+            "measured": {"checked": list(SYNC_TASKS_EXPECTED + SYNC_TASKS_OPTIONAL), "not_firing": not_firing, "why": why}}
+
+
+def detect_sync_task_missing(health=None, tasks=None):
+    """An armed sync task that should be present is gone entirely. ARMING IS CONFIG, so there is NO remedy here:
+    re-arming a task is PROPOSE_ONLY by construction (the operator is holding that decision), and
+    detect_schedule_stalled cannot see an ABSENT task -- it grades only tasks the scheduler still lists. Only the
+    two tasks armed as part of the fabric are expected; CovenantRepoAutopush is optional (it may not be armed yet)
+    and its absence is never flagged."""
+    tasks = _read_sync_tasks(SYNC_TASKS_EXPECTED) if tasks is None else tasks
+    if tasks is None:
+        return {"state": UNKNOWN, "measured": {"why": "the scheduler could not be read here (not Windows, or the read failed)"}}
+    missing = [n for n in SYNC_TASKS_EXPECTED if not (tasks.get(n) or {}).get("present")]
+    return {"state": PRESENT if missing else ABSENT,
+            "measured": {"expected": list(SYNC_TASKS_EXPECTED), "missing": missing,
+                         "note": "re-arming a scheduled task is config -- a person's call, never auto (A244 line held)"}}
+
+
+def _load_lag_cache(path=None):
+    try:
+        with io.open(path or UNISON_LAG_CACHE, encoding="utf-8") as fh:
+            d = json.load(fh)
+        return d if isinstance(d, dict) else {}
+    except (OSError, ValueError):
+        return {}
+
+
+def _save_lag_cache(d, path=None):
+    p = path or UNISON_LAG_CACHE
+    try:
+        os.makedirs(os.path.dirname(p), exist_ok=True)
+        with io.open(p, "w", encoding="utf-8") as fh:
+            json.dump(d, fh, indent=1)
+    except OSError:
+        pass
+
+
+def _mirror_lag_cycles(feed, now=None, cache_path=None, save=True):
+    """Update and return the per-(repo/copy) lag cycle counts from the feed. A 'cycle' is one DISTINCT feed
+    reading (`checked`) during which a copy's head != the repo's target. The highway passes many times per feed
+    reading; only a NEW `checked` advances a count, so N cycles is ~N feed intervals behind, not N highway passes
+    (the A20/M34 trap: an alert whose threshold is really 'the loop ran again' goes off for nothing). Returns
+    {"<repo>/<copy>": {cycles, since, last_checked, target}} for every copy currently behind; copies now caught up
+    are dropped, and a changed target restarts the count (a fresh lag instance). Writes the cache unless save=False
+    (the suite). Within one feed reading the update is idempotent, so the two lag detectors calling it in one pass
+    cannot double-count."""
+    cache = _load_lag_cache(cache_path)
+    cur = str((feed or {}).get("checked") or "")
+    repos = (feed or {}).get("repos") or {}
+    if isinstance(repos, list):
+        repos = {r.get("name"): r for r in repos if isinstance(r, dict) and r.get("name")}
+    behind_now = {}
+    for name, entry in (repos.items() if isinstance(repos, dict) else []):
+        if not isinstance(entry, dict):
+            continue
+        target = str(entry.get("target") or "")
+        heads = entry.get("heads") or {}
+        if not target:
+            continue
+        for copy in UNISON_COPIES:
+            head = str(heads.get(copy) or "")
+            if not head or head == target:
+                continue
+            key = "%s/%s" % (name, copy)
+            prev = cache.get(key) or {}
+            if not prev or prev.get("target") != target:
+                behind_now[key] = {"cycles": 1, "since": cur, "last_checked": cur, "target": target}
+            elif cur and prev.get("last_checked") != cur:
+                behind_now[key] = {"cycles": int(prev.get("cycles", 0)) + 1, "since": prev.get("since", cur),
+                                   "last_checked": cur, "target": target}
+            else:
+                behind_now[key] = dict(prev)      # same feed reading (or no `checked`) -> no new cycle
+    if save:
+        _save_lag_cache(behind_now, cache_path)
+    return behind_now
+
+
+def detect_mirror_lag(health=None, feed=None, url=None, get=None, now=None, cache_path=None, save=True):
+    """A mirror copy the loop CAN reversibly re-sync -- Hugging Face or the website -- has stayed behind the unison
+    target across more than MIRROR_LAG_CYCLES distinct feed readings: the task that owns it is firing, yet the copy
+    is not catching up. Remedy AUTO_REVERSIBLE (trigger_mirror_sync): re-run the guarded, idempotent task that owns
+    the lagging copy (CovenantHFMirror for hf, CovenantSiteAutoDeploy for the website). The GitHub copy is NOT here:
+    a public push belongs to repo_autopush and is PROPOSE_ONLY (detect_github_push_lag).
+
+    PERSISTENCE, NOT A BLIP. A copy one reading behind is a mirror mid-catch-up, not a fault; only a copy behind
+    across >N readings -- the owning task has had N chances and the copy still trails -- is the finding. UNKNOWN
+    (never OK) when the feed is not ready, so a feed outage reads as unison_feed_stale's finding, not as 'no lag'."""
+    feed, note = _resolve_feed(feed, url, get)
+    if not isinstance(feed, dict) or feed.get("status") != "ok":
+        return {"state": UNKNOWN, "measured": {"why": note or "feed not ready (no status ok)"}}
+    lag = _mirror_lag_cycles(feed, now=now, cache_path=cache_path, save=save)
+    laggards = {k: v for k, v in lag.items()
+                if k.rsplit("/", 1)[-1] in ("hf", "website") and int(v.get("cycles", 0)) > MIRROR_LAG_CYCLES}
+    rerun = sorted({COPY_TASK[k.rsplit("/", 1)[-1]] for k in laggards})
+    return {"state": PRESENT if laggards else ABSENT,
+            "measured": {"over_%d_cycles" % MIRROR_LAG_CYCLES: {k: v.get("cycles") for k, v in laggards.items()},
+                         "rerun_tasks": rerun, "cycles_needed": MIRROR_LAG_CYCLES}}
+
+
+def detect_github_push_lag(health=None, feed=None, url=None, get=None, now=None, cache_path=None, save=True):
+    """The GitHub copy of a repo has stayed behind the unison target across more than MIRROR_LAG_CYCLES feed
+    readings. NO AUTO REMEDY HERE: pushing to a public GitHub repo is never AUTO_REVERSIBLE. It belongs to the
+    repo_autopush pusher (CovenantRepoAutopush), itself gated on open-PRs, opsec and fast-forward, and whose
+    arming is the operator's call. This SENSES the lag and names who owns the push; it never performs one."""
+    feed, note = _resolve_feed(feed, url, get)
+    if not isinstance(feed, dict) or feed.get("status") != "ok":
+        return {"state": UNKNOWN, "measured": {"why": note or "feed not ready (no status ok)"}}
+    lag = _mirror_lag_cycles(feed, now=now, cache_path=cache_path, save=save)
+    behind = {k: int(v.get("cycles", 0)) for k, v in lag.items()
+              if k.rsplit("/", 1)[-1] == "github" and int(v.get("cycles", 0)) > MIRROR_LAG_CYCLES}
+    return {"state": PRESENT if behind else ABSENT,
+            "measured": {"github_behind_over_%d_cycles" % MIRROR_LAG_CYCLES: behind,
+                         "owner": "repo_autopush (CovenantRepoAutopush) -- gated public push, PROPOSE_ONLY; never run here"}}
+
+
 DETECTORS = {
     "node_down": detect_node_down,
     "tetsu_cannot_answer": detect_tetsu_cannot_answer,
@@ -2401,6 +2768,14 @@ DETECTORS = {
     "sweep_not_current": detect_sweep_not_current,
     "ambassador_stalled": detect_ambassador_stalled,
     "schedule_stalled": detect_schedule_stalled,
+    # The sync fabric (A405, 2026-11-02): the unison feed and the operator-armed tasks that keep the repo
+    # mirrors in step. Feed-stale and github-lag propose only; a present-but-stalled task and HF/website lag
+    # are reversibly healed by re-running the already-armed, guarded, idempotent task that owns them.
+    "unison_feed_stale": detect_unison_feed_stale,
+    "sync_task_not_firing": detect_sync_task_not_firing,
+    "sync_task_missing": detect_sync_task_missing,
+    "mirror_lag": detect_mirror_lag,
+    "github_push_lag": detect_github_push_lag,
     "moltbook_strikes": detect_moltbook_strikes,
     "threefold_witness": detect_threefold_witness,
     # "public_email_exposure" RETIRED the day it was added (A306), his words 2026-10-07: "they aren't issues idgaf".
@@ -3051,7 +3426,86 @@ def remedy_refresh_defender(measured, dry_run=True):
                                                   scanned, "; ".join(mendable))
 
 
+def _run_scheduled_task(name):
+    """(ok, detail) -- ask Windows to run an already-armed task NOW (schtasks /run). This starts the SAME guarded,
+    idempotent run the operator's schedule already makes on its timer; it changes no task config and no file here."""
+    import subprocess
+    try:
+        p = subprocess.run(["schtasks", "/run", "/tn", name], capture_output=True, text=True, timeout=60,
+                           creationflags=_NOWIN)
+        return p.returncode == 0, ((p.stdout or "") + (p.stderr or "")).strip()[:200]
+    except Exception as e:                                        # noqa: BLE001
+        return False, "%s: %s" % (type(e).__name__, e)
+
+
+def _rerun_named(names, dry_run):
+    """(ok, detail) for re-running a list of armed sync tasks. ok=False ('held', never counted as a failure) when
+    there is nothing to re-run or every /run was refused; ok=True when at least one started."""
+    names = list(names or [])
+    if not names:
+        return False, "no armed sync task to re-run"
+    if dry_run:
+        return True, "would re-run: " + ", ".join(names)
+    ran, failed = [], []
+    for n in names:
+        ok, detail = _run_scheduled_task(n)
+        (ran if ok else failed).append(n if ok else "%s (%s)" % (n, detail))
+    if not ran:
+        return False, "schtasks /run refused: " + "; ".join(failed)
+    return True, "re-ran " + ", ".join(ran) + ("; failed: " + "; ".join(failed) if failed else "")
+
+
+def remedy_rerun_sync_task(measured, dry_run=True):
+    """STATELESS: re-run the armed, guarded, idempotent sync task(s) the detector found present-and-not-firing.
+    This does only what the operator-armed schedule already does on its own timer -- it changes no config and no
+    file here, and under a DRY/staged pass (A244) it runs nothing."""
+    return _rerun_named((measured or {}).get("not_firing"), dry_run)
+
+
+def remedy_trigger_mirror_sync(measured, dry_run=True):
+    """STATELESS: re-run the guarded, idempotent task that owns a lagging HF/website mirror copy -- the same run
+    its schedule makes. A public GitHub push is never here; that is repo_autopush's, and PROPOSE_ONLY."""
+    return _rerun_named((measured or {}).get("rerun_tasks"), dry_run)
+
+
+def remedy_propose_unison_feed(measured, dry_run=True):
+    """PROPOSE_ONLY, never executed by the engine. The feed runs on TETSU; this machine cannot restart it, and
+    reaching across to his machine to restart a service is not reversible from here. apply_remedy refuses the
+    class and turns this into a proposal carrying the covenant's reading and the cost beside the gain."""
+    return False, "the unison feed runs on TETSU; a person restarts or unsticks it there"
+
+
 REMEDIES = {
+    "rerun_sync_task": {"fn": remedy_rerun_sync_task, "klass": AUTO_REVERSIBLE,
+                        "for": ["sync_task_not_firing"], "kind": "stateless",
+                        # Async, and graded by a later pass: schtasks /run returns at once while the task runs on,
+                        # and a run in flight reports result 0x41301 ('currently running'), which a synchronous
+                        # re-sense would read as still-not-firing and quarantine a remedy that works -- the exact
+                        # fetch_build / restart_nodes trap. The day's budget keeps a persistently-failing task
+                        # from being hammered every pass; a task the re-run cannot revive is graded 'did not fix'
+                        # and quarantined for a person, which is correct.
+                        "async_for": ["sync_task_not_firing"], "grade_after_s": 900.0, "cooldown_s": 1800,
+                        "touches": ["an already-armed, guarded sync task's next run (re-run only; the schedule and its config are untouched)"],
+                        "benefit": {"gains": ["a mirror task that stalled or last failed runs again without waiting for its timer or a person",
+                                              "the sync fabric keeps moving"],
+                                    "cost": ["one extra guarded, idempotent run (the task's own ff-only / opsec gates still apply); a task that keeps failing is quarantined for a person"],
+                                    "irreversible": []}},
+    "trigger_mirror_sync": {"fn": remedy_trigger_mirror_sync, "klass": AUTO_REVERSIBLE,
+                            "for": ["mirror_lag"], "kind": "stateless",
+                            # Two feed readings to catch up before grading (the feed moves every interval_s), same
+                            # async reasoning as rerun_sync_task.
+                            "async_for": ["mirror_lag"], "grade_after_s": 1800.0, "cooldown_s": 1800,
+                            "touches": ["the guarded sync task that owns a lagging HF/website copy (re-run only; no config change, no GitHub push)"],
+                            "benefit": {"gains": ["an HF or website copy stuck behind the unison target is re-synced by its own task, without a person"],
+                                        "cost": ["one extra guarded, idempotent run; the github copy is never touched here"],
+                                        "irreversible": []}},
+    "propose_unison_feed": {"fn": remedy_propose_unison_feed, "klass": PROPOSE_ONLY,
+                            "for": ["unison_feed_stale"], "kind": "needs a person",
+                            "touches": ["the unison feed on TETSU"],
+                            "benefit": {"gains": ["the whole sync fabric stops being frozen by a stuck or dead feed",
+                                                  "downstream tasks stop reading a stale 'nothing due' as quiet"],
+                                        "cost": ["a person's attention on TETSU to restart or unstick the feed"],
+                                        "irreversible": []}},
     "refresh_defender": {"fn": remedy_refresh_defender, "klass": AUTO_REVERSIBLE,
                          "for": ["defense_lapse"], "kind": "stateless",
                          "touches": ["Defender's signature update, judged by covenant_immune.update(), and a scan of our own tree; never its settings"],
@@ -4084,6 +4538,22 @@ def run_once(dry_run=False, exclude=("restart_watchdog",), ledger=None, health=N
                 # repairs it" without that is a nudge, not a finding.
                 alerts.append("highway: %s is present and nothing here repairs it -- %s"
                               % (name, json.dumps(c["measured"])[:400]))
+    # A405 (2026-11-02, "loosen autonomy ... create its own lane"): the AUTO_GATED
+    # auto-onboard lane runs as its OWN pass in covenant_onboard.py -- a third class
+    # between AUTO_REVERSIBLE and PROPOSE_ONLY, gated on an operator-rooted roster and
+    # a verified node identity. This hook is the whole of its footprint here: it does
+    # not touch apply_remedy or the existing remedies. dry_run carries A244 (the
+    # watchdog passes dry_run=not highway_may_act(), so a staged copy runs it DRY), and
+    # a paused highway has already returned above, so the lane never acts when paused.
+    # It SHIPS DARK: no roster and not armed, so this adds one "DARK" info line and
+    # repairs nothing. Guarded -- a notice from a new lane must never fell the road.
+    try:
+        import covenant_onboard as _ob
+        ob_alerts, ob_infos = _ob.run_once(dry_run=dry_run, ledger=ledger)
+        alerts.extend(ob_alerts)
+        infos.extend(ob_infos)
+    except Exception as _e:                                      # noqa: BLE001
+        infos.append("highway: auto-onboard lane unavailable -- %s: %s" % (type(_e).__name__, _e))
     return alerts, infos
 
 

covenant_onboard.py

@@ -0,0 +1,738 @@
+#!/usr/bin/env python3
+"""covenant_onboard.py -- the highway's AUTO_GATED lane: auto-onboard a KNOWN node.
+
+ASKED 2026-11-02: the operator chose to "loosen autonomy" for the mycelial
+highway and to "create its own lane." The highway's founding line (see
+covenant_highway.py) is two classes: AUTO_REVERSIBLE may be run by the engine;
+PROPOSE_ONLY -- config, install, money, rule, a narrowing of choice -- may never
+be. Onboarding a node sat in PROPOSE_ONLY, so a device the operator already owns
+could appear on the tailnet and the highway would only ever PROPOSE peering it,
+never do it. This file adds a THIRD class between the two, and nothing else:
+
+    AUTO_GATED -- the engine MAY act, but only through a gate rooted in the
+                  operator, and only on a reversible, local action.
+
+WHY A SEPARATE MODULE, NOT A NEW BRANCH IN apply_remedy. The highway's
+apply_remedy dispatches on one axis -- is the class AUTO_REVERSIBLE? -- and its
+invariants (NEVER_AUTOMATIC, the operator's choices, stateless-or-undo) are
+written for repairs of a sensed fault. AUTO_GATED is a different shape: its
+"condition" is a roster the operator signed, its gate verifies an identity over
+the network, and its actions peer or arm rather than repair. Folding that into
+apply_remedy would widen the one function every repair flows through, on the
+same day a sibling session is editing the detectors beside it. So the lane lives
+here, with its own gate and its own ledger rows (written to the SAME highway
+ledger, klass AUTO_GATED, so one audit sees everything), and the highway's hook
+is a single call. apply_remedy is not touched; the existing autonomous execution
+is unchanged.
+
+THE GATE. The engine auto-acts ONLY when ALL of these hold (any miss -> it does
+not act; it PROPOSES or REFUSES, and says why):
+
+  1. A model's output never triggers it. Data is not authority -- the same line
+     as the Jang-woo reply, and it binds our own fleet, the uncensored 27B
+     included. The trigger's source must be a roster scan or tailnet presence,
+     never a model.
+  2. ARMED. The operator wrote ops/AUTO_ONBOARD_ARMED choosing a trust root, AND
+     that trust root is satisfiable now. Absent either, the lane is DARK: it
+     senses and proposes, it acts on nothing. It ships DARK.
+  3. A roster the operator ROOTED. Either (a) ops/NODE_ROSTER.json signed by the
+     operator's OWN key and the signature verifies here, or (b) the roster is
+     committed to the repo and every entry verifies as owned by the operator's
+     own Tailscale account (whois). Strongest present wins; neither -> dark. The
+     highway only READS and VERIFIES the roster. It never writes it.
+  4. The target is ON that roster, and the action is one the roster allows for
+     it. An unknown node is never auto-onboarded.
+  5. Identity verified, not minted. `tailscale whois <ip>` must name the
+     operator's account and the roster's Machine ID, and the node key pinned
+     from `tailscale status --json` must equal the roster's. An IP alone is
+     never trusted. Keys are only ever verified here, never created or copied.
+  6. Reversible, LOCAL, exactly undoable. Auto-peer a known node (a line in
+     ops/local_peers.txt; un-peer undoes it) and auto-arm a vetted task that
+     keeps its OWN gate (disable undoes it). Anything that publishes, pushes,
+     posts or sends is NOT reversible and stays PROPOSE_ONLY.
+  7. Still PROPOSE_ONLY, forever: installing software; creating or granting keys
+     or signing authority; the consent roster; onboarding an unknown node; any
+     firewall / power / security / system-setting change on any node, even a
+     reversible one; anything touching Tetsu's key, consent or path; any change
+     to the roster itself, removals included.
+  8. A per-node refusal is honoured. A node may declare "no system-setting
+     auto-action" and the engine respects it whatever the class. TETSU declares
+     it (ops/node_refusals.json) -- its Claude session's standing line on that
+     machine.
+  9. The operator's choices (ops/OPERATOR_CHOICES.json) and the pause switch are
+     honoured, and the lane is DRY wherever highway_may_act() is False (a staged
+     copy or a fresh clone -- A244). Every auto-action is logged AND announced
+     with its undo command, rate-limited per node.
+
+SOVEREIGNTY IS IMMOVABLE. Onboarding grants no signing or consent authority --
+that is the owner's alone (Tetsu, 10-31: "a decision for the OWNER"). The host
+must not be able to command the signer. Key_id ...dd885ef9... is Tetsu's own
+CONSENT key, not a node key; nothing here gates on it, signs with it, or touches
+it.
+
+USE
+  python covenant_onboard.py                 # what the lane sees, and whether it is armed
+  python covenant_onboard.py --plan          # the candidates and what the gate says of each (DRY)
+LICENCE: public domain.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import time
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+_NOWIN = 0x08000000 if os.name == "nt" else 0
+
+# The third class. A string, like the other two in covenant_highway.py, so a
+# ledger row and a test can name it without importing the engine.
+AUTO_GATED = "AUTO_GATED"
+
+# Files the operator owns. The highway READS these; it writes none of them.
+ROSTER = os.path.join(HERE, "ops", "NODE_ROSTER.json")       # operator-signed / operator-committed
+OPERATOR_PUBKEY = os.path.join(HERE, "ops", "operator_pubkey.pem")  # for trust root (a)
+ARMED_FILE = os.path.join(HERE, "ops", "AUTO_ONBOARD_ARMED")  # the operator's arm + trust-root choice
+# This file the highway MAY write, because a refusal only ever TIGHTENS -- it can
+# forbid an action, never permit one. Revoking is easier than granting (the spec's
+# own rule), and a per-node "do not touch my system settings" is a revocation.
+NODE_REFUSALS = os.path.join(HERE, "ops", "node_refusals.json")
+# The highway's own ledger and choices, reused so one audit sees the whole road.
+LEDGER = os.path.join(HERE, "ops", "highway.jsonl")
+CHOICES = os.path.join(HERE, "ops", "OPERATOR_CHOICES.json")
+# off-box peers, one comma-separated line of host:port -- the same file
+# covenant_watchdog._local_peers() and covenant_prod.bat read.
+LOCAL_PEERS = os.path.join(HERE, "ops", "local_peers.txt")
+ARMED_TASKS = os.path.join(HERE, "ops", "armed_tasks.json")  # tasks this lane has armed; each undoable
+
+NOTIFY_COOLDOWN_S = 3600     # one announcement per (node, action) an hour, so the lane cannot spam his phone
+_PEER_RE = re.compile(r"[\w.\-]+:\d+")     # one host:port, the watchdog's own grammar
+
+# Decisions a gate pass can return.
+ACT = "ACT"           # armed, gated, reversible: the engine performs it
+DRY = "DRY"           # it WOULD act, but a pause / may_act==False / dry_run holds it
+PROPOSE = "PROPOSE"   # fails closed OR the action is PROPOSE_ONLY: a person decides
+REFUSE = "REFUSE"     # a hard line: a model trigger, a bad signature, a failed identity, a refusal flag
+
+# The only trigger sources that may ever reach ACT. A model is NOT here, and an
+# unrecognised source is treated as a model would be: data, not authority.
+TRIGGER_SOURCES = frozenset({"roster_scan", "tailnet_presence", "operator"})
+
+# Action classes that are PROPOSE_ONLY forever, matched against what an action
+# declares it IS. These never become AUTO_GATED however rooted the roster, so a
+# careless roster entry naming one of them cannot make the engine run it.
+PROPOSE_ONLY_CLASSES = frozenset({
+    "install", "key", "sign", "consent", "roster-change", "onboard-unknown",
+    "system-setting", "firewall", "power", "security",
+})
+
+# Reuse the highway's subject list so the two lanes refuse the same subjects.
+try:  # pragma: no cover - a trivial import guard
+    import covenant_highway as _H
+    NEVER_AUTOMATIC = _H.NEVER_AUTOMATIC
+except Exception:                                            # noqa: BLE001
+    NEVER_AUTOMATIC = ("money", "holdings", "wallet", "trader", "venue", "order",
+                       "rule", "principle", "judge", "seat", "model", "phone",
+                       "consent", "key", "grid", "breaker", "relay", "feeder",
+                       "substation", "setpoint", "load_shed", "actuat", "scada", "intertie")
+
+
+# ---- the actions this lane can perform, each reversible and local ----------
+#
+# `reversible` False means outward (publish/push/post/send): the gate refuses it
+# to PROPOSE before it is ever reached, so an outward action can never be
+# AUTO_GATED even if a roster lists it. `cls` is matched against
+# PROPOSE_ONLY_CLASSES. `touches` is matched against NEVER_AUTOMATIC, the same
+# way apply_remedy matches a remedy's touches -- a label is one careless edit
+# from wrong, so the subject is checked, not the class alone.
+
+def _do_peer(target, dry_run):
+    """Add the node's address to ops/local_peers.txt. LOCAL state, no outward
+    effect; un-peer (remove the line) undoes it exactly."""
+    addr = str((target or {}).get("peer_addr") or "")
+    if not _PEER_RE.fullmatch(addr):
+        return False, "no valid host:port peer address on the roster entry"
+    cur = _read_peers()
+    if addr in cur:
+        return False, "already peered: %s" % addr
+    if dry_run:
+        return True, "would add peer %s to ops/local_peers.txt" % addr
+    _write_peers(cur + [addr])
+    return True, "peered %s" % addr
+
+
+def _undo_peer(target):
+    addr = str((target or {}).get("peer_addr") or "")
+    return "remove %s from ops/local_peers.txt (un-peer)" % addr
+
+
+def _do_arm_task(target, dry_run):
+    """Arm a vetted task by recording it in ops/armed_tasks.json. Allowed ONLY
+    if the task keeps its own gate (ff-only / opsec / dry); disabling it undoes
+    it. This lane never removes the task's gate and never runs the task -- it
+    flips a local 'may run' flag the task itself still gates behind."""
+    task = (target or {}).get("task") or {}
+    name = str(task.get("name") or "")
+    gate = str(task.get("gate") or "")
+    if not name:
+        return False, "no task name"
+    if gate not in ("ff-only", "opsec", "dry"):
+        return False, "task does not keep its own gate (ff-only/opsec/dry): %r" % gate
+    cur = _read_armed_tasks()
+    if name in cur:
+        return False, "already armed: %s" % name
+    if dry_run:
+        return True, "would arm task %s (keeps its %s gate)" % (name, gate)
+    cur[name] = {"gate": gate, "armed_at": time.strftime("%Y-%m-%dT%H:%M:%S%z")}
+    _write_json(ARMED_TASKS, cur)
+    return True, "armed %s (keeps its %s gate)" % (name, gate)
+
+
+def _undo_arm_task(target):
+    name = str(((target or {}).get("task") or {}).get("name") or "")
+    return "disable %s (remove it from ops/armed_tasks.json)" % name
+
+
+ACTIONS = {
+    "peer": {"fn": _do_peer, "undo": _undo_peer, "reversible": True, "cls": "peer",
+             "touches": ["a node address in ops/local_peers.txt"]},
+    "arm_task": {"fn": _do_arm_task, "undo": _undo_arm_task, "reversible": True, "cls": "arm_task",
+                 "touches": ["a vetted task's armed flag in ops/armed_tasks.json"]},
+    # Declared so the gate and the tests can name them; all refuse before acting.
+    "install": {"fn": None, "undo": None, "reversible": True, "cls": "install",
+                "touches": ["software on a node"]},
+    "set_firewall": {"fn": None, "undo": None, "reversible": True, "cls": "firewall",
+                     "touches": ["a node's firewall (a system setting)"]},
+    "push": {"fn": None, "undo": None, "reversible": False, "cls": "publish",
+             "touches": ["an outward publish"]},
+}
+
+# Which per-node refusal a given action class trips. A node that refuses
+# "system-setting" refuses firewall/power/security actions whatever their class.
+_REFUSAL_OF_CLASS = {"firewall": "system-setting", "power": "system-setting",
+                     "security": "system-setting", "system-setting": "system-setting"}
+
+
+# ----------------------------------------------------------------- readers
+
+def _read_json(path, default):
+    try:
+        with open(path, encoding="utf-8") as fh:
+            return json.load(fh)
+    except (OSError, ValueError):
+        return default
+
+
+def _write_json(path, obj):
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    tmp = path + ".tmp"
+    with open(tmp, "w", encoding="utf-8") as fh:
+        json.dump(obj, fh, ensure_ascii=False, indent=2, sort_keys=True)
+    os.replace(tmp, path)
+
+
+def _read_peers(path=None):
+    """The current off-box peers as a list of host:port, read from the one line
+    covenant_watchdog also reads. A malformed line reads as empty (no guessed peer)."""
+    path = path or LOCAL_PEERS
+    try:
+        with open(path, encoding="utf-8") as fh:
+            s = fh.readline().strip().strip(",")
+    except OSError:
+        return []
+    if not s:
+        return []
+    parts = [p for p in s.split(",") if p]
+    return parts if all(_PEER_RE.fullmatch(p) for p in parts) else []
+
+
+def _write_peers(peers, path=None):
+    path = path or LOCAL_PEERS
+    line = ",".join(peers)
+    if line and not re.fullmatch(r"[\w.\-]+:\d+(,[\w.\-]+:\d+)*", line):
+        raise ValueError("refusing to write a malformed peers line: %r" % line)
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    tmp = path + ".tmp"
+    with open(tmp, "w", encoding="utf-8") as fh:
+        fh.write(line + "\n")
+    os.replace(tmp, path)
+
+
+def _read_armed_tasks(path=None):
+    d = _read_json(path or ARMED_TASKS, {})
+    return d if isinstance(d, dict) else {}
+
+
+def _operator_choices(path=None):
+    d = _read_json(path or CHOICES, {})
+    return d if isinstance(d, dict) else {}
+
+
+def _node_refusals(path=None):
+    """{node_id: [refused classes]} -- refusals only tighten, so this file is
+    safe for the highway to carry. Absent -> no refusals."""
+    d = _read_json(path or NODE_REFUSALS, {})
+    out = {}
+    if isinstance(d, dict):
+        for k, v in d.items():
+            if isinstance(v, list):
+                out[str(k)] = [str(x) for x in v]
+            elif isinstance(v, dict) and isinstance(v.get("refuses"), list):
+                out[str(k)] = [str(x) for x in v["refuses"]]
+    return out
+
+
+# ----------------------------------------------------------------- roster + trust root
+
+def _canonical_payload(roster):
+    """The bytes a signature covers: the roster without its own signature meta.
+    sort_keys so the operator's signer and this verifier agree byte for byte."""
+    body = {k: v for k, v in (roster or {}).items()
+            if k not in ("signature", "signed_by", "alg")}
+    return json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def _verify_signature(roster, pubkey_pem):
+    """True only if the roster carries a signature that `pubkey_pem` verifies over
+    its canonical payload. Any missing piece, any error, any mismatch -> False.
+    Supports whatever key type the operator chose (Ed25519 or RSA PEM); a key this
+    cannot load is a key it cannot verify, which is False, not a pass."""
+    sig_hex = (roster or {}).get("signature")
+    if not sig_hex or not pubkey_pem:
+        return False
+    try:
+        from cryptography.hazmat.primitives import hashes, serialization
+        from cryptography.hazmat.primitives.asymmetric import ed25519, padding
+        key = serialization.load_pem_public_key(pubkey_pem.encode()
+                                                 if isinstance(pubkey_pem, str) else pubkey_pem)
+        sig = bytes.fromhex(sig_hex)
+        payload = _canonical_payload(roster)
+        if isinstance(key, ed25519.Ed25519PublicKey):
+            key.verify(sig, payload)
+        else:  # RSA (or any key whose verify takes a padding+hash); PSS is what a signer should use
+            key.verify(sig, payload, padding.PSS(mgf=padding.MGF1(hashes.SHA256()),
+                                                 salt_length=padding.PSS.MAX_LENGTH), hashes.SHA256())
+        return True
+    except Exception:                                           # noqa: BLE001 -- unverifiable == not verified
+        return False
+
+
+def load_roster(path=None):
+    """The roster dict, or {} when absent/unreadable. {} is the fail-closed shape:
+    no roster means no node is known, so nothing is ever auto-onboarded."""
+    d = _read_json(path or ROSTER, {})
+    return d if isinstance(d, dict) else {}
+
+
+def _operator_pubkey(path=None):
+    try:
+        with open(path or OPERATOR_PUBKEY, encoding="utf-8") as fh:
+            return fh.read()
+    except OSError:
+        return None
+
+
+def roster_trust_root(roster=None, pubkey=None, committed=None):
+    """(root_name, reason). root_name is 'signature' | 'whois' | None.
+
+    Strongest present wins. 'signature' needs the operator's public key and a
+    signature over the roster that verifies. 'whois' needs the roster committed
+    to the repo (so its authenticity rides the repo) AND is then only as strong
+    as the per-entry whois done at identity time. None -> the lane is dark.
+    """
+    roster = load_roster() if roster is None else roster
+    if not roster:
+        return None, "no roster on disk -- auto-onboard is dark"
+    pk = _operator_pubkey() if pubkey is None else pubkey
+    if pk and _verify_signature(roster, pk):
+        return "signature", "roster signed by the operator's key and the signature verifies"
+    if pk and roster.get("signature"):
+        return None, "a signature is present but does NOT verify against the operator's key -- fail closed"
+    committed = _roster_is_committed() if committed is None else committed
+    if committed:
+        return "whois", "no valid signature; roster is repo-committed, anchored per-entry by whois ownership"
+    return None, ("roster is present but neither signed-and-verified nor repo-committed -- "
+                  "no trust root, fail closed")
+
+
+def _roster_is_committed(path=None):
+    """True only if NODE_ROSTER.json is tracked by git (the whois anchor requires
+    the roster's authenticity to ride the repo, not a file anyone could drop in)."""
+    path = path or ROSTER
+    env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")}
+    try:
+        r = subprocess.run(["git", "-C", HERE, "ls-files", "--error-unmatch", os.path.relpath(path, HERE)],
+                           capture_output=True, text=True, timeout=20, env=env, creationflags=_NOWIN)
+        return r.returncode == 0
+    except Exception:                                           # noqa: BLE001
+        return False
+
+
+# ----------------------------------------------------------------- identity (tailscale)
+
+def _tailscale_whois(ip, whois_fn=None):
+    """{'machine_id':.., 'user':..} for an IP, or None. Never trusts an IP alone:
+    the CALLER checks machine id and owning user against the roster. Any failure
+    -> None -> fail closed."""
+    if whois_fn is not None:
+        return whois_fn(ip)
+    try:
+        r = subprocess.run(["tailscale", "whois", "--json", str(ip)],
+                           capture_output=True, text=True, timeout=20, creationflags=_NOWIN)
+        if r.returncode != 0:
+            return None
+        d = json.loads(r.stdout or "null")
+        node = (d or {}).get("Node") or {}
+        user = (d or {}).get("UserProfile") or {}
+        return {"machine_id": node.get("StableID") or node.get("ID"),
+                "user": user.get("LoginName") or user.get("ID"),
+                "key": node.get("Key")}
+    except Exception:                                           # noqa: BLE001
+        return None
+
+
+def _tailscale_nodekey(ip, status_fn=None):
+    """The node's public key pinned from `tailscale status --json`, or None."""
+    if status_fn is not None:
+        return status_fn(ip)
+    try:
+        r = subprocess.run(["tailscale", "status", "--json"],
+                           capture_output=True, text=True, timeout=20, creationflags=_NOWIN)
+        if r.returncode != 0:
+            return None
+        d = json.loads(r.stdout or "null") or {}
+        for peer in list((d.get("Peer") or {}).values()) + [d.get("Self") or {}]:
+            for a in (peer.get("TailscaleIPs") or []):
+                if str(a) == str(ip):
+                    return peer.get("PublicKey") or peer.get("Key")
+        return None
+    except Exception:                                           # noqa: BLE001
+        return None
+
+
+def verify_identity(node_id, entry, operator_account, whois_fn=None, status_fn=None):
+    """(ok, reason). ok only when whois names the operator's account AND the
+    roster's Machine ID, and the node key pinned from status equals the roster's.
+    An IP alone is never enough; any missing or mismatched piece is False."""
+    ip = entry.get("ip")
+    if not ip:
+        return False, "roster entry for %s has no ip" % node_id
+    who = _tailscale_whois(ip, whois_fn=whois_fn)
+    if not who:
+        return False, "whois failed for %s (%s) -- fail closed" % (node_id, ip)
+    if operator_account and str(who.get("user")) != str(operator_account):
+        return False, "whois owner %r != operator %r" % (who.get("user"), operator_account)
+    if entry.get("machine_id") and str(who.get("machine_id")) != str(entry["machine_id"]):
+        return False, "whois machine id %r != rostered %r" % (who.get("machine_id"), entry["machine_id"])
+    pinned = _tailscale_nodekey(ip, status_fn=status_fn)
+    if entry.get("pubkey"):
+        if not pinned:
+            return False, "could not pin a node key for %s -- fail closed" % node_id
+        if str(pinned) != str(entry["pubkey"]):
+            return False, "pinned node key != rostered node key for %s" % node_id
+    return True, "identity verified by whois (owner, machine id) and pinned node key"
+
+
+# ----------------------------------------------------------------- the gate
+
+def _grid_route(action):
+    """A reason string if `action` is a grid/ICS effect that must go to the grid
+    guard's together-vote, else None. The class is DERIVED from covenant_grid_guard's
+    ACTION_CLASS allowlist -- the one source of truth, reused not forked. If the guard
+    cannot be imported, this returns None: the known-action check still refuses
+    anything that is not one of this lane's own (non-grid) reversible actions, so a
+    missing guard opens no hole."""
+    try:
+        import covenant_grid_guard as G
+        cls = G.required_class(action)
+    except Exception:                                           # noqa: BLE001
+        return None
+    if cls is not None:
+        holders = "+".join(getattr(G, "AUTHORITY_ROLES", ("operator", "tetsu")))
+        return ("grid/ICS action %r (class %s) -> covenant_grid_guard together-vote (%s), "
+                "fail-closed, never AUTO_GATED (data/anomaly is ALARM-only, never authority)"
+                % (action, cls, holders))
+    return None
+
+
+def is_armed(roster=None, armed_cfg=None, pubkey=None, committed=None):
+    """(armed, trust_root, reason). The lane is armed ONLY when the operator wrote
+    ops/AUTO_ONBOARD_ARMED choosing a trust root AND that trust root is satisfiable
+    now. Ships DARK: the file does not exist, so this returns (False, None, ...)."""
+    cfg = _read_json(ARMED_FILE, None) if armed_cfg is None else armed_cfg
+    if not isinstance(cfg, dict) or not cfg.get("armed"):
+        return False, None, "ops/AUTO_ONBOARD_ARMED absent or armed!=true -- the lane is dark"
+    want = str(cfg.get("trust_root") or "auto")
+    root, why = roster_trust_root(roster=roster, pubkey=pubkey, committed=committed)
+    if root is None:
+        return False, None, "armed, but %s" % why
+    if want not in ("auto", root):
+        return False, None, "operator asked trust root %r but the strongest present is %r" % (want, root)
+    return True, root, "armed; trust root %s (%s)" % (root, why)
+
+
+def evaluate(action, target, trigger, *, roster=None, choices=None, refusals=None,
+             operator_account=None, armed_cfg=None, pubkey=None, committed=None,
+             whois_fn=None, status_fn=None, paused=None, may_act=True, now=None):
+    """The whole gate for one (action, target, trigger), in order, fail-closed.
+
+    Returns {decision, reason, undo, action, node, klass}. decision is ACT only
+    when every gate passes AND the lane is armed AND nothing holds it dry; DRY
+    when it would act but a pause / may_act False / a declared dry holds it;
+    PROPOSE when it fails closed or the action is PROPOSE_ONLY; REFUSE on a hard
+    line (a model trigger, a bad signature, a failed identity, a refusal flag).
+    """
+    node_id = str((target or {}).get("node") or (target or {}).get("id") or "?")
+    spec = ACTIONS.get(action)
+    base = {"action": action, "node": node_id, "klass": AUTO_GATED, "undo": "n/a"}
+
+    def out(decision, reason):
+        return dict(base, decision=decision, reason=reason)
+
+    # 1. A MODEL'S OUTPUT NEVER TRIGGERS AUTO_GATED. Data is not authority.
+    src = (trigger or {}).get("source") if isinstance(trigger, dict) else None
+    if src not in TRIGGER_SOURCES:
+        return out(REFUSE, "trigger source %r is not operator-rooted -- a model's output can never "
+                           "trigger AUTO_GATED (data is not authority)" % src)
+
+    # 2. GRID / ICS DEFERS TO covenant_grid_guard -- BEFORE the known-action check, so a
+    #    grid action named on a roster is routed to the guard rather than called "unknown".
+    #    The class is DERIVED from the grid guard's own ACTION_CLASS allowlist (the single
+    #    source of truth -- no forked grid word list, per the PG1/rule-6 lesson): any action
+    #    it recognises is grid-domain and goes to the together-vote (operator + Tetsu),
+    #    fail-closed, never AUTO_GATED. An action the allowlist does not know is not treated
+    #    as grid here; it falls to the known-action check below, which refuses the unknown.
+    gr = _grid_route(action)
+    if gr:
+        return dict(base, decision=REFUSE, reason=gr, route="grid_guard")
+
+    # 3. a known action
+    if spec is None:
+        return out(REFUSE, "unknown action %r" % action)
+
+    # 4. a per-node refusal, honoured whatever the class and BEFORE the
+    #    PROPOSE_ONLY short-circuit below, so it is a guard in its own right: a
+    #    node that refuses 'system-setting' REFUSES (a hard no), where an
+    #    unflagged node's system-setting action merely PROPOSES. TETSU sets this.
+    refusals = _node_refusals() if refusals is None else refusals
+    refused = set(refusals.get(node_id, []))
+    cls_refusal = _REFUSAL_OF_CLASS.get(spec.get("cls"))
+    if spec.get("cls") in refused or (cls_refusal and cls_refusal in refused):
+        return out(REFUSE, "%s refuses %s auto-actions (per-node refusal)" % (node_id, cls_refusal or spec["cls"]))
+
+    # 5. PROPOSE_ONLY forever -- by the action's declared class, and by its subject
+    if spec.get("cls") in PROPOSE_ONLY_CLASSES:
+        return out(PROPOSE, "action class %r is PROPOSE_ONLY forever -- a person decides" % spec["cls"])
+    crossed = sorted({w for w in NEVER_AUTOMATIC for t in spec.get("touches", []) if w in t.lower()})
+    if crossed:
+        return out(PROPOSE, "touches %s -- refused whatever class it claims" % ", ".join(crossed))
+
+    # 6. outward actions are not reversible, so never AUTO_GATED
+    if not spec.get("reversible"):
+        return out(PROPOSE, "action %r is not reversible (outward: cannot be unsent) -- PROPOSE_ONLY" % action)
+
+    # 7. ARMED and a trust root present, else DARK (fail closed -> propose)
+    armed, root, why = is_armed(roster=roster, armed_cfg=armed_cfg, pubkey=pubkey, committed=committed)
+    if not armed:
+        return out(PROPOSE, "auto-onboard dark: %s" % why)
+    base["root"] = root
+
+    # 8. the target is on the roster (an unknown node is never auto-onboarded)
+    r = load_roster() if roster is None else roster
+    nodes = (r.get("nodes") or {}) if isinstance(r, dict) else {}
+    entry = nodes.get(node_id)
+    if not isinstance(entry, dict):
+        return out(PROPOSE, "node %r is not on the roster -- onboarding an unknown node is PROPOSE_ONLY" % node_id)
+
+    # 9. the roster allows THIS action for THIS node
+    allowed = entry.get("allowed") or entry.get("actions") or []
+    if action not in allowed:
+        return out(REFUSE, "the roster does not allow %r for %s (allowed: %s)" % (action, node_id, allowed))
+
+    # 10. identity verified, not minted
+    acct = operator_account if operator_account is not None else r.get("operator_account")
+    ok, idreason = verify_identity(node_id, entry, acct, whois_fn=whois_fn, status_fn=status_fn)
+    if not ok:
+        return out(REFUSE, "identity not verified: %s" % idreason)
+
+    # 11. the operator's explicit choices are untouchable
+    choices = _operator_choices() if choices is None else choices
+    clash = sorted(set(spec.get("touches", [])) & set(choices or {}))
+    if node_id in (choices or {}):
+        clash = sorted(set(clash) | {node_id})
+    if clash:
+        return out(REFUSE, "the operator chose %s: %s" % (clash[0], (choices or {}).get(clash[0], "")))
+
+    # the undo is on record from here
+    undo_fn = spec.get("undo")
+    base["undo"] = undo_fn(target) if callable(undo_fn) else "n/a"
+
+    # 12. pause / may_act==False -> it would act, but DRY (A244)
+    if paused is None:
+        paused = _paused()
+    if paused or not may_act:
+        return out(DRY, "gated and armed, but held DRY (%s)" % ("paused" if paused else "highway_may_act() is False"))
+
+    return out(ACT, "armed, rostered, identity-verified, reversible -- the engine may act (%s)" % idreason)
+
+
+def _paused():
+    try:
+        import covenant_pause as _p
+        is_paused, _why = _p.paused("highway")
+        return bool(is_paused)
+    except Exception:                                           # noqa: BLE001
+        return False
+
+
+# ----------------------------------------------------------------- announce + ledger
+
+def _recent_notify(node_id, action, ledger=None, now=None):
+    now = time.time() if now is None else now
+    for row in reversed(_read_ledger(ledger)):
+        if row.get("klass") == AUTO_GATED and row.get("node") == node_id and row.get("action") == action \
+                and row.get("announced"):
+            return (now - float(row.get("at", 0))) < NOTIFY_COOLDOWN_S
+    return False
+
+
+def announce(decision_row, ledger=None, notify_fn=None, now=None):
+    """Announce an auto-action with its undo command, rate-limited per (node, action).
+    A notifier that can bring down what it reports on has inverted its purpose, so
+    this never raises and returns whether it sent."""
+    node_id, action = decision_row.get("node"), decision_row.get("action")
+    if _recent_notify(node_id, action, ledger=ledger, now=now):
+        return False
+    title = "highway auto-onboard: %s %s" % (action, node_id)
+    body = "%s\nundo: %s" % (decision_row.get("reason", ""), decision_row.get("undo", "n/a"))
+    try:
+        if notify_fn is not None:
+            notify_fn(title, body)
+        else:
+            import covenant_notify
+            covenant_notify.notify(title, body, priority="default")
+        return True
+    except Exception:                                           # noqa: BLE001
+        return False
+
+
+def _read_ledger(path=None):
+    out = []
+    try:
+        with open(path or LEDGER, encoding="utf-8") as fh:
+            for line in fh:
+                try:
+                    out.append(json.loads(line))
+                except ValueError:
+                    continue
+    except OSError:
+        pass
+    return out
+
+
+def _write_ledger(row, path=None):
+    path = path or LEDGER
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "a", encoding="utf-8") as fh:
+        fh.write(json.dumps(row, ensure_ascii=False) + "\n")
+    return row
+
+
+def act(action, target, trigger, *, dry_run=True, ledger=None, notify_fn=None, now=None, **gate):
+    """Evaluate the gate and, only on ACT and not dry_run, perform the reversible
+    action -- then log the row (klass AUTO_GATED, to the highway ledger) and
+    announce it with its undo. DRY/PROPOSE/REFUSE are logged too: a refusal nobody
+    can see is indistinguishable from a bug that swallowed the work."""
+    d = evaluate(action, target, trigger, **gate)
+    row = {"t": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "at": round(now if now is not None else time.time(), 1),
+           "klass": AUTO_GATED, "action": action, "node": d.get("node"), "detector": "auto_onboard",
+           "decision": d["decision"], "why": d["reason"], "undo": d.get("undo", "n/a"),
+           "trigger_source": (trigger or {}).get("source") if isinstance(trigger, dict) else None,
+           "dry_run": bool(dry_run)}
+    if d["decision"] != ACT or dry_run:
+        row["outcome"] = {ACT: "dry run", DRY: "held (dry)", PROPOSE: "proposed", REFUSE: "refused"}[d["decision"]]
+        row["ran"] = False
+        return _write_ledger(row, ledger)
+    spec = ACTIONS.get(action) or {}
+    fn = spec.get("fn")
+    if not callable(fn):
+        row.update(outcome="refused", ran=False, why="no executable function for %r" % action)
+        return _write_ledger(row, ledger)
+    ok, detail = fn(target, dry_run=False)
+    row.update(ran=bool(ok), detail=str(detail)[:300],
+               outcome="done" if ok else "held", undo=(spec["undo"](target) if ok and callable(spec.get("undo")) else row["undo"]))
+    _write_ledger(row, ledger)
+    if ok:
+        row_sent = dict(d, undo=row["undo"])
+        if announce(row_sent, ledger=ledger, notify_fn=notify_fn, now=now):
+            _write_ledger(dict(row, announced=True, outcome="announced",
+                               at=round(now if now is not None else time.time(), 1)), ledger)
+    return row
+
+
+# ----------------------------------------------------------------- the pass
+
+def _candidates(roster):
+    """[(action, target, trigger)] the roster offers: a rostered node not yet
+    peered -> peer it; a vetted task not yet armed -> arm it. The trigger source
+    is 'roster_scan' -- never a model. An empty/absent roster yields nothing."""
+    out = []
+    nodes = (roster.get("nodes") or {}) if isinstance(roster, dict) else {}
+    peers = set(_read_peers())
+    for node_id, entry in sorted(nodes.items()):
+        if not isinstance(entry, dict):
+            continue
+        allowed = entry.get("allowed") or entry.get("actions") or []
+        if "peer" in allowed and str(entry.get("peer_addr") or "") not in peers:
+            out.append(("peer", {"node": node_id, "peer_addr": entry.get("peer_addr")},
+                        {"source": "roster_scan"}))
+        for task in (entry.get("tasks") or []):
+            if "arm_task" in allowed and task.get("name") not in _read_armed_tasks():
+                out.append(("arm_task", {"node": node_id, "task": task}, {"source": "roster_scan"}))
+    return out
+
+
+def run_once(dry_run=True, ledger=None, now=None, **gate):
+    """One auto-onboard pass, in the highway's (alerts, infos) shape so its hook
+    can merge the lines. Honours pause and may_act (both default from the gate).
+    DARK and empty by default: no roster, not armed -> one info line, no action."""
+    alerts, infos = [], []
+    roster = gate.get("roster")
+    roster = load_roster() if roster is None else roster
+    armed, root, why = is_armed(roster=roster, pubkey=gate.get("pubkey"),
+                                armed_cfg=gate.get("armed_cfg"), committed=gate.get("committed"))
+    if not armed:
+        infos.append("highway auto-onboard: DARK -- %s" % why)
+        return alerts, infos
+    infos.append("highway auto-onboard: armed (trust root %s)" % root)
+    for action, target, trigger in _candidates(roster):
+        row = act(action, target, trigger, dry_run=dry_run, ledger=ledger, now=now, roster=roster, **{
+            k: v for k, v in gate.items() if k in
+            ("choices", "refusals", "operator_account", "armed_cfg", "pubkey", "committed",
+             "whois_fn", "status_fn", "paused", "may_act")})
+        line = "highway auto-onboard: %s %s -> %s (%s)" % (action, target.get("node"), row.get("outcome"), row.get("why", "")[:120])
+        (infos if row.get("outcome") in ("done", "announced", "dry run") else alerts).append(line)
+    return alerts, infos
+
+
+# ----------------------------------------------------------------- CLI
+
+def main(argv=None):
+    import argparse
+    ap = argparse.ArgumentParser(description="the highway's AUTO_GATED auto-onboard lane")
+    ap.add_argument("--plan", action="store_true", help="show the candidates and the gate's verdict (DRY)")
+    args = ap.parse_args(argv)
+    roster = load_roster()
+    armed, root, why = is_armed(roster=roster)
+    print("auto-onboard: %s" % ("ARMED (trust root %s)" % root if armed else "DARK -- %s" % why))
+    if args.plan:
+        alerts, infos = run_once(dry_run=True)
+        for ln in infos + alerts:
+            print("  " + ln)
+    return 0
+
+
+if __name__ == "__main__":
+    import sys
+    sys.exit(main())

covenant_one.py

@@ -619,10 +619,22 @@ SUITES = [
     # mutation-tested. Hermetic: fixture conditions, temp ledgers, no node and
     # no network, so it measures the rules rather than the machine.
     ("test_h1_highway.py", 180, "DAILY + GUARDS"),
+    # SF1 (2026-11-02, A405): the highway's sync-fabric detectors and remedies, each driven both ways. The
+    # unison feed's staleness, a sync task that fails or stops firing (re-run, AUTO_REVERSIBLE), a task gone
+    # entirely (config, PROPOSE_ONLY), and HF/website vs github mirror lag -- github never auto-pushed. Hermetic:
+    # feed and scheduler are fixtures, the cache a temp file, no node and no network.
+    ("test_sf1_sync_fabric.py", 120, "DAILY + GUARDS"),
     # NA1 (2026-10-20): H1i names six protected subjects; the engine's NEVER_AUTOMATIC tuple had fourteen
     # (twenty-four since 2026-10-30: the grid/ICS-OT actuation words, A392).
     # NA1 drives every one through apply_remedy so the money/seat/rule/consent/key list cannot silently shrink.
     ("test_na1_never_automatic.py", 120, "DAILY + GUARDS"),
+    # OG1 (2026-11-02, "loosen autonomy ... create its own lane"): the highway's AUTO_GATED auto-onboard lane
+    # (covenant_onboard.py) -- a third class between AUTO_REVERSIBLE and PROPOSE_ONLY. Every gate driven both
+    # ways: a model trigger refused, an off-roster node never onboarded, an absent/unsigned/bad-signature roster
+    # dark, a failed whois refused, a verified node auto-peered AND un-peered, a system-setting action refused on
+    # a flagged node, an outward action never AUTO_GATED. Hermetic: an in-process key signs a fixture roster,
+    # identity is stubbed, every path is a temp file. Ships DARK and NOT ARMED.
+    ("test_og1_onboard.py", 120, "DAILY + GUARDS"),
     # PB1 (2026-09-26, A231): the highway's two phone-build signals measure what they name -- a look that finds
     # nothing newer counts as a look (fetch_build was quarantined for it), and "behind" means behind in the
     # files the APK ships, not in every docs commit.
@@ -824,6 +836,12 @@ SUITES = [
     # HM1 (2026-10-31, his "fix hf_mirror.py"): the HF mirror pushes fast-forward only, behind his opsec scan, and
     # snapshot-only repos never send history -- it used to force-push full history from a hook-less clone (A396).
     ("test_hm1_hf_mirror.py", 120, "SECURITY"),
+    # AP1/HD1 (2026-11-02, A404, his "Enable auto-push" + "as much redundancy as possible"): the laptop pusher
+    # off TETSU's unison feed (ff-only vs current GitHub main, open-PR guard, no --force/tags/unify-branch, never
+    # a snapshot repo's history) and hf_mirror's fail-closed holds-drift guard (refuse a history push when the
+    # feed can't confirm a repo was not marked HF-snapshot elsewhere). Both ways; each guard mutated -> suite RED.
+    ("test_ap1_repo_autopush.py", 120, "SECURITY"),
+    ("test_hd1_hf_mirror_holds.py", 60, "SECURITY"),
     ("test_sw1_study_web.py", 120, "JUDGE"),
     ("test_r2v_frames.py", 240, "JUDGE"),
     # PC2 (2026-10-03, the core's pin; 2026-10-06, A315, the other four through tools/pin_deploy.py): every