fix(decode): fail-closed false-singleton + consume min_terminals bound

docs/design/hybrid-unmask-hx5-20260804.md

@@ -104,11 +104,15 @@ over the E1 corpus (accepts-set equality at every terminal prefix: 8
 programs, 56 steps, 14 states). `require_certified_static_lalr()` is
 fail-closed and cached per artifact digest.
 
-**Consumed by nothing**: `completion_kernel.py` and `engine.py` are
-untouched; no default decode path changed. The artifact was regenerated
-(53824 → 56224 bytes) so its sha256 changed — checkpoints declaring the old
-`completion_artifact` identity must be re-stamped (fail-closed contract, not
-a regression).
+**Now consumed (negative-direction only, 2026-08-05):**
+`completion_kernel.terminal_witness` may call
+`require_certified_static_lalr` + `StaticLalrAdapter.min_terminals` and return
+`UNSUPPORTED` when remaining room is strictly below the certified lower
+bound. Counter: `min_terminals_prunes`. Lark remains the live parser
+authority; the bound never forces a commit or widens legality. The artifact
+was regenerated (53824 → 56224 bytes) so its sha256 changed — checkpoints
+declaring the old `completion_artifact` identity must be re-stamped
+(fail-closed contract, not a regression).
 
 ### (iv) W3 duration-aware sharding — PASS, with a caveat that matters
 
@@ -158,10 +162,13 @@ adapter and bound are certified but unconsumed. Any run using
 ## Follow-ups
 
 - Repair-heavy fixture to actually test HX5 (attempts > 1).
-- Kernel consumption of the certified adapter + `state_min_terminals` under
-  the campaign's E2 cold-speed gate.
-- Remediate the two >3-minute test files; consider per-node weighting if the
-  goal shifts from heavy-file isolation to minimizing max shard wall.
+- ~~Kernel consumption of the certified adapter + `state_min_terminals`~~ —
+  done as negative-direction prune in `terminal_witness` (still open: E2
+  cold-speed measurement under a preregistered campaign).
+- Remediate the two >3-minute test files further (topology live apply paths
+  still `@pytest.mark.slow`; dsh5 default CI now uses synthetic probes with
+  live preflight kept slow). Duration table carries `measured_at` +
+  `integrity.irreducible_files` so partial/stale weights fail validation.
 - Hybrid quality screening (preregistered) before any default change; the
   cluster-mode (V7) comparison remains open — hybrid is *structural* region
   typing, cluster is *attention-derived*, and they are mutually exclusive by

src/slm_training/dsl/grammar/fastpath/completion_artifact.py

