Merge pull request #736 from AssetOverflow/feat/b4-leeway-producer

Wave M B4: engine-side leeway producer (clears the block)

core/cognition/leeway.py

@@ -0,0 +1,107 @@
+"""Per-turn leeway record — the engine-side B4 producer.
+
+When a turn is governed (ADR-0206 ``response_governance``), the reliability gate
+either grants latitude to a class (a licensed ``Action.SERVE`` converse-guess,
+surfaced as a DISCLOSED ``[approximate]`` estimate) or withholds it (the STRICT
+default).  That decision is computed on the serving path and was previously
+discarded.  This module turns the decision the runtime ALREADY made into a small
+observational record on the turn result — it never calls the gate, never
+authorizes anything, and never alters the served surface.
+
+The record mirrors the workbench ``LeewayEvidence`` tuple field-for-field, so the
+workbench maps it with a trivial projection and gains no serving-path import.
+The ``license_decision`` is read duck-typed (a
+``core.reliability_gate.LicenseDecision`` when present) so this stays a pure
+cognition-layer leaf with no new cross-package coupling.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True, slots=True)
+class LeewayRecord:
+    """Observational record of the leeway the gate granted (or withheld)."""
+
+    class_name: str
+    license: str  # "SERVE" | "PROPOSE" | "blocked" | "unknown"
+    theta: float | None
+    claim_disclosure: str  # "approximate" | "verified" | "proposal_only" | "none"
+    source_digest: str | None
+    calibration_evidence_ref: str | None
+
+
+def _decision_digest(license_decision: Any) -> str:
+    """Content-address the licensing decision — provenance, deterministic.
+
+    Hashes only the decision's deterministic fields (never a timestamp), so a
+    replayed turn produces the same digest.
+    """
+
+    payload = {
+        "class_name": str(getattr(license_decision, "class_name", "")),
+        "action": str(getattr(getattr(license_decision, "action", None), "name", "")),
+        "checker": str(getattr(license_decision, "checker", "")),
+        "measured": float(getattr(license_decision, "measured", 0.0) or 0.0),
+        "required": float(getattr(license_decision, "required", 0.0) or 0.0),
+        "licensed": bool(getattr(license_decision, "licensed", False)),
+    }
+    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+    return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+def build_leeway_record(
+    *,
+    reach_level: str,
+    license_decision: Any | None,
+) -> LeewayRecord:
+    """Map the governed turn's reach level + license decision to a record.
+
+    - No decision consulted (the STRICT main path): ``license="unknown"``, no
+      latitude.
+    - A decision exists but was denied: ``license="blocked"`` — the gate was
+      consulted and said no.
+    - A licensed SERVE/PROPOSE that widened: the real class, θ, and the
+      ``[approximate]`` disclosure — the "engine earned the right to guess" case.
+
+    ``claim_disclosure`` reflects what was actually disclosed this turn
+    (``approximate`` iff the surface widened); ``"verified"`` is intentionally
+    never emitted (it is a RESERVED epistemic state — claiming it would
+    over-state).
+    """
+
+    disclosure = "approximate" if reach_level == "approximate" else "none"
+
+    if license_decision is None:
+        return LeewayRecord(
+            class_name="none",
+            license="unknown",
+            theta=None,
+            claim_disclosure=disclosure,
+            source_digest=None,
+            calibration_evidence_ref=None,
+        )
+
+    action_name = str(getattr(getattr(license_decision, "action", None), "name", ""))
+    licensed = bool(getattr(license_decision, "licensed", False))
+    if licensed and action_name == "SERVE":
+        license = "SERVE"
+    elif licensed and action_name == "PROPOSE":
+        license = "PROPOSE"
+    else:
+        license = "blocked"
+
+    class_name = str(getattr(license_decision, "class_name", "") or "none")
+    required = getattr(license_decision, "required", None)
+    return LeewayRecord(
+        class_name=class_name,
+        license=license,
+        theta=float(required) if required is not None else None,
+        claim_disclosure=disclosure,
+        source_digest=_decision_digest(license_decision),
+        calibration_evidence_ref=class_name if class_name != "none" else None,
+    )

core/cognition/pipeline.py

@@ -19,6 +19,7 @@ import json
 from collections import OrderedDict
 
 from field.state import FieldState
+from core.cognition.leeway import build_leeway_record
 from core.cognition.result import CognitiveTurnResult
 from core.cognition.surface_resolution import resolve_surface
 from core.cognition.trace import compute_trace_hash, hash_admissibility_trace
@@ -457,6 +458,19 @@ class CognitiveTurnPipeline:
         # ``source_turn_trace``.  See runtime.finalize_turn_trace_hash.
         self.runtime.finalize_turn_trace_hash(trace_hash)
 
+        # B4 producer: capture the leeway the response path already decided —
+        # observational only (reads the runtime's introspection-only accrual and
+        # the response's reach level; never alters the surface, never gates).
+        accrual = (
+            self.runtime.last_turn_accrual()
+            if hasattr(self.runtime, "last_turn_accrual")
+            else None
+        )
+        leeway = build_leeway_record(
+            reach_level=str(getattr(response, "reach_level", "strict") or "strict"),
+            license_decision=getattr(accrual, "license", None),
+        )
+
         return CognitiveTurnResult(
             input_text=text,
             input_tokens=raw_tokens,
@@ -489,6 +503,7 @@ class CognitiveTurnPipeline:
             dropped_compound_clauses=dropped_compound_clauses,
             versor_condition=response.versor_condition,
             trace_hash=trace_hash,
+            leeway=leeway,
         )
 
     # ------------------------------------------------------------------

core/cognition/result.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 
+from core.cognition.leeway import LeewayRecord
 from field.state import FieldState
 from generate.articulation import ArticulationPlan
 from generate.dialogue import DialogueRole
@@ -138,3 +139,6 @@ class CognitiveTurnResult:
     # --- invariant bookkeeping ---
     versor_condition: float = 0.0   # must be < 1e-6
     trace_hash: str = ""            # SHA-256 over deterministic key fields
+
+    # --- response-governance leeway evidence (B4; observational, not in trace_hash) ---
+    leeway: LeewayRecord | None = None

docs/workbench/b4-leeway-feasibility-gate.md

@@ -1,7 +1,16 @@
 # Wave M B3.5-c — B4 Leeway Feasibility Gate
 
 Date: 2026-06-13
-Status: B4a read model created; full B4 annotations not yet admitted
+Status: **GATE CLEARED (2026-06-13)** — the engine-side producer now exists.
+`core/cognition/leeway.py::build_leeway_record` turns the reach-policy +
+`LicenseDecision` the response path already computes into an observational
+`LeewayRecord` on `CognitiveTurnResult`; `workbench/api.py::_leeway_evidence_from_result`
+maps it to `LeewayEvidence` (no `reliability_gate` import — firewall intact);
+the journal persists it and the existing B4a UI (Replay / Proposals /
+RightInspector) renders it. Every turn now carries an honest record (STRICT →
+"no latitude"; a licensed SERVE widening → the real class/θ/`[approximate]`).
+Scope + design: `b4-leeway-producer-scope-2026-06-13.md`. Original gate text
+below, kept for provenance.
 
 ## Finding
 

docs/workbench/phase-a-residue-ledger.md

@@ -12,4 +12,4 @@ Status: active consolidation ledger
 | Calibration evidence subject | implemented | `calibration_class` is addressable via `/calibration?inspect=calibration:<className>` and renders in RightInspector/EvidenceChainRail. | None. |
 | UI/UX guide | implemented | `docs/workbench/UI-UX-GUIDE.md` now records the current 12-route map, evidence grammar, route proofs, and absences. | Keep updated when route registry changes. |
 | Route registry | implemented | `workbench-ui/src/app/routes.ts` is the route source for App, LeftNav, palette, shortcuts, landing prefs, and route tests. | None. |
-| B4 source tuple | blocked | Full B4 producer is absent. B4a nullable `LeewayEvidence` read model now exists, but no backend path populates it yet. | B4 producer wiring from engine-owned approximation/calibration evidence. |
+| B4 source tuple | implemented | Engine producer landed (2026-06-13): `core/cognition/leeway.py` emits an observational `LeewayRecord` on the turn result from the reach-policy + `LicenseDecision` the response path already computes; `workbench/api.py` maps it to `LeewayEvidence` across the read-only firewall. Every turn now carries an honest record; the existing B4a UI renders it. | — (follow-up: a SERVE-licensed fixture to exercise the earned-`APPROXIMATE` path end-to-end). |

tests/test_cognition_leeway.py

@@ -0,0 +1,91 @@
+"""B4 producer — the observational leeway record (engine side).
+
+Non-vacuous: each test fails under a violation of the honest mapping — a STRICT
+turn claiming latitude, a denied license read as granted, ``verified`` ever
+emitted, or a digest that drifts on identical decisions.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from core.cognition.leeway import LeewayRecord, build_leeway_record
+
+
+def _decision(*, action: str, licensed: bool, class_name: str = "addition.converse",
+              required: float = 0.99, measured: float = 0.995) -> SimpleNamespace:
+    return SimpleNamespace(
+        class_name=class_name,
+        action=SimpleNamespace(name=action),
+        checker="reliability",
+        measured=measured,
+        required=required,
+        licensed=licensed,
+    )
+
+
+class TestStrictDefault:
+    def test_no_decision_is_no_latitude(self) -> None:
+        rec = build_leeway_record(reach_level="strict", license_decision=None)
+        assert rec == LeewayRecord(
+            class_name="none",
+            license="unknown",
+            theta=None,
+            claim_disclosure="none",
+            source_digest=None,
+            calibration_evidence_ref=None,
+        )
+
+
+class TestEarnedLeeway:
+    def test_licensed_serve_widening_is_the_real_story(self) -> None:
+        rec = build_leeway_record(
+            reach_level="approximate",
+            license_decision=_decision(action="SERVE", licensed=True),
+        )
+        assert rec.license == "SERVE"
+        assert rec.claim_disclosure == "approximate"
+        assert rec.theta == 0.99
+        assert rec.class_name == "addition.converse"
+        assert rec.calibration_evidence_ref == "addition.converse"
+        assert rec.source_digest is not None and rec.source_digest.startswith("sha256:")
+
+    def test_denied_license_is_blocked_not_granted(self) -> None:
+        # The gate was consulted and said no — never read as latitude.
+        rec = build_leeway_record(
+            reach_level="strict",
+            license_decision=_decision(action="SERVE", licensed=False),
+        )
+        assert rec.license == "blocked"
+        assert rec.claim_disclosure == "none"
+
+    def test_propose_license(self) -> None:
+        rec = build_leeway_record(
+            reach_level="strict",
+            license_decision=_decision(action="PROPOSE", licensed=True),
+        )
+        assert rec.license == "PROPOSE"
+
+
+class TestHonesty:
+    def test_verified_is_never_emitted(self) -> None:
+        # "verified" is a RESERVED epistemic state; the producer must not claim it.
+        for reach in ("strict", "approximate"):
+            for ld in (None, _decision(action="SERVE", licensed=True),
+                       _decision(action="SERVE", licensed=False)):
+                rec = build_leeway_record(reach_level=reach, license_decision=ld)
+                assert rec.claim_disclosure != "verified"
+
+    def test_digest_is_deterministic(self) -> None:
+        a = build_leeway_record(reach_level="approximate",
+                                license_decision=_decision(action="SERVE", licensed=True))
+        b = build_leeway_record(reach_level="approximate",
+                                license_decision=_decision(action="SERVE", licensed=True))
+        assert a.source_digest == b.source_digest
+
+    def test_different_decision_flips_digest(self) -> None:
+        a = build_leeway_record(reach_level="approximate",
+                                license_decision=_decision(action="SERVE", licensed=True, measured=0.995))
+        b = build_leeway_record(reach_level="approximate",
+                                license_decision=_decision(action="SERVE", licensed=True, measured=0.999))
+        assert a.source_digest != b.source_digest

