Answer two independent reviews: three regressions and a fallback

src/ctrlrun/authority.py

@@ -1350,8 +1350,17 @@ def evaluate(
                 try:
                     chain = self._check_chain(delegation, store=store, now=now)
                 except _UnreadableError as unreadable:
+                    # SPEC-v0.10 §9 — `hop` is on **every** result an action under one produces,
+                    # passing or failing, which is what lets §6.3 print a command for each. This
+                    # path is the chain walk's own unreadable record, one frame below the
+                    # identical handler above, and it was the one return that dropped it: an
+                    # independent review found a hop with an unreadable ANCESTOR refusing with
+                    # `hop=None`, leaving §6.3 no argument to print.
                     return AuthorityResult(
-                        False, AUTHORITY_UNREADABLE, delegation_id=unreadable.delegation_id
+                        False,
+                        AUTHORITY_UNREADABLE,
+                        delegation_id=unreadable.delegation_id,
+                        hop=hop,
                     )
                 depth = chain.depth
                 if chain.failure is not None:

src/ctrlrun/gateway/__init__.py

@@ -15,18 +15,20 @@
 from __future__ import annotations
 
 import importlib
+import logging
 import sys
 from pathlib import Path
 from types import ModuleType
 from typing import Any, Final
 
-from ..errors import MissingDependency
+from ..errors import InvalidArgument, MissingDependency
 
 #: The extra's HTTP client, imported by name so a missing one is a `MissingDependency`
 #: rather than a `ModuleNotFoundError` from halfway down an import chain.
 _HTTP_CLIENT: Final = "httpx"
 
 #: The extra that carries it, for the install command in the error.
+_LOG: Final = logging.getLogger("ctrlrun.gateway")
 _EXTRA: Final = "gateway"
 
 __all__ = ["serve", "serve_operator"]
@@ -109,6 +111,14 @@ def serve(*, upstream: str, alias: str, **options: Any) -> None:
         # and §4.4 refuses a pinned action there.
         upstream=config.upstream,
     )
+    # SPEC-v0.10 §4.3's check 1, and the only one of the three where the operator is present.
+    # Without it the observation register is empty in every shipped process, check 2 answers
+    # `upstream_unverified` for ever, and a gateway that pins refuses every pinned action. A
+    # review found exactly that: the register was written only by tests.
+    #
+    # **A mismatch refuses to start**, printing observed beside pinned, because a pin an operator
+    # got wrong should fail on a console rather than on production traffic.
+    _observe_the_upstream(control, config)
     forwarder = httpx_forwarder(config, control.policy)
     gateway = Gateway(config, control, forwarder)
     _announce(control, config, gateway.identity, authority_path)
