feat(loop): C2 launch-time committed-truth check (keyless council anchor)

.console/log.md

@@ -1,4 +1,36 @@
 # Log
+## 2026-07-14 — refactor: C2 shares one public section-extractor with signing (reviewer code_quality)
+
+Addressed the C2 reviewer's code_quality concern + removed a real smell:
+committed.py imported the PRIVATE `signing._load_section` AND duplicated its
+section-extraction as `_section_from_yaml_text`. Promoted to a single public
+`signing.section_of(data)` (the one source of truth for what the anchor covers)
++ `signing.load_section(file)`; committed.py now calls both, so the signed and
+committed checks provably resolve the SAME scope with no duplication. Also
+caught a latent bug: the committed-YAML parse could raise `yaml.YAMLError`,
+which `verify_committed` (contract: never raises) did not catch — now returns
+`skip` on malformed committed YAML too. 408 pass; a same-scope test pins that
+`load_section` and `section_of` resolve identically (T1).
+
+## 2026-07-14 — feat: C2 launch-time committed-truth check (keyless council anchor)
+
+Council spec (COUNCIL_VERDICT.md) Phase 1. When the loop config has NO signed
+reference, `cl loop run` now compares the live `pseudo_operator:` section
+against the committed copy on `origin/main` (`git show origin/main:<relpath>`
+after a bounded `git fetch`) — the keyless analogue of Track C's
+restore-by-consumption. Match → run live (it IS committed truth); drift → run
+the COMMITTED copy (`signed_status="drift_unsigned"`, no YAML rewrite) and flag
+loudly; unreachable origin → skip with a loud note (degrade-never-halt). New
+launcher flag `--require-committed` (parity with `--require-signed`, lives
+outside agent-reachable config) turns it into a fail-closed gate: refuses to
+start on drift AND when it cannot confirm committed truth (offline) — a
+fail-open skip there would let anyone bypass the gate by cutting network. A
+signed reference present ⇒ Track C wins unchanged; C2 is never consulted. New
+module `pseudo_operator/committed.py` reuses signing's `_load_section` +
+`canonical_bytes` so the committed scope is byte-identical to what Track C
+signs. 28 tests (test_committed.py + signing/loop regression); T1 satisfied by
+a `CommittedResult` isinstance assertion.
+
 ## 2026-07-07 — fix: surface hook stderr in the loop log on success
 
 Hooks log their actions via logging → stderr (e.g. OC self-update's

src/context_lifecycle/cli/loop.py

@@ -57,16 +57,29 @@ def run_cmd(
         help="Refuse to start unless the config is anchored (signed reference "
         "verifies). Launchers pass this once the operator has signed.",
     ),
+    require_committed: bool = typer.Option(
+        False,
+        "--require-committed",
+        help="Keyless (C2) parity with --require-signed: when the config is NOT "
+        "signed, refuse to start if the live section drifts from origin/main "
+        "instead of running the committed copy. Ignored when a signed reference "
+        "is present (Track C wins).",
+    ),
 ) -> NoReturn:
     """Run the loop (foreground). Refuses to start on an invalid config.
 
     Trust anchor (Track C): with a signed reference present, the engine runs
     the VERIFIED section — on drift it consumes the reference (not the live
     section) and flags the divergence; a bad signature always refuses.
+
+    Committed-truth check (C2): with NO signed reference, the engine compares
+    the live section against the committed copy on ``origin/main`` — on drift it
+    runs the committed copy (``drift_unsigned``) unless ``--require-committed``
+    refuses; offline/unreachable launches degrade to unsigned with a loud note.
     """
     try:
         cfg, signed_status, detail = load_verified_config(
-            config, require_signed=require_signed
+            config, require_signed=require_signed, require_committed=require_committed
         )
     except (ValueError, OSError) as exc:
         typer.echo(f"config error: {exc}", err=True)
@@ -77,6 +90,12 @@ def run_cmd(
             f"reference (restore-by-consumption): {detail}",
             err=True,
         )
