c9a644e496
Adds referent tracking, session graph traversal, unknown-domain gating, correction propagation, compositional surface assembly, and regression coverage. Follow-up fixes included before merge: - split probe/commit/finalize turn flow so unknown-domain checks run before current-query vault writes - record real input tokens and input versors for sync and async session paths - return true graph distances from backward walks and consume them in correction decay - synchronize corrected graph outputs into vault-backed recall and live referent state - regenerate correction responses from corrected context rather than correction text - keep coreference pronouns lowercase in question bodies - centralize elaboration-string construction to avoid plan/surface drift - add targeted dialogue fluency regression tests
107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
|
|
from generate.articulation import ArticulationPlan
|
|
from generate.surface import SentenceAssembler, SurfaceContext
|
|
from session.correction import CorrectionPass
|
|
from session.graph import SessionGraph
|
|
from session.referents import ReferentRegistry
|
|
from vault.decompose import default_gate
|
|
|
|
|
|
def _v(slot: int, scale: float = 1.0) -> np.ndarray:
|
|
arr = np.zeros(32, dtype=np.float32)
|
|
arr[slot] = scale
|
|
return arr
|
|
|
|
|
|
def test_backward_walk_returns_true_graph_distances_for_branching_dag() -> None:
|
|
graph = SessionGraph()
|
|
graph.add_turn(0, _v(0), _v(0), ("a",), ("a",), "assert")
|
|
graph.add_turn(1, _v(1), _v(1), ("b",), ("b",), "assert")
|
|
graph.add_turn(2, _v(2), _v(2), ("c",), ("c",), "assert", backward_edges=[0, 1])
|
|
graph.add_turn(3, _v(3), _v(3), ("d",), ("d",), "assert", backward_edges=[2])
|
|
|
|
walked = graph.backward_walk(3)
|
|
|
|
assert [(dist, node.turn_idx) for dist, node in walked] == [(1, 2), (2, 0), (2, 1)]
|
|
|
|
|
|
def test_correction_pass_uses_graph_distance_not_bfs_ordinal() -> None:
|
|
graph = SessionGraph()
|
|
base = _v(0)
|
|
graph.add_turn(0, base, base, ("a",), ("a",), "assert")
|
|
graph.add_turn(1, base, base, ("b",), ("b",), "assert")
|
|
graph.add_turn(2, base, base, ("c",), ("c",), "assert", backward_edges=[0, 1])
|
|
graph.add_turn(3, base, base, ("d",), ("d",), "assert", backward_edges=[2])
|
|
|
|
result = CorrectionPass(min_alignment=0.0).apply(graph, base, from_turn=3)
|
|
by_turn = {record.turn_idx: record.graph_distance for record in result.records}
|
|
|
|
assert by_turn[3] == 0
|
|
assert by_turn[2] == 1
|
|
assert by_turn[0] == 2
|
|
assert by_turn[1] == 2
|
|
|
|
|
|
def test_referent_registry_tracks_only_currently_consumed_sources() -> None:
|
|
registry = ReferentRegistry()
|
|
registry.register("light", _v(4), turn=7)
|
|
|
|
assert registry.resolve(["what", "is", "it"]) == ["what", "is", "light"]
|
|
assert registry.consumed_turns() == [7]
|
|
assert registry.consumed_slots() == {"neut_sg": 7}
|
|
|
|
assert registry.resolve(["light"]) == ["light"]
|
|
assert registry.consumed_turns() == []
|
|
assert registry.consumed_slots() == {}
|
|
|
|
|
|
def test_surface_coreference_keeps_question_pronoun_lowercase() -> None:
|
|
plan = ArticulationPlan(
|
|
subject="Light",
|
|
predicate="form",
|
|
object=None,
|
|
surface="Light form",
|
|
output_language="en",
|
|
frame_id="test",
|
|
)
|
|
ctx = SurfaceContext(active_referent_surface="Light", active_referent_slot="neut_sg")
|
|
|
|
sentence = SentenceAssembler().assemble(plan, (), role="question", context=ctx)
|
|
|
|
assert sentence.surface == "Given that Light, does it form?"
|
|
|
|
|
|
def test_surface_elaboration_matches_rendered_elaboration_string() -> None:
|
|
plan = ArticulationPlan(
|
|
subject="light",
|
|
predicate="reveals",
|
|
object="truth",
|
|
surface="light reveals truth",
|
|
output_language="en",
|
|
frame_id="test",
|
|
)
|
|
ctx = SurfaceContext(valence_delta=-1.0)
|
|
|
|
sentence = SentenceAssembler().assemble(
|
|
plan,
|
|
("mercy", "justice"),
|
|
role="elaborate",
|
|
context=ctx,
|
|
)
|
|
|
|
assert sentence.elaboration == "mercy but justice"
|
|
assert "mercy but justice" in sentence.surface
|
|
|
|
|
|
def test_unknown_gate_fires_on_empty_vault_without_self_storage() -> None:
|
|
class EmptyVault:
|
|
def __len__(self) -> int:
|
|
return 0
|
|
|
|
decision = default_gate.check(0.0, vault=EmptyVault(), query=_v(0))
|
|
|
|
assert decision.fire is True
|
|
assert decision.source == "empty_vault"
|