config/mimo-hooks/builder-bridge.ts

@@ -0,0 +1,124 @@
+import { execFile } from "node:child_process"
+import { existsSync, appendFileSync } from "node:fs"
+
+// builder-bridge — Rune Core preflight for the MiMo builder (handoff
+// 2026-09-10 §3–§7, Option C bridge tranche). For each user turn, the hook
+// extracts the user's message from the outgoing LLM request, runs the Rune
+// Core builder-bridge preflight (deterministic: route classification +
+// memory selection + active-Scaffolding projection, ZERO model calls), and
+// appends a compact capsule to the final user message. Same Rune Core the
+// TUI/operator path uses — this file is a thin adapter with no cognition.
+//
+// CANONICAL SOURCE: config/mimo-hooks/builder-bridge.ts
+// Installed to ~/.mimocode/hooks/builder-bridge.ts by
+// `scripts/install_mimo_hook.py` (hash-verified, drift-detecting — the same
+// contract as event-wait-guard.ts). T22: this file was previously
+// UNVERSIONED and hardcoded an ephemeral interpreter path (/tmp/ci-venv),
+// which silently killed the whole edge; both defects are repaired here.
+//
+// Overhead contract (§7/§8): capsule ≤ ~300 tokens; a repeat/empty task
+// costs ~0ms (TTL cache); a fresh task ~150ms (bounded retrieval). The hook
+// degrades silently to no-injection on any failure — it must never block
+// the builder.
+//
+// Injection policy: only append when the capsule is non-empty. A plain
+// "hello" gets nothing (§5: if nothing relevant exists, inject nothing).
+
+const BRIDGE = "/mnt/workdrive/issue161-harness/src/rune/builder_bridge_cli.py"
+const MARKER_PREFIX = "/tmp/rune-bridge-"
+
+// Interpreter resolution (T22 repair): the previous hardcoded
+// "/tmp/ci-venv/bin/python" is an EPHEMERAL path that no longer exists, so
+// every bridge call failed and the hook silently degraded — the MiMo edge
+// received no Rune Core preflight at all (zero fire-log lines). Resolve
+// against durable candidates, first existing wins; env override wins over
+// all of them. If nothing exists the hook degrades as before.
+const PY_CANDIDATES = [
+  process.env.MIMO_BRIDGE_PY,
+  "/mnt/workdrive/issue161-harness/.test-venv/bin/python",
+  "/home/nick/.venvs/memory/bin/python3",
+  "/usr/bin/python3",
+].filter((p) => typeof p === "string" && p.length > 0)
+
+function resolvePy() {
+  for (const p of PY_CANDIDATES) {
+    try {
+      if (existsSync(p)) return p
+    } catch {}
+  }
+  return null
+}
+
+// Per-process dedupe: identical task text within the cache TTL does not need
+// a second bridge call — the CLI has its own TTL cache, but skipping the
+// process spawn entirely for the same message text within 10 minutes keeps
+// repeats at ~0ms.
+const lastText = new Map() // text -> timestamp
+const TTL = 600_000
+
+function extractUserText(messages) {
+  // messages: Array<{ info: Message, parts: Part[] }> — newest last.
+  for (let i = messages.length - 1; i >= 0; i--) {
+    const m = messages[i]
+    if (m?.info?.role !== "user") continue
+    const texts = []
+    for (const p of m.parts ?? []) {
+      if (p?.type === "text" && typeof p.text === "string") texts.push(p.text)
+    }
+    if (texts.length) return texts.join("\n").slice(0, 4000)
+  }
+  return ""
+}
+
+export default {
+  "experimental.chat.messages.transform": async (input, output) => {
+    try {
+      if (!Array.isArray(output.messages) || output.messages.length === 0) return
+      const userText = extractUserText(output.messages)
+      if (!userText || userText.length < 12) return // trivial/noise guard
+      // Harness-internal steps (checkpoint writer, compaction, system
+      // reminders) are not builder tasks — no capsule (measured: the fire
+      // log showed checkpoint-writer steps receiving capsules 2026-09-10).
+      if (/^(<system-reminder>|EXTRACTION REQUIRED)/.test(userText.trim())) return
+      const now = Date.now()
+      const prev = lastText.get(userText)
+      if (prev !== undefined && now - prev < TTL) return // dedupe repeats
+      lastText.set(userText, now)
+      if (lastText.size > 64) {
+        // evict oldest
+        let oldestKey = null, oldestVal = Infinity
+        for (const [k, v] of lastText) if (v < oldestVal) { oldestVal = v; oldestKey = k }
+        if (oldestKey) lastText.delete(oldestKey)
+      }
+
+      const py = resolvePy()
+      if (!py) return // no durable interpreter — degrade (never block)
+
+      const capsule = await new Promise((resolve) => {
+        execFile(py, [BRIDGE, userText], { timeout: 15_000 }, (err, stdout) => {
+          resolve(err ? "" : (stdout?.toString() ?? ""))
+        })
+      })
+      if (!capsule.trim()) return
+
+      // Durable fire-trace: one line per ACTUAL injection into an outgoing
+      // model request (handoff §1 proof requirement). Not a canary — this is
+      // standing observability for the bridge's consumer edge.
+      try {
+        appendFileSync("/tmp/rune-bridge-fire.log",
+          JSON.stringify({ ts: new Date().toISOString(), task: userText.slice(0, 80), capsule_tokens: Math.round(capsule.length / 4), py }) + "\n")
+      } catch {}
+
+      // Append the capsule to the LAST user message (the one being answered).
+      const last = output.messages[output.messages.length - 1]
+      if (last?.parts && Array.isArray(last.parts)) {
+        last.parts.push({
+          type: "text",
+          text: "\n\n<rune-preflight>\n" + capsule.trim() + "\n</rune-preflight>\n(Injected by Rune Core builder bridge — deterministic preflight, zero model calls. Relevant state above; ignore if not applicable to this task.)",
+        })
+      }
+    } catch {
+      // never block the builder on bridge failure
+    }
+  },
+}

