Codex
82e2c91f42
Publish documentation / publish (push) Successful in 1m27s
Add PostgreSQL-backed memory, editable evidence with source review and activation, and human-approved archive repairs across the harness, API, and UI. Include migrations, deployment support, regression coverage, and validation documentation. Refresh permissions from validated session roles so existing administrator logins can access newly deployed archive management features.
122 lines
4.9 KiB
Python
122 lines
4.9 KiB
Python
"""Deterministic retrieval policy tests, separate from actual embedding/Qdrant recovery."""
|
|
|
|
from contextlib import nullcontext
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from tht.memory.models import Card, MemoryNotFound
|
|
from tht.memory.retrieval import RecallScope, expand_and_rank
|
|
|
|
|
|
def card(identity, **values):
|
|
return Card(id=identity, workspace_id="sales", family="domain_clarification",
|
|
subject=identity, scope="Sales", origin="manual", revision="r", indexed=True,
|
|
created_at=datetime.now(UTC), updated_at=datetime.now(UTC), **values)
|
|
|
|
|
|
class Archive:
|
|
def __init__(self, *cards):
|
|
self.cards = {c.id: c for c in cards}
|
|
self.reads = []
|
|
|
|
def get(self, identity):
|
|
self.reads.append(identity)
|
|
if identity not in self.cards:
|
|
raise MemoryNotFound(identity)
|
|
return self.cards[identity]
|
|
|
|
def operation(self):
|
|
return nullcontext(self)
|
|
|
|
|
|
def hit(identity, **metadata):
|
|
return SimpleNamespace(ref=identity, metadata={"memory_revision": "r", "memory_format": 2,
|
|
**metadata})
|
|
|
|
|
|
def link(identity):
|
|
return {"target_id": identity, "meaning": "Requires the grain clarification"}
|
|
|
|
|
|
def rank(repo, hits, **kwargs):
|
|
return expand_and_rank(repo, hits, scope=kwargs.pop("scope", RecallScope()),
|
|
family="domain_clarification", excluded=kwargs.pop("excluded", set()), top=100, **kwargs)
|
|
|
|
|
|
def test_link_only_candidates_cycles_duplicates_depth_and_current_content():
|
|
repo = Archive(card("a", links=[link("b")]),
|
|
card("b", detail="Current correction", links=[link("a"), link("c")]),
|
|
card("c", links=[link("d")]), card("d"))
|
|
results = rank(repo, [hit("a"), hit("a")])
|
|
assert [r.card.id for r in results] == ["a", "b", "c"]
|
|
assert results[1].card.detail == "Current correction"
|
|
assert results[2].path == ("a", "b", "c")
|
|
assert len(repo.reads) == len(set(repo.reads)) == 3
|
|
assert rank(repo, [hit("a")]) == results # duplicate seeds do not boost a score
|
|
|
|
|
|
def test_direct_and_graph_candidates_are_reranked_together_without_cycle_boost():
|
|
repo = Archive(card("a", links=[link("c")]), card("b"), card("c", links=[link("a")]))
|
|
results = rank(repo, [hit("a"), hit("b"), hit("c")])
|
|
assert [r.card.id for r in results] == ["a", "c", "b"]
|
|
assert len(results) == 3
|
|
|
|
|
|
@pytest.mark.parametrize("invalid", ["missing", "pending", "excluded", "wrong_scope", "family"])
|
|
def test_ineligible_link_targets_cannot_be_returned_or_used_as_bridges(invalid):
|
|
target = card("b", links=[link("c")])
|
|
if invalid == "pending":
|
|
target.indexed = False
|
|
if invalid == "wrong_scope":
|
|
target.scope = "Purchases"
|
|
if invalid == "family":
|
|
target.family = "sql_rule"
|
|
repo = Archive(card("a", links=[link("b")]), card("c"),
|
|
*([] if invalid == "missing" else [target]))
|
|
results = rank(repo, [hit("a")], scope=RecallScope(scope="Sales"),
|
|
excluded={"b"} if invalid == "excluded" else set())
|
|
assert [r.card.id for r in results] == ["a"]
|
|
|
|
|
|
@pytest.mark.parametrize("metadata", [{"memory_revision": "old"}, {"memory_format": 1}])
|
|
def test_stale_seeds_do_not_expand(metadata):
|
|
repo = Archive(card("a", links=[link("b")]), card("b"))
|
|
assert rank(repo, [hit("a", **metadata)]) == []
|
|
|
|
|
|
def test_physical_scope_matches_one_dependency_and_includes_workspace_and_parent_rules():
|
|
scope = RecallScope(database="sales", schema_name="public", table="orders", column="id")
|
|
assert scope.matches(card("global"))
|
|
assert scope.matches(card("database", dependencies=[{"database": "sales"}]))
|
|
assert scope.matches(card("table", dependencies=[{"database": "sales",
|
|
"schema_name": "public", "table": "orders"}]))
|
|
assert not scope.matches(card("split", dependencies=[
|
|
{"database": "sales", "schema_name": "public", "table": "orders", "column": "amount"},
|
|
{"database": "purchases", "schema_name": "public", "table": "orders", "column": "id"},
|
|
]))
|
|
assert not scope.matches(card("schema", dependencies=[
|
|
{"database": "sales", "schema_name": "audit", "table": "orders", "column": "id"},
|
|
]))
|
|
|
|
|
|
def test_scope_and_concepts_are_explicit_and_combined():
|
|
scope = RecallScope(scope="Sales", concepts=["orders", "grain"])
|
|
assert scope.matches(card("a", concepts=["orders", "grain"]))
|
|
assert not scope.matches(card("b", concepts=["orders"]))
|
|
with pytest.raises(ValueError):
|
|
RecallScope(column="id")
|
|
|
|
|
|
def test_fanout_and_total_visits_are_bounded():
|
|
cards = [card(f"c{i:04}", links=[link(f"c{j:04}") for j in range(i+1, i+101)])
|
|
for i in range(500)]
|
|
repo = Archive(*cards)
|
|
results = rank(repo, [hit("c0000")])
|
|
assert "c0100" not in {r.card.id for r in results}
|
|
assert len(repo.reads) <= 200
|
|
repo.reads.clear()
|
|
rank(repo, [hit(c.id) for c in cards[:100]])
|
|
assert len(repo.reads) <= 200
|