tests/test_workbench_chat_turn.py

@@ -118,6 +118,13 @@ def test_chat_turn_happy_path_real_prompt(api: WorkbenchApi) -> None:
     assert isinstance(data["checkpoint_emitted"], bool)
     assert isinstance(data["turn_cost_ms"], int)
     assert "walk_surface" in data and "articulation_surface" in data
+    # B4 producer: every turn now carries an honest leeway record. A normal
+    # STRICT turn grants no latitude (no longer the null "No evidence recorded").
+    leeway = data["leeway_evidence"]
+    assert leeway is not None
+    assert leeway["license"] in {"unknown", "blocked", "PROPOSE", "SERVE"}
+    assert leeway["claim_disclosure"] in {"none", "approximate", "proposal_only"}
+    assert leeway["claim_disclosure"] != "verified"
 
 
 @pytest.mark.parametrize(

tests/test_workbench_leeway_mapping.py

@@ -0,0 +1,67 @@
+"""B4 workbench mapping — engine LeewayRecord -> LeewayEvidence.
+
+A pure projection across the read-only firewall: no reliability_gate import,
+honest absence preserved, and unexpected enum values clamped to safe defaults.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from core.cognition.leeway import build_leeway_record
+from workbench.api import _leeway_evidence_from_result
+
+
+def test_absent_record_maps_to_none() -> None:
+    # Pre-B4 results carry no leeway -> honest absence (UI shows "not recorded").
+    assert _leeway_evidence_from_result(SimpleNamespace(leeway=None)) is None
+    assert _leeway_evidence_from_result(SimpleNamespace()) is None
+
+
+def test_strict_record_maps_field_for_field() -> None:
+    rec = build_leeway_record(reach_level="strict", license_decision=None)
+    ev = _leeway_evidence_from_result(SimpleNamespace(leeway=rec))
+    assert ev is not None
+    assert ev.license == "unknown"
+    assert ev.claim_disclosure == "none"
+    assert ev.class_name == "none"
+    assert ev.theta is None
+
+
+def test_earned_serve_record_maps() -> None:
+    ld = SimpleNamespace(
+        class_name="addition.converse",
+        action=SimpleNamespace(name="SERVE"),
+        checker="reliability",
+        measured=0.995,
+        required=0.99,
+        licensed=True,
+    )
+    rec = build_leeway_record(reach_level="approximate", license_decision=ld)
+    ev = _leeway_evidence_from_result(SimpleNamespace(leeway=rec))
+    assert ev is not None
+    assert ev.license == "SERVE"
+    assert ev.theta == 0.99
+    assert ev.claim_disclosure == "approximate"
+    assert ev.calibration_evidence_ref == "addition.converse"
+    assert ev.source_digest is not None
+
+
+def test_unexpected_enums_clamp_to_safe_defaults() -> None:
+    # The workbench faithfully maps any *schema-valid* value (the "never emit
+    # verified" guarantee is engine-side); only genuinely invalid enum values
+    # are clamped to the safe default.
+    bad = SimpleNamespace(
+        leeway=SimpleNamespace(
+            class_name="x",
+            license="WIDE_OPEN",  # not a valid license -> unknown
+            theta=0.5,
+            claim_disclosure="wild_guess",  # not a valid disclosure -> none
+            source_digest=None,
+            calibration_evidence_ref=None,
+        )
+    )
+    ev = _leeway_evidence_from_result(bad)
+    assert ev is not None
+    assert ev.license == "unknown"
+    assert ev.claim_disclosure == "none"

workbench/api.py

@@ -34,6 +34,7 @@ from workbench.readers import ArtifactTooLargeError, EvidenceUnavailableError
 from workbench.replay import replay_turn
 from workbench.schemas import (
     ChatTurnResult,
+    LeewayEvidence,
     MathRatifyResult,
     ProposalRef,
     TurnVerdict,
@@ -598,6 +599,36 @@ class WorkbenchApi:
             return ApiResponse(200, ok(result_with_cost))
 
 
+_VALID_LICENSES = frozenset({"PROPOSE", "SERVE", "blocked", "unknown"})
+_VALID_DISCLOSURES = frozenset({"approximate", "verified", "proposal_only", "none"})
+
+
+def _leeway_evidence_from_result(result: object) -> LeewayEvidence | None:
+    """Map the engine's observational ``LeewayRecord`` to ``LeewayEvidence``.
+
+    A pure projection of a plain dataclass off the turn result — no
+    ``reliability_gate`` import, so the read-only firewall holds. Returns ``None``
+    only for results predating the B4 producer (the UI then shows honest
+    absence). Unexpected enum values fall back to the safe ``unknown``/``none``.
+    """
+
+    record = getattr(result, "leeway", None)
+    if record is None:
+        return None
+    license = str(getattr(record, "license", "unknown"))
+    disclosure = str(getattr(record, "claim_disclosure", "none"))
+    return LeewayEvidence(
+        class_name=str(getattr(record, "class_name", "none")),
+        license=license if license in _VALID_LICENSES else "unknown",  # type: ignore[arg-type]
+        theta=getattr(record, "theta", None),
+        claim_disclosure=(  # type: ignore[arg-type]
+            disclosure if disclosure in _VALID_DISCLOSURES else "none"
+        ),
+        source_digest=getattr(record, "source_digest", None),
+        calibration_evidence_ref=getattr(record, "calibration_evidence_ref", None),
+    )
+
+
 def _with_turn_cost_and_id(
     result: ChatTurnResult,
     turn_cost_ms: int,
@@ -741,4 +772,5 @@ def _run_chat_turn(prompt: str, runtime: ChatRuntime | None = None) -> ChatTurnR
         checkpoint_emitted=checkpoint_emitted,
         pipeline_record=cognitive_pipeline_record_from_result(result),
         field_evidence=field_evidence_from_result(result),
+        leeway_evidence=_leeway_evidence_from_result(result),
     )