requirements-ci.txt

@@ -1,5 +1,5 @@
 # This file was autogenerated by uv via the following command:
-#    uv pip compile pyproject.toml -o /tmp/req_new.txt --extra dev --extra memory --python-version 3.11
+#    uv pip compile pyproject.toml -o requirements-ci.txt --extra dev --extra memory --python-version 3.11
 aiohappyeyeballs==2.7.1
     # via aiohttp
 aiohttp==3.14.3
@@ -20,7 +20,7 @@ chardet==7.6.0
     # via diff-cover
 coverage==7.16.1
     # via rune (pyproject.toml)
-diff-cover==10.5.1
+diff-cover==10.6.0
     # via rune (pyproject.toml)
 frozenlist==1.8.0
     # via
@@ -56,7 +56,7 @@ mdit-py-plugins==0.6.1
     # via textual
 mdurl==0.1.2
     # via markdown-it-py
-multidict==6.9.0
+multidict==6.9.1
     # via
     #   aiohttp
     #   yarl
@@ -64,7 +64,7 @@ nodeenv==1.10.0
     # via pyright
 packaging==26.3
     # via pytest
-platformdirs==4.11.11
+platformdirs==4.11.12
     # via textual
 pluggy==1.6.0
     # via

scripts/install_mimo_hook.py

@@ -1,43 +1,71 @@
 #!/usr/bin/env python3
-"""install_mimo_hook.py — T21.2 §21.2D: canonical MiMo hook installer.
+"""install_mimo_hook.py — T21.2 §21.2D (T22: multi-hook): canonical MiMo
+hook installer.
 
