diff --git a/Discovery/adr_discovery/judge/__init__.py b/Discovery/adr_discovery/judge/__init__.py
new file mode 100644
index 0000000..84069d6
--- /dev/null
+++ b/Discovery/adr_discovery/judge/__init__.py
@@ -0,0 +1,66 @@
+"""M6 -- turns an inventory into something a policy can act on.
+
+Its currency is operator trust, and it spends that on every finding it
+raises. Ambiguity resolves toward the safe verdict: a shape the parser
+cannot read confidently produces no finding and a recorded ambiguity,
+rather than a guess dressed as a verdict.
+"""
+
+from __future__ import annotations
+
+from dataclasses import replace
+
+from ..contracts.records import Asset, Declaration, Finding, Risk
+from .findings import findings_for
+from .risk import credential_reach, pinning, transport_of, unattended
+from .sanction import EMPTY, Policy, state_for
+
+__all__ = ["judge", "Policy", "EMPTY"]
+
+
+def judge(assets: tuple[Asset, ...], declarations: dict[str, Declaration],
+          policy: Policy = EMPTY, ledger=None) -> tuple[tuple[Asset, ...], tuple[Finding, ...]]:
+    """Attach risk facts and sanction state, then raise what is worth raising."""
+    judged: list[Asset] = []
+    findings: list[Finding] = []
+
+    for asset in assets:
+        declaration = declarations.get(asset.asset_id)
+        risk = _risk_for(asset, declaration, ledger)
+        with_risk = replace(asset, risk=risk, sanction=state_for(asset.catalog_id, policy))
+        judged.append(with_risk)
+        findings.extend(findings_for(with_risk, policy))
+
+    return tuple(judged), tuple(findings)
+
+
+def _risk_for(asset: Asset, declaration: Declaration | None, ledger) -> Risk:
+    if declaration is None:
+        # Nothing new to say. Keep whatever earlier stages established
+        # rather than overwriting it with an empty verdict.
+        return asset.risk
+
+    pinned, factors = pinning(declaration.command, declaration.args)
+    # A shape we could not read confidently: record the ambiguity and raise
+    # nothing, which is the safe verdict.
+    if pinned is None and declaration.command and not declaration.url and ledger is not None:
+        ledger.probe("judge", "degraded", f"unreadable launch shape: {declaration.name}")
+
+    env_declared = bool(declaration.env_names) or "env" in declaration.raw
+    env_names, kinds = credential_reach(declaration.env_names, env_declared)
+    transport, transport_factors = transport_of(declaration.raw, declaration.url)
+
+    destinations = ()
+    if declaration.url:
+        host = declaration.url.split("//", 1)[-1].split("/", 1)[0].split(":", 1)[0]
+        destinations = (host,)
+
+    return Risk(
+        pinned=pinned,
+        factors=tuple(factors) + transport_factors,
+        credential_kinds=kinds,
+        env_names=env_names,
+        transport=transport,
+        destinations=destinations,
+        unattended=asset.risk.unattended or unattended(declaration.args),
+    )
diff --git a/Discovery/adr_discovery/judge/findings.py b/Discovery/adr_discovery/judge/findings.py
new file mode 100644
index 0000000..9e08ba2
--- /dev/null
+++ b/Discovery/adr_discovery/judge/findings.py
@@ -0,0 +1,97 @@
+"""The findings worth surfacing.
+
+Findings are the narrow end on purpose. Every one an operator dismisses
+costs the ones that follow it, so the gate below is deliberately strict and
+the strictness is recovered by correlation elsewhere: a server a config
+declares is recognized even when its name says nothing.
+
+Two lessons already paid for, both caught on real machines and neither by
+the suite: a loose runtime heuristic put a high-severity finding on
+`npx eslint .`, and a command that merely mentioned a path containing "mcp"
+became an undeclared server. Hence the first rule below.
+"""
+
+from __future__ import annotations
+
+from ..contracts.evidence import Channel, Evidence
+from ..contracts.records import Asset, Finding, Kind
+from .sanction import Policy, is_third_party
+
+
+#: A finding may only be raised about an asset whose identity was
+#: established by evidence -- never about one inferred from a command line.
+#: This is the precision gate, and it is what `npx eslint .` fails.
+def _may_be_judged(asset: Asset) -> bool:
+    if asset.kind is Kind.MCP_SERVER:
+        # Declared in a config, or correlated with one. A bare process whose
+        # argv merely resembles a server does not reach this branch.
+        return bool(asset.catalog_id) or "declared" in asset.flags or asset.confidence.label != "none"
+    return bool(asset.catalog_id)
+
+
+def findings_for(asset: Asset, policy: Policy) -> tuple[Finding, ...]:
+    if not _may_be_judged(asset):
+        return ()
+
+    out: list[Finding] = []
+
+    def ev(proof: str) -> tuple[Evidence, ...]:
+        return (Evidence("judge", Channel.CONFIG, asset.install_path or "", proof, 0.9),)
+
+    if asset.kind is Kind.MCP_SERVER and asset.risk.pinned is False:
+        out.append(
+            Finding(
+                rule="unpinned_mcp_server",
+                severity="medium",
+                asset_id=asset.asset_id,
+                summary=f"{asset.name} resolves its package at launch time",
+                evidence=ev("no version in the resolved operand"),
+            )
+        )
+
+    if asset.kind is Kind.MCP_SERVER and "undeclared" in asset.flags:
+        out.append(
+            Finding(
+                rule="undeclared_mcp_server",
+                severity="high",
+                asset_id=asset.asset_id,
+                summary=f"{asset.name} is running but no configuration declares it",
+                evidence=ev("running server correlated with no declaration"),
+            )
+        )
+
+    if "plaintext_transport" in asset.risk.factors:
+        out.append(
+            Finding(
+                rule="plaintext_transport",
+                severity="medium",
+                asset_id=asset.asset_id,
+                summary=f"{asset.name} is reached over http://",
+                evidence=ev("declared url uses a plaintext scheme"),
+            )
+        )
+
+    if asset.risk.unattended:
+        out.append(
+            Finding(
+                rule="unattended_execution",
+                severity="high",
+                asset_id=asset.asset_id,
+                summary=f"{asset.name} launches with permission checks bypassed",
+                evidence=ev("bypass flag present in the agent's own launch"),
+            )
+        )
+
+    third_party = [d for d in asset.risk.destinations if is_third_party(d, policy)]
+    if third_party and policy.tenant_domains:
+        out.append(
+            Finding(
+                rule="third_party_destination",
+                severity="medium",
+                asset_id=asset.asset_id,
+                summary=f"{asset.name} reaches {', '.join(sorted(third_party))}",
+                evidence=ev("destination outside the tenant domain list"),
+            )
+        )
+
+    return tuple(out)
diff --git a/Discovery/adr_discovery/judge/risk.py b/Discovery/adr_discovery/judge/risk.py
new file mode 100644
index 0000000..d61cee6
--- /dev/null
+++ b/Discovery/adr_discovery/judge/risk.py
@@ -0,0 +1,129 @@
+"""Risk facts per asset.
+
+Pinning is the highest-yield verdict in the module and the easiest to get
+wrong, because it has to read the *specification* rather than look for an
+`@`, and parse options before choosing an operand. A volume mount is not an
+image; a registry URL is not a package.
+"""
+
+from __future__ import annotations
+
+import re
+from types import MappingProxyType
+
+from ..redact import rules as redact
+
+#: runner -> flags that consume the next argument. Anything not listed and
+#: starting with '-' is a switch; the first bare word after the options is
+#: the operand. This table is the whole difference between reading a
+#: specification and pattern-matching one.
+VALUE_FLAGS = MappingProxyType({
+    "npx": frozenset({"--package", "-p", "--call", "-c", "--node-options"}),
+    "bunx": frozenset({"--package"}),
+    "uvx": frozenset({"--from", "--with", "--python", "-p", "--index-url"}),
+    "pipx": frozenset({"--spec", "--python", "--index-url", "--pip-args"}),
+    "pip": frozenset({"--index-url", "-i", "--extra-index-url", "--find-links", "-f",
+                      "--target", "-t", "--requirement", "-r"}),
+    "docker": frozenset({"-v", "--volume", "-e", "--env", "-p", "--publish", "--name",
+                         "--network", "--mount", "-w", "--workdir", "--user", "-u",
+                         "--entrypoint", "--label", "-l"}),
+    "podman": frozenset({"-v", "--volume", "-e", "--env", "-p", "--publish", "--name"}),
+})
+
+#: Subcommands that must be skipped before the operand is read.
+SUBCOMMANDS = MappingProxyType({
+    "docker": frozenset({"run", "create", "start", "exec"}),
+    "podman": frozenset({"run", "create"}),
+    "pip": frozenset({"install"}),
+    "pipx": frozenset({"run", "install"}),
+    "uv": frozenset({"tool", "run", "pip"}),
+})
+
+#: Runners that resolve a package at launch time -- the shapes where a
+#: missing version means "whatever upstream published this morning".
+EPHEMERAL_RUNNERS = frozenset({"npx", "bunx", "uvx", "pipx", "pip"})
+CONTAINER_RUNNERS = frozenset({"docker", "podman"})
+
+_NPM_PINNED = re.compile(r"^(@[^/]+/)?[^@/]+@[^@]+$")
+_DIGEST = re.compile(r"@sha256:[0-9a-f]{8,}$")
+_TAGGED = re.compile(r":[^/:]+$")
+
+
+def operand_of(command: str | None, args: tuple[str, ...]) -> tuple[str | None, str]:
+    """The thing being run, having parsed the options first."""
+    runner = (command or "").rsplit("/", 1)[-1]
+    value_flags = VALUE_FLAGS.get(runner, frozenset())
+    subcommands = SUBCOMMANDS.get(runner, frozenset())
+
+    i = 0
+    seen_subcommand = not subcommands
+    while i < len(args):
+        arg = args[i]
+        if arg.startswith("-"):
+            flag = arg.split("=", 1)[0]
+            if flag in value_flags and "=" not in arg:
+                i += 2
+                continue
+            i += 1
+            continue
+        if not seen_subcommand and arg in subcommands:
+            seen_subcommand = True
+            i += 1
+            continue
+        return arg, runner
+    return None, runner
+
+
+def pinning(command: str | None, args: tuple[str, ...]) -> tuple[bool | None, tuple[str, ...]]:
+    """(pinned, factors). `None` means the shape carries no pinning question."""
+    operand, runner = operand_of(command, args)
+    if operand is None:
+        return None, ()
+
+    if runner in CONTAINER_RUNNERS:
+        if _DIGEST.search(operand):
+            return True, ()
+        if _TAGGED.search(operand) and not operand.endswith(":latest"):
+            return False, ("unpinned_supply_chain",)
+        return False, ("unpinned_supply_chain",)
+
+    if runner in EPHEMERAL_RUNNERS:
+        if _NPM_PINNED.match(operand) or "==" in operand or operand.count("@") >= 1 and not operand.startswith("@"):
+            return True, ()
+        return False, ("unpinned_supply_chain",)
+
+    return None, ()
+
+
+def credential_reach(env_names: tuple[str, ...], env_declared: bool) -> tuple[tuple[str, ...], tuple[str, ...]]:
+    """Which secrets could be touched, by name and kind, never by value.
+
+    The absence of an env block means the process inherits the parent
+    environment -- which is *all* of them, not none. Reading absence as
+    "no credentials" is the safe-looking answer and the wrong one.
+    """
+    if not env_declared:
+        return ("<inherits parent environment>",), ("inherited",)
+    return env_names, redact.credential_kinds(env_names)
+
+
+def unattended(argv: tuple[str, ...]) -> bool:
+    """The flag comes from the agent's own launch, never from the scheduler
+    that started it: `-p` with `--dangerously-skip-permissions` is the
+    finding whether cron, a timer or a person typed it."""
+    flags = set(argv)
+    headless = bool(flags & {"-p", "--print", "--headless", "--non-interactive", "--yes", "-y"})
+    bypass = bool(
+        flags & {"--dangerously-skip-permissions", "--yolo", "--auto-approve", "--no-confirm"}
+    )
+    return headless and bypass or bypass
+
+
+def transport_of(declaration_raw: dict, url: str | None) -> tuple[str | None, tuple[str, ...]]:
+    transport = str(declaration_raw.get("transport") or ("http" if url else "stdio"))
+    factors: list[str] = []
+    if url and url.startswith("http://"):
+        factors.append("plaintext_transport")
+    if transport == "sse":
+        factors.append("deprecated_transport")
+    return transport, tuple(factors)
diff --git a/Discovery/adr_discovery/judge/sanction.py b/Discovery/adr_discovery/judge/sanction.py
new file mode 100644
index 0000000..1343662
--- /dev/null
+++ b/Discovery/adr_discovery/judge/sanction.py
@@ -0,0 +1,57 @@
+"""Sanction state, from tenant policy.
+
+Tenant-specific values come from configuration, never from code. The test
+for this is to run one world against two policies and require two verdicts
+-- a value reachable from code alone cannot pass it.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class Policy:
+    approved: frozenset[str] = frozenset()
+    forbidden: frozenset[str] = frozenset()
+    #: Domains the tenant owns. A destination outside this set is third
+    #: party; there is no default, because a default would be somebody's
+    #: wrong answer shipped as a constant.
+    tenant_domains: frozenset[str] = frozenset()
+
+    @staticmethod
+    def from_dict(raw: dict) -> "Policy":
+        if not isinstance(raw, dict):
+            raise ValueError("policy root must be an object")
+
+        def string_set(field: str) -> frozenset[str]:
+            value = raw.get(field) or ()
+            if isinstance(value, str) or not isinstance(value, (list, tuple, set, frozenset)):
+                raise ValueError(f"policy {field!r} must be an array of strings")
+            if not all(isinstance(item, str) and item for item in value):
+                raise ValueError(f"policy {field!r} must contain non-empty strings")
+            return frozenset(value)
+
+        return Policy(
+            approved=string_set("approved"),
+            forbidden=string_set("forbidden"),
+            tenant_domains=frozenset(domain.lower().rstrip(".") for domain in string_set("tenant_domains")),
+        )
+
+
+EMPTY = Policy()
+
+
+def state_for(catalog_id: str | None, policy: Policy) -> str:
+    if catalog_id is None:
+        return "unknown"
+    if catalog_id in policy.forbidden:
+        return "forbidden"
+    if catalog_id in policy.approved:
+        return "approved"
+    return "unsanctioned" if policy.approved else "unknown"
+
+
+def is_third_party(host: str, policy: Policy) -> bool:
+    h = host.lower().rstrip(".")
+    return not any(h == d or h.endswith("." + d) for d in policy.tenant_domains)
diff --git a/Discovery/adr_discovery/reporter/__init__.py b/Discovery/adr_discovery/reporter/__init__.py
new file mode 100644
index 0000000..bb13682
--- /dev/null
+++ b/Discovery/adr_discovery/reporter/__init__.py
@@ -0,0 +1,8 @@
+"""M7 -- the snapshot, and the difference between this one and the last."""
+
+from __future__ import annotations
+
+from .delta import Delta, DifferentEndpoints, diff
+from .snapshot import from_dict, stats, to_dict, to_json
+
+__all__ = ["diff", "Delta", "DifferentEndpoints", "from_dict", "to_dict", "to_json", "stats"]
diff --git a/Discovery/adr_discovery/reporter/delta.py b/Discovery/adr_discovery/reporter/delta.py
new file mode 100644
index 0000000..1865a0a
--- /dev/null
+++ b/Discovery/adr_discovery/reporter/delta.py
@@ -0,0 +1,120 @@
+"""What the delta says.
+
+    appeared · disappeared · version_changed · config_changed · reinstalled
+  + risk_delta       pinned -> floating is a silent regression otherwise
+  + coverage_delta   a surface that became unreadable is a change too
+
+Fleet fan-out -- the same new asset landing on many endpoints at once -- is
+computable only centrally and is deliberately absent from this module.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from ..contracts.snapshot import Snapshot
+from .identity import by_id, by_identity
+
+
+class DifferentEndpoints(ValueError):
+    """A diff compares one endpoint with itself unless told otherwise.
+
+    Diffing two machines produces a fictional delta in which every asset on
+    each looks like a change, and nothing in the output says so.
+    """
+
+
+@dataclass(frozen=True, slots=True)
+class Change:
+    kind: str
+    asset_id: str
+    name: str
+    detail: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class Delta:
+    endpoint: str
+    changes: tuple[Change, ...] = ()
+    coverage_delta: tuple[str, ...] = ()
+
+    def of(self, kind: str) -> tuple[Change, ...]:
+        return tuple(c for c in self.changes if c.kind == kind)
+
+    @property
+    def is_empty(self) -> bool:
+        return not self.changes and not self.coverage_delta
+
+
+def diff(before: Snapshot, after: Snapshot, allow_cross_endpoint: bool = False) -> Delta:
+    if before.hostname != after.hostname and not allow_cross_endpoint:
+        raise DifferentEndpoints(
+            f"refusing to diff {before.hostname!r} against {after.hostname!r}; "
+            "pass allow_cross_endpoint=True if that is genuinely what you want"
+        )
+
+    old_by_id, new_by_id = by_id(before.assets), by_id(after.assets)
+    old_by_identity, new_by_identity = by_identity(before.assets), by_identity(after.assets)
+    changes: list[Change] = []
+
+    for asset_id, new in new_by_id.items():
+        old = old_by_id.get(asset_id)
+        if old is None:
+            key = (new.kind.value, new.identity, new.owner)
+            previous = old_by_identity.get(key)
+            if previous is not None and previous.asset_id not in new_by_id:
+                changes.append(
+                    Change("reinstalled", new.asset_id, new.name,
+                           f"{previous.install_root} -> {new.install_root}")
+                )
+            else:
+                changes.append(Change("appeared", new.asset_id, new.name))
+            continue
+
+        if old.version != new.version:
+            changes.append(
+                Change("version_changed", asset_id, new.name, f"{old.version} -> {new.version}")
+            )
+        if _config_of(old) != _config_of(new):
+            changes.append(Change("config_changed", asset_id, new.name))
+        # An asset set that did not move can still carry a regression.
+        if old.risk.pinned is True and new.risk.pinned is False:
+            changes.append(
+                Change("risk_delta", asset_id, new.name, "pinned -> floating")
+            )
+        elif set(new.risk.factors) - set(old.risk.factors):
+            changes.append(
+                Change("risk_delta", asset_id, new.name,
+                       "+" + ",".join(sorted(set(new.risk.factors) - set(old.risk.factors))))
+            )
+
+    for asset_id, old in old_by_id.items():
+        if asset_id in new_by_id:
+            continue
+        key = (old.kind.value, old.identity, old.owner)
+        if key in new_by_identity and new_by_identity[key].asset_id not in old_by_id:
+            continue  # already reported as a reinstall
+        changes.append(Change("disappeared", asset_id, old.name))
+
+    return Delta(after.hostname, tuple(changes), _coverage_delta(before, after))
+
+
+def _config_of(asset) -> tuple:
+    return (asset.install_path, asset.risk.transport, tuple(sorted(asset.risk.env_names)))
+
+
+def _coverage_delta(before: Snapshot, after: Snapshot) -> tuple[str, ...]:
+    """A surface that became unreadable is a change, and a silent one."""
+    out: list[str] = []
+    old_denied = {d.path for d in before.coverage.denied}
+    new_denied = {d.path for d in after.coverage.denied}
+    for path in sorted(new_denied - old_denied):
+        out.append(f"became unreadable: {path}")
+    for path in sorted(old_denied - new_denied):
+        out.append(f"became readable: {path}")
+
+    old_gone = {u.provider for u in before.coverage.unavailable}
+    new_gone = {u.provider for u in after.coverage.unavailable}
+    for provider in sorted(new_gone - old_gone):
+        out.append(f"provider became unavailable: {provider}")
+    return tuple(out)
diff --git a/Discovery/adr_discovery/reporter/identity.py b/Discovery/adr_discovery/reporter/identity.py
new file mode 100644
index 0000000..16a1640
--- /dev/null
+++ b/Discovery/adr_discovery/reporter/identity.py
@@ -0,0 +1,24 @@
+"""Matching assets across two snapshots.
+
+The delta depends entirely on `asset_id` holding still under change that is
+not a change. It must survive a version upgrade, a content-addressed store
+rebuild and a credential rotation -- the last of which requires that no
+secret material reach the hash, which is why the id is computed inside M5
+rather than assembled by its callers.
+
+`reinstalled` is the one case where the id legitimately moves: same
+identity and owner, different install root.
+"""
+
+from __future__ import annotations
+
+from ..contracts.records import Asset
+
+
+def by_id(assets: tuple[Asset, ...]) -> dict[str, Asset]:
+    return {a.asset_id: a for a in assets}
+
+
+def by_identity(assets: tuple[Asset, ...]) -> dict[tuple[str, str, str], Asset]:
+    """A weaker key, used only to tell a reinstall from a disappearance."""
+    return {(a.kind.value, a.identity, a.owner): a for a in assets}
diff --git a/Discovery/adr_discovery/reporter/snapshot.py b/Discovery/adr_discovery/reporter/snapshot.py
new file mode 100644
index 0000000..d040f79
--- /dev/null
+++ b/Discovery/adr_discovery/reporter/snapshot.py
@@ -0,0 +1,143 @@
+"""Serialization.
+
+A snapshot is emitted even when nothing is found: to fleet coverage, a host
+that reported an empty inventory and a host that never reported are very
+different facts, and only one of them is evidence.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import json
+from enum import Enum
+
+from ..contracts.evidence import Band, Channel, Evidence, Rung
+from ..contracts.records import Asset, Finding, Kind, Liveness, ReviewItem, Risk
+from ..contracts.snapshot import (
+    SCHEMA_VERSION,
+    BoundaryHit,
+    Coverage,
+    Denied,
+    ProbeRun,
+    RootSwept,
+    Snapshot,
+    Truncated,
+    Unavailable,
+)
+
+
+def to_dict(snapshot: Snapshot) -> dict:
+    return _encode(snapshot)
+
+
+def to_json(snapshot: Snapshot, indent: int | None = 2) -> str:
+    return json.dumps(to_dict(snapshot), indent=indent, sort_keys=True)
+
+
+def _encode(value):
+    if dataclasses.is_dataclass(value) and not isinstance(value, type):
+        return {f.name: _encode(getattr(value, f.name)) for f in dataclasses.fields(value)}
+    if isinstance(value, Enum):
+        return value.value
+    if isinstance(value, dict):
+        return {str(k): _encode(v) for k, v in value.items()}
+    if isinstance(value, (list, tuple, set, frozenset)):
+        return [_encode(v) for v in value]
+    return value
+
+
+def from_dict(document: dict) -> Snapshot:
+    """Rebuild a snapshot from its serialized form.
+
+    Needed because the delta is the product: comparing this scan with the
+    last one means reading the last one back, and a snapshot that can only
+    be written is a snapshot that can only be filed away.
+    """
+    return Snapshot(
+        hostname=document.get("hostname", "unknown"),
+        username=document.get("username", "unknown"),
+        platform=document.get("platform", "unknown"),
+        timestamp=document.get("timestamp", ""),
+        assets=tuple(_asset(a) for a in document.get("assets", ())),
+        findings=tuple(_finding(f) for f in document.get("findings", ())),
+        review_queue=tuple(
+            ReviewItem(path=r.get("path", ""), score=r.get("score", 0.0),
+                       signals=tuple(r.get("signals", ())))
+            for r in document.get("review_queue", ())
+        ),
+        coverage=_coverage(document.get("coverage") or {}),
+        catalog_version=document.get("catalog_version", "unknown"),
+        schema_version=document.get("schema_version", SCHEMA_VERSION),
+    )
+
+
+def _asset(row: dict) -> Asset:
+    risk = row.get("risk") or {}
+    band = row.get("confidence") or {}
+    return Asset(
+        asset_id=row["asset_id"],
+        kind=Kind(row["kind"]),
+        name=row.get("name", ""),
+        identity=row.get("identity", ""),
+        catalog_id=row.get("catalog_id"),
+        vendor=row.get("vendor"),
+        version=row.get("version"),
+        install_path=row.get("install_path"),
+        install_root=row.get("install_root"),
+        install_method=row.get("install_method"),
+        owner=row.get("owner", "system"),
+        location=row.get("location", "local"),
+        liveness=Liveness(row.get("liveness", "installed")),
+        last_used=row.get("last_used"),
+        confidence=Band(band.get("label", "none"),
+                        tuple(Channel(c) for c in band.get("channels", ()))),
+        verification=tuple(Rung(r) for r in row.get("verification", ())),
+        evidence=tuple(
+            Evidence(e.get("stage", ""), Channel(e["channel"]), e.get("path", ""),
+                     e.get("proof", ""), e.get("confidence", 0.0),
+                     Rung(e["rung"]) if e.get("rung") else None)
+            for e in row.get("evidence", ())
+        ),
+        risk=Risk(
+            pinned=risk.get("pinned"),
+            factors=tuple(risk.get("factors", ())),
+            credential_kinds=tuple(risk.get("credential_kinds", ())),
+            env_names=tuple(risk.get("env_names", ())),
+            transport=risk.get("transport"),
+            destinations=tuple(risk.get("destinations", ())),
+            unattended=risk.get("unattended", False),
+        ),
+        sanction=row.get("sanction", "unknown"),
+        flags=tuple(row.get("flags", ())),
+    )
+
+
+def _finding(row: dict) -> Finding:
+    return Finding(rule=row.get("rule", ""), severity=row.get("severity", "medium"),
+                   asset_id=row.get("asset_id", ""), summary=row.get("summary", ""))
+
+
+def _coverage(row: dict) -> Coverage:
+    return Coverage(
+        roots_swept=tuple(RootSwept(**r) for r in row.get("roots_swept", ())),
+        boundaries_hit=tuple(BoundaryHit(**b) for b in row.get("boundaries_hit", ())),
+        denied=tuple(Denied(**d) for d in row.get("denied", ())),
+        unavailable=tuple(Unavailable(**u) for u in row.get("unavailable", ())),
+        truncated=tuple(Truncated(**t) for t in row.get("truncated", ())),
+        probes=tuple(ProbeRun(**p) for p in row.get("probes", ())),
+        out_of_scope=tuple(row.get("out_of_scope", ())),
+    )
+
+
+def stats(snapshot: Snapshot) -> dict[str, int]:
+    return {
+        "asset_count": len(snapshot.assets),
+        "finding_count": len(snapshot.findings),
+        "review_queue_count": len(snapshot.review_queue),
+        "coverage_gaps": (
+            len(snapshot.coverage.denied)
+            + len(snapshot.coverage.unavailable)
+            + len(snapshot.coverage.boundaries_hit)
+            + len(snapshot.coverage.truncated)
+        ),
+    }
diff --git a/Discovery/adr_discovery/resolver/__init__.py b/Discovery/adr_discovery/resolver/__init__.py
new file mode 100644
index 0000000..693deba
--- /dev/null
+++ b/Discovery/adr_discovery/resolver/__init__.py
@@ -0,0 +1,159 @@
+"""M5 -- one real thing becomes exactly one asset.
+
+The count is the product, and both ways of getting it wrong are invisible
+from the inside: a false split inflates the inventory, a false merge hides
+a tool, and each produces a plausible-looking answer.
+"""
+
+from __future__ import annotations
+
+from types import MappingProxyType
+
+from ..contracts.records import Asset, Liveness, Observation, Risk
+from .confidence import band_for
+from .keys import asset_id, identity_of, normalize_root
+from .merge import group
+
+__all__ = ["resolve", "asset_id", "normalize_root"]
+
+#: Strongest liveness wins when observations of one install disagree: a
+#: process is proof the thing runs, whatever the config says.
+_LIVENESS_RANK = MappingProxyType(
+    {Liveness.RUNNING: 2, Liveness.INSTALLED: 1, Liveness.DECLARED_ONLY: 0}
+)
+
+
+def resolve(observations: tuple[Observation, ...], telemetry: dict[str, str] | None = None,
+            ledger=None) -> tuple[Asset, ...]:
+    """Merge observations into assets, derive confidence, decide liveness."""
+    if not observations:
+        return ()
+
+    observations = bind_contained(observations)
+    groups, refused = group(observations)
+    if ledger is not None and refused:
+        ledger.probe("resolver", "ran", f"{refused} bridging merge(s) refused")
+
+    assets: list[Asset] = []
+    for members in groups:
+        rows = [observations[i] for i in members]
+        primary = _primary(rows)
+
+        evidence = tuple(e for row in rows for e in row.evidence)
+        liveness = max((r.liveness for r in rows), key=_LIVENESS_RANK.__getitem__)
+        install_root = next((r.install_root for r in rows if r.install_root), None)
+        identity = identity_of(primary)
+        owner = next((r.owner for r in rows if r.owner), "system")
+
+        rungs = tuple(sorted({e.rung for e in evidence if e.rung is not None},
+                             key=lambda r: r.value))
+        last_used = (telemetry or {}).get(primary.catalog_id or identity)
+
+        assets.append(
+            Asset(
+                asset_id=asset_id(primary.kind.value, identity, owner, install_root),
+                kind=primary.kind,
+                name=primary.detail.get("name") or identity,
+                identity=identity,
+                catalog_id=primary.catalog_id,
+                vendor=primary.detail.get("vendor"),
+                version=next((r.version for r in rows if r.version), None),
+                install_path=primary.path,
+                install_root=install_root,
+                install_method=primary.detail.get("install_method"),
+                owner=owner,
+                location=primary.detail.get("location", "local"),
+                liveness=liveness,
+                last_used=last_used,
+                confidence=band_for(evidence),
+                verification=rungs,
+                evidence=evidence,
+                risk=_runtime_risk(rows),
+                flags=_flags(rows),
+            )
+        )
+    return tuple(assets)
+
+
+def bind_contained(observations: tuple[Observation, ...]) -> tuple[Observation, ...]:
+    """An observation that lives inside another install describes it.
+
+    A package directory and the `bin` symlink pointing into it are one
+    install reached two ways. Neither shares a content hash (one is a
+    directory), an inode, or a normalized root -- so without this pass they
+    are two assets that agree about everything, which is the false split in
+    its most ordinary form.
+
+    Containment is only allowed to bind observations whose catalogued
+    identities do not disagree; the group rule in `merge` still has the
+    final say.
+    """
+    from dataclasses import replace
+
+    roots = [
+        (obs.path.rstrip("/"), obs)
+        for obs in observations
+        if obs.path and not obs.attribute_of
+    ]
+    roots.sort(key=lambda pair: len(pair[0]), reverse=True)
+
+    bound: list[Observation] = []
+    for obs in observations:
+        if obs.attribute_of or not obs.path:
+            bound.append(obs)
+            continue
+        inside = _containing(obs, roots)
+        bound.append(replace(obs, attribute_of=inside) if inside else obs)
+    return tuple(bound)
+
+
+def _containing(obs: Observation, roots) -> str | None:
+    probes = [p for p in (obs.real_path, obs.path) if p]
+    for root, other in roots:
+        if other is obs or not root:
+            continue
+        if obs.catalog_id and other.catalog_id and obs.catalog_id != other.catalog_id:
+            continue
+        if any(probe.startswith(root + "/") for probe in probes):
+            return other.path
+    return None
+
+
+def _primary(rows: list[Observation]) -> Observation:
+    """The observation that best describes the install itself.
+
+    An attribute -- a model directory, a state folder -- describes something
+    about an install and must never become the row the asset is named from.
+    """
+    real = [r for r in rows if not r.attribute_of]
+    pool = real or rows
+    return max(pool, key=lambda r: (bool(r.catalog_id), bool(r.content_hash), bool(r.path)))
+
+
+def _flags(rows: list[Observation]) -> tuple[str, ...]:
+    flags: set[str] = set()
+    if all(r.attribute_of for r in rows):
+        flags.add("state_only")
+    if any(r.detail.get("alias") for r in rows):
+        flags.add("alias")
+    if len(rows) > 1:
+        flags.add(f"merged_{len(rows)}")
+    for row in rows:
+        flags.update(str(flag) for flag in (row.detail.get("flags") or ()))
+        scope = row.detail.get("scope")
+        if scope:
+            flags.add(f"config_scope={scope}")
+    return tuple(sorted(flags))
+
+
+def _runtime_risk(rows: list[Observation]) -> Risk:
+    """Carry derived live-process facts, never raw argv or environment values."""
+    env_names = tuple(sorted({
+        str(name)
+        for row in rows
+        for name in (row.detail.get("env_names") or ())
+    }))
+    return Risk(
+        env_names=env_names,
+        unattended=any(bool(row.detail.get("unattended")) for row in rows),
+    )
diff --git a/Discovery/adr_discovery/resolver/confidence.py b/Discovery/adr_discovery/resolver/confidence.py
new file mode 100644
index 0000000..b4f8922
--- /dev/null
+++ b/Discovery/adr_discovery/resolver/confidence.py
@@ -0,0 +1,19 @@
+"""Confidence counts independent channels, never repeated observations.
+
+A binary and the symlink pointing at it are one FILESYSTEM sighting of one
+fact. Multiplying their confidences manufactures certainty out of a single
+look, which is how an inventory becomes confidently wrong.
+"""
+
+from __future__ import annotations
+
+from ..contracts.evidence import Band, Channel, Evidence
+
+
+def band_for(evidence: tuple[Evidence, ...]) -> Band:
+    """Distinct channels only -- repetition inside one channel adds nothing."""
+    return Band.from_channels([e.channel for e in evidence])
+
+
+def independent_channels(evidence: tuple[Evidence, ...]) -> tuple[Channel, ...]:
+    return tuple(sorted({e.channel for e in evidence}, key=lambda c: c.value))
diff --git a/Discovery/adr_discovery/resolver/keys.py b/Discovery/adr_discovery/resolver/keys.py
new file mode 100644
index 0000000..0a2afe9
--- /dev/null
+++ b/Discovery/adr_discovery/resolver/keys.py
@@ -0,0 +1,93 @@
+"""Merge keys, strongest first.
+
+    content hash          same bytes, wherever they sit
+  > real path / inode     same file, reached by any number of links
+  > package identity      same package record
+  > signature identity    same publisher and product
+  > catalog + owner + normalized install root
+
+The first line is the addition. Ollama was counted twice because the
+model-directory observation had no path to merge on and the binary
+observation did -- two rows, one identity, disagreeing about version and
+liveness. An observation that describes an *attribute* of an install binds
+to the install rather than standing alone.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+
+from ..contracts.records import Observation
+
+#: Content-addressed store paths carry a build hash that changes on every
+#: rebuild without the install changing. Normalizing it is what lets
+#: `asset_id` survive a store rebuild (U5-11).
+_STORE_PATTERNS = (
+    re.compile(r"^(/nix/store)/[a-z0-9]{32}-(.+?)(/.*)?$"),
+    re.compile(r"^(/gnu/store)/[a-z0-9]{32}-(.+?)(/.*)?$"),
+    re.compile(r"^(.*/\.pnpm)/[^/]+@[^/]+(/.*)?$"),
+)
+
+
+def normalize_root(path: str | None) -> str | None:
+    if not path:
+        return path
+    for pattern in _STORE_PATTERNS:
+        m = pattern.match(path)
+        if m:
+            return f"{m.group(1)}/*-{m.group(2)}"
+    return path
+
+
+def keys_for(obs: Observation) -> tuple[tuple[str, str], ...]:
+    """Every key this observation can be merged on, strongest first."""
+    out: list[tuple[str, str]] = []
+    if obs.content_hash:
+        out.append(("content", obs.content_hash))
+    if obs.real_path:
+        out.append(("realpath", obs.real_path))
+    if obs.inode:
+        out.append(("inode", obs.inode))
+    if obs.package_id:
+        out.append(("package", obs.package_id))
+    if obs.signature_id:
+        out.append(("signature", obs.signature_id))
+    if obs.catalog_id:
+        root = normalize_root(obs.install_root) or ""
+        out.append(("catalog", f"{obs.catalog_id}|{obs.owner or 'system'}|{root}"))
+    else:
+        # Uncatalogued things still have an identity. Without this key a
+        # server declared in two scopes becomes two assets -- a false split
+        # that looks exactly like two servers.
+        out.append(("identity", f"{obs.kind.value}|{obs.identity}|{obs.owner or 'system'}"))
+    # An attribute binds to *its* install and never stands alone, so both
+    # sides of the binding have to emit the same key. Keying on the
+    # parent's path rather than its identity is what keeps two installs of
+    # one tool -- a release and a vendored alpha -- from collapsing into one.
+    if obs.attribute_of:
+        out.append(("install", obs.attribute_of))
+    elif obs.path:
+        out.append(("install", obs.path))
+    return tuple(out)
+
+
+def identity_of(obs: Observation) -> str:
+    """What must stay singular within a merged group.
+
+    Two observations that name different tools may never merge, however
+    many weak keys they happen to share.
+    """
+    return obs.catalog_id or obs.identity
+
+
+def asset_id(kind: str, identity: str, owner: str, install_root: str | None) -> str:
+    """Stable across benign change.
+
+    Deliberately excludes the version (an upgrade is not a new asset), any
+    credential material (a rotation is not a new asset), and the build hash
+    inside a content-addressed store path (a rebuild is not a new asset).
+    """
+    root = normalize_root(install_root) or ""
+    payload = f"{kind}\0{identity}\0{owner}\0{root}"
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:32]
diff --git a/Discovery/adr_discovery/resolver/merge.py b/Discovery/adr_discovery/resolver/merge.py
new file mode 100644
index 0000000..9b1a626
--- /dev/null
+++ b/Discovery/adr_discovery/resolver/merge.py
@@ -0,0 +1,75 @@
+"""Group formation, and the refusal to bridge two identities.
+
+Checking conflict pairwise is not enough. A bridging observation shares a
+key with each of two unrelated tools, passes both pairwise checks, and
+transitivity then unites them -- one tool silently disappears, and the
+count, which is the product, is wrong in the direction nobody notices.
+
+So conflict is evaluated on the *group*: a merge is applied only if the
+union's identity set stays singular.
+"""
+
+from __future__ import annotations
+
+from ..contracts.records import Observation
+from .keys import identity_of, keys_for
+
+
+class _Groups:
+    """Union-find, with the union guarded by the group identity rule."""
+
+    def __init__(self, n: int) -> None:
+        self.parent = list(range(n))
+        self.identities: list[set[str]] = [set() for _ in range(n)]
+
+    def find(self, i: int) -> int:
+        while self.parent[i] != i:
+            self.parent[i] = self.parent[self.parent[i]]
+            i = self.parent[i]
+        return i
+
+    def union(self, a: int, b: int) -> bool:
+        ra, rb = self.find(a), self.find(b)
+        if ra == rb:
+            return True
+        merged = self.identities[ra] | self.identities[rb]
+        if len(merged) > 1:
+            return False  # would make the group's identity plural
+        self.parent[rb] = ra
+        self.identities[ra] = merged
+        return True
+
+
+def group(observations: tuple[Observation, ...]) -> tuple[tuple[int, ...], ...]:
+    """Return index groups. Order within a group follows input order."""
+    groups = _Groups(len(observations))
+    for i, obs in enumerate(observations):
+        ident = identity_of(obs)
+        if ident:
+            groups.identities[i].add(ident)
+
+    # Strongest key kind first, so a strong merge is applied before a weak
+    # one can be refused for a conflict the strong merge would have settled.
+    by_kind: dict[str, dict[str, list[int]]] = {}
+    for i, obs in enumerate(observations):
+        for kind, value in keys_for(obs):
+            by_kind.setdefault(kind, {}).setdefault(value, []).append(i)
+
+    refused = 0
+    for kind in ("content", "realpath", "inode", "package", "signature",
+                 "install", "catalog", "identity"):
+        for members in by_kind.get(kind, {}).values():
+            anchor = members[0]
+            for other in members[1:]:
+                if not groups.union(anchor, other):
+                    refused += 1
+
+    buckets: dict[int, list[int]] = {}
+    for i in range(len(observations)):
+        buckets.setdefault(groups.find(i), []).append(i)
+    return tuple(tuple(v) for v in buckets.values()), refused
+
+
+def group_only(observations: tuple[Observation, ...]) -> tuple[tuple[int, ...], ...]:
+    grouped, _ = group(observations)
+    return grouped
diff --git a/Discovery/adr_discovery/tests_unit/test_m5_resolver.py b/Discovery/adr_discovery/tests_unit/test_m5_resolver.py
new file mode 100644
index 0000000..98c1263
--- /dev/null
+++ b/Discovery/adr_discovery/tests_unit/test_m5_resolver.py
@@ -0,0 +1,193 @@
+"""M5 -- resolver.
+
+The count is the product, so the count is asserted -- but a count alone
+cannot distinguish a correct merge from two compensating errors, so every
+case also asserts which observations landed together.
+"""
+
+from __future__ import annotations
+
+from adr_discovery.contracts.evidence import Channel, Evidence, Rung
+from adr_discovery.contracts.records import Kind, Liveness, Observation
+from adr_discovery.resolver import asset_id, resolve
+
+
+def ev(channel, proof="seen"):
+    return (Evidence("t", channel, "/p", proof, 0.9, Rung.PROVENANCE),)
+
+
+def test_u5_01_the_ollama_split_resolves_to_one_asset():
+    observations = (
+        Observation(Kind.MODEL_RUNTIME, "ollama", path="/usr/local/bin/ollama",
+                    catalog_id="ollama", install_root="/usr/local", version="0.5.1",
+                    real_path="/usr/local/bin/ollama", evidence=ev(Channel.FILESYSTEM)),
+        Observation(Kind.MODEL_RUNTIME, "ollama", path="/home/a/.ollama/models",
+                    catalog_id="ollama", attribute_of="ollama", evidence=ev(Channel.FILESYSTEM)),
+        Observation(Kind.MODEL_RUNTIME, "ollama", path="/usr/local/bin/ollama",
+                    catalog_id="ollama", real_path="/usr/local/bin/ollama",
+                    liveness=Liveness.RUNNING, evidence=ev(Channel.RUNTIME)),
+    )
+
+    assets = resolve(observations)
+
+    assert len(assets) == 1
+    assert assets[0].liveness is Liveness.RUNNING, "a process is proof it runs"
+    assert assets[0].version == "0.5.1"
+
+
+def test_u5_02_a_symlink_is_not_a_second_witness():
+    same = dict(kind=Kind.CLI_AGENT, identity="claude-code", catalog_id="claude-code",
+                real_path="/opt/x/cli.js")
+    assets = resolve((
+        Observation(**same, path="/opt/homebrew/bin/claude", evidence=ev(Channel.FILESYSTEM)),
+        Observation(**same, path="/usr/local/bin/claude", evidence=ev(Channel.FILESYSTEM)),
+    ))
+
+    assert len(assets) == 1
+    assert assets[0].confidence.label == "low", "one channel seen twice is one channel"
+
+
+def test_u5_03_a_bridging_observation_may_not_unite_two_tools():
+    assets = resolve((
+        Observation(Kind.CLI_AGENT, "tool-x", path="/x", catalog_id="tool-x",
+                    package_id="npm:shared", evidence=ev(Channel.PACKAGE)),
+        Observation(Kind.CLI_AGENT, "tool-y", path="/y", catalog_id="tool-y",
+                    package_id="npm:shared", evidence=ev(Channel.PACKAGE)),
+    ))
+
+    assert len(assets) == 2, "pairwise checks pass; the group identity must stay singular"
+    assert {a.identity for a in assets} == {"tool-x", "tool-y"}
+
+
+def test_u5_04_same_bytes_in_two_places_is_one_asset():
+    assets = resolve((
+        Observation(Kind.CLI_AGENT, "a", path="/one", content_hash="deadbeef",
+                    catalog_id="a", evidence=ev(Channel.FILESYSTEM)),
+        Observation(Kind.CLI_AGENT, "a", path="/two", content_hash="deadbeef",
+                    catalog_id="a", evidence=ev(Channel.FILESYSTEM)),
+    ))
+
+    assert len(assets) == 1
+
+
+def test_u5_05_one_inode_reached_by_two_paths_is_one_asset():
+    assets = resolve((
+        Observation(Kind.CLI_AGENT, "a", path="/one", inode="1:42", catalog_id="a",
+                    evidence=ev(Channel.FILESYSTEM)),
+        Observation(Kind.CLI_AGENT, "a", path="/two", inode="1:42", catalog_id="a",
+                    evidence=ev(Channel.FILESYSTEM)),
+    ))
+
+    assert len(assets) == 1
+
+
+def test_u5_06_two_tools_sharing_a_root_do_not_merge():
+    assets = resolve((
+        Observation(Kind.CLI_AGENT, "a", path="/r/a", catalog_id="a",
+                    install_root="/r", owner="alice", evidence=ev(Channel.FILESYSTEM)),
+        Observation(Kind.CLI_AGENT, "b", path="/r/b", catalog_id="b",
+                    install_root="/r", owner="alice", evidence=ev(Channel.FILESYSTEM)),
+    ))
+
+    assert len(assets) == 2
+
+
+def test_u5_07_liveness_has_three_distinct_values():
+    seen = set()
+    for liveness in (Liveness.RUNNING, Liveness.INSTALLED, Liveness.DECLARED_ONLY):
+        (asset,) = resolve((
+            Observation(Kind.CLI_AGENT, "a", path="/a", catalog_id="a",
+                        liveness=liveness, evidence=ev(Channel.FILESYSTEM)),
+        ))
+        seen.add(asset.liveness)
+
+    assert len(seen) == 3
+
+
+def test_u5_08_configured_but_never_invoked_is_its_own_state():
+    (asset,) = resolve((
+        Observation(Kind.MCP_SERVER, "srv", path="/cfg", catalog_id=None,
+                    liveness=Liveness.DECLARED_ONLY, evidence=ev(Channel.CONFIG)),
+    ), telemetry={})
+
+    assert asset.liveness is Liveness.DECLARED_ONLY
+    assert asset.last_used is None, "a cleanup candidate, not a threat"
+
+
+def test_u5_09_an_upgrade_keeps_the_asset_id():
+    before = asset_id("cli_agent", "claude-code", "alice", "/opt/x")
+    after = asset_id("cli_agent", "claude-code", "alice", "/opt/x")
+
+    assert before == after
+
+
+def test_u5_10_a_credential_rotation_keeps_the_asset_id():
+    """Requires that no secret material reaches the hash, which is why the
+    id is computed inside M5 rather than assembled by its callers."""
+    payload = ("cli_agent", "claude-code", "alice", "/opt/x")
+
+    assert asset_id(*payload) == asset_id(*payload)
+
+
+def test_u5_11_a_store_rebuild_keeps_the_asset_id():
+    a = asset_id("cli_agent", "claude-code", "alice", "/nix/store/" + "a" * 32 + "-claude")
+    b = asset_id("cli_agent", "claude-code", "alice", "/nix/store/" + "b" * 32 + "-claude")
+
+    assert a == b
+
+
+def test_u5_12_independent_channels_beat_repetition():
+    three = resolve((
+        Observation(Kind.CLI_AGENT, "a", path="/a", catalog_id="a",
+                    evidence=ev(Channel.FILESYSTEM) + ev(Channel.PACKAGE) + ev(Channel.RUNTIME)),
+    ))
+    once = resolve((
+        Observation(Kind.CLI_AGENT, "b", path="/b", catalog_id="b",
+                    evidence=ev(Channel.FILESYSTEM) * 3),
+    ))
+
+    assert three[0].confidence.label == "high"
+    assert once[0].confidence.label == "low"
+
+
+def test_u5_13_a_bin_symlink_binds_to_the_package_it_points_into():
+    """A package directory and the `bin` entry pointing into it are one
+    install reached two ways. They share no hash (one is a directory), no
+    inode and no normalized root, so containment is the only thing that
+    can join them -- and without it they are two assets that agree about
+    everything, which is the false split in its most ordinary form."""
+    package = Observation(
+        Kind.CLI_AGENT, "codex-cli", path="/opt/lib/node_modules/@openai/codex",
+        catalog_id="codex-cli", package_id="npm:@openai/codex", version="0.147.0",
+        evidence=ev(Channel.PACKAGE),
+    )
+    launcher = Observation(
+        Kind.CLI_AGENT, "codex-cli", path="/opt/bin/codex", catalog_id="codex-cli",
+        real_path="/opt/lib/node_modules/@openai/codex/bin/codex.js",
+        evidence=ev(Channel.FILESYSTEM),
+    )
+
+    assets = resolve((package, launcher))
+
+    assert len(assets) == 1
+    assert assets[0].confidence.label == "medium", "package and filesystem are two channels"
+
+
+def test_u5_14_a_second_install_of_one_tool_stays_separate():
+    """Containment must key on the parent install, not on its identity --
+    otherwise a vendored copy collapses into the release."""
+    release = Observation(
+        Kind.CLI_AGENT, "codex-cli", path="/opt/lib/node_modules/@openai/codex",
+        install_root="/opt/lib/node_modules/@openai", catalog_id="codex-cli",
+        version="0.147.0", evidence=ev(Channel.PACKAGE),
+    )
+    vendored = Observation(
+        Kind.CLI_AGENT, "codex-cli", path="/home/a/.codex/plugins/appserver/codex",
+        install_root="/home/a/.codex/plugins/appserver", catalog_id="codex-cli",
+        version="0.147.0-alpha.6.5", evidence=ev(Channel.FILESYSTEM),
+    )
+
+    assets = resolve((release, vendored))
+
+    assert len(assets) == 2
+    assert {a.version for a in assets} == {"0.147.0", "0.147.0-alpha.6.5"}
diff --git a/Discovery/adr_discovery/tests_unit/test_m6_judge.py b/Discovery/adr_discovery/tests_unit/test_m6_judge.py
new file mode 100644
index 0000000..bbe6cef
--- /dev/null
+++ b/Discovery/adr_discovery/tests_unit/test_m6_judge.py
@@ -0,0 +1,160 @@
+"""M6 -- judge.
+
+Precision is asserted, not just recall, so the negative cases are the
+load-bearing half. U6-09 in particular is two false positives caught on
+real machines and by no fixture.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from adr_discovery.contracts.evidence import Band, Channel
+from adr_discovery.contracts.records import Asset, Declaration, Kind, Liveness
+from adr_discovery.judge import Policy, judge
+from adr_discovery.judge.risk import credential_reach, operand_of, pinning, unattended
+
+
+def declaration(command, args=(), **kwargs):
+    return Declaration(kind=Kind.MCP_SERVER, name=kwargs.pop("name", "srv"), path="/cfg",
+                       command=command, args=tuple(args), **kwargs)
+
+
+def asset(name="srv", kind=Kind.MCP_SERVER, **kwargs):
+    return Asset(asset_id="id-" + name, kind=kind, name=name, identity=name,
+                 catalog_id=kwargs.pop("catalog_id", "srv"),
+                 confidence=Band("medium", (Channel.CONFIG,)), **kwargs)
+
+
+# ------------------------------------------------------------------ pinning
+
+
+def test_u6_01_unpinned_then_pinned():
+    assert pinning("npx", ("-y", "server-github"))[0] is False
+    assert pinning("npx", ("-y", "@modelcontextprotocol/server-github@1.4.2"))[0] is True
+
+
+def test_u6_02_a_volume_mount_is_not_an_image():
+    operand, _ = operand_of("docker", ("run", "-v", "/data:/srv", "img:tag"))
+
+    assert operand == "img:tag"
+
+
+def test_u6_03_a_registry_url_is_not_a_package():
+    operand, _ = operand_of("pip", ("install", "--index-url", "https://r/simple", "pkg"))
+
+    assert operand == "pkg"
+
+
+def test_u6_04_a_digest_is_pinned():
+    assert pinning("docker", ("run", "ghcr.io/x/y@sha256:ab34cd56"))[0] is True
+
+
+# -------------------------------------------------------------- credentials
+
+
+def test_u6_05_an_absent_env_block_means_all_not_none():
+    names, kinds = credential_reach((), env_declared=False)
+
+    assert kinds == ("inherited",)
+    assert names == ("<inherits parent environment>",)
+
+
+def test_u6_06_names_never_values():
+    names, kinds = credential_reach(("ANTHROPIC_API_KEY", "PATH"), env_declared=True)
+
+    assert names == ("ANTHROPIC_API_KEY", "PATH")
+    assert kinds == ("anthropic",)
+
+
+# ------------------------------------------------------------------- policy
+
+
+def test_u6_07_tenant_values_come_from_configuration():
+    """One world, two policies, two verdicts. A value reachable from code
+    alone cannot pass this."""
+    from dataclasses import replace
+
+    from adr_discovery.contracts.records import Risk
+
+    subject = replace(asset(), risk=Risk(destinations=("api.vendor.example",)))
+
+    ours = Policy.from_dict({"tenant_domains": ["vendor.example"]})
+    theirs = Policy.from_dict({"tenant_domains": ["corp.internal"]})
+
+    _, none_raised = judge((subject,), {}, ours)
+    _, raised = judge((subject,), {}, theirs)
+
+    assert [f.rule for f in none_raised] == []
+    assert [f.rule for f in raised] == ["third_party_destination"]
+
+
+@pytest.mark.parametrize("value", ["claude-code", ["claude-code", 7], {"id": True}])
+def test_u6_07b_policy_sets_require_arrays_of_strings(value):
+    with pytest.raises(ValueError):
+        Policy.from_dict({"approved": value})
+
+
+def test_u6_07c_tenant_domains_are_normalized():
+    policy = Policy.from_dict({"tenant_domains": ["EXAMPLE.COM."]})
+
+    assert policy.tenant_domains == frozenset({"example.com"})
+
+
+def test_u6_08_unattended_comes_from_the_agents_own_launch():
+    """cron, a timer or a person -- the same argv is the same finding."""
+    argv = ("-p", "--dangerously-skip-permissions")
+
+    assert unattended(argv)
+    assert unattended(("--dangerously-skip-permissions",))
+    assert not unattended(("-p",)), "headless alone is not a bypass"
+
+
+# ---------------------------------------------------------------- precision
+
+
+def test_u6_09_the_two_false_positives_stay_silent():
+    from dataclasses import replace
+
+    from adr_discovery.contracts.records import Risk
+
+    # `npx eslint .` -- a shell's argv is arbitrary user text.
+    eslint = replace(
+        asset(name="eslint", kind=Kind.CLI_AGENT, catalog_id=None),
+        risk=Risk(pinned=False),
+    )
+    # A command that merely mentions a path containing "mcp".
+    mentions = replace(
+        asset(name="build", kind=Kind.CLI_AGENT, catalog_id=None),
+        risk=Risk(pinned=False),
+    )
+
+    _, findings = judge((eslint, mentions), {}, Policy())
+
+    assert findings == (), "neither was caught by the suite the first time"
+
+
+def test_u6_10_declared_is_silent_and_undeclared_is_not():
+    from dataclasses import replace
+
+    declared = asset(name="github", liveness=Liveness.RUNNING)
+    undeclared = replace(declared, asset_id="id-undeclared", flags=("undeclared",))
+
+    _, quiet = judge((declared,), {}, Policy())
+    _, loud = judge((undeclared,), {}, Policy())
+
+    assert [f.rule for f in quiet] == []
+    assert "undeclared_mcp_server" in [f.rule for f in loud]
+
+
+def test_u6_11_an_unreadable_shape_raises_nothing(world):
+    """Ambiguity resolves toward the safe verdict, and is recorded."""
+    gate = world.gate()
+    subject = asset(name="odd")
+    declarations = {subject.asset_id: declaration("some-unknown-runner", ("--weird",), name="odd")}
+
+    judged, findings = judge((subject,), declarations, Policy(), gate.ledger)
+
+    assert judged[0].risk.pinned is None
+    assert [f.rule for f in findings if f.rule == "unpinned_mcp_server"] == []
+    assert any(p.name == "judge" and p.status == "degraded" for p in gate.ledger.freeze().probes)
diff --git a/Discovery/adr_discovery/tests_unit/test_m7_reporter.py b/Discovery/adr_discovery/tests_unit/test_m7_reporter.py
new file mode 100644
index 0000000..8cc6a67
--- /dev/null
+++ b/Discovery/adr_discovery/tests_unit/test_m7_reporter.py
@@ -0,0 +1,115 @@
+"""M7 -- reporter.
+
+Half of these assert on a single snapshot; the rest need two, so the
+harness stores a before and an after and diffs them the way production
+does. The empty-machine case runs first, because a module that writes
+nothing when it finds nothing is indistinguishable from one that failed.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import replace
+
+import pytest
+
+from adr_discovery.contracts.records import Asset, Kind, Risk
+from adr_discovery.contracts.snapshot import Coverage, Denied, Snapshot, Unavailable
+from adr_discovery.reporter import DifferentEndpoints, diff, to_json
+
+
+def snapshot(assets=(), hostname="host-a", coverage=None):
+    return Snapshot(hostname=hostname, username="alice", platform="darwin",
+                    timestamp="2026-08-22T00:00:00+00:00", assets=tuple(assets),
+                    coverage=coverage or Coverage())
+
+
+def asset(name="claude-code", version="2.1.3", root="/opt/x", **kwargs):
+    return Asset(asset_id=kwargs.pop("asset_id", "id-" + name), kind=Kind.CLI_AGENT,
+                 name=name, identity=name, install_root=root, version=version, **kwargs)
+
+
+def test_u7_01_a_clean_machine_still_emits_a_snapshot():
+    empty = snapshot()
+
+    document = json.loads(to_json(empty))
+
+    assert document["assets"] == []
+    assert "coverage" in document
+    assert document["schema_version"]
+
+
+def test_u7_02_asset_ids_are_unique():
+    many = [asset(name=f"t{i}", asset_id=f"id-{i}") for i in range(500)]
+
+    ids = {a.asset_id for a in many}
+
+    assert len(ids) == 500
+
+
+def test_u7_03_an_upgrade_is_one_change_not_two():
+    before = snapshot([asset(version="2.1.3")])
+    after = snapshot([asset(version="2.2.0")])
+
+    delta = diff(before, after)
+
+    assert [c.kind for c in delta.changes] == ["version_changed"]
+    assert delta.of("appeared") == () and delta.of("disappeared") == ()
+
+
+def test_u7_04_a_reinstall_is_not_a_disappearance():
+    before = snapshot([asset(root="/opt/old", asset_id="id-old")])
+    after = snapshot([asset(root="/opt/new", asset_id="id-new")])
+
+    delta = diff(before, after)
+
+    assert [c.kind for c in delta.changes] == ["reinstalled"]
+
+
+def test_u7_05_risk_moves_even_when_the_inventory_does_not():
+    """pinned -> floating is a silent regression if only membership is diffed."""
+    before = snapshot([replace(asset(), risk=Risk(pinned=True))])
+    after = snapshot([replace(asset(), risk=Risk(pinned=False))])
+
+    delta = diff(before, after)
+
+    assert [c.detail for c in delta.of("risk_delta")] == ["pinned -> floating"]
+
+
+def test_u7_06_a_surface_that_went_dark_is_a_change():
+    before = snapshot(coverage=Coverage())
+    after = snapshot(coverage=Coverage(denied=(Denied("/opt/secret", "eacces"),),
+                                       unavailable=(Unavailable("dpkg", "gone"),)))
+
+    delta = diff(before, after)
+
+    assert any("became unreadable" in c for c in delta.coverage_delta)
+    assert any("provider became unavailable" in c for c in delta.coverage_delta)
+
+
+def test_u7_07_two_endpoints_are_refused():
+    with pytest.raises(DifferentEndpoints):
+        diff(snapshot(hostname="host-a"), snapshot(hostname="host-b"))
+
+
+def test_u7_07b_cross_endpoint_is_possible_but_must_be_asked_for():
+    delta = diff(snapshot(hostname="host-a"), snapshot(hostname="host-b"),
+                 allow_cross_endpoint=True)
+
+    assert delta.endpoint == "host-b"
+
+
+def test_u7_08_fleet_fan_out_is_not_computed_on_the_endpoint():
+    document = json.loads(to_json(snapshot([asset()])))
+
+    assert not any("fleet" in key or "fan_out" in key for key in document)
+
+
+def test_u7_09_a_snapshot_round_trips():
+    original = snapshot([asset()])
+
+    document = json.loads(to_json(original))
+
+    assert document["assets"][0]["asset_id"] == "id-claude-code"
+    assert document["assets"][0]["kind"] == "cli_agent"
+    assert json.loads(to_json(original)) == document