#000027 + #000028: canonical projections persist; STRICT-WITNESSED reachable

Closes #000027. Closes #000028 (cache-leg wired).

#000027 — canonical projections persist to providence_cache
============================================================

Math/logic π* answers (arithmetic@v1, logic-kernel@v1,
time-series-quantized@v1, …) are now first-class providence rows.
Pre-fix: question → kernel → answer → return. No cache, no audit
event, no run_dag, no inspect/burn/replay surface.

Post-fix: question → cache_key (8-dim, synthetic for the three
RAG-shaped dims) → lookup → on miss persist (providence_cache row +
providence_canonical audit event + canonical run_dag) → return.

Synthetic cache_key dimensions for canonical rows (per ticket §2.2):

- source_root        = sha256("pi_star_source:" + pi_star_ref)
- model_profile_hash = sha256("pi_star_model:"  + pi_star_ref)
- conversation_hash  = sha256("pi_star_conv:"   + canonical_q + ":" + ref)
- chunking_version   = literal "n/a-canonical" — chunker bumps on
                       wikipedia path don't stale math answers.

The other dims (question_hash, governance_policy_hash, schema_version,
canonicalization_version) are real and shared with the RAG path.

Schema: audit_mode CHECK widened to admit 'CANONICAL_PROJECTION';
verifier_method CHECK widened to admit 'canonical_projection'. New
_rebuild_providence_cache_canonical_projection migration helper
follows the existing _rebuild_providence_cache_* pattern (temp-table
dance, additive value-space, fully idempotent). Wired into connect()
migration block alongside the prior CHECK extensions.

Cache-hit policy: trust the row. Kernel-version drift is handled by
pi_star_ref bumping (synthetic source_root changes → fresh row,
prior row stays in DB but unreachable via the live cache_key).
Re-running on every hit would defeat the optimization without
adding audit value the version-pin doesn't already provide.

Policy gate: canonical_projection_preflight_persist (default True).
Operators who want the legacy transient render-only behavior set it
to False — keeps the existing canon-CLI experience for tests /
probes / scripts that don't want audit-chain entries for math
questions.

CLI render: `CANONICAL · via canonical_projection` for persisted
rows. Works through the existing cache_hit / cache_miss_then_written
render path; no new render branch needed.

`arborist canon <key> "<input>"` stays transient — direct one-shot
probe, never persists. Boundary preserved per ticket §2.6.

#000028 — multi-modality witness cache-leg
==========================================

Pre-#000027 the witness cache-leg closure always returned None;
STRICT-WITNESSED (3-of-3 byte-equal) was structurally unreachable.
Post-#000027 the closure now returns the persisted answer bytes
when a prior canonical row exists. Three-way agreement
(kernel == cache == canonicalize(LLM)) is now reachable on the
second canonical-witness call.

New test test_query_canonical_witness_reaches_strict_after_persist
covers it end-to-end: first call writes the row + KERNEL-LLM-AGREE;
second call hits cache + STRICT-WITNESSED.

Tests
=====

- tests/test_canonical_cache.py: 16 new tests covering ticket §7
  acceptance criteria (cache_key shape, persist round-trip, audit
  event, hit-count increments, chain integrity, pi_star version
  bump orphans old row, distinct refs namespace separately,
  chunking_version sentinel, governance policy invalidates lookup,
  canon stays transient, synthetic source_root encodes ref).
- tests/test_canonical_projection.py: assertions updated — status
  is now cache_miss_then_written / cache_hit instead of
  canonical_projection. Added a transient-mode test pinning the
  policy gate.
- tests/test_witness.py: status assertions updated to reflect
  persistence; new STRICT-WITNESSED test.
- tests/test_directives.py: D7 audit_mode enum test now admits
  CANONICAL_PROJECTION (governance event — admissibility class
  added).

Full suite: 1367 passed, 36 skipped (was 1306; +61 new).

Real-shard smoke
================

  $ make query Q="0.1 + 0.2" BURN=1
  → cache_miss_then_written, ~300ms wall, row written
  $ make query Q="0.1 + 0.2"
  → cache_hit, ~40ms wall, hit_count++

  $ make chain-check-shards
  → 0 breaks per shard

arborist/cli.py

