Repair Daybreak adversarial discovery invariants

src/lumi_eggcracker/approvals.py

@@ -86,6 +86,8 @@ def approved(
     *,
     executable_metadata: os.stat_result | tuple[int, int] | None = None,
 ) -> bool:
+    if not getattr(snapshot, "argv_complete", True):
+        return False
     command_hash = argv_digest(snapshot.argv)
     if executable_metadata is None:
         try:

src/lumi_eggcracker/artifacts.py

@@ -16,6 +16,7 @@ from .jsonio import JsonInputError
 MAX_ARTIFACT_BYTES = 1 * 1024 * 1024
 MAX_ARTIFACTS = 16
 MAX_TOTAL_BYTES = 4 * 1024 * 1024
+MAX_FD_PROBES_PER_SCAN = 256
 MAX_GGUF_ITEMS = 10_000_000
 MAX_SAFETENSORS_TENSORS = 100_000
 MAX_SAFETENSORS_DIMS = 64
@@ -292,19 +293,43 @@ def from_process_fds(
     *,
     proc: Path = Path("/proc"),
     cache: ArtifactCache | None = None,
+    start_index: int = 0,
+    max_probes: int = MAX_FD_PROBES_PER_SCAN,
 ) -> tuple[ArtifactEvidence, ...]:
-    """Inspect a bounded set of currently open target descriptors.
+    """Inspect one fair bounded window of currently open target descriptors.
 
     The descriptor is opened through ``/proc/<pid>/fd/<n>``.  The link name is
     used only to find a descriptor number; it is never accepted as evidence.
