feat(operator): forge projection wired into the failure path (T5.15 slice 5)

The final orphan: testing_mode.project_failure_to_forge was TEST_PROVEN with zero runtime callers. _project_failure_to_forge() invokes it from the operator failure path using the already-wired forge client (create_forge_client → NullAdapter when unconfigured = dormant by construction); a thin shim adapts create_issue to the projection's create_or_patch_release contract; ForgeUnavailable degrades to a marker, never breaking the failure path (rule 19P). Three contracts: projection fires, dormant-not-error, forge-down-never-breaks. Completion wave wire-up complete: all 5 orphans now have live callers.
feat(scaffold): iteration impact-delta recorded at close (T5.15 slice 4)
2026-10-06 04:09:22 +01:00 · 2026-10-06 03:54:01 +01:00 · 2026-10-06 03:46:24 +01:00 · 2026-10-06 03:27:08 +01:00 · 2026-10-06 03:27:02 +01:00 · 2026-10-06 03:19:13 +01:00
26 changed files with 10059 additions and 8446 deletions
@@ -1,6 +1,6 @@
 # Rune

-**Current release: v0.97.0**
+**Current release: v0.99.5**

 **Rune is a persistent personal agent designed to accumulate experience, develop reusable competence, and remain recognisably the same agent even when its underlying foundation model changes.**

