core-labs/core
core-labs/core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity
core/chat/runtime.py
Shay 1a95acb692 feat(ask): strengthen Q1-D artifact validation, add rejection tests, bypass justification 
- evaluate_served_ask now validates the full Q1-D DeliveredQuestion contract:
  status == 'question_only', requires_review is True, served is False,
  answer_binding is None, question.text non-empty, question.slot_name present,
  question_path != proposal_path.  Each clause is individually testable via
  _validate_artifact() which is a pure predicate with no side effects.

- Added _MISSING sentinel to distinguish absent keys from explicit None/False
  values in 'served' and 'answer_binding' checks.

- Added load-bearing comment in _stub_response explaining why served ASK
  bypasses register decoration and realizer guard: Q1-C/Q1-D is the
  authoritative grounded renderer; mutating pre-rendered intake prose would
  be semantically incorrect.

- Expanded test_ask_serving_integration.py with:
  - Rejection tests: status='proposal_only', served=True, requires_review=False,
    non-null answer_binding, missing/blank question.text, missing/blank slot_name.
  - Exact-text bypass safety test proving stub surface == artifact text exactly
    with no register/decorator mutation and realizer_guard_status='ok'.
  - Disposition telemetry tests: default main-path stability, proposal terminal
    not reclassified, ASK branch emits 'ask', refused terminal stays 'refuse',
    ChatResponse.disposition defaults '' for backward compat.

All 440 tests pass.
2026-06-09 12:03:15 -07:00