+    Callers advance ``start_index`` between scans so a held-open artifact
+    cannot be hidden permanently behind harmless lower-numbered descriptors.
     """
     identity = snapshot.identity
-    entries = tuple(getattr(snapshot, "fd_entries", ()))[:MAX_ARTIFACTS]
+    if isinstance(start_index, bool) or not isinstance(start_index, int) or start_index < 0:
+        raise JsonInputError("artifact descriptor cursor is invalid")
+    if isinstance(max_probes, bool) or not isinstance(max_probes, int) or max_probes < 1:
+        raise JsonInputError("artifact descriptor probe limit is invalid")
+    directory = proc / str(identity.pid) / "fd"
+    try:
+        numbers = sorted(
+            int(item.name) for item in directory.iterdir() if item.name.isdigit()
+        )
+    except (AttributeError, OSError):
+        numbers = sorted(
+            {
+                number
+                for number, _target in tuple(getattr(snapshot, "fd_entries", ()))
+                if isinstance(number, int) and not isinstance(number, bool) and number >= 0
+            }
+        )
+    if numbers:
+        offset = start_index % len(numbers)
+        probe_count = min(max_probes, len(numbers))
+        selected = tuple(numbers[(offset + index) % len(numbers)] for index in range(probe_count))
+    else:
+        selected = ()
     result: list[ArtifactEvidence] = []
     consumed = 0
-    for number, _target in entries:
-        if not isinstance(number, int) or number < 0:
-            continue
+    for number in selected:
         # /proc/<pid>/fd/<n> is necessarily a procfs symlink.  We open that
         # exact descriptor target, then validate the opened inode; O_NOFOLLOW
         # would reject every legitimate procfs descriptor.
@@ -349,9 +374,19 @@ def from_snapshot(
     *,
     proc: Path = Path("/proc"),
     cache: ArtifactCache | None = None,
+    fd_start_index: int = 0,
+    fd_max_probes: int = MAX_FD_PROBES_PER_SCAN,
 ) -> tuple[ArtifactEvidence, ...]:
     """Collect bounded GGUF evidence from open descriptors and mapped files."""
-    result = list(from_process_fds(snapshot, proc=proc, cache=cache))
+    result = list(
+        from_process_fds(
+            snapshot,
+            proc=proc,
+            cache=cache,
+            start_index=fd_start_index,
+            max_probes=fd_max_probes,
+        )
+    )
     for raw in tuple(getattr(snapshot, "map_paths", ())):
         if not isinstance(raw, str) or not raw.startswith("/"):
             continue

src/lumi_eggcracker/discovery.py

@@ -13,7 +13,8 @@ from .jsonio import JsonInputError
 
 PROC = Path("/proc")
 MAX_READ = 64 * 1024
-MAX_MAP_BYTES = 256 * 1024
+MAX_CMDLINE_BYTES = 8 * 1024 * 1024
+MAX_MAP_BYTES = 8 * 1024 * 1024
 MAX_FDS = 256
 MAX_MAPS = 512
 
@@ -37,6 +38,7 @@ class ProcessSnapshot:
     fd_entries: tuple[tuple[int, str], ...] = ()
     map_paths: tuple[str, ...] = ()
     parent: ProcessIdentity | None = None
+    argv_complete: bool = True
 
 
 def _read(path: Path, limit: int = MAX_READ) -> bytes:
@@ -79,7 +81,9 @@ def identity(pid: int, *, proc: Path = PROC) -> ProcessIdentity | None:
 
 
 def argv_digest(argv: tuple[str, ...] | list[str]) -> str:
-    return hashlib.sha256("\0".join(argv).encode("utf-8")).hexdigest()
+    return hashlib.sha256(
+        "\0".join(argv).encode("utf-8", errors="surrogateescape")
+    ).hexdigest()
 
 
 def executable_digest(path: Path) -> str:
@@ -214,9 +218,11 @@ def _links(directory: Path, *, limit: int) -> tuple[str, ...]:
 
 def _fd_entries(directory: Path, *, limit: int) -> tuple[tuple[int, str], ...]:
     values: list[tuple[int, str]] = []
-    for item in sorted(directory.iterdir(), key=lambda value: value.name)[:limit]:
-        if not item.name.isdigit():
-            continue
+    entries = sorted(
+        (item for item in directory.iterdir() if item.name.isdigit()),
+        key=lambda value: int(value.name),
+    )
+    for item in entries[:limit]:
         try:
             target = os.readlink(item)
         except OSError:
@@ -229,7 +235,7 @@ def _fd_entries(directory: Path, *, limit: int) -> tuple[tuple[int, str], ...]:
 def _maps(path: Path) -> tuple[str, ...]:
     try:
         raw = _read(path, MAX_MAP_BYTES).decode("utf-8", errors="replace")
-    except OSError:
+    except (OSError, JsonInputError):
         return ()
     values: list[str] = []
     for line in raw.splitlines()[:MAX_MAPS]:
@@ -242,7 +248,7 @@ def _maps(path: Path) -> tuple[str, ...]:
 def _map_paths(path: Path) -> tuple[str, ...]:
     try:
         raw = _read(path, MAX_MAP_BYTES).decode("utf-8", errors="replace")
-    except OSError:
+    except (OSError, JsonInputError):
         return ()
     values: list[str] = []
     for line in raw.splitlines()[:MAX_MAPS]:
@@ -252,7 +258,12 @@ def _map_paths(path: Path) -> tuple[str, ...]:
     return tuple(values)
 
 
-def snapshot(value: ProcessIdentity, *, proc: Path = PROC) -> ProcessSnapshot | None:
+def snapshot(
+    value: ProcessIdentity,
+    *,
+    proc: Path = PROC,
+    include_evidence: bool = True,
+) -> ProcessSnapshot | None:
     current = identity(value.pid, proc=proc)
     if current != value:
         return None
@@ -260,20 +271,45 @@ def snapshot(value: ProcessIdentity, *, proc: Path = PROC) -> ProcessSnapshot |
     try:
         parent_pid, _start = parse_stat(_read(root / "stat", 4096).decode("utf-8"))
         exe = os.readlink(root / "exe")
-        command = tuple(
-            item for item in _read(root / "cmdline").decode("utf-8").split("\0") if item
-        )
         cgroups = tuple(
             line for line in _read(root / "cgroup", 8192).decode("utf-8").splitlines() if line
         )
         uid = _uid(_read(root / "status", 8192).decode("utf-8"))
-        fd_entries = _fd_entries(root / "fd", limit=MAX_FDS)
-        fd_paths = tuple(item[1] for item in fd_entries)
     except (OSError, UnicodeDecodeError, JsonInputError):
         return None
-    if not command or not exe or len(exe) > 4096:
+    if not exe or len(exe) > 4096:
         return None
-    maps = _map_paths(root / "maps")
+
+    # Linux argv is a byte sequence.  Invalid UTF-8 must not erase the whole
+    # process from discovery, and a bounded-read failure must make approval
+    # unavailable rather than making the process invisible.
+    try:
+        command = tuple(
+            item
+            for item in _read(root / "cmdline", MAX_CMDLINE_BYTES)
+            .decode("utf-8", errors="surrogateescape")
+            .split("\0")
+            if item
+        )
+        argv_complete = True
+    except (OSError, JsonInputError):
+        command = ()
+        argv_complete = False
+
+    fd_entries: tuple[tuple[int, str], ...] = ()
+    maps: tuple[str, ...] = ()
+    if include_evidence:
+        try:
+            fd_entries = _fd_entries(root / "fd", limit=MAX_FDS)
+        except OSError:
+            fd_entries = ()
+        # Native runtime discovery uses stable /proc/<pid>/map_files handles.
+        # This fallback is retained for kernels/test fixtures without it and
+        # is deliberately isolated so one oversized maps file cannot abort a
+        # global discovery pass.
+        if not (root / "map_files").is_dir():
+            maps = _map_paths(root / "maps")
+    fd_paths = tuple(item[1] for item in fd_entries)
     parent = identity(parent_pid, proc=proc) if parent_pid > 0 else None
     return ProcessSnapshot(
         value,
@@ -287,11 +323,15 @@ def snapshot(value: ProcessIdentity, *, proc: Path = PROC) -> ProcessSnapshot |
         fd_entries,
         maps,
         parent,
+        argv_complete,
     )
 
 
 def scan(
-    *, proc: Path = PROC, exclude: Callable[[ProcessSnapshot], bool] | None = None
+    *,
+    proc: Path = PROC,
+    exclude: Callable[[ProcessSnapshot], bool] | None = None,
+    include_evidence: bool = True,
 ) -> list[ProcessSnapshot]:
     result: list[ProcessSnapshot] = []
     try:
@@ -302,7 +342,7 @@ def scan(
         item = identity(pid, proc=proc)
         if item is None:
             continue
-        current = snapshot(item, proc=proc)
+        current = snapshot(item, proc=proc, include_evidence=include_evidence)
         if current is not None and not (exclude and exclude(current)):
             result.append(current)
     return result

src/lumi_eggcracker/elfmarkers.py

@@ -30,7 +30,7 @@ PINNED_PYTORCH_ATEN_BUILD_IDS = frozenset({"ad9ab6eeec3b28a0ec3f12f266627610de90
 PYTORCH_BRIDGE_EVIDENCE_ID = "pytorch-bridge-build-id-pinned-cpu"
 PYTORCH_ATEN_EVIDENCE_ID = "pytorch-aten-build-id-pinned-cpu"
 PYTORCH_PAIR_EVIDENCE_ID = "pytorch-bridge-aten-pair-pinned-cpu"
-MAX_RUNTIME_CANDIDATES = 128
+MAX_RUNTIME_CANDIDATES = 256
 
 
 @dataclass(frozen=True)
@@ -182,13 +182,8 @@ def _build_id(descriptor: int, size: int) -> str | None:
     return None
 
 
-def inspect_path(path: Path) -> RuntimeEvidence | None:
-    """Recognise a llama/GGML ELF by two valid symbol-table markers, not name."""
-    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
-    try:
-        descriptor = os.open(path, flags)
-    except OSError:
-        return None
+def _inspect_llama_descriptor(descriptor: int) -> RuntimeEvidence | None:
+    """Recognise llama/GGML through one already-open, stable descriptor."""
     try:
         before = os.fstat(descriptor)
         if not stat.S_ISREG(before.st_mode) or before.st_size < 64 or before.st_size > (1 << 40):
@@ -214,17 +209,10 @@ def inspect_path(path: Path) -> RuntimeEvidence | None:
         return None
     except (JsonInputError, OSError, struct.error):
         return None
-    finally:
-        os.close(descriptor)
 
 
-def inspect_pytorch_path(path: Path) -> RuntimeEvidence | None:
-    """Recognise one exact pinned PyTorch bridge/ATen ELF identity."""
-    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
-    try:
-        descriptor = os.open(path, flags)
-    except OSError:
-        return None
+def _inspect_pytorch_descriptor(descriptor: int) -> RuntimeEvidence | None:
+    """Recognise one exact pinned PyTorch bridge/ATen descriptor identity."""
     try:
         before = os.fstat(descriptor)
         if not stat.S_ISREG(before.st_mode) or before.st_size < 64:
@@ -248,15 +236,50 @@ def inspect_pytorch_path(path: Path) -> RuntimeEvidence | None:
         return None
     except (JsonInputError, OSError, struct.error):
         return None
+
+
+def _inspect_descriptor(descriptor: int) -> tuple[RuntimeEvidence, ...]:
+    values: list[RuntimeEvidence] = []
+    llama = _inspect_llama_descriptor(descriptor)
+    if llama is not None:
+        values.append(llama)
+    pytorch = _inspect_pytorch_descriptor(descriptor)
+    if pytorch is not None and pytorch not in values:
+        values.append(pytorch)
+    return tuple(values)
+
+
+def inspect_path(path: Path) -> RuntimeEvidence | None:
+    """Recognise a llama/GGML ELF by two valid symbol-table markers, not name."""
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except OSError:
+        return None
+    try:
+        return _inspect_llama_descriptor(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def inspect_pytorch_path(path: Path) -> RuntimeEvidence | None:
+    """Recognise one exact pinned PyTorch bridge/ATen ELF identity."""
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except OSError:
+        return None
+    try:
+        return _inspect_pytorch_descriptor(descriptor)
     finally:
         os.close(descriptor)
 
 
 def _inspect_candidate(path: Path) -> tuple[RuntimeEvidence, ...]:
     values: list[RuntimeEvidence] = []
-    evidence = inspect_path(path)
-    if evidence is not None:
-        values.append(evidence)
+    llama = inspect_path(path)
+    if llama is not None:
+        values.append(llama)
     pytorch = inspect_pytorch_path(path)
     if pytorch is not None and pytorch not in values:
         values.append(pytorch)
@@ -276,38 +299,122 @@ def with_pytorch_pair(values: Iterable[RuntimeEvidence]) -> tuple[RuntimeEvidenc
     return tuple(result)
 
 
+def _cached_descriptor(
+    descriptor: int, cache: RuntimeCache | None
+) -> tuple[RuntimeEvidence, ...]:
+    try:
+        metadata = os.fstat(descriptor)
+        key = _cache_key(metadata)
+    except OSError:
+        return ()
+    if cache is not None and key in cache:
+        return cache[key]
+    values = _inspect_descriptor(descriptor)
+    if cache is not None:
+        cache[key] = values
+    return values
+
+
+def _mapping_references(snapshot: object, proc: Path) -> tuple[str, ...]:
+    try:
+        pid = snapshot.identity.pid
+        directory = proc / str(pid) / "map_files"
+        values = [
+            item.name
+            for item in directory.iterdir()
+            if len(item.name.split("-", 1)) == 2
+            and all(
+                part and all(character in "0123456789abcdefABCDEF" for character in part)
+                for part in item.name.split("-", 1)
+            )
+        ]
+    except (AttributeError, OSError):
+        return ()
+    return tuple(sorted(values, key=lambda value: int(value.split("-", 1)[0], 16)))
+
+
 def from_snapshot(
-    snapshot: object, *, cache: RuntimeCache | None = None
+    snapshot: object,
+    *,
+    cache: RuntimeCache | None = None,
+    proc: Path = Path("/proc"),
+    start_index: int = 0,
+    max_candidates: int = MAX_RUNTIME_CANDIDATES,
 ) -> tuple[RuntimeEvidence, ...]:
-    """Inspect executable and mapped regular files; their names are ignored."""
-    # ``/proc/<pid>/maps`` repeats each shared object once per mapped segment.
-    # Deduplicate before applying the bounded candidate cap; otherwise a busy
-    # Python/AI process can exhaust the cap on unrelated NumPy/locale segments
-    # and hide the exact PyTorch bridge/ATen objects that are mapped later.
-    candidates: list[str] = []
-    seen: set[str] = set()
-    for raw in (getattr(snapshot, "exe_path", ""), *getattr(snapshot, "map_paths", ())):
-        if isinstance(raw, str) and raw not in seen:
-            seen.add(raw)
-            candidates.append(raw)
+    """Inspect one fair window of live executable/mapping descriptors.
+
+    Native Linux inspection opens ``/proc/<pid>/map_files/<range>`` directly.
+    The mapped inode therefore remains inspectable after its pathname is
+    renamed or deleted.  Path reopening is only a compatibility fallback for
+    fixtures or kernels where ``map_files`` is unavailable.
+    """
+    if isinstance(start_index, bool) or not isinstance(start_index, int) or start_index < 0:
+        raise JsonInputError("runtime mapping cursor is invalid")
+    if isinstance(max_candidates, bool) or not isinstance(max_candidates, int) or max_candidates < 1:
+        raise JsonInputError("runtime candidate limit is invalid")
+
     result: list[RuntimeEvidence] = []
-    for raw in candidates[:MAX_RUNTIME_CANDIDATES]:
-        if not isinstance(raw, str) or not raw.startswith("/"):
-            continue
-        path = Path(raw)
-        if cache is None:
-            values = _inspect_candidate(path)
-        else:
-            try:
-                key = _cache_key(path.stat())
-            except OSError:
-                continue
-            if key in cache:
-                values = cache[key]
-            else:
-                values = _inspect_candidate(path)
-                cache[key] = values
+
+    def add(values: tuple[RuntimeEvidence, ...]) -> None:
         for evidence in values:
             if evidence not in result:
                 result.append(evidence)
+
+    # Bind the executable through procfs as well; a deleted executable path is
+    # not an excuse to lose an otherwise exact runtime identity.
+    try:
+        executable = proc / str(snapshot.identity.pid) / "exe"
+        descriptor = os.open(executable, os.O_RDONLY | getattr(os, "O_CLOEXEC", 0))
+    except (AttributeError, OSError):
+        descriptor = -1
+    if descriptor >= 0:
+        try:
+            add(_cached_descriptor(descriptor, cache))
+        finally:
+            os.close(descriptor)
+
+    references = _mapping_references(snapshot, proc)
+    if references:
+        offset = start_index % len(references)
+        count = min(max_candidates, len(references))
+        selected = tuple(
+            references[(offset + index) % len(references)] for index in range(count)
+        )
+        for reference in selected:
+            try:
+                descriptor = os.open(
+                    proc / str(snapshot.identity.pid) / "map_files" / reference,
+                    os.O_RDONLY | getattr(os, "O_CLOEXEC", 0),
+                )
+            except (AttributeError, OSError):
+                continue
+            try:
+                add(_cached_descriptor(descriptor, cache))
+            finally:
+                os.close(descriptor)
+    else:
+        candidates: list[str] = []
+        seen: set[str] = set()
+        for raw in (getattr(snapshot, "exe_path", ""), *getattr(snapshot, "map_paths", ())):
+            if isinstance(raw, str) and raw.startswith("/") and raw not in seen:
+                seen.add(raw)
+                candidates.append(raw)
+        if candidates:
+            offset = start_index % len(candidates)
+            count = min(max_candidates, len(candidates))
+            for index in range(count):
+                path = Path(candidates[(offset + index) % len(candidates)])
+                if cache is None:
+                    values = _inspect_candidate(path)
+                else:
+                    try:
+                        key = _cache_key(path.stat())
+                    except OSError:
+                        continue
+                    if key in cache:
+                        values = cache[key]
+                    else:
+                        values = _inspect_candidate(path)
+                        cache[key] = values
+                add(values)
     return with_pytorch_pair(result)

src/lumi_eggcracker/observation.py

@@ -39,6 +39,16 @@ class ObservationStore:
                 self._values.pop(oldest, None)
         return value
 
+    def get(
+        self, identity: Hashable, *, now_ns: int | None = None
+    ) -> Observation | None:
+        self.expire(now_ns=now_ns)
+        return self._values.get(identity)
+
+    def identities(self, *, now_ns: int | None = None) -> frozenset[Hashable]:
+        self.expire(now_ns=now_ns)
+        return frozenset(self._values)
+
     def expire(self, *, now_ns: int | None = None) -> None:
         now = time.monotonic_ns() if now_ns is None else now_ns
         for key, value in tuple(self._values.items()):

src/lumi_eggcracker/supervisor.py

@@ -5,7 +5,6 @@ from __future__ import annotations
 import argparse
 import datetime as dt
 import errno
-import hashlib
 import json
 import os
 import re
@@ -27,7 +26,7 @@ from .approvals import approved, revoke
 from .approvals import create as create_approval
 from .approvals import load_all as load_approvals
 from .approvals import public as public_approval
-from .artifacts import ArtifactEvidence
+from .artifacts import MAX_FD_PROBES_PER_SCAN, ArtifactEvidence
 from .artifacts import from_snapshot as artifacts_from_snapshot
 from .containment import (
     capture_identity,
@@ -40,11 +39,12 @@ from .detectors import Catalogue, DetectionMatch, load_catalogue, match, public_
 from .discovery import (
     ProcessIdentity,
     ProcessSnapshot,
+    argv_digest,
     executable_digest_for_identity,
     executable_metadata_for_identity,
     scan,
 )
-from .elfmarkers import RuntimeEvidence, with_pytorch_pair
+from .elfmarkers import MAX_RUNTIME_CANDIDATES, RuntimeEvidence, with_pytorch_pair
 from .elfmarkers import from_snapshot as runtime_from_snapshot
 from .jsonio import JsonInputError, load_regular_json
 from .observation import ObservationStore
@@ -217,6 +217,10 @@ class Supervisor:
         ] = {}
         self.enforcement_slots = threading.BoundedSemaphore(MAX_ENFORCEMENT_TASKS)
         self.observations = ObservationStore()
+        self.artifact_fd_offsets: dict[ProcessIdentity, int] = {}
+        self.runtime_map_offsets: dict[ProcessIdentity, int] = {}
+        self.observed_content: dict[ProcessIdentity, dict[str, ArtifactEvidence]] = {}
+        self.observed_runtimes: dict[ProcessIdentity, dict[str, RuntimeEvidence]] = {}
         self.content_scan_tick = 0
         self.last_scan_completed_ns = 0
         self.last_scan_duration_ns = 0
@@ -423,7 +427,11 @@ class Supervisor:
         candidates: list[_EvidenceCandidate],
         snapshots: dict[ProcessIdentity, ProcessSnapshot],
     ) -> list[tuple[tuple[_EvidenceCandidate, ...], str]]:
-        """Union only small evidence sets joined by an exact live relation."""
+        """Union evidence sets joined by an exact live relation.
+
+        Component size is never interpreted as absence.  Callers reduce a
+        component to a minimal complete evidence set before enforcement.
+        """
         if not candidates:
             return []
         parent = list(range(len(candidates)))
@@ -452,8 +460,6 @@ class Supervisor:
             groups.setdefault(find(index), []).append(index)
         result: list[tuple[tuple[_EvidenceCandidate, ...], str]] = []
         for indexes in groups.values():
-            if len(indexes) > MAX_CORRELATED_PROCESSES:
-                continue
             values = tuple(candidates[index] for index in indexes)
             boundary_types = {
                 value
@@ -464,6 +470,97 @@ class Supervisor:
             result.append((values, boundary))
         return result
 
+    def _minimal_content_groups(
+        self,
+        candidates: list[_EvidenceCandidate],
+        snapshots: dict[ProcessIdentity, ProcessSnapshot],
+    ) -> list[tuple[tuple[_EvidenceCandidate, ...], str]]:
+        """Return one minimal complete evidence set per related component.
+
+        Repeated partial evidence does not revoke an independently complete
+        approval, and a component larger than the historical containment cap
+        cannot suppress a complete profile.  Only evidence identities that
+        contribute a new predicate are selected.
+        """
+        result: list[tuple[tuple[_EvidenceCandidate, ...], str]] = []
+        for component, boundary in self._correlate(candidates, snapshots):
+            selected: list[_EvidenceCandidate] = []
+            seen_content: set[str] = set()
+            seen_runtime: set[str] = set()
+            detected: DetectionMatch | None = None
+            for candidate in sorted(component, key=lambda item: item.snapshot.identity):
+                content_ids = {item.evidence_id for item in candidate.content}
+                runtime_ids = {item.evidence_id for item in candidate.runtimes}
+                if content_ids <= seen_content and runtime_ids <= seen_runtime:
+                    continue
+                selected.append(candidate)
+                seen_content.update(content_ids)
+                seen_runtime.update(runtime_ids)
+                paired = with_pytorch_pair(
+                    item for value in selected for item in value.runtimes
+                )
+                detected = match(
+                    self.catalogue,
+                    selected[0].snapshot,
+                    evidence={
+                        "MODEL_CONTENT": {
+                            item.evidence_id for value in selected for item in value.content
+                        },
+                        "MODEL_RUNTIME": {item.evidence_id for item in paired},
+                    },
+                )
+                if detected is not None:
+                    break
+                if len(selected) >= MAX_CORRELATED_PROCESSES:
+                    # There are currently only two independent evidence groups
+                    # and at most three raw identities are required for the
+                    # pinned profiles.  Reaching this guard means no bounded
+                    # minimal set was established; do not reinterpret partial
+                    # evidence as a destructive match.
+                    break
+            if detected is not None:
+                result.append((tuple(selected), boundary))
+        return result
+
+    def _content_groups(
+        self,
+        candidates: list[_EvidenceCandidate],
+        snapshots: dict[ProcessIdentity, ProcessSnapshot],
+    ) -> list[tuple[tuple[_EvidenceCandidate, ...], str]]:
+        """Prefer exact same-process matches before any relation aggregate."""
+        complete: list[_EvidenceCandidate] = []
+        partial: list[_EvidenceCandidate] = []
+        for item in candidates:
+            own_runtime = with_pytorch_pair(item.runtimes)
+            own_match = match(
+                self.catalogue,
+                item.snapshot,
+                evidence={
+                    "MODEL_CONTENT": {value.evidence_id for value in item.content},
+                    "MODEL_RUNTIME": {value.evidence_id for value in own_runtime},
+                },
+            )
+            if own_match is None:
+                partial.append(item)
+            else:
+                complete.append(
+                    _EvidenceCandidate(
+                        item.snapshot,
+                        item.content,
+                        tuple(own_runtime),
+                        item.first_seen_ns,
+                        own_match,
+                    )
+                )
+        return [((item,), "same-process") for item in complete] + self._minimal_content_groups(
+            partial, snapshots
+        )
+
+    @staticmethod
+    def _discovery_excluded(snapshot: ProcessSnapshot) -> bool:
+        """Exclude only the supervisor itself; a run record is not approval."""
+        return snapshot.identity.pid == os.getpid()
+
     def _cached_executable_digest(self, snapshot: ProcessSnapshot) -> str:
         metadata = executable_metadata_for_identity(snapshot.identity)
         key = (
@@ -571,7 +668,8 @@ class Supervisor:
             "executable": executable_record,
             "observed": {
                 "argv_count": len(snapshot.argv),
-                "argv_sha256": hashlib.sha256("\0".join(snapshot.argv).encode("utf-8")).hexdigest(),
+                "argv_sha256": argv_digest(snapshot.argv),
+                "argv_complete": snapshot.argv_complete,
                 "pid": snapshot.identity.pid,
                 "start_time": snapshot.identity.start_time,
                 "uid": snapshot.uid,
@@ -739,9 +837,20 @@ class Supervisor:
         except JsonInputError:
             approvals = []  # A corrupt approval never authorizes a matching workload.
         snapshots = scan(
-            exclude=lambda item: item.identity.pid == os.getpid() or self._managed(item)
+            exclude=self._discovery_excluded,
+            include_evidence=content_due,
         )
         snapshot_map = {item.identity: item for item in snapshots}
+        live_identities = set(snapshot_map)
+        for cache in (
+            getattr(self, "artifact_fd_offsets", {}),
+            getattr(self, "runtime_map_offsets", {}),
+            getattr(self, "observed_content", {}),
+            getattr(self, "observed_runtimes", {}),
+        ):
+            for identity in tuple(cache):
+                if identity not in live_identities:
+                    cache.pop(identity, None)
         candidates: list[_EvidenceCandidate] = []
         for snapshot in snapshots:
             fast_match = match(self.catalogue, snapshot)
@@ -749,33 +858,82 @@ class Supervisor:
             runtimes: tuple[RuntimeEvidence, ...] = ()
             first_seen_ns = time.monotonic_ns()
             if content_due:
-                content = artifacts_from_snapshot(snapshot, cache=self.artifact_cache)
+                artifact_offsets = getattr(self, "artifact_fd_offsets", None)
+                if artifact_offsets is None:
+                    self.artifact_fd_offsets = {}
+                    artifact_offsets = self.artifact_fd_offsets
+                runtime_offsets = getattr(self, "runtime_map_offsets", None)
+                if runtime_offsets is None:
+                    self.runtime_map_offsets = {}
+                    runtime_offsets = self.runtime_map_offsets
+                content_now = artifacts_from_snapshot(
+                    snapshot,
+                    cache=self.artifact_cache,
+                    fd_start_index=artifact_offsets.get(snapshot.identity, 0),
+                )
+                artifact_offsets[snapshot.identity] = (
+                    artifact_offsets.get(snapshot.identity, 0) + MAX_FD_PROBES_PER_SCAN
+                )
                 # Runtime evidence is intentionally collected independently so
                 # it can be correlated with content held by a bounded peer.
-                runtimes = runtime_from_snapshot(snapshot, cache=self.runtime_cache)
-                supplied = {
-                    "MODEL_CONTENT": {item.evidence_id for item in content},
-                    "MODEL_RUNTIME": {item.evidence_id for item in runtimes},
+                runtimes_now = runtime_from_snapshot(
+                    snapshot,
+                    cache=self.runtime_cache,
+                    start_index=runtime_offsets.get(snapshot.identity, 0),
+                )
+                runtime_offsets[snapshot.identity] = (
+                    runtime_offsets.get(snapshot.identity, 0) + MAX_RUNTIME_CANDIDATES
+                )
+                observed_content = getattr(self, "observed_content", None)
+                if observed_content is None:
+                    self.observed_content = {}
+                    observed_content = self.observed_content
+                observed_runtimes = getattr(self, "observed_runtimes", None)
+                if observed_runtimes is None:
+                    self.observed_runtimes = {}
+                    observed_runtimes = self.observed_runtimes
+                content_values = observed_content.setdefault(snapshot.identity, {})
+                runtime_values = observed_runtimes.setdefault(snapshot.identity, {})
+                content_values.update({item.evidence_id: item for item in content_now})
+                runtime_values.update({item.evidence_id: item for item in runtimes_now})
+                current_ids = {
+                    *(item.evidence_id for item in content_now),
+                    *(item.evidence_id for item in runtimes_now),
                 }
-                if content or runtimes:
+                if current_ids:
                     observation = self.observations.observe(
-                        snapshot.identity, set().union(*supplied.values())
+                        snapshot.identity, current_ids
                     )
                     first_seen_ns = observation.first_seen_ns
+                else:
+                    observation = self.observations.get(snapshot.identity)
+                if observation is not None:
+                    active = observation.evidence
+                    content = tuple(
+                        item for key, item in content_values.items() if key in active
+                    )
+                    runtimes = with_pytorch_pair(
+                        item for key, item in runtime_values.items() if key in active
+                    )
             if fast_match is None and not content and not runtimes:
                 continue
             candidates.append(
                 _EvidenceCandidate(snapshot, content, runtimes, first_seen_ns, fast_match)
             )
 
-        content_groups = self._correlate(
-            [item for item in candidates if (item.content or item.runtimes) and item.fast_match is None], snapshot_map
+        content_groups = self._content_groups(
+            [
+                item
+                for item in candidates
+                if (item.content or item.runtimes) and item.fast_match is None
+            ],
+            snapshot_map,
         )
-        groups = content_groups + [
+        groups = [
             ((item,), "same-process")
             for item in candidates
             if item.fast_match is not None
-        ]
+        ] + content_groups
         for group, boundary_type in groups:
             trigger_candidate = group[0]
             aggregate_content: list[ArtifactEvidence] = []
@@ -1313,7 +1471,20 @@ class Supervisor:
                 and "pids"
                 in Path("/sys/fs/cgroup/cgroup.controllers").read_text(encoding="ascii").split()
             )
-            ready = available and pidfd_available() and self.quarantine_root is not None
+            now_ns = time.monotonic_ns()
+            scan_healthy = (
+                self.last_scan_completed_ns > 0
+                and now_ns - self.last_scan_completed_ns <= SCAN_HEALTH_TIMEOUT_NS
+                and self.discovery_failures < MAX_DISCOVERY_FAILURES
+                and self.discovery_thread is not None
+                and self.discovery_thread.is_alive()
+            )
+            ready = (
+                available
+                and pidfd_available()
+                and self.quarantine_root is not None
+                and scan_healthy
+            )
             return {
                 "autonomous_discovery": ready,
                 "backend": "root-supervisor",
@@ -1323,6 +1494,7 @@ class Supervisor:
                     "consecutive_failures": self.discovery_failures,
                     "last_scan_duration_ms": self.last_scan_duration_ns / 1_000_000,
                     "last_scan_completed": self.last_scan_completed_ns > 0,
+                    "healthy": scan_healthy,
                     "scan_health_timeout_ms": SCAN_HEALTH_TIMEOUT_NS / 1_000_000,
                 },
                 "pidfd": pidfd_available(),

src/lumi_eggcracker/watchdog.py

@@ -90,12 +90,45 @@ def _kill_target(path: Path) -> dict[str, object]:
     _write(path, "cgroup.kill", b"1\n")
     deadline = time.monotonic() + 2.0
     while time.monotonic() < deadline:
-        if _events(path).get("populated") == "0":
-            return {"cgroup": str(path), "frozen": frozen, "populated": 0, "started_ns": before}
+        try:
+            empty = _events(path).get("populated") == "0"
+        except JsonInputError:
+            # systemd may collect an empty transient cgroup immediately after
+            # the authoritative kill write.  Collection after that write is
+            # equivalent to an empty proof, not a containment failure.
+            if not path.exists():
+                return {
+                    "cgroup": str(path),
+                    "collected": True,
+                    "frozen": frozen,
+                    "populated": 0,
+                    "started_ns": before,
+                }
+            raise
+        if empty:
+            return {
+                "cgroup": str(path),
+                "collected": False,
+                "frozen": frozen,
+                "populated": 0,
+                "started_ns": before,
+            }
         time.sleep(0.005)
-    value = _events(path)
+    try:
+        value = _events(path)
+    except JsonInputError:
+        if not path.exists():
+            return {
+                "cgroup": str(path),
+                "collected": True,
+                "frozen": frozen,
+                "populated": 0,
+                "started_ns": before,
+            }
+        raise
     return {
         "cgroup": str(path),
+        "collected": False,
         "frozen": frozen,
         "populated": int(value.get("populated", "1")),
         "started_ns": before,

tests/test_approvals.py

@@ -14,6 +14,7 @@ class Sample:
     uid: int
     exe_path: str
     argv: tuple[str, ...]
+    argv_complete: bool = True
 
 
 class ApprovalTests(unittest.TestCase):
@@ -28,6 +29,13 @@ class ApprovalTests(unittest.TestCase):
             self.assertTrue(approved(sample, executable_digest(binary), values))
             self.assertFalse(approved(Sample(1002, str(binary), sample.argv), executable_digest(binary), values))
             self.assertFalse(approved(Sample(1001, str(binary), (str(binary), "-m", "/models/other.gguf")), executable_digest(binary), values))
+            self.assertFalse(
+                approved(
+                    Sample(1001, str(binary), sample.argv, argv_complete=False),
+                    executable_digest(binary),
+                    values,
+                )
+            )
             self.assertNotIn("/models/qwen.gguf", record.values())
             self.assertEqual(argv_digest(sample.argv), record["argv_sha256"])
 

tests/test_artifacts.py

@@ -7,7 +7,8 @@ import tempfile
 import unittest
 from pathlib import Path
 
-from lumi_eggcracker.artifacts import from_snapshot, validate_path
+from lumi_eggcracker.artifacts import from_process_fds, from_snapshot, validate_path
+from lumi_eggcracker.discovery import ProcessIdentity
 
 
 def gguf(*, version: int = 3, tensors: int = 1, metadata: int = 0, padding: int = 64) -> bytes:
@@ -26,6 +27,23 @@ def safetensors(*, dtype: str = "F32", shape: list[int] | None = None, offsets:
 
 
 class ArtifactTests(unittest.TestCase):
+    def test_descriptor_windows_eventually_cover_late_model_fd(self) -> None:
+        with tempfile.TemporaryDirectory() as raw:
+            proc = Path(raw)
+            directory = proc / "42" / "fd"
+            directory.mkdir(parents=True)
+            for number in range(20):
+                (directory / str(number)).write_bytes(b"harmless")
+            (directory / "20").write_bytes(gguf())
+            sample = type("Snapshot", (), {"identity": ProcessIdentity(42, 100), "fd_entries": ()})()
+            self.assertEqual(
+                (), from_process_fds(sample, proc=proc, start_index=0, max_probes=16)
+            )
+            evidence = from_process_fds(
+                sample, proc=proc, start_index=16, max_probes=16
+            )
+            self.assertEqual({"gguf-v3"}, {item.evidence_id for item in evidence})
+
     def test_valid_gguf_content_does_not_depend_on_filename(self) -> None:
         with tempfile.TemporaryDirectory() as raw:
             path = Path(raw) / "opaque-input"

tests/test_discovery.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import tempfile
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 
 from lumi_eggcracker.discovery import (
     ProcessIdentity,
@@ -11,11 +12,25 @@ from lumi_eggcracker.discovery import (
     executable_digest_for_identity,
     identity,
     parse_stat,
+    snapshot,
 )
 from lumi_eggcracker.jsonio import JsonInputError
 
 
 class DiscoveryTests(unittest.TestCase):
+    def _proc_entry(self, proc: Path, command: bytes) -> ProcessIdentity:
+        entry = proc / "42"
+        entry.mkdir()
+        (entry / "stat").write_text(
+            "42 (runner) S 1 " + "0 " * 17 + "100\n", encoding="utf-8"
+        )
+        (entry / "cgroup").write_text("0::/user.slice/test\n", encoding="utf-8")
+        (entry / "status").write_text("Uid:\t1000\t1000\t1000\t1000\n", encoding="utf-8")
+        (entry / "cmdline").write_bytes(command)
+        (entry / "fd").mkdir()
+        (entry / "map_files").mkdir()
+        return ProcessIdentity(42, 100)
+
     def test_stat_parser_handles_spaces_in_command_name(self) -> None:
         raw = "77 (worker with spaces) S 4 " + "0 " * 17 + "901\n"
         self.assertEqual((4, 901), parse_stat(raw))
@@ -52,6 +67,40 @@ class DiscoveryTests(unittest.TestCase):
             self.assertEqual(512, len(values))
             self.assertTrue(values[0].startswith("/opt/runtime/"))
 
+    def test_oversized_maps_file_is_localized_to_one_empty_fallback(self) -> None:
+        with tempfile.TemporaryDirectory() as raw:
+            path = Path(raw) / "maps"
+            path.write_bytes(b"x" * 64)
+            with patch("lumi_eggcracker.discovery.MAX_MAP_BYTES", 16):
+                self.assertEqual((), _map_paths(path))
+
+    def test_invalid_utf8_argv_does_not_erase_process_snapshot(self) -> None:
+        with tempfile.TemporaryDirectory() as raw:
+            proc = Path(raw)
+            value = self._proc_entry(proc, b"/usr/bin/python3\0bad\xffarg\0")
+            with patch(
+                "lumi_eggcracker.discovery.os.readlink", return_value="/usr/bin/python3"
+            ):
+                current = snapshot(value, proc=proc)
+            self.assertIsNotNone(current)
+            assert current is not None
+            self.assertTrue(current.argv_complete)
+            self.assertEqual(2, len(current.argv))
+            self.assertEqual(64, len(argv_digest(current.argv)))
+
+    def test_oversized_argv_is_unapprovable_not_invisible(self) -> None:
+        with tempfile.TemporaryDirectory() as raw:
+            proc = Path(raw)
+            value = self._proc_entry(proc, b"/usr/bin/python3\0" + b"x" * 64)
+            with patch("lumi_eggcracker.discovery.MAX_CMDLINE_BYTES", 16), patch(
+                "lumi_eggcracker.discovery.os.readlink", return_value="/usr/bin/python3"
+            ):
+                current = snapshot(value, proc=proc)
+            self.assertIsNotNone(current)
+            assert current is not None
+            self.assertFalse(current.argv_complete)
+            self.assertEqual((), current.argv)
+
     def test_process_executable_is_hashed_from_bound_proc_identity(self) -> None:
         with tempfile.TemporaryDirectory() as raw:
             proc = Path(raw)

tests/test_elfmarkers.py

@@ -7,6 +7,7 @@ import unittest
 from pathlib import Path
 from unittest.mock import patch
 
+from lumi_eggcracker.discovery import ProcessIdentity
 from lumi_eggcracker.elfmarkers import (
     PINNED_LLAMA_BUILD_IDS,
     PYTORCH_ATEN_EVIDENCE_ID,
@@ -44,7 +45,55 @@ def elf(*, markers: tuple[str, ...], append_decoys: bool = False) -> bytes:
     return bytes(body)
 
 
+def build_id_elf(identifier: bytes) -> bytes:
+    note = struct.pack("<III", 4, len(identifier), 3) + b"GNU\0" + identifier
+    note += b"\0" * ((4 - len(identifier) % 4) % 4)
+    body = bytearray(128 + len(note))
+    body[:16] = b"\x7fELF\x02\x01\x01" + b"\0" * 9
+    body[16:64] = struct.pack(
+        "<HHIQQQIHHHHHH", 3, 62, 1, 0, 64, 0, 0, 64, 56, 1, 0, 0, 0
+    )
+    body[64:120] = struct.pack("<IIQQQQQQ", 4, 0, 128, 0, 0, len(note), 0, 4)
+    body[128:] = note
+    return bytes(body)
+
+
 class ElfMarkerTests(unittest.TestCase):
+    def test_proc_mapping_descriptors_are_inspected_without_pathnames(self) -> None:
+        bridge_id = bytes.fromhex("44" * 20)
+        aten_id = bytes.fromhex("55" * 20)
+        with tempfile.TemporaryDirectory() as raw:
+            proc = Path(raw)
+            mappings = proc / "42" / "map_files"
+            mappings.mkdir(parents=True)
+            (mappings / "1000-2000").write_bytes(build_id_elf(bridge_id))
+            (mappings / "3000-4000").write_bytes(build_id_elf(aten_id))
+            sample = type(
+                "Snapshot",
+                (),
+                {
+                    "identity": ProcessIdentity(42, 100),
+                    "exe_path": "/deleted-runtime",
+                    "map_paths": (),
+                },
+            )()
+            with patch(
+                "lumi_eggcracker.elfmarkers.PINNED_PYTORCH_BRIDGE_BUILD_IDS",
+                {bridge_id.hex()},
+            ), patch(
+                "lumi_eggcracker.elfmarkers.PINNED_PYTORCH_ATEN_BUILD_IDS",
+                {aten_id.hex()},
+            ):
+                evidence = from_snapshot(sample, proc=proc, max_candidates=2)
+            self.assertEqual(
+                {
+                    PYTORCH_BRIDGE_EVIDENCE_ID,
+                    PYTORCH_ATEN_EVIDENCE_ID,
+                    PYTORCH_PAIR_EVIDENCE_ID,
+                },
+                {item.evidence_id for item in evidence},
+            )
+
     def test_two_valid_symbol_markers_qualify_regardless_of_filename(self) -> None:
         with tempfile.TemporaryDirectory() as raw:
             path = Path(raw) / "unrelated-name"

tests/test_supervisor.py

@@ -67,6 +67,65 @@ class SupervisorTests(unittest.TestCase):
             (),
         )
         self.assertTrue(supervisor._managed(snapshot))
+        self.assertFalse(supervisor._discovery_excluded(snapshot))
+
+    def test_complete_identity_is_not_suppressed_or_broadened_by_64_partials(self) -> None:
+        supervisor = self._instance()
+        supervisor.catalogue = load_bundled()
+        parent = ProcessIdentity(10, 100)
+        parent_snapshot = ProcessSnapshot(
+            parent, 2001, "/usr/bin/worker", "worker", ("worker",), (), (), ()
+        )
+        content = ArtifactEvidence(
+            "safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64
+        )
+        runtime = RuntimeEvidence(
+            "pytorch-bridge-aten-pair-pinned-cpu",
+            "PyTorch/ATen",
+            "BUILD_ID_PAIR",
+            (),
+        )
+        snapshots = {parent: parent_snapshot}
+        candidates: list[_EvidenceCandidate] = []
+        for offset in range(64):
+            identity = ProcessIdentity(100 + offset, 1000 + offset)
+            current = ProcessSnapshot(
+                identity,
+                2001,
+                "/usr/bin/python3",
+                "python3",
+                ("python3",),
+                (),
+                (),
+                (),
+                parent=parent,
+            )
+            snapshots[identity] = current
+            candidates.append(_EvidenceCandidate(current, (content,), (), 1))
+        complete_identity = ProcessIdentity(999, 1999)
+        complete_snapshot = ProcessSnapshot(
+            complete_identity,
+            2001,
+            "/usr/bin/python3",
+            "python3",
+            ("python3",),
+            (),
+            (),
+            (),
+            parent=parent,
+        )
+        snapshots[complete_identity] = complete_snapshot
+        candidates.append(
+            _EvidenceCandidate(complete_snapshot, (content,), (runtime,), 2)
+        )
+
+        correlated = supervisor._correlate(candidates, snapshots)
+        self.assertEqual(1, len(correlated))
+        self.assertEqual(65, len(correlated[0][0]))
+
+        groups = supervisor._content_groups(candidates, snapshots)
+        self.assertEqual(1, len(groups))
+        self.assertEqual((complete_identity,), tuple(item.snapshot.identity for item in groups[0][0]))
 
     def test_heartbeat_stops_when_no_scan_completed_within_health_bound(self) -> None:
         supervisor = self._instance()

tests/test_watchdog.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from lumi_eggcracker.jsonio import JsonInputError
+from lumi_eggcracker.watchdog import _kill_target
+
+
+class WatchdogTests(unittest.TestCase):
+    def test_collection_after_kill_write_is_successful_empty_containment(self) -> None:
+        path = Path("/definitely-absent-eggcracker-cgroup")
+        with patch("lumi_eggcracker.watchdog._write") as write, patch(
+            "lumi_eggcracker.watchdog._events",
+            side_effect=JsonInputError("cannot inspect watchdog cgroup"),
+        ):
+            result = _kill_target(path)
+        write.assert_called_once_with(path, "cgroup.kill", b"1\n")
+        self.assertEqual(0, result["populated"])
+        self.assertTrue(result["collected"])
+