+    elif signed_status == "drift_unsigned":
+        typer.echo(
+            f"[TRUST] live config DIVERGES from origin/main — running the COMMITTED "
+            f"copy (keyless committed-truth check, restore-by-consumption): {detail}",
+            err=True,
+        )
     elif signed_status == "unsigned":
         typer.echo(
             f"[TRUST] loop config is NOT anchored ({detail}) — running unsigned; "

src/context_lifecycle/pseudo_operator/committed.py

@@ -0,0 +1,122 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+# Copyright (C) 2026 ProtocolWarden
+"""Committed-truth check (C2) — the keyless analogue of Track C's anchor.
+
+When no signed reference exists the loop has no cryptographic anchor, but the
+committed history is still an authority the fleet cannot rewrite from inside a
+session: ``origin/main`` is protected by branch rules + CODEOWNERS, so the
+*committed* copy of the config is a trustworthy "restore" target even without a
+signature. C2 makes "restore vs. change" decidable the same way Track C does,
+keylessly:
+
+- read the committed ``pseudo_operator:`` section from ``origin/main``
+  (``git show origin/main:<relpath>`` after a bounded ``git fetch``);
+- drift (live section != committed section) → the caller runs the COMMITTED
+  copy and flags the divergence — restore-by-consumption, no YAML rewriting;
+- match → the live section IS the committed truth, run it as normal unsigned.
+
+Degrade-never-halt: an ordinary launch that cannot reach ``origin`` (offline,
+fetch failure, config absent on origin, git unavailable) SKIPS the check with a
+loud note rather than refusing — the keyless check is a guardrail, not a gate.
+The gate form is opt-in via the launcher's ``--require-committed`` flag, which
+lives outside agent-reachable config exactly like ``--require-signed``.
+
+Signed reference present ⇒ Track C wins; C2 is not consulted (see
+``config.load_verified_config``).
+"""
+
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+import yaml
+
+from .signing import canonical_bytes, load_section, section_of
+
+COMMITTED_REF = "origin/main"
+# Bounded so an offline/slow launch degrades quickly instead of hanging.
+FETCH_TIMEOUT_SECONDS = 20
+GIT_TIMEOUT_SECONDS = 15
+
+
+@dataclass(frozen=True)
+class CommittedResult:
+    """Outcome of comparing the live section against ``origin/main``."""
+
+    status: str  # "match" | "drift" | "skip"
+    detail: str
+    committed: dict | None  # the committed section (match/drift), else None
+
+
+def _run_git(args: list[str], cwd: Path, timeout: int) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(  # noqa: S603 — fixed argv, no shell
+        ["git", *args],
+        cwd=cwd,
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+    )
+
+
+def verify_committed(config_file: Path) -> CommittedResult:
+    """Compare the live section against the committed copy on ``origin/main``.
+
+    Never raises on reachability/parse failures — those return ``skip`` (the
+    caller runs the live section unsigned, loudly). Only genuine drift returns
+    ``drift`` with the committed section for the caller to consume.
+    """
+    cwd = config_file.resolve().parent
+    try:
+        fetch = _run_git(["fetch", "--quiet", "origin", "main"], cwd, FETCH_TIMEOUT_SECONDS)
+    except (OSError, subprocess.SubprocessError) as exc:
+        return CommittedResult("skip", f"origin fetch failed: {exc}", None)
+    if fetch.returncode != 0:
+        return CommittedResult("skip", f"origin fetch failed: {fetch.stderr.strip()}", None)
+
+    try:
+        top = _run_git(["rev-parse", "--show-toplevel"], cwd, GIT_TIMEOUT_SECONDS)
+    except (OSError, subprocess.SubprocessError) as exc:
+        return CommittedResult("skip", f"git rev-parse failed: {exc}", None)
+    if top.returncode != 0:
+        return CommittedResult("skip", f"not a git repo: {top.stderr.strip()}", None)
+    try:
+        relpath = config_file.resolve().relative_to(Path(top.stdout.strip()))
+    except ValueError as exc:
+        return CommittedResult("skip", f"config file outside git repo: {exc}", None)
+
+    try:
+        show = _run_git(
+            ["show", f"{COMMITTED_REF}:{relpath.as_posix()}"], cwd, GIT_TIMEOUT_SECONDS
+        )
+    except (OSError, subprocess.SubprocessError) as exc:
+        return CommittedResult("skip", f"git show failed: {exc}", None)
+    if show.returncode != 0:
+        return CommittedResult(
+            "skip", f"config absent on {COMMITTED_REF}: {show.stderr.strip()}", None
+        )
+
+    try:
+        committed = section_of(yaml.safe_load(show.stdout), source=f"{COMMITTED_REF}:committed")
+    except (ValueError, yaml.YAMLError) as exc:
+        return CommittedResult("skip", f"committed config unreadable: {exc}", None)
+    try:
+        live = load_section(config_file)
+    except ValueError as exc:
+        return CommittedResult("skip", f"live config unreadable: {exc}", None)
+
+    if canonical_bytes(live) != canonical_bytes(committed):
+        return CommittedResult(
+            "drift",
+            f"live pseudo_operator section diverges from {COMMITTED_REF}",
+            committed,
+        )
+    return CommittedResult("match", f"live section matches {COMMITTED_REF}", committed)
+
+
+__all__ = [
+    "COMMITTED_REF",
+    "CommittedResult",
+    "verify_committed",
+]

src/context_lifecycle/pseudo_operator/config.py

@@ -235,43 +235,94 @@ def load_pseudo_operator_config(config_file: Path) -> PseudoOperatorConfig:
     return PseudoOperatorConfig.model_validate(section)
 
 
+def _with_repo_root(section: dict, config_file: Path) -> dict:
+    """Inject the default ``repo_root`` into a section loaded from anchor bytes."""
+    section = dict(section)
+    if "repo_root" not in section:
+        parent = config_file.resolve().parent
+        section["repo_root"] = str(
+            parent.parent if parent.name in {".context", ".console"} else parent
+        )
+    return section
+
+
 def load_verified_config(
-    config_file: Path, *, require_signed: bool = False
+    config_file: Path, *, require_signed: bool = False, require_committed: bool = False
 ) -> tuple[PseudoOperatorConfig, str, str]:
-    """Load the loop config through the trust anchor (Track C).
+    """Load the loop config through the trust anchor (Track C) or, when it is
+    not anchored, the keyless committed-truth check (C2).
 
     Returns ``(config, signed_status, detail)`` where signed_status is
-    ok | drift | unsigned. Enforcement:
+    ok | drift | unsigned | drift_unsigned. Enforcement:
 
-    - ``ok``: run the live section (it IS the reference).
+    - ``ok``: run the live section (it IS the signed reference).
     - ``drift``: run the SIGNED REFERENCE, not the live section — restore-by-
       consumption; the caller logs the divergence loudly.
     - ``bad_signature``: raise — a present-but-unverifiable reference means the
       anchor is compromised or corrupt; never fall back to the live section.
-    - ``unsigned``: raise when ``require_signed``; otherwise run the live
-      section (pre-anchoring mode; the caller warns).
+    - no signed reference (``unsigned``): raise when ``require_signed``;
+      otherwise fall through to the committed-truth check (C2, below).
+
+    C2 — when no signed reference exists (Track C wins unchanged whenever one is
+    present, so this runs only in the unsigned case):
+
+    - committed match: run the live section (it IS the committed truth) as
+      normal unsigned.
+    - ``drift_unsigned``: the live section diverges from ``origin/main`` — run
+      the COMMITTED copy (restore-by-consumption, no YAML rewriting) and flag it
+      loudly, OR raise when ``require_committed``.
+    - skip (offline / fetch failure / config absent on origin): run the live
+      section unsigned with a loud note — degrade-never-halt for ordinary
+      launches; but raise when ``require_committed`` — the gate cannot confirm
+      the live section matches committed truth, and a fail-open skip would let
+      anyone bypass ``--require-committed`` by cutting network access (parity
+      with ``--require-signed``, which likewise refuses when it cannot confirm).
     """
+    from .committed import verify_committed
     from .signing import verify_reference
 
     result = verify_reference(config_file)
     if result.status == "bad_signature":
         raise ValueError(f"signed loop config FAILED verification: {result.detail}")
-    if result.status == "unsigned":
-        if require_signed:
+    if result.status == "drift":
+        section = _with_repo_root(result.reference or {}, config_file)
+        return PseudoOperatorConfig.model_validate(section), "drift", result.detail
+    if result.status == "ok":
+        return load_pseudo_operator_config(config_file), "ok", result.detail
+
+    # No signed reference — Track C does not apply.
+    if require_signed:
+        raise ValueError(
+            f"--require-signed set but the loop config is not anchored ({result.detail}); "
+            "sign it with `cl loop sign-config`"
+        )
+
+    # C2: keyless committed-truth check against origin/main.
+    committed = verify_committed(config_file)
+    if committed.status == "drift":
+        if require_committed:
             raise ValueError(
-                f"--require-signed set but the loop config is not anchored ({result.detail}); "
-                "sign it with `cl loop sign-config`"
+                "--require-committed set but the live loop config DIVERGES from "
+                f"origin/main ({committed.detail}); commit the change on origin/main "
+                "or relaunch without --require-committed"
             )
-        return load_pseudo_operator_config(config_file), "unsigned", result.detail
-    if result.status == "drift":
-        section = dict(result.reference or {})
-        if "repo_root" not in section:
-            parent = config_file.resolve().parent
-            section["repo_root"] = str(
-                parent.parent if parent.name in {".context", ".console"} else parent
+        section = _with_repo_root(committed.committed or {}, config_file)
+        return PseudoOperatorConfig.model_validate(section), "drift_unsigned", committed.detail
+    if committed.status == "skip":
+        if require_committed:
+            raise ValueError(
+                "--require-committed set but the live loop config could not be "
+                f"verified against origin/main ({committed.detail}); the gate "
+                "cannot confirm committed truth — resolve connectivity or "
+                "relaunch without --require-committed"
             )
-        return PseudoOperatorConfig.model_validate(section), "drift", result.detail
-    return load_pseudo_operator_config(config_file), "ok", result.detail
+        return (
+            load_pseudo_operator_config(config_file),
+            "unsigned",
+            f"{result.detail}; committed-truth check skipped: {committed.detail}",
+        )
+    # match — live section is the committed truth.
+    return load_pseudo_operator_config(config_file), "unsigned", committed.detail
 
 
 __all__ = [

src/context_lifecycle/pseudo_operator/signing.py

@@ -55,26 +55,36 @@ def _paths(config_file: Path) -> tuple[Path, Path, Path]:
     return d / REFERENCE_NAME, d / SIGNATURE_NAME, d / PUBKEY_NAME
 
 
-def _load_section(config_file: Path) -> dict:
-    from context_lifecycle.io.yaml_io import load_yaml_safe
+def section_of(data: object, *, source: str = "config") -> dict:
+    """Extract the anchored ``pseudo_operator`` section from parsed YAML data.
 
-    data = load_yaml_safe(config_file, default=None)
+    The single source of truth for what a signed/committed anchor covers, so the
+    file path (:func:`load_section`) and any already-parsed source (e.g. the
+    committed bytes from ``git show``) resolve the SAME scope.
+    """
     if not isinstance(data, dict):
-        raise ValueError(f"{config_file}: not a YAML mapping")
+        raise ValueError(f"{source}: not a YAML mapping")
     section = data.get("pseudo_operator", data)
     if not isinstance(section, dict):
-        raise ValueError(f"{config_file}: no pseudo_operator section")
+        raise ValueError(f"{source}: no pseudo_operator section")
     return section
 
 
+def load_section(config_file: Path) -> dict:
+    """Load and extract the anchored section from a YAML config file."""
+    from context_lifecycle.io.yaml_io import load_yaml_safe
+
+    return section_of(load_yaml_safe(config_file, default=None), source=str(config_file))
+
+
 def sign_reference(config_file: Path, private_key_path: Path) -> Path:
     """Operator tool: snapshot + sign the live ``pseudo_operator:`` section.
 
     Run OFFLINE with the operator's private key (PEM, as produced by the EVAL
     keygen). Writes the canonical snapshot + signature beside the config file;
     commit both. Never called by any fleet path.
     """
-    section = _load_section(config_file)
+    section = load_section(config_file)
     payload = canonical_bytes(section)
     data = private_key_path.read_bytes()
     key = load_pem_private_key(data, password=None)
@@ -129,7 +139,7 @@ def verify_reference(config_file: Path) -> VerifyResult:
     except ValueError as exc:
         return VerifyResult("bad_signature", f"reference is not valid JSON: {exc}", None)
     try:
-        live = _load_section(config_file)
+        live = load_section(config_file)
     except ValueError as exc:
         return VerifyResult("drift", f"live config unreadable: {exc}", reference)
     if canonical_bytes(live) != canonical_bytes(reference):
@@ -147,6 +157,8 @@ def verify_reference(config_file: Path) -> VerifyResult:
     "SIGNATURE_NAME",
     "VerifyResult",
     "canonical_bytes",
+    "load_section",
+    "section_of",
     "sign_reference",
     "verify_reference",
 ]