diff --git a/core/cognition/pipeline.py b/core/cognition/pipeline.py index 1bc4cc91..d76dcdcc 100644 --- a/core/cognition/pipeline.py +++ b/core/cognition/pipeline.py @@ -17,8 +17,10 @@ from __future__ import annotations from field.state import FieldState 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 from generate.intent import classify_intent +from generate.intent_bridge import _is_useful_surface from generate.intent_ratifier import ( RatificationOutcome, ratify_intent, @@ -126,88 +128,39 @@ class CognitiveTurnPipeline: # ChatResponse is the stable contract surface. response = self.runtime.chat(text, max_tokens=max_tokens) - # Override surfaces when semantic realizer produced a result. - # The ChatResponse contract fields are preserved; we select - # the better articulation surface from the semantic path. - # - # Exception: when the unknown-domain gate fired, ChatRuntime - # returns either the universal "insufficient grounding" stub - # (ADR-0035) or a pack-grounded surface (ADR-0048) — in both - # cases the realizer's fallback would be template-noise that - # contradicts the gate's honest "no_grounding" signal, so we - # keep ChatRuntime's stub-path surface user-visible. walk_surface - # is unaffected either way. Addresses calibration gaps.md - # Finding 2. gate_fired = ( response.vault_hits == 0 and getattr(response, "grounding_source", "vault") != "vault" ) - # ADR-0077 (R6) — read register_canonical_surface for trace_hash - # so substantive register transforms (terse compress_gloss / - # drop_provenance / drop_articles, convivial - # append_semantic_domain_clause) cannot leak into the truth - # path either. Falls back through pre_decoration_surface (R4, - # ADR-0071) and finally response.surface — historical trace - # hashes stay byte-identical for any pre-R6 caller that does - # not set the new field, because both pre_decoration and the - # canonical degrade to empty in that case. canonical = getattr(response, "register_canonical_surface", "") or "" pre_decoration = getattr(response, "pre_decoration_surface", "") or "" - if canonical: - surface = canonical - elif pre_decoration: - surface = pre_decoration - else: - surface = response.surface - articulation_surface = response.articulation_surface - # Pipeline override usefulness gate (2026-05-19 design review, - # Finding P0 #1). The previous gate only required - # ``realized_plan.surface`` to be non-empty, so a realizer - # output of "Truth is defined as ..." (with rendered - # as ...) would silently override a perfectly-good runtime - # pack-grounded surface and the telemetry would record yet a - # third surface. The warmed_session_consistency lane catches - # this exactly. - # - # Use the same usefulness predicate the bridge uses to gate - # ``articulate_with_intent`` output (generate/intent_bridge.py), - # which rejects empty surfaces and surfaces containing the - # placeholder sentinels (, , "..."). An - # ungrounded realizer surface cannot honestly override a - # grounded runtime surface — when the realizer cannot produce - # a useful surface, we keep the runtime answer the user sees. - from generate.intent_bridge import _is_useful_surface - if ( - realized_plan.surface - and not gate_fired - and _is_useful_surface(realized_plan.surface) - ): - surface = realized_plan.surface - articulation_surface = realized_plan.surface - # 7b. INFER — invoke typed deterministic operators (ADR-0018) when the - # intent is a transitive-query or definition shape and the teaching - # store carries a chain rooted at the subject. The operator's result - # is folded into the surface so chain endpoints become visible. walk_result: WalkResult | None = self._maybe_transitive_walk(intent) + walk_surface = "" if walk_result is not None and len(walk_result.path) > 1: - surface, articulation_surface = self._fold_walk_into_surface( - walk_result, surface, articulation_surface, - ) + walk_surface = CognitiveTurnPipeline._render_walk_surface(walk_result) - # 7c. INFER (frame transfer) — for "What does X R in Y?" probes, - # compose_relations reports the tails of R(X, ?) and R(Y, ?) so - # the realizer surface names both endpoints. Fires only on the - # FRAME_TRANSFER intent shape so the generic transitive-query - # surface is unaffected. compose_result: FrameComposeResult | None = self._maybe_compose_relations(intent) + compose_surface = "" if compose_result is not None and ( compose_result.subject_tail is not None or compose_result.frame_tail is not None ): - surface, articulation_surface = self._fold_compose_into_surface( - compose_result, surface, articulation_surface, - ) + compose_surface = CognitiveTurnPipeline._render_compose_surface(compose_result) + + resolved = resolve_surface( + canonical_surface=canonical, + pre_decoration_surface=pre_decoration, + response_surface=response.surface, + response_articulation_surface=response.articulation_surface, + realized_surface=realized_plan.surface, + realizer_useful=_is_useful_surface(realized_plan.surface), + gate_fired=gate_fired, + walk_surface=walk_surface, + compose_surface=compose_surface, + ) + surface = resolved.surface + articulation_surface = resolved.articulation_surface # Track last node id for correction-intent chaining if graph.nodes: @@ -507,6 +460,20 @@ class CognitiveTurnPipeline: relation=intent.relation, ) + @staticmethod + def _render_compose_surface(compose: FrameComposeResult) -> str: + """Render a frame-transfer composition suffix without selecting authority.""" + parts: list[str] = [] + if compose.subject_tail is not None: + parts.append( + f"{compose.head} {compose.relation.replace('_', ' ')} {compose.subject_tail}" + ) + if compose.frame_tail is not None: + parts.append( + f"in {compose.frame} {compose.relation.replace('_', ' ')} {compose.frame_tail}" + ) + return "; ".join(parts) + @staticmethod def _fold_compose_into_surface( compose: FrameComposeResult, @@ -520,18 +487,9 @@ class CognitiveTurnPipeline: as the expected answer. Deterministic; identical inputs yield identical output. """ - parts: list[str] = [] - if compose.subject_tail is not None: - parts.append( - f"{compose.head} {compose.relation.replace('_', ' ')} {compose.subject_tail}" - ) - if compose.frame_tail is not None: - parts.append( - f"in {compose.frame} {compose.relation.replace('_', ' ')} {compose.frame_tail}" - ) - if not parts: + compose_surface = CognitiveTurnPipeline._render_compose_surface(compose) + if not compose_surface: return surface, articulation_surface - compose_surface = "; ".join(parts) new_surface = ( f"{surface} — {compose_surface}" if surface else compose_surface ) @@ -558,6 +516,16 @@ class CognitiveTurnPipeline: import json return json.dumps(compose.as_dict(), sort_keys=True, ensure_ascii=False) + @staticmethod + def _render_walk_surface(walk: WalkResult) -> str: + """Render a chain-aware walk suffix without selecting authority.""" + chain = " ".join(walk.path) + endpoint = walk.path[-1] + return ( + f"{walk.head} {walk.relation.replace('_', ' ')} {endpoint} " + f"(via {chain})" + ) + @staticmethod def _fold_walk_into_surface( walk: WalkResult, @@ -570,14 +538,7 @@ class CognitiveTurnPipeline: identical output. The chain endpoint is the load-bearing token for the inference-closure / multi-step-reasoning eval lanes. """ - chain = " ".join(walk.path) - endpoint = walk.path[-1] - chain_surface = ( - f"{walk.head} {walk.relation.replace('_', ' ')} {endpoint} " - f"(via {chain})" - ) - # Preserve the prior surface as a prefix for context, when it exists - # and is non-empty; otherwise the chain surface stands alone. + chain_surface = CognitiveTurnPipeline._render_walk_surface(walk) if surface: new_surface = f"{surface} — {chain_surface}" else: diff --git a/core/cognition/surface_resolution.py b/core/cognition/surface_resolution.py new file mode 100644 index 00000000..ea24b185 --- /dev/null +++ b/core/cognition/surface_resolution.py @@ -0,0 +1,108 @@ +"""Explicit user-facing surface resolution for cognitive turns. + +The pipeline produces several candidate surfaces in one turn: + +* runtime/canonical surface from ChatRuntime +* semantic realizer surface from the proposition graph +* deterministic operator folds from walk / compose inference + +Historically these mutated one string in evaluation order. This module +centralizes the policy so fold behavior is declared and unit-testable. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True, slots=True) +class SurfaceResolution: + """Resolved user-facing and articulation surfaces. + + ``authority`` records the prefix authority before deterministic folds + are appended. ``fold_sources`` records which inference suffixes were + appended, in deterministic order. + """ + + surface: str + articulation_surface: str + authority: str + fold_sources: tuple[str, ...] = () + + +def _base_runtime_surface( + *, + canonical_surface: str, + pre_decoration_surface: str, + response_surface: str, + response_articulation_surface: str, +) -> tuple[str, str, str]: + """Select the runtime-owned base surface by declared precedence.""" + + if canonical_surface: + return canonical_surface, response_articulation_surface, "runtime_canonical" + if pre_decoration_surface: + return pre_decoration_surface, response_articulation_surface, "runtime_pre_decoration" + return response_surface, response_articulation_surface, "runtime" + + +def resolve_surface( + *, + canonical_surface: str = "", + pre_decoration_surface: str = "", + response_surface: str = "", + response_articulation_surface: str = "", + realized_surface: str = "", + realizer_useful: bool = False, + gate_fired: bool = False, + walk_surface: str = "", + compose_surface: str = "", +) -> SurfaceResolution: + """Resolve the final turn surface under one explicit policy. + + Policy: + 1. Runtime/canonical/pre-decoration selects the base authority. + 2. A useful realizer surface may replace the prefix only when the + unknown-domain gate did not fire. + 3. Walk and compose suffixes are deterministic inference folds. They + append after prefix authority is selected and are never allowed to + re-run or reinterpret the prefix decision. + """ + + surface, articulation_surface, authority = _base_runtime_surface( + canonical_surface=canonical_surface or "", + pre_decoration_surface=pre_decoration_surface or "", + response_surface=response_surface or "", + response_articulation_surface=response_articulation_surface or "", + ) + + if realizer_useful and not gate_fired: + surface = realized_surface + articulation_surface = realized_surface + authority = "realizer" + + fold_sources: list[str] = [] + if walk_surface: + surface = f"{surface} — {walk_surface}" if surface else walk_surface + articulation_surface = ( + f"{articulation_surface} — {walk_surface}" + if articulation_surface + else walk_surface + ) + fold_sources.append("walk") + + if compose_surface: + surface = f"{surface} — {compose_surface}" if surface else compose_surface + articulation_surface = ( + f"{articulation_surface} — {compose_surface}" + if articulation_surface + else compose_surface + ) + fold_sources.append("compose") + + return SurfaceResolution( + surface=surface, + articulation_surface=articulation_surface, + authority=authority, + fold_sources=tuple(fold_sources), + ) diff --git a/tests/test_surface_resolution.py b/tests/test_surface_resolution.py new file mode 100644 index 00000000..fda866cf --- /dev/null +++ b/tests/test_surface_resolution.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +from core.cognition.surface_resolution import resolve_surface + + +def test_runtime_canonical_surface_has_base_precedence() -> None: + resolved = resolve_surface( + canonical_surface="canonical", + pre_decoration_surface="pre-decoration", + response_surface="runtime", + response_articulation_surface="articulation", + ) + + assert resolved.surface == "canonical" + assert resolved.articulation_surface == "articulation" + assert resolved.authority == "runtime_canonical" + assert resolved.fold_sources == () + + +def test_useful_realizer_replaces_prefix_when_gate_did_not_fire() -> None: + resolved = resolve_surface( + response_surface="runtime", + response_articulation_surface="runtime articulation", + realized_surface="realizer", + realizer_useful=True, + gate_fired=False, + ) + + assert resolved.surface == "realizer" + assert resolved.articulation_surface == "realizer" + assert resolved.authority == "realizer" + + +def test_gate_fired_keeps_runtime_surface_even_when_realizer_is_useful() -> None: + resolved = resolve_surface( + response_surface="runtime refusal", + response_articulation_surface="runtime refusal articulation", + realized_surface="realizer noise", + realizer_useful=True, + gate_fired=True, + ) + + assert resolved.surface == "runtime refusal" + assert resolved.articulation_surface == "runtime refusal articulation" + assert resolved.authority == "runtime" + + +def test_useless_realizer_keeps_runtime_surface() -> None: + resolved = resolve_surface( + response_surface="runtime", + response_articulation_surface="runtime articulation", + realized_surface="Truth is defined as ...", + realizer_useful=False, + ) + + assert resolved.surface == "runtime" + assert resolved.articulation_surface == "runtime articulation" + assert resolved.authority == "runtime" + + +def test_walk_and_compose_fold_after_selected_authority() -> None: + resolved = resolve_surface( + response_surface="runtime", + response_articulation_surface="runtime articulation", + realized_surface="realizer", + realizer_useful=True, + walk_surface="walk chain", + compose_surface="compose transfer", + ) + + assert resolved.surface == "realizer — walk chain — compose transfer" + assert resolved.articulation_surface == "realizer — walk chain — compose transfer" + assert resolved.authority == "realizer" + assert resolved.fold_sources == ("walk", "compose") + + +def test_folds_stand_alone_when_base_surface_is_empty() -> None: + resolved = resolve_surface(walk_surface="walk chain", compose_surface="compose transfer") + + assert resolved.surface == "walk chain — compose transfer" + assert resolved.articulation_surface == "walk chain — compose transfer" + assert resolved.authority == "runtime" + assert resolved.fold_sources == ("walk", "compose")