@@ -133,7 +143,6 @@ def _authority(control: Any, path: str | None) -> Any:
     operator who edited the wrong file saw no effect and no error.
     """
     from ..authority import Authority
-    from ..errors import InvalidArgument
 
     if path is None:
         return control.authority
@@ -307,3 +316,30 @@ def _announce_operator(control: Any, config: Any, identity: Any, store: Any) ->
             f"authority    {len(control.authority.grants)} grant(s), evaluated by the agent",
             flush=True,
         )
+
+
+def _observe_the_upstream(control: Any, config: Any) -> None:
+    """SPEC-v0.10 §4.3, check 1. One connection, one comparison, before the listener opens."""
+    from ..upstream import observe_upstream, pinned_context
+
+    pins = [control.policy.upstream_pin(name) for name in control.policy.actions]
+    pinned = [pin for pin in pins if pin]
+    if not pinned:
+        return
+    certs = tuple(sorted({path for pin in pinned for path in pin.certs}))
+    observed = observe_upstream(
+        config.upstream,
+        verify=pinned_context(certs) if certs else None,
+        timeout=config.upstream_timeout,
+    )
+    expected: set[str] = set()
+    for pin in pinned:
+        expected.update(pin.cert_sha256)
+    if expected and observed not in expected:
+        raise InvalidArgument(
+            f"{config.upstream} presented {observed}, which is in no 'tls_cert_sha256' this "
+            f"policy pins ({', '.join(sorted(expected))}). A swapped server behind the same "
+            "name is what the pin exists to catch, so the gateway does not start "
+            "(SPEC-v0.10 §4.3)"
+        )
+    _LOG.info("upstream %s observed as %s", config.upstream, observed)

src/ctrlrun/gateway/server.py

@@ -432,8 +432,37 @@ def _relay(self, parsed: ParsedRequest, headers: Mapping[str, str]) -> _Response
         if payload is None:
             _LOG.warning("relaying %s failed: %s", parsed.method, observed)
             return _Response(502)
+        self._observe_tools(parsed, payload)
         return _Response(status, payload, response_headers)
 
+    def _observe_tools(self, parsed: ParsedRequest, payload: bytes) -> None:
+        """Record what the upstream advertises, from the `tools/list` it just relayed.
+
+        SPEC-v0.10 §4.2: the tool-schema pin is over the **whole** advertised entry, name,
+        description and input schema together, because a description that changed is a tool whose
+        behaviour an operator has not reviewed. §4.3's check 2 compares against what this process
+        observed, and this is the only place the gateway sees it: `tools/list` is relayed rather
+        than intercepted (`v0.2 §6.3` -- it is not an action), so the observation rides the relay.
+
+        **Best effort, and never a refusal.** A malformed or absent `tools` array leaves the
+        register untouched, which leaves a pinned action `upstream_unverified`: the fail-closed
+        direction, and the same answer as never having called `tools/list` at all.
+        """
+        if parsed.method != "tools/list":
+            return
+        from ..upstream import observe_tool_schema
+
+        try:
+            document = json.loads(payload)
+            tools = document.get("result", {}).get("tools", [])
+        except (ValueError, AttributeError):
+            return
+        if not isinstance(tools, list):
+            return
+        for entry in tools:
+            if isinstance(entry, Mapping) and isinstance(entry.get("name"), str):
+                observe_tool_schema(self._config.upstream, entry["name"], entry)
+
     def relay_method(self, method: str, body: bytes, headers: Mapping[str, str]) -> _Response:
         """Relay GET/DELETE transport operations without inventing an action."""
         origin = _header(headers, "origin")

src/ctrlrun/policy.py

@@ -1592,9 +1592,62 @@ def _parse_upstream(value: object, where: str) -> UpstreamPin:
         raise PolicyError(
             f"{where}: 'upstream.tool_schema_sha256' must be 'sha256:' followed by 64 hex chars"
         )
-    return UpstreamPin(
+    pin = UpstreamPin(
         cert_sha256=tuple(digests), certs=tuple(files), tool_schema_sha256=schema_hash
     )
+    _check_pin_correspondence(pin, where)
+    return pin
+
+
+def _check_pin_correspondence(pin: UpstreamPin, where: str) -> None:
+    """The two TLS halves must agree, checked at load (SPEC-v0.10 §4.2).
+
+    §4.2 measured what a half-moved rotation costs and then this check was not written, which an
+    independent review found: a document whose `tls_cert_file` still held only the old certificate
+    while `tls_cert_sha256` had both loaded cleanly and failed at the **handshake**, on the day an
+    operator believed they had prepared for. That is the outage §4.2 says the list prevents,
+    arriving one layer down.
+
+    So: every certificate `tls_cert_file` holds hashes to a digest `tls_cert_sha256` names, and a
+    path that does not exist is a load error rather than an empty trust store discovered at the
+    first connection.
+    """
+    if not pin.certs:
+        return
+    import hashlib
+    import ssl
+
+    for path in pin.certs:
+        try:
+            der_list = [
+                ssl.PEM_cert_to_DER_cert(block + "-----END CERTIFICATE-----")
+                for block in Path(path).read_text().split("-----END CERTIFICATE-----")
+                if "BEGIN CERTIFICATE" in block
+            ]
+        except OSError as unreadable:
+            raise PolicyError(
+                f"{where}: 'upstream.tls_cert_file' names {path!r}, which could not be read "
+                f"({unreadable.strerror}); a pin whose certificate is missing builds an empty "
+                "trust store and refuses every connection (SPEC-v0.10 §4.2)"
+            ) from unreadable
+        except ValueError as malformed:
+            raise PolicyError(
+                f"{where}: 'upstream.tls_cert_file' names {path!r}, which is not PEM: {malformed}"
+            ) from malformed
+        if not der_list:
+            raise PolicyError(
+                f"{where}: 'upstream.tls_cert_file' names {path!r}, which holds no certificate"
+            )
+        if not pin.cert_sha256:
+            continue
+        digests = {"sha256:" + hashlib.sha256(der).hexdigest() for der in der_list}
+        if not digests & set(pin.cert_sha256):
+            raise PolicyError(
+                f"{where}: 'upstream.tls_cert_file' {path!r} hashes to "
+                f"{sorted(digests)[0]}, which 'upstream.tls_cert_sha256' does not name. The two "
+                "halves pin the same certificates or a rotation that moves one fails at the "
+                "handshake (SPEC-v0.10 §4.2)"
+            )
 
 
 def _parse_entry(

src/ctrlrun/scan.py

@@ -190,6 +190,17 @@ class ScanReport:
     vocabulary: tuple[str, ...]
     policy_path: str | None
     policy_read: bool
+    #: SPEC-v0.10 §6.4 — the principals holding a grant no hop bounds, in codepoint order.
+    #:
+    #: §2.3.2's residual is that CTRLRun cannot make a receiving agent present the hop it was
+    #: given: one holding a grant of its own can decline and act on that instead. The deployment
+    #: rule that collapses it is *an agent that only ever acts on handed-over work holds no root
+    #: grant of its own*, and without a surface that rule is advice. This is the surface.
+    #:
+    #: **It reports and does not score.** `v0.4 §3.9`'s rule that CTRLRun never grades an
+    #: operator's document holds here: a principal on this line is a fact, not a finding, and it
+    #: does not move `exit_code`.
+    root_grant_holders: tuple[str, ...] = ()
 
     @property
     def exit_code(self) -> int:
@@ -570,6 +581,7 @@ def scan(
         vocabulary=words,
         policy_path=str(policy_path) if policy_path else None,
         policy_read=policy_path is not None,
+        root_grant_holders=_root_grant_holders(policy_path),
     )
 
 
@@ -587,6 +599,26 @@ def _finding_line(finding: Finding) -> str:
     return f"  {where}  {subject}  [{finding.kind}: {finding.rule}]{detail}"
 
 
+def _root_grant_holders(policy_path: Path | None) -> tuple[str, ...]:
+    """Which principals the document grants authority no hop bounds (SPEC-v0.10 §6.4).
+
+    A **root** grant, meaning one written in the document rather than delegated at runtime: those
+    are the ones an agent holds whether or not anybody handed it work. A document with no
+    `authority:` section grants nothing and answers with nothing.
+    """
+    if policy_path is None:
+        return ()
+    from .authority import _optional_from_yaml
+
+    try:
+        authority = _optional_from_yaml(policy_path.read_text(), source=str(policy_path))
+    except Exception:
+        return ()
+    if authority is None:
+        return ()
+    return tuple(sorted({grant.subject.agent or "*" for grant in authority.grants.values()}))
+
+
 def report_lines(report: ScanReport) -> list[str]:
     """The human rendering. Every finding in the document has a line here (§5.3, T204)."""
     lines = [f"ctrlrun scan — {report.root}", ""]
@@ -597,6 +629,15 @@ def report_lines(report: ScanReport) -> list[str]:
         lines.append(f"{kind} ({len(found)})")
         lines.extend(_finding_line(finding) for finding in found)
         lines.append("")
+    if report.root_grant_holders:
+        # SPEC-v0.10 §6.4. A fact about the document, not a finding: it does not move the exit
+        # code, and `v0.4 §3.9` is why there is no verdict attached to it.
+        lines.append(f"holds a root grant ({len(report.root_grant_holders)})")
+        lines.extend(f"  {agent}" for agent in report.root_grant_holders)
+        lines.append(
+            "  an agent that only ever acts on handed-over work holds none (SPEC-v0.10 §2.3.2)"
+        )
+        lines.append("")
     if report.undetermined:
         lines.append(f"undetermined ({len(report.undetermined)})")
         lines.extend(
@@ -652,6 +693,9 @@ def report_document(report: ScanReport) -> dict[str, Any]:
             {"file": call.file, "line": call.line, "expression": call.expression}
             for call in report.undetermined
         ],
+        # SPEC-v0.10 §6.4 — a fact about the document, additive, and outside `findings` because
+        # it is not one: `v0.4 §3.9` keeps `scan` from grading an operator's choices.
+        "root_grant_holders": list(report.root_grant_holders),
         "totals": {
             "files_read": report.files_read,
             "files_excluded": report.files_excluded,

src/ctrlrun/upstream.py

@@ -23,6 +23,7 @@
     import ssl
 
 from .action import canonical_bytes
+from .errors import InvalidArgument
 from .policy import UPSTREAM_MISMATCH, UPSTREAM_UNVERIFIED, UpstreamPin
 
 #: SPEC-v0.10 §4.2 — the tool-schema hash's domain tag. `canonical_bytes` is the one
@@ -80,6 +81,56 @@
             del _TOOLS[key]
 
 
+def observe_upstream(url: str, *, verify: object | None = None, timeout: float = 10.0) -> str:
+    """Open one TLS connection to `url`, record its leaf certificate, and return the digest.
+
+    **This is §4.3's check 1, and it is what makes check 2 answerable at all.** Without it the
+    register is empty in every shipped process, so `check` answers `upstream_unverified` for ever
+    and §10's `upstream_mismatch` row describes an outcome nothing can produce. A review found
+    exactly that: the register was written only by tests and by `verify`'s own scenario.
+
+    It is the **legible** check, and the only one where the operator is present: a gateway that
+    calls this at startup fails on a console rather than on production traffic.
+
+    `verify` is the pinned `SSLContext` where §4.2's certificate half is configured, so a swapped
+    server fails this handshake too and the gateway never starts. Where only the digest half is
+    configured there is no context to build, the handshake is ordinary, and the comparison is
+    check 2's job.
+
+    A plain `http://` upstream has no certificate to observe and is left unrecorded, so a pin on
+    it stays `upstream_unverified`: a pin is a claim about a server's identity and an unencrypted
+    hop carries none.
+    """
+    import socket
+    import ssl
+    from urllib.parse import urlsplit
+
+    split = urlsplit(url)
+    if split.scheme != "https":
+        raise InvalidArgument(
+            f"{url!r} is not https, so it presents no certificate to pin against; a pin is a "
+            "claim about a server's identity and an unencrypted hop carries none "
+            "(SPEC-v0.10 §4.3)"
+        )
+    host = split.hostname or ""
+    port = split.port or 443
+    context = verify if isinstance(verify, ssl.SSLContext) else ssl.create_default_context()
+    with (
+        socket.create_connection((host, port), timeout=timeout) as raw,
+        context.wrap_socket(raw, server_hostname=host) as tls,
+    ):
+        der = tls.getpeercert(binary_form=True)
+    if not der:
+        raise InvalidArgument(
+            f"{url!r} presented no certificate this process could read, so nothing can be "
+            "pinned against it (SPEC-v0.10 §4.3)"
+        )
+    # Keyed by the URL, because that is what `GatewayConfig.upstream` holds and what
+    # `Control._upstream` passes to `check`. A register keyed by host and read by URL is
+    # two registers.
+    return observe_certificate(url, der)
+
+
 def check(pin: UpstreamPin, upstream: str, tool: str | None = None) -> str | None:
     """§4.3's check 2: the reason this action is refused, or `None` where the pin is satisfied.
 
@@ -142,6 +193,7 @@
     "forget",
     "observe_certificate",
     "observe_tool_schema",
+    "observe_upstream",
     "pinned_context",
     "tool_schema_hash",
 ]