core-labs/core
core-labs/core
1
0
Fork 0
Code Issues 9 Pull requests Projects Releases Packages Wiki Activity Actions

fix(cognition): declare surface authority resolution (#76)

Browse source 
* fix(cognition): add explicit surface resolution policy

* test(cognition): cover explicit surface resolution policy

* fix(cognition): route pipeline surfaces through resolver

* fix(cognition): address PR #76 review comments

- hoist `_is_useful_surface` import from inside `run()` to module top
- call `_render_walk_surface` / `_render_compose_surface` via the class
  name (both are @staticmethod) for consistency with the existing
  `_fold_*_into_surface` helpers
- drop redundant `realized_surface` truthiness check in
  `resolve_surface` — `realizer_useful` already excludes empty /
  placeholder surfaces via `_is_useful_surface`

Tests: tests/test_surface_resolution.py + tests/test_cognitive_turn_pipeline.py
green (16 passed); cognition suite 120/1s, smoke suite 67/0.
This commit is contained in:
Shay 2026-05-20 19:42:10 -07:00 committed by GitHub
parent 886fc4553f
commit ff1dcb2594
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 238 additions and 86 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

133
core/cognition/pipeline.py
View file

@ -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 <pending> 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 (<pending>, <prior>, "..."). 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:

108
core/cognition/surface_resolution.py Normal file
View file

@ -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),
)

83
tests/test_surface_resolution.py Normal file
View file

@ -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")