generated from coulomb/repo-seed
96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
from phase_memory.contracts import graph_from_markitect
|
|
from phase_memory.models import MemoryEvent, MemoryPath, MemoryPathState
|
|
from phase_memory.retrieval import (
|
|
WordCountTokenEstimator,
|
|
activation_quality_report,
|
|
plan_neighborhood_activation,
|
|
retrieve_graph_neighborhood,
|
|
select_event_path,
|
|
)
|
|
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def _graph():
|
|
return graph_from_markitect(json.loads((FIXTURES / "memory-graph.json").read_text(encoding="utf-8"))).value
|
|
|
|
|
|
def test_retrieve_graph_neighborhood_is_stable_and_filterable() -> None:
|
|
candidates = retrieve_graph_neighborhood(
|
|
_graph(),
|
|
seed_node_ids=("decision.boundary",),
|
|
max_hops=1,
|
|
edge_kinds=("governs",),
|
|
direction="out",
|
|
)
|
|
|
|
assert [candidate.node_id for candidate in candidates] == ["decision.boundary", "artifact.profile"]
|
|
assert candidates[0].reasons[:2] == ("graph_distance:0", "explicit_priority")
|
|
|
|
|
|
def test_event_path_selection_respects_path_state_and_budget() -> None:
|
|
events = (
|
|
MemoryEvent("event.a", "user_turn"),
|
|
MemoryEvent("event.b", "agent_turn"),
|
|
MemoryEvent("event.c", "tool_call"),
|
|
)
|
|
active = MemoryPath("path.active", event_ids=("event.a", "event.b", "event.c"))
|
|
abandoned = MemoryPath("path.abandoned", event_ids=("event.a",), state=MemoryPathState.ABANDONED)
|
|
|
|
assert select_event_path(events, active, max_events=2) == ("event.a", "event.b")
|
|
assert select_event_path(events, abandoned, max_events=2) == ()
|
|
assert select_event_path(events, abandoned, max_events=2, include_inactive=True) == ("event.a",)
|
|
|
|
|
|
def test_neighborhood_activation_uses_retrieval_order_and_estimator_label() -> None:
|
|
plan, candidates = plan_neighborhood_activation(
|
|
_graph(),
|
|
seed_node_ids=("decision.boundary",),
|
|
max_hops=1,
|
|
max_items=2,
|
|
max_tokens=40,
|
|
profile_id="phase-memory-fixture-profile",
|
|
)
|
|
|
|
assert [candidate.node_id for candidate in candidates][:2] == ["decision.boundary", "artifact.profile"]
|
|
assert plan.selected_node_ids[:2] == ("decision.boundary", "artifact.profile")
|
|
assert plan.selection["metadata"]["estimator"] == "WordCountTokenEstimator"
|
|
|
|
|
|
def test_token_estimator_accounts_for_nodes_and_events() -> None:
|
|
graph = _graph()
|
|
estimator = WordCountTokenEstimator()
|
|
|
|
assert estimator.estimate_node(graph.nodes[0]) == 9
|
|
assert estimator.estimate_event(graph.events[0]) >= 2
|
|
|
|
|
|
def test_activation_quality_report_is_deterministic() -> None:
|
|
plan, _ = plan_neighborhood_activation(
|
|
_graph(),
|
|
seed_node_ids=("decision.boundary",),
|
|
max_hops=1,
|
|
max_items=1,
|
|
max_tokens=20,
|
|
profile_id="phase-memory-fixture-profile",
|
|
)
|
|
|
|
report = activation_quality_report(
|
|
plan,
|
|
expected_node_ids=("decision.boundary", "artifact.profile"),
|
|
policy_denied_node_ids=("artifact.profile",),
|
|
)
|
|
|
|
expected = json.loads((FIXTURES / "activation-quality-report.json").read_text(encoding="utf-8"))
|
|
assert report["selected_expected_nodes"] == ["decision.boundary"]
|
|
assert report["omitted_required_nodes"] == ["artifact.profile"]
|
|
assert report["policy_denied_required_nodes"] == ["artifact.profile"]
|
|
assert report["token_budget_utilization"] == 0.45
|
|
assert report["source_span_coverage"] == 1.0
|
|
assert report["explanation_coverage"] == 1.0
|
|
assert report == expected
|