@@ -1,318 +1,319 @@
 {
-  "_comment": "GENERATED by rune.spine \u2014 compatibility view for legacy umbrella consumers. DO NOT EDIT: structural truth lives in store/spine/current.json; edits here are flagged by spine drift.",
-  "_generated_by_spine": true,
-  "generated_at": "2026-09-16T18:23:38Z",
-  "umbrellas": {
-    "capability": {
-      "description": "Tool selection and execution authority",
-      "modules": [
-        "adaptive_router",
-        "builder_bridge",
-        "builder_bridge_cli",
-        "capability_broker",
-        "connection_pool",
-        "core_server",
-        "cs_loader",
-        "evidence_probe",
-        "forge",
-        "forge_client",
-        "intent_continuity",
-        "local_tool_runtime",
-        "message_queue",
-        "native_units",
-        "openai_compat_transport",
-        "provider_context",
-        "recall_uplift",
-        "route_risk_flags",
-        "tool_executor",
-        "tool_protocol"
-      ],
-      "depends_on": [
-        "memory"
-      ],
-      "mirrors": [],
-      "canaries": [
-        "canary_native",
-        "canary_p0_05",
-        "canary_p1_01",
-        "canary_rune_entry",
-        "canary_rune_serve"
-      ],
-      "go_surfaces": [
-        "rune-entry"
-      ],
-      "configs": [],
-      "rust_crates": [
-        "capability_scoring"
-      ]
-    },
-    "governance": {
-      "description": "Quality control, drift detection, promotion, audit",
-      "modules": [
-        "audit_ledger",
-        "build_pipeline_policy",
-        "cc_loader",
-        "claim_audit",
-        "contracts",
-        "curator",
-        "drift_core",
-        "drift_ledger",
-        "drift_observability",
-        "failure_auto_filer",
-        "programme_gates",
-        "sc_loader"
-      ],
-      "depends_on": [],
-      "mirrors": [],
-      "canaries": [
-        "canary_full_system",
-        "canary_schema_migration",
-        "canary_audit_ledger",
-        "canary_spine_meta"
-      ],
-      "rust_crates": [
-        "evidence_provenance",
-        "scaffold_core"
-      ],
-      "configs": [
-        ".coveragerc",
-        "go-version.txt",
-        "lint-baseline.json",
-        "pyright-baseline.json"
-      ]
-    },
-    "growth": {
-      "description": "Learning and improvement",
-      "modules": [
-        "autonomy_loop",
-        "cognitive_loop",
-        "dream_cycle",
-        "elastic_cognition",
-        "goal",
-        "growth_delta",
-        "growth_simulator",
-        "self_improve",
-        "subconscious_pass"
-      ],
-      "depends_on": [
-        "governance",
-        "memory"
-      ],
-      "mirrors": [],
-      "canaries": [
-        "canary_full_system"
-      ]
-    },
-    "identity": {
-      "description": "Who the agent is \u2014 persona, voice, anchors",
-      "modules": [
-        "behavior",
-        "identity_anchors",
-        "persona_definition",
-        "persona_gate",
-        "prompt_sync",
-        "voice_hybrid",
-        "voice_injector",
-        "voice_sync",
-        "voicebox"
-      ],
-      "depends_on": [
-        "governance"
-      ],
-      "mirrors": [],
-      "canaries": [
-        "canary_full_system"
-      ]
-    },
-    "inference": {
-      "description": "Model routing and outbound requests",
-      "modules": [
-        "inference_trace",
-        "model_connect",
-        "model_profiles",
-        "model_registry",
-        "model_routing",
-        "provider_budget",
-        "provider_registry",
-        "response_adapter",
-        "standardized_providers",
-        "tc_loader",
-        "token_optimizer"
-      ],
-      "depends_on": [
-        "capability"
-      ],
-      "mirrors": [],
-      "canaries": [
-        "canary_p1_02",
-        "canary_p1_06",
-        "canary_resource_authority"
-      ],
-      "rust_crates": [
-        "token_counting"
-      ]
-    },
-    "memory": {
-      "description": "Storing, retrieving, and managing knowledge",
-      "modules": [
-        "abstraction",
-        "cache",
-        "drift_cache",
-        "drift_compression",
-        "ffi_query_memory_smoke",
-        "graph_memory",
-        "graph_shard_bridge",
-        "memory_consolidation",
-        "memory_env",
-        "memory_selector",
-        "memory_store",
-        "mems_loader",
-        "mr_loader",
-        "ms_loader",
-        "query_memory",
-        "recall_router",
-        "retrieval_benchmark",
-        "shard_manager",
-        "tier_manager",
-        "warm_neuron_store"
-      ],
-      "depends_on": [
-        "governance"
-      ],
-      "mirrors": [],
-      "canaries": [
-        "canary_full_system",
-        "canary_memory_synthesis",
-        "canary_query_memory"
-      ],
-      "rust_crates": [
-        "memory_retrieval",
-        "memory_scoring",
-        "memory_storage",
-        "query_memory"
-      ]
-    },
-    "orchestration": {
-      "description": "Task lifecycle, operator, TUI presentation",
-      "modules": [
-        "auto_detect",
-        "build_mode",
-        "build_orchestrator",
-        "chat_bubble",
-        "cron_engine",
-        "custodian",
-        "dependency_tracker",
-        "event_bus",
-        "event_envelope",
-        "gitea_auth",
-        "gitea_pipeline",
-        "gitea_provision",
-        "hook_engine",
-        "message_widgets",
-        "microkernel",
-        "night_mode",
-        "outcome_ledger",
-        "parallel_processor",
-        "pins",
-        "progressive_escalation",
-        "rune_graphic",
-        "rune_operator",
-        "rune_profiler",
-        "rune_spinner",
-        "rune_tui",
-        "safety_override",
-        "sleep_time",
-        "subsystem_health",
-        "system_registry",
-        "task_gate",
-        "task_persist",
-        "tiered_executor",
-        "time_awareness",
-        "tui_transport",
-        "tui_validator",
-        "umbrella_update",
-        "verification_layer",
-        "warm_router",
-        "work_manager"
-      ],
-      "depends_on": [
-        "capability",
-        "growth",
-        "identity",
-        "inference",
-        "perception"
-      ],
-      "mirrors": [],
-      "canaries": [
-        "canary_cognitive_strata",
-        "canary_conversation_state",
-        "canary_gitea_pipeline",
-        "canary_orchestration_foundation",
-        "canary_p1_07",
-        "canary_surface_parity",
-        "canary_tui_drift",
-        "canary_tui_transport"
-      ]
-    },
-    "perception": {
-      "description": "Input processing and context assembly",
-      "modules": [
-        "attention_capsule",
-        "auto_context",
-        "auto_tasks",
-        "behaviour_ledger",
-        "build_basis",
-        "capability_fit",
-        "capability_ledger",
-        "cognitive_budgeter",
-        "construction_activation",
-        "construction_commands",
-        "construction_decision",
-        "construction_depth",
-        "construction_intent",
-        "construction_interface",
-        "construction_mode",
-        "construction_pattern",
-        "construction_patterns_builtin",
-        "construction_projection",
-        "context_broker",
-        "context_pressure",
-        "drift_resume",
-        "event_wait",
-        "failure_recovery",
-        "graph_projection",
-        "impact_preview",
-        "iteration",
-        "layout_contract",
-        "programme_continuation",
-        "project_registry",
-        "release_notes",
-        "salience_gate",
-        "scaffold",
-        "spine",
-        "spine_loader",
-        "stage_guards",
-        "structural_migration",
-        "structural_model",
-        "structural_store",
-        "testing_mode",
-        "toolchain",
-        "tui_backlog",
-        "wait_guard",
-        "wait_guard_corpus",
-        "witness"
-      ],
-      "depends_on": [
-        "memory"
-      ],
-      "mirrors": [],
-      "canaries": [
-        "canary_context_broker",
-        "canary_full_system"
-      ],
-      "rust_crates": [
-        "rune_spine"
-      ]
-    }
+ "_comment": "GENERATED by rune.spine \u2014 compatibility view for legacy umbrella consumers. DO NOT EDIT: structural truth lives in store/spine/current.json; edits here are flagged by spine drift.",
+ "_generated_by_spine": true,
+ "generated_at": "2026-09-16T18:23:38Z",
+ "umbrellas": {
+  "capability": {
+   "canaries": [
+    "canary_native",
+    "canary_p0_05",
+    "canary_p1_01",
+    "canary_rune_entry",
+    "canary_rune_serve"
+   ],
+   "configs": [],
+   "depends_on": [
+    "memory"
+   ],
+   "description": "Tool selection and execution authority",
+   "go_surfaces": [
+    "rune-entry"
+   ],
+   "mirrors": [],
+   "modules": [
+    "adaptive_router",
+    "builder_bridge",
+    "builder_bridge_cli",
+    "capability_broker",
+    "connection_pool",
+    "core_server",
+    "cs_loader",
+    "evidence_probe",
+    "forge",
+    "forge_client",
+    "intent_continuity",
+    "local_tool_runtime",
+    "message_queue",
+    "native_units",
+    "openai_compat_transport",
+    "provider_context",
+    "recall_uplift",
+    "route_risk_flags",
+    "tool_executor",
+    "tool_protocol"
+   ],
+   "rust_crates": [
+    "capability_scoring"
+   ]
+  },
+  "governance": {
+   "canaries": [
+    "canary_full_system",
+    "canary_schema_migration",
+    "canary_audit_ledger",
+    "canary_spine_meta"
+   ],
+   "configs": [
+    ".coveragerc",
+    "go-version.txt",
+    "lint-baseline.json",
+    "pyright-baseline.json"
+   ],
+   "depends_on": [],
+   "description": "Quality control, drift detection, promotion, audit",
+   "mirrors": [],
+   "modules": [
+    "audit_ledger",
+    "build_pipeline_policy",
+    "cc_loader",
+    "claim_audit",
+    "contracts",
+    "curator",
+    "drift_core",
+    "drift_ledger",
+    "drift_observability",
+    "failure_auto_filer",
+    "programme_gates",
+    "sc_loader"
+   ],
+   "rust_crates": [
+    "evidence_provenance",
+    "scaffold_core"
+   ]
+  },
+  "growth": {
+   "canaries": [
+    "canary_full_system"
+   ],
+   "depends_on": [
+    "governance",
+    "memory"
+   ],
+   "description": "Learning and improvement",
+   "mirrors": [],
+   "modules": [
+    "autonomy_loop",
+    "cognitive_loop",
+    "dream_cycle",
+    "elastic_cognition",
+    "goal",
+    "growth_delta",
+    "growth_simulator",
+    "self_improve",
+    "subconscious_pass"
+   ]
+  },
+  "identity": {
+   "canaries": [
+    "canary_full_system"
+   ],
+   "depends_on": [
+    "governance"
+   ],
+   "description": "Who the agent is \u2014 persona, voice, anchors",
+   "mirrors": [],
+   "modules": [
+    "behavior",
+    "identity_anchors",
+    "persona_definition",
+    "persona_gate",
+    "prompt_sync",
+    "voice_hybrid",
+    "voice_injector",
+    "voice_sync",
+    "voicebox"
+   ]
+  },
+  "inference": {
+   "canaries": [
+    "canary_p1_02",
+    "canary_p1_06",
+    "canary_resource_authority"
+   ],
+   "depends_on": [
+    "capability"
+   ],
+   "description": "Model routing and outbound requests",
+   "mirrors": [],
+   "modules": [
+    "inference_trace",
+    "model_connect",
+    "model_profiles",
+    "model_registry",
+    "model_routing",
+    "provider_budget",
+    "provider_registry",
+    "response_adapter",
+    "standardized_providers",
+    "tc_loader",
+    "token_optimizer"
+   ],
+   "rust_crates": [
+    "token_counting"
+   ]
+  },
+  "memory": {
+   "canaries": [
+    "canary_full_system",
+    "canary_memory_synthesis",
+    "canary_query_memory"
+   ],
+   "depends_on": [
+    "governance"
+   ],
+   "description": "Storing, retrieving, and managing knowledge",
+   "mirrors": [],
+   "modules": [
+    "abstraction",
+    "cache",
+    "drift_cache",
+    "drift_compression",
+    "ffi_query_memory_smoke",
+    "graph_memory",
+    "graph_shard_bridge",
+    "memory_consolidation",
+    "memory_env",
+    "memory_selector",
+    "memory_store",
+    "mems_loader",
+    "mr_loader",
+    "ms_loader",
+    "query_memory",
+    "recall_router",
+    "retrieval_benchmark",
+    "shard_manager",
+    "tier_manager",
+    "warm_neuron_store"
+   ],
+   "rust_crates": [
+    "memory_retrieval",
+    "memory_scoring",
+    "memory_storage",
+    "query_memory"
+   ]
+  },
+  "orchestration": {
+   "canaries": [
+    "canary_cognitive_strata",
+    "canary_conversation_state",
+    "canary_gitea_pipeline",
+    "canary_orchestration_foundation",
+    "canary_p1_07",
+    "canary_surface_parity",
+    "canary_tui_drift",
+    "canary_tui_transport"
+   ],
+   "depends_on": [
+    "capability",
+    "growth",
+    "identity",
+    "inference",
+    "perception"
+   ],
+   "description": "Task lifecycle, operator, TUI presentation",
+   "mirrors": [],
+   "modules": [
+    "auto_detect",
+    "build_mode",
+    "build_orchestrator",
+    "chat_bubble",
+    "cron_engine",
+    "custodian",
+    "dependency_tracker",
+    "event_bus",
+    "event_envelope",
+    "gitea_auth",
+    "gitea_pipeline",
+    "gitea_provision",
+    "hook_engine",
+    "message_widgets",
+    "microkernel",
+    "night_mode",
+    "outcome_ledger",
+    "parallel_processor",
+    "pins",
+    "progressive_escalation",
+    "rune_graphic",
+    "rune_operator",
+    "rune_profiler",
+    "rune_spinner",
+    "rune_tui",
+    "safety_override",
+    "sleep_time",
+    "subsystem_health",
+    "system_registry",
+    "task_gate",
+    "task_persist",
+    "tiered_executor",
+    "time_awareness",
+    "tui_transport",
+    "tui_validator",
+    "umbrella_update",
+    "verification_layer",
+    "warm_router",
+    "work_manager"
+   ]
+  },
+  "perception": {
+   "canaries": [
+    "canary_context_broker",
+    "canary_full_system"
+   ],
+   "depends_on": [
+    "memory"
+   ],
+   "description": "Input processing and context assembly",
+   "mirrors": [],
+   "modules": [
+    "attention_capsule",
+    "auto_context",
+    "auto_tasks",
+    "behaviour_ledger",
+    "build_basis",
+    "capability_fit",
+    "capability_ledger",
+    "cognitive_budgeter",
+    "construction_activation",
+    "construction_commands",
+    "construction_decision",
+    "construction_depth",
+    "construction_intent",
+    "construction_interface",
+    "construction_mode",
+    "construction_pattern",
+    "construction_patterns_builtin",
+    "construction_projection",
+    "context_broker",
+    "context_pressure",
+    "drift_resume",
+    "event_wait",
+    "failure_recovery",
+    "graph_projection",
+    "impact_preview",
+    "iteration",
+    "layout_contract",
+    "programme_continuation",
+    "project_registry",
+    "release_notes",
+    "salience_gate",
+    "scaffold",
+    "spine",
+    "spine_language",
+    "spine_loader",
+    "stage_guards",
+    "structural_migration",
+    "structural_model",
+    "structural_store",
+    "testing_mode",
+    "toolchain",
+    "tui_backlog",
+    "wait_guard",
+    "wait_guard_corpus",
+    "witness"
+   ],
+   "rust_crates": [
+    "rune_spine"
+   ]
  }
+ }
 }
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

 [project]
 name = "rune"
-version = "0.97.0"
+version = "0.99.5"
 description = "Rune — a stateful coding assistant with persistent memory, orchestration, and TUI"
 requires-python = ">=3.10"

@@ -6,7 +6,7 @@ This is the main entry point for the Rune program.
 import json
 from pathlib import Path

-__version__ = "0.97.0"
+__version__ = "0.99.5"

 # One build identity (handoff 2026-09-10 §4): the semantic version above is
 # the repo-level truth (auto-bumped at merge); the DEPLOYED revision (exact
@@ -43,6 +43,42 @@ LOCAL_MODEL = "qwen2.5:3b"
 OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434")


+_ABSTRACTION_STOPWORDS_FROZEN = frozenset(
+    "the a an is are was were be been being have has had do does did will "
+    "would could should may might shall can to of in for on with at by from "
+    "as into through during before after this that these those it its what "
+    "which who how when where why and or but not no".split()
+)
+
+
+def _ABSTRACTION_STOPWORDS() -> list:
+    """Gather seam (W3.4): the fact-extraction stopword table as a sorted
+    list — data received by the kernel; identical filter on both engines."""
+    return sorted(_ABSTRACTION_STOPWORDS_FROZEN)
+
+
+def _core():
+    """The Rust cognition_core when available (strangler: Python fallback
+    load-bearing). RUNE_COGNITION_CORE=python pins the fallback."""
+    return None if cc_loader_pin() else _cc_probe()
+
+
+def _cc_probe():
+    from rune import cc_loader as _cl
+    try:
+        return _cl if _cl.available() else None
+    except Exception:  # noqa: BLE001, S110 — loader probe never crashes the fallback
+        return None
+
+
+def cc_loader_pin() -> bool:
+    """The env pin: RUNE_COGNITION_CORE=python → kernels OFF (shared probe).
+"""
+    import os
+
+    return os.environ.get("RUNE_COGNITION_CORE", "auto") == "python"
+
+
 def _now():
    return datetime.now(timezone.utc).isoformat()

@@ -241,17 +277,41 @@ Extract 1-5 key facts. Be precise, not verbose."""
        response = llm_call(prompt, max_tokens=800)

        # Parse response
+        facts: list = []
        try:
            # Try to extract JSON from response
            start = response.find("[")
            end = response.rfind("]") + 1
            if start >= 0 and end > start:
                facts = json.loads(response[start:end])
-            else:
-                facts = []
        except json.JSONDecodeError:
            facts = []

+        # LLM unavailable / unparseable → DETERMINISTIC KERNEL FALLBACK
+        # (W3.4: cc_abstraction_extract_facts — sentence split, stopwords
+        # filtered, min-length gate; the same facts schema it already emits:
+        # text/words/content_words mapped onto what= text with empty dims).
+        if not facts:
+            core = _core()
+            if core is not None:
+                try:
+                    from rune import cc_loader as _cl
+
+                    raw = _cl.abstraction_extract_facts(
+                        episode_text, _ABSTRACTION_STOPWORDS(), 10)
+                    for item in raw or []:
+                        text_frag = str(item.get("text", "")).strip()
+                        if not text_frag:
+                            continue
+                        facts.append({
+                            "what": text_frag, "when": "", "where": "", "who": "",
+                            "why": "", "type": "experience", "entities": [],
+                            "kernel": True,
+                            "content_words": item.get("content_words"),
+                        })
+                except Exception:  # noqa: BLE001, S110 — kernel failure degrades to empty, intentional silence
+                    pass
+
        if not facts:
            return {"extracted": 0, "error": "LLM did not return valid JSON", "raw": response[:200]}

@@ -233,7 +233,7 @@ def main() -> int:
    )
    check(
        "Rune minor version advertises provisioning and restore verification",
-        '__version__ = "0.97.0"' in rune_version,
+        '__version__ = "0.99.5"' in rune_version,
        failures,
    )
    # #169 policy update: auto-bump on merge is sanctioned (Nick directive:
@@ -21,6 +21,21 @@ _LIB_PATH = Path(
 _lib = None


+def core():
+    """The shared kernel probe (strangler law): the delegation entry the
+    kernel-consuming modules use. RUNE_COGNITION_CORE=python pins the
+    fallback — THIS is where the pin is honored, so every consumer gets it
+    (a module-local probe that ignores the pin silently defeats it)."""
+    import os
+
+    if os.environ.get("RUNE_COGNITION_CORE", "auto") == "python":
+        return None
+    try:
+        return _load()
+    except Exception:  # noqa: BLE001 — build/loader failures mean fallback
+        return None
+
+
 def available() -> bool:
    return _LIB_PATH.is_file()

@@ -171,3 +171,98 @@ def _accept(command_id: str, construction_id: str, correlation_id: str,
            "recorded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ",
                                         time.gmtime()),
            "idempotent_replay": False}
+
+
+def dispatch(command_type: str, *, construction_id: str,
+             correlation_id: str = "", command_id: str = "",
+             payload: dict | None = None,
+             store_path: str = "store/scaffold") -> dict:
+    """The SERVING layer (T5.15 slice 2): validate the command against the
+    canonical store, execute the accepted intent through the Construction
+    Interface, and return the interface's envelope with the command
+    provenance attached.
+
+    Single entry point for `rune scaffold <command>` (the Go proxy) and
+    any future transport. Idempotency: a repeated command_id replays the
+    recorded outcome without re-executing.
+
+    Refusal still returns success=False envelope? NO — a refused command
+    is a VALID protocol outcome: the envelope carries accepted=False and
+    success=True (the dispatch itself worked; the lifecycle refused the
+    intent). Execution failures carry success=False.
+    """
+    iface_path = None
+    if store_path and store_path != "store/scaffold":
+        iface_path = store_path
+    try:
+        from rune import construction_interface as ci
+
+        # store_path carries the serving layer's "path|project_id" contract
+        # (a bare path = last segment is the project id, matching how the
+        # Go proxy and the interface's own default name it).
+        if iface_path:
+            path_part, _, proj_part = iface_path.partition("|")
+            proj = proj_part or path_part.rstrip("/").split("/")[-1]
+            iface = ci.ConstructionInterface(builder="substrate",
+                                             project_store=path_part,
+                                             project_id=proj)
+        else:
+            iface = ci.ConstructionInterface(builder="substrate")
+    except Exception as exc:  # noqa: BLE001 — serving failure is an envelope, not a crash
+        return {"accepted": False, "executed": False,
+                "success": False, "error": f"serving unavailable: {exc}"[:200],
+                "command_type": command_type,
+                "construction_id": construction_id}
+
+    intent = command(command_type, construction_id=construction_id,
+                     correlation_id=correlation_id, command_id=command_id,
+                     payload=payload, scaffold=iface.scaffold)
+    if intent.get("idempotent_replay"):
+        # Repeated delivery: return the recorded outcome WITHOUT executing
+        # again (the ledger's whole point). The first execution's envelope
+        # is not re-derivable here, so the replay carries the intent and
+        # the flag — the caller knows the effect already happened.
+        return {"accepted": True, "executed": False,
+                "success": True, "idempotent_replay": True,
+                "op": intent.get("op"),
+                "command_type": command_type,
+                "construction_id": construction_id,
+                "correlation_id": intent.get("correlation_id", ""),
+                "command_id": intent.get("command_id", "")}
+    if not intent.get("accepted"):
+        return {"accepted": False, "executed": False, "success": True,
+                "reason": intent.get("reason"),
+                "command_type": command_type,
+                "construction_id": construction_id,
+                "correlation_id": intent.get("correlation_id", ""),
+                "command_id": intent.get("command_id", ""),
+                "idempotent_replay": intent.get("idempotent_replay", False)}
+
+    op = intent["op"]
+    args = intent["args"]
+    ops = {
+        "resume": lambda: iface.resume(args["contract_id"]),
+        "transition": lambda: iface.transition(
+            args["contract_id"], args["phase"],
+            evidence_ref=f"command:{intent['command_id']}"
+            + (f" approved_by:{args['approved_by']}" if args.get("approved_by") else "")),
+        "amend_scope": lambda: iface.amend_scope(
+            args["contract_id"], args["add_paths"],
+            reason=args.get("reason", "substrate context"),
+            authority=args.get("authority", "substrate")),
+    }
+    fn = ops.get(op)
+    if fn is None:
+        return {"accepted": True, "executed": False, "success": False,
+                "error": f"no executor for op {op!r}",
+                "command_type": command_type,
+                "construction_id": construction_id}
+    env = fn()
+    return {"accepted": True, "executed": True,
+            "success": bool(env.get("success")),
+            "envelope": env,
+            "command_type": command_type,
+            "construction_id": construction_id,
+            "correlation_id": intent.get("correlation_id", ""),
+            "command_id": intent.get("command_id", ""),
+            "idempotent_replay": False}
@@ -36,6 +36,7 @@ from typing import Any

 import json
 import sys
+import uuid
 from pathlib import Path

 from rune import structural_store as ss
@@ -88,6 +89,39 @@ def _envelope(
    return env


+def _to_stage_check(spec: dict):
+    """One guard spec → a stage_guards.StageCheck (T5.15 wire)."""
+    from rune import stage_guards as sg
+
+    return sg.StageCheck(
+        name=str(spec.get("name", "guard")),
+        kind=str(spec.get("kind", "assert")),
+        argv=spec.get("argv"),
+        output_pattern=str(spec.get("output_pattern", "")),
+        path=str(spec.get("path", "")),
+        check=spec.get("check"),
+        description=str(spec.get("description", "")),
+    )
+
+
+def _guard_specs_from_obligations(c: dict) -> list[dict]:
+    """The contract's guard declarations: verification_obligations rows
+    shaped `guard:<json-spec>` (the decision path records them; anything
+    else stays a free-text obligation for the projection). JSONDecodeError
+    → skipped with no invented facts."""
+    specs: list[dict] = []
+    for item in (c.get("identities") or {}).get("verification_obligations", []) or []:
+        if not isinstance(item, str) or not item.startswith("guard:"):
+            continue
+        try:
+            spec = json.loads(item[6:])
+            if isinstance(spec, dict):
+                specs.append(spec)
+        except json.JSONDecodeError:
+            continue
+    return specs
+
+
 class ConstructionInterface:
    """The builder-facing construction lifecycle for ONE project target.

@@ -291,6 +325,13 @@ class ConstructionInterface:
                    "transition": {
                        "phase": {"type": "str", "required": True,
                                  "allowed": list(sc.PHASES)},
+                        "evidence_ref": {"type": "str", "required": False},
+                        "guard_checks": {
+                            "type": "list[dict] | None", "required": False,
+                            "notes": ("stage-guard specs run on the "
+                                      "completion edges (verifying/closed); "
+                                      "a failure BLOCKS with "
+                                      "VERIFICATION_FAILED — T5.15 wire")},
                    },
                    "close": {
                        "observed_evidence": {
@@ -381,7 +422,8 @@ class ConstructionInterface:
            errors.append({"error": Errors.INSUFFICIENT_EVIDENCE, "detail": trace["rationale"][:200]})
        contract = None
        if generate and trace["resolution"] not in ("INSUFFICIENT_EVIDENCE",):
-            cid = contract_id or f"BUILDER-{ss.now()}-{self.builder}"
+            cid = (contract_id
+                   or f"BUILDER-{ss.now()}-{uuid.uuid4().hex[:6]}-{self.builder}")
            cd.generate_contract(
                self.scaffold,
                doc,
@@ -607,8 +649,57 @@ class ConstructionInterface:
            warnings=warnings,
        )

-    def transition(self, contract_id: str, phase: str, *, evidence_ref: str = "") -> dict:
+    def transition(self, contract_id: str, phase: str, *, evidence_ref: str = "",
+                   guard_checks: list[dict] | None = None) -> dict:
+        """Move a contract phase. Completion-wave slice (T5.15, the
+        stage_guards wire): entering `verifying` and `closed` runs the
+        §15F/§15G guards — readiness/exit StageChecks and test-truth
+        verification — when the contract declares them
+        (identities.verification_obligations carries `guard:<spec>` rows,
+        or guard_checks passes inline specs).
+
+        guard_checks spec: {"name", "kind": "command"|"path_exists"|"assert",
+        "argv"/"path"/"output_pattern", "check": callable (in-process only),
+        "description"}. A guard failure is a VERIFICATION_FAILED envelope:
+        the phase does NOT move.
+
+        The legality map stays the scaffold's authority; guards ADD evidence
+        obligations on the completion-facing edges (building→verifying,
+        verifying→closed) — a tiny bounded change can pass guards=[] freely.
+        """
        doc = self.scaffold.load()
+        c = self.scaffold.contract(doc, contract_id)
+        if c is None:
+            return _envelope(
+                "transition",
+                success=False,
+                ids={"contract_id": contract_id},
+                errors=[{"error": Errors.UNKNOWN_CONTRACT, "detail": f"no contract {contract_id}"}],
+            )
+        if phase in ("verifying", "closed"):
+            from rune import stage_guards as sg
+
+            specs = guard_checks
+            if specs is None:
+                specs = _guard_specs_from_obligations(c)
+            if specs:
+                stage = sg.StageContract(
+                    stage_id=f"{contract_id}:{phase}",
+                    contract_id=contract_id,
+                    ready_when=[_to_stage_check(s) for s in specs],
+                    done_when=[_to_stage_check(s) for s in specs],
+                )
+                canary = sg.readiness_canary(stage)
+                if canary["verdict"] == "BLOCKED":
+                    return _envelope(
+                        "transition",
+                        success=False,
+                        ids={"contract_id": contract_id},
+                        state={"phase": c.get("phase"), "guard": canary},
+                        errors=[{"error": Errors.VERIFICATION_FAILED,
+                                 "detail": "stage guards blocked the transition",
+                                 "failures": canary["failures"]}],
+                    )
        try:
            self.scaffold.transition_contract(
                doc, contract_id, phase, evidence_ref=evidence_ref or f"builder:{self.builder}"
@@ -679,21 +770,87 @@ class ConstructionInterface:
        material_closure = self._auto_material_change(
            scope_files or [],
            note=f"contract {contract_id} closed (material change)")
+        self._record_impact_delta(contract_id, scope_files or [])
+        continuation = self._programme_continuation(contract_id)
+        state = {
+            "phase": "closed",
+            "material_change_closure": material_closure,
+            "contract_quality": {
+                k: q[k]
+                for k in ("correctly_predicted", "expected_new_observed", "genuinely_unexpected_new", "scope_check")
+                if k in q
+            },
+        }
+        if continuation is not None:
+            state["programme_continuation"] = continuation
        return _envelope(
            "close",
            success=True,
            ids={"contract_id": contract_id},
-            state={
-                "phase": "closed",
-                "material_change_closure": material_closure,
-                "contract_quality": {
-                    k: q[k]
-                    for k in ("correctly_predicted", "expected_new_observed", "genuinely_unexpected_new", "scope_check")
-                    if k in q
-                },
-            },
+            state=state,
        )

+    def _record_impact_delta(self, contract_id: str, scope_files: list) -> None:
+        """T5.15 slice 4 (the iteration wire): a material closure records a
+        predicted-vs-observed impact delta (rule 19H — deltas compound into
+        better future prediction). Governance-only closes (no scope files)
+        record nothing. Failure degrades: the ledger never fails a close."""
+        if not scope_files:
+            return
+        try:
+            from rune import iteration as it
+
+            it.record_impact_delta(
+                Path(self.scaffold.current).parent,
+                {
+                    "contract_id": contract_id,
+                    "scope_files": sorted(scope_files),
+                    "predicted_frontier": "from-contract-scope",
+                    "recorded_at": ss.now(),
+                },
+            )
+        except Exception as exc:  # noqa: BLE001 — delta failure never fails a closure
+            self._record_session(
+                self.scaffold.load(), "impact_delta_degraded",
+                f"{contract_id}: {str(exc)[:120]}")
+
+    def _programme_continuation(self, contract_id: str) -> dict | None:
+        """T5.15 slice 3 (the programme_continuation wire): after a legal
+        close, any persisted Programme stage bound to this contract is
+        marked COMPLETE and the continuation canary's verdict rides the
+        close envelope — 'task done' ≠ 'programme done', and the next
+        eligible stage is SURFACED, never guessed. No programme bound →
+        None (no continuation block in the envelope). Failures degrade:
+        a broken programme ledger never fails a real closure."""
+        try:
+            from pathlib import Path as _Path
+
+            from rune import programme_continuation as pc
+
+            base = _Path(self.scaffold.current).parent / pc.DEFAULT_PROGRAMME_AREA
+            if not base.is_dir():
+                return None
+            for prog_file in sorted(base.glob("*.json")):
+                prog = pc.Programme.load(base, prog_file.stem)
+                if prog is None:
+                    continue
+                stage = next((s for s in prog.stages
+                              if s.stage_id == contract_id
+                              and s.state != pc.COMPLETE), None)
+                if stage is None:
+                    continue
+                stage.state = pc.COMPLETE
+                stage.result["closed_at_contract"] = contract_id
+                prog.save(base)
+                return {
+                    "programme_id": prog.programme_id,
+                    "stage_marked_complete": contract_id,
+                    "canary": prog.continuation_canary(),
+                }
+        except Exception as exc:  # noqa: BLE001 — ledger failure never fails a closure
+            return {"error": f"programme continuation degraded: {exc}"[:200]}
+        return None
+
    def _auto_material_change(self, scope_files: list[str], *,
                              note: str) -> dict:
        """T21.4 §21.4H (rule 19Q.2): governed construction closes → the
@@ -1019,7 +1176,9 @@ def main(argv=None) -> int:
    if cmd == "verify":
        return _out(iface.verify(args[0], changed_files=(json.loads(args[1]) if len(args) > 1 else None)))
    if cmd == "transition":
-        return _out(iface.transition(args[0], args[1], evidence_ref=args[2] if len(args) > 2 else ""))
+        guard_checks = json.loads(args[3]) if len(args) > 3 else None
+        return _out(iface.transition(args[0], args[1], evidence_ref=args[2] if len(args) > 2 else "",
+                                     guard_checks=guard_checks))
    if cmd == "close":
        return _out(
            iface.close(
@@ -2525,6 +2525,53 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
 _operator = None


+def _forge_client_for_projection():
+    """The client seam for the forge projection (test-overridable)."""
+    try:
+        from .forge_client import create_forge_client
+    except ImportError:  # TUI top-level deploy (script mode)
+        from forge_client import create_forge_client
+    return create_forge_client()
+
+
+def _project_failure_to_forge(*, behaviour_id: str, observation: dict,
+                              repo: str) -> dict:
+    """T5.15 slice 5 (the forge projection wire): a recorded failure
+    projects to the human-visible forge (rule 19P — the projection is a
+    VIEW; canonical truth stays in the ledger). Uses the ALREADY-WIRED
+    forge client (create_forge_client → NullAdapter when unconfigured, so
+    the projection is dormant by construction). A thin shim adapts the
+    client's create_issue to the testing-mode projection's
+    create_or_patch_release contract. Forge failure NEVER breaks the
+    failure path — it degrades to a recorded marker."""
+    import logging
+
+    log = logging.getLogger(__name__)
+    try:
+        client = _forge_client_for_projection()
+
+        class _Shim:
+            """Adapts create_issue → the projection's release contract."""
+
+            @staticmethod
+            def create_or_patch_release(repo_: str, tag: str,
+                                        title: str, body: str) -> dict:
+                return client.create_issue(title, body)
+
+        try:
+            from .testing_mode import project_failure_to_forge
+        except ImportError:
+            from testing_mode import project_failure_to_forge
+
+        return project_failure_to_forge(
+            _Shim(), repo, behaviour_id=behaviour_id,
+            observation=observation)
+    except Exception as exc:  # noqa: BLE001 — projection failure never rewrites truth
+        log.debug("forge projection degraded: %s", exc)
+        return {"projected": False,
+                "error": str(exc)[:200]}
+
+
 def _wire_auto_filer() -> None:
    """Composition-root wiring for the failure auto-filer (#122).

@@ -8,7 +8,7 @@ mode system, subject header, shared memory and identity.

 This is not an interface. This is how I see, speak, and interact with the world.

-Version: v0.97.0
+Version: v0.99.5
 Features: 41 (24 original + 17 new native commands)

 Usage:
@@ -110,6 +110,56 @@ try:
 except ImportError:
    BACKEND_AVAILABLE = False

+
+def collect_post_response_signals(response: str, messages_count: int) -> list[str]:
+    """Gather the operational signals that used to each print a dim
+    system message after every task/build response (the dim-spam source,
+    T5.13): budget pressure, degraded claims, growth delta, the periodic
+    self-improvement cycle.
+
+    PURE and headless — no channel, no app. The caller renders ONE
+    collapsed line (or none when nothing fired). Queued-message execution
+    and on-complete cron firing stay in the worker: they move real
+    messages, not chatter, and keep their own echoes.
+
+    Returns short human-readable fragments; empty list = silence.
+    """
+    signals: list[str] = []
+    if not BACKEND_AVAILABLE:
+        return signals
+    try:
+        budget_data = provider_budget.get_status()
+        ratios = []
+        for provider in budget_data.get("providers", {}).values():
+            if provider.get("monthly_limit", 0) > 0:
+                ratios.append(provider.get("monthly_used", 0) / provider["monthly_limit"])
+            if provider.get("token_limit", 0) > 0:
+                ratios.append(provider.get("tokens_used", 0) / provider["token_limit"])
+        usage_percent = max(ratios, default=0) * 100
+        if usage_percent > 80:
+            signals.append(f"⚠ Budget: {usage_percent:.0f}% used")
+    except Exception:
+        pass
+    try:
+        claims_result = claim_audit.scan_text(response[:500])
+        if claims_result and claims_result.get("degraded"):
+            signals.append(f"⚠ Claims degraded: {len(claims_result['degraded'])}")
+    except Exception:
+        pass
+    try:
+        growth = growth_delta.check()
+        if growth and growth.get("delta"):
+            signals.append(f"📈 Growth: {growth['delta']}")
+    except Exception:
+        pass
+    try:
+        if messages_count % 10 == 0:
+            self_improve.run()
+            signals.append("🔄 Self-improvement cycle triggered")
+    except Exception:
+        pass
+    return signals
+
 from textual.app import App, ComposeResult
 from textual.containers import Horizontal, Vertical, VerticalScroll
 from textual.widgets import Button, Header, Input, ProgressBar, RichLog, Static
@@ -1533,18 +1583,21 @@ class RuneApp(App):
    def compose(self) -> ComposeResult:
        yield Header()
        with Horizontal(id="channels"):
-            yield ChannelView(
-                "task",
-                "[b]TASK[/b] — focused work",
-                classes="channel-view",
-                id="task-channel",
-            )
+            # T5.14 (Nick 2026-10-05): chat LEFT, task RIGHT — feels more
+            # natural (conversation first, focused work second). WorkPane
+            # stays third.
            yield ChannelView(
                "chat",
                "[b]CHAT[/b] — free conversation",
                classes="channel-view",
                id="chat-channel",
            )
+            yield ChannelView(
+                "task",
+                "[b]TASK[/b] — focused work",
+                classes="channel-view",
+                id="task-channel",
+            )
            # P1-07 (#30): third collapsible column — the operational index.
            yield WorkPane(id="work-pane")
        with Horizontal(id="footer-bar"):
@@ -1895,7 +1948,7 @@ class RuneApp(App):
        # Show version info
        self.chat_channel.add_message(
            "system",
-            f"[dim]Rune v0.97.0 | Textual TUI | {len(loaded)} native modules[/dim]",
+            f"[dim]Rune v0.99.5 | Textual TUI | {len(loaded)} native modules[/dim]",
        )

        # Auto-show night mode suggestion if evening/night hours
@@ -3737,69 +3790,13 @@ class RuneApp(App):
                        except Exception:
                            pass

-                        # Auto-warn budget when >80% used
-                        try:
-                            if BACKEND_AVAILABLE:
-                                budget_data = provider_budget.get_status()
-                                ratios = []
-                                for provider in budget_data.get("providers", {}).values():
-                                    if provider.get("monthly_limit", 0) > 0:
-                                        ratios.append(provider.get("monthly_used", 0) / provider["monthly_limit"])
-                                    if provider.get("token_limit", 0) > 0:
-                                        ratios.append(provider.get("tokens_used", 0) / provider["token_limit"])
-                                usage_percent = max(ratios, default=0) * 100
-                                if usage_percent > 80:
-                                    ch.add_message(
-                                        "system",
-                                        f"[dim]⚠ Budget: {usage_percent:.0f}% used[/dim]",
-                                    )
-                        except Exception:
-                            pass
-
-                        # Auto-check claims after response
-                        try:
-                            if BACKEND_AVAILABLE:
-                                claims_result = claim_audit.scan_text(response[:500])
-                                if claims_result and claims_result.get("degraded"):
-                                    ch.add_message(
-                                        "system",
-                                        f"[dim]⚠ Claims degraded: {len(claims_result['degraded'])}[/dim]",
-                                    )
-                        except Exception:
-                            pass
-
-                        # NOTE (Runtime Reachability audit 2026-09-10): the
-                        # old hook_engine.fire("post_message", ...) call here
-                        # was dead — hook_engine defines no fire(); the
-                        # AttributeError was silently swallowed, so "hooks"
-                        # never ran. Removed rather than left as a false-
-                        # liveness surface. Real hook dispatch is unwired
-                        # (microkernel.fire has zero callers) — see the
-                        # reachability audit, P3.
-
-                        # Auto-check growth delta
-                        try:
-                            if BACKEND_AVAILABLE:
-                                growth = growth_delta.check()
-                                if growth and growth.get("delta"):
-                                    ch.add_message(
-                                        "system",
-                                        f"[dim]📈 Growth: {growth['delta']}[/dim]",
-                                    )
-                        except Exception:
-                            pass
-
-                        # Auto-improve every 10 messages
-                        try:
-                            if self.status_bar.messages_count % 10 == 0:
-                                if BACKEND_AVAILABLE:
-                                    self_improve.run()
-                                    ch.add_message(
-                                        "system",
-                                        f"[dim]🔄 Self-improvement cycle triggered[/dim]",
-                                    )
-                        except Exception:
-                            pass
+                        # T5.13 dim-spam consolidation: previously five
+                        # separate dim system messages printed after every
+                        # response. ONE collapsed line now — empty means
+                        # silence, not a dash.
+                        signals = collect_post_response_signals(response[:500], self.status_bar.messages_count)
+                        if signals:
+                            ch.add_message("system", "[dim]" + " · ".join(signals) + "[/dim]")
                elif result.get("error") == "RESEARCH_REQUIRED":
                    # Nonterminal routing disposition, not failure (2026-09-10):
                    # the task hit the deep-research floor and was dispatched.
@@ -3980,13 +3977,13 @@ class RuneApp(App):
            pass

    def action_resize_left(self) -> None:
-        """Shrink task channel, grow chat channel."""
-        self.task_width = max(20, self.task_width - 5)
+        """Shrink the LEFT channel (chat, post-swap), grow the RIGHT (task)."""
+        self.task_width = min(80, self.task_width + 5)
        self._apply_resize()

    def action_resize_right(self) -> None:
-        """Grow task channel, shrink chat channel."""
-        self.task_width = min(80, self.task_width + 5)
+        """Grow the LEFT channel (chat), shrink the RIGHT (task)."""
+        self.task_width = max(20, self.task_width - 5)
        self._apply_resize()

    def _apply_resize(self):
@@ -86,6 +86,110 @@ PATTERN_MAX_PATTERNS = 200
 # ═══════════════════════════════════════════════════════════════════════════════


+def _keyword_stopwords() -> list:
+    """Gather seam (W3.4): the pattern-keyword stopword table as a sorted
+    list — data, not logic; both engines read this module-owned table."""
+    return sorted(_KEYWORD_STOPWORDS)
+
+
+_KEYWORD_STOPWORDS = frozenset({
+    "the",
+    "and",
+    "for",
+    "are",
+    "but",
+    "not",
+    "you",
+    "all",
+    "can",
+    "had",
+    "her",
+    "was",
+    "one",
+    "our",
+    "out",
+    "day",
+    "get",
+    "has",
+    "him",
+    "his",
+    "how",
+    "its",
+    "may",
+    "new",
+    "now",
+    "old",
+    "see",
+    "way",
+    "who",
+    "did",
+    "let",
+    "say",
+    "she",
+    "too",
+    "use",
+    "this",
+    "that",
+    "with",
+    "have",
+    "from",
+    "they",
+    "been",
+    "said",
+    "each",
+    "make",
+    "like",
+    "just",
+    "over",
+    "such",
+    "take",
+    "year",
+    "them",
+    "some",
+    "time",
+    "very",
+    "when",
+    "come",
+    "could",
+    "what",
+    "there",
+    "their",
+    "will",
+    "would",
+    "about",
+    "which",
+    "these",
+    "other",
+    "into",
+    "more",
+    "than",
+    "also",
+    "then",
+})
+
+
+def _core():
+    """The Rust cognition_core when available (strangler: Python fallback
+    load-bearing). RUNE_COGNITION_CORE=python pins the fallback."""
+    return None if cc_loader_pin() else _cc_probe()
+
+
+def _cc_probe():
+    from rune import cc_loader as _cl
+    try:
+        return _cl if _cl.available() else None
+    except Exception:  # noqa: BLE001, S110 — loader probe never crashes the fallback
+        return None
+
+
+def cc_loader_pin() -> bool:
+    """The env pin: RUNE_COGNITION_CORE=python → kernels OFF (shared probe).
+"""
+    import os
+
+    return os.environ.get("RUNE_COGNITION_CORE", "auto") == "python"
+
+
@dataclass
 class OverridePattern:
    """A learned pattern that requires override."""
@@ -105,7 +209,14 @@ class OverridePattern:
        return self.successes / total

    def matches(self, text: str) -> bool:
-        """Check if text matches this pattern's keywords."""
+        """Check if text matches this pattern's keywords. Deterministic
+        keyword-count check (kernel port NOT owed here — the W3.4 safety
+        kernels cover the params/rows surfaces; this tiny comparison stays
+        Python-side per the membrane law: gather/IO in Python, kernels for
+        deterministic row work already ported)."""
+        return self._py_matches(text)
+
+    def _py_matches(self, text: str) -> bool:
        text_lower = text.lower()
        matches = sum(1 for kw in self.keywords if kw.lower() in text_lower)
        return matches >= PATTERN_MIN_KEYWORDS
@@ -293,84 +404,12 @@ class SafetyOverride:
            self.stats.patterns_learned += 1

    def _extract_keywords(self, text: str) -> list[str]:
-        """Extract distinctive keywords from text for pattern matching."""
-        # Remove common stop words, keep words > 3 chars
-        stop_words = {
-            "the",
-            "and",
-            "for",
-            "are",
-            "but",
-            "not",
-            "you",
-            "all",
-            "can",
-            "had",
-            "her",
-            "was",
-            "one",
-            "our",
-            "out",
-            "day",
-            "get",
-            "has",
-            "him",
-            "his",
-            "how",
-            "its",
-            "may",
-            "new",
-            "now",
-            "old",
-            "see",
-            "way",
-            "who",
-            "did",
-            "let",
-            "say",
-            "she",
-            "too",
-            "use",
-            "this",
-            "that",
-            "with",
-            "have",
-            "from",
-            "they",
-            "been",
-            "said",
-            "each",
-            "make",
-            "like",
-            "just",
-            "over",
-            "such",
-            "take",
-            "year",
-            "them",
-            "some",
-            "time",
-            "very",
-            "when",
-            "come",
-            "could",
-            "what",
-            "there",
-            "their",
-            "will",
-            "would",
-            "about",
-            "which",
-            "these",
-            "other",
-            "into",
-            "more",
-            "than",
-            "also",
-            "then",
-        }
+        """Extract distinctive keywords from text for pattern matching.
+        Keyword extraction stays Python-side (W3.4 scope note): the safety
+        kernels cover the params/rows surfaces; this extractor is a gather
+        (IO-adjacent list build) per the membrane law."""
        words = re.findall(r"[a-z]{4,}", text.lower())
-        keywords = [w for w in words if w not in stop_words]
+        keywords = [w for w in words if w not in _KEYWORD_STOPWORDS]
        # Return top 5 most distinctive words
        return keywords[:5]

@@ -53,6 +53,12 @@ from rune.structural_store import (  # shared generic core (SCAFFOLD-TRANCHE-001
    reconcile_authorities, mark_conflict, resolve_status, authority_health,
    dedupe_observations as _ss_dedupe_observations,
 )
+from rune.spine_language import (  # noqa: F401 — re-exported (import-compat): the
+    # #2465 doc-level transforms live here now (giant-file carve); tests and
+    # the canary_spine_meta read them THROUGH spine, exactly as before.
+    _classification_sides, _place_languages, _normalize_discovery_ids,
+    _apply_contract_discoveries, _contract_discoveries,
+)

 SCHEMA_VERSION = 1

@@ -337,181 +343,6 @@ def _rust_core():
        return None


-def _classification_sides() -> dict:
-    """The measured dichotomy map (W0): module stem → RUST|GO|TUI-LAST|...
-    Loaded once per _place_languages call, best-effort — a missing or
-    unreadable classification.json degrades to {}/unassessed verdicts,
-    never to invented facts."""
-    try:
-        data = json.loads((ROOT / "store" / "rebase" / "classification.json").read_text())
-        return {name: rec.get("side", "") for name, rec in data.items() if isinstance(rec, dict)}
-    except (OSError, json.JSONDecodeError, AttributeError):
-        return {}
-
-
-def _place_languages(doc: dict) -> int:
-    """Stamp a language verdict per entity (no-op for entities that carry
-    one). Deterministic: substring classification over the entity id + the
-    kind; the exact same rule construction_decision.language_placement
-    implements (kept inline here to avoid a per-entity FFI crossing for
-    the spine's own 218 entities; the FFI path is used in contracts)."""
-    GO_HINTS = ("server", "transport", "dispatch", "entry", "bridge",
-                "hook_engine", "provider", "watch", "notif", "voice",
-                "gitea", "event_", "message_", "tool_exec", "work_manager")
-    RUST_HINTS = ("memory", "drift", "warm", "capab", "token", "spine",
-                  "scaffold", "construct", "structural", "curator",
-                  "consolidat", "elastic", "cognit", "abstract", "contract",
-                  "ledger", "audit", "claim", "evidence", "behavio",
-                  "verif", "pins", "witness", "anchor", "growth", "goal",
-                  "salience", "store", "tui_", "rune_tui")
-    placed = 0
-    CLASSIFICATION_SIDES = _classification_sides()
-    for e in doc.get("entities", []):
-        eid = e.get("id", "")
-        kind = e.get("kind", "")
-        if eid.startswith("contract:"):
-            lang = "n/a"
-        elif kind == "rust-crate":
-            lang = "rust"
-        elif kind == "go-surface":
-            lang = "go"
-        elif kind == "workflow" or kind == "config":
-            lang = "n/a"
-        elif kind == "deployment_surface":
-            lang = "go"
-        elif "tui" in eid:
-            lang = "python-tui (Wave 6)"
-        elif any(h in eid for h in GO_HINTS):
-            lang = "go"
-        elif any(h in eid for h in RUST_HINTS):
-            lang = "rust"
-        elif kind == "module":
-            lang = "rust"
-        elif kind == "subsystem":
-            # A subsystem is a CONTAINER — its language is its members'
-            # majority side per the measured dichotomy. The evidence map is
-            # classification.json (measured, 174 modules; honest note: newer
-            # files may be unclassified, so the count of classified members
-            # rides the verdict). Members arrive as belongs_to relationships
-            # whose from-ids carry module:<name> — map through classification.
-            members = [
-                r.get("from", "").split(":", 1)[1]
-                for r in doc.get("relationships", [])
-                if r.get("type") == "belongs_to" and r.get("to") == eid
-                and r.get("from", "").startswith("module:")
-            ]
-            sides = [
-                str(CLASSIFICATION_SIDES.get(m) or "").upper()
-                for m in members
-            ]
-            rust = sides.count("RUST")
-            go = sides.count("GO")
-            if rust and go:
-                lang = f"mixed (rust {rust} / go {go} members)"
-            elif rust:
-                lang = "rust"
-            elif go:
-                lang = "go"
-            else:
-                lang = "unassessed"
-        else:
-            lang = "unassessed"
-        e.setdefault("identities", {})["language"] = lang
-        if lang != "unassessed":
-            placed += 1
-    return placed
-
-
-def _normalize_discovery_ids(doc: dict) -> int:
-    """Migrate the old unqualified contract references, preserving evidence.
-
-    A file path (or 'suite') is not a module ID. Only migrate the malformed
-    module records produced by the earlier discovery projector; leave all
-    other open-world entity classes alone.
-    """
-    renamed = {}
-    observed = {
-        eid for _, entities in _contract_discoveries()
-        for eid, _, provenance in entities if provenance == "observed"
-    }
-    for e in doc.get("entities", []):
-        if (e.get("kind") == "module" and ":" not in e.get("id", "")
-                and any(ref.startswith("contract-discovery:")
-                        for ref in e.get("evidence_refs", []))):
-            old = e["id"]
-            e["id"] = f"construction-surface:{old}"
-            e["kind"] = "construction-surface"
-            e.setdefault("identities", {})["source_ref"] = old
-            renamed[old] = e["id"]
-        if e.get("kind") == "construction-surface" and e.get("id", "").startswith("construction-surface:"):
-            # Earlier code promoted every *candidate* to PRESENT. Undo only
-            # that unsupported claim; an actually observed surface retains it.
-            if e["id"].split(":", 1)[1] not in observed:
-                e["maturity"] = "DECLARED"
-            e["tags"] = [tag for tag in e.get("tags", []) if tag != "code"]
-    for rel in doc.get("relationships", []):
-        for endpoint in ("from", "to"):
-            if rel.get(endpoint) in renamed:
-                rel[endpoint] = renamed[rel[endpoint]]
-    return len(renamed)
-
-
-def _apply_contract_discoveries(doc: dict) -> int:
-    """Contract-discovery write-back (issue #2465 addition 2): entities
-    DISCOVERED by closed contracts become modeled spine entities — the
-    spine learns what construction actually built. Doc-level transform,
-    applied to BOTH cores' output. The CANDIDATE law holds: only CLOSED
-    contracts write back."""
-    placed = 0
-    for cid, entities in _contract_discoveries():
-        for eid, kind, provenance in entities:
-            before = len(doc.get("entities", []))
-            if kind == "construction-surface":
-                eid = f"construction-surface:{eid}"
-            add_entity(doc, id=eid, kind=kind,
-                       tags=["contract-discovery"],
-                       identities={"source_ref": eid.split(":", 1)[1]} if kind == "construction-surface" else None,
-                       evidence_refs=[f"contract-discovery:{cid}"],
-                       maturity="PRESENT" if provenance == "observed" else "DECLARED")
-            if len(doc.get("entities", [])) > before:
-                placed += 1
-            else:
-                placed += 0  # high-water merge — no new entity, still evidence
-    return placed
-
-
-def _contract_discoveries():
-    """Closed contracts' DISCOVERED entities (issue #2465 addition 2):
-    expected_new_observed + candidate corrections, gathered from the
-    scaffold store. Returns [(contract_id, [(entity_id, kind), ...])]."""
-    try:
-        doc = json.loads((ROOT / "store" / "scaffold" / "current.json").read_text())
-    except (OSError, json.JSONDecodeError):
-        return []
-    out = []
-    for e in doc.get("entities", []):
-        cid = e.get("id", "")
-        if not cid.startswith("contract:") or e.get("phase") != "closed":
-            continue
-        q = e.get("ext", {}).get("contract_quality", {})
-        found = []
-        # expected-new surfaces observed at close (REAL delivered entities)
-        for eid in q.get("expected_new_observed", []):
-            kind = eid.split(":", 1)[0] if ":" in eid else "construction-surface"
-            if kind in ("rust-crate", "go-surface", "config", "workflow", "module"):
-                found.append((eid, kind, "observed"))
-            elif kind == "construction-surface":
-                found.append((eid, kind, "observed"))
-        # candidate corrections: the closure flagged them for modeling
-        for cand in q.get("candidate_corrections", []):
-            eid = cand.get("entity", "")
-            kind = eid.split(":", 1)[0] if ":" in eid else "construction-surface"
-            if eid and kind in ("rust-crate", "go-surface", "config", "workflow", "module", "construction-surface"):
-                found.append((eid, kind, "candidate"))
-        if found:
-            out.append((cid[9:], found))
-    return out
-

 def _gather_build_inputs(manifest_path: Path | None = None) -> dict:
    """Every I/O read build() needs, in one place (zero-dep split law:
@@ -0,0 +1,218 @@
+"""spine_language.py — the #2465 doc-level transforms (carved from spine.py
+to honor the giant-file henxel: keep it under 1200 lines).
+
+Three transforms, applied to BOTH cores' output at build time:
+  1. language placement — every entity gets a language verdict (kind
+     defaults; subsystems derive from member-majority over the measured
+     classification map);
+  2. discovery-ID normalization — migrate legacy unqualified contract
+     discoveries to construction-surface kind without losing evidence;
+  3. contract-discovery write-back — closed contracts teach the spine
+     what construction actually built (observed → PRESENT, candidate →
+     DECLARED; the CANDIDATE law holds).
+
+Same dependency laws as spine.py: structural_store primitives only, no
+I/O beyond the classification/scaffold store reads, deterministic.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from rune.structural_store import (
+    add_entity, add_relationship, entity, new_doc, relationships, _sorted, now,
+)
+
+ROOT = Path(__file__).resolve().parents[2]
+SRC_DIR = ROOT / "src" / "rune"
+
+def _classification_sides() -> dict:
+    """The measured dichotomy map (W0): module stem → RUST|GO|TUI-LAST|...
+    Loaded once per _place_languages call, best-effort — a missing or
+    unreadable classification.json degrades to {}/unassessed verdicts,
+    never to invented facts."""
+    try:
+        data = json.loads((ROOT / "store" / "rebase" / "classification.json").read_text())
+        return {name: rec.get("side", "") for name, rec in data.items() if isinstance(rec, dict)}
+    except (OSError, json.JSONDecodeError, AttributeError):
+        return {}
+
+
+def _place_languages(doc: dict) -> int:
+    """Stamp a language verdict per entity (no-op for entities that carry
+    one). Deterministic: substring classification over the entity id + the
+    kind; the exact same rule construction_decision.language_placement
+    implements (kept inline here to avoid a per-entity FFI crossing for
+    the spine's own 218 entities; the FFI path is used in contracts)."""
+    GO_HINTS = ("server", "transport", "dispatch", "entry", "bridge",
+                "hook_engine", "provider", "watch", "notif", "voice",
+                "gitea", "event_", "message_", "tool_exec", "work_manager")
+    RUST_HINTS = ("memory", "drift", "warm", "capab", "token", "spine",
+                  "scaffold", "construct", "structural", "curator",
+                  "consolidat", "elastic", "cognit", "abstract", "contract",
+                  "ledger", "audit", "claim", "evidence", "behavio",
+                  "verif", "pins", "witness", "anchor", "growth", "goal",
+                  "salience", "store", "tui_", "rune_tui")
+    placed = 0
+    CLASSIFICATION_SIDES = _classification_sides()
+    for e in doc.get("entities", []):
+        eid = e.get("id", "")
+        kind = e.get("kind", "")
+        # AUTHORITY ORDER (Nick's dichotomy, 2026-10-06 correction): the
+        # MEASURED classification map (import-evidence per module) decides
+        # FIRST. The name-hint lists are only a fallback for entities with
+        # no measurement — never allowed to override the evidence. A module
+        # with neither measurement nor hint is "unclassified", not a silent
+        # rust default.
+        measured = str(CLASSIFICATION_SIDES.get(
+            eid.split(":", 1)[1] if ":" in eid else eid, "") or "").upper()
+        if eid.startswith("contract:"):
+            lang = "n/a"
+        elif kind == "rust-crate":
+            lang = "rust"
+        elif kind == "go-surface":
+            lang = "go"
+        elif kind == "workflow" or kind == "config":
+            lang = "n/a"
+        elif kind == "deployment_surface":
+            lang = "go"
+        elif measured == "RUST":
+            lang = "rust"
+        elif measured == "GO":
+            lang = "go"
+        elif measured == "TUI-LAST":
+            lang = "python-tui (Wave 6)"
+        elif measured == "TEST-INFRA" or measured == "RETIRES":
+            lang = "test-infra" if measured == "TEST-INFRA" else "retires"
+        elif "tui" in eid:
+            lang = "python-tui (Wave 6)"
+        elif any(h in eid for h in GO_HINTS):
+            lang = "go (name hint)"
+        elif any(h in eid for h in RUST_HINTS):
+            lang = "rust (name hint)"
+        elif kind == "module":
+            lang = "unclassified"
+        elif kind == "subsystem":
+            # A subsystem is a CONTAINER — its language is its members'
+            # majority side per the measured dichotomy. The evidence map is
+            # classification.json (measured, 174 modules; honest note: newer
+            # files may be unclassified, so the count of classified members
+            # rides the verdict). Members arrive as belongs_to relationships
+            # whose from-ids carry module:<name> — map through classification.
+            members = [
+                r.get("from", "").split(":", 1)[1]
+                for r in doc.get("relationships", [])
+                if r.get("type") == "belongs_to" and r.get("to") == eid
+                and r.get("from", "").startswith("module:")
+            ]
+            sides = [
+                str(CLASSIFICATION_SIDES.get(m) or "").upper()
+                for m in members
+            ]
+            rust = sides.count("RUST")
+            go = sides.count("GO")
+            if rust and go:
+                lang = f"mixed (rust {rust} / go {go} members)"
+            elif rust:
+                lang = "rust"
+            elif go:
+                lang = "go"
+            else:
+                lang = "unassessed"
+        else:
+            lang = "unassessed"
+        e.setdefault("identities", {})["language"] = lang
+        if lang != "unassessed":
+            placed += 1
+    return placed
+
+
+def _normalize_discovery_ids(doc: dict) -> int:
+    """Migrate the old unqualified contract references, preserving evidence.
+
+    A file path (or 'suite') is not a module ID. Only migrate the malformed
+    module records produced by the earlier discovery projector; leave all
+    other open-world entity classes alone.
+    """
+    renamed = {}
+    observed = {
+        eid for _, entities in _contract_discoveries()
+        for eid, _, provenance in entities if provenance == "observed"
+    }
+    for e in doc.get("entities", []):
+        if (e.get("kind") == "module" and ":" not in e.get("id", "")
+                and any(ref.startswith("contract-discovery:")
+                        for ref in e.get("evidence_refs", []))):
+            old = e["id"]
+            e["id"] = f"construction-surface:{old}"
+            e["kind"] = "construction-surface"
+            e.setdefault("identities", {})["source_ref"] = old
+            renamed[old] = e["id"]
+        if e.get("kind") == "construction-surface" and e.get("id", "").startswith("construction-surface:"):
+            # Earlier code promoted every *candidate* to PRESENT. Undo only
+            # that unsupported claim; an actually observed surface retains it.
+            if e["id"].split(":", 1)[1] not in observed:
+                e["maturity"] = "DECLARED"
+            e["tags"] = [tag for tag in e.get("tags", []) if tag != "code"]
+    for rel in doc.get("relationships", []):
+        for endpoint in ("from", "to"):
+            if rel.get(endpoint) in renamed:
+                rel[endpoint] = renamed[rel[endpoint]]
+    return len(renamed)
+
+
+def _apply_contract_discoveries(doc: dict) -> int:
+    """Contract-discovery write-back (issue #2465 addition 2): entities
+    DISCOVERED by closed contracts become modeled spine entities — the
+    spine learns what construction actually built. Doc-level transform,
+    applied to BOTH cores' output. The CANDIDATE law holds: only CLOSED
+    contracts write back."""
+    placed = 0
+    for cid, entities in _contract_discoveries():
+        for eid, kind, provenance in entities:
+            before = len(doc.get("entities", []))
+            if kind == "construction-surface":
+                eid = f"construction-surface:{eid}"
+            add_entity(doc, id=eid, kind=kind,
+                       tags=["contract-discovery"],
+                       identities={"source_ref": eid.split(":", 1)[1]} if kind == "construction-surface" else None,
+                       evidence_refs=[f"contract-discovery:{cid}"],
+                       maturity="PRESENT" if provenance == "observed" else "DECLARED")
+            if len(doc.get("entities", [])) > before:
+                placed += 1
+            else:
+                placed += 0  # high-water merge — no new entity, still evidence
+    return placed
+
+
+def _contract_discoveries():
+    """Closed contracts' DISCOVERED entities (issue #2465 addition 2):
+    expected_new_observed + candidate corrections, gathered from the
+    scaffold store. Returns [(contract_id, [(entity_id, kind), ...])]."""
+    try:
+        doc = json.loads((ROOT / "store" / "scaffold" / "current.json").read_text())
+    except (OSError, json.JSONDecodeError):
+        return []
+    out = []
+    for e in doc.get("entities", []):
+        cid = e.get("id", "")
+        if not cid.startswith("contract:") or e.get("phase") != "closed":
+            continue
+        q = e.get("ext", {}).get("contract_quality", {})
+        found = []
+        # expected-new surfaces observed at close (REAL delivered entities)
+        for eid in q.get("expected_new_observed", []):
+            kind = eid.split(":", 1)[0] if ":" in eid else "construction-surface"
+            if kind in ("rust-crate", "go-surface", "config", "workflow", "module"):
+                found.append((eid, kind, "observed"))
+            elif kind == "construction-surface":
+                found.append((eid, kind, "observed"))
+        # candidate corrections: the closure flagged them for modeling
+        for cand in q.get("candidate_corrections", []):
+            eid = cand.get("entity", "")
+            kind = eid.split(":", 1)[0] if ":" in eid else "construction-surface"
+            if eid and kind in ("rust-crate", "go-surface", "config", "workflow", "module", "construction-surface"):
+                found.append((eid, kind, "candidate"))
+        if found:
+            out.append((cid[9:], found))
+    return out
@@ -20,6 +20,33 @@ HOT_THRESHOLD = 7  # accessed within 7 days = hot
 WARM_THRESHOLD = 30  # accessed within 30 days = warm
 # > 30 days = cold

+
+def _core():
+    """The Rust cognition_core when available (strangler: Python fallback is
+    load-bearing). RUNE_COGNITION_CORE=python pins the fallback."""
+    return None if cc_loader_pin() else _cc_probe()
+
+
+def _cc_probe():
+    from rune import cc_loader as _cl
+    try:
+        return _cl if _cl.available() else None
+    except Exception:  # noqa: BLE001, S110 — loader probe never crashes the fallback
+        return None
+
+
+def cc_loader_pin() -> bool:
+    """The env pin: RUNE_COGNITION_CORE=python → kernels OFF (shared probe).
+"""
+    import os
+
+    return os.environ.get("RUNE_COGNITION_CORE", "auto") == "python"
+
+
+def _py_core_now() -> str:
+    """The injected clock (iso Z) — module-level for the kernel boundary."""
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
 # Tier storage characteristics
 TIER_CONFIG = {
    "hot": {
@@ -56,7 +83,24 @@ class TierManager:
        """Determine the tier for an entry based on its access patterns.

        Returns 'hot', 'warm', or 'cold'.
+        Delegates to the Rust kernel when built; the Python body below is
+        the parity oracle (W3.4: identical thresholds injected, naive
+        timestamps refuse per the PARITY TRAP law).
        """
+        core = _core()
+        if core is not None:
+            try:
+                from rune import cc_loader as _cl
+
+                raw = _cl.tier_get(entry, _py_core_now(),
+                                   float(HOT_THRESHOLD), float(WARM_THRESHOLD))
+                return str(raw)
+            except Exception:  # noqa: BLE001, S110 — kernel failure → oracle, never a broken call
+                pass
+        return self._py_get_tier(entry)
+
+    def _py_get_tier(self, entry: dict) -> str:
+        """Python parity oracle (W3.4)."""
        wash_count = entry.get("wash_count", 0) or 0
        last_accessed = entry.get("last_accessed")
        rock_status = entry.get("rock_status", "washable")
@@ -88,7 +132,26 @@ class TierManager:
        return "warm"

    def promote(self, entry: dict) -> str:
-        """Promote an entry to a higher tier. Returns new tier."""
+        """Promote an entry to a higher tier. Returns new tier.
+        Delegates to the Rust kernel when built (Python oracle below)."""
+        core = _core()
+        if core is not None:
+            try:
+                from rune import cc_loader as _cl
+
+                # The kernel's promote/demote take the FULL ENTRY (plus
+                # thresholds + clock) — not a tier string. They recompute
+                # the current tier internally, so a tier-string call was
+                # silently landing on defaults (the live probe caught it).
+                raw = _cl.tier_promote(entry, _py_core_now(),
+                                       float(HOT_THRESHOLD), float(WARM_THRESHOLD))
+                return str(raw)
+            except Exception:  # noqa: BLE001, S110 — kernel failure → oracle fallback, intentional silence
+                pass
+        return self._py_promote(entry)
+
+    def _py_promote(self, entry: dict) -> str:
+        """Python parity oracle."""
        current = self.get_tier(entry)
        if current == "cold":
            return "warm"
@@ -97,7 +160,24 @@ class TierManager:
        return "hot"  # already hot

    def demote(self, entry: dict) -> str:
-        """Demote an entry to a lower tier. Returns new tier."""
+        """Demote an entry to a lower tier. Returns new tier.
+        Delegates to the Rust kernel when built (Python oracle below)."""
+        core = _core()
+        if core is not None:
+            try:
+                from rune import cc_loader as _cl
+
+                # Same full-entry contract as promote; the kernel handles
+                # rock_status internally (rock never demotes).
+                raw = _cl.tier_demote(entry, _py_core_now(),
+                                      float(HOT_THRESHOLD), float(WARM_THRESHOLD))
+                return str(raw)
+            except Exception:  # noqa: BLE001, S110 — kernel failure → oracle fallback, intentional silence
+                pass
+        return self._py_demote(entry)
+
+    def _py_demote(self, entry: dict) -> str:
+        """Python parity oracle."""
        current = self.get_tier(entry)
        rock_status = entry.get("rock_status", "washable")

@@ -84,76 +84,129 @@ def is_correction(text):
    return False


-def extract_topic(text):
-    """Extract topic keywords from text. Deterministic."""
-    stopwords = {
-        "the",
-        "a",
-        "an",
-        "is",
-        "are",
-        "was",
-        "were",
-        "be",
-        "been",
-        "being",
-        "have",
-        "has",
-        "had",
-        "do",
-        "does",
-        "did",
-        "will",
-        "would",
-        "could",
-        "should",
-        "may",
-        "might",
-        "shall",
-        "can",
-        "to",
-        "of",
-        "in",
-        "for",
-        "on",
-        "with",
-        "at",
-        "by",
-        "from",
-        "as",
-        "into",
-        "through",
-        "during",
-        "before",
-        "after",
-        "this",
-        "that",
-        "these",
-        "those",
-        "it",
-        "its",
-        "what",
-        "which",
-        "who",
-        "how",
-        "when",
-        "where",
-        "why",
-        "and",
-        "or",
-        "but",
-        "not",
-        "no",
+def _stopwords() -> list:
+    """Gather seam (W3.4 delegation): the topic-extraction stopword table
+    as a SORTED list — sorted so JSON travel and kernel-side matching see
+    a deterministic order regardless of the set's iteration order."""
+    return sorted(_STOPWORDS)
+
+
+_STOPWORDS = {
+    "the", "a", "an", "is", "are", "was", "were", "be", "been", "being",
+    "have", "has", "had", "do", "does", "did", "will", "would", "could",
+    "should", "may", "might", "shall", "can", "to", "of", "in", "for",
+    "on", "with", "at", "by", "from", "as", "into", "through", "during",
+    "before", "after", "this", "that", "these", "those", "it", "its",
+    "what", "which", "who", "how", "when", "where", "why", "and", "or",
+    "but", "not", "no",
+}
+
+
+def _tables() -> dict:
+    """Gather seam: ALL regex pattern tables as data in one dict — the
+    exact contract the cc_witness_signals kernel consumes (parity-proven
+    per-table-row; regex_lite_search is the faithful subset for THESE
+    shapes). Sorted lists keep travel deterministic."""
+    return {
+        "stopwords": _stopwords(),
+        "correction": sorted(CORRECTION_PATTERNS),
+        "commitment": sorted([
+            r"\bI'?ll\s+(send|do|fix|build|check|update|create|write)\b",
+            r"\bI will\s+(send|do|fix|build|check|update|create|write)\b",
+            r"\blet me\s+(send|do|fix|build|check|update|create|write)\b",
+            r"\bremind me\b",
+            r"\bI promise\b",
+            r"\bI need to\b",
+        ]),
+        "contradiction": sorted([r"\b(but|however|actually|wait|that contradicts)\b"]),
+        "preference": sorted([
+            r"\bI (prefer|like|love|hate|dislike)\b",
+            r"\bI'?m (not )?(a fan of|into|interested in)\b",
+            r"\b(don'?t|do not) (like|want|need)\b",
+            r"\b(always|never) (use|do|say)\b",
+        ]),
+        "urgency": sorted([
+            r"\b(urgent|asap|immediately|now|quick|hurry)\b",
+            r"\b(deadline|due|overdue|late)\b",
+            r"\b(important|critical|crucial|vital)\b",
+        ]),
+        "learning": sorted([
+            r"\b(learn|understand|explain|teach|show me)\b",
+            r"\b(how (does|do|did)|why (does|do|did)|what (is|are|was|were))\b",
+            r"\b(I don'?t (understand|know|get))\b",
+        ]),
    }
+
+
+def _core():
+    """The Rust cognition_core when available (strangler: Python fallback
+    load-bearing). RUNE_COGNITION_CORE=python pins the fallback."""
+    return None if cc_loader_pin() else _cc_probe()
+
+
+def _cc_probe():
+    from rune import cc_loader as _cl
+    try:
+        return _cl if _cl.available() else None
+    except Exception:  # noqa: BLE001, S110 — loader probe never crashes the fallback
+        return None
+
+
+def cc_loader_pin() -> bool:
+    """The env pin: RUNE_COGNITION_CORE=python → kernels OFF (shared probe).
+"""
+    import os
+
+    return os.environ.get("RUNE_COGNITION_CORE", "auto") == "python"
+
+
+def extract_topic(text):
+    """Extract topic keywords from text. Deterministic.
+    Delegates to the kernel when built; Python oracle below."""
+    core = _core()
+    if core is not None:
+        try:
+            from rune import cc_loader as _cl
+
+            raw = _cl.witness_extract_topic(text, _stopwords())
+            return list(raw) if isinstance(raw, list) else list(raw or [])
+        except Exception:  # noqa: BLE001, S110 — kernel failure → oracle, intentional silence
+            pass
    words = re.findall(r"\w+", text.lower())
-    return [w for w in words if len(w) > 3 and w not in stopwords][:5]
+    return [w for w in words if len(w) > 3 and w not in _STOPWORDS][:5]
+
+
+def _py_extract_topic(text):
+    """Python parity oracle (W3.4)."""
+    words = re.findall(r"\w+", text.lower())
+    return [w for w in words if len(w) > 3 and w not in _STOPWORDS][:5]


 def detect_signals(message, tool_error=None, task_status=None, recent_topics=None):
    """Tier 1: Reflex — extract signals from the turn. Pure Python, <1ms.

    Returns dict of signals. Empty dict = nothing meaningful happened.
+    Delegates to the kernel when built (tables travel as data per the
+    REGEX-TABLE PORT LAW); Python oracle below is the parity authority.
    """
+    core = _core()
+    if core is not None:
+        try:
+            from rune import cc_loader as _cl
+
+            raw = _cl.witness_signals(message, _tables(), tool_error or "",
+                                      task_status or "", recent_topics or [])
+            if isinstance(raw, dict):
+                return {k: v for k, v in raw.items() if v not in (False, "")}
+            # non-dict → fall through to the oracle
+        except Exception:  # noqa: BLE001, S110 — kernel failure → oracle, intentional silence
+            pass
+    return _py_detect_signals(message, tool_error, task_status, recent_topics)
+
+
+def _py_detect_signals(message, tool_error=None, task_status=None,
+                       recent_topics=None):
+    """Python parity oracle (W3.4). Local tables via the gather seam."""
    signals = {}

    # Error signal
@@ -179,62 +232,24 @@ def detect_signals(message, tool_error=None, task_status=None, recent_topics=Non
            signals["topic_drift"] = True
            signals["new_topics"] = list(current_topics - recent_set)[:3]

-    # Commitment detection
-    commitment_patterns = [
-        r"\bI'?ll\s+(send|do|fix|build|check|update|create|write)\b",
-        r"\bI will\s+(send|do|fix|build|check|update|create|write)\b",
-        r"\blet me\s+(send|do|fix|build|check|update|create|write)\b",
-        r"\bremind me\b",
-        r"\bI promise\b",
-        r"\bI need to\b",
-    ]
-    if message:
-        for pattern in commitment_patterns:
-            if re.search(pattern, message, re.IGNORECASE):
-                signals["commitment_detected"] = True
-                break
+    # Commitment detection (tables live in _tables() — the gather seam;
+    # the oracle reads the SAME table data the kernel receives)
+    for signal_name, table_name in (
+        ("commitment_detected", "commitment"),
+        ("preference_detected", "preference"),
+        ("urgency_detected", "urgency"),
+        ("learning_opportunity", "learning"),
+    ):
+        if message:
+            for pattern in _tables().get(table_name, []):
+                if re.search(pattern, message, re.IGNORECASE):
+                    signals[signal_name] = True
+                    break

    # Contradiction signal (simple: message contains "but" or "however" after a statement)
    if message and re.search(r"\b(but|however|actually|wait|that contradicts)\b", message.lower()):
        signals["possible_contradiction"] = True

-    # Preference detection
-    preference_patterns = [
-        r"\bI (prefer|like|love|hate|dislike)\b",
-        r"\bI'?m (not )?(a fan of|into|interested in)\b",
-        r"\b(don'?t|do not) (like|want|need)\b",
-        r"\b(always|never) (use|do|say)\b",
-    ]
-    if message:
-        for pattern in preference_patterns:
-            if re.search(pattern, message, re.IGNORECASE):
-                signals["preference_detected"] = True
-                break
-
-    # Urgency detection
-    urgency_patterns = [
-        r"\b(urgent|asap|immediately|now|quick|hurry)\b",
-        r"\b(deadline|due|overdue|late)\b",
-        r"\b(important|critical|crucial|vital)\b",
-    ]
-    if message:
-        for pattern in urgency_patterns:
-            if re.search(pattern, message, re.IGNORECASE):
-                signals["urgency_detected"] = True
-                break
-
-    # Learning opportunity detection
-    learning_patterns = [
-        r"\b(learn|understand|explain|teach|show me)\b",
-        r"\b(how (does|do|did)|why (does|do|did)|what (is|are|was|were))\b",
-        r"\b(I don'?t (understand|know|get))\b",
-    ]
-    if message:
-        for pattern in learning_patterns:
-            if re.search(pattern, message, re.IGNORECASE):
-                signals["learning_opportunity"] = True
-                break
-
    return signals


@@ -132,27 +132,67 @@ def main():
    check("demote rock=hot (immobile)",
          cc_loader.tier_demote({"wash_count": 0, "rock_status": "rock"}, NOW_G) == "hot")

-    # ── witness + safety + abstraction (W3.4 continued) ──────────────────
+    # ── witness + safety + abstraction (W3.4, FULL-TABLE parity) ──────────
    print("\n--- witness/safety/abstraction kernels ---")
    from rune import witness as wt, safety_override as so, abstraction as ab
+    import re as _re

-    # the module's REAL tables as data (stopwords/travel tables)
-    sw = sorted(wt.extract_topic.__doc__ or "") # placeholder — use the real stopwords from a probe call
-    # extract topics via the PYTHON path first to get the stopwords table
-    topics_py = wt.extract_topic("the deployment runner needs repair now")
-    rs_topics = cc_loader.witness_extract_topic("the deployment runner needs repair now", [])
-    check("witness extract topic", topics_py == rs_topics,
+    # REAL gather seams (PATTERN-TABLE GATHER-SEAM LAW: canaries call the
+    # module's actual seams, not hand-typed tables)
+    sw = wt._stopwords()
+    topics_py = wt._py_extract_topic("the deployment runner needs repair now")
+    rs_topics = cc_loader.witness_extract_topic("the deployment runner needs repair now", sw)
+    check("witness extract topic (real stopwords)",
+          topics_py == rs_topics and topics_py == ["deployment", "runner", "needs", "repair"],
          f"py={topics_py!r} rs={rs_topics!r}"[:200])

    signals_py = wt.detect_signals("I'll fix this now, actually that's wrong; I prefer the other way. urgent!")
-    tables = {"stopwords": [], "correction": [], "commitment": [],
-              "contradiction": [], "preference": [], "urgency": [], "learning": []}
+    tables = wt._tables()
    rs_sigs = cc_loader.witness_signals(
        "I'll fix this now, actually that's wrong; I prefer the other way. urgent!",
        tables)
-    check("witness signals (kernel subset)",
-          isinstance(rs_sigs, dict),
-          f"py={signals_py!r} rs={rs_sigs!r}"[:300])
+    rs_sigs_f = {k: v for k, v in rs_sigs.items() if v not in (False, "")}
+    check("witness signals (real tables, full parity)",
+          signals_py == rs_sigs_f,
+          f"py={signals_py!r} rs={rs_sigs_f!r}"[:300])
+
+    # PER-TABLE-ROW parity matrix (REGEX-TABLE PORT LAW: the faithful subset
+    # is proven PER ROW, not by one lucky combined message; truth = real
+    # Python re over the SAME table data)
+    _KEY = {"correction": "correction", "commitment": "commitment_detected",
+            "contradiction": "possible_contradiction", "preference": "preference_detected",
+            "urgency": "urgency_detected", "learning": "learning_opportunity"}
+    _ROWS = {
+        "correction": [("no, that's wrong", True), ("actually it's not what I said", True),
+                       ("fix this please", True), ("wrong answer", True),
+                       ("try again tomorrow", True),
+                       ("nope, try again", True), ("stop doing that", True),
+                       ("the deployment ran fine", False)],
+        "commitment": [("I'll send the report", True), ("let me fix it", True),
+                       ("remind me later", True), ("I promise it", True),
+                       ("I need to check", True), ("we should check", False)],
+        "contradiction": [("but the test passed", True), ("actually, wait", True),
+                          ("plain message", False)],
+        "preference": [("I prefer rust", True), ("I don't like eggs", True),
+                       ("always use argv", True), ("he likes it", False)],
+        "urgency": [("asap please", True), ("deadline tomorrow", True),
+                    ("critical bug", True), ("calm message", False)],
+        "learning": [("explain this", True), ("how does it work", True),
+                     ("I don't understand", True), ("it works", False)],
+    }
+    row_bad = []
+    for table, rows in _ROWS.items():
+        for msg, expect in rows:
+            exp = any(_re.search(p, msg, _re.IGNORECASE) for p in tables[table])
+            py = wt._py_detect_signals(msg)
+            got = bool(py.get(_KEY[table]))
+            rs = {k: v for k, v in cc_loader.witness_signals(msg, tables).items()
+                  if v not in (False, "")}
+            rs_got = bool(rs.get(_KEY[table]))
+            if not (exp == got == rs_got == expect):
+                row_bad.append((table, msg[:30], expect, got, rs_got))
+    check(f"witness per-row parity matrix ({sum(len(r) for r in _ROWS.values())} rows)",
+          not row_bad, f"{row_bad[:3]!r}"[:300])

    params = cc_loader.safety_get_params([{"key": "k1", "value": "v1"}])
    check("safety get_params", params == {"key": "k1", "value": "v1"}, f"{params!r}")
@@ -553,31 +553,188 @@ pub fn witness_signals(message: &str, tables: &Json,
    Json::Obj(sig)
 }

-fn regex_lite_search(pattern: &str, text: &str) -> bool {
-    let low = text.to_lowercase();
-    let low_pat = pattern.to_lowercase();
-    // faithful subset: word tokens from the pattern (alternation groups
-    // split on |), every alternative must appear as a phrase
+/// Expand top-level alternation of a pattern into full alternative strings
+/// (groups become inline cartesian products of their inner alternatives).
+fn regex_lite_alternatives(pattern: &str) -> Vec<String> {
    let mut alts: Vec<String> = vec![];
+    let mut prefixes: Vec<String> = vec![String::new()];
    let mut cur = String::new();
-    let mut depth = 0;
-    for ch in low_pat.chars() {
+    let mut depth = 0usize;
+    let chars: Vec<char> = pattern.to_lowercase().chars().collect();
+    for ch in chars {
        match ch {
-            '(' => { depth += 1; if depth == 1 { alts.push(std::mem::take(&mut cur)); } }
-            ')' => { depth -= 1; if depth == 0 { alts.push(std::mem::take(&mut cur)); } }
-            '|' if depth == 1 => { alts.push(std::mem::take(&mut cur)); }
-            _ => { cur.push(ch); }
+            '(' => {
+                depth += 1;
+                if depth == 1 {
+                    if !cur.is_empty() {
+                        prefixes = prefixes
+                            .iter()
+                            .map(|p| format!("{p}{cur}"))
+                            .collect();
+                        cur.clear();
+                    }
+                } else {
+                    cur.push('(');
+                }
+            }
+            ')' => {
+                depth = depth.saturating_sub(1);
+                if depth == 0 {
+                    // closing a top-level group: its content is an
+                    // alternation set — expand against the accumulated
+                    // prefixes (cartesian). prefixes.len() may be 1.
+                    let group_alts: Vec<String> = cur.split('|').map(String::from).collect();
+                    let mut next = vec![];
+                    for p in &prefixes {
+                        for g in &group_alts {
+                            next.push(format!("{p}{g}"));
+                        }
+                    }
+                    prefixes = next;
+                    cur.clear();
+                } else if depth >= 1 {
+                    cur.push(')');
+                }
+            }
+            '|' if depth == 0 => {
+                for p in &prefixes {
+                    alts.push(format!("{p}{cur}"));
+                }
+                cur.clear();
+            }
+            _ => cur.push(ch),
        }
    }
-    alts.push(cur);
+    if depth == 0 {
+        for p in &prefixes {
+            alts.push(format!("{p}{cur}"));
+        }
+    } else {
+        // unbalanced group: treat literally
+        alts.push(cur);
+    }
    alts.retain(|a| !a.is_empty());
-    alts.iter().all(|alt| {
-        let cleaned: String = alt.chars()
-            .filter(|c| c.is_alphanumeric() || *c == ' ' || *c == '_')
-            .collect::<String>()
-            .split_whitespace().collect::<Vec<_>>().join(" ");
-        cleaned.is_empty() || low.contains(&cleaned)
-    })
+    if alts.is_empty() {
+        alts.push(String::new());
+    }
+    alts
+}
+
+/// Split ONE alternative into ordered PHRASE-SEGMENTS, each segment being
+/// a set of variant strings (from `x?` optional chars — both shapes).
+fn regex_lite_segments(alt: &str) -> Vec<Vec<String>> {
+    let mut segments: Vec<Vec<String>> = vec![];
+    let mut cur: Vec<String> = vec![String::new()]; // variants of current phrase
+    let pc: Vec<char> = alt.chars().collect();
+    let mut j = 0usize;
+    while j < pc.len() {
+        let c = pc[j];
+        if c == '\\' && j + 1 < pc.len() {
+            let n = pc[j + 1];
+            match n {
+                'b' | 'B' => {} // zero-width anchor: drop
+                's' => {
+                    if cur.iter().any(|v| !v.is_empty()) {
+                        segments.push(std::mem::take(&mut cur));
+                        cur = vec![String::new()];
+                    }
+                    if j + 2 < pc.len() && matches!(pc[j + 2], '*' | '+' | '?') {
+                        j += 3;
+                    } else {
+                        j += 2;
+                    }
+                    continue;
+                }
+                other => {
+                    for v in cur.iter_mut() {
+                        v.push(other)
+                    }
+                }
+            }
+            j += 2;
+            continue;
+        }
+        match c {
+            '?' => {
+                // optional previous char: each variant gains a SHORTER twin
+                let twinned: Vec<String> = cur
+                    .iter()
+                    .flat_map(|v| {
+                        let mut out = vec![v.clone()];
+                        if !v.is_empty() {
+                            out.push(v[..v.len() - 1].to_string());
+                        }
+                        out
+                    })
+                    .collect();
+                cur = twinned;
+            }
+            '.' | '[' => {
+                if cur.iter().any(|v| !v.is_empty()) {
+                    segments.push(std::mem::take(&mut cur));
+                    cur = vec![String::new()];
+                }
+                if c == '[' {
+                    let mut k = j + 1;
+                    while k < pc.len() && pc[k] != ']' {
+                        k += 1;
+                    }
+                    j = k;
+                }
+            }
+            '*' | '+' => {} // gap quantifier
+            '^' | '$' => {}
+            c if c.is_alphanumeric() || c == '\'' || c == ' ' || c == '-' || c == '_' => {
+                for v in cur.iter_mut() {
+                    v.push(c)
+                }
+            }
+            _ => {
+                if cur.iter().any(|v| !v.is_empty()) {
+                    segments.push(std::mem::take(&mut cur));
+                    cur = vec![String::new()];
+                }
+            }
+        }
+        j += 1;
+    }
+    if cur.iter().any(|v| !v.is_empty()) {
+        segments.push(cur);
+    }
+    segments
+}
+
+fn regex_lite_search(pattern: &str, text: &str) -> bool {
+    let low = text.to_lowercase();
+    for alt in regex_lite_alternatives(pattern) {
+        let segments = regex_lite_segments(&alt);
+        if segments.is_empty() {
+            continue;
+        }
+        // DFS: every segment contributes ONE of its variants, in order,
+        // gap-matched against the text.
+        fn walk(segments: &[Vec<String>], low: &str, cursor: usize) -> bool {
+            if segments.is_empty() {
+                return true;
+            }
+            for variant in &segments[0] {
+                if variant.is_empty() {
+                    return walk(&segments[1..], low, cursor);
+                }
+                if let Some(ix) = low[cursor..].find(variant.as_str()) {
+                    let next = cursor + ix + variant.len();
+                    if walk(&segments[1..], low, next) {
+                        return true;
+                    }
+                }
+            }
+            false
+        }
+        if walk(&segments, &low, 0) {
+            return true;
+        }
+    }
+    false
 }

 // ── safety_override kernels (W3.4) ─────────────────────────────────────────
@@ -6,6 +6,7 @@ import json
 import os
 import subprocess
 import sys
+from pathlib import Path

 import pytest

@@ -585,3 +586,211 @@ def test_15u_discover_active_contract_migrated_to_interface(tmp_path):
    # capabilities() advertises the new op
    assert "discover_active_contract" in \
        iface.capabilities()["state"]["operations"]
+
+
+# ── T5.15 completion wave: stage_guards wired into transition ───────────────
+
+
+def _guard_contract(iface, tmp_path):
+    """A contract whose obligations declare a PATH guard and a failing
+    command guard — proving both BLOCK shapes."""
+    doc = iface.scaffold.load()
+    ss.add_entity(
+        doc, id="contract:guarded-1", kind="contract",
+        identities={
+            "objective": "guarded stage",
+            "verification_obligations": [
+                "guard:" + json.dumps({
+                    "name": "artifact present", "kind": "path_exists",
+                    "path": str(tmp_path / "artifact.txt"),
+                }),
+            ],
+        },
+        evidence_refs=["e"],
+    )
+    iface.scaffold.commit(doc, what="seed guarded contract")
+    doc = iface.scaffold.load()
+    # move through the legal path so guards face verifying from building
+    sc.Scaffold.transition_contract(iface.scaffold, doc, "guarded-1", "approved",
+                                    evidence_ref="t")
+    sc.Scaffold.transition_contract(iface.scaffold, doc, "guarded-1", "building",
+                                    evidence_ref="t")
+    iface.scaffold.commit(doc, what="advance to building")
+    return "guarded-1"
+
+
+def test_15f_transition_to_verifying_blocked_when_guard_fails(tmp_path):
+    iface = _iface(tmp_path)
+    cid = _guard_contract(iface, tmp_path)
+    env = iface.transition(cid, "verifying")
+    assert not env["success"]
+    assert env["errors"][0]["error"] == ci.Errors.VERIFICATION_FAILED
+    assert env["state"]["guard"]["verdict"] == "BLOCKED"
+    # phase UNCHANGED — the blocked stage does not move
+    doc = iface.scaffold.load()
+    c = iface.scaffold.contract(doc, cid)
+    assert c["phase"] == "building"
+
+
+def test_15f_transition_passes_when_guard_satisfied(tmp_path):
+    iface = _iface(tmp_path)
+    cid = _guard_contract(iface, tmp_path)
+    (tmp_path / "artifact.txt").write_text("real artifact")
+    env = iface.transition(cid, "verifying")
+    assert env["success"], env["errors"]
+    doc = iface.scaffold.load()
+    assert iface.scaffold.contract(doc, cid)["phase"] == "verifying"
+
+
+def test_15f_inline_guard_checks_override_obligations(tmp_path):
+    iface = _iface(tmp_path)
+    cid = _guard_contract(iface, tmp_path)  # has the path guard, NOT satisfied
+    env = iface.transition(
+        cid, "verifying",
+        guard_checks=[{"name": "inline truth", "kind": "command",
+                       "argv": [sys.executable, "-c", "print('ok')"]}],
+    )
+    assert env["success"], env["errors"]
+    assert iface.scaffold.contract(iface.scaffold.load(), cid)["phase"] == "verifying"
+
+
+def test_15f_transition_to_building_runs_no_guards(tmp_path):
+    """Completion-facing edges ONLY (verifying/closed): building→verifying
+    is guarded, but draft→approved passes freely even with obligations."""
+    iface = _iface(tmp_path)
+    doc = iface.scaffold.load()
+    ss.add_entity(doc, id="contract:guarded-2", kind="contract",
+                  identities={"objective": "2"}, evidence_refs=["e"])
+    iface.scaffold.commit(doc, what="seed guarded-2")
+    cid = "guarded-2"
+    doc = iface.scaffold.load()
+    c = iface.scaffold.contract(doc, cid)
+    c["identities"]["verification_obligations"] = [
+        "guard:" + json.dumps({"name": "never", "kind": "path_exists",
+                               "path": "/nonexistent/for-sure"})
+    ]
+    iface.scaffold.commit(doc, what="attach guard")
+    env = iface.transition(cid, "approved")  # NOT a completion edge
+    assert env["success"]
+
+
+# ── T5.15 slice 3: programme_continuation wired into close() ────────────────
+
+
+def _seed_programme(iface, programme_id, stage_contract_ids, tmp_path):
+    """Persist a real Programme bound to the interface's scaffold store."""
+    from rune import programme_continuation as pc
+
+    base = Path(iface.scaffold.current).parent / pc.DEFAULT_PROGRAMME_AREA
+    prog = pc.Programme(programme_id=programme_id, objective="wave test")
+    for i, cid in enumerate(stage_contract_ids):
+        prog.stages.append(pc.ProgrammeStage(
+            stage_id=cid, objective=f"stage {i}", depends_on=[]))
+    prog.save(base)
+    return prog, base
+
+
+def test_close_marks_programme_stage_complete_and_surfaces_continuation(tmp_path):
+    """Closing a contract that is a stage of a persisted Programme marks
+    that stage COMPLETE and the envelope carries the continuation canary —
+    the programme's next eligible stage is surfaced, not guessed."""
+    from rune import programme_continuation as pc
+
+    iface = _iface(tmp_path)
+    env1 = _submit(iface, objective="wave stage one")
+    cid1 = env1["ids"]["contract_id"]
+    env2 = _submit(iface, objective="wave stage two")
+    cid2 = env2["ids"]["contract_id"]
+    # drive both through to verifying so close is legal
+    for cid in (cid1, cid2):
+        iface.transition(cid, "approved")
+        iface.transition(cid, "building")
+        iface.transition(cid, "verifying")
+    prog, base = _seed_programme(iface, "wave-x", [cid1, cid2], tmp_path)
+    env = iface.close(cid1, observed_evidence=[{"entity": "capability:exporter", "detail": "done"}])
+    assert env["success"], env["errors"]
+    cont = env["state"].get("programme_continuation")
+    assert cont, "close envelope must carry the programme continuation"
+    assert cont["programme_id"] == "wave-x"
+    assert cont["stage_marked_complete"] == cid1
+    reloaded = pc.Programme.load(base, "wave-x")
+    stage1 = next(s for s in reloaded.stages if s.stage_id == cid1)
+    assert stage1.state == pc.COMPLETE
+    assert cont["canary"]["verdict"] == "PROGRAMME_OPEN"
+    assert cont["canary"]["next_action"] == ("start", cid2)
+
+
+def test_close_without_programme_has_no_continuation_block(tmp_path):
+    iface = _iface(tmp_path)
+    env = _submit(iface)
+    cid = env["ids"]["contract_id"]
+    iface.transition(cid, "approved")
+    iface.transition(cid, "building")
+    iface.transition(cid, "verifying")
+    env = iface.close(cid, observed_evidence=[{"entity": "capability:exporter", "detail": "done"}])
+    assert env["success"]
+    assert "programme_continuation" not in env["state"]
+
+
+def test_close_programme_completion_verdict_surfaces(tmp_path):
+    """Closing the LAST stage yields PROGRAMME_COMPLETE — the canary says
+    stop, the runner never guesses."""
+    iface = _iface(tmp_path)
+    env = _submit(iface)
+    cid = env["ids"]["contract_id"]
+    iface.transition(cid, "approved")
+    iface.transition(cid, "building")
+    iface.transition(cid, "verifying")
+    prog, base = _seed_programme(iface, "solo", [cid], tmp_path)
+    env = iface.close(cid, observed_evidence=[{"entity": "capability:exporter", "detail": "done"}])
+    cont = env["state"]["programme_continuation"]
+    assert cont["canary"]["verdict"] == "PROGRAMME_COMPLETE"
+
+
+def test_two_submits_same_second_get_distinct_contract_ids(tmp_path):
+    """Second-granularity timestamps collide when two needs are filed in
+    the same second (real builder-pair scenario) — ids must be distinct."""
+    iface = _iface(tmp_path)
+    e1 = _submit(iface, objective="wave stage one")
+    e2 = _submit(iface, objective="wave stage two")
+    assert e1["success"] and e2["success"], (e1["errors"], e2["errors"])
+    assert e1["ids"]["contract_id"] != e2["ids"]["contract_id"]
+
+
+# ── T5.15 slice 4: iteration impact-delta recorded at close ─────────────────
+
+
+def test_close_records_impact_delta_when_scope_changed(tmp_path):
+    """A closed contract with scope_files records a predicted-vs-observed
+    impact delta into the project store's impact_deltas.jsonl (rule 19H:
+    deltas compound into better future prediction)."""
+    from rune import iteration as it
+
+    iface = _iface(tmp_path)
+    env = _submit(iface)
+    cid = env["ids"]["contract_id"]
+    for ph in ("approved", "building", "verifying"):
+        iface.transition(cid, ph)
+    iface.close(cid, observed_evidence=[{"entity": "capability:exporter", "detail": "done"}],
+                scope_files=["src/export_module.py"])
+    delta_file = tmp_path / ".rune" / "scaffold" / "impact_deltas.jsonl"
+    assert delta_file.exists(), "close must persist the impact delta"
+    import json as _json
+    deltas = [_json.loads(l) for l in delta_file.read_text().splitlines() if l.strip()]
+    assert any(d.get("contract_id") == cid for d in deltas)
+    d = next(d for d in deltas if d.get("contract_id") == cid)
+    assert d.get("scope_files") == ["src/export_module.py"]
+    assert "recorded_at" in d
+
+
+def test_close_without_scope_records_no_delta(tmp_path):
+    """A governance-only close (no scope files) records nothing — the
+    delta ledger only carries material changes."""
+    iface = _iface(tmp_path)
+    env = _submit(iface)
+    cid = env["ids"]["contract_id"]
+    for ph in ("approved", "building", "verifying"):
+        iface.transition(cid, ph)
+    iface.close(cid, observed_evidence=[{"entity": "capability:exporter", "detail": "done"}])
+    delta_file = tmp_path / ".rune" / "scaffold" / "impact_deltas.jsonl"
+    assert not delta_file.exists()
@@ -0,0 +1,84 @@
+"""T5.15 slice 5: the forge projection wired into the operator failure path.
+
+The projection (testing_mode.project_failure_to_forge) was TEST_PROVEN with
+zero runtime callers. This wire invokes it from the operator's recorded-
+failure sink using the ALREADY-WIRED forge client (create_forge_client —
+NullAdapter when unconfigured → the projection is dormant by construction,
+never an error source).
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+from rune import rune_operator as ro  # noqa: E402
+from rune import testing_mode as tm  # noqa: E402
+
+
+class RecordingAdapter:
+    """Records create_issue calls — proves the projection fired."""
+
+    def __init__(self):
+        self.calls: list[tuple[str, str]] = []
+
+    def create_issue(self, title: str, body: str) -> dict:
+        self.calls.append((title, body))
+        return {"created": True, "number": len(self.calls)}
+
+
+def test_recorded_failure_projects_to_forge(monkeypatch):
+    adapter = RecordingAdapter()
+    monkeypatch.setattr(ro, "_forge_client_for_projection", lambda: adapter)
+    result = ro._project_failure_to_forge(
+        behaviour_id="behaviour-test-1",
+        observation={"expected": "tests pass", "observed": "2 failed",
+                     "observation_id": "obs-1", "reproduction": "pytest -k x"},
+        repo="conductor/rune",
+    )
+    # the projection returns the adapter's own dict (create_issue result) —
+    # the wire is proven by the adapter having been CALLED
+    assert len(adapter.calls) == 1
+    assert result.get("created") is True
+    title, _body = adapter.calls[0]
+    assert "behaviour-test-1" in title
+
+
+def test_null_adapter_projection_is_dormant_not_error(monkeypatch):
+    """Unconfigured forge → the projection returns a dormant marker; the
+    failure path is untouched."""
+    monkeypatch.setattr(ro, "_forge_client_for_projection", lambda: _NullAdapterProbe())
+    result = ro._project_failure_to_forge(
+        behaviour_id="b-2",
+        observation={"expected": "x", "observed": "y"},
+        repo="r",
+    )
+    assert result.get("created") is False
+    assert result.get("reason") == "no forge configured"
+
+
+def test_forge_unavailable_never_breaks_the_failure_path(monkeypatch):
+    from rune import forge as _forge
+
+    def boom():
+        raise _forge.ForgeUnavailable("forge down")
+    monkeypatch.setattr(ro, "_forge_client_for_projection", lambda: boom())
+    result = ro._project_failure_to_forge(
+        behaviour_id="b-3",
+        observation={"expected": "x", "observed": "y"},
+        repo="r",
+    )
+    assert result["projected"] is False
+    assert "forge down" in result.get("error", "")
+
+
+class _NullAdapterProbe:
+    """Mimics forge_client.NullAdapter's shape for the dormant test."""
+
+    def create_issue(self, title: str, body: str) -> dict:
+        return {"created": False, "reason": "no forge configured"}
@@ -0,0 +1,143 @@
+"""spine_language — the #2465 doc-level transforms (T5.15 keep-green:
+the new-module henxel demands these ship WITH tests; they were carved
+from spine.py whose battery covered them, but the carved module is the
+unit under test now)."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+from rune import spine_language as sl  # noqa: E402
+
+
+@pytest.fixture
+def doc():
+    return sl.new_doc()
+
+
+def test_classification_sides_reads_the_measured_map():
+    sides = sl._classification_sides()
+    assert sides, "classification.json must be readable in-repo"
+    assert sides.get("spine") == "RUST"
+    assert sides.get("spine_language") == "RUST"
+
+
+def test_classification_sides_degrades_to_empty_when_unreadable(tmp_path, monkeypatch):
+    monkeypatch.setattr(sl, "ROOT", tmp_path)
+    assert sl._classification_sides() == {}
+
+
+def test_language_verdicts_kind_defaults(doc):
+    sl.add_entity(doc, id="rust-crate:cognition_core", kind="rust-crate")
+    sl.add_entity(doc, id="go-surface:rune-entry", kind="go-surface")
+    sl.add_entity(doc, id="workflow:ci.yml", kind="workflow")
+    sl.add_entity(doc, id="contract:BUILDER-x", kind="contract")
+    sl._place_languages(doc)
+    by = {e["id"]: e["identities"]["language"] for e in doc["entities"]}
+    assert by["rust-crate:cognition_core"] == "rust"
+    assert by["go-surface:rune-entry"] == "go"
+    assert by["workflow:ci.yml"] == "n/a"
+    assert by["contract:BUILDER-x"] == "n/a"
+
+
+def test_language_verdicts_measured_authoritative_then_hints(doc):
+    """Nick's dichotomy correction: the measured map decides; hints only
+    label UNMEASURED entities and say so; nothing silently defaults."""
+    sl.add_entity(doc, id="module:rune_tui", kind="module")   # TUI-LAST measured
+    sl.add_entity(doc, id="module:spine", kind="module")      # RUST measured
+    sl.add_entity(doc, id="module:totally_unknown_name", kind="module")
+    sl._place_languages(doc)
+    by = {e["id"]: e["identities"]["language"] for e in doc["entities"]}
+    assert by["module:rune_tui"] == "python-tui (Wave 6)"   # from the MAP
+    assert by["module:spine"] == "rust"                      # from the MAP
+    assert by["module:totally_unknown_name"] == "unclassified"
+
+
+def test_subsystem_language_from_member_majority(doc):
+    """identity: 7 RUST / 3 GO members per classification.json → mixed with
+    counts carried in the verdict text (never a silent guess)."""
+    for m in ("behavior", "identity_anchors", "persona_definition",
+              "persona_gate", "prompt_sync", "voice_hybrid", "voice_injector",
+              "voice_sync", "voicebox", "safety_override"):
+        sl.add_entity(doc, id=f"module:{m}", kind="module")
+        sl.add_relationship(doc, type="belongs_to",
+                            from_id=f"module:{m}", to_id="subsystem:identity",
+                            evidence_refs=["t"])
+    sl.add_entity(doc, id="subsystem:identity", kind="subsystem")
+    sl._place_languages(doc)
+    e = sl.entity(doc, "subsystem:identity")
+    lang = e["identities"]["language"]
+    assert lang.startswith("mixed (rust ")
+    assert "go" in lang
+
+
+def test_unqualified_discovery_becomes_construction_surface(doc):
+    """A bare file path from a closed contract is NOT a module — provenance
+    preserved, kind honest, candidate stays DECLARED."""
+    monkey_specs = [("example", [("src/rune/audit_ledger.py",
+                                  "construction-surface", "candidate")])]
+    original = sl._contract_discoveries
+    sl._contract_discoveries = lambda: monkey_specs
+    try:
+        placed = sl._apply_contract_discoveries(doc)
+    finally:
+        sl._contract_discoveries = original
+    assert placed == 1
+    e = doc["entities"][0]
+    assert e["id"] == "construction-surface:src/rune/audit_ledger.py"
+    assert e["kind"] == "construction-surface"
+    assert e["maturity"] == "DECLARED"
+    assert e["evidence_refs"] == ["contract-discovery:example"]
+
+
+def test_legacy_unqualified_module_normalized_with_edges(doc):
+    sl.add_entity(doc, id="src/rune/claim_audit.py", kind="module",
+                  maturity="PRESENT",
+                  evidence_refs=["contract-discovery:old"])
+    sl.add_relationship(doc, type="depends_on",
+                        from_id="src/rune/claim_audit.py",
+                        to_id="module:other", evidence_refs=["t"])
+    sl._normalize_discovery_ids(doc)
+    assert doc["entities"][0]["id"] == \
+        "construction-surface:src/rune/claim_audit.py"
+    assert doc["entities"][0]["kind"] == "construction-surface"
+    assert doc["relationships"][0]["from"] == \
+        "construction-surface:src/rune/claim_audit.py"
+    # Not in the observed set (no closed contract recorded it) → the
+    # unsupported PRESENT claim is demoted to DECLARED; evidence survives.
+    assert doc["entities"][0]["maturity"] == "DECLARED"
+    assert doc["entities"][0]["evidence_refs"] == ["contract-discovery:old"]
+
+
+def test_discoveries_gather_reads_real_scaffold_store(tmp_path, monkeypatch):
+    """The gatherer only trusts CLOSED contracts' ext.contract_quality."""
+    store = tmp_path / "scaffold"
+    store.mkdir()
+    doc = sl.new_doc()
+    sl.add_entity(doc, id="contract:C1", kind="contract",
+                  ext={"contract_quality": {
+                      "expected_new_observed": ["rust-crate:kernel_x"],
+                      "candidate_corrections": [
+                          {"entity": "suite"}]}})
+    sl.add_entity(doc, id="contract:C2", kind="contract",
+                  ext={"contract_quality": {
+                      "expected_new_observed": ["rust-crate:never"]}})
+    for e in doc["entities"]:
+        if e["id"] == "contract:C1":
+            e["phase"] = "closed"
+        if e["id"] == "contract:C2":
+            e["phase"] = "building"
+    monkeypatch.setattr(sl, "ROOT", tmp_path)
+    (tmp_path / "store" / "scaffold").mkdir(parents=True, exist_ok=True)
+    (tmp_path / "store" / "scaffold" / "current.json").write_text(
+        json.dumps(doc))
+    got = sl._contract_discoveries()
+    assert got == [("C1", [("rust-crate:kernel_x", "rust-crate", "observed"),
+                           ("suite", "construction-surface", "candidate")])]
@@ -174,3 +174,76 @@ def test_16k_projection_schema_validated(tmp_path):
    assert proj.validate(bad) != []
    bad2 = dict(p, schema_version=999)
    assert proj.validate(bad2) != []
+
+
+# ── T5.15 slice 2: the dispatch wire (command → interface execution) ────────
+
+
+def test_dispatch_executes_accepted_transition_end_to_end(tmp_path):
+    """A START command on a draft contract executes through the interface:
+    the phase actually moves — not just an accepted intent."""
+    store = str(tmp_path / ".rune" / "scaffold")
+    iface = cmds if False else None  # noqa: F841 — readability anchor
+    from rune import construction_interface as ci
+    ciface = ci.ConstructionInterface(builder="seed", project_store=store, project_id="proj")
+    ciface.scaffold.init(purpose="t")
+    doc = ciface.scaffold.load()
+    sc.add_entity(doc, id="contract:disp-1", kind="contract",
+                  identities={"objective": "dispatch"}, evidence_refs=["e"])
+    doc["entities"][-1]["phase"] = "draft"
+    ciface.scaffold.commit(doc, what="seed")
+    out = cmds.dispatch("START", construction_id="disp-1",
+                        store_path=store + "|proj")
+    # external-project store path: dispatch builds its own interface bound
+    # to the same store
+    assert out["executed"] is True, out
+    assert out["success"] is True, out
+    doc2 = ciface.scaffold.load()
+    c = sc.entity(doc2, "contract:disp-1")
+    assert c["phase"] == "approved"
+
+
+def test_dispatch_refusal_is_a_valid_protocol_outcome(tmp_path):
+    """START on a non-draft contract: the lifecycle refuses — the dispatch
+    itself succeeds (success=True) with accepted=False."""
+    store = str(tmp_path / ".rune" / "scaffold")
+    from rune import construction_interface as ci
+    ciface = ci.ConstructionInterface(builder="seed", project_store=store, project_id="proj")
+    ciface.scaffold.init(purpose="t")
+    doc = ciface.scaffold.load()
+    sc.add_entity(doc, id="contract:disp-2", kind="contract",
+                  identities={"objective": "x"}, evidence_refs=["e"])
+    doc["entities"][-1]["phase"] = "draft"
+    ciface.scaffold.commit(doc, what="seed")
+    out = cmds.dispatch("START", construction_id="disp-2",
+                        payload={"phase": "building"},
+                        store_path=store + "|proj")
+    # contract starts in draft so START is ACCEPTED here; drive to building
+    # first to prove the refusal shape:
+    assert out["executed"] and out["success"]
+    out2 = cmds.dispatch("START", construction_id="disp-2",
+                         store_path=store + "|proj")
+    assert out2["accepted"] is False and out2["executed"] is False
+    assert out2["success"] is True
+    assert "draft" in out2["reason"]
+
+
+def test_dispatch_idempotent_replay_does_not_double_execute(tmp_path):
+    """Same command_id twice: the second is a replay — the phase does not
+    move twice and idempotent_replay=True comes back."""
+    store = str(tmp_path / ".rune" / "scaffold")
+    from rune import construction_interface as ci
+    ciface = ci.ConstructionInterface(builder="seed", project_store=store, project_id="proj")
+    ciface.scaffold.init(purpose="t")
+    doc = ciface.scaffold.load()
+    sc.add_entity(doc, id="contract:disp-3", kind="contract",
+                  identities={"objective": "idem"}, evidence_refs=["e"])
+    doc["entities"][-1]["phase"] = "draft"
+    ciface.scaffold.commit(doc, what="seed")
+    first = cmds.dispatch("START", construction_id="disp-3",
+                          command_id="fixed-cmd-1", store_path=store + "|proj")
+    assert first["executed"] and first["success"]
+    second = cmds.dispatch("START", construction_id="disp-3",
+                           command_id="fixed-cmd-1", store_path=store + "|proj")
+    assert second["idempotent_replay"] is True
+    assert second.get("executed") in (False, None)  # replay ≠ execution
@@ -0,0 +1,115 @@
+"""T5.13/T5.14 frozen contracts: pane order and post-response silence.
+
+The pane swap is structural (compose order); the dim-spam consolidation
+is behavioral (one collapsed line instead of five dim messages).
+"""
+
+from __future__ import annotations
+
+import re
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+from rune import rune_tui  # noqa: E402 — the TUI lives in the package
+
+
+def test_chat_channel_is_yielded_first_in_compose():
+    src = (ROOT / "src" / "rune" / "rune_tui.py").read_text()
+    body = re.search(
+        r'with Horizontal\(id="channels"\):(.*?)with Horizontal\(id="footer-bar"\)',
+        src,
+        re.DOTALL,
+    ).group(1)
+    order = re.findall(r'ChannelView\(\s*"(\w+)"', body)
+    assert order == ["chat", "task"], f"compose yields {order}"
+
+
+def test_resize_actions_follow_the_new_visual_positions():
+    """After the swap LEFT=chat, RIGHT=task. resize_left shrinks the LEFT
+    (chat) — which GROWS the task share; resize_right is the reverse."""
+    assert "grow the RIGHT (task)" in rune_tui.RuneApp.action_resize_left.__doc__
+    assert "shrink the RIGHT (task)" in rune_tui.RuneApp.action_resize_right.__doc__
+    # Behavioral proof against an instance WITHOUT constructing the App
+    # (reactive initialization needs a running textual app; the docstring
+    # carries the contract, the arithmetic is what matters).
+    app = SimpleNamespace(
+        task_width=50,
+        action_resize_left=None,
+        action_resize_right=None,
+        _apply_resize=None,
+    )
+    app._apply_resize = lambda: None
+    # Bind the real methods onto the stand-in and call them.
+    rune_tui.RuneApp.action_resize_left(app)
+    assert app.task_width == 55
+    rune_tui.RuneApp.action_resize_right(app)
+    assert app.task_width == 50
+
+
+def _isolate_collector(monkeypatch, **modules):
+    monkeypatch.setattr(rune_tui, "BACKEND_AVAILABLE", True)
+    for name, value in modules.items():
+        monkeypatch.setattr(rune_tui, name, value)
+
+
+def test_collector_returns_all_signals_in_one_pass(monkeypatch):
+    _isolate_collector(
+        monkeypatch,
+        provider_budget=SimpleNamespace(
+            get_status=lambda: {
+                "providers": {"p": {"monthly_limit": 100, "monthly_used": 90, "token_limit": 0, "tokens_used": 0}}
+            }
+        ),
+        claim_audit=SimpleNamespace(scan_text=lambda _t: {"degraded": ["a", "b"]}),
+        growth_delta=SimpleNamespace(check=lambda: {"delta": "focus: +1"}),
+        self_improve=SimpleNamespace(run=lambda: None),
+    )
+    signals = rune_tui.collect_post_response_signals("response text", messages_count=10)
+    assert signals == ["⚠ Budget: 90% used", "⚠ Claims degraded: 2", "📈 Growth: focus: +1", "🔄 Self-improvement cycle triggered"]
+
+
+def test_collector_silence_when_nothing_fires(monkeypatch):
+    _isolate_collector(
+        monkeypatch,
+        provider_budget=Mock(get_status=lambda: {"providers": {}}),
+        claim_audit=Mock(scan_text=lambda _t: None),
+        growth_delta=Mock(check=lambda: None),
+        self_improve=Mock(run=lambda: None),
+    )
+    collector = rune_tui.collect_post_response_signals
+    app_signals = collector("plain", messages_count=7)  # not a cycle multiple
+    assert app_signals == []
+
+
+def test_collector_never_raises_on_backend_noise(monkeypatch):
+    def boom(*_a, **_k):
+        raise RuntimeError("backend gone")
+
+    _isolate_collector(
+        monkeypatch,
+        provider_budget=Mock(get_status=boom),
+        claim_audit=Mock(scan_text=boom),
+        growth_delta=Mock(check=boom),
+        self_improve=Mock(run=boom),
+    )
+    assert rune_tui.collect_post_response_signals("x", 10) == []
+
+
+def test_self_improve_only_at_cycle_multiple(monkeypatch):
+    ran = []
+    _isolate_collector(
+        monkeypatch,
+        provider_budget=Mock(get_status=lambda: {"providers": {}}),
+        claim_audit=Mock(scan_text=lambda _t: None),
+        growth_delta=Mock(check=lambda: None),
+        self_improve=Mock(run=lambda: ran.append(1)),
+    )
+    rune_tui.collect_post_response_signals("x", 7)
+    assert ran == []
+    rune_tui.collect_post_response_signals("x", 10)
+    assert ran == [1]