feat(kernel): bound, hash-verified evidence and the verified-evidence completion policy (slice 3/5)

README.md

@@ -33,7 +33,10 @@ ships, on disk and runnable today:
   named reason otherwise — e.g. `python3 -m pytest`), policy digest — and
   `loop doctor` fails when a record declares that its own
   producer verified it; identity is recorded, not proven: a worker can write a
-  false verifier name, and the records live outside the hash chain.
+  false verifier name. A verified dispatch **binds its evidence digests into the
+  hash chain**, and `loop doctor` re-hashes them — binding makes tampering
+  detectable against an anchor, not impossible: without `--expect-chain-head` a
+  worker who can rewrite `.loop/` can rewrite the chain too.
 
 ![The inspector scores a self-asserted DIY loop 0/weak, then the gate-backed example 90/strong — both runs live](docs/demo.gif)
 

loop/completion.py

@@ -11,9 +11,15 @@
 from collections.abc import Mapping
 from typing import Final, Literal, TypeAlias, cast
 
-CompletionMode: TypeAlias = Literal["all_required"]
+CompletionMode: TypeAlias = Literal["all_required", "all_required_verified_evidence"]
 DEFAULT_COMPLETION_MODE: Final[CompletionMode] = "all_required"
-SUPPORTED_COMPLETION_MODES: Final[tuple[CompletionMode, ...]] = (DEFAULT_COMPLETION_MODE,)
+VERIFIED_EVIDENCE_MODE: Final[CompletionMode] = "all_required_verified_evidence"
+SUPPORTED_COMPLETION_MODES: Final[tuple[CompletionMode, ...]] = (
+    DEFAULT_COMPLETION_MODE,
+    VERIFIED_EVIDENCE_MODE,
+)
+
+_RECORD_PREFIX = ".loop/evidence/"
 
 
 class CompletionPolicyError(ValueError):