@@ -466,11 +466,14 @@ def _state_min_terminals(
 class StaticLalrAdapter:
     """Executable shift/reduce over the certified control arrays.
 
-    This is a *checked* adapter, never an authority: it is import-only until a
+    This is a *checked* adapter, never an authority: it is usable only after a
     lockstep certification against the live Lark ``InteractiveParser`` passes
-    (see ``static_control_domain.require_certified_static_lalr``).  Nothing in
-    the decode path consumes it.  ``$END`` follows Lark's end handling: the
-    reduction chain runs until the end state is exposed.
+    (see ``static_control_domain.require_certified_static_lalr``).  The decode
+    path may consume ``min_terminals`` as a **negative-direction** prune inside
+    ``completion_kernel.terminal_witness`` (reject when room is strictly below
+    the certified lower bound); Lark remains the live parser authority.
+    ``$END`` follows Lark's end handling: the reduction chain runs until the
+    end state is exposed.
     """
 
     symbols: tuple[str, ...]

src/slm_training/dsl/grammar/fastpath/completion_kernel.py

@@ -150,6 +150,12 @@ def __init__(
         # stores draft results only — semantic filtering and path labeling
         # stay live per state; authority is unchanged.
         self._branch_memo: dict[tuple, tuple[str, tuple[int, ...] | None]] = {}
+        # Certified LALR acceptance-length bound (negative-direction only).
+        # Lazy: first successful certify is reused for the session; a failed
+        # or unavailable certification disables bound pruning for the session
+        # (Lark remains the live authority either way).
+        self._lalr_adapter: Any | None = None
+        self._lalr_bound_disabled: bool = False
         self._counters: dict[str, int] = {
             "session_starts": 1,
             "state_intern_hits": 0,
@@ -173,6 +179,7 @@ def __init__(
             "scope_reference_scans_avoided": 0,
             "branch_memo_hits": 0,
             "branch_memo_misses": 0,
+            "min_terminals_prunes": 0,
         }
 
     # --- introspection ----------------------------------------------------
@@ -475,6 +482,45 @@ def outgoing(self, state_id: int, *, state: Any | None = None) -> CompletionFore
 
     # --- bounded terminal-witness DP -----------------------------------------
 
+    def _min_terminals_bound(self, state_id: int) -> int | None:
+        """Certified lower bound on remaining terminals, or None if unknown.
+
+        Uses ``StaticLalrAdapter.min_terminals`` after lockstep certification.
+        Negative-direction only: a bound of ``b`` means at least ``b`` more
+        terminals are required; never used to force a commit or widen legality.
+        """
+        if self._lalr_bound_disabled:
+            return None
+        try:
+            record = self._states[int(state_id)]
+        except (IndexError, TypeError, ValueError):
+            return None
+        engine = record.engine
+        ip = getattr(engine, "_ip", None)
+        parser_state = getattr(ip, "parser_state", None) if ip is not None else None
+        stack = getattr(parser_state, "state_stack", None)
+        if not stack:
+            return None
+        if self._lalr_adapter is None:
+            try:
+                from slm_training.dsl.grammar.fastpath.static_control_domain import (
+                    require_certified_static_lalr,
+                    static_lalr_adapter,
+                )
+
+                require_certified_static_lalr(self._tokenizer, engine=engine)
+                self._lalr_adapter = static_lalr_adapter(
+                    self._tokenizer, engine=engine
+                )
+            except Exception:  # noqa: BLE001 - bound is optional; fail open to search
+                self._lalr_bound_disabled = True
+                return None
+        try:
+            bound = int(self._lalr_adapter.min_terminals(int(stack[-1])))
+        except Exception:  # noqa: BLE001
+            return None
+        return bound if bound >= 0 else None
+
     def terminal_witness(self, state_id: int, room: int) -> WitnessResult:
         """Return V1's first-in-path-order witness with typed uncertainty.
 
@@ -483,6 +529,10 @@ def terminal_witness(self, state_id: int, room: int) -> WitnessResult:
         behavior-visible expansion order. Results affected by partial
         authority or budget depletion are UNKNOWN and never enter the
         session-wide memo.
+
+        When a certified ``state_min_terminals`` lower bound proves the
+        remaining room is insufficient, the query returns UNSUPPORTED without
+        expanding further (sound negative prune only).
         """
         sid = int(state_id)
         rm = max(0, int(room))
@@ -507,35 +557,40 @@ def _eval(current: int, remaining: int) -> WitnessResult:
                 return cached
             if remaining <= 0:
                 result = WitnessResult(WitnessStatus.UNSUPPORTED)
-            elif nodes_left <= 0:
-                result = WitnessResult(WitnessStatus.UNKNOWN)
             else:
-                nodes_left -= 1
-                self._counters["reachability_cache_misses"] += 1
-                self._counters["witness_states_expanded"] += 1
-                forest = self.outgoing(current)
-                saw_unknown = forest.coverage != "complete"
-                result = WitnessResult(WitnessStatus.UNSUPPORTED)
-                for path in forest.paths:
-                    tokens = tuple(int(token_id) for token_id in path.token_ids)
-                    if not tokens or len(tokens) > remaining:
-                        continue
-                    if path.kind == "eos":
-                        result = WitnessResult(WitnessStatus.SUPPORTED, tokens)
-                        break
-                    child = self.advance_path(current, tokens)
-                    if child is None:
-                        continue
-                    suffix = _eval(child, remaining - len(tokens))
-                    if suffix.status is WitnessStatus.SUPPORTED:
-                        result = WitnessResult(
-                            WitnessStatus.SUPPORTED, tokens + suffix.witness
-                        )
-                        break
-                    saw_unknown |= suffix.status is WitnessStatus.UNKNOWN
+                bound = self._min_terminals_bound(current)
+                if bound is not None and bound > remaining:
+                    self._counters["min_terminals_prunes"] += 1
+                    result = WitnessResult(WitnessStatus.UNSUPPORTED)
+                elif nodes_left <= 0:
+                    result = WitnessResult(WitnessStatus.UNKNOWN)
                 else:
-                    if saw_unknown:
-                        result = WitnessResult(WitnessStatus.UNKNOWN)
+                    nodes_left -= 1
+                    self._counters["reachability_cache_misses"] += 1
+                    self._counters["witness_states_expanded"] += 1
+                    forest = self.outgoing(current)
+                    saw_unknown = forest.coverage != "complete"
+                    result = WitnessResult(WitnessStatus.UNSUPPORTED)
+                    for path in forest.paths:
+                        tokens = tuple(int(token_id) for token_id in path.token_ids)
+                        if not tokens or len(tokens) > remaining:
+                            continue
+                        if path.kind == "eos":
+                            result = WitnessResult(WitnessStatus.SUPPORTED, tokens)
+                            break
+                        child = self.advance_path(current, tokens)
+                        if child is None:
+                            continue
+                        suffix = _eval(child, remaining - len(tokens))
+                        if suffix.status is WitnessStatus.SUPPORTED:
+                            result = WitnessResult(
+                                WitnessStatus.SUPPORTED, tokens + suffix.witness
+                            )
+                            break
+                        saw_unknown |= suffix.status is WitnessStatus.UNKNOWN
+                    else:
+                        if saw_unknown:
+                            result = WitnessResult(WitnessStatus.UNKNOWN)
             query_cache[key] = result
             query_cache.move_to_end(key)
             if len(query_cache) > 64:

src/slm_training/dsl/grammar/fastpath/static_control_domain.py

@@ -4,11 +4,12 @@
 accepts the request-independent projection. Scope and semantic (Γ) constraints
 remain request-local leaf filters that may only **tighten** the static domain.
 Lark remains the live parser authority. ``StaticLalrAdapter`` is an executable
-projection of the certified control arrays, but it is import-only: it becomes
-usable evidence only after ``require_certified_static_lalr`` proves lockstep
-accepts-set equality against the live ``InteractiveParser`` over the canonical
-corpus, and no decode path consumes it. Certificate payload is never rebranded
-as a DPDA executor.
+projection of the certified control arrays: it becomes usable evidence only
+after ``require_certified_static_lalr`` proves lockstep accepts-set equality
+against the live ``InteractiveParser`` over the canonical corpus.
+``completion_kernel`` may then use ``min_terminals`` for negative-direction
+room pruning only — never to force commits or widen legality. Certificate
+payload is never rebranded as a DPDA executor.
 
 See ``docs/design/adr-constrained-diffusion-topology-split.md``.
 """

src/slm_training/dsl/pack.py

@@ -528,9 +528,20 @@ def _finish(domain: Any) -> Any:
     if query_witness_tally[0] and query_witness_tally[1] == 1:
         # Exactly one proven candidate survived while an unproven one was
         # dropped: downstream singleton detection cannot distinguish this from
-        # a genuine deterministic force, so the model is bypassed on the
-        # strength of a budget limit rather than a proof.
+        # a genuine deterministic force, so the model would be bypassed on the
+        # strength of a budget limit rather than a proof (I2 hazard).
+        # Fail closed: refuse a complete domain. Incomplete maps to forest
+        # coverage "none", so exact_forced_token_id will not commit the
+        # budget-manufactured singleton.
         _note_witness(false_singleton=True)
+        return _finish(
+            CompletionDomainV1(
+                status="incomplete",
+                scope_fingerprint=fingerprint,
+                terminals=initial.terminals,
+                reason="witness_false_singleton_risk",
+            )
+        )
     return _finish(
         CompletionDomainV1(
             status="complete",
@@ -775,6 +786,9 @@ def _tail(current: tuple[int, ...], room: int) -> tuple[int, ...] | None:
 
     candidates: list[Any] = []
     unwitnessed = False
+    # [unknown_dropped, proven_kept] — same false-singleton shape as the
+    # packed session path.
+    query_witness_tally = [0, 0]
     for path in initial.paths:
         tokens = tuple(int(token_id) for token_id in path.token_ids)
         if not tokens or len(tokens) > budget:
@@ -789,9 +803,11 @@ def _tail(current: tuple[int, ...], room: int) -> tuple[int, ...] | None:
             # Bounded tail search found nothing within budget: same
             # UNKNOWN-flavoured false-reject risk as the session path.
             _note_witness(materialized=True, pruned_unknown=True)
+            query_witness_tally[0] += 1
             unwitnessed = True
             continue
         _note_witness(materialized=True, kept=True)
+        query_witness_tally[1] += 1
         candidates.append(
             CompletionDomainCandidateV1(
                 token_ids=tokens,
@@ -806,6 +822,14 @@ def _tail(current: tuple[int, ...], room: int) -> tuple[int, ...] | None:
             terminals=initial.terminals,
             reason="terminal_witness_unavailable",
         )
+    if query_witness_tally[0] and query_witness_tally[1] == 1:
+        _note_witness(false_singleton=True)
+        return CompletionDomainV1(
+            status="incomplete",
+            scope_fingerprint=fingerprint,
+            terminals=initial.terminals,
+            reason="witness_false_singleton_risk",
+        )
     return CompletionDomainV1(
         status="complete",
         candidates=tuple(candidates),

src/slm_training/resources/versions.json

@@ -1168,7 +1168,7 @@
       ]
     },
     "dsl.completion_kernel": {
-      "version": "v20",
+      "version": "v21",
       "kind": "harness",
       "paths": [
         "src/slm_training/dsl/grammar/fastpath/engine.py",
@@ -1193,9 +1193,15 @@
         "src/slm_training/dsl/grammar/fastpath/residual_support.py",
         "tests/test_dsl/test_static_control_domain.py",
         "tests/test_dsl/test_residual_support.py",
-        "docs/design/adr-constrained-diffusion-topology-split.md"
+        "docs/design/adr-constrained-diffusion-topology-split.md",
+        "docs/design/hybrid-unmask-hx5-20260804.md"
       ],
       "history": [
+        {
+          "version": "v21",
+          "date": "2026-08-05",
+          "note": "Consume certified StaticLalrAdapter.min_terminals in completion_kernel.terminal_witness as a negative-direction prune (UNSUPPORTED when room < bound). Counter min_terminals_prunes. Adapter is no longer import-only; Lark remains live authority."
+        },
         {
           "version": "v20",
           "date": "2026-08-04",
@@ -1354,14 +1360,21 @@
       ]
     },
     "dsl.operators.registry": {
-      "version": "v12",
+      "version": "v13",
       "kind": "harness",
       "paths": [
         "src/slm_training/dsl/operators/registry.py",
         "src/slm_training/dsl/pack.py",
-        "tests/test_dsl/test_operator_registry.py"
+        "tests/test_dsl/test_operator_registry.py",
+        "docs/design/false-singleton-and-shard-makespan-20260804.md",
+        "tests/test_models/test_decode_stats.py"
       ],
       "history": [
+        {
+          "version": "v13",
+          "date": "2026-08-05",
+          "note": "Fail closed on witness false-singleton risk: when UNKNOWN drops leave exactly one proven candidate, completion domain status becomes incomplete with reason witness_false_singleton_risk so I2 exact_forced cannot commit a budget-manufactured singleton. Counter still fires."
+        },
         {
           "version": "v12",
           "date": "2026-08-04",

tests/test_dsl/test_static_lalr_adapter.py

@@ -30,6 +30,8 @@
 _OWNING_MODULES = {
     "src/slm_training/dsl/grammar/fastpath/completion_artifact.py",
     "src/slm_training/dsl/grammar/fastpath/static_control_domain.py",
+    # Negative-direction room prune via certified min_terminals.
+    "src/slm_training/dsl/grammar/fastpath/completion_kernel.py",
 }
 
 
@@ -215,15 +217,16 @@ def _grep_python_sources(pattern: str) -> set[str]:
     return {line for line in found.stdout.splitlines() if line}
 
 
-def test_adapter_is_import_only_with_no_production_consumer() -> None:
-    """Nothing under src/ or scripts/ wires the adapter into a decode path."""
+def test_adapter_consumers_are_allowlisted() -> None:
+    """Only the certified kernel may consume the adapter in production paths."""
 
     adapter_hits = _grep_python_sources(
         "StaticLalrAdapter|require_certified_static_lalr|static_lalr_adapter"
     )
     assert adapter_hits == _OWNING_MODULES, sorted(adapter_hits)
 
-    # The acceptance-length bound is serialized and reported, never consumed.
+    # Bound may be serialized, reported, or used as a negative prune in the
+    # completion kernel — nowhere else under src/ or scripts/.
     bound_hits = _grep_python_sources("state_min_terminals|min_terminals")
     assert bound_hits <= _OWNING_MODULES | {"scripts/build_completion_artifact.py"}, (
         sorted(bound_hits)