@@ -662,6 +662,11 @@ def _render_audit_label(
     if is_claim_lattice:
         rung = _ladder_rung_for_lattice(audit_mode, violations)
         return f"{rung} · via {verifier_method}"
+    if verifier_method == "canonical_projection":
+        # Display label for #000027 — persisted canonical answers
+        # (math/logic π* outputs). The line-rendering site appends
+        # pi_star_ref as a tail since this helper doesn't have it.
+        return "CANONICAL · via canonical_projection"
     # Quote / span / entity / paraphrase: keep audit_mode as the
     # primary token; append method for clarity.
     return f"{audit_mode} · via {verifier_method}"

arborist/qa/canonical_cache.py

@@ -0,0 +1,350 @@
+"""Persistence + cache lookup for canonical-projection answers.
+
+#000027 — implementation.
+
+Math/logic π* answers (`arithmetic@v1`, `logic-kernel@v1`,
+`time-series-quantized@v1`, …) used to short-circuit RAG and
+return without ever touching the audit chain. This module makes
+them first-class providence_cache rows: cache_key minted from the
+8-dim invariant with synthetic-but-deterministic values for the
+three RAG-shaped dimensions (`source_root`, `model_profile_hash`,
+`conversation_hash`); persisted with `audit_mode =
+CANONICAL_PROJECTION` and `verifier_method = canonical_projection`;
+audit-chain entry under event_type ``providence_canonical``.
+
+The hard-constraint contract from the ticket: the canonical bytes
+ARE the answer. Round-trip MUST be byte-identical — re-canonicalizing
+the stored input through the same `pi_star_ref` produces the
+stored output, every time. We trust the row on cache hit (do not
+re-run the kernel); kernel-version drift is handled by `pi_star_ref`
+bumping (`@v1` → `@v2`), which mints a new synthetic source_root
+and naturally orphans prior records.
+
+Synthetic-dim choices (per ticket §2.2):
+
+- ``source_root``         = sha256("pi_star_source:" + pi_star_ref)
+- ``model_profile_hash``  = sha256("pi_star_model:"  + pi_star_ref)
+- ``conversation_hash``   = sha256("pi_star_conv:"   + canonical_question + ":" + pi_star_ref)
+- ``chunking_version``    = literal "n/a-canonical" — chunking is
+  irrelevant to math; using the live CHUNKING_VERSION would
+  mass-stale every canonical row whenever the wikipedia chunker
+  changes (unrelated to math correctness).
+
+Other dims (`question_hash`, `governance_policy_hash`,
+`schema_version`, `canonicalization_version`) are real and shared
+with the RAG path.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sqlite3
+import time
+from typing import Callable
+
+from arborist import (
+    CANONICALIZATION_VERSION,
+    CHUNKING_VERSION,  # noqa: F401 — referenced by docstring; not used
+    SCHEMA_VERSION,
+)
+from arborist.qa.dag import _canonical_json, _sha256_hex
+from arborist.qa.keys import (
+    cache_key as _cache_key,
+    canonical_question as _canonical_question,
+    governance_policy_hash,
+    question_hash as _question_hash,
+)
+from arborist.store import append_audit, transaction
+
+
+CANONICAL_CHUNKING_VERSION = "n/a-canonical"
+CANONICAL_AUDIT_EVENT_TYPE = "providence_canonical"
+
+
+def canonical_synthetic_source_root(pi_star_ref: str) -> str:
+    return _sha256_hex(f"pi_star_source:{pi_star_ref}")
+
+
+def canonical_synthetic_model_hash(pi_star_ref: str) -> str:
+    return _sha256_hex(f"pi_star_model:{pi_star_ref}")
+
+
+def canonical_synthetic_conversation_hash(
+    canonical_question_str: str, pi_star_ref: str
+) -> str:
+    return _sha256_hex(
+        f"pi_star_conv:{canonical_question_str}:{pi_star_ref}"
+    )
+
+
+def canonical_cache_key(
+    *,
+    question: str,
+    pi_star_ref: str,
+    policy: dict,
+    mode: str = "equivalence_class",
+) -> str:
+    """8-dim cache_key for a canonical projection answer.
+
+    Reuses the standard `cache_key` function from `arborist.qa.keys`
+    so the canonical path lives in the same identity space as RAG
+    rows. The three RAG-shaped dimensions get synthetic values
+    pinned to `pi_star_ref`.
+    """
+    qhash = _question_hash(question, mode=mode)
+    src = canonical_synthetic_source_root(pi_star_ref)
+    mhash = canonical_synthetic_model_hash(pi_star_ref)
+    chash = canonical_synthetic_conversation_hash(
+        _canonical_question(question, mode=mode), pi_star_ref
+    )
+    ghash = governance_policy_hash(policy)
+    return _cache_key(
+        src,
+        qhash,
+        mhash,
+        chash,
+        ghash,
+        SCHEMA_VERSION,
+        CANONICALIZATION_VERSION,
+        CANONICAL_CHUNKING_VERSION,
+    )
+
+
+def build_canonical_run_dag(
+    *,
+    question_hash_value: str,
+    pi_star_ref: str,
+    canonical_input_sha256: str,
+    canonical_output_sha256: str,
+) -> dict:
+    """3-stage run-DAG for canonical projections.
+
+    question → canonical_projection → final_label
+
+    Each node carries an opaque ``hash`` (stage payload sha256). The
+    Merkle root of the three hashes is the run_dag_root we persist.
+    """
+    canonical_payload = {
+        "stage": "canonical_projection",
+        "pi_star_ref": pi_star_ref,
+        "canonical_input_sha256": canonical_input_sha256,
+        "canonical_output_sha256": canonical_output_sha256,
+    }
+    canonical_hash = _sha256_hex(_canonical_json(canonical_payload))
+    final_payload = {
+        "stage": "final_label",
+        "audit_mode": "CANONICAL_PROJECTION",
+        "verifier_method": "canonical_projection",
+    }
+    final_hash = _sha256_hex(_canonical_json(final_payload))
+    nodes = [
+        {"stage": "question", "hash": question_hash_value},
+        {"stage": "canonical_projection", "hash": canonical_hash,
+         "payload": canonical_payload},
+        {"stage": "final_label", "hash": final_hash,
+         "payload": final_payload},
+    ]
+    # Root: sha256 of concatenated hashes (cheap, deterministic, and
+    # consistent with the lighter run-DAG shape used by other paths
+    # that don't need full Merkle proofs).
+    root = hashlib.sha256(
+        "".join(n["hash"] for n in nodes).encode("utf-8")
+    ).hexdigest()
+    return {"root": root, "nodes": nodes}
+
+
+def lookup_canonical(
+    qa_conn: sqlite3.Connection, cache_key_value: str
+) -> sqlite3.Row | None:
+    """Return the live providence_cache row for `cache_key_value`,
+    or None on miss. Filters on falsification_state='live' the same
+    way the RAG path does.
+    """
+    row = qa_conn.execute(
+        "SELECT cache_key, source_root, document_uri, question_text, "
+        " answer_text, audit_mode, verifier_method, n_quotes, n_verified, "
+        " falsification_state, run_dag_root, run_dag_blob, "
+        " audit_event_hash, created_at "
+        "FROM providence_cache "
+        "WHERE cache_key = ? AND falsification_state = 'live'",
+        (cache_key_value,),
+    ).fetchone()
+    return row
+
+
+def increment_hit_count(
+    qa_conn: sqlite3.Connection, cache_key_value: str
+) -> None:
+    """Mirror the RAG path's last_hit_at + hit_count bump on cache hit."""
+    qa_conn.execute(
+        "UPDATE providence_cache "
+        "SET last_hit_at = ?, hit_count = hit_count + 1 "
+        "WHERE cache_key = ?",
+        (int(time.time()), cache_key_value),
+    )
+
+
+def persist_canonical(
+    qa_conn: sqlite3.Connection,
+    *,
+    cache_key_value: str,
+    question: str,
+    pi_star_ref: str,
+    canonical_input_bytes: bytes,
+    canonical_output_bytes: bytes,
+    policy: dict,
+    chain: str = "private",
+    mode: str = "equivalence_class",
+) -> tuple[str, str, dict]:
+    """Write one canonical providence_cache row + matching audit event.
+
+    Returns (event_hash, run_dag_root, run_dag).
+    """
+    canonical_input_sha = _sha256_hex(
+        canonical_input_bytes.decode("utf-8", errors="surrogatepass")
+    )
+    canonical_output_sha = _sha256_hex(
+        canonical_output_bytes.decode("utf-8", errors="surrogatepass")
+    )
+    qhash = _question_hash(question, mode=mode)
+    src = canonical_synthetic_source_root(pi_star_ref)
+    mhash = canonical_synthetic_model_hash(pi_star_ref)
+    chash = canonical_synthetic_conversation_hash(
+        _canonical_question(question, mode=mode), pi_star_ref
+    )
+    ghash = governance_policy_hash(policy)
+
+    run_dag = build_canonical_run_dag(
+        question_hash_value=qhash,
+        pi_star_ref=pi_star_ref,
+        canonical_input_sha256=canonical_input_sha,
+        canonical_output_sha256=canonical_output_sha,
+    )
+    run_dag_blob = json.dumps(run_dag, separators=(",", ":"))
+
+    answer_text = canonical_output_bytes.decode("utf-8", errors="replace")
+    document_uri = f"pi_star://{pi_star_ref}"
+    merkle_proof = json.dumps(
+        {"context_root": None, "sources": []}, separators=(",", ":")
+    )
+    now = int(time.time())
+
+    audit_body = {
+        "pi_star_ref": pi_star_ref,
+        "question_text": question,
+        "canonical_input_sha256": canonical_input_sha,
+        "canonical_output_bytes_hex": canonical_output_bytes.hex(),
+        "canonical_output_text": answer_text,
+        "kernel_audit_mode": "CANONICAL_PROJECTION",
+    }
+
+    with transaction(qa_conn):
+        event_hash = append_audit(
+            qa_conn,
+            event_type=CANONICAL_AUDIT_EVENT_TYPE,
+            subject_root=cache_key_value,
+            body=audit_body,
+            ts=now,
+        )
+        qa_conn.execute(
+            "INSERT INTO providence_cache "
+            "(cache_key, source_root, document_uri, question_hash, question_text, "
+            " answer_text, merkle_proof, model_profile_hash, conversation_hash, "
+            " governance_policy_hash, schema_version, canonicalization_version, "
+            " chunking_version, falsification_state, chain, audit_event_hash, "
+            " created_at, hit_count, audit_mode, n_quotes, n_verified, "
+            " unverified_quotes, verifier_method, run_dag_root, run_dag_blob) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0, "
+            " ?, ?, ?, ?, ?, ?, ?)",
+            (
+                cache_key_value,
+                src,
+                document_uri,
+                qhash,
+                question,
+                answer_text,
+                merkle_proof,
+                mhash,
+                chash,
+                ghash,
+                SCHEMA_VERSION,
+                CANONICALIZATION_VERSION,
+                CANONICAL_CHUNKING_VERSION,
+                chain,
+                event_hash,
+                now,
+                "CANONICAL_PROJECTION",
+                1,  # n_quotes — the kernel output IS the one "quote"
+                1,  # n_verified — kernel-verified by definition
+                None,  # unverified_quotes
+                "canonical_projection",
+                run_dag["root"],
+                run_dag_blob,
+            ),
+        )
+    return event_hash, run_dag["root"], run_dag
+
+
+def lookup_or_persist(
+    qa_conn: sqlite3.Connection,
+    *,
+    question: str,
+    pi_star_ref: str,
+    canonical_output_bytes: bytes,
+    policy: dict,
+    chain: str = "private",
+    mode: str = "equivalence_class",
+) -> tuple[str, sqlite3.Row | None, bool, dict]:
+    """One-shot helper: mint cache_key, look up, and persist on miss.
+
+    Returns (cache_key, row_or_None, was_hit, run_dag).
+    On hit, ``row`` is the cached row and ``was_hit=True``;
+    ``run_dag`` is reconstructed from the row's run_dag_blob (or {}
+    if absent on legacy rows).
+    On miss, the row is written; ``row`` is None (caller has the
+    canonical bytes already), ``was_hit=False``; ``run_dag`` is
+    the freshly-built DAG.
+    """
+    ckey = canonical_cache_key(
+        question=question,
+        pi_star_ref=pi_star_ref,
+        policy=policy,
+        mode=mode,
+    )
+    row = lookup_canonical(qa_conn, ckey)
+    if row is not None:
+        increment_hit_count(qa_conn, ckey)
+        try:
+            run_dag = json.loads(row["run_dag_blob"]) if row["run_dag_blob"] else {}
+        except (json.JSONDecodeError, KeyError):
+            run_dag = {}
+        return ckey, row, True, run_dag
+    _, _, run_dag = persist_canonical(
+        qa_conn,
+        cache_key_value=ckey,
+        question=question,
+        pi_star_ref=pi_star_ref,
+        canonical_input_bytes=question.encode("utf-8"),
+        canonical_output_bytes=canonical_output_bytes,
+        policy=policy,
+        chain=chain,
+        mode=mode,
+    )
+    return ckey, None, False, run_dag
+
+
+# Public exports.
+__all__ = [
+    "CANONICAL_AUDIT_EVENT_TYPE",
+    "CANONICAL_CHUNKING_VERSION",
+    "build_canonical_run_dag",
+    "canonical_cache_key",
+    "canonical_synthetic_conversation_hash",
+    "canonical_synthetic_model_hash",
+    "canonical_synthetic_source_root",
+    "increment_hit_count",
+    "lookup_canonical",
+    "lookup_or_persist",
+    "persist_canonical",
+]

arborist/qa/query.py

@@ -2008,18 +2008,106 @@ def query(
                 pi_star_ref=pi_star_ref,
                 ms=int(canonical_ms),
             )
+            # Ticket #000027 — persist canonical answers to
+            # providence_cache. Default ON; operators can disable per
+            # call via policy["canonical_projection_preflight_persist"]
+            # = False (keeps the legacy transient render-only behavior
+            # for tests / probes / scripts that don't want audit-chain
+            # entries for math questions).
+            canonical_persist_on = bool(
+                policy.get("canonical_projection_preflight_persist", True)
+            )
+            ckey: str | None = None
+            audit_event_hash_v: str | None = None
+            run_dag_root_v: str | None = None
+            run_dag_blob_v: str | None = None
+            cached_row = None
+            lookup_path_v = "preflight_canonical"
+            status_v = "canonical_projection"
+
+            if canonical_persist_on:
+                from arborist.qa.canonical_cache import (
+                    canonical_cache_key,
+                    increment_hit_count,
+                    lookup_canonical,
+                    persist_canonical,
+                )
+                primary_dedup = policy.get(
+                    "question_dedup", "equivalence_class"
+                )
+                ckey = canonical_cache_key(
+                    question=question,
+                    pi_star_ref=pi_star_ref,
+                    policy=policy,
+                    mode=primary_dedup,
+                )
+                qa_conn = connect(qa_db)
+                try:
+                    cached_row = lookup_canonical(qa_conn, ckey)
+                    if cached_row is not None:
+                        # Trust the row — kernel-version drift is
+                        # handled by pi_star_ref bumping (synthetic
+                        # source_root changes), not by re-running the
+                        # kernel on hit.
+                        answer_text = cached_row["answer_text"]
+                        run_dag_root_v = cached_row["run_dag_root"]
+                        run_dag_blob_v = cached_row["run_dag_blob"]
+                        audit_event_hash_v = cached_row["audit_event_hash"]
+                        increment_hit_count(qa_conn, ckey)
+                        lookup_path_v = "canonical_cache_hit"
+                        status_v = "cache_hit"
+                        progress.emit(
+                            "cache.hit", lookup_path="canonical"
+                        )
+                    else:
+                        progress.emit("cache.miss", lookup_path="canonical")
+                        audit_event_hash_v, run_dag_root_v, run_dag = (
+                            persist_canonical(
+                                qa_conn,
+                                cache_key_value=ckey,
+                                question=question,
+                                pi_star_ref=pi_star_ref,
+                                canonical_input_bytes=question.encode(
+                                    "utf-8"
+                                ),
+                                canonical_output_bytes=canonical_bytes,
+                                policy=policy,
+                                chain=chain,
+                                mode=primary_dedup,
+                            )
+                        )
+                        run_dag_blob_v = json.dumps(
+                            run_dag, separators=(",", ":")
+                        )
+                        lookup_path_v = "canonical_cache_miss"
+                        status_v = "cache_miss_then_written"
+                finally:
+                    qa_conn.close()
+
             # Ticket #000028 — multi-modality witness. Default OFF;
             # operator opts in via policy["canonical_witness_enabled"]
             # (CLI: --witness). Fans out cache + LLM in parallel against
             # the kernel ground truth and records cross-modality
-            # agreement. Cache leg is a no-op closure pre-Ticket #000027.
+            # agreement. Cache leg uses the real persisted-row bytes
+            # (post-#000027) when persistence is on; otherwise None.
             witness_dict = None
             if bool(policy.get("canonical_witness_enabled", False)):
                 from arborist.qa.witness import run_witness
-                # Pre-#000027: no canonical persistence, cache always
-                # misses. The closure shape stays so #000027 can wire
-                # the real lookup without changing this call site.
-                _cache_lookup = lambda: None  # noqa: E731
+                # Cache-leg closure: returns the persisted answer
+                # bytes if we found a prior row at the top of this
+                # branch (so the witness compares against the
+                # already-committed canonical answer), else None.
+                # We use the row from BEFORE we wrote — comparing
+                # against a row we just wrote in the same call would
+                # be tautological.
+                _cached_bytes = (
+                    cached_row["answer_text"].encode(
+                        "utf-8", errors="surrogatepass"
+                    )
+                    if cached_row is not None
+                    else None
+                )
+                _cache_lookup = lambda: _cached_bytes  # noqa: E731
                 _witness = run_witness(
                     question=question,
                     pi_star_ref=pi_star_ref,
@@ -2034,7 +2122,7 @@ def query(
                 )
                 witness_dict = _witness.to_dict()
             return {
-                "status": "canonical_projection",
+                "status": status_v,
                 "audit_mode": "CANONICAL_PROJECTION",
                 "verifier_method": "canonical_projection",
                 "pi_star_ref": pi_star_ref,
@@ -2042,9 +2130,12 @@ def query(
                 "n_quotes": 1,
                 "n_verified": 1,
                 "violations": [],
-                "lookup_path": "preflight_canonical",
+                "lookup_path": lookup_path_v,
                 "sources": [],
-                "cache_key": None,
+                "cache_key": ckey,
+                "audit_event_hash": audit_event_hash_v,
+                "run_dag_root": run_dag_root_v,
+                "run_dag_blob": run_dag_blob_v,
                 "witness": witness_dict,
                 "timings": {
                     "canonical_preflight_ms": canonical_ms,

arborist/store.py

@@ -174,7 +174,7 @@ CREATE TABLE IF NOT EXISTS providence_cache (
     -- (no verbatim grounding — purely emergent from training).
     -- Default UNGROUNDED: an unclassified record is the weakest claim.
     audit_mode               TEXT NOT NULL DEFAULT 'UNGROUNDED'
-                                CHECK (audit_mode IN ('STRICT','HYBRID','UNGROUNDED')),
+                                CHECK (audit_mode IN ('STRICT','HYBRID','UNGROUNDED','CANONICAL_PROJECTION')),
     n_quotes                 INTEGER NOT NULL DEFAULT 0,
     n_verified               INTEGER NOT NULL DEFAULT 0,
     -- JSON array of quoted spans the model produced but we couldn't find
@@ -187,10 +187,12 @@ CREATE TABLE IF NOT EXISTS providence_cache (
     -- spans matched but multi-word proper nouns did. 'paraphrase' =
     -- token-coverage match (soft signal). 'claim_lattice' = quote-by-
     -- pointer mode (model emitted JSON; verifier checked evidence_id
-    -- resolution + source_role + manual-quote prohibition). 'none' = no
-    -- evidence at all (truly emergent).
+    -- resolution + source_role + manual-quote prohibition).
+    -- 'canonical_projection' = answer produced by a deterministic π*
+    -- kernel (arithmetic@v1, logic-kernel@v1, …) — the canonical bytes
+    -- ARE the proof. 'none' = no evidence at all (truly emergent).
     verifier_method          TEXT NOT NULL DEFAULT 'none'
-                                CHECK (verifier_method IN ('quote','span','entity','paraphrase','claim_lattice','none')),
+                                CHECK (verifier_method IN ('quote','span','entity','paraphrase','claim_lattice','canonical_projection','none')),
     -- Per-run Merkle-DAG. run_dag_root = MerkleTree over ordered stage
     -- hashes (question / retrieval / context / prompt / answer / verify /
     -- final_label). run_dag_blob carries the full {root, nodes} JSON so
@@ -609,6 +611,20 @@ def _migrate_audit_mode(conn: sqlite3.Connection) -> None:
     if ddl_row and "'claim_lattice'" not in (ddl_row[0] or ""):
         _rebuild_providence_cache_claim_lattice(conn)
 
+    # canonical_projection (#000027) — extends BOTH audit_mode and
+    # verifier_method CHECK constraints. CANONICAL_PROJECTION is the
+    # admissibility class for deterministic π* kernel answers
+    # (arithmetic@v1, logic-kernel@v1, …); 'canonical_projection' is the
+    # verifier-method token. Detect either missing → rebuild once.
+    ddl_row = conn.execute(
+        "SELECT sql FROM sqlite_master WHERE type='table' AND name='providence_cache'"
+    ).fetchone()
+    if ddl_row and (
+        "'CANONICAL_PROJECTION'" not in (ddl_row[0] or "")
+        or "'canonical_projection'" not in (ddl_row[0] or "")
+    ):
+        _rebuild_providence_cache_canonical_projection(conn)
+
     conn.execute(
         "CREATE INDEX IF NOT EXISTS idx_providence_audit "
         "ON providence_cache(audit_mode)"
@@ -1035,6 +1051,84 @@ def _rebuild_providence_cache_claim_lattice(conn: sqlite3.Connection) -> None:
         raise
 
 
+def _rebuild_providence_cache_canonical_projection(conn: sqlite3.Connection) -> None:
+    """One-time table rebuild: extend audit_mode CHECK to admit
+    'CANONICAL_PROJECTION' AND verifier_method CHECK to admit
+    'canonical_projection' (#000027 — canonical projections persist).
+
+    Same temp-table dance as the paraphrase / claim_lattice rebuilds.
+    SQLite cannot modify a CHECK constraint in place; create new table
+    with both extended CHECKs, copy data verbatim (the new tokens are
+    additive — no value translation), drop old, rename new.
+
+    Caller probes existing CHECK from sqlite_master and only invokes
+    this helper when EITHER token is missing from the constraint
+    string. Idempotent.
+    """
+    new_create = """
+    CREATE TABLE providence_cache_new (
+        cache_key                TEXT PRIMARY KEY,
+        source_root              TEXT NOT NULL,
+        document_uri             TEXT NOT NULL,
+        question_hash            TEXT NOT NULL,
+        question_text            TEXT NOT NULL,
+        answer_text              TEXT NOT NULL,
+        merkle_proof             TEXT NOT NULL,
+        model_profile_hash       TEXT NOT NULL,
+        conversation_hash        TEXT NOT NULL,
+        governance_policy_hash   TEXT NOT NULL,
+        schema_version           TEXT NOT NULL,
+        canonicalization_version TEXT NOT NULL,
+        chunking_version         TEXT NOT NULL,
+        falsification_state      TEXT NOT NULL DEFAULT 'live'
+                                    CHECK (falsification_state IN ('live','failed','stale','quarantined')),
+        chain                    TEXT NOT NULL DEFAULT 'private'
+                                    CHECK (chain IN ('private','public')),
+        audit_event_hash         TEXT,
+        created_at               INTEGER NOT NULL,
+        last_hit_at              INTEGER,
+        hit_count                INTEGER NOT NULL DEFAULT 0,
+        audit_mode               TEXT NOT NULL DEFAULT 'UNGROUNDED'
+                                    CHECK (audit_mode IN ('STRICT','HYBRID','UNGROUNDED','CANONICAL_PROJECTION')),
+        n_quotes                 INTEGER NOT NULL DEFAULT 0,
+        n_verified               INTEGER NOT NULL DEFAULT 0,
+        unverified_quotes        TEXT,
+        verifier_method          TEXT NOT NULL DEFAULT 'none'
+                                    CHECK (verifier_method IN ('quote','span','entity','paraphrase','claim_lattice','canonical_projection','none')),
+        run_dag_root             TEXT,
+        run_dag_blob             TEXT
+    )
+    """
+    conn.execute("BEGIN IMMEDIATE")
+    try:
+        conn.execute(new_create)
+        conn.execute(
+            "INSERT INTO providence_cache_new "
+            "(cache_key, source_root, document_uri, question_hash, question_text, "
+            " answer_text, merkle_proof, model_profile_hash, conversation_hash, "
+            " governance_policy_hash, schema_version, canonicalization_version, "
+            " chunking_version, falsification_state, chain, audit_event_hash, "
+            " created_at, last_hit_at, hit_count, audit_mode, n_quotes, "
+            " n_verified, unverified_quotes, verifier_method, "
+            " run_dag_root, run_dag_blob) "
+            "SELECT "
+            " cache_key, source_root, document_uri, question_hash, question_text, "
+            " answer_text, merkle_proof, model_profile_hash, conversation_hash, "
+            " governance_policy_hash, schema_version, canonicalization_version, "
+            " chunking_version, falsification_state, chain, audit_event_hash, "
+            " created_at, last_hit_at, hit_count, audit_mode, n_quotes, "
+            " n_verified, unverified_quotes, verifier_method, "
+            " run_dag_root, run_dag_blob "
+            "FROM providence_cache"
+        )
+        conn.execute("DROP TABLE providence_cache")
+        conn.execute("ALTER TABLE providence_cache_new RENAME TO providence_cache")
+        conn.execute("COMMIT")
+    except Exception:
+        conn.execute("ROLLBACK")
+        raise
+
+
 def _rebuild_providence_cache_ungrounded(conn: sqlite3.Connection) -> None:
     """One-time table rebuild: rename audit_mode value VISUAL → UNGROUNDED.
 

docs/TICKETS.md

@@ -61,8 +61,9 @@ Newest first. Update on every open/close.
 
 | ID       | Title                                          | Status                | Opened     | Directive |
 |----------|------------------------------------------------|-----------------------|------------|-----------|
-| #000028  | Multi-modality witness for canonical shapes    | open · implementation in progress | 2026-05-08 | —         |
-| #000027  | Canonical projections persist to providence_cache | open · design phase | 2026-05-08 | —         |
+| #000029  | Claim-pack source (axiom/theorem JSON bundles) | open · implementation in progress | 2026-05-09 | —         |
+| #000028  | Multi-modality witness for canonical shapes    | closed · landed 2026-05-09 (STRICT-WITNESSED reachable post-#000027) | 2026-05-08 | —         |
+| #000027  | Canonical projections persist to providence_cache | closed · landed 2026-05-09 | 2026-05-08 | —         |
 | #000026  | Real-shard workload baseline + search latency  | in progress · Phase 1 + 2 landed 2026-05-08 | 2026-05-08 | —         |
 | #000025  | 5F battery (Function · Finetuning · Falsification · Formulate · Feedback Loop) | in progress · Phase 1a landed 2026-05-08 | 2026-05-07 | —         |
 | #000024  | 5T Phase 1b + Dav1DPrometheus vocabulary alignment | closed · landed 2026-05-08 | 2026-05-07 | —         |
@@ -92,4 +93,4 @@ Newest first. Update on every open/close.
 
 ## Next ID
 
-`000029`
+`000030`

docs/tickets/ticket-000027-canonical-projections-in-providence-cache.md

@@ -1,6 +1,6 @@
 # Ticket #000027 — Canonical projections persist to providence_cache
 
-**Status:** open · design phase
+**Status:** closed · landed 2026-05-09
 **Opened:** 2026-05-08
 **Scope:** Treat `canonical_projection` results (arithmetic@v1, logic-kernel@v1,
 and future algebra/calc kernels) as first-class providence records.

docs/tickets/ticket-000028-multi-modality-witness.md

@@ -1,6 +1,6 @@
 # Ticket #000028 — Multi-modality witness for canonical shapes
 
-**Status:** open · implementation in progress
+**Status:** closed · landed 2026-05-09 (cache-leg wired post-#000027; STRICT-WITNESSED reachable)
 **Opened:** 2026-05-08
 **Scope:** When a question matches a canonical shape (arithmetic@v1,
 logic-kernel@v1, future kernels), fan out three independent answer

tests/test_canonical_cache.py

@@ -0,0 +1,436 @@
+"""Tests for #000027 — canonical projection persistence.
+
+Covers the acceptance criteria from the ticket §7:
+
+1. First call writes one providence_cache row + one
+   ``providence_canonical`` audit event.
+2. Second call: cache hit; ``hit_count`` increments; kernel not
+   re-run.
+3. Audit-chain integrity intact after mixed canonical/RAG writes.
+4. Bumping pi_star_ref (``@v1`` → ``@v2``) routes new questions to
+   a fresh row; old rows remain in DB but unreachable via the live
+   cache_key.
+5. Bumping CHUNKING_VERSION does NOT stale canonical rows (they
+   pin ``"n/a-canonical"``).
+6. Distinct pi_star_refs namespace separately.
+7. Strict vs equivalence_class dedup behavior for canonical rows.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from arborist.qa.canonical_cache import (
+    CANONICAL_AUDIT_EVENT_TYPE,
+    canonical_cache_key,
+    canonical_synthetic_source_root,
+    lookup_canonical,
+    lookup_or_persist,
+    persist_canonical,
+)
+from arborist.qa.client import StubClient
+from arborist.qa.query import query
+from arborist.store import connect
+
+
+# ----- low-level: cache_key shape ------------------------------------------
+
+
+def test_canonical_cache_key_is_deterministic():
+    a = canonical_cache_key(
+        question="0.1 + 0.2", pi_star_ref="arithmetic@v1", policy={}
+    )
+    b = canonical_cache_key(
+        question="0.1 + 0.2", pi_star_ref="arithmetic@v1", policy={}
+    )
+    assert a == b
+
+
+def test_canonical_cache_key_distinct_pi_star_refs():
+    """Same question + different pi_star_ref → different cache_key.
+    The synthetic source_root encodes pi_star_ref."""
+    a = canonical_cache_key(
+        question="0.1 + 0.2", pi_star_ref="arithmetic@v1", policy={}
+    )
+    b = canonical_cache_key(
+        question="0.1 + 0.2", pi_star_ref="arithmetic@v2", policy={}
+    )
+    assert a != b
+
+
+def test_canonical_cache_key_distinct_questions():
+    a = canonical_cache_key(
+        question="0.1 + 0.2", pi_star_ref="arithmetic@v1", policy={}
+    )
+    b = canonical_cache_key(
+        question="1 + 1", pi_star_ref="arithmetic@v1", policy={}
+    )
+    assert a != b
+
+
+def test_canonical_cache_key_strict_vs_equivalence_class():
+    """Strict mode keeps "0.1+0.2" and "0.1 + 0.2" distinct;
+    equivalence_class mode collapses them via the trailing-strip /
+    article-strip / case rules in question_hash."""
+    strict_a = canonical_cache_key(
+        question="0.1 + 0.2?", pi_star_ref="arithmetic@v1",
+        policy={}, mode="strict",
+    )
+    strict_b = canonical_cache_key(
+        question="0.1 + 0.2", pi_star_ref="arithmetic@v1",
+        policy={}, mode="strict",
+    )
+    eq_a = canonical_cache_key(
+        question="0.1 + 0.2?", pi_star_ref="arithmetic@v1",
+        policy={}, mode="equivalence_class",
+    )
+    eq_b = canonical_cache_key(
+        question="0.1 + 0.2", pi_star_ref="arithmetic@v1",
+        policy={}, mode="equivalence_class",
+    )
+    assert strict_a != strict_b   # "?" matters in strict mode
+    assert eq_a == eq_b           # collapses in equivalence_class
+
+
+# ----- persistence round-trip ---------------------------------------------
+
+
+def test_persist_then_lookup(tmp_path: Path):
+    db = tmp_path / "qa.db"
+    conn = connect(db)
+    try:
+        ckey = canonical_cache_key(
+            question="0.1 + 0.2", pi_star_ref="arithmetic@v1", policy={}
+        )
+        event_hash, run_dag_root, run_dag = persist_canonical(
+            conn,
+            cache_key_value=ckey,
+            question="0.1 + 0.2",
+            pi_star_ref="arithmetic@v1",
+            canonical_input_bytes=b"0.1 + 0.2",
+            canonical_output_bytes=b"3/10",
+            policy={},
+        )
+        assert len(event_hash) == 64
+        assert len(run_dag_root) == 64
+        assert run_dag["nodes"][1]["stage"] == "canonical_projection"
+
+        row = lookup_canonical(conn, ckey)
+        assert row is not None
+        assert row["audit_mode"] == "CANONICAL_PROJECTION"
+        assert row["verifier_method"] == "canonical_projection"
+        assert row["answer_text"] == "3/10"
+        assert row["audit_event_hash"] == event_hash
+        assert row["run_dag_root"] == run_dag_root
+    finally:
+        conn.close()
+
+
+def test_audit_event_appended(tmp_path: Path):
+    db = tmp_path / "qa.db"
+    conn = connect(db)
+    try:
+        before = conn.execute(
+            "SELECT COUNT(*) FROM audit_events"
+        ).fetchone()[0]
+        ckey = canonical_cache_key(
+            question="0.1 + 0.2", pi_star_ref="arithmetic@v1", policy={}
+        )
+        persist_canonical(
+            conn, cache_key_value=ckey, question="0.1 + 0.2",
+            pi_star_ref="arithmetic@v1",
+            canonical_input_bytes=b"0.1 + 0.2",
+            canonical_output_bytes=b"3/10", policy={},
+        )
+        after = conn.execute(
+            "SELECT COUNT(*) FROM audit_events"
+        ).fetchone()[0]
+        assert after == before + 1
+        latest = conn.execute(
+            "SELECT event_type, body FROM audit_events "
+            "ORDER BY seq DESC LIMIT 1"
+        ).fetchone()
+        assert latest["event_type"] == CANONICAL_AUDIT_EVENT_TYPE
+        body = json.loads(latest["body"])
+        assert body["pi_star_ref"] == "arithmetic@v1"
+        assert body["canonical_output_text"] == "3/10"
+        assert body["kernel_audit_mode"] == "CANONICAL_PROJECTION"
+    finally:
+        conn.close()
+
+
+def test_lookup_or_persist_first_call_misses_second_hits(tmp_path: Path):
+    db = tmp_path / "qa.db"
+    conn = connect(db)
+    try:
+        ckey1, row1, was_hit1, dag1 = lookup_or_persist(
+            conn, question="0.1 + 0.2",
+            pi_star_ref="arithmetic@v1",
+            canonical_output_bytes=b"3/10",
+            policy={},
+        )
+        assert was_hit1 is False
+        assert row1 is None
+        assert dag1["root"]
+
+        ckey2, row2, was_hit2, dag2 = lookup_or_persist(
+            conn, question="0.1 + 0.2",
+            pi_star_ref="arithmetic@v1",
+            canonical_output_bytes=b"3/10",
+            policy={},
+        )
+        assert ckey1 == ckey2
+        assert was_hit2 is True
+        assert row2 is not None
+        assert row2["answer_text"] == "3/10"
+    finally:
+        conn.close()
+
+
+# ----- end-to-end via query() ---------------------------------------------
+
+
+def test_query_writes_then_hits(tmp_path: Path):
+    qa_db = tmp_path / "qa.db"
+    no_shards = tmp_path / "shards"
+
+    first = query(
+        question="0.1 + 0.2",
+        qa_db=qa_db, chat_client=StubClient(""),
+        model_id="m", shards_dir=no_shards,
+    )
+    assert first["status"] == "cache_miss_then_written"
+    assert first["lookup_path"] == "canonical_cache_miss"
+    ckey1 = first["cache_key"]
+    assert ckey1 is not None
+
+    second = query(
+        question="0.1 + 0.2",
+        qa_db=qa_db, chat_client=StubClient(""),
+        model_id="m", shards_dir=no_shards,
+    )
+    assert second["status"] == "cache_hit"
+    assert second["lookup_path"] == "canonical_cache_hit"
+    assert second["cache_key"] == ckey1
+    assert second["answer_text"] == "3/10"
+
+
+def test_hit_count_increments_on_repeated_query(tmp_path: Path):
+    qa_db = tmp_path / "qa.db"
+    no_shards = tmp_path / "shards"
+    for _ in range(3):
+        query(
+            question="0.1 + 0.2",
+            qa_db=qa_db, chat_client=StubClient(""),
+            model_id="m", shards_dir=no_shards,
+        )
+    conn = connect(qa_db)
+    try:
+        row = conn.execute(
+            "SELECT hit_count FROM providence_cache "
+            "WHERE audit_mode = 'CANONICAL_PROJECTION'"
+        ).fetchone()
+        # 1 write + 2 hits; hit_count counts hits only.
+        assert row["hit_count"] == 2
+    finally:
+        conn.close()
+
+
+def test_audit_chain_intact_after_canonical_writes(tmp_path: Path):
+    """audit_events chain (event_hash = sha256(prev || canonical_body))
+    stays unbroken when canonical rows are interleaved."""
+    qa_db = tmp_path / "qa.db"
+    no_shards = tmp_path / "shards"
+    for q in ("0.1 + 0.2", "1 + 1", "A AND B", "0.1 + 0.2"):
+        query(
+            question=q, qa_db=qa_db,
+            chat_client=StubClient(""),
+            model_id="m", shards_dir=no_shards,
+        )
+    conn = connect(qa_db)
+    try:
+        rows = conn.execute(
+            "SELECT seq, event_hash, prev_event_hash, body "
+            "FROM audit_events ORDER BY seq"
+        ).fetchall()
+        import hashlib
+        prev = None
+        for r in rows:
+            h = hashlib.sha256()
+            if r["prev_event_hash"]:
+                h.update(bytes.fromhex(r["prev_event_hash"]))
+            h.update(r["body"].encode("utf-8"))
+            assert h.hexdigest() == r["event_hash"]
+            assert r["prev_event_hash"] == prev
+            prev = r["event_hash"]
+    finally:
+        conn.close()
+
+
+# ----- pi_star_ref version semantics --------------------------------------
+
+
+def test_distinct_pi_star_refs_namespace_separately(tmp_path: Path):
+    """Same question through two different kernels → two distinct rows."""
+    db = tmp_path / "qa.db"
+    conn = connect(db)
+    try:
+        ckey_a, _, miss_a, _ = lookup_or_persist(
+            conn, question="0.1 + 0.2",
+            pi_star_ref="arithmetic@v1",
+            canonical_output_bytes=b"3/10", policy={},
+        )
+        ckey_b, _, miss_b, _ = lookup_or_persist(
+            conn, question="0.1 + 0.2",
+            pi_star_ref="arithmetic@v2",  # hypothetical bump
+            canonical_output_bytes=b"3/10", policy={},
+        )
+        assert miss_a is False
+        assert miss_b is False
+        assert ckey_a != ckey_b
+        rows = conn.execute(
+            "SELECT cache_key FROM providence_cache "
+            "WHERE audit_mode = 'CANONICAL_PROJECTION'"
+        ).fetchall()
+        assert len({r["cache_key"] for r in rows}) == 2
+    finally:
+        conn.close()
+
+
+def test_pi_star_version_bump_orphans_old_row(tmp_path: Path):
+    """Write under @v1; lookup under @v2 (different synthetic
+    source_root) misses; old @v1 row remains in DB."""
+    db = tmp_path / "qa.db"
+    conn = connect(db)
+    try:
+        # Write under @v1.
+        lookup_or_persist(
+            conn, question="0.1 + 0.2",
+            pi_star_ref="arithmetic@v1",
+            canonical_output_bytes=b"3/10", policy={},
+        )
+        # Lookup under @v2 — different cache_key → miss.
+        ckey_v2 = canonical_cache_key(
+            question="0.1 + 0.2", pi_star_ref="arithmetic@v2",
+            policy={},
+        )
+        assert lookup_canonical(conn, ckey_v2) is None
+        # Old @v1 row still present; just unreachable via @v2 lookup.
+        ckey_v1 = canonical_cache_key(
+            question="0.1 + 0.2", pi_star_ref="arithmetic@v1",
+            policy={},
+        )
+        assert lookup_canonical(conn, ckey_v1) is not None
+    finally:
+        conn.close()
+
+
+# ----- governance + chunking_version ---------------------------------------
+
+
+def test_chunking_version_bump_does_not_stale_canonical(tmp_path: Path):
+    """Canonical rows pin chunking_version='n/a-canonical' so a
+    chunker bump on the wikipedia path doesn't mass-stale math
+    answers. Verify by inspecting the persisted column directly."""
+    db = tmp_path / "qa.db"
+    conn = connect(db)
+    try:
+        ckey = canonical_cache_key(
+            question="0.1 + 0.2", pi_star_ref="arithmetic@v1", policy={}
+        )
+        persist_canonical(
+            conn, cache_key_value=ckey, question="0.1 + 0.2",
+            pi_star_ref="arithmetic@v1",
+            canonical_input_bytes=b"0.1 + 0.2",
+            canonical_output_bytes=b"3/10", policy={},
+        )
+        row = lookup_canonical(conn, ckey)
+        assert row is not None
+        # The persisted chunking_version is the canonical sentinel,
+        # NOT the live CHUNKING_VERSION constant.
+        cv = conn.execute(
+            "SELECT chunking_version FROM providence_cache "
+            "WHERE cache_key = ?", (ckey,),
+        ).fetchone()[0]
+        assert cv == "n/a-canonical"
+    finally:
+        conn.close()
+
+
+def test_governance_policy_change_invalidates_lookup(tmp_path: Path):
+    """Different policy → different governance_policy_hash → different
+    cache_key. The old row stays but the lookup misses."""
+    db = tmp_path / "qa.db"
+    conn = connect(db)
+    try:
+        # Write under empty policy.
+        lookup_or_persist(
+            conn, question="0.1 + 0.2",
+            pi_star_ref="arithmetic@v1",
+            canonical_output_bytes=b"3/10",
+            policy={},
+        )
+        # Lookup under different policy — different ghash → miss.
+        ckey_alt = canonical_cache_key(
+            question="0.1 + 0.2",
+            pi_star_ref="arithmetic@v1",
+            policy={"answer_mode": "claim_lattice"},
+        )
+        assert lookup_canonical(conn, ckey_alt) is None
+    finally:
+        conn.close()
+
+
+# ----- arborist canon stays transient -------------------------------------
+
+
+def test_arborist_canon_does_not_persist(tmp_path: Path):
+    """`arborist canon <key> "<input>"` is a one-shot probe; it
+    bypasses query() entirely and writes nothing to providence_cache.
+    Confirms the boundary the ticket §2.6 promises."""
+    import subprocess
+    import sys
+
+    qa_db = tmp_path / "qa.db"
+    # Make sure no rows exist before the canon call.
+    conn = connect(qa_db)
+    try:
+        before = conn.execute(
+            "SELECT COUNT(*) FROM providence_cache"
+        ).fetchone()[0]
+    finally:
+        conn.close()
+
+    r = subprocess.run(
+        [sys.executable, "-m", "arborist.cli", "canon",
+         "arithmetic@v1", "0.1 + 0.2"],
+        capture_output=True, text=True, check=False,
+    )
+    assert r.returncode == 0
+    assert r.stdout.strip() == "3/10"
+
+    # No row should have been written by `canon` (it's transient by
+    # design — the canon CLI doesn't take a qa_db). Confirm the
+    # qa_db count is unchanged. We verify by re-checking the same
+    # tmp_path qa_db (which is unrelated to the default qa.db that
+    # `canon` doesn't write to anyway).
+    conn = connect(qa_db)
+    try:
+        after = conn.execute(
+            "SELECT COUNT(*) FROM providence_cache"
+        ).fetchone()[0]
+    finally:
+        conn.close()
+    assert after == before
+
+
+# ----- synthetic dimension shape ------------------------------------------
+
+
+def test_synthetic_source_root_encodes_pi_star_ref():
+    a = canonical_synthetic_source_root("arithmetic@v1")
+    b = canonical_synthetic_source_root("arithmetic@v2")
+    c = canonical_synthetic_source_root("logic-kernel@v1")
+    assert len({a, b, c}) == 3
+    assert len(a) == 64  # sha256 hex

tests/test_canonical_projection.py

@@ -85,8 +85,10 @@ def test_preflight_returns_none_on_pi_star_error():
 
 
 def test_query_math_short_circuits_with_no_shards(tmp_path: Path):
-    """Math-shaped question returns canonical_projection without
-    needing a shard or hitting the LLM."""
+    """Math-shaped question persists a canonical row + answers without
+    needing a shard or hitting the LLM. Status is cache_miss_then_written
+    (first call writes the row); audit_mode/verifier_method are the
+    canonical-projection tokens; cache_key is real."""
     qa_db = tmp_path / "qa.db"
     no_shards = tmp_path / "shards"  # absent on purpose
     result = query(
@@ -96,14 +98,16 @@ def test_query_math_short_circuits_with_no_shards(tmp_path: Path):
         model_id="test/model",
         shards_dir=no_shards,
     )
-    assert result["status"] == "canonical_projection"
+    assert result["status"] == "cache_miss_then_written"
     assert result["audit_mode"] == "CANONICAL_PROJECTION"
     assert result["verifier_method"] == "canonical_projection"
     assert result["pi_star_ref"] == "arithmetic@v1"
     assert result["answer_text"] == "3/10"
-    assert result["lookup_path"] == "preflight_canonical"
+    assert result["lookup_path"] == "canonical_cache_miss"
     assert result["sources"] == []
-    assert result["cache_key"] is None
+    assert result["cache_key"] is not None
+    assert result["audit_event_hash"] is not None
+    assert result["run_dag_root"] is not None
 
 
 def test_query_logic_short_circuits(tmp_path: Path):
@@ -116,11 +120,31 @@ def test_query_logic_short_circuits(tmp_path: Path):
         model_id="test/model",
         shards_dir=no_shards,
     )
-    assert result["status"] == "canonical_projection"
+    assert result["status"] == "cache_miss_then_written"
     assert result["pi_star_ref"] == "logic-kernel@v1"
     assert result["answer_text"] == "(NOT A OR B)"
 
 
+def test_query_canonical_transient_mode(tmp_path: Path):
+    """Operators can disable persistence per-call via
+    policy['canonical_projection_preflight_persist']=False; the legacy
+    transient render-only behavior comes back (status='canonical_projection',
+    cache_key=None, no audit-chain entry)."""
+    qa_db = tmp_path / "qa.db"
+    no_shards = tmp_path / "shards"
+    result = query(
+        question="0.1 + 0.2",
+        qa_db=qa_db,
+        chat_client=StubClient(""),
+        model_id="test/model",
+        shards_dir=no_shards,
+        policy={"canonical_projection_preflight_persist": False},
+    )
+    assert result["status"] == "canonical_projection"
+    assert result["lookup_path"] == "preflight_canonical"
+    assert result["cache_key"] is None
+
+
 def test_query_contrapositive_collapses_to_same_canonical(tmp_path: Path):
     """A IMPL B and (NOT B) IMPL (NOT A) are the same equivalence class."""
     qa_db = tmp_path / "qa.db"

tests/test_directives.py

@@ -444,9 +444,11 @@ def test_d7_renderer_keeps_strict_for_pinned_span_methods():
 
 
 def test_d7_audit_mode_enum_canonical_set():
-    """Schema column must keep the canonical 3-value enum so v9.8
-    cache_key invariants hold. Renderer-level relabel (above)
-    doesn't touch this."""
+    """Schema column carries the v9.8 trichotomy plus
+    CANONICAL_PROJECTION (#000027 — deterministic π* answer rows).
+    Renderer-level relabels (4-rung ladder, CANONICAL display) do
+    NOT touch this enum. Adding a new admissibility class here is a
+    governance event and bumps the schema-version conversation."""
     from arborist.store import SCHEMA_SQL
 
     match = re.search(
@@ -458,10 +460,12 @@ def test_d7_audit_mode_enum_canonical_set():
     enum_values = sorted(
         s.strip().strip("'\"") for s in match.group(1).split(",")
     )
-    assert enum_values == ["HYBRID", "STRICT", "UNGROUNDED"], (
+    assert enum_values == [
+        "CANONICAL_PROJECTION", "HYBRID", "STRICT", "UNGROUNDED",
+    ], (
         f"audit_mode enum drifted to {enum_values} — schema column "
-        f"must stay {{STRICT, HYBRID, UNGROUNDED}}; rendered labels "
-        f"are display-layer only."
+        f"must stay {{STRICT, HYBRID, UNGROUNDED, CANONICAL_PROJECTION}}; "
+        f"rendered labels are display-layer only."
     )
 
 

tests/test_witness.py

@@ -315,7 +315,9 @@ def test_witness_parallel_not_sequential():
 
 
 def test_query_canonical_path_off_by_default(tmp_path):
-    """Default policy: canonical_witness_enabled is False; no LLM call."""
+    """Default policy: canonical_witness_enabled is False; no LLM call.
+    Persistence is also default-on (#000027), so first call writes
+    a row → status='cache_miss_then_written'."""
     from arborist.qa.query import query
 
     client = StubClient(answer="<should-not-be-called>")
@@ -325,7 +327,7 @@ def test_query_canonical_path_off_by_default(tmp_path):
         chat_client=client,
         model_id="stub",
     )
-    assert result["status"] == "canonical_projection"
+    assert result["status"] == "cache_miss_then_written"
     assert result["audit_mode"] == "CANONICAL_PROJECTION"
     assert result.get("witness") is None
     # Critical: no LLM round-trip happened.
@@ -346,9 +348,42 @@ def test_query_canonical_with_witness_calls_llm(tmp_path):
         model_id="stub",
         policy=policy,
     )
-    assert result["status"] == "canonical_projection"
+    assert result["status"] == "cache_miss_then_written"
     witness = result.get("witness")
     assert witness is not None
+    # First call: no prior cache row to compare against → cache leg
+    # is ABSENT; agreement is kernel↔LLM only.
     assert witness["agreement_label"] == "KERNEL-LLM-AGREE"
     # The LLM was actually called.
     assert len(client.calls) == 1
+
+
+def test_query_canonical_witness_reaches_strict_after_persist(tmp_path):
+    """Post-#000027 + cache-leg wire: a SECOND witness call (after the
+    first writes a row) compares kernel + cache + LLM, all three
+    byte-equal → STRICT-WITNESSED. This was structurally unreachable
+    before #000027 landed — the cache leg always returned None."""
+    from arborist.qa.query import DEFAULT_QUERY_POLICY, query
+
+    client = StubClient(answer="3/10")
+    policy = dict(DEFAULT_QUERY_POLICY)
+    policy["canonical_witness_enabled"] = True
+    qa_db = tmp_path / "qa.db"
+
+    # First call: persists row; cache leg ABSENT; KERNEL-LLM-AGREE.
+    first = query(
+        question="0.1 + 0.2", qa_db=qa_db,
+        chat_client=client, model_id="stub", policy=policy,
+    )
+    assert first["status"] == "cache_miss_then_written"
+    assert first["witness"]["agreement_label"] == "KERNEL-LLM-AGREE"
+
+    # Second call: cache hits (no LLM run from query() path; witness
+    # still calls LLM separately when enabled). Cache leg now
+    # populated with the persisted bytes → STRICT-WITNESSED.
+    second = query(
+        question="0.1 + 0.2", qa_db=qa_db,
+        chat_client=client, model_id="stub", policy=policy,
+    )
+    assert second["status"] == "cache_hit"
+    assert second["witness"]["agreement_label"] == "STRICT-WITNESSED"