2975 lines
141 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from __future__ import annotations
from dataclasses import dataclass, replace
import hashlib
import json
import re
import warnings
from collections.abc import Sequence
from typing import Any, List
import numpy as np
from algebra.versor import versor_condition
from chat.pack_grounding import (
pack_grounded_surface,
pack_grounded_comparison_surface,
pack_grounded_correction_surface,
pack_grounded_procedure_surface,
pack_grounded_relation_confirmation_surface,
pack_grounded_unknown_surface,
gloss_aware_cause_surface,
PACK_ID as _COGNITION_PACK_ID,
)
from chat.teaching_grounding import (
teaching_grounded_surface,
teaching_grounded_surface_composed,
teaching_grounded_surface_transitive,
TEACHING_CORPUS_ID as _TEACHING_CORPUS_ID,
)
from chat.refusal import (
build_hedge_prefix,
build_refusal_surface,
inject_hedge,
should_inject_hedge,
)
from core.epistemic_state import (
clearance_from_verdicts,
epistemic_state_for_grounding_source,
normative_detail_from_verdicts,
)
from core.proposal_review.summary import ProposalReviewIdleSummary, idle_summary
from core.response_governance import govern_response, shape_surface
from chat.telemetry import (
TurnEventSink,
format_correction_event_jsonl,
format_reboot_event_jsonl,
format_turn_event_jsonl,
)
from chat.verdicts import TurnVerdicts
from chat.dispatch_trace import DispatchAttempt, DispatchTrace
from teaching.discovery import (
DiscoveryCandidate,
extract_discovery_candidates,
format_candidate_jsonl,
)
from teaching.discovery_sink import DiscoveryCandidateSink
from engine_state import EngineStateStore, get_git_revision
from core.engine_identity import engine_identity_for_config
from recognition.anti_unifier import derive_recognizer
from recognition.outcome import FeatureBundle
from recognition.registry import RecognizerRegistry
from core.config import DEFAULT_CONFIG, DEFAULT_IDENTITY_PACK, RuntimeConfig
from core.physics.drive import DriveGradientMap, GradientField
from core.physics.energy import EnergyClass, EnergyProfile
from core.physics.exertion import CycleCost, ExertionMeter
from core.physics.learning import VaultPromotionPolicy
from core.physics.identity import (
CharacterProfile,
IdentityCheck,
IdentityScore,
TurnEvent,
)
from packs.ethics.check import EthicsCheck, EthicsContext
from packs.ethics.loader import (
DEFAULT_ETHICS_PACK as _DEFAULT_ETHICS_PACK,
EthicsPackError,
load_ethics_pack,
)
from packs.identity.loader import load_identity_manifold
from chat.register_substantive import apply_substantive_register
from chat.register_variation import decorate_surface
from chat.atom_equivalence import atoms_for_graph_nodes, compare_atom_sets
from generate.realizer_guard import (
DISCLOSURE_SURFACE as _GUARD_DISCLOSURE_SURFACE,
check_surface as _check_realizer_surface,
)
from packs.anchor_lens.loader import AnchorLens, load_anchor_lens
from packs.register.loader import RegisterPack, load_register_pack
from packs.safety.check import SafetyCheck, SafetyContext
from packs.safety.loader import load_safety_pack
from field.state import FieldState
from generate.articulation import ArticulationPlan, realize
from generate.dialogue import DialogueRole, classify_dialogue_blade, propose_dialogue
from generate.graph_constraint import build_graph_constraint
from generate.intent_bridge import articulate_with_intent, build_graph_from_input
from generate.proposition import FrameRegistry, Proposition, propose
from generate.result import GenerationResult
from generate.stream import generate
from generate.surface import SentenceAssembler, SentencePlan, SurfaceContext
from ingest.gate import inject
from language_packs import OOVPolicy, load_mounted_packs, load_pack, load_pack_entries
from persona.motor import PersonaMotor
from session.context import SessionContext
from session.correction import CorrectionPass
from vault.decompose import default_decomposer, default_gate
_TOKEN_RE = re.compile(r"\w+", re.UNICODE)
# ADR-0073d (L1.4) — extracts the engaged ``cognitive_mode_label`` from a
# composer-emitted ``[lens(<lens_id>):<mode>]`` annotation. The runtime
# uses this read-only to populate the TurnEvent telemetry field; the
# composer remains the only source of truth for engagement.
_ANCHOR_LENS_ANNOTATION_RE = re.compile(r"\[lens\(([^):]+)\):([^\]]+)\]")
def _auto_propose_from_candidates(candidates: list[DiscoveryCandidate]) -> None:
from teaching.proposals import (
ProposalError,
ProposalLog,
_current_revision,
propose_from_candidate,
)
from teaching.source import ProposalSource
log = ProposalLog()
for candidate in candidates:
source = ProposalSource(
kind="contemplation",
source_id=candidate.candidate_id,
emitted_at_revision=_current_revision(),
)
try:
propose_from_candidate(candidate, log=log, source=source)
except ProposalError:
pass
def _extract_anchor_lens_mode_label(surface: str, lens_id: str) -> str:
"""Return the engaged mode_label if *surface* carries a
``[lens(<lens_id>):<mode>]`` annotation for the given ``lens_id``.
Returns ``""`` when:
* surface is empty or contains no lens annotation
* lens_id is empty (no lens loaded)
* the annotation in surface is for a different lens_id (defensive)
Pure read; no side effects. Telemetry-only — the composer is the
sole source of truth for engagement (ADR-0073c).
"""
if not surface or not lens_id:
return ""
for match in _ANCHOR_LENS_ANNOTATION_RE.finditer(surface):
if match.group(1) == lens_id:
return match.group(2)
return ""
_SEED_ALIASES = {
"logos": "\u03bb\u03cc\u03b3\u03bf\u03c2",
"dabar": "\u05d3\u05d1\u05e8",
"or": "\u05d0\u05d5\u05e8",
"phos": "\u03c6\u03c9\u03c2",
"zoe": "\u03b6\u03c9\u03ae",
"arche": "\u1f00\u03c1\u03c7\u03ae",
"aletheia": "\u1f00\u03bb\u03ae\u03b8\u03b5\u03b9\u03b1",
}
_QUESTION_WORDS = frozenset({"what", "who", "how", "why", "when", "where", "which"})
# Comb pass 2026-05-21 — module-level constant so ``_prefer_prompt_anchor``
# does not allocate a fresh set on every English turn. Aux-verbs that
# precede the prompt's content noun ("is", "are", "was", "were") get
# filtered out so the content-noun search lands on the actual subject.
_BE_FORMS: frozenset[str] = frozenset({"is", "are", "was", "were"})
_TERMINALS = frozenset({".", "?", ";", "!"})
_UNKNOWN_DOMAIN_SURFACE = "I don't know — insufficient grounding for that yet."
def _build_vault_probe(vault, vocab):
"""Return a vault probe for discovery contemplation (W-016).
Queries the session vault at EpistemicStatus.COHERENT using the
subject lemma's versor as the lookup key. The probe is a pure
read; it never writes to the vault or changes runtime state.
Trust boundary: vault entries from SPECULATIVE/CONTESTED/FALSIFIED
tiers are excluded by passing min_status=COHERENT to vault.recall.
The probe returns only ``vault_coherent`` EvidencePointers per
_VaultProbe contract (teaching/contemplation.py).
"""
from teaching.discovery import EvidencePointer
from teaching.epistemic import EpistemicStatus
def _probe(subject: str, obj: str) -> tuple[EvidencePointer, ...]:
try:
query = vocab.get_versor(subject)
except KeyError:
return ()
hits = vault.recall(query, top_k=3, min_status=EpistemicStatus.COHERENT)
return tuple(
EvidencePointer(
source="vault_coherent",
ref=str(hit["index"]),
polarity="affirms",
epistemic_status="coherent",
)
for hit in hits
)
return _probe
def _vault_probe_for_context(context: SessionContext | None) -> Any | None:
"""Return a _VaultProbe callable or None for the given session context."""
if context is None:
return None
return _build_vault_probe(context.vault, context.vocab)
def _energy_scalar(energy_obj) -> float:
if energy_obj is None:
return 1.0
if isinstance(energy_obj, EnergyProfile):
return float(energy_obj.raw)
try:
return float(energy_obj)
except (TypeError, ValueError):
return 1.0
def _recall_energy_class_from_hits(hits: Sequence[dict]) -> str | None:
if not hits:
return None
profile = hits[0].get("energy_profile")
energy_class = getattr(profile, "energy_class", None)
value = getattr(energy_class, "value", None)
return value if isinstance(value, str) else None
def _is_question_input(raw_text: str, tokens: Sequence[str]) -> bool:
if raw_text.strip().endswith("?"):
return True
return bool(tokens and tokens[0].casefold() in _QUESTION_WORDS)
def _stable_dialogue_role(role: DialogueRole, *, raw_text: str, tokens: Sequence[str]) -> DialogueRole:
if role in {"question", "refute"} and not _is_question_input(raw_text, tokens):
return "elaborate"
return role
def _terminal_for_role(role: DialogueRole, output_language: str) -> str:
if role == "question":
return ";" if output_language == "grc" else "?"
return "."
def _terminate_surface(surface: str, *, role: DialogueRole, output_language: str) -> str:
stripped = surface.strip()
if not stripped:
return stripped
if stripped[-1] in _TERMINALS:
return stripped
return f"{stripped}{_terminal_for_role(role, output_language)}"
def _prefer_prompt_anchor(
articulation: ArticulationPlan,
filtered_tokens: Sequence[str],
*,
output_language: str,
) -> ArticulationPlan:
if output_language != "en" or len(filtered_tokens) < 2:
return articulation
# Comb pass 2026-05-21 — find the last content-bearing token by
# reverse iteration with short-circuit; pre-fix this built a full
# ``content_tokens`` list and then took ``[-1]``. Also: cache
# ``token.casefold()`` once per token via walrus operator instead
# of calling it twice (against ``_QUESTION_WORDS`` and the
# historical inline ``{"is", "are", "was", "were"}`` literal).
anchor: str | None = None
for token in reversed(filtered_tokens):
lower = token.casefold()
if lower in _QUESTION_WORDS or lower in _BE_FORMS:
continue
anchor = token
break
if anchor is None or anchor == articulation.subject:
return articulation
return replace(
articulation,
subject=anchor,
surface=" ".join(part for part in (anchor, articulation.predicate, articulation.object) if part),
)
@dataclass
class _StubBindingFrame:
frame_id: str
coherence_magnitude: float
region_ids: frozenset
cycle_index: int
@dataclass(frozen=True, slots=True)
class _FieldStateWithVersor:
"""Adapter exposing ``versor_condition`` for SafetyContext.
``FieldState`` itself does not carry a precomputed
``versor_condition`` attribute; it is computed on demand from
``versor_condition(state.F)``. The SafetyCheck predicate for
``preserve_versor_closure`` reads ``ctx.field_state.versor_condition``
via ``getattr``. This adapter exposes the precomputed value so the
predicate is runtime-checkable each turn.
"""
versor_condition: float
def _hash_identity_manifold(manifold) -> str:
"""Deterministic SHA-256 of the load-bearing identity-manifold fields.
ADR-0035 — feeds the ``no_identity_override`` predicate in
:class:`SafetyCheck`. The runtime never mutates ``identity_manifold``
after composition, so before- and after-turn hashes are equal by
construction; an unequal hash would indicate the predicate's exact
failure mode.
"""
payload = {
"value_axes": [
{
"axis_id": axis.axis_id,
"name": axis.name,
"direction": list(axis.direction),
"weight": axis.weight,
}
for axis in manifold.value_axes
],
"boundary_ids": sorted(manifold.boundary_ids),
"alignment_threshold": manifold.alignment_threshold,
}
blob = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
return hashlib.sha256(blob).hexdigest()
def _surface_contains_hedge(surface: str, manifold) -> bool:
"""Detect whether the realized surface emitted a hedge phrase.
Compares case-insensitively against the manifold's preferred hedge
phrases (ADR-0028). False when surface is empty. Coarse but
deterministic: the predicate downstream is observational, so
occasional false negatives are surfaced as
``acknowledge_uncertainty`` violations in audit and corrected by
refining hedge detection, not by silently passing.
"""
if not surface:
return False
prefs = getattr(manifold, "surface_preferences", None)
if prefs is None:
return False
candidates: list[str] = []
for field_name in (
"preferred_hedge_strong",
"preferred_hedge_soft",
"preferred_qualifier",
):
value = getattr(prefs, field_name, "")
if value:
candidates.append(value)
for _, hedge in getattr(prefs, "axis_hedges", ()) or ():
for sub in ("strong", "soft", "qualifier"):
value = getattr(hedge, sub, "")
if value:
candidates.append(value)
surface_fold = surface.casefold()
return any(c.casefold() in surface_fold for c in candidates if c)
def _make_trajectory_from_result(result, turn: int):
from core.physics.reasoning import TrajectoryOperator
operator = TrajectoryOperator()
states = result.trajectory or (result.final_state,)
frames = [
_StubBindingFrame(
frame_id=f"t{turn}_s{i}",
coherence_magnitude=_energy_scalar(getattr(fs, "energy", None)),
region_ids=frozenset({str(getattr(fs, "node", 0))}),
cycle_index=turn,
)
for i, fs in enumerate(states)
]
return operator.build(frames, trajectory_id=f"turn_{turn}")
@dataclass(frozen=True, slots=True)
class TurnAccrual:
"""The inline-realization outcome of one turn (Step B / E).
``kind`` is ``"realized"`` (a declarative fact was accrued into the held self),
``"determined"`` (a question was answered over realized knowledge), ``"estimated"``
(Step E — a converse query DETERMINE refused, for which a calibrated converse-guess
exists), or ``"none"`` (nothing comprehensible to accrue/determine). The payload
carries the typed result for introspection. B-1 records, does not surface; B-2 and E
surface only when their flag is on.
"""
kind: str
realized: Any = None # generate.realize.Realized | NotRealized | None
determination: Any = None # generate.determine.Determined | Undetermined | None
# Step E — the converse-guess candidate and its SERVE license, when ``kind`` is
# ``"estimated"``. ``estimate`` is a ``ConverseEstimate``; ``license`` a
# ``LicenseDecision`` (None if the predicate-class is absent from the ratified ledger).
estimate: Any = None
license: Any = None
@dataclass(frozen=True, slots=True)
class ChatResponse:
surface: str
proposition: Proposition
articulation: ArticulationPlan
articulation_surface: str
dialogue_role: DialogueRole
versor_condition: float
output_language: str
frame_pack: str
walk_surface: str
salience_top_k: int | None
candidates_used: int | None
vault_hits: int
identity_score: IdentityScore | None
character_profile: CharacterProfile
flagged: bool
recall_energy_class: str | None = None
# ADR-0023 §2 — per-transition admissibility evidence and region
# provenance flag. An empty tuple is the contract for "no
# admissibility was checked this turn" (cold start, refusal, stub).
admissibility_trace: tuple = ()
region_was_unconstrained: bool = True
# ADR-0035 — verdicts surfaced from SafetyCheck and EthicsCheck.
# ``None`` only on stub/refusal paths that bypass the turn loop.
safety_verdict: object = None
ethics_verdict: object = None
# ADR-0039 — unified TurnVerdicts bundle carrying identity / safety
# / ethics verdicts and the two remediation flags
# (refusal_emitted, hedge_injected). Typed as ``object`` to avoid
# coupling at module-resolution time; downcast at use site.
verdicts: object = None
# ADR-0048 / ADR-0050 / ADR-0052 — provenance tag for the surface's
# grounding. One of:
# "vault" — answer drawn from session vault evidence (main path).
# "pack" — answer drawn from the ratified language pack
# (cold-start DEFINITION/RECALL/COMPARISON on pack-known
# lemmas — ADR-0048 / ADR-0050).
# "teaching" — answer drawn from a reviewed teaching-chain corpus
# (cold-start CAUSE/VERIFICATION — ADR-0052).
# "none" — universal "insufficient grounding" disclosure on stub.
# The string is preserved verbatim in TurnEvent for downstream audit.
grounding_source: str = "none"
# ADR-0071 (R4) — pre-decoration surface. ``surface`` is the
# user-facing string AFTER seeded discourse-marker decoration;
# ``pre_decoration_surface`` is the realizer's output BEFORE the
# decoration step. The cognition pipeline reads this field to
# compute ``trace_hash`` so register decoration cannot leak into
# the truth path (ADR-0069 invariant C). Empty string ⇒ identical
# to ``surface`` (no decoration applied this turn).
pre_decoration_surface: str = ""
# Phase 3 epistemic taxonomy — first-class state axes per turn.
# Values are lower_snake_case strings matching core.epistemic_state
# enum values so the field serializes stably without importing the
# enum here. Defaults match TurnEvent defaults (undetermined /
# unassessable) so pre-Phase-3 callers that omit these fields are
# treated conservatively.
epistemic_state: str = "undetermined"
normative_clearance: str = "unassessable"
# Comma-separated violated boundary/commitment IDs when normative
# clearance is VIOLATED or SUPPRESSED; empty string otherwise.
normative_detail: str = ""
# ADR-0206 — Response Governance Bridge reach level for this turn.
# Mirrors the TurnEvent field so callers (CLI, demos, tests) can read
# the governed reach level from ChatResponse. Scaffold contract:
# always "strict" (govern_response is STRICT-only until widening is
# built). ``"strict"`` default preserves byte-identity for callers
# that construct ChatResponse without this field.
reach_level: str = "strict"
# ADR-0072 (R5) — operator-visible register identity per turn.
# Mirrors the TurnEvent fields so callers (CLI, demos, tests) can
# read the register state from ChatResponse without re-parsing the
# telemetry JSONL. ``""`` defaults preserve pre-R5 byte-identity
# for callers that construct ChatResponse without these fields.
register_id: str = ""
register_variant_id: str = ""
# ADR-0073d (L1.4) — operator-visible anchor-lens identity per turn.
# Mirrors the TurnEvent fields so callers (CLI, demos, tests) can
# read the lens state from ChatResponse without re-parsing the
# telemetry JSONL. ``""`` defaults preserve pre-L1.4 byte-identity.
anchor_lens_id: str = ""
anchor_lens_mode_label: str = ""
# ADR-0075 (C1) — realizer slot-type guard verdict. Mirrors the
# TurnEvent fields so callers (CLI, demos, tests) can read the
# guard state from ChatResponse without re-parsing the telemetry
# JSONL. ``""`` defaults preserve pre-C1 byte-identity.
realizer_guard_status: str = ""
realizer_guard_rule: str = ""
# ADR-0077 (R6) — register layering boundary surface. Carries the
# composer output BEFORE any register transformation (substantive
# or decorative). The cognition pipeline hashes this field for
# ``trace_hash`` when present, preserving R5's load-bearing
# invariant — substantive register transforms must not move
# ``trace_hash``. Empty string ⇒ pre-R6 caller; pipeline falls
# back to ``pre_decoration_surface`` (byte-identity preserved).
register_canonical_surface: str = ""
# ADR-0078 (Phase 1) — observational composer/graph atom
# equivalence telemetry mirrored from TurnEvent.
composer_graph_atom_status: str = ""
composer_atom_set_hash: str = ""
graph_atom_set_hash: str = ""
composer_graph_atom_overlap_count: int = 0
# ADR-0088 Phase B (audit Finding 2, 2026-05-20) — alphabetic-
# filtered walk tokens from the recall step. Populated only on
# the main path; the stub / refusal paths leave this empty.
# Consumed by ``CognitiveTurnPipeline`` when
# ``RuntimeConfig.realizer_grounded_authority`` is True so the
# proposition graph can be grounded before ``realize_semantic``
# is invoked. Empty tuple preserves pre-ADR-0088 byte-identity
# for every caller that constructs ChatResponse without this
# field.
recalled_words: tuple[str, ...] = ()
# ADR-0024 Phase 2 — stable refusal reason value
refusal_reason: str = ""
disposition: str = ""
dispatch_trace: DispatchTrace | None = None
@dataclass(frozen=True, slots=True)
class IdleTickResult:
"""Outcome of one ``idle_tick``.
The proposal pass is PROPOSAL-ONLY learning, never ratification (HITL ratifies). The
Step D consolidation pass is SESSION-memory learning: ``facts_consolidated`` derived
facts were written back into the held self so the next ``determine`` reaches them
directly — still never corpus mutation, never a proposal. The proposal-review sub-pass
(``proposal_review``) is READ-ONLY: it surfaces pending comprehension-failure proposals for
review and mutates nothing.
"""
candidates_contemplated: int
proposals_created: int
pending_proposals: int
#: Step D — derived facts consolidated into the held self this tick (0 unless
#: ``config.consolidate_determinations`` and the closure had a new layer to add).
facts_consolidated: int = 0
#: IT — read-only proposal-review summary (None unless ``config.review_pending_proposals``).
#: Surfaces pending comprehension-failure proposals for review; mutates nothing.
proposal_review: ProposalReviewIdleSummary | None = None
class ChatRuntime:
def __init__(
self,
pack_id: str | Sequence[str] | None = None,
*,
frame_pack: str | None = None,
config: RuntimeConfig = DEFAULT_CONFIG,
no_load_state: bool = False,
engine_state_path: Any | None = None,
) -> None:
if pack_id is not None or frame_pack is not None:
pack_ids = (pack_id,) if isinstance(pack_id, str) else tuple(pack_id or config.input_packs)
# Use dataclasses.replace so newer RuntimeConfig fields
# (identity_pack, ethics_pack, forward_graph_constraint,
# composed_surface, thread_anaphora, etc.) survive the
# pack_id / frame_pack override path. The previous manual
# reconstruction silently dropped any field not enumerated
# here, which would let a caller like
# ``ChatRuntime(pack_id="x", config=RuntimeConfig(composed_surface=True))``
# lose composed_surface without warning.
from dataclasses import replace as _dc_replace
resolved_config = _dc_replace(
config,
input_packs=pack_ids,
frame_pack=frame_pack or config.frame_pack,
)
else:
resolved_config = config
pack_ids = tuple(config.input_packs)
self.config = resolved_config
manifests = []
manifolds = []
entries = []
for mounted_pack_id in pack_ids:
manifest, manifold = load_pack(mounted_pack_id)
manifests.append(manifest)
manifolds.append(manifold)
entries.extend(load_pack_entries(mounted_pack_id))
manifold = manifolds[0] if len(pack_ids) == 1 else load_mounted_packs(pack_ids)
self._manifests = tuple(manifests)
# Comb pass 2026-05-21 — precompute OOV-policy aggregates so
# ``_apply_oov_policy`` doesn't rescan every manifest per OOV
# token. Manifests are immutable post-construction, so a
# one-time aggregate is safe and cuts the hot path from
# O(packs × OOV) to O(OOV).
self._all_manifests_fail_closed: bool = all(
m.oov_policy is OOVPolicy.FAIL_CLOSED for m in self._manifests
)
self._any_manifest_proposes_vocab: bool = any(
m.oov_policy is OOVPolicy.PROPOSE_VOCAB_EXPANSION for m in self._manifests
)
identity_pack_id = resolved_config.identity_pack or DEFAULT_IDENTITY_PACK
identity_manifold = load_identity_manifold(identity_pack_id)
self.safety_pack = load_safety_pack()
ethics_pack_id = resolved_config.ethics_pack or _DEFAULT_ETHICS_PACK
try:
self.ethics_pack = load_ethics_pack(ethics_pack_id)
except EthicsPackError:
if ethics_pack_id == _DEFAULT_ETHICS_PACK:
raise
self.ethics_pack = load_ethics_pack(_DEFAULT_ETHICS_PACK)
ethics_pack_id = _DEFAULT_ETHICS_PACK
self.ethics_pack_id = ethics_pack_id
# ADR-0068 / ADR-0069 — register pack load. None resolves to the
# in-memory unregistered sentinel (structurally identical to
# default_neutral_v1). Invalid ids fail-fast at runtime init,
# not at first turn. At R2 the register is loaded but no
# composer consumes it; byte-identity invariants pin this.
if resolved_config.register_pack_id is None:
self.register_pack: RegisterPack = RegisterPack.unregistered()
else:
self.register_pack = load_register_pack(
resolved_config.register_pack_id
)
self.register_pack_id = resolved_config.register_pack_id
# ADR-0073b — anchor-lens load. ``None`` resolves to the
# in-memory unanchored sentinel (structurally identical to
# ``default_unanchored_v1``). Invalid ids fail-fast at
# runtime init, not at first turn. At L1.2 the lens is
# loaded and stored but no composer consumes it; the
# ``anchor_lens_byte_identity_null_lift`` invariant pins this.
if resolved_config.anchor_lens_id is None:
self.anchor_lens: AnchorLens = AnchorLens.unanchored()
else:
self.anchor_lens = load_anchor_lens(
resolved_config.anchor_lens_id
)
self.anchor_lens_id = resolved_config.anchor_lens_id
self.identity_manifold = type(identity_manifold)(
value_axes=identity_manifold.value_axes,
boundary_ids=(
identity_manifold.boundary_ids
| self.safety_pack.boundary_ids
| self.ethics_pack.commitment_ids
),
alignment_threshold=identity_manifold.alignment_threshold,
surface_preferences=identity_manifold.surface_preferences,
)
self.identity_pack_id = identity_pack_id
persona_motor = PersonaMotor.identity()
self._context = SessionContext(
manifold,
persona=persona_motor,
vault_reproject_interval=resolved_config.vault_reproject_interval,
)
self._frame_registry = FrameRegistry.from_pack(resolved_config.frame_pack, self._context.vocab)
self._surface_by_fold = {e.surface.casefold(): e.surface for e in entries}
self._surface_by_fold.update(_SEED_ALIASES)
self._pos_by_surface = {e.surface: (e.pos or e.part_of_speech or "X") for e in entries}
self.exertion_meter = ExertionMeter(capacity_ceiling=128.0)
self.drive_gradients = tuple(GradientField(axis=axis, magnitude=0.75) for axis in self.identity_manifold.value_axes)
self._drive_map = DriveGradientMap(gradients=self.drive_gradients)
self.character_profile = CharacterProfile.from_manifold(
self.identity_manifold,
drive_summaries={g.axis.name: g.magnitude for g in self.drive_gradients},
fatigue_index=0.0,
)
self._identity_check = IdentityCheck()
self.safety_check = SafetyCheck()
self.ethics_check = EthicsCheck()
self._identity_manifold_hash: str = _hash_identity_manifold(
self.identity_manifold,
)
self._last_refusal_was_typed: bool = True
self.turn_log: List[TurnEvent] = []
from chat.thread_context import ThreadContext
self.thread_context = ThreadContext()
self._telemetry_sink: TurnEventSink | None = None
self._telemetry_include_content: bool = False
self._discovery_sink: DiscoveryCandidateSink | None = None
self._oov_sink: Any = None
self._contemplate_discoveries: bool = False
self._correction_pass = CorrectionPass()
self._last_valence: float = 0.0
# Phase 3 — most-recent plan-contemplation findings (tuple of
# SPECULATIVE ``ContemplationFinding`` records). Reset to ``()``
# on every turn; populated only when ``config.discourse_contemplation``
# is True AND the planner actually engaged on the turn. Exposed
# via the ``last_plan_findings`` property below.
self._last_plan_findings: tuple[Any, ...] = ()
# Phase 4 — most-recent plan-articulation metrics (PlanMetrics).
# Reset to ``None`` between turns. Populated under the same
# gating discipline as ``_last_plan_findings``: requires
# ``config.discourse_contemplation`` + an engaged planner.
self._last_plan_metrics: Any | None = None
# Phase 5 — articulation-observation sink (per-turn JSONL stream
# consumed by the offline ``mine_articulation_observations``
# miner). Attached via ``attach_articulation_sink``; ``None``
# by default so the runtime emits nothing until an operator
# opts in. Behaviour mirrors ``attach_telemetry_sink``:
# append-only, fail-fast on sink errors, deterministic JSONL.
self._articulation_sink: Any | None = None
self._articulation_turn_counter: int = 0
# W-013 — last classified intent, updated each turn for /explain REPL use.
self._last_intent: Any | None = None
self._last_input_text: str = ""
# Step B (inline realization) — the last turn's accrual outcome (what the turn
# realized into the held self, or determined over it). Introspectable; the
# surface contract is unchanged (slice B-1 records, does not surface).
self._last_turn_accrual: TurnAccrual | None = None
self._relational_pack_lemmas: frozenset[str] | None = None
self._engine_state_store: EngineStateStore | None = (
None if no_load_state else EngineStateStore(engine_state_path)
)
self._recognizer_registry: RecognizerRegistry = RecognizerRegistry()
self._turn_count: int = 0
self._pending_candidates: list[DiscoveryCandidate] = []
self._pending_recognizer_examples: list[
tuple[tuple[str, ...], FeatureBundle]
] = []
# W-024 / ADR-0158 — reboot event JSONL line buffered here when a
# checkpoint is loaded, flushed to the sink on attach_telemetry_sink.
# None means no reboot was detected this session.
self._pending_reboot_payload: str | None = None
# L11 — the engine's content-derived identity (who am I), and the
# identity stamped in the loaded checkpoint (the lineage parent for the
# next checkpoint). ``_loaded_engine_identity`` stays "" at genesis.
self._engine_identity: str = engine_identity_for_config(
self.config, get_git_revision()
)
self._loaded_engine_identity: str = ""
# CL — the persistent reviewed-learning proposal log. ``idle_tick()``
# advances it during idle (proposal-only); it lives alongside the engine
# state so the learning backlog survives reboot. None for no_load_state
# (ephemeral runtimes don't accumulate a learning lineage).
self._proposal_log: Any | None = None
if self._engine_state_store is not None:
from teaching.proposals import ProposalLog
self._proposal_log = ProposalLog(
path=self._engine_state_store.path / "proposals.jsonl"
)
# L11 — set True on reboot when the stamped checkpoint identity differs
# from the recomputed identity (the ratified substrate changed while down).
self.identity_continuity_break: bool = False
if self._engine_state_store is not None and self._engine_state_store.exists():
self._load_engine_state()
def _load_engine_state(self) -> None:
store = self._engine_state_store
if store is None:
return
# Schema-version compatibility gates the whole load: load_manifest()
# refuses (raises) a checkpoint written by newer code BEFORE we read any
# recognizers/candidates (L10 step-2 migration discipline).
manifest = store.load_manifest() or {}
recognizers = store.load_recognizers()
self._recognizer_registry = RecognizerRegistry.from_recognizers(recognizers)
self._pending_candidates = store.load_discovery_candidates()
self._turn_count = int(manifest.get("turn_count", 0))
# L11 — the identity this checkpoint was written under becomes the lineage
# parent of the next checkpoint we write. If it differs from the identity
# we recomputed at boot, the ratified substrate changed during downtime:
# we would resume the lived state under a DIFFERENT identity. Surface it
# (warn + flag); refuse only under strict_identity_continuity.
self._loaded_engine_identity = str(manifest.get("engine_identity", ""))
if (
self._loaded_engine_identity
and self._loaded_engine_identity != self._engine_identity
):
self.identity_continuity_break = True
message = (
"engine identity continuity break: checkpoint was written under "
f"{self._loaded_engine_identity[:12]}… but this build computes "
f"{self._engine_identity[:12]}… — the ratified identity substrate "
"changed while the engine was down. Resuming would carry the lived "
"state into a different identity."
)
if self.config.strict_identity_continuity:
from core.engine_identity import IdentityContinuityError
raise IdentityContinuityError(message)
warnings.warn(message, RuntimeWarning, stacklevel=2)
# Shape B+ (schema v2): restore the lived session state into the live
# context so a reboot resumes the SAME life (field/vault/anchor/graph/
# referents/dialogue). Opt-in (config.persist_session_state); None for a
# v1 checkpoint -> fresh session (the historical Shape B behavior), so old
# checkpoints stay loadable.
if self.config.persist_session_state and self._context is not None:
session_snapshot = store.load_session_state()
if session_snapshot is not None:
self._context.restore(session_snapshot)
# W-024 / ADR-0158 — buffer reboot event for emission when sink attaches.
from engine_state import _git_revision
self._pending_reboot_payload = format_reboot_event_jsonl(
restored_turn_count=self._turn_count,
stored_revision=str(manifest.get("written_at_revision", "unknown")),
current_revision=_git_revision(),
recognizers_count=len(recognizers),
candidates_count=len(self._pending_candidates),
)
if self.config.auto_proposal_enabled and self._pending_candidates:
_auto_propose_from_candidates(self._pending_candidates)
def checkpoint_engine_state(self) -> None:
store = self._engine_state_store
if store is None:
return
if (
self.config.recognition_grounded_graph
and self._pending_recognizer_examples
):
recognizer = derive_recognizer(tuple(self._pending_recognizer_examples))
self._recognizer_registry.register(recognizer)
self._pending_recognizer_examples.clear()
store.save_recognizers(self._recognizer_registry.all())
candidates_to_save = self._pending_candidates
if self.config.auto_contemplate and candidates_to_save:
from teaching.contemplation import contemplate
vault_probe = _vault_probe_for_context(self._context) if self._context else None
candidates_to_save = [
contemplate(c, vault_probe=vault_probe)
for c in candidates_to_save
]
store.save_discovery_candidates(candidates_to_save)
# Shape B+ (schema v2): persist the lived session state (field, vault,
# anchor, graph, referents, dialogue) BEFORE the manifest, so the
# manifest stays the last durable act — the commit marker for the turn.
# Opt-in (config.persist_session_state): a deliberate resume mode, off by
# default so one-shot runtimes don't pay the per-turn snapshot cost.
if self._context is not None and self.config.persist_session_state:
store.save_session_state(self._context.snapshot())
# L11 — stamp the engine's identity and its lineage parent (the identity
# of the prior checkpoint). Same substrate -> identity == parent (a stable
# life); a ratified substrate change -> identity != parent (the bump).
store.save_manifest(
self._turn_count,
engine_identity=self._engine_identity,
parent_engine_identity=self._loaded_engine_identity,
)
self._loaded_engine_identity = self._engine_identity
def _count_pending_proposals(self) -> int:
if self._proposal_log is None:
return 0
return sum(
1
for entry in self._proposal_log.current_state().values()
if entry.get("state") == "pending"
)
def idle_tick(self) -> "IdleTickResult":
"""Advance learning during idle (NO user turn). Two disjoint passes:
1. PROPOSAL pass (the reviewed-learning flywheel): turn the pending discovery
backlog into reviewable teaching proposals. Contemplate each pending
candidate (enrichment) and run the replay-gated ``propose_from_candidate``,
which leaves a PROPOSAL-ONLY ``pending`` entry in the persistent proposal
log. An idle tick NEVER ratifies — ratification (appending to the corpus)
stays HITL via ``teaching/review`` (CLAUDE.md teaching safety). The tick only
*proposes*; the reviewed loop is not bypassed or duplicated.
2. CONSOLIDATION pass (Step D — CLOSE): the loop learns from *determined* facts.
Run one semi-naive layer of the member/subset deductive closure over the held
self (``generate.determine.consolidate_once``); each soundly-derived,
proof_chain-verified fact is written back as a SPECULATIVE realized record so
the next ``determine`` reaches it directly. SESSION memory (immediate,
allowed) — NOT corpus mutation, NOT a proposal; the HITL path is untouched.
Gated by ``config.consolidate_determinations``. The closure converges (a
saturated tick consolidates nothing — ``at_fixed_point``).
The proposal log, the candidate backlog, and (with ``persist_session_state``)
the consolidated facts all live in the engine-state dir, so this learning
progress persists across reboot (CL-2).
"""
contemplated_count = 0
created = 0
facts_consolidated = 0
did_work = False
# 1. Proposal pass — unchanged behavior, runs only with a log + backlog.
if self._proposal_log is not None and self._pending_candidates:
from teaching.contemplation import contemplate
from teaching.proposals import (
ProposalError,
TeachingChainProposal,
_current_revision,
propose_from_candidate,
)
from teaching.source import ProposalSource
vault_probe = (
_vault_probe_for_context(self._context) if self._context else None
)
contemplated = [
contemplate(candidate, vault_probe=vault_probe)
for candidate in self._pending_candidates
]
contemplated_count = len(contemplated)
for candidate in contemplated:
source = ProposalSource(
kind="contemplation",
source_id=candidate.candidate_id,
emitted_at_revision=_current_revision(),
)
try:
result = propose_from_candidate(
candidate, log=self._proposal_log, source=source
)
except ProposalError:
continue
if isinstance(result, TeachingChainProposal):
created += 1
did_work = True
# 2. Consolidation pass (Step D) — runs independently of the backlog.
if self.config.consolidate_determinations and self._context is not None:
from generate.determine import consolidate_once
facts_consolidated = consolidate_once(self._context).consolidated
did_work = True
# 3. Proposal-review sub-pass (IT) — READ-ONLY. Surfaces pending comprehension-failure
# proposals (the contemplation pass's N5 artifacts) for review. It NEVER mutates an
# artifact, NEVER sets ``did_work`` (no state change → no checkpoint), NEVER ratifies
# or mounts. A reporter failure is CAPTURED, not propagated, so it cannot corrupt the
# tick. Gated off by default.
proposal_review = None
if self.config.review_pending_proposals:
try:
proposal_review = idle_summary()
except Exception as exc: # noqa: BLE001 — a reporter failure must not corrupt the tick
proposal_review = ProposalReviewIdleSummary(
safe=False, total=0, review_needed=0, malformed=0, by_family=(),
errors=(f"proposal_review_failed:{type(exc).__name__}",),
)
# Persist the advanced state once (backlog + lineage +, with
# persist_session_state, the consolidated facts). Skipped on a no-op tick so an
# idle engine with nothing to learn does not churn the checkpoint.
if did_work:
self.checkpoint_engine_state()
return IdleTickResult(
candidates_contemplated=contemplated_count,
proposals_created=created,
pending_proposals=self._count_pending_proposals(),
facts_consolidated=facts_consolidated,
proposal_review=proposal_review,
)
def record_recognition_example(
self,
tokens: tuple[str, ...],
bundle: FeatureBundle,
) -> None:
self._pending_recognizer_examples.append((tuple(tokens), bundle))
def finalize_turn_trace_hash(self, trace_hash: str) -> None:
"""ADR-0153 (W-020a) — back-stamp the canonical trace_hash.
Called by ``CognitiveTurnPipeline.process`` after
``compute_trace_hash`` produces the turn's canonical
SHA-256. Stamps the trace_hash onto the most recent
TurnEvent and any DiscoveryCandidate emitted during this
turn (i.e., the unstamped tail of ``_pending_candidates``),
then re-persists the candidates checkpoint so the on-disk
audit trail names the originating turn instead of the
prior empty-string default.
No-op when ``trace_hash`` is empty (pre-pipeline call sites,
refusal stub path). Idempotent: stamping a tail whose
``source_turn_trace`` is already non-empty halts the back-walk.
"""
if not trace_hash:
return
from dataclasses import replace
if self.turn_log:
last_event = self.turn_log[-1]
if not last_event.trace_hash:
self.turn_log[-1] = replace(last_event, trace_hash=trace_hash)
stamped = False
for idx in range(len(self._pending_candidates) - 1, -1, -1):
cand = self._pending_candidates[idx]
if cand.source_turn_trace:
break
self._pending_candidates[idx] = replace(
cand, source_turn_trace=trace_hash
)
stamped = True
if stamped and self._engine_state_store is not None:
self._engine_state_store.save_discovery_candidates(
self._pending_candidates
)
def first_admitted_recognizer(self):
if not self.config.recognition_grounded_graph:
return None
return self._recognizer_registry.first_admitted()
def _checkpointed_response(self, response: ChatResponse) -> ChatResponse:
self._turn_count += 1
# Step B — inline realization, BEFORE the checkpoint so an accrued fact is in
# the snapshot this turn (survives reboot with persist_session_state). Gated;
# off by default. The surface contract is unchanged (the outcome is recorded,
# not surfaced).
if self.config.accrue_realized_knowledge:
self._accrue_in_turn(self._last_input_text)
response = self._maybe_surface_determination(response)
self.checkpoint_engine_state()
return response
def _maybe_surface_determination(self, response: ChatResponse) -> ChatResponse:
"""Step B-2 / E — select the user-facing surface from the turn's accrual.
B-2: when the turn DETERMINED an answer over realized knowledge, select the
rendered determination as ``surface`` (the realizer's ``articulation_surface``
is retained as evidence). An ``Undetermined`` turn keeps the default surface.
E: when the turn is ``estimated`` (a refused converse query with a calibrated
guess) AND ``estimation_enabled``, route the guess through the ADR-0206 bridge —
``govern_response`` widens to APPROXIMATE iff the predicate-class holds a genuine
SERVE license, and ``shape_surface`` DISCLOSES it as ``[approximate] …``. An
unlicensed class stays STRICT (the surface is unchanged — the honest refusal).
Off-flag turns never reach here. See ``docs/runtime_contracts.md``.
"""
accrual = self._last_turn_accrual
if accrual is None:
return response
if accrual.kind == "determined":
from generate.determine import Determined, render_determination
if not isinstance(accrual.determination, Determined):
return response # Undetermined → keep the default surface
return replace(response, surface=render_determination(accrual.determination))
if accrual.kind == "estimated" and self.config.estimation_enabled:
return self._surface_estimate(response, accrual)
return response
def _surface_estimate(self, response: ChatResponse, accrual: "TurnAccrual") -> ChatResponse:
"""Surface a licensed converse-guess as a DISCLOSED ``[approximate]`` estimate.
The license gates the widening (``govern_response`` returns STRICT for an
unlicensed class → surface unchanged); ``shape_surface`` guarantees the
disclosure prefix because a converse guess is ``UNVERIFIED_POSSIBLE``, never in
APPROXIMATE's admissible (fully-grounded) set. So a wrong estimate is always a
DISCLOSED wrong — wrong=0 (silent) is preserved.
"""
from core.epistemic_state import EpistemicState
from core.response_governance import ReachLevel, govern_response, shape_surface
from generate.determine import ConverseEstimate, render_estimate
estimate, license_decision = accrual.estimate, accrual.license
if not isinstance(estimate, ConverseEstimate):
return response
policy = govern_response(
epistemic_state=EpistemicState.UNVERIFIED_POSSIBLE,
license_decision=license_decision,
)
if policy.level is ReachLevel.STRICT:
return response # unlicensed → no widening, honest refusal stands
disclosed = shape_surface(
policy,
committed_surface=response.surface,
decode_state=EpistemicState.UNVERIFIED_POSSIBLE,
disclosed_alternative=render_estimate(estimate),
)
return replace(response, surface=disclosed, reach_level=policy.level.value)
def last_turn_accrual(self) -> TurnAccrual | None:
"""The most recent turn's inline-realization outcome (Step B), or None when
accrual is off or did not complete. Introspection only — never surfaced."""
return self._last_turn_accrual
def _accrue_in_turn(self, text: str) -> None:
"""Inline realization (Step B): a comprehensible declarative turn accrues a
realized fact into the held self (session vault, SPECULATIVE / as-told); a
comprehensible question turn is determined over realized knowledge. Records the
typed outcome on ``self._last_turn_accrual``.
Never raises into the turn — accrual is ADDITIVE, so any failure is a clean
no-op (the turn's response is untouched). This is SESSION memory (immediate),
NOT ratified learning: it proposes nothing and leaves the teaching/review HITL
path untouched. Realization writes go through ``generate.realize`` (the INV-21
allowed vault writer); DETERMINE and the readers are total (typed results, no
raises), so the broad guard is a defensive backstop, not expected control flow.
"""
self._last_turn_accrual = None
if self._context is None or not text or not text.strip():
return
try:
from generate.determine import determine
from generate.meaning_graph.reader import Comprehension, comprehend
from generate.meaning_graph.relational import comprehend_relational
from generate.realize import realize_comprehension
if self._relational_pack_lemmas is None:
from generate.meaning_graph.relational import load_relational_pack_lemmas
self._relational_pack_lemmas = load_relational_pack_lemmas()
readings = (
comprehend(text),
comprehend_relational(text, self._relational_pack_lemmas),
)
comprehensions = [c for c in readings if isinstance(c, Comprehension)]
# A question turn (query-bearing) is DETERMINED over realized knowledge.
for c in comprehensions:
if c.queries:
determination = determine(c, self._context)
self._last_turn_accrual = self._accrue_estimate_if_refused(
c, determination
)
return
# A declarative turn (a single told fact) is REALIZED into the held self.
for c in comprehensions:
if not c.queries and c.meaning_graph.relations:
self._last_turn_accrual = TurnAccrual(
kind="realized", realized=realize_comprehension(c, self._context)
)
return
self._last_turn_accrual = TurnAccrual(kind="none")
except Exception: # additive: accrual must never crash a turn # noqa: BLE001
self._last_turn_accrual = None
def _accrue_estimate_if_refused(self, comprehension: Any, determination: Any) -> "TurnAccrual":
"""Step E: turn a REFUSED converse query into an ``estimated`` accrual.
When DETERMINE refused (``Undetermined``) a single non-negated binary query whose
converse was told (``p(a,b)`` realized, ``p(b,a)`` asked), produce the calibrated
converse-guess + its SERVE license. Off-flag (or any non-converse refusal) returns
the plain ``determined`` accrual unchanged — nothing widens. The license is only
*attached* here; the surface decision (and the disclosure) is the bridge's, in
``_maybe_surface_determination``.
"""
from generate.determine import Undetermined
if not (self.config.estimation_enabled and isinstance(determination, Undetermined)):
return TurnAccrual(kind="determined", determination=determination)
queries = getattr(comprehension, "queries", ())
if len(queries) != 1:
return TurnAccrual(kind="determined", determination=determination)
query = queries[0]
if getattr(query, "negated", False) or len(getattr(query, "arguments", ())) != 2:
return TurnAccrual(kind="determined", determination=determination)
from generate.determine import estimate_converse, serve_license
subject, target = query.arguments[0], query.arguments[1]
estimate = estimate_converse(self._context, query.predicate, subject, target)
if estimate is None: # no told converse to generalize from → plain refusal
return TurnAccrual(kind="determined", determination=determination)
return TurnAccrual(
kind="estimated",
determination=determination,
estimate=estimate,
license=serve_license(query.predicate),
)
@property
def session(self) -> SessionContext:
return self._context
@property
def last_plan_findings(self) -> tuple[Any, ...]:
"""Phase 3 — most-recent plan-contemplation findings.
Tuple of ``core.contemplation.schema.ContemplationFinding``
records (always SPECULATIVE per ADR-0080). Populated only
when ``config.discourse_contemplation`` is True and the
discourse planner engaged on the turn — empty tuple
otherwise. Read-only observation surface; the runtime
itself never acts on findings, the offline contemplation
miner does.
"""
return self._last_plan_findings
@property
def last_plan_metrics(self) -> Any | None:
"""Phase 4 — most-recent plan articulation metrics.
``core.contemplation.plan_metrics.PlanMetrics`` instance
when the discourse planner engaged on the most recent turn
AND ``config.discourse_contemplation`` is True; ``None``
otherwise. Read-only quantitative companion to
``last_plan_findings`` (which carries the qualitative
SPECULATIVE concerns). Designed for downstream aggregation
— Phase 5's offline contemplation miner streams these
across turns to score plan-quality patterns the runtime
never tries to act on alone.
"""
return self._last_plan_metrics
def explain_last_turn(self) -> str:
"""Return a canonical natural-language restatement of the last turn (W-013).
Feeds the last classified intent through ``core.cognition.explain``'s
dispatch table and returns the resulting canonical prompt string.
This is the ``/explain`` REPL command's backing method.
Returns an empty string when no turn has been processed yet or when
the intent could not be classified (UNKNOWN tag).
"""
from core.cognition.explain import explain_from_intent
return explain_from_intent(
self._last_intent,
correction_text=self._last_input_text,
)
def attach_telemetry_sink(
self,
sink: TurnEventSink | None,
*,
include_content: bool = False,
) -> None:
"""ADR-0040 — attach a structured-logging sink."""
self._telemetry_sink = sink
self._telemetry_include_content = bool(include_content)
# W-024 / ADR-0158 — flush buffered reboot event now that sink is live.
if sink is not None and self._pending_reboot_payload is not None:
sink.emit(self._pending_reboot_payload)
self._pending_reboot_payload = None
def attach_articulation_sink(self, sink: Any | None) -> None:
"""Phase 5 — attach a sink for per-turn articulation observations.
``sink`` must satisfy
``chat.articulation_telemetry.ArticulationObservationSink``
(any object with ``def emit(line: str) -> None``). Pass
``None`` to detach.
The sink receives one canonical JSONL line per turn that
engages the discourse planner AND has
``config.discourse_contemplation == True``; non-engaged turns
emit nothing. Lines are byte-identical for byte-equal plans
— the offline miner relies on this for deterministic
aggregation.
"""
self._articulation_sink = sink
def attach_oov_sink(self, sink: Any) -> None:
"""Phase 2.3 — attach an OOV candidate sink."""
self._oov_sink = sink
def attach_discovery_sink(
self,
sink: DiscoveryCandidateSink | None,
) -> None:
"""ADR-0055 Phase B — attach a DiscoveryCandidate sink."""
self._discovery_sink = sink
def attach_contemplation(self, *, enabled: bool = True) -> None:
"""ADR-0056 Phase C1 — opt-in inline contemplation."""
self._contemplate_discoveries = bool(enabled)
def _push_thread_summary(
self,
*,
turn_event: TurnEvent,
intent_tag: Any,
intent_subject: str | None,
grounding_source: str | None,
surface: str | None = None,
) -> None:
"""P3.1 — append one TurnSummary to the bounded session-thread context."""
from chat.thread_context import TurnSummary
turn_index = len(self.turn_log) - 1
if intent_tag is not None and hasattr(intent_tag, "name"):
intent_name = str(intent_tag.name).lower()
else:
intent_name = ""
subject = (intent_subject or "").strip().lower()
source = (grounding_source or "none").lower()
chain_id: str | None = None
corpus_id: str | None = None
if source == "teaching" and subject and intent_name in {"cause", "verification"}:
from chat.teaching_grounding import _all_chains_index
chain = _all_chains_index().get((subject, intent_name))
if chain is not None:
chain_id = chain.chain_id
corpus_id = chain.corpus_id
_ = surface
self.thread_context.push(
TurnSummary(
turn_index=turn_index,
intent_tag_name=intent_name,
subject=subject,
grounding_source=source,
chain_id=chain_id,
corpus_id=corpus_id,
)
)
def _emit_oov_candidate(
self,
*,
turn_event: TurnEvent,
intent_tag: Any,
token: str | None,
) -> None:
"""P2.3 — emit one OOVCandidate per OOV-grounded turn."""
sink = self._oov_sink
if sink is None or not token:
return
from teaching.oov_sink import (
OOVCandidate,
format_oov_candidate_jsonl,
hash_oov_candidate_id,
)
from generate.intent import IntentTag
if intent_tag is None or not isinstance(intent_tag, IntentTag):
return
intent_name = intent_tag.name.lower()
trace_hash = getattr(turn_event, "trace_hash", "") or ""
boundary_clean = (
not getattr(turn_event, "refusal_emitted", False)
and not getattr(turn_event, "hedge_injected", False)
)
cleaned_token = (token or "").strip().lower()
if not cleaned_token:
return
candidate_id = hash_oov_candidate_id(cleaned_token, intent_name, trace_hash)
candidate = OOVCandidate(
candidate_id=candidate_id,
token=cleaned_token,
intent=intent_name, # type: ignore[arg-type]
trigger="unresolved_subject",
source_turn_trace=trace_hash,
boundary_clean=boundary_clean,
)
sink.emit(format_oov_candidate_jsonl(candidate))
def _emit_discovery_candidates(
self,
*,
turn_event: TurnEvent,
intent_tag: Any,
intent_subject: str | None,
grounding_source: str | None,
) -> None:
candidates = extract_discovery_candidates(
turn_event,
intent_tag,
intent_subject,
grounding_source=grounding_source,
)
if self._contemplate_discoveries and candidates:
from teaching.contemplation import contemplate
vault_probe = (
_build_vault_probe(self._context.vault, self._context.vocab)
if self.config.vault_probe_discoveries
else None
)
candidates = tuple(
contemplate(c, vault_probe=vault_probe) for c in candidates
)
self._pending_candidates.extend(candidates)
sink = self._discovery_sink
if sink is None:
return
for candidate in candidates:
sink.emit(format_candidate_jsonl(candidate))
def _emit_turn_event(self, event: TurnEvent) -> None:
sink = self._telemetry_sink
if sink is None:
return
line = format_turn_event_jsonl(
event,
safety_pack_id=self.safety_pack.pack_id,
ethics_pack_id=self.ethics_pack_id,
identity_pack_id=self.identity_pack_id,
include_content=self._telemetry_include_content,
)
sink.emit(line)
def _tokenize(self, text: str) -> list[str]:
return [self._surface_by_fold.get(m.group(0).casefold(), m.group(0)) for m in _TOKEN_RE.finditer(text)]
def tokenize(self, text: str) -> list[str]:
return self._tokenize(text)
def _apply_oov_policy(self, tokens: list[str]) -> list[str]:
# Comb pass 2026-05-21 — OOV-policy aggregates are precomputed
# at ``__init__`` so this method stays O(OOV tokens) rather
# than O(packs × OOV tokens). See ``_all_manifests_fail_closed``
# / ``_any_manifest_proposes_vocab``.
kept: list[str] = []
for token in tokens:
try:
self._context.vocab.get_versor(token)
kept.append(token)
except KeyError:
if self._all_manifests_fail_closed:
raise
if self._any_manifest_proposes_vocab:
raise KeyError(f"OOV token requires vocab proposal: {token}")
kept.append(token)
return kept
def _syntactic_guard(self, tokens: tuple[str, ...]) -> list[str]:
out: list[str] = []
prev_pos: str | None = None
for token in tokens:
pos = self._pos_by_surface.get(token, "X")
if pos == prev_pos:
continue
out.append(token)
prev_pos = pos
return out
def _dialogue_reference(self) -> np.ndarray | None:
blade = self._context.last_dialogue_blade
if blade is None or float(np.linalg.norm(blade)) < 1e-8:
return None
return blade
def _apply_drive_bias(self, field_state: FieldState) -> FieldState:
return field_state
def _build_surface_context(self, identity_score, current_valence: float) -> SurfaceContext:
active = self._context.referents.active_referent()
alignment = float(identity_score.alignment) if identity_score is not None else 1.0
deviation_axes = (
frozenset(identity_score.deviation_axes)
if identity_score is not None
else frozenset()
)
prefs = self.identity_manifold.surface_preferences
axis_hedges = tuple(
(axis_id, hedge.strong, hedge.soft, hedge.qualifier)
for axis_id, hedge in prefs.axis_hedges
)
return SurfaceContext(
active_referent_surface=active.surface if active is not None else "",
active_referent_slot=active.slot if active is not None else "neut_sg",
identity_alignment=alignment,
valence_delta=current_valence - self._last_valence,
elab_conjunction="",
hedge_threshold_strong=prefs.hedge_threshold_strong,
hedge_threshold_soft=prefs.hedge_threshold_soft,
preferred_hedge_strong=prefs.preferred_hedge_strong,
preferred_hedge_soft=prefs.preferred_hedge_soft,
claim_strength=prefs.claim_strength,
qualified_band_high=prefs.qualified_band_high,
preferred_qualifier=prefs.preferred_qualifier,
deviation_axes=deviation_axes,
axis_hedges=axis_hedges,
)
def _maybe_pack_grounded_surface(
self, text: str, gate_source: str, *, allow_warm: bool = False, attempts: list[DispatchAttempt] | None = None
) -> tuple[str, str, tuple[str, ...]] | None:
"""Return ``(surface, grounding_source)`` or ``None``.
ADR-0048 / ADR-0050 / ADR-0052 — three reviewed sources of
cold-start grounding share this dispatcher.
``allow_warm=True`` bypasses the empty-vault gate so the warm
path can engage pack-grounding for pack-resident DEFINITION /
RECALL / NARRATIVE / EXAMPLE / COMPARISON / PROCEDURE intents
— addresses ``warm_grounding_stability`` regression where
turn-2 of the same prompt drifted from a coherent pack surface
to a walk fragment. CAUSE / VERIFICATION still return None
when no teaching chain exists, preserving the discovery signal.
"""
if not allow_warm and gate_source != "empty_vault":
if attempts is not None:
for src in ("pack", "teaching", "partial", "oov"):
attempts.append(DispatchAttempt(source=src, outcome="skipped", reason="warm_path_disabled"))
return None
if self.config.output_language != "en":
if attempts is not None:
for src in ("pack", "teaching", "partial", "oov"):
attempts.append(DispatchAttempt(source=src, outcome="skipped", reason="non_english_output"))
return None
from generate.intent import IntentTag
from generate.intent_bridge import classify_intent_from_input
intent = classify_intent_from_input(text)
self._last_intent = intent # W-013: expose for /explain
if intent.tag is IntentTag.COMPARISON:
lemma_a = (intent.subject or "").strip().rstrip(".,?!;:")
lemma_b = (intent.secondary_subject or "").strip().rstrip(".,?!;:")
if lemma_a and lemma_b:
surface = pack_grounded_comparison_surface(
lemma_a, lemma_b, register=self.register_pack,
)
if surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="admitted", reason="pack_comparison_surface_found"))
return (surface, "pack", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="no_pack_comparison_surface"))
from chat.partial_surface import partial_comparison_surface
partial = partial_comparison_surface(lemma_a, lemma_b)
if partial is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="partial", outcome="admitted", reason="partial_comparison_surface_found"))
return (partial[0], "partial", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="partial", outcome="fell_through", reason="no_partial_comparison_surface"))
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="missing_comparison_lemmas"))
attempts.append(DispatchAttempt(source="partial", outcome="fell_through", reason="missing_comparison_lemmas"))
if intent.tag is IntentTag.NARRATIVE:
lemma = (intent.subject or "").strip()
if lemma:
from chat.narrative_surface import narrative_grounded_surface
surface = narrative_grounded_surface(
lemma, register=self.register_pack,
)
if surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="teaching", outcome="admitted", reason="narrative_surface_found"))
return (surface, "teaching", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="teaching", outcome="fell_through", reason="no_narrative_surface"))
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="teaching", outcome="fell_through", reason="missing_subject"))
if intent.tag is IntentTag.EXAMPLE:
lemma = (intent.subject or "").strip()
if lemma:
from chat.example_surface import example_grounded_surface
surface = example_grounded_surface(
lemma, register=self.register_pack,
)
if surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="teaching", outcome="admitted", reason="example_surface_found"))
return (surface, "teaching", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="teaching", outcome="fell_through", reason="no_example_surface"))
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="teaching", outcome="fell_through", reason="missing_subject"))
if intent.tag in (IntentTag.CAUSE, IntentTag.VERIFICATION):
lemma = (intent.subject or "").strip()
if lemma:
if (
intent.tag is IntentTag.VERIFICATION
and intent.relation
and intent.secondary_subject
):
surface = pack_grounded_relation_confirmation_surface(
lemma,
intent.relation,
intent.object or intent.secondary_subject,
negated=intent.negated,
)
if surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="admitted", reason="pack_relation_confirmation_found"))
return (surface, "pack", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="no_pack_relation_confirmation"))
elif intent.tag is IntentTag.VERIFICATION:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="skipped", reason="missing_relation_or_secondary_subject"))
# ADR-0085 — gloss-aware CAUSE surface (opt-in). Tried
# FIRST so a lemma with a ratified gloss gets an
# explanation-shaped answer drawn from the gloss text
# instead of the chain-walk's structurally-correct-but-
# bureaucratic domain-tag walk. Falls through to the
# chain-walk on None (no gloss for this lemma), so the
# null-drop invariant holds: every case that lifted
# pre-ADR-0085 still lifts; only the *frame* shifts on
# lemmas where a gloss exists.
if (
self.config.gloss_aware_cause
and intent.tag is IntentTag.CAUSE
):
surface = gloss_aware_cause_surface(
lemma, register=self.register_pack,
anchor_lens=self.anchor_lens,
)
if surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="admitted", reason="gloss_aware_cause_found"))
return (surface, "pack", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="no_gloss_aware_cause"))
elif intent.tag is IntentTag.CAUSE:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="skipped", reason="gloss_aware_cause_disabled"))
if self.config.transitive_surface:
# ADR-0083 — transitive supersedes composed. At
# max_depth=1 this degrades byte-identically to the
# single-chain surface; at max_depth=2 byte-identical
# to ADR-0062 when no second hop exists.
surface = teaching_grounded_surface_transitive(
lemma,
intent.tag,
register=self.register_pack,
max_depth=self.config.transitive_max_depth,
)
reason_type = "transitive"
elif self.config.composed_surface:
surface = teaching_grounded_surface_composed(
lemma, intent.tag, register=self.register_pack,
)
reason_type = "composed"
else:
surface = teaching_grounded_surface(
lemma, intent.tag, register=self.register_pack,
)
reason_type = "standard"
if surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="teaching", outcome="admitted", reason=f"teaching_{reason_type}_surface_found"))
return (surface, "teaching", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="teaching", outcome="fell_through", reason=f"no_teaching_{reason_type}_surface"))
from chat.cross_pack_grounding import cross_pack_grounded_surface
surface = cross_pack_grounded_surface(
lemma, intent.tag, register=self.register_pack,
)
if surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="teaching", outcome="admitted", reason="cross_pack_grounded_surface_found"))
return (surface, "teaching", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="teaching", outcome="fell_through", reason="no_cross_pack_grounded_surface"))
# Deliberate non-fallback: when CAUSE / VERIFICATION
# has no teaching chain or cross-pack chain rooted on
# the subject, return None so the discovery layer logs
# a "would_have_grounded" candidate identifying the
# teaching-content gap. Emitting the bare pack
# disclosure here would mask that signal and give the
# user a non-answer (a definition rather than a cause).
# See ``tests/test_discovery_candidates``.
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="missing_subject"))
attempts.append(DispatchAttempt(source="teaching", outcome="fell_through", reason="missing_subject"))
if intent.tag is IntentTag.CORRECTION:
surface = pack_grounded_correction_surface(
text, register=self.register_pack,
)
if surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="admitted", reason="pack_correction_surface_found"))
return (surface, "pack", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="no_pack_correction_surface"))
if intent.tag is IntentTag.PROCEDURE:
subject_text = (intent.subject or "").strip()
if subject_text:
surface = pack_grounded_procedure_surface(
subject_text, register=self.register_pack,
)
if surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="admitted", reason="pack_procedure_surface_found"))
return (surface, "pack", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="no_pack_procedure_surface"))
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="missing_subject"))
if intent.tag in (IntentTag.DEFINITION, IntentTag.RECALL):
lemma = (intent.subject or "").strip()
if not lemma:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="missing_subject"))
return None
surface = pack_grounded_surface(
lemma,
register=self.register_pack,
anchor_lens=self.anchor_lens,
)
if surface is not None:
# ADR-0077 (R6) — expose the resolving lemma's
# semantic_domains so the runtime's substantive-register
# hook can fuel ``append_semantic_domain_clause``. All
# other composers return ``()`` because only the gloss
# DEFINITION/RECALL path participates in convivial's
# bounded propositional expansion in R6.
from chat.pack_resolver import resolve_lemma
resolved = resolve_lemma(lemma)
domains = resolved[1] if resolved is not None else ()
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="admitted", reason="pack_grounded_surface_found"))
return (surface, "pack", domains)
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="no_pack_grounded_surface"))
if intent.tag is IntentTag.UNKNOWN:
# ADR-0086 — UNKNOWN intent with pack-resident prompt
# tokens. The classifier could not assign a known dialogue
# shape, but the prompt itself may contain lemmas that are
# ratified in mounted lexicon packs (e.g. ``"light logos"``,
# ``"spirit wisdom truth"``). Surface those lemmas with
# their semantic_domains rather than emit the bare
# _UNKNOWN_DOMAIN_SURFACE disclosure. Null-lift invariant:
# when no prompt token resolves, composer returns None and
# the caller falls through to the universal disclosure
# byte-identically (preserves the ADR-0053 honesty contract
# for fully-OOV prompts).
surface = pack_grounded_unknown_surface(
text, register=self.register_pack,
)
if surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="admitted", reason="pack_grounded_unknown_surface_found"))
return (surface, "pack", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="no_pack_grounded_unknown_surface"))
# Check if any attempts have been recorded for pack/teaching/partial. If not, record them as skipped.
if attempts is not None:
has_pack = any(a.source == "pack" for a in attempts)
has_teaching = any(a.source == "teaching" for a in attempts)
has_partial = any(a.source == "partial" for a in attempts)
if not has_pack:
attempts.append(DispatchAttempt(source="pack", outcome="skipped", reason=f"intent_tag_{intent.tag.name.lower()}_not_targeted"))
if not has_teaching:
attempts.append(DispatchAttempt(source="teaching", outcome="skipped", reason=f"intent_tag_{intent.tag.name.lower()}_not_targeted"))
if not has_partial:
attempts.append(DispatchAttempt(source="partial", outcome="skipped", reason=f"intent_tag_{intent.tag.name.lower()}_not_targeted"))
oov_lemma = (intent.subject or "").strip()
if oov_lemma:
from chat.oov_surface import oov_learning_invitation_surface
oov_surface = oov_learning_invitation_surface(oov_lemma, intent.tag)
if oov_surface is not None:
if attempts is not None:
attempts.append(DispatchAttempt(source="oov", outcome="admitted", reason="oov_learning_invitation_surface_found"))
return (oov_surface, "oov", ())
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="oov", outcome="fell_through", reason="no_oov_learning_invitation_surface"))
else:
if attempts is not None:
attempts.append(DispatchAttempt(source="oov", outcome="skipped", reason="missing_subject"))
return None
def _graph_atom_context(
self,
text: str,
articulation: ArticulationPlan,
*,
region=None,
) -> tuple[tuple[str, ...], bool]:
"""Return ``(graph_atoms, graph_unconstrained)`` for observational telemetry."""
if self.config.output_language != "en":
return ((), True)
graph = build_graph_from_input(text, articulation)
graph_atoms = atoms_for_graph_nodes(graph)
unconstrained = len(graph_atoms) == 0
if region is not None:
unconstrained = unconstrained or getattr(region, "allowed_indices", None) is None
return (graph_atoms, unconstrained)
def _composer_graph_atom_equivalence(
self,
*,
grounding_source: str,
composer_atoms: tuple[str, ...],
graph_atoms: tuple[str, ...],
graph_unconstrained: bool,
):
applicable = grounding_source in {"pack", "teaching"}
return compare_atom_sets(
composer_atoms=composer_atoms,
graph_atoms=graph_atoms,
graph_unconstrained=graph_unconstrained,
applicable=applicable,
)
def _maybe_apply_discourse_planner(
self, text: str, source_tag: str
) -> tuple[str, str] | None:
"""Build and render a :class:`DiscoursePlan` for *text*.
Returns ``(rendered_surface, new_source_tag)`` when the planner
engages and produces more than one move, else ``None``. Callers
own assignment. The returned ``new_source_tag`` is the source
the planner actually used (``"teaching"`` when the plan
contains any teaching fact, else ``"pack"``) so downstream
labels reflect the surface's true provenance — particularly
important when the planner engaged via the compound bypass
(upstream tagged "oov" but rendered output is pack/teaching
content).
Gating discipline (must match both cold-start and warm hooks):
* Returns ``None`` unless ``self.config.discourse_planner`` is True.
* Returns ``None`` unless *source_tag* is one of ``pack`` or
``teaching``. Vault / none / oov / empty paths are not
replaced — the discovery-signal disclosure and the existing
vault-grounded walk surfaces stay intact.
* Returns ``None`` when the classified intent carries no
subject (no head noun ⇒ no grounding bundle to plan over).
* Returns ``None`` when the resulting plan has ≤ 1 move (BRIEF
mode or empty bundle) — render in that case would just
duplicate the existing single-sentence pack-grounded surface.
* Returns ``None`` when the renderer produces an empty string.
"""
# Phase 3 + 4 — reset plan-contemplation findings AND plan
# metrics at the start of every call so they never leak across
# turns; only successfully rendered plans (with contemplation
# enabled) repopulate them.
self._last_plan_findings = ()
self._last_plan_metrics = None
if not self.config.discourse_planner:
return None
from generate.discourse_planner import (
GroundingBundle,
plan_compound_discourse,
plan_discourse,
render_plan,
)
from generate.grounding_accessors import grounding_bundle_for
from generate.intent import (
classify_compound_intent,
classify_response_mode,
)
from generate.intent_bridge import classify_intent_from_input
compound = classify_compound_intent(text)
mode = classify_response_mode(text)
# Compound prompts implicitly request more depth than BRIEF
# can express — a multi-part compound in BRIEF mode produces
# one ANCHOR per part, which on shared-subject compounds
# ("What is X, and why does it matter?") would emit duplicate
# anchor sentences. Upgrade to EXPLAIN so each sub-plan has
# ANCHOR+SUPPORT+RELATION budget and the parts differentiate.
from generate.intent import ResponseMode as _ResponseMode
if compound.is_compound() and mode is _ResponseMode.BRIEF:
mode = _ResponseMode.EXPLAIN
# Fast path: BRIEF mode on a non-compound prompt can never
# emit > 1 move (``_MODE_BUDGETS[BRIEF] = (1, 1)``). The
# downstream ``len(plan.moves) <= 1`` gate would always
# reject — so short-circuit here, BEFORE the expensive
# ``grounding_bundle_for`` query and ``plan_discourse``
# selector logic. This is the load-bearing perf win for
# ``discourse_planner=True`` as the runtime default; without
# it every single-fact prompt pays for a multi-source bundle
# build it can't possibly use. Confirmed empirically:
# ``tests/test_cognition_eval_register_matrix.py`` runtime
# collapsed from ~14 minutes to seconds after this gate
# landed.
if mode is _ResponseMode.BRIEF and not compound.is_compound():
return None
# Standard gate: when upstream grounded the surface in pack or
# teaching, the planner is free to engage.
standard_gate = source_tag in {"pack", "teaching"}
# Compound bypass: when upstream produced an OOV / none surface
# because the flat classifier saw a polluted subject (e.g.
# ``"truth, and why does it matter"``), but the compound
# decomposition reveals at least one pack-resident primary
# part, the substrate exists — the planner engages on the
# decomposed parts rather than the polluted flat surface.
compound_bypass = False
if not standard_gate and compound.is_compound():
primary = compound.primary
if primary.subject:
probe = grounding_bundle_for(primary.subject)
if not probe.is_empty():
compound_bypass = True
if not standard_gate and not compound_bypass:
return None
if compound.is_compound():
bundles = tuple(
grounding_bundle_for(part.subject)
if part.subject
else GroundingBundle()
for part in compound.parts
)
plan = plan_compound_discourse(compound, mode, bundles)
else:
# Use the intent_bridge classifier on single-part prompts to
# preserve the pre-compound behavior exactly.
intent = classify_intent_from_input(text)
if not intent.subject:
return None
bundle = grounding_bundle_for(intent.subject)
plan = plan_discourse(intent, mode, bundle)
if len(plan.moves) <= 1:
return None
# Phase 3 + 4 — plan-level contemplation pre-flight + metrics.
# Read-only, SPECULATIVE-only on the findings side; pure
# measurements on the metrics side. Stores both on the
# runtime for offline miner consumption. Does not mutate the
# plan or block rendering — emits side observations only.
if self.config.discourse_contemplation:
from core.contemplation.plan_metrics import compute_plan_metrics
from core.contemplation.plan_preflight import contemplate_plan
self._last_plan_findings = contemplate_plan(plan)
self._last_plan_metrics = compute_plan_metrics(plan)
else:
self._last_plan_findings = ()
self._last_plan_metrics = None
# Phase 2 — reflective rendering pronominalizes the focus
# subject across consecutive same-subject moves, eliminating
# the mechanical "Truth ... Truth ... Truth ..." cascade the
# Phase 1 flat renderer produced. Deterministic, replayable,
# adds no new content — purely a rendering improvement.
rendered = render_plan(plan, reflective=True)
if not rendered:
return None
from generate.discourse_planner import FactSource
plan_uses_teaching = any(
m.fact is not None and m.fact.source is FactSource.TEACHING
for m in plan.moves
)
new_source = "teaching" if plan_uses_teaching else "pack"
# Phase 5 — emit one articulation observation per engaged turn.
# Gated by both ``discourse_contemplation`` (so metrics +
# findings exist to package) AND the presence of an attached
# sink (so the runtime does no JSON work when nobody is
# listening). Sink errors are NOT swallowed — same fail-fast
# contract as the telemetry sink.
if (
self._articulation_sink is not None
and self.config.discourse_contemplation
and self._last_plan_metrics is not None
):
from chat.articulation_telemetry import (
ArticulationObservation,
format_articulation_observation_jsonl,
prompt_hash,
)
anchor = plan.anchor()
anchor_subject = (
anchor.fact.subject
if anchor is not None and anchor.fact is not None
else (plan.intent.subject or "")
)
import hashlib as _hashlib
plan_substrate_hash = _hashlib.sha256(
plan.to_json().encode("utf-8")
).hexdigest()[:16]
observation = ArticulationObservation(
turn_id=self._articulation_turn_counter,
anchor_subject=anchor_subject,
prompt_hash=prompt_hash(text),
plan_substrate_hash=plan_substrate_hash,
metrics=self._last_plan_metrics.as_dict(),
findings=tuple(
{
"kind": f.kind.value,
"subject": f.subject,
"predicate": f.predicate,
"object": f.object,
}
for f in self._last_plan_findings
),
)
self._articulation_sink.emit(
format_articulation_observation_jsonl(observation)
)
self._articulation_turn_counter += 1
return rendered, new_source
def _stub_response(
self,
field_state: FieldState,
*,
tokens: tuple[str, ...] = (),
pack_grounded_surface: str | None = None,
grounded_source_tag: str = "pack",
pack_semantic_domains: tuple[str, ...] = (),
graph_atoms: tuple[str, ...] = (),
graph_unconstrained: bool = True,
discovery_intent_tag: Any = None,
discovery_intent_subject: str | None = None,
dispatch_trace: DispatchTrace | None = None,
contemplation_result: Any | None = None,
) -> ChatResponse:
zero = np.zeros(field_state.F.shape, dtype=np.float32)
prop = Proposition(
subject="",
predicate="",
object_=None,
surface=_UNKNOWN_DOMAIN_SURFACE,
frame_id="unknown_domain",
subject_versor=zero,
predicate_versor=zero,
object_versor=None,
relation=zero,
)
art = ArticulationPlan(
subject="",
predicate="",
object=None,
surface=_UNKNOWN_DOMAIN_SURFACE,
output_language=self.config.output_language,
frame_id="unknown_domain",
)
safety_ctx = SafetyContext(
field_state=_FieldStateWithVersor(
versor_condition=float(versor_condition(field_state.F)),
),
last_refusal_was_typed=self._last_refusal_was_typed,
identity_manifold_hash_before=self._identity_manifold_hash,
identity_manifold_hash_after=_hash_identity_manifold(self.identity_manifold),
)
safety_verdict = self.safety_check.check(safety_ctx, self.safety_pack)
ethics_ctx = EthicsContext(
alignment_score=0.0,
hedge_threshold_soft=float(
self.identity_manifold.surface_preferences.hedge_threshold_soft
),
hedge_emitted=False,
grounded_in_evidence=False,
disclosure_emitted=True,
)
ethics_verdict = self.ethics_check.check(ethics_ctx, self.ethics_pack)
refusal_surface = build_refusal_surface(
safety_verdict, ethics_verdict, self.ethics_pack,
)
refusal_emitted = refusal_surface is not None
if refusal_emitted:
response_surface = refusal_surface
self._last_refusal_was_typed = True
elif pack_grounded_surface is not None:
response_surface = pack_grounded_surface
if (
self.config.thread_anaphora
and grounded_source_tag in {"pack", "teaching"}
and discovery_intent_subject
and discovery_intent_tag is not None
):
from chat.anaphora import thread_anaphora_prefix
prefix = thread_anaphora_prefix(
self.thread_context,
discovery_intent_subject,
discovery_intent_tag.name.lower(),
grounded_source_tag,
)
if prefix is not None:
response_surface = prefix + response_surface
else:
response_surface = _UNKNOWN_DOMAIN_SURFACE
if pack_grounded_surface is not None and not refusal_emitted:
grounding_source = grounded_source_tag
else:
grounding_source = "none"
# Determine the fallback disposition
if refusal_emitted:
fallback_disposition = "refuse"
elif grounding_source in ("vault", "pack", "teaching"):
fallback_disposition = "commit"
else:
fallback_disposition = "refuse"
from core.epistemic_disclosure.ask_serving import evaluate_served_ask
decision = evaluate_served_ask(
self.config,
contemplation_result,
response_surface,
)
if decision.served and not refusal_emitted:
# ASK serving bypass justification (Q1-C/Q1-D authoritative renderer):
#
# The Q1-C/Q1-D delivery path is the *authoritative grounded renderer*
# for served questions: the question prose was already validated, slot-
# checked, and written to disk by the ASK pass_manager before this point.
# Applying register decoration or the realizer slot-type guard here would
# risk mutating pre-rendered question prose that must reach the user
# exactly as grounded — register suffixes, discourse markers, and guard-
# replacement strings are semantically wrong in an intake-request context.
#
# Safety contract:
# - evaluate_served_ask validates the full Q1-D contract (status,
# requires_review, served=False, answer_binding=None, slot_name) before
# setting decision.served = True.
# - The text consumed here is question_text.strip() from the artifact,
# not constructed in runtime.
# - realizer_guard_status_stub = "ok" is set manually because the pre-
# rendered question surface is Q1-C-grounded and does not require the
# slot-type guard's structural check (which is designed for realizer
# output, not pre-rendered intake prose).
# - tests/test_ask_serving_integration.py::
# test_served_ask_surface_is_consumed_exactly_from_artifact asserts
# that the served surface equals artifact text without any mutation.
response_surface = decision.surface
stub_disposition = decision.disposition.value
stub_epistemic_state = "undetermined"
pre_decoration_surface_stub = response_surface
register_canonical_surface_stub = response_surface
class _DummyDecoration:
surface = decision.surface
variant_id = ""
decoration_stub = _DummyDecoration()
realizer_guard_status_stub = "ok"
realizer_guard_rule_stub = ""
walk_surface_stub = _UNKNOWN_DOMAIN_SURFACE
else:
if contemplation_result is not None:
stub_disposition = decision.disposition.value
else:
stub_disposition = fallback_disposition
stub_epistemic_state = epistemic_state_for_grounding_source(grounding_source).value
# ADR-0075 (C1) — realizer slot-type guard. Runs BEFORE
# register decoration so a register cannot accidentally heal
# an illegal articulation by wrapping it, and BEFORE anchor-
# lens annotation extraction so the lens annotation never
# rides on a guard-rejected surface. On rejection, route to
# the bounded disclosure string and force grounding_source to
# ``"none"`` (an illegal surface is ungrounded by construction).
# The pre-guard candidate is preserved on walk_surface_stub
# for telemetry — the stub path normally leaves walk_surface as
# _UNKNOWN_DOMAIN_SURFACE, so this swap strictly increases
# observability under rejection.
guard_verdict_stub = _check_realizer_surface(
response_surface,
pos_lookup=self._pos_by_surface.get,
)
realizer_guard_status_stub = guard_verdict_stub.status
realizer_guard_rule_stub = guard_verdict_stub.rule_id
walk_surface_stub = _UNKNOWN_DOMAIN_SURFACE
if guard_verdict_stub.status == "rejected":
walk_surface_stub = response_surface
response_surface = _GUARD_DISCLOSURE_SURFACE
grounding_source = "none"
# ADR-0077 (R6) — register layering separation.
register_canonical_surface_stub = response_surface
if grounding_source == "none":
substantive_surface_stub = response_surface
else:
substantive_surface_stub = apply_substantive_register(
response_surface,
self.register_pack,
semantic_domains=pack_semantic_domains,
)
response_surface = substantive_surface_stub
# ADR-0071 (R4) — apply seeded discourse-marker decoration to
# the realized surface AFTER substantive register transforms.
pre_decoration_surface_stub = response_surface
decoration_stub = decorate_surface(
response_surface,
self.register_pack,
turn_idx=len(self.turn_log),
)
response_surface = decoration_stub.surface
register_id_stub = (
"" if self.register_pack.is_unregistered()
else self.register_pack.register_id
)
# ADR-0073d — anchor-lens telemetry. ``id`` reflects the loaded
# pack (empty for UNANCHORED); ``mode_label`` reflects the
# engaged label this turn (empty when the lens didn't fire on
# this turn's lemma). Mode is extracted from the pre-decoration
# surface so register decoration cannot interfere.
anchor_lens_id_stub = (
"" if self.anchor_lens.is_unanchored()
else self.anchor_lens.lens_id
)
anchor_lens_mode_label_stub = _extract_anchor_lens_mode_label(
pre_decoration_surface_stub, anchor_lens_id_stub,
)
atom_equivalence_stub = self._composer_graph_atom_equivalence(
grounding_source=grounding_source,
composer_atoms=pack_semantic_domains,
graph_atoms=graph_atoms,
graph_unconstrained=graph_unconstrained,
)
verdicts_bundle = TurnVerdicts(
identity_score=None,
safety_verdict=safety_verdict,
ethics_verdict=ethics_verdict,
refusal_emitted=refusal_emitted,
hedge_injected=False,
)
stub_normative_clearance = clearance_from_verdicts(verdicts_bundle).value
stub_normative_detail = normative_detail_from_verdicts(verdicts_bundle)
if tokens:
stub_event = TurnEvent(
turn=max(self._context.turn - 1, 0),
input_tokens=tokens,
surface=response_surface,
walk_surface=walk_surface_stub,
articulation_surface=_UNKNOWN_DOMAIN_SURFACE,
dialogue_role="assert",
identity_score=None,
cycle_cost_total=0.0,
vault_hits=0,
versor_condition=float(versor_condition(field_state.F)),
flagged=False,
elaboration=None,
safety_verdict=safety_verdict,
ethics_verdict=ethics_verdict,
verdicts=verdicts_bundle,
grounding_source=grounding_source,
register_id=register_id_stub,
register_variant_id=decoration_stub.variant_id,
anchor_lens_id=anchor_lens_id_stub,
anchor_lens_mode_label=anchor_lens_mode_label_stub,
realizer_guard_status=realizer_guard_status_stub,
realizer_guard_rule=realizer_guard_rule_stub,
register_canonical_surface=register_canonical_surface_stub,
composer_graph_atom_status=atom_equivalence_stub.status,
composer_atom_set_hash=atom_equivalence_stub.composer_atom_set_hash,
graph_atom_set_hash=atom_equivalence_stub.graph_atom_set_hash,
composer_graph_atom_overlap_count=atom_equivalence_stub.overlap_count,
epistemic_state=stub_epistemic_state,
normative_clearance=stub_normative_clearance,
normative_detail=stub_normative_detail,
disposition=stub_disposition,
)
self.turn_log.append(stub_event)
self._emit_turn_event(stub_event)
if discovery_intent_tag is not None:
self._emit_discovery_candidates(
turn_event=stub_event,
intent_tag=discovery_intent_tag,
intent_subject=discovery_intent_subject,
grounding_source=grounding_source,
)
if grounding_source == "oov":
self._emit_oov_candidate(
turn_event=stub_event,
intent_tag=discovery_intent_tag,
token=discovery_intent_subject,
)
self._push_thread_summary(
turn_event=stub_event,
intent_tag=discovery_intent_tag,
intent_subject=discovery_intent_subject,
grounding_source=grounding_source,
surface=response_surface,
)
return ChatResponse(
surface=response_surface,
proposition=prop,
articulation=art,
articulation_surface=_UNKNOWN_DOMAIN_SURFACE,
dialogue_role="assert",
versor_condition=versor_condition(field_state.F),
output_language=self.config.output_language,
frame_pack=self.config.frame_pack,
walk_surface=walk_surface_stub,
salience_top_k=None,
candidates_used=None,
vault_hits=0,
identity_score=None,
character_profile=self.character_profile,
flagged=False,
safety_verdict=safety_verdict,
ethics_verdict=ethics_verdict,
verdicts=verdicts_bundle,
grounding_source=grounding_source,
pre_decoration_surface=pre_decoration_surface_stub,
register_id=register_id_stub,
register_variant_id=decoration_stub.variant_id,
anchor_lens_id=anchor_lens_id_stub,
anchor_lens_mode_label=anchor_lens_mode_label_stub,
realizer_guard_status=realizer_guard_status_stub,
realizer_guard_rule=realizer_guard_rule_stub,
register_canonical_surface=register_canonical_surface_stub,
composer_graph_atom_status=atom_equivalence_stub.status,
composer_atom_set_hash=atom_equivalence_stub.composer_atom_set_hash,
graph_atom_set_hash=atom_equivalence_stub.graph_atom_set_hash,
composer_graph_atom_overlap_count=atom_equivalence_stub.overlap_count,
epistemic_state=stub_epistemic_state,
normative_clearance=stub_normative_clearance,
normative_detail=stub_normative_detail,
refusal_reason=refusal_surface if refusal_emitted else "",
disposition=stub_disposition,
dispatch_trace=dispatch_trace,
)
def chat(
self,
text: str,
max_tokens: int | None = None,
contemplation_result: Any | None = None,
) -> ChatResponse:
self._last_input_text = text # W-013: store for explain_last_turn()
tokens = self._tokenize(text)
filtered = self._apply_oov_policy(tokens)
if not filtered:
raise ValueError("ChatRuntime.chat() received no in-vocabulary tokens.")
# ADR-0090 — unified-ingest path is flag-gated. Default (False)
# preserves the historical probe-then-commit behavior; True
# commits first so the gate and the walk see the same field.
# ``committed`` is materialized eagerly on the unified path and
# lazily on the stub path of the historical flow; the explicit
# ``FieldState | None`` declaration documents that and silences
# Pyright's reportPossiblyUnbound across the conditional flow.
committed: FieldState | None = None
if self.config.unified_ingest:
committed = self._context.commit_ingest(filtered)
committed = self._apply_drive_bias(committed)
gate_query = committed.F
else:
probe_state = self._context.probe_ingest(filtered)
gate_query = probe_state.F
direct_hits = self._context.vault.recall(gate_query, top_k=3)
direct_recall_energy_class = _recall_energy_class_from_hits(direct_hits)
direct_best = max((h["score"] for h in direct_hits), default=0.0)
gate_decision = default_gate.check(
direct_best,
vault=self._context.vault,
query=gate_query,
decomposer=default_decomposer,
)
if gate_decision.fire:
if not self.config.unified_ingest:
committed = self._context.commit_ingest(filtered)
assert committed is not None # set above on both flag paths
empty_result = GenerationResult(tokens=(), final_state=committed, vault_hits=0)
attempts: list[DispatchAttempt] = []
pack_result = self._maybe_pack_grounded_surface(
text, gate_decision.source, attempts=attempts
)
if pack_result is None:
pack_surface = None
pack_source_tag = "none"
pack_semantic_domains: tuple[str, ...] = ()
final_attempts = []
final_attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="no_pack_resident_lemmas"))
final_attempts.append(DispatchAttempt(source="teaching", outcome="fell_through", reason="no_teaching_chains"))
final_attempts.append(DispatchAttempt(source="partial", outcome="fell_through", reason="no_partial_match"))
final_attempts.append(DispatchAttempt(source="oov", outcome="fell_through", reason="no_oov_learning_invitation"))
final_attempts.append(DispatchAttempt(source="universal_disclosure", outcome="admitted", reason="fallback_to_universal_disclosure"))
attempts = final_attempts
selected_source = "universal_disclosure"
else:
pack_surface, pack_source_tag, pack_semantic_domains = pack_result
planned = self._maybe_apply_discourse_planner(
text, pack_source_tag
)
if planned is not None:
pack_surface, pack_source_tag = planned
# ADR-0077 — planner-rendered surfaces are outside
# the gloss DEFINITION/RECALL convivial-expansion
# path; drop the carried semantic_domains so the
# ``append_semantic_domain_clause`` knob is a no-op
# over planner output.
pack_semantic_domains = ()
selected_source = pack_source_tag
dispatch_trace = DispatchTrace(attempts=tuple(attempts), selected=selected_source)
self._context.finalize_turn(
empty_result,
tokens_in=tuple(filtered),
input_versor=committed.F,
dialogue_role="assert",
metadata={
"unknown": True,
"unknown_source": gate_decision.source,
"grounding_source": pack_source_tag if pack_surface else "none",
},
)
# ADR-0148 — post-finalize promotion scan (flag-gated, null-drop when False).
if self.config.vault_promotion_enabled:
self._context.vault.promote_eligible_entries(VaultPromotionPolicy())
discovery_intent_tag = None
discovery_intent_subject: str | None = None
stub_graph_atoms: tuple[str, ...] = ()
stub_graph_unconstrained = True
if (
gate_decision.source == "empty_vault"
and self.config.output_language == "en"
):
from generate.intent_bridge import classify_intent_from_input
_intent = classify_intent_from_input(text)
self._last_intent = _intent # W-013
discovery_intent_tag = _intent.tag
discovery_intent_subject = _intent.subject
stub_articulation = ArticulationPlan(
subject=_intent.subject or "",
predicate="",
object=None,
surface="",
output_language=self.config.output_language,
frame_id="unknown_domain",
)
stub_graph_atoms, stub_graph_unconstrained = self._graph_atom_context(
text,
stub_articulation,
)
return self._checkpointed_response(
self._stub_response(
committed,
tokens=tuple(filtered),
pack_grounded_surface=pack_surface,
grounded_source_tag=pack_source_tag,
pack_semantic_domains=pack_semantic_domains,
graph_atoms=stub_graph_atoms,
graph_unconstrained=stub_graph_unconstrained,
discovery_intent_tag=discovery_intent_tag,
discovery_intent_subject=discovery_intent_subject,
dispatch_trace=dispatch_trace,
contemplation_result=contemplation_result,
)
)
if self.config.unified_ingest:
# ADR-0090 — commit + drive bias already ran before the gate
# check; reuse the same field the gate decided against so the
# walk navigates the manifold position the gate ratified.
assert committed is not None # set in the unified-ingest branch above
field_state = committed
else:
field_state = self._context.commit_ingest(filtered)
field_state = self._apply_drive_bias(field_state)
reference_blade = self._dialogue_reference()
base_proposition = propose(
field_state,
None,
self._context.vocab,
self._frame_registry,
output_lang=self.config.output_language,
)
dialogue_role = _stable_dialogue_role(
classify_dialogue_blade(base_proposition.relation, reference_blade),
raw_text=text,
tokens=tokens,
)
proposition = propose_dialogue(
field_state,
self._context.vault,
self._context.vocab,
self._frame_registry,
reference_blade,
output_lang=self.config.output_language,
)
articulation = realize(proposition, self._context.vocab, output_language=self.config.output_language)
articulation = _prefer_prompt_anchor(articulation, filtered, output_language=self.config.output_language)
self._context.record_dialogue(proposition)
forward_region = None
graph_atoms_main: tuple[str, ...] = ()
graph_unconstrained_main = True
if self.config.output_language == "en":
pre_gen_graph = build_graph_from_input(text, articulation)
graph_atoms_main = atoms_for_graph_nodes(pre_gen_graph)
if self.config.forward_graph_constraint:
forward_region = build_graph_constraint(pre_gen_graph, self._context.vocab)
graph_unconstrained_main = (
len(graph_atoms_main) == 0
or (
forward_region is not None
and getattr(forward_region, "allowed_indices", None) is None
)
)
# W-012 — catch InnerLoopExhaustion so the caller receives a
# typed refusal ChatResponse instead of an unhandled exception.
from generate.exhaustion import InnerLoopExhaustion as _ILE
try:
result = generate(
field_state,
self._context.vocab,
self._context.persona,
max_tokens=self.config.max_tokens if max_tokens is None else max_tokens,
record_trajectory=True,
vault=self._context.vault,
recall_top_k=3 if self.config.allow_cross_language_recall else 0,
output_lang=self.config.output_language,
allow_cross_language_generation=self.config.allow_cross_language_generation,
use_salience=self.config.use_salience,
salience_top_k=self.config.salience_top_k,
inhibition_threshold=self.config.inhibition_threshold,
region=forward_region,
inner_loop_admissibility=self.config.inner_loop_admissibility,
admissibility_threshold=self.config.admissibility_threshold,
admissibility_mode=self.config.admissibility_mode,
admissibility_margin=self.config.admissibility_margin,
stop_tokens=(
frozenset(self.config.stop_tokens)
if self.config.stop_tokens is not None
else None
),
)
except _ILE as _exhaustion_exc:
self._context.finalize_turn(
GenerationResult(tokens=(), final_state=field_state, vault_hits=0),
tokens_in=tuple(filtered),
input_versor=field_state.F,
dialogue_role="assert",
metadata={"exhaustion": True, "refusal_reason": _exhaustion_exc.reason.value},
)
stub = self._stub_response(
field_state,
tokens=tuple(filtered),
contemplation_result=contemplation_result,
)
return self._checkpointed_response(
replace(stub, refusal_reason=_exhaustion_exc.reason.value)
)
# --- Articulation fidelity: replace bare S-P-O join with intent-aware surface ---
# Phase 2: pass proposition so the bridge grounds <pending> obj slots
# from pack-resolved proposition slots (primary) rather than walk
# tokens (supplemental backfill only). walk_tokens still participates
# as a fallback when proposition.object_ is None/empty.
# ADR-0088 Phase B (audit Finding 2, 2026-05-20) — compute
# walk_tokens unconditionally so non-English packs can also
# surface them via ``ChatResponse.recalled_words`` for the
# pipeline's opt-in ``ground_graph`` step. English keeps
# using them for ``articulate_with_intent`` grounding as
# before.
walk_tokens = tuple(
tok for tok in (result.tokens or ()) if tok and tok.isalpha()
)
if self.config.output_language == "en":
intent_surface = articulate_with_intent(
text,
articulation,
walk_tokens,
proposition=proposition,
)
if intent_surface:
articulation = replace(articulation, surface=intent_surface)
# --- end articulation fidelity ---
reasoning_trajectory = _make_trajectory_from_result(result, self._context.turn)
identity_score = self._identity_check.check(reasoning_trajectory, self.identity_manifold)
flagged = identity_score.flagged
cycle_cost = CycleCost(
cycle_index=self._context.turn,
attention_cost=float(result.candidates_used or 0),
inhibition_cost=float(self.config.inhibition_threshold),
digest_cost=0.0,
trajectory_cost=float(len(result.trajectory or ())),
)
self.exertion_meter.record(cycle_cost)
fatigue = self.exertion_meter.fatigue(at_cycle=self._context.turn)
self.character_profile = CharacterProfile.from_manifold(
self.identity_manifold,
drive_summaries={g.axis.name: g.magnitude * (1.0 - fatigue.value) for g in self.drive_gradients},
fatigue_index=fatigue.value,
)
self._context.finalize_turn(
result,
tokens_in=tuple(filtered),
dialogue_role=str(dialogue_role),
)
# ADR-0148 — post-finalize promotion scan (flag-gated, null-drop when False).
if self.config.vault_promotion_enabled:
self._context.vault.promote_eligible_entries(VaultPromotionPolicy())
current_valence = _energy_scalar(getattr(result.final_state, "valence", None))
surface_ctx = self._build_surface_context(identity_score, current_valence)
self._last_valence = current_valence
surface = _terminate_surface(articulation.surface, role=dialogue_role, output_language=self.config.output_language)
articulation = replace(articulation, surface=surface)
sentence_plan: SentencePlan = SentenceAssembler().assemble(
articulation,
result.tokens,
role=dialogue_role,
context=surface_ctx,
)
walk_surface = sentence_plan.surface
vault_hits = int(result.vault_hits)
is_grounded = walk_surface != _UNKNOWN_DOMAIN_SURFACE
hedge_emitted = _surface_contains_hedge(walk_surface, self.identity_manifold)
safety_ctx = SafetyContext(
field_state=_FieldStateWithVersor(
versor_condition=float(versor_condition(result.final_state.F)),
),
last_refusal_was_typed=self._last_refusal_was_typed,
identity_manifold_hash_before=self._identity_manifold_hash,
identity_manifold_hash_after=_hash_identity_manifold(self.identity_manifold),
)
safety_verdict = self.safety_check.check(safety_ctx, self.safety_pack)
ethics_ctx = EthicsContext(
alignment_score=float(getattr(identity_score, "alignment", 0.0)),
hedge_threshold_soft=float(
self.identity_manifold.surface_preferences.hedge_threshold_soft
),
hedge_emitted=hedge_emitted,
grounded_in_evidence=is_grounded,
disclosure_emitted=not is_grounded,
)
ethics_verdict = self.ethics_check.check(ethics_ctx, self.ethics_pack)
refusal_surface = build_refusal_surface(
safety_verdict, ethics_verdict, self.ethics_pack,
)
refusal_emitted = refusal_surface is not None
hedge_injected = False
warm_grounding_source: str | None = None
warm_pack_subject: str | None = None
warm_pack_intent_tag: Any = None
warm_pack_semantic_domains: tuple[str, ...] = ()
attempts: list[DispatchAttempt] = []
selected_source = "vault"
if refusal_emitted:
response_surface = refusal_surface
self._last_refusal_was_typed = True
for src in ("pack", "teaching", "partial", "oov", "universal_disclosure"):
attempts.append(DispatchAttempt(source=src, outcome="skipped", reason="refusal_emitted"))
selected_source = "none"
else:
response_surface = walk_surface
warm_pack_result = self._maybe_pack_grounded_surface(
text, "warm", allow_warm=True, attempts=attempts
)
if warm_pack_result is None:
from generate.intent import IntentTag
from generate.intent_bridge import classify_intent_from_input
_wintent = classify_intent_from_input(text)
# Discovery-signal preservation on warm path: when CAUSE /
# VERIFICATION lacks both a teaching chain and a cross-pack
# chain, the cold path emits the unknown-domain disclosure.
# The warm path must match — fabricating a vault-grounded
# walk fragment ("Work infer.") would mask the very gap
# the discovery layer is meant to surface.
final_attempts = []
final_attempts.append(DispatchAttempt(source="pack", outcome="fell_through", reason="no_pack_resident_lemmas"))
final_attempts.append(DispatchAttempt(source="teaching", outcome="fell_through", reason="no_teaching_chains"))
final_attempts.append(DispatchAttempt(source="partial", outcome="fell_through", reason="no_partial_match"))
final_attempts.append(DispatchAttempt(source="oov", outcome="fell_through", reason="no_oov_learning_invitation"))
if _wintent.tag in (IntentTag.CAUSE, IntentTag.VERIFICATION):
response_surface = _UNKNOWN_DOMAIN_SURFACE
articulation = replace(articulation, surface=_UNKNOWN_DOMAIN_SURFACE)
warm_grounding_source = "none"
final_attempts.append(DispatchAttempt(source="universal_disclosure", outcome="admitted", reason="cause_verification_warm_fallback"))
selected_source = "universal_disclosure"
else:
final_attempts.append(DispatchAttempt(source="universal_disclosure", outcome="skipped", reason="used_walk_surface"))
selected_source = "vault"
attempts = final_attempts
elif warm_pack_result is not None:
warm_pack_surface, warm_grounding_source, warm_pack_semantic_domains = warm_pack_result
if self.config.thread_anaphora and warm_grounding_source in {"pack", "teaching"}:
from chat.anaphora import thread_anaphora_prefix
from generate.intent_bridge import classify_intent_from_input
_wintent = classify_intent_from_input(text)
warm_pack_intent_tag = _wintent.tag
warm_pack_subject = _wintent.subject
if warm_pack_subject and warm_pack_intent_tag is not None:
prefix = thread_anaphora_prefix(
self.thread_context,
warm_pack_subject,
warm_pack_intent_tag.name.lower(),
warm_grounding_source,
)
if prefix is not None:
warm_pack_surface = prefix + warm_pack_surface
response_surface = warm_pack_surface
articulation = replace(articulation, surface=warm_pack_surface)
# Step 5 — discourse planner. Opt-in; engages only on
# pack/teaching-grounded turns where the response mode
# asks for more than a single-sentence brief. When the
# planner returns a multi-move plan, replace the warm
# surface with the deterministic multi-clause rendering.
# BRIEF mode always collapses to a single ANCHOR move so
# the flag-off path stays byte-identical to the existing
# composer.
planned = self._maybe_apply_discourse_planner(
text, warm_grounding_source or ""
)
if planned is not None:
planned_surface, planned_source = planned
response_surface = planned_surface
articulation = replace(articulation, surface=planned_surface)
warm_grounding_source = planned_source
# ADR-0077 — planner-rendered surfaces are outside
# the gloss DEFINITION/RECALL convivial-expansion
# path; drop the carried semantic_domains so the
# ``append_semantic_domain_clause`` knob is a no-op
# over planner output.
warm_pack_semantic_domains = ()
selected_source = warm_grounding_source
if should_inject_hedge(ethics_verdict, self.ethics_pack):
hedge_prefix = build_hedge_prefix(self.identity_manifold)
before = response_surface
response_surface = inject_hedge(response_surface, hedge_prefix)
hedge_injected = response_surface != before
dispatch_trace = DispatchTrace(attempts=tuple(attempts), selected=selected_source)
# ADR-0075 (C1) — realizer slot-type guard (main path). Runs
# AFTER all composer / planner / hedge transformations and
# BEFORE register decoration so a single seam covers every
# articulation path. On rejection: surface is replaced with
# the bounded disclosure string, grounding_source forced to
# ``"none"``, and walk_surface preserves the rejected
# candidate so the manifold-walk evidence is overwritten only
# in the rejection branch (the contract says illegal
# articulation evidence is the relevant telemetry).
guard_verdict_main = _check_realizer_surface(
response_surface,
pos_lookup=self._pos_by_surface.get,
)
realizer_guard_status_main = guard_verdict_main.status
realizer_guard_rule_main = guard_verdict_main.rule_id
if guard_verdict_main.status == "rejected":
walk_surface = response_surface
response_surface = _GUARD_DISCLOSURE_SURFACE
warm_grounding_source = "none"
main_grounding_source = warm_grounding_source or "vault"
recall_energy_class_main = (
direct_recall_energy_class
if main_grounding_source == "vault"
else None
)
if recall_energy_class_main:
from generate.realizer import energy_modulated_surface
try:
ec = EnergyClass(recall_energy_class_main)
except ValueError:
pass
else:
response_surface = energy_modulated_surface(response_surface, ec)
articulation = replace(articulation, surface=response_surface)
# ADR-0077 (R6) — register layering separation (main path). See
# the stub-path equivalent for full semantics: the canonical
# surface is captured pre-substantive so the cognition pipeline
# can hash it for ``trace_hash``, preserving register
# invariance under R6's stronger consumer set. Substantive
# transforms are skipped on ungrounded turns so the bounded
# disclosure stays sacrosanct under terse's drop_articles.
register_canonical_surface_main = response_surface
if main_grounding_source == "none":
substantive_surface_main = response_surface
else:
substantive_surface_main = apply_substantive_register(
response_surface,
self.register_pack,
semantic_domains=warm_pack_semantic_domains,
)
response_surface = substantive_surface_main
# ADR-0071 (R4) — seeded discourse-marker decoration runs AFTER
# substantive register transforms and is the last step before
# TurnEvent is sealed. Applies uniformly to every grounding
# path (vault / pack / teaching / planner / hedge-prefixed).
# No-op for registers with empty marker buckets (UNREGISTERED /
# default_neutral_v1 / terse_v1). Pre-decoration surface is
# preserved separately so the cognition pipeline can hash the
# truth-path surface and trace_hash stays invariant under
# register (ADR-0069 inv C, strengthened by ADR-0077).
pre_decoration_surface_main = response_surface
decoration_main = decorate_surface(
response_surface,
self.register_pack,
turn_idx=len(self.turn_log),
)
response_surface = decoration_main.surface
register_id_main = (
"" if self.register_pack.is_unregistered()
else self.register_pack.register_id
)
# ADR-0073d — anchor-lens telemetry (main path). See stub-path
# comment above for semantics.
anchor_lens_id_main = (
"" if self.anchor_lens.is_unanchored()
else self.anchor_lens.lens_id
)
anchor_lens_mode_label_main = _extract_anchor_lens_mode_label(
pre_decoration_surface_main, anchor_lens_id_main,
)
atom_equivalence_main = self._composer_graph_atom_equivalence(
grounding_source=main_grounding_source,
composer_atoms=warm_pack_semantic_domains,
graph_atoms=graph_atoms_main,
graph_unconstrained=graph_unconstrained_main,
)
verdicts_bundle = TurnVerdicts(
identity_score=identity_score,
safety_verdict=safety_verdict,
ethics_verdict=ethics_verdict,
refusal_emitted=refusal_emitted,
hedge_injected=hedge_injected,
)
main_epistemic = epistemic_state_for_grounding_source(main_grounding_source)
main_epistemic_state = main_epistemic.value
# ADR-0206 — Response Governance Bridge seam (cognition path).
# govern_response is STRICT-only (scaffold), so shape_surface is the
# identity transform and response_surface is returned verbatim —
# byte-identical to the pre-bridge path. This is live wiring, not
# dead code: the response surface now flows through the policy
# consumer, and the ONLY thing keeping it strict is the STRICT
# return value (proven by the live-wiring test). The risk-reward
# widening pathway and the math-serving seam are deferred to their
# own PRs (ADR-0206 §5); wrong==0 is untouched here.
main_reach_policy = govern_response(epistemic_state=main_epistemic)
main_reach_level = main_reach_policy.level.value
response_surface = shape_surface(
main_reach_policy,
committed_surface=response_surface,
decode_state=main_epistemic,
)
from core.epistemic_disclosure.disposition import choose_served_disposition
from core.epistemic_disclosure.disclosure_claim import DisclosureClaim
main_disposition = choose_served_disposition(
epistemic_state=main_epistemic,
limitation=None,
disclosure_claim=DisclosureClaim.NONE,
).value
main_normative_clearance = clearance_from_verdicts(verdicts_bundle).value
main_normative_detail = normative_detail_from_verdicts(verdicts_bundle)
turn_event = TurnEvent(
turn=self._context.turn - 1,
input_tokens=tuple(filtered),
surface=response_surface,
walk_surface=walk_surface,
articulation_surface=articulation.surface,
dialogue_role=str(dialogue_role),
identity_score=identity_score,
cycle_cost_total=cycle_cost.total,
vault_hits=vault_hits,
versor_condition=versor_condition(result.final_state.F),
flagged=flagged,
elaboration=sentence_plan.elaboration,
safety_verdict=safety_verdict,
ethics_verdict=ethics_verdict,
verdicts=verdicts_bundle,
grounding_source=main_grounding_source,
register_id=register_id_main,
register_variant_id=decoration_main.variant_id,
anchor_lens_id=anchor_lens_id_main,
anchor_lens_mode_label=anchor_lens_mode_label_main,
realizer_guard_status=realizer_guard_status_main,
realizer_guard_rule=realizer_guard_rule_main,
register_canonical_surface=register_canonical_surface_main,
composer_graph_atom_status=atom_equivalence_main.status,
composer_atom_set_hash=atom_equivalence_main.composer_atom_set_hash,
graph_atom_set_hash=atom_equivalence_main.graph_atom_set_hash,
composer_graph_atom_overlap_count=atom_equivalence_main.overlap_count,
epistemic_state=main_epistemic_state,
normative_clearance=main_normative_clearance,
normative_detail=main_normative_detail,
reach_level=main_reach_level,
disposition=main_disposition,
)
self.turn_log.append(turn_event)
self._emit_turn_event(turn_event)
self._push_thread_summary(
turn_event=turn_event,
intent_tag=warm_pack_intent_tag,
intent_subject=warm_pack_subject or articulation.subject,
grounding_source=main_grounding_source,
surface=response_surface,
)
return self._checkpointed_response(
ChatResponse(
surface=response_surface,
proposition=proposition,
articulation=articulation,
articulation_surface=articulation.surface,
dialogue_role=dialogue_role,
versor_condition=versor_condition(result.final_state.F),
output_language=self.config.output_language,
frame_pack=self.config.frame_pack,
walk_surface=walk_surface,
salience_top_k=result.salience_top_k,
candidates_used=result.candidates_used,
vault_hits=vault_hits,
identity_score=identity_score,
character_profile=self.character_profile,
flagged=flagged,
recall_energy_class=recall_energy_class_main,
admissibility_trace=result.admissibility_trace,
region_was_unconstrained=result.region_was_unconstrained,
safety_verdict=safety_verdict,
ethics_verdict=ethics_verdict,
verdicts=verdicts_bundle,
grounding_source=main_grounding_source,
pre_decoration_surface=pre_decoration_surface_main,
register_id=register_id_main,
register_variant_id=decoration_main.variant_id,
anchor_lens_id=anchor_lens_id_main,
anchor_lens_mode_label=anchor_lens_mode_label_main,
realizer_guard_status=realizer_guard_status_main,
realizer_guard_rule=realizer_guard_rule_main,
register_canonical_surface=register_canonical_surface_main,
composer_graph_atom_status=atom_equivalence_main.status,
composer_atom_set_hash=atom_equivalence_main.composer_atom_set_hash,
graph_atom_set_hash=atom_equivalence_main.graph_atom_set_hash,
composer_graph_atom_overlap_count=atom_equivalence_main.overlap_count,
recalled_words=walk_tokens,
epistemic_state=main_epistemic_state,
normative_clearance=main_normative_clearance,
normative_detail=main_normative_detail,
reach_level=main_reach_level,
refusal_reason=refusal_surface if refusal_emitted else "",
disposition=main_disposition,
dispatch_trace=dispatch_trace,
)
)
def _unknown_domain_response(self, field_state: FieldState, filtered: list[str]) -> ChatResponse:
return self._stub_response(field_state)
def respond(self, text: str, max_tokens: int | None = None) -> str:
"""Return only the user-facing surface string for *text*.
Convenience wrapper around :meth:`chat` for callers that need
the raw surface without ChatResponse provenance — REPLs, simple
scripts, and the existing test_language_pack_runtime suite.
For audit / telemetry / verdict access, call :meth:`chat`.
"""
return self.chat(text, max_tokens=max_tokens).surface
async def achat(self, text: str, max_tokens: int | None = None) -> ChatResponse:
"""Async-compatible convenience wrapper around :meth:`chat`.
This is a thin async surface; the underlying call is still
synchronous CPU-bound work (versor walk, vault recall, surface
composition). Use this only for integration with asyncio-based
callers that need an awaitable. No real off-thread execution
is performed — if true non-blocking concurrency is required,
wrap calls in :func:`asyncio.to_thread` at the call site.
"""
return self.chat(text, max_tokens=max_tokens)
async def arespond(self, text: str, max_tokens: int | None = None) -> str:
"""Async-compatible convenience wrapper around :meth:`respond`.
Same caveats as :meth:`achat` — wrapper, not true async.
"""
return self.respond(text, max_tokens=max_tokens)
def correct(self, text: str, target_turn: int = -1, max_tokens: int | None = None) -> ChatResponse:
tokens = self._tokenize(text)
filtered = self._apply_oov_policy(tokens)
if not filtered:
raise ValueError("correct() received no in-vocabulary tokens.")
correction_state = inject(filtered, self._context.vocab)
correction_result = self._correction_pass.apply(
self._context.graph,
correction_state.F,
from_turn=target_turn,
)
self._context.apply_corrected_outputs(correction_result.records)
self._emit_correction_event(correction_result, target_turn=target_turn)
regen_tokens = self._context.last_input_tokens
if not regen_tokens:
return self._stub_response(correction_state)
return self.chat(" ".join(regen_tokens), max_tokens=max_tokens)
def _emit_correction_event(
self, correction_result, *, target_turn: int,
) -> None:
"""ADR-0059 — emit one JSONL correction event to the telemetry sink."""
sink = self._telemetry_sink
if sink is None:
return
line = format_correction_event_jsonl(
correction_result,
target_turn=target_turn,
identity_pack_id=self.identity_pack_id,
safety_pack_id=self.safety_pack.pack_id,
ethics_pack_id=self.ethics_pack_id,
)
sink.emit(line)
Reference in a new issue View git blame Copy permalink