@@ -63,11 +69,24 @@ def criteria_satisfy_completion(
     ``True``; truthy substitutes such as ``1`` are deliberately rejected.
     """
     normalized = normalize_completion_policy(policy)
-    if normalized["mode"] == "all_required":
+    if normalized["mode"] in SUPPORTED_COMPLETION_MODES:
+        # The criteria half is deliberately shared: the verified-evidence mode raises
+        # the EVIDENCE bar, never the criteria bar.
         return bool(criteria_met) and all(value is True for value in criteria_met.values())
     raise AssertionError(f"unhandled completion policy: {normalized!r}")
 
 
+def policy_requires_verified_evidence(policy: object | None = None) -> bool:
+    """True when this policy's evidence bar is hash-verified records, not path strings."""
+    return normalize_completion_policy(policy)["mode"] == VERIFIED_EVIDENCE_MODE
+
+
+def evidence_entry_is_record_shaped(entry: object) -> bool:
+    """The structural half of the bar — the most a pure, I/O-free layer can honestly check."""
+    return (isinstance(entry, str) and entry.startswith(_RECORD_PREFIX)
+            and entry.endswith(".json") and ".." not in entry)
+
+
 def unmet_required_criteria(criteria_met: Mapping[str, object]) -> tuple[str, ...]:
     """Return stable string identifiers for criteria not proven true."""
     return tuple(sorted(key for key, value in criteria_met.items() if value is not True))

loop/emit.py

@@ -11,25 +11,29 @@
 import json
 import os
 import tempfile
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Mapping, Sequence
 
 from . import fsm
 from .chain import ChainHashError
 from .completion import (
+    VERIFIED_EVIDENCE_MODE,
     CompletionPolicyError,
     criteria_satisfy_completion,
     normalize_completion_policy,
+    policy_requires_verified_evidence,
     unmet_required_criteria,
 )
 from .contract import (
     TERMINAL_STATES,
+    _strict_evidence_failure,
     _validate_record,
     _validate_terminal,
     _validation_mode,
 )
-from .evidence import EVIDENCE_SCHEMA_ID
+from .evidence import EVIDENCE_SCHEMA_ID, artifact_object_path, validate_evidence
 from .paths import ARTIFACTS_DIR_NAME, EVIDENCE_DIR_NAME, resolve_loop_paths
 from .scaffold import scaffold
 from .verifier import criterion_partition, verification_policy_digest
@@ -242,6 +246,28 @@ def append_receipt(
     return receipts
 
 
+def _refuse_unverified_evidence(paths, evidence: Sequence[str]) -> None:
+    """Enforce the verified-evidence bar at write time, through contract's one definition.
+
+    An unreadable store is a REFUSAL, never a skip: the whole point of the mode is that
+    a cited record is bound into a chain, and a chain nobody can read proves nothing.
+    """
+    from .runtime import RuntimeStoreError, bound_artifact_digests  # local: runtime -> events -> emit
+
+    try:
+        bound = bound_artifact_digests(paths.workspace)
+    except RuntimeStoreError as exc:
+        raise EmitError(
+            f"refusing Succeeded under {VERIFIED_EVIDENCE_MODE}: the event store could not be "
+            f"read ({exc}) — chain-boundness is unestablished, and unestablished is not proof"
+        ) from exc
+    for entry in evidence:
+        detail = _strict_evidence_failure(entry, paths, bound)
+        if detail is not None:
+            raise EmitError(
+                f"refusing Succeeded under {VERIFIED_EVIDENCE_MODE}: {entry} {detail}")
+
+
 def terminate(
     target: str | Path,
     *,
@@ -302,6 +328,10 @@ def terminate(
             raise EmitError(
                 "refusing Succeeded because not all required criteria are proven true: " + detail
             )
+        if policy_requires_verified_evidence(normalized_policy):
+            # Resolved here rather than by hoisting the assignment below, so the order in
+            # which terminate reports its refusals is exactly what it was.
+            _refuse_unverified_evidence(_require_contract(target), evidence)
 
     paths = _require_contract(target)
     terminal_path = paths.loop_dir / "terminal_state.json"
@@ -407,23 +437,36 @@ def sync_state_to_projection(target: str | Path, projection: dict[str, Any]) ->
 DEFAULT_VERIFIER_IDENTITY = "loop.run"
 
 
-def write_verify_evidence(
+@dataclass(frozen=True)
+class BuiltEvidence:
+    """Every byte and digest of one verify write, computed without touching disk.
+
+    Separating the computation from the write lets a caller commit to the exact
+    artifact hashes BEFORE anything is placed, so the digests a durable event
+    carries and the bytes on disk cannot disagree.
+    """
+
+    bundle_path: Path
+    bundle_text: str
+    sha256: str
+    record_path: Path
+    record_text: str
+    record_sha256: str
+    object_path: Path
+    artifact_hashes: tuple[dict[str, str], ...]
+
+
+def build_verify_evidence(
     target: str | Path, *, run_id: str, iteration_id: int, task: Mapping[str, Any], passed: bool,
     code_identity: Mapping[str, Any], summary: str = "", executor: str | None = None,
     verifier_identity: str | None = None, attempt: int | None = None,
-) -> dict[str, Any]:
-    """Write one verify bundle and its hashed evidence@1 record.
-
-    ``code_identity`` is REQUIRED and is never derived here: only the caller knows
-    which verifier actually ran (see loop.verifier.executed_verifier_identity vs
-    injected_verifier_identity). Deriving it from ``task['verify']`` would record a
-    command that an injected verifier never executed.
+) -> BuiltEvidence:
+    """Render one verify bundle, its evidence@1 record, and their bound digests.
 
-    The bundle is the artifact (metrics-compatible; carries verifier identity, the
-    verdict's source, and the DECLARED criterion partition); the record is the
-    schema-validated pointer that commits to the bundle's bytes. Identities are
-    recorded, never inferred: an unsupplied executor is the literal ``unattributed``
-    and an unsupplied attempt is ``null``.
+    Pure with respect to the workspace: it reads the contract and writes nothing.
+    The record it renders is byte-identical to what ``write_verify_evidence``
+    places, and is schema-validated here so an invalid record is refused before
+    any file exists rather than after.
     """
     paths = _require_contract(target)
     iteration_id = _require_iteration_id(iteration_id)
@@ -450,20 +493,16 @@ def write_verify_evidence(
         "partition": criterion_partition(task),
     }
     bundle_path = paths.loop_dir / ARTIFACTS_DIR_NAME / f"verify-iter{iteration_id}.json"
-    bundle_path.parent.mkdir(parents=True, exist_ok=True)
     bundle_text = json.dumps(bundle, indent=2, sort_keys=True) + "\n"
-    # Staged under a name metrics does not rglob, so a failure before the record is
-    # written cannot leave an orphan green bundle for FCR to count.
-    staged = bundle_path.with_name(bundle_path.name + ".staged")
-    _atomic_write_text(staged, bundle_text)
+    bundle_sha256 = hashlib.sha256(bundle_text.encode("utf-8")).hexdigest()
 
     now = datetime.now(timezone.utc).isoformat(timespec="seconds")
     record = {
         "schema": EVIDENCE_SCHEMA_ID,
         "id": f"{run_id}:{iteration_id}:verify",
         "kind": "verify-bundle",
         "uri": bundle_path.relative_to(paths.workspace).as_posix(),
-        "sha256": hashlib.sha256(bundle_text.encode("utf-8")).hexdigest(),
+        "sha256": bundle_sha256,
         "media_type": "application/json",
         "created_at": now,
         "produced_by": {"run_id": run_id, "task_id": task.get("id"), "attempt": attempt,
@@ -473,12 +512,90 @@ def write_verify_evidence(
                         "code_digest_basis": identity["code_digest_basis"],
                         "policy_digest": identity["policy_digest"]},
     }
+    validation = validate_evidence(record, mode="basic")
+    if not validation["ok"]:
+        raise EmitError(f"evidence record failed schema validation: {validation['issues']}")
+
     record_path = paths.loop_dir / EVIDENCE_DIR_NAME / f"evidence-iter{iteration_id}.json"
-    record_path.parent.mkdir(parents=True, exist_ok=True)
+    record_text = json.dumps(record, indent=2, sort_keys=True) + "\n"
+    record_sha256 = hashlib.sha256(record_text.encode("utf-8")).hexdigest()
+    object_path = artifact_object_path(paths.workspace, bundle_sha256)
+    artifact_hashes = tuple(
+        {"path": path.relative_to(paths.workspace).as_posix(), "sha256": digest}
+        for path, digest in ((bundle_path, bundle_sha256), (record_path, record_sha256),
+                             (object_path, bundle_sha256))
+    )
+    return BuiltEvidence(
+        bundle_path=bundle_path, bundle_text=bundle_text, sha256=bundle_sha256,
+        record_path=record_path, record_text=record_text, record_sha256=record_sha256,
+        object_path=object_path, artifact_hashes=artifact_hashes,
+    )
+
+
+def write_verify_evidence(
+    target: str | Path, *, run_id: str, iteration_id: int, task: Mapping[str, Any], passed: bool,
+    code_identity: Mapping[str, Any], summary: str = "", executor: str | None = None,
+    verifier_identity: str | None = None, attempt: int | None = None,
+    built: BuiltEvidence | None = None,
+) -> dict[str, Any]:
+    """Write one verify bundle, its content-addressed object, and its evidence@1 record.
+
+    ``code_identity`` is REQUIRED and is never derived here: only the caller knows
+    which verifier actually ran (see loop.verifier.executed_verifier_identity vs
+    injected_verifier_identity). Deriving it from ``task['verify']`` would record a
+    command that an injected verifier never executed.
+
+    The bundle is the artifact (metrics-compatible; carries verifier identity, the
+    verdict's source, and the DECLARED criterion partition); the record is the
+    schema-validated pointer that commits to the bundle's bytes. Identities are
+    recorded, never inferred: an unsupplied executor is the literal ``unattributed``
+    and an unsupplied attempt is ``null``.
+
+    ``built`` lets a caller that already rendered the write place exactly those
+    bytes, so the digests it committed to elsewhere describe what landed.
+    """
+    if built is None:
+        built = build_verify_evidence(
+            target, run_id=run_id, iteration_id=iteration_id, task=task, passed=passed,
+            code_identity=code_identity, summary=summary, executor=executor,
+            verifier_identity=verifier_identity, attempt=attempt,
+        )
+
+    # The object is placed FIRST and never replaced: it is the recovery copy the
+    # evidence record's digest still resolves to after the mutable bundle path is
+    # overwritten.
+    try:
+        built.object_path.parent.mkdir(parents=True, exist_ok=True)
+        _atomic_create_text(built.object_path, built.bundle_text)
+    except FileExistsError:
+        # The sibling `except OSError` cannot catch what is raised INSIDE this handler,
+        # so the collision read is wrapped here: an IsADirectoryError or PermissionError
+        # at the object path must leave write_verify_evidence as a typed EmitError.
+        try:
+            existing = hashlib.sha256(built.object_path.read_bytes()).hexdigest()
+        except OSError as exc:
+            raise EmitError(
+                f"object store entry for {built.sha256} exists but cannot be read at "
+                f"{built.object_path}: {exc}") from exc
+        if existing != built.sha256:
+            raise EmitError(
+                f"object store holds different bytes for {built.sha256}: refusing to overwrite "
+                f"a content-addressed object (found {existing})")
+    except OSError as exc:
+        raise EmitError(f"object store write failed at {built.object_path}: {exc}") from exc
+
+    built.bundle_path.parent.mkdir(parents=True, exist_ok=True)
+    # Staged under a name metrics does not rglob, so a failure before the record is
+    # written cannot leave an orphan green bundle for FCR to count.
+    staged = built.bundle_path.with_name(built.bundle_path.name + ".staged")
+    _atomic_write_text(staged, built.bundle_text)
+
+    built.record_path.parent.mkdir(parents=True, exist_ok=True)
     try:
-        _atomic_write_text(record_path, json.dumps(record, indent=2, sort_keys=True) + "\n")
+        _atomic_write_text(built.record_path, built.record_text)
     except BaseException:
         staged.unlink(missing_ok=True)
         raise
-    os.replace(staged, bundle_path)
-    return {"bundle": bundle_path, "evidence": record_path, "sha256": record["sha256"]}
+    os.replace(staged, built.bundle_path)
+    return {"bundle": built.bundle_path, "evidence": built.record_path,
+            "sha256": built.sha256, "object": built.object_path}