From 471131cabfc403924043dd4e02f415433780b0b7 Mon Sep 17 00:00:00 2001 From: Shay Date: Sat, 13 Jun 2026 20:22:39 -0700 Subject: [PATCH] =?UTF-8?q?feat(b4):=20engine-side=20leeway=20producer=20?= =?UTF-8?q?=E2=80=94=20populate=20LeewayEvidence=20(Wave=20M=20B4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Clears the long-standing B4 block: the leeway decision was already made on the serving path (chat/runtime.py::_surface_estimate has a real LicenseDecision + the ReachPolicy) and then DISCARDED — never threaded to the result, and the workbench can't import reliability_gate. This wires it through. Producer (observational, never authorizing): - core/cognition/leeway.py: LeewayRecord + build_leeway_record(reach_level, license_decision) — duck-typed on the decision, zero new cross-package coupling. Maps to: no decision -> "unknown" (STRICT, no latitude); denied -> "blocked" (gate consulted, said no); licensed SERVE/PROPOSE widening -> the real class / theta / "[approximate]" disclosure. "verified" is never emitted (RESERVED state). source_digest content-addresses the decision (deterministic, no wall-clock). - core/cognition/result.py: additive `leeway: LeewayRecord | None = None` on CognitiveTurnResult. - core/cognition/pipeline.py: build it at result construction from the data the runtime ALREADY exposes (response.reach_level + runtime.last_turn_accrual().license). chat/runtime.py is UNTOUCHED. - workbench/api.py: _leeway_evidence_from_result maps result.leeway -> LeewayEvidence (pure projection; no reliability_gate import — firewall intact). The journal already persists it; the B4a UI (Replay / Proposals / RightInspector) already renders it — no frontend or schema change needed. Safety (this touches the serving path, so proven, not asserted): - trace_hash is a NAMED-field hash (core/cognition/trace.py); `leeway` is not in it -> byte-identical serving. All provenance/trace tests pass. - response_governance STRICT stays byte-identical (375 governance/serving/ provenance tests green, incl. the live-wiring + estimation-lane + ADR-0206 seam tests). - core eval cognition: 13 cases, 100% intent / groundedness / versor closure. - replay determinism holds (leeway is in CRITICAL_FIELDS; deterministic). Tests: engine (build_leeway_record: strict/blocked/SERVE/PROPOSE, no "verified", deterministic digest) + workbench mapping (field-for-field, honest absence, invalid-enum clamp) + integration (a real turn now carries an honest leeway record, not the null "No evidence recorded"). 151 workbench/leeway Python + 68 frontend (leeway/replay/proposals/schemaDrift) green; schema-snapshot unchanged. Docs: gate cleared (b4-leeway-feasibility-gate.md), residue ledger flipped to implemented, scope brief is b4-leeway-producer-scope-2026-06-13.md. --- core/cognition/leeway.py | 107 +++++++++++++++++++ core/cognition/pipeline.py | 15 +++ core/cognition/result.py | 4 + docs/workbench/b4-leeway-feasibility-gate.md | 11 +- docs/workbench/phase-a-residue-ledger.md | 2 +- tests/test_cognition_leeway.py | 91 ++++++++++++++++ tests/test_workbench_chat_turn.py | 7 ++ tests/test_workbench_leeway_mapping.py | 67 ++++++++++++ workbench/api.py | 32 ++++++ 9 files changed, 334 insertions(+), 2 deletions(-) create mode 100644 core/cognition/leeway.py create mode 100644 tests/test_cognition_leeway.py create mode 100644 tests/test_workbench_leeway_mapping.py diff --git a/core/cognition/leeway.py b/core/cognition/leeway.py new file mode 100644 index 00000000..4dcf5931 --- /dev/null +++ b/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, + ) diff --git a/core/cognition/pipeline.py b/core/cognition/pipeline.py index f9c041d4..eb9078df 100644 --- a/core/cognition/pipeline.py +++ b/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, ) # ------------------------------------------------------------------ diff --git a/core/cognition/result.py b/core/cognition/result.py index c7daae7f..bcf9ddbe 100644 --- a/core/cognition/result.py +++ b/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 diff --git a/docs/workbench/b4-leeway-feasibility-gate.md b/docs/workbench/b4-leeway-feasibility-gate.md index afb6eb75..cfaa4ed8 100644 --- a/docs/workbench/b4-leeway-feasibility-gate.md +++ b/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 diff --git a/docs/workbench/phase-a-residue-ledger.md b/docs/workbench/phase-a-residue-ledger.md index 2757fea5..d150e41f 100644 --- a/docs/workbench/phase-a-residue-ledger.md +++ b/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:` 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). | diff --git a/tests/test_cognition_leeway.py b/tests/test_cognition_leeway.py new file mode 100644 index 00000000..1fcd8ab0 --- /dev/null +++ b/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 diff --git a/tests/test_workbench_chat_turn.py b/tests/test_workbench_chat_turn.py index 13232ef0..9163717f 100644 --- a/tests/test_workbench_chat_turn.py +++ b/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( diff --git a/tests/test_workbench_leeway_mapping.py b/tests/test_workbench_leeway_mapping.py new file mode 100644 index 00000000..454f2777 --- /dev/null +++ b/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" diff --git a/workbench/api.py b/workbench/api.py index 4e67601e..a99f1ff9 100644 --- a/workbench/api.py +++ b/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), )