-Canonical source: config/mimo-hooks/event-wait-guard.ts
-Target: ~/.mimocode/hooks/event-wait-guard.ts
-Verifies hash after install. Detects drift in both directions."""
+Canonical sources: config/mimo-hooks/*.ts
+Targets:          ~/.mimocode/hooks/<same-name>.ts
+
+Every hook is hash-verified after install; drift is detected in both
+directions. T22 added builder-bridge.ts to the managed set — it was
+previously installed by hand, unversioned, which is how it silently rotted
+(hardcoded ephemeral interpreter). A hook that reaches the builder must be
+version-controlled and drift-checked like any other authority.
+
+Usage:
+    python3 scripts/install_mimo_hook.py            # sync all managed hooks
+    python3 scripts/install_mimo_hook.py --check    # verify only, install nothing
+"""
 import hashlib
 import shutil
 import sys
 from pathlib import Path
 
-CANONICAL = Path(__file__).resolve().parent.parent / "config" / "mimo-hooks" / "event-wait-guard.ts"
-TARGET = Path.home() / ".mimocode" / "hooks" / "event-wait-guard.ts"
+REPO = Path(__file__).resolve().parent.parent
+CANONICAL_DIR = REPO / "config" / "mimo-hooks"
+TARGET_DIR = Path.home() / ".mimocode" / "hooks"
+
+# The managed set. Adding a file here is what makes a hook version-controlled.
+MANAGED = ["event-wait-guard.ts", "builder-bridge.ts"]
 
 
 def sha256(p: Path) -> str:
     return hashlib.sha256(p.read_bytes()).hexdigest()
 
 
-def main() -> int:
-    if not CANONICAL.is_file():
-        print(f"canonical source missing: {CANONICAL}")
-        return 1
-    canon_hash = sha256(CANONICAL)
-    if TARGET.is_file():
-        target_hash = sha256(TARGET)
-        if target_hash == canon_hash:
-            print(f"in-sync: {target_hash[:12]}")
-            return 0
-        print(f"DRIFT: canonical={canon_hash[:12]} installed={target_hash[:12]}")
+def sync_one(name: str, check_only: bool) -> bool:
+    canonical = CANONICAL_DIR / name
+    target = TARGET_DIR / name
+    if not canonical.is_file():
+        print(f"canonical source missing: {canonical}")
+        return False
+    canon_hash = sha256(canonical)
+    if target.is_file() and sha256(target) == canon_hash:
+        print(f"in-sync: {name} {canon_hash[:12]}")
+        return True
+    if check_only:
+        state = "DRIFT" if target.is_file() else "MISSING"
+        inst = sha256(target)[:12] if target.is_file() else "-"
+        print(f"{state}: {name} canonical={canon_hash[:12]} installed={inst}")
+        return False
+    if target.is_file():
+        print(f"DRIFT: {name} canonical={canon_hash[:12]} installed={sha256(target)[:12]}")
     else:
-        print(f"hook missing at {TARGET}")
-    TARGET.parent.mkdir(parents=True, exist_ok=True)
-    shutil.copy2(CANONICAL, TARGET)
-    installed_hash = sha256(TARGET)
-    if installed_hash != canon_hash:
-        print(f"INSTALL FAILED: hash mismatch after copy")
-        return 1
-    print(f"installed: {installed_hash[:12]}")
-    return 0
+        print(f"hook missing at {target}")
+    TARGET_DIR.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(canonical, target)
+    if sha256(target) != canon_hash:
+        print(f"INSTALL FAILED: {name} hash mismatch after copy")
+        return False
+    print(f"installed: {name} {canon_hash[:12]}")
+    return True
+
+
+def main() -> int:
+    check_only = "--check" in sys.argv
+    ok = True
+    for name in MANAGED:
+        ok = sync_one(name, check_only) and ok
+    return 0 if ok else 1
 
 
 if __name__ == "__main__":

src/rune/builder_bridge.py

@@ -64,6 +64,7 @@ class BuilderCapsule:
     memory_lines: list = field(default_factory=list)
     lesson_lines: list = field(default_factory=list)
     constraints: list = field(default_factory=list)
+    construction_lines: list = field(default_factory=list)
     generated_at: str = ""
     core_version: str = "rune-core/operator-path"
 
@@ -75,6 +76,7 @@ class BuilderCapsule:
             "memory_lines": self.memory_lines[:MAX_MEMORY_LINES],
             "lesson_lines": self.lesson_lines[:MAX_LESSON_LINES],
             "constraints": self.constraints,
+            "construction_lines": self.construction_lines,
             "generated_at": self.generated_at,
             "core_version": self.core_version,
         }
@@ -87,10 +89,14 @@ class BuilderCapsule:
             and not d["lesson_lines"]
             and not d["unknowns"]
             and not d["constraints"]
+            and not d["construction_lines"]
             and d["task_route"] in ("LOCAL_FIX", "STANDARD_CHANGE")
         ):
             return ""  # nothing useful — inject nothing (handoff §5)
         lines = [f"### Rune Core preflight ({d['task_route']})"]
+        if d["construction_lines"]:
+            lines.append("**Active Scaffolding construction (read-only projection):**")
+            lines += [f"- {c}" for c in d["construction_lines"]]
         if d["memory_lines"]:
             lines.append("**Relevant Rune memory:**")
             lines += [f"- {m}" for m in d["memory_lines"]]
@@ -107,6 +113,56 @@ class BuilderCapsule:
         return "\n".join(lines)
 
 
+def _construction_store() -> str:
+    """The SAME construction store the operator path uses (env override
+    honoured; default derives from the repo root, never a bare relative
+    path). Kept identical to rune_operator._construction_store."""
+    import os as _os
+    env = _os.environ.get("RUNE_CONSTRUCTION_STORE")
+    if env:
+        return env
+    from pathlib import Path as _P
+    return str(_P(__file__).resolve().parent.parent.parent / "store" / "scaffold")
+
+
+def _active_construction_lines() -> list:
+    """Read-only projection of the ACTIVE Scaffolding construction into
+    builder-consumable lines. This is the MiMo edge of the shared
+    Construction Context (§15) — it surfaces canonical state; it is NOT a
+    second construction authority and never mutates. Degrade-safe: any
+    failure returns [] (the bridge must never block the builder)."""
+    try:
+        from .construction_interface import ConstructionInterface
+        env = ConstructionInterface(builder="mimo-bridge",
+                                    project_store=_construction_store(),
+                                    project_id="issue161-harness"
+                                    ).construction_context()
+        st = env.get("state") or {}
+        if not st.get("active"):
+            return []
+        out = [f"contract {st.get('contract_id')} · phase {st.get('phase')}"
+               f" · status {st.get('status')}"
+               f" · mode {st.get('construction_mode')}"]
+        if st.get("project"):
+            out.append(f"project: {st['project']}")
+        if st.get("need"):
+            out.append(f"need: {st['need']}")
+        scope = st.get("scope") or []
+        if scope:
+            out.append("scope: " + ", ".join(str(x) for x in scope[:8]))
+        obl = st.get("obligations") or []
+        if obl:
+            out.append(f"outstanding obligations: {len(obl)}")
+            for o in obl[:3]:
+                txt = o.get("obligation") or o.get("kind") or o
+                out.append(f"  - {str(txt)[:120]}")
+        if st.get("next_action"):
+            out.append(f"next action: {st['next_action']}")
+        return out
+    except Exception:  # noqa: BLE001 — projection only; never block the builder
+        return []
+
+
 def _default_selector():
     """The SAME MemorySelector the operator path uses (lazy, degrade-safe)."""
     try:
@@ -166,6 +222,11 @@ def preflight(task_text: str, memory_selector=None) -> BuilderCapsule:
     # design (handoff §9: raw SEED records are outcome telemetry, not lessons).
     capsule.lesson_lines = _validated_lessons_for(task_text or "")
 
+    # Active Scaffolding construction state (read-only projection of the
+    # shared Construction Context). Empty when no contract is active, so
+    # the inject-nothing policy is preserved for ordinary turns.
+    capsule.construction_lines = _active_construction_lines()
+
     return capsule
 
 

src/rune/construction_interface.py

@@ -181,9 +181,16 @@ class ConstructionInterface:
         if contract_id:
             target = self.scaffold.contract(doc, contract_id)
         else:
-            # most recent non-closed contract = the active construction
+            # Most recent ACTIVE contract = status open AND phase non-terminal.
+            # T22 repair: a contract whose phase is terminal (abandoned/closed
+            # — see scaffold.TERMINAL_PHASES) is NOT active construction, even
+            # if its status field was left "open". Without the phase test the
+            # projection reported stale abandoned contracts as the active
+            # construction (8 such records observed in the live store).
+            from .scaffold import TERMINAL_PHASES
             open_contracts = [e for e in contracts
-                              if e.get("status") == "open"]
+                              if e.get("status") == "open"
+                              and e.get("phase") not in TERMINAL_PHASES]
             target = (sorted(open_contracts,
                              key=lambda e: e.get("observed_at", ""))[-1]
                       if open_contracts else None)

src/rune/rune_operator.py

@@ -805,13 +805,22 @@ class RuneOperator:
         except Exception:  # noqa: S110 — notes are best-effort by design
             pass
 
-    def _construction_activation(self, description: str, task_id: str) -> None:
+    def _construction_activation(self, description: str, task_id: str,
+                                 channel: str = OPERATOR_CHANNEL) -> None:
         """SCAFFOLD-TRANCHE-0011/0012 §11: contextual construction
         classification + activation on the NORMAL input path. Deterministic;
         the mode lands in the durable scaffold store where the shared
         Construction Context (and the future TUI) reads it. Delegates all
         lifecycle decisions to construction_mode/construction_activation —
-        NO duplicate intelligence here."""
+        NO duplicate intelligence here.
+
+        §17J criterion 1 (T22 wiring): the canonical intent layer
+        (construction_intent) resolves MEANING before the ONE
+        classifier decides mode — the boundary the intent module
+        documents ("the intent layer resolves MEANING; classify_mode
+        decides mode"). A turn the intent layer resolves as
+        DISCUSSION is not a construction demand, so no construction
+        state is recorded for it."""
         from .construction_mode import classify_mode, _STANDALONE_WORDS
         from .construction_activation import activate
         # §15J case F: manual override feeds the SAME decision path — a
@@ -849,6 +858,56 @@ class RuneOperator:
         from .scaffold import Scaffold
         s = Scaffold(self._construction_store(), "issue161-harness")
         doc = s.load()
+        # §17J criterion 1: meaning gate BEFORE the classifier.
+        # Context = the operator's own canonical conversation window
+        # (no new state store); the current message is excluded so it
+        # is not double-recorded as its own referent.
+        try:
+            from .construction_intent import (
+                ConversationContext, resolve_construction_intent)
+            ctx = ConversationContext()
+            for _m in self.conversation_window(channel, n=8):
+                _txt = str(_m.get("content", ""))
+                if _txt and _txt != description:
+                    ctx.record_turn(str(_m.get("role", "user")), _txt)
+            _intent = resolve_construction_intent(description, ctx)
+        except Exception as exc:  # noqa: BLE001 — intent layer
+            # unavailable: PRESERVE the pre-existing behaviour exactly
+            # (classify+activate) rather than silently changing
+            # governance in an unproven direction. The note is appended to
+            # THIS doc (not via _construction_note, which does its own
+            # load/save) because the normal path saves `doc` afterwards —
+            # a separate write would be clobbered (lost update).
+            _intent = None
+            doc.setdefault("observations", []).append({
+                "kind": "construction_note",
+                "text": (f"intent layer unavailable ({exc}); "
+                         f"classifier path unchanged")[:300],
+                "observed_at": __import__("datetime").datetime.utcnow()
+                .strftime("%Y-%m-%dT%H:%M:%SZ")})
+        # An explicit user override is USER AUTHORITY and outranks the
+        # meaning gate (§15J case F: the override is already carried with
+        # user provenance into the SAME classify_mode call).
+        if (override is None and _intent is not None
+                and _intent.action_intent == "DISCUSSION"):
+            # The intent layer resolved this turn as DISCUSSION: it is not a
+            # construction demand, so no construction state is created. The
+            # record keeps the SAME canonical observation kind/shape every
+            # other input produces (mode vocabulary unchanged — DISCUSSION
+            # maps to NOT_CONSTRUCTION, "no construction state"), with the
+            # intent evidence retained so the verdict is auditable.
+            doc.setdefault("observations", []).append({
+                "kind": "construction_activated",
+                "task": task_id,
+                "mode": "NOT_CONSTRUCTION",
+                "activated": False,
+                "lifecycle": "",
+                "classification_evidence": _intent.classification_evidence[:3],
+                "intent_layer": "DISCUSSION",
+                "observed_at": __import__("datetime").datetime.utcnow()
+                .strftime("%Y-%m-%dT%H:%M:%SZ")})
+            s.save(doc)
+            return
         rec = activate(mode_decision, need=type("N", (), {
             "objective": description,
             "project": "rune",
@@ -929,7 +988,8 @@ class RuneOperator:
         # failure NEVER blocks the ordinary task path (same lifecycle
         # guarantee as the route above).
         try:
-            self._construction_activation(description, state_task.task_id)
+            self._construction_activation(description, state_task.task_id,
+                                          channel=channel)
         except Exception as exc:  # noqa: BLE001 — construction must not break tasks
             try:
                 self._task_state.transition(

src/rune/scaffold.py

@@ -57,7 +57,7 @@ from rune.structural_store import (  # shared generic core — no Rune constants
     task_entities,
 )
 
-__all__ = ["Scaffold", "PHASES", "LEGAL_TRANSITIONS", "REPO_PLACEMENTS", "REPO_SECRET_FIELDS"]
+__all__ = ["Scaffold", "PHASES", "LEGAL_TRANSITIONS", "TERMINAL_PHASES", "REPO_PLACEMENTS", "REPO_SECRET_FIELDS"]
 
 # ── constants ─────────────────────────────────────────────────────────────
 
@@ -73,6 +73,10 @@ LEGAL_TRANSITIONS = {
     "failed": {"building", "abandoned"},  # retry path
     "abandoned": set(),
 }
+# Terminal phases are DERIVED from the canonical transition table: a
+# phase with no legal successor ends the contract. One authority —
+# consumers must not re-declare their own terminal list.
+TERMINAL_PHASES = frozenset(p for p, nxt in LEGAL_TRANSITIONS.items() if not nxt)
 STATUSES = {"open", "closed", "failed", "abandoned"}
 REPO_PLACEMENTS = {"RUNE_REPO", "EXISTING_PROJECT_REPO", "NEW_MANAGED_REPO"}
 REPO_SECRET_FIELDS = {"token", "api_key", "password", "secret", "auth_header", "credential", "api_token"}

tests/test_builder_bridge.py

@@ -23,6 +23,83 @@ from rune.builder_bridge import (  # noqa: E402
 )
 
 
+class ConstructionProjectionActive(unittest.TestCase):
+    """T22: with a LIVE contract the projection surfaces canonical state
+    (contract/phase/scope/obligations/next-action) and renders it in the
+    capsule markdown; the read-only projection never mutates the store."""
+
+    def _active_store(self, td):
+        from pathlib import Path
+        from rune.scaffold import Scaffold
+        store = Path(td) / "scaffold"
+        sc = Scaffold(store, "fixture-proj")
+        sc.init(purpose="fixture")
+        doc = sc.load()
+        sc.new_contract(doc, contract_id="c-t22", need="add the retry backoff",
+                        origin="user", scope=["src/rune/retry.py"],
+                        requirement="prove the projection")
+        sc.save(doc)
+        return str(store)
+
+    def test_active_contract_projects_and_renders(self):
+        import tempfile
+        from pathlib import Path
+        with tempfile.TemporaryDirectory() as td:
+            store = self._active_store(td)
+            old = os.environ.get("RUNE_CONSTRUCTION_STORE")
+            os.environ["RUNE_CONSTRUCTION_STORE"] = store
+            try:
+                lines = builder_bridge._active_construction_lines()
+                self.assertTrue(lines, "an open non-terminal contract must project")
+                self.assertIn("c-t22", lines[0])
+                self.assertTrue(any("scope" in ln for ln in lines), lines)
+                cap = builder_bridge.preflight("add the retry backoff")
+                self.assertTrue(cap.construction_lines)
+                md = cap.to_markdown()
+                self.assertIn("Active Scaffolding construction", md)
+                # read-only: the store is unchanged by projection
+                before = (Path(store) / "current.json").read_text()
+                builder_bridge._active_construction_lines()
+                self.assertEqual(before, (Path(store) / "current.json").read_text())
+            finally:
+                if old is None:
+                    os.environ.pop("RUNE_CONSTRUCTION_STORE", None)
+                else:
+                    os.environ["RUNE_CONSTRUCTION_STORE"] = old
+
+    def test_store_resolution_honours_env_override(self):
+        self.assertEqual(builder_bridge._construction_store(),
+                         os.environ.get("RUNE_CONSTRUCTION_STORE"))
+
+
+class ConstructionProjection(unittest.TestCase):
+    """T22: the bridge surfaces the ACTIVE Scaffolding construction (the
+    shared Construction Context) to the builder — read-only projection,
+    inject-nothing preserved when no contract is active."""
+
+    def test_no_active_contract_projects_nothing(self):
+        import tempfile
+        from pathlib import Path
+        with tempfile.TemporaryDirectory() as td:
+            old = os.environ.get("RUNE_CONSTRUCTION_STORE")
+            os.environ["RUNE_CONSTRUCTION_STORE"] = str(Path(td) / "scaffold")
+            try:
+                self.assertEqual(builder_bridge._active_construction_lines(), [])
+            finally:
+                if old is None:
+                    os.environ.pop("RUNE_CONSTRUCTION_STORE", None)
+                else:
+                    os.environ["RUNE_CONSTRUCTION_STORE"] = old
+
+    def test_terminal_phase_is_not_active_construction(self):
+        """A contract left status='open' with a TERMINAL phase must not be
+        reported as active (live store had 8 such records)."""
+        from rune.scaffold import TERMINAL_PHASES
+        self.assertIn("abandoned", TERMINAL_PHASES)
+        self.assertIn("closed", TERMINAL_PHASES)
+        self.assertNotIn("building", TERMINAL_PHASES)
+
+
 class BridgeContract(unittest.TestCase):
     def test_preflight_local_task_is_cheap_and_bounded(self):
         cap = preflight("Fix the telemetry summary log level constant in stage_telemetry.py")

tests/test_hook_parity.py

@@ -4,14 +4,17 @@ from __future__ import annotations
 import hashlib
 from pathlib import Path
 REPO = Path(__file__).resolve().parent.parent
-CANONICAL = REPO / "config" / "mimo-hooks" / "event-wait-guard.ts"
+CANONICAL_DIR = REPO / "config" / "mimo-hooks"
+# T22: the managed set is the installer's authority — a hook that reaches the
+# builder must be version-controlled and drift-checked like any other.
+MANAGED = ["event-wait-guard.ts", "builder-bridge.ts"]
 # T21.3: the hook lives at the ACTUAL controller user's home, not the
 # test runner's. Use an env override or skip when unavailable.
 import os as _os
-INSTALLED = Path(_os.environ.get(
-    "MIMOCODE_HOOK_TARGET",
+HOOK_DIR = Path(_os.environ.get(
+    "MIMOCODE_HOOK_DIR",
     str(Path(_os.environ.get("RUNE_HOOK_TARGET_HOME", "/home/nick"))
-        / ".mimocode" / "hooks" / "event-wait-guard.ts")))
+        / ".mimocode" / "hooks")))
 
 
 def _sha(p: Path) -> str:
@@ -19,15 +22,19 @@ def _sha(p: Path) -> str:
 
 
 def test_213c_canonical_exists():
-    assert CANONICAL.is_file()
+    for name in MANAGED:
+        assert (CANONICAL_DIR / name).is_file(), f"canonical missing: {name}"
 
 
 def test_213c_installed_matches_canonical():
-    assert INSTALLED.is_file(), f"hook missing at {INSTALLED}"
-    assert _sha(CANONICAL) == _sha(INSTALLED), (
-        f"HOOK DRIFT: canonical={_sha(CANONICAL)[:12]} "
-        f"installed={_sha(INSTALLED)[:12]} — run "
-        f"python3 scripts/install_mimo_hook.py to sync")
+    for name in MANAGED:
+        canonical = CANONICAL_DIR / name
+        installed = HOOK_DIR / name
+        assert installed.is_file(), f"hook missing at {installed}"
+        assert _sha(canonical) == _sha(installed), (
+            f"HOOK DRIFT: {name} canonical={_sha(canonical)[:12]} "
+            f"installed={_sha(installed)[:12]} — run "
+            f"python3 scripts/install_mimo_hook.py to sync")
 
 
 def test_213c_installer_exists():

tests/test_native_default.py

@@ -97,6 +97,61 @@ def test_15j_native_not_construction_touches_nothing(op):
     assert not projects.exists() or not any(projects.glob("*/current.json"))
 
 
+def test_15j_intent_layer_unavailable_preserves_classifier_path(op, monkeypatch):
+    """If the meaning layer is unavailable the PRE-EXISTING behaviour is
+    preserved exactly (classify + activate) and the reason is recorded —
+    governance is never silently changed in an unproven direction."""
+    import rune.construction_intent as cint
+
+    def _boom(*a, **k):
+        raise RuntimeError("intent layer offline")
+
+    monkeypatch.setattr(cint, "resolve_construction_intent", _boom)
+    op, tmp = op
+    op.send_task("fix the broken retry handling in this project")
+    rec = _last_activation(tmp)
+    assert rec["mode"] == "EXISTING_PROJECT_CHANGE", rec
+    assert rec["activated"] is True
+    assert rec.get("intent_layer") is None
+    doc = json.loads((tmp / ".rune" / "scaffold" / "current.json").read_text())
+    notes = [o for o in doc.get("observations", [])
+             if o.get("kind") == "construction_note"]
+    assert any("intent layer unavailable" in str(n.get("text", "")) for n in notes), notes
+
+
+def test_15j_native_question_with_target_is_not_construction(op):
+    """§17J criterion 1 on the NORMAL path (T22 wiring repair).
+
+    A pure question that happens to carry a target reference
+    ("this module") must NOT record construction state. Before the
+    intent-layer gate this classified EXISTING_PROJECT_CHANGE (a target
+    reference was read as scope evidence) and recorded an activation for
+    a discussion turn. The canonical meaning layer resolves it as
+    DISCUSSION; the classifier is never reached.
+    """
+    op, tmp = op
+    op.send_task("explain how this module works")
+    rec = _last_activation(tmp)
+    assert rec is not None, "every input records its classification outcome"
+    assert rec["mode"] == "NOT_CONSTRUCTION", rec
+    assert rec["activated"] is False
+    assert rec.get("intent_layer") == "DISCUSSION", rec
+    # no project invention
+    projects = (tmp / ".rune" / "scaffold" / "projects")
+    assert not projects.exists() or not any(projects.glob("*/current.json"))
+
+
+def test_15j_native_real_construction_still_activates(op):
+    """The meaning gate must not suppress genuine construction demands —
+    the guard is a question/discussion gate, not an activation blocker."""
+    op, tmp = op
+    op.send_task("fix the broken retry handling in this project")
+    rec = _last_activation(tmp)
+    assert rec["mode"] == "EXISTING_PROJECT_CHANGE", rec
+    assert rec["activated"] is True
+    assert rec.get("intent_layer") is None, rec
+
+
 def test_15j_native_unknown_probes_without_invention(op):
     op, tmp = op
     op.send_task("Build something useful for this")