User
d6cdffea62
Publish documentation / publish (push) Successful in 34s
Cap the embedded shell at its portal container height so steering and stop controls remain accessible. Skip vector retrieval for an empty authoritative Memory archive and compute SQL-rule embeddings lazily. Validated with 54 Memory tests, 90 frontend tests, five browser scenarios, frontend and Docker builds, and a read-only comparison against the real empty Memory archive.
1103 lines
55 KiB
Python
1103 lines
55 KiB
Python
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
import requests
|
|
from sqlalchemy import create_engine, text
|
|
from testcontainers.core.container import DockerContainer
|
|
from testcontainers.core.waiting_utils import wait_for_logs
|
|
from testcontainers.postgres import PostgresContainer
|
|
|
|
from tht.adapters.vector.qdrant import QdrantVectorStore
|
|
from tht.memory.migrate import migrate
|
|
from tht.memory.models import (
|
|
CardInput,
|
|
CardQuery,
|
|
MemoryConflict,
|
|
MemoryForbidden,
|
|
MemoryNotFound,
|
|
MemoryUnavailable,
|
|
)
|
|
from tht.memory.repository import MemoryRepository
|
|
from tht.memory.service import MemoryService
|
|
from tht.session.models import PrincipalContext, SessionManifest, SessionSnapshot
|
|
from tht.vectorstore.store import VectorHit
|
|
|
|
pytestmark = pytest.mark.l0
|
|
ADMIN = PrincipalContext(issuer="test", subject="admin", is_admin=True)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def database_url():
|
|
with PostgresContainer("postgres:16-alpine") as postgres:
|
|
url = postgres.get_connection_url()
|
|
migrate(url)
|
|
migrate(url)
|
|
yield url
|
|
|
|
|
|
class Vectors:
|
|
def __init__(self):
|
|
self.records = {}
|
|
self.fail = False
|
|
|
|
def upsert(self, collection, rows):
|
|
if self.fail:
|
|
raise RuntimeError("injected vector outage")
|
|
for row in rows:
|
|
self.records[row.record.ref] = row.record
|
|
return len(rows)
|
|
|
|
def delete_memory_records(self, keys):
|
|
if self.fail:
|
|
raise RuntimeError("injected vector outage")
|
|
for ref, record in list(self.records.items()):
|
|
if record.id in keys:
|
|
del self.records[ref]
|
|
|
|
def delete_kinds(self, collection, kinds):
|
|
if self.fail:
|
|
raise RuntimeError("injected vector outage")
|
|
self.records.clear()
|
|
|
|
def prepare_memory_index(self):
|
|
if self.fail:
|
|
raise RuntimeError("injected vector outage")
|
|
|
|
def search(self, embedding, top_n, kinds, **kwargs):
|
|
return [VectorHit(id=r.id, kind=r.kind, ref=r.ref, title=r.title,
|
|
content=r.content, metadata=r.metadata, similarity=0.9)
|
|
for r in self.records.values() if r.kind in kinds][:top_n]
|
|
|
|
|
|
class Embeddings:
|
|
def embed_documents(self, docs):
|
|
return [[1.0, 0.0, 0.0] for _ in docs]
|
|
|
|
def embed_query(self, query):
|
|
return [1.0, 0.0, 0.0]
|
|
|
|
|
|
@pytest.fixture
|
|
def service(database_url):
|
|
vectors = Vectors()
|
|
s = MemoryService(MemoryRepository(database_url, "test-"+uuid4().hex), ADMIN,
|
|
store_factory=lambda: vectors, embedder_factory=Embeddings)
|
|
yield s, vectors
|
|
s.close()
|
|
|
|
|
|
def card(**overrides):
|
|
return CardInput.model_validate({"family": "domain_clarification", "subject": "Order",
|
|
"detail": "All lines shipped", "scope": "Sales", **overrides})
|
|
|
|
|
|
def recall(service, vectors, **kwargs):
|
|
return service.recall("Order", searcher=vectors, embedder=Embeddings(), **kwargs)
|
|
|
|
|
|
def test_real_qdrant_update_delete_rebuild_and_reference_isolation(database_url):
|
|
with DockerContainer("qdrant/qdrant:v1.18.2").with_exposed_ports(6333) as container:
|
|
wait_for_logs(container, "Qdrant HTTP listening on 6333")
|
|
url = f"http://{container.get_container_host_ip()}:{container.get_exposed_port(6333)}"
|
|
workspace = "qdrant-" + uuid4().hex
|
|
store = QdrantVectorStore(base_url=url, workspace_id=workspace,
|
|
collections={"reference": workspace+"-reference", "memory": workspace+"-memory"},
|
|
expected_dimension=3)
|
|
service = MemoryService(MemoryRepository(database_url, workspace), ADMIN,
|
|
store_factory=lambda: store, embedder_factory=Embeddings)
|
|
|
|
class Searcher:
|
|
def search(self, embedding, top_n, kinds, **kwargs):
|
|
return store.search(["memory"], embedding, limit=top_n, kinds=kinds, **kwargs)
|
|
|
|
try:
|
|
identity = service.save(card())["id"]
|
|
assert recall(service, Searcher())[0]["id"] == identity
|
|
assert service.save(card(detail="Corrected"), identity)["indexed"]
|
|
assert recall(service, Searcher())[0]["detail"] == "Corrected"
|
|
assert service.save(card(family="solved_question", question="Orders?", sql="select 2"),
|
|
identity)["indexed"]
|
|
assert recall(service, Searcher()) == []
|
|
assert recall(service, Searcher(), solved=True)[0]["sql"] == "select 2"
|
|
assert service.delete(identity)["indexed"]
|
|
assert recall(service, Searcher(), solved=True) == []
|
|
surviving = service.save(card(subject="Keep"))["id"]
|
|
# Reference preprocessing and repeated rebuilds never resurrect removed cards.
|
|
store.clear_reference()
|
|
service.rebuild()
|
|
service.rebuild()
|
|
assert [c["id"] for c in recall(service, Searcher())] == [surviving]
|
|
response = requests.post(f"{url}/collections/{workspace}-memory/points/count",
|
|
json={"exact": True}, timeout=5)
|
|
response.raise_for_status()
|
|
assert response.json()["result"]["count"] == 1
|
|
response = requests.delete(f"{url}/collections/{workspace}-memory", timeout=5)
|
|
response.raise_for_status()
|
|
assert all(result["indexed"] for result in service.rebuild())
|
|
assert [c["id"] for c in recall(service, Searcher())] == [surviving]
|
|
finally:
|
|
service.close()
|
|
|
|
|
|
def test_admin_cli_uses_protected_snapshot_without_dwh(database_url, tmp_path, monkeypatch):
|
|
import json
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app
|
|
|
|
monkeypatch.setenv("THT_CATALOG_RUNTIME_DATABASE_URL", database_url)
|
|
monkeypatch.setenv("THT_PRINCIPAL_ISSUER", "test")
|
|
monkeypatch.setenv("THT_PRINCIPAL_SUBJECT", "admin")
|
|
monkeypatch.setenv("THT_PRINCIPAL_IS_ADMIN", "true")
|
|
workspace = "cli-" + uuid4().hex
|
|
snapshot = tmp_path / "request.json"
|
|
snapshot.write_text(json.dumps({"action": "create", "runtime": {},
|
|
"request": {"card": card().model_dump()}}))
|
|
command = ["memory", "admin", "--workspace", workspace, "-c", str(snapshot)]
|
|
created = CliRunner().invoke(app, command)
|
|
assert created.exit_code == 0, created.output
|
|
result = json.loads(created.stdout)
|
|
assert result["saved"] and not result["indexed"]
|
|
snapshot.write_text(json.dumps({"action": "list", "runtime": {}, "request": {}}))
|
|
listed = CliRunner().invoke(app, command)
|
|
assert listed.exit_code == 0
|
|
assert json.loads(listed.stdout)["items"][0]["id"] == result["id"]
|
|
|
|
|
|
def test_persistence_and_all_families_after_reopen(service, database_url):
|
|
s, _vectors = service
|
|
values = [card(), card(family="sql_rule", rationale="Use both keys"),
|
|
card(family="explained_error", rationale="Avoid multiplying the header total"),
|
|
card(family="solved_question", question="Total orders?", sql="select 1")]
|
|
for value in values:
|
|
result = s.save(value)
|
|
assert result["saved"] and result["indexed"]
|
|
assert result["card"]["session_id"] is None
|
|
reopened = MemoryRepository(database_url, s.repository.workspace_id)
|
|
try:
|
|
page = reopened.list(CardQuery())
|
|
assert page["total"] == 4
|
|
assert {c["family"] for c in page["items"]} == {v.family for v in values}
|
|
finally:
|
|
reopened.close()
|
|
|
|
|
|
def test_filters_precede_pagination_and_literal_text(service):
|
|
s, _ = service
|
|
for n in range(5):
|
|
s.save(card(subject=f"Order {n}"))
|
|
wanted = s.save(card(subject="100%_special", concepts=["target"], dependencies=[{
|
|
"database": "dwh", "schema_name": "sales", "table": "orders", "column": "id",
|
|
}]))
|
|
page = s.list(CardQuery(q="%_", concept="target", database="dwh", table="orders",
|
|
column="id", page_size=1, origin="manual"))
|
|
assert page["total"] == 1
|
|
assert page["items"][0]["id"] == wanted["id"]
|
|
assert s.list(CardQuery(column="missing"))["total"] == 0
|
|
|
|
|
|
def test_invalid_link_rolls_back_card_and_dependencies(service):
|
|
s, _ = service
|
|
first = s.save(card())["card"]
|
|
with pytest.raises(MemoryConflict):
|
|
s.save(card(subject="Changed", dependencies=[{"database": "new"}],
|
|
links=[{"target_id": "missing", "meaning": "Related"}]), first["id"])
|
|
assert s.get(first["id"])["subject"] == first["subject"]
|
|
assert s.get(first["id"])["dependencies"] == []
|
|
assert s.get(first["id"])["indexed"]
|
|
|
|
|
|
def test_delete_cascades_links_only_and_preserves_other_cards(service):
|
|
s, _ = service
|
|
b = s.save(card(subject="B"))["id"]
|
|
a = s.save(card(subject="A", links=[{"target_id": b, "meaning": "Clarifies"}]))["id"]
|
|
s.delete(b)
|
|
assert s.get(a)["links"] == []
|
|
with pytest.raises(MemoryNotFound):
|
|
s.get(b)
|
|
assert s.list(CardQuery())["total"] == 1
|
|
|
|
|
|
def test_cross_workspace_read_write_links_and_retry_are_rejected(service, database_url):
|
|
s, _ = service
|
|
identity = s.save(card())["id"]
|
|
other = MemoryService(MemoryRepository(database_url, "other-"+uuid4().hex), ADMIN,
|
|
store_factory=Vectors, embedder_factory=Embeddings)
|
|
try:
|
|
for action in [lambda: other.get(identity), lambda: other.delete(identity),
|
|
lambda: other.save(card(), identity), lambda: other.retry(identity)]:
|
|
with pytest.raises(MemoryNotFound):
|
|
action()
|
|
with pytest.raises(MemoryConflict):
|
|
other.save(card(links=[{"target_id": identity, "meaning": "Invalid"}]))
|
|
assert other.list(CardQuery())["total"] == 0
|
|
finally:
|
|
other.close()
|
|
|
|
|
|
def test_non_admin_cannot_use_any_administrative_action(service):
|
|
s, _ = service
|
|
identity = s.save(card())["id"]
|
|
s.principal = PrincipalContext(issuer="test", subject="user")
|
|
for action in [lambda: s.list(CardQuery()), lambda: s.get(identity), lambda: s.save(card()),
|
|
lambda: s.delete(identity), lambda: s.retry(identity), s.pending, s.rebuild]:
|
|
with pytest.raises(MemoryForbidden):
|
|
action()
|
|
|
|
|
|
@pytest.mark.parametrize("solved", [False, True])
|
|
def test_outage_invalidates_stale_recall_then_retry_uses_current_content(service, solved):
|
|
s, vectors = service
|
|
value = card(family="solved_question", question="Orders?", sql="select 1") if solved else card()
|
|
identity = s.save(value)["id"]
|
|
assert recall(s, vectors, solved=solved)
|
|
vectors.fail = True
|
|
updated = value.model_copy(update={"detail": "Current", "sql": "select 2" if solved else ""})
|
|
outcome = s.save(updated, identity)
|
|
assert outcome["saved"] and not outcome["indexed"]
|
|
assert s.get(identity)["detail"] == "Current"
|
|
assert recall(s, vectors, solved=solved) == []
|
|
vectors.fail = False
|
|
assert s.retry(identity)["indexed"]
|
|
result = recall(s, vectors, solved=solved)[0]
|
|
assert result["sql"] == "select 2" if solved else result["detail"] == "Current"
|
|
|
|
|
|
def test_delete_outage_survives_reopen_and_cannot_resurrect(service, database_url):
|
|
s, vectors = service
|
|
identity = s.save(card())["id"]
|
|
vectors.fail = True
|
|
assert not s.delete(identity)["indexed"]
|
|
assert recall(s, vectors) == []
|
|
reopened = MemoryService(MemoryRepository(database_url, s.repository.workspace_id), ADMIN,
|
|
store_factory=lambda: vectors, embedder_factory=Embeddings)
|
|
try:
|
|
assert reopened.pending()[0]["card_id"] == identity
|
|
vectors.fail = False
|
|
assert reopened.retry(identity)["indexed"]
|
|
assert reopened.retry(identity)["indexed"]
|
|
reopened.rebuild()
|
|
assert vectors.records == {}
|
|
assert reopened.list(CardQuery())["total"] == 0
|
|
finally:
|
|
reopened.close()
|
|
|
|
|
|
def test_crash_after_sql_commit_can_be_recovered(service, database_url):
|
|
s, vectors = service
|
|
with s.repository.operation() as repo:
|
|
identity = repo.save(card())
|
|
assert s.pending()[0]["card_id"] == identity
|
|
assert recall(s, vectors) == []
|
|
assert s.retry(identity)["indexed"]
|
|
assert recall(s, vectors)[0]["id"] == identity
|
|
|
|
|
|
def test_failed_archive_never_becomes_empty_success(service):
|
|
s, vectors = service
|
|
s.repository = MemoryRepository("postgresql+psycopg2://invalid@127.0.0.1:1/absent", "w")
|
|
with pytest.raises(MemoryUnavailable):
|
|
s.list(CardQuery())
|
|
with pytest.raises(MemoryUnavailable):
|
|
recall(s, vectors)
|
|
|
|
|
|
def test_runtime_role_enforces_workspace_even_without_sql_predicate(service):
|
|
s, _ = service
|
|
s.save(card())
|
|
with s.repository.transaction() as c:
|
|
c.execute(text("SELECT set_config('thoth.memory_workspace', 'foreign', true)"))
|
|
assert c.execute(text("SELECT count(*) FROM thoth_memory.cards")).scalar_one() == 0
|
|
|
|
|
|
def test_solved_producer_receipt_preserves_curated_edits_and_deletion(service):
|
|
from tht.decisions import DecisionRecord
|
|
s, vectors = service
|
|
snapshot = SessionSnapshot(manifest=SessionManifest(
|
|
id="session", created_at=datetime.now(UTC), question="Orders?", database="dwh",
|
|
schema="sales", workspace_id=s.repository.workspace_id, author="admin", status="finalized",
|
|
), artifacts={"sql_final": "select 1"}, decisions=[DecisionRecord(
|
|
seq=1, ts=datetime.now(UTC), type="sql_approved", subject="sql", detail="", rationale="",
|
|
)])
|
|
result = s.save_solved(snapshot, {"orders"})
|
|
identity = result["id"]
|
|
assert result["card"]["dependencies"] == [{
|
|
"database": "dwh", "schema_name": "sales", "table": "orders", "column": "",
|
|
}]
|
|
assert s.save_solved(snapshot)["id"] == identity
|
|
s.save(card(family="solved_question", question="Orders?", sql="select 2"), identity)
|
|
s.retry_solved(snapshot)
|
|
assert recall(s, vectors, solved=True)[0]["sql"] == "select 2"
|
|
s.delete(identity)
|
|
s.save_solved(snapshot)
|
|
s.retry_solved(snapshot)
|
|
s.rebuild()
|
|
assert s.list(CardQuery())["total"] == 0
|
|
|
|
|
|
def test_runtime_role_has_no_ddl_rights(service):
|
|
from sqlalchemy.exc import ProgrammingError
|
|
s, _ = service
|
|
engine = create_engine(s.repository.engine.url)
|
|
try:
|
|
with engine.connect() as c:
|
|
c.execute(text("SET ROLE thoth_memory_runtime"))
|
|
with pytest.raises(ProgrammingError):
|
|
c.execute(text("CREATE TABLE thoth_memory.forbidden (id int)"))
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def test_incompatible_schema_is_an_operational_error(service):
|
|
s, _ = service
|
|
engine = create_engine(s.repository.engine.url)
|
|
try:
|
|
with engine.begin() as c:
|
|
c.execute(text("INSERT INTO thoth_memory.migrations VALUES ('future.sql','future')"))
|
|
with pytest.raises(MemoryUnavailable, match="incompatible"):
|
|
s.list(CardQuery())
|
|
finally:
|
|
with engine.begin() as c:
|
|
c.execute(text("DELETE FROM thoth_memory.migrations WHERE version='future.sql'"))
|
|
engine.dispose()
|
|
|
|
|
|
def test_dense_m1_projection_is_pending_until_explicit_retry(service):
|
|
s, vectors = service
|
|
identity = s.save(card())["id"]
|
|
with s.repository.transaction() as c:
|
|
c.execute(text("UPDATE thoth_memory.projections SET format=1 WHERE card_id=:id"),
|
|
{"id": identity})
|
|
assert not s.get(identity)["indexed"]
|
|
assert [p["card_id"] for p in s.pending()] == [identity]
|
|
assert recall(s, vectors) == []
|
|
assert s.retry(identity)["indexed"]
|
|
assert recall(s, vectors)[0]["id"] == identity
|
|
|
|
|
|
def review_snapshot(service, proposals=()):
|
|
import json
|
|
|
|
from tht.decisions import DecisionRecord
|
|
|
|
def decision(seq, kind, subject, **kw):
|
|
return DecisionRecord(seq=seq, ts=datetime.now(UTC), type=kind, subject=subject, **kw)
|
|
return SessionSnapshot(manifest=SessionManifest(id=str(uuid4()), created_at=datetime.now(UTC),
|
|
question="Order totals?", database="dwh", schema="sales", workspace_id=service.repository.workspace_id,
|
|
author="admin"), decisions=[*[decision(i, "phase_approved", f"phase:{i}") for i in range(1, 8)],
|
|
decision(8, "sql_approved", "sql"), decision(9, "datamart_declined", "phase:8"),
|
|
decision(10, "concept_clarified", "Order", detail="All lines shipped", rationale="Domain definition"),
|
|
decision(11, "join_modified", "Order grain", rationale="The reviewer confirmed one row per order")],
|
|
artifacts={"sql_final": "select 1", "memory_proposals": json.dumps(list(proposals))})
|
|
|
|
|
|
def repair_setup(service, tmp_path):
|
|
from types import SimpleNamespace
|
|
|
|
from tht.archive_repair import RepairProposal, prepare
|
|
from tht.evidence.canonical import CuratedEvidence, dump_curated_markdown
|
|
from tht.evidence.local_archive import LocalEvidenceArchive, _content
|
|
|
|
s, vectors = service
|
|
root = tmp_path / s.repository.workspace_id
|
|
path = root / "evidence/curated/domain/order.md"
|
|
path.parent.mkdir(parents=True)
|
|
unit = CuratedEvidence.model_validate({"schema_version": 4, "id": "evidence:order",
|
|
"title": "Order grain", "kind": "domain", "language": "en",
|
|
"purposes": ["sql_generation"], "payload": {"rule": "Use order ID"},
|
|
"provenance": {"kind": "manual", "declared_by": "curator"}})
|
|
path.write_text(dump_curated_markdown(unit))
|
|
archive = LocalEvidenceArchive(root)
|
|
archive.consolidate(actor="curator", activate=lambda _: None)
|
|
unit = archive.get(unit.id)["unit"]
|
|
existing = s.save(card())["card"]
|
|
snapshot = review_snapshot(s)
|
|
cfg = SimpleNamespace(evidence=SimpleNamespace(local_archive_root=root))
|
|
proposal = RepairProposal.model_validate({"reason": "Order definitions conflict", "options": [
|
|
{"id": "fix-memory", "label": "Use the document definition", "archive": "memory",
|
|
"target_id": existing["id"], "revision": existing["revision"],
|
|
"content": card(detail="Use order ID").model_dump(mode="json")},
|
|
{"id": "fix-evidence", "label": "Correct the document rule", "archive": "evidence",
|
|
"target_id": unit.id, "revision": _content(unit),
|
|
"content": unit.model_copy(update={"payload": {"rule": "All lines shipped"}})
|
|
.model_dump(mode="json")}]})
|
|
repair = prepare(s, snapshot, cfg, proposal)
|
|
return s, vectors, snapshot, cfg, archive, path, repair, proposal
|
|
|
|
|
|
@pytest.mark.parametrize("choice", ["fix-memory", "fix-evidence"])
|
|
def test_archive_repair_selected_target_activation_failure_and_retry(service, tmp_path, choice):
|
|
from tht.archive_repair import apply, show
|
|
|
|
s, vectors, snapshot, cfg, archive, path, repair, _ = repair_setup(service, tmp_path)
|
|
before_files = path.read_bytes()
|
|
before_card = s.repository.get(repair["options"][0]["target_id"])
|
|
before_active = archive.active_snapshot()
|
|
vectors.fail = True
|
|
def fail(_):
|
|
raise RuntimeError("private provider error")
|
|
result = apply(s, snapshot, cfg, repair["repair_id"], choice, activate=fail)
|
|
assert result["saved"] and not result["indexed"]
|
|
assert result["status"] == "pending_activation"
|
|
if choice == "fix-memory":
|
|
assert path.read_bytes() == before_files
|
|
else:
|
|
assert s.repository.get(before_card.id) == before_card
|
|
assert archive.active_snapshot() == before_active
|
|
assert archive.get("evidence:order")["unit"].payload.rule == "All lines shipped"
|
|
with pytest.raises(MemoryConflict, match="different recorded choice"):
|
|
apply(s, snapshot, cfg, repair["repair_id"], "reject")
|
|
vectors.fail = False
|
|
result = apply(s, snapshot, cfg, repair["repair_id"], choice, activate=lambda _: None)
|
|
assert result["saved"] and result["indexed"] and result["status"] == "active"
|
|
saved_card = s.repository.get(before_card.id)
|
|
saved_files = path.read_bytes()
|
|
assert apply(s, snapshot, cfg, repair["repair_id"], choice,
|
|
activate=lambda _: None)["indexed"]
|
|
assert s.repository.get(before_card.id) == saved_card
|
|
assert path.read_bytes() == saved_files
|
|
assert show(s, snapshot, cfg, repair["repair_id"])["indexed"]
|
|
|
|
|
|
def test_archive_repair_reject_and_nonadmin_cannot_write_either_archive(service, tmp_path):
|
|
from tht.archive_repair import apply, prepare, show
|
|
|
|
s, _, snapshot, cfg, _, path, repair, proposal = repair_setup(service, tmp_path)
|
|
original = path.read_bytes()
|
|
s.principal = ADMIN.model_copy(update={"is_admin": False})
|
|
assert not show(s, snapshot, cfg, repair["repair_id"])["can_apply"]
|
|
for choice in ["fix-memory", "fix-evidence"]:
|
|
with pytest.raises(MemoryForbidden):
|
|
apply(s, snapshot, cfg, repair["repair_id"], choice)
|
|
assert apply(s, snapshot, cfg, repair["repair_id"], "reject")["status"] == "rejected"
|
|
assert path.read_bytes() == original
|
|
assert prepare(s, snapshot, cfg, proposal)["status"] == "rejected"
|
|
proposal.reason = "A different explanation after reviewer feedback"
|
|
assert prepare(s, snapshot, cfg, proposal)["status"] == "proposed"
|
|
s.principal = ADMIN
|
|
|
|
|
|
def test_archive_repair_protects_other_workspace_session_and_changed_context(service, tmp_path):
|
|
from tht.archive_repair import apply, show
|
|
|
|
s, _, snapshot, cfg, _, _, repair, _ = repair_setup(service, tmp_path)
|
|
foreign = snapshot.model_copy(deep=True)
|
|
foreign.manifest.workspace_id = "foreign"
|
|
with pytest.raises(MemoryForbidden):
|
|
show(s, foreign, cfg, repair["repair_id"])
|
|
foreign.manifest.workspace_id = s.repository.workspace_id
|
|
foreign.manifest.id = str(uuid4())
|
|
with pytest.raises(MemoryNotFound):
|
|
apply(s, foreign, cfg, repair["repair_id"], "fix-memory")
|
|
snapshot.artifacts["question"] = "Changed question"
|
|
with pytest.raises(MemoryConflict, match="Session decisions changed"):
|
|
apply(s, snapshot, cfg, repair["repair_id"], "fix-memory")
|
|
|
|
|
|
@pytest.mark.parametrize("choice", ["fix-memory", "fix-evidence"])
|
|
def test_archive_repair_never_overwrites_intervening_edits(service, tmp_path, choice):
|
|
from tht.archive_repair import apply, show
|
|
|
|
s, _, snapshot, cfg, archive, _, repair, _ = repair_setup(service, tmp_path)
|
|
apply(s, snapshot, cfg, repair["repair_id"], choice, activate=lambda _: None)
|
|
if choice == "fix-memory":
|
|
s.save(card(detail="Newer curator correction"), repair["options"][0]["target_id"])
|
|
else:
|
|
current = archive.get("evidence:order")
|
|
archive.save(current["unit"].model_copy(update={"payload":
|
|
current["unit"].payload.model_copy(update={"rule": "Newer rule"})}),
|
|
expected_revision=current["revision"], actor="other")
|
|
assert show(s, snapshot, cfg, repair["repair_id"])["status"] == "superseded"
|
|
with pytest.raises(MemoryConflict):
|
|
apply(s, snapshot, cfg, repair["repair_id"], choice, activate=lambda _: None)
|
|
|
|
|
|
def test_evidence_repair_recovers_file_write_before_receipt_and_preserves_unrelated_edits(
|
|
service, tmp_path, monkeypatch):
|
|
from tht.archive_repair import apply
|
|
from tht.evidence.local_archive import LocalEvidenceArchive
|
|
|
|
s, _, snapshot, cfg, archive, _, repair, _ = repair_setup(service, tmp_path)
|
|
original = LocalEvidenceArchive._save
|
|
def interrupted(*args, **kwargs):
|
|
original(*args, **kwargs)
|
|
raise OSError("interrupted after file replacement")
|
|
monkeypatch.setattr(LocalEvidenceArchive, "_save", interrupted)
|
|
result = apply(s, snapshot, cfg, repair["repair_id"], "fix-evidence", activate=lambda _: None)
|
|
assert result["status"] == "applying" and not result["indexed"]
|
|
monkeypatch.setattr(LocalEvidenceArchive, "_save", original)
|
|
unrelated = archive.evidence / "curated/domain/unrelated.md"
|
|
unrelated.write_text("Unfinished external edit")
|
|
with pytest.raises(MemoryConflict, match="Other Evidence files changed"):
|
|
apply(s, snapshot, cfg, repair["repair_id"], "fix-evidence", activate=lambda _: None)
|
|
unrelated.unlink()
|
|
assert apply(s, snapshot, cfg, repair["repair_id"], "fix-evidence",
|
|
activate=lambda _: None)["indexed"]
|
|
|
|
|
|
def test_repair_cli_persists_and_retrieves_both_archives_with_real_qdrant(
|
|
service, tmp_path, monkeypatch, database_url):
|
|
import json
|
|
|
|
import yaml
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app
|
|
from tht.evidence.corpus.store import CorpusStore
|
|
from tht.evidence.search import ActiveEvidenceSearcher, EvidenceSearchContext, search_evidence
|
|
|
|
s, _, snapshot, _, archive, _, _, proposal = repair_setup(service, tmp_path)
|
|
with DockerContainer("qdrant/qdrant:v1.18.2").with_exposed_ports(6333) as container:
|
|
wait_for_logs(container, "Qdrant HTTP listening on 6333")
|
|
url = f"http://{container.get_container_host_ip()}:{container.get_exposed_port(6333)}"
|
|
workspace = s.repository.workspace_id
|
|
vectors = QdrantVectorStore(base_url=url, workspace_id=workspace,
|
|
collections={"reference": workspace+"-reference", "memory": workspace+"-memory"},
|
|
expected_dimension=3)
|
|
for suffix in ("reference", "memory"):
|
|
requests.put(f"{url}/collections/{workspace}-{suffix}", json={
|
|
"vectors": {"size": 3, "distance": "Cosine"},
|
|
"sparse_vectors": {"bm25": {"modifier": "idf"}},
|
|
}, timeout=10).raise_for_status()
|
|
runtime = tmp_path / "runtime.yaml"
|
|
runtime.write_text(yaml.safe_dump({
|
|
"runtime_identity": {"workspace_id": workspace, "workspace_revision": "a"*40},
|
|
"dwh": {"type": "postgres_direct", "connection": {"database": "unused",
|
|
"schema": "public", "user": "unused", "password": "unused"}},
|
|
"vectors": {"type": "qdrant", "base_url": "http://qdrant:6333", "collection": workspace},
|
|
"embeddings": {"provider": "ollama_internal", "base_url": "http://embedding:11434",
|
|
"model": "fixture", "dim": 3},
|
|
"evidence": {"schema_version": 2, "local_archive_root": str(archive.root),
|
|
"sources": [{"type": "http", "urls": ["https://must-not-fetch.invalid/source.md"]}]},
|
|
"roots": {"sessions": str(tmp_path / "sessions"), "artifacts": str(tmp_path / "artifacts"),
|
|
"indexes": str(tmp_path / "indexes")},
|
|
}))
|
|
monkeypatch.setenv("THT_CATALOG_DATABASE_URL", database_url)
|
|
monkeypatch.setenv("THT_PRINCIPAL_ISSUER", "test")
|
|
monkeypatch.setenv("THT_PRINCIPAL_SUBJECT", "admin")
|
|
monkeypatch.setenv("THT_PRINCIPAL_IS_ADMIN", "true")
|
|
monkeypatch.setattr("tht.cli.memory_cmd.load_snapshot_or_exit", lambda *_: snapshot)
|
|
monkeypatch.setattr("tht.adapters.factory.build_vector_store", lambda *a, **kw: vectors)
|
|
monkeypatch.setattr("tht.cli.vector_cmd.make_embedder", lambda _: Embeddings())
|
|
|
|
def command(*args):
|
|
result = CliRunner().invoke(app, ["memory", *args, "--session", snapshot.manifest.id,
|
|
"--json", "-c", str(runtime)])
|
|
assert result.exit_code == 0, (result.output, result.exception)
|
|
return json.loads(result.stdout)
|
|
|
|
current = command("repair-target", "--archive", "memory", "--target-id",
|
|
proposal.options[0].target_id)
|
|
assert current["revision"] == proposal.options[0].revision
|
|
prepared = command("repair-prepare", "--proposal-json", proposal.model_dump_json())
|
|
assert command("repair-apply", "--repair-id", prepared["repair_id"],
|
|
"--choice", "fix-memory")["indexed"]
|
|
|
|
class MemorySearch:
|
|
def search(self, embedding, top_n=10, kinds=None, **kwargs):
|
|
return vectors.search(["memory"], embedding, limit=top_n, kinds=kinds, **kwargs)
|
|
assert s.recall("Order", searcher=MemorySearch(), embedder=Embeddings())[0]["detail"] \
|
|
== "Use order ID"
|
|
proposal.options = [proposal.options[1]]
|
|
prepared = command("repair-prepare", "--proposal-json", proposal.model_dump_json())
|
|
applied = command("repair-apply", "--repair-id", prepared["repair_id"],
|
|
"--choice", "fix-evidence")
|
|
assert applied["indexed"] and applied["saved"]
|
|
|
|
class EvidenceSearch:
|
|
def search(self, embedding, top_n=10, kinds=None, **kwargs):
|
|
return vectors.search(["evidence"], embedding, limit=top_n, kinds=kinds, **kwargs)
|
|
found = search_evidence("Order", "sql_generation", EvidenceSearchContext(),
|
|
searcher=ActiveEvidenceSearcher(CorpusStore(tmp_path / "corpus"), EvidenceSearch(),
|
|
workspace, "english"), embedder=Embeddings())
|
|
assert found.status == "available"
|
|
assert any("All lines shipped" in " ".join(hit.excerpts) for hit in found.results)
|
|
assert command("repair-show", "--repair-id", prepared["repair_id"])["status"] == "active"
|
|
assert s.recall("Order", searcher=MemorySearch(), embedder=Embeddings())[0]["detail"] \
|
|
== "Use order ID"
|
|
|
|
|
|
def test_final_review_edits_selected_cards_and_does_not_save_declined_exemplar(service):
|
|
from tht.memory.review import ReviewResponse, apply, prepare
|
|
s, _ = service
|
|
snapshot = review_snapshot(s)
|
|
summary = prepare(s, snapshot)
|
|
item = next(p for p in summary["items"] if p["card"]["family"] == "domain_clarification")
|
|
item["card"]["detail"] = "Reviewer-edited definition"
|
|
response = ReviewResponse.model_validate({"summary_id": summary["summary_id"],
|
|
"items": [{"id": item["id"], "card": item["card"]}]})
|
|
outcome = apply(s, snapshot, response)
|
|
assert outcome["saved"] == 1 and outcome["indexed"]
|
|
assert s.list(CardQuery())["items"][0]["detail"] == "Reviewer-edited definition"
|
|
assert s.list(CardQuery(family="solved_question"))["total"] == 0
|
|
assert apply(s, snapshot, response)["saved"] == 1
|
|
assert prepare(s, snapshot)["reviewed"]
|
|
|
|
|
|
def test_review_can_decline_everything_and_recover_without_another_prompt(service):
|
|
from tht.memory.review import ReviewResponse, apply, prepare
|
|
s, _ = service
|
|
snapshot = review_snapshot(s)
|
|
summary = prepare(s, snapshot)
|
|
assert apply(s, snapshot, ReviewResponse(summary_id=summary["summary_id"], items=[]))["saved"] == 0
|
|
assert prepare(s, snapshot)["reviewed"]
|
|
assert s.list(CardQuery())["total"] == 0
|
|
|
|
|
|
def test_selected_new_cards_and_links_are_atomic_and_idempotent(service):
|
|
from tht.memory.review import ReviewResponse, apply, prepare
|
|
s, _ = service
|
|
proposals = [{"id": "grain", "source_seqs": [11], "reason": "Avoid repeated totals",
|
|
"card": card(family="sql_rule", subject="Order grain").model_dump()}]
|
|
snapshot = review_snapshot(s, proposals)
|
|
summary = prepare(s, snapshot)
|
|
selected = [{"id": p["id"], "card": p["card"]} for p in summary["items"] if p["id"] != "solved-question"]
|
|
selected[0]["card"]["links"] = [{"target_id": "proposal:decision-10", "meaning": "Definition"}]
|
|
response = ReviewResponse.model_validate({"summary_id": summary["summary_id"], "items": selected})
|
|
apply(s, snapshot, response)
|
|
cards = s.list(CardQuery())["items"]
|
|
source = next(c for c in cards if c["family"] == "sql_rule")
|
|
assert len(source["links"]) == 1
|
|
assert source["links"][0]["target_id"] in {c["id"] for c in cards}
|
|
s.delete(source["id"])
|
|
apply(s, snapshot, response)
|
|
assert s.list(CardQuery())["total"] == 1 # no resurrection on retry
|
|
|
|
|
|
def test_invalid_selected_link_rolls_back_all_selected_cards(service):
|
|
from tht.memory.review import ReviewResponse, apply, prepare
|
|
s, _ = service
|
|
snapshot = review_snapshot(s)
|
|
summary = prepare(s, snapshot)
|
|
selected = [{"id": p["id"], "card": p["card"]} for p in summary["items"]]
|
|
selected[-1]["card"]["links"] = [{"target_id": "missing", "meaning": "Invalid target"}]
|
|
with pytest.raises(MemoryConflict):
|
|
apply(s, snapshot, ReviewResponse.model_validate({"summary_id": summary["summary_id"], "items": selected}))
|
|
assert s.list(CardQuery())["total"] == 0
|
|
|
|
|
|
def test_review_update_detects_manual_changes_and_preserves_origin(service):
|
|
from tht.memory.review import ReviewResponse, apply, prepare
|
|
s, _ = service
|
|
identity = s.save(card())["id"]
|
|
current = s.get(identity)
|
|
proposal = {"id": "update", "source_seqs": [10], "reason": "Correct the definition",
|
|
"target_id": identity, "target_revision": current["revision"], "card": card(detail="Corrected").model_dump()}
|
|
snapshot = review_snapshot(s, [proposal])
|
|
summary = prepare(s, snapshot)
|
|
s.save(card(detail="Manual correction"), identity)
|
|
with pytest.raises(MemoryConflict):
|
|
apply(s, snapshot, ReviewResponse.model_validate({"summary_id": summary["summary_id"],
|
|
"items": [{"id": "update", "card": proposal["card"]}]}))
|
|
assert s.get(identity)["detail"] == "Manual correction"
|
|
assert s.get(identity)["origin"] == "manual"
|
|
|
|
|
|
def test_review_reuses_exact_manual_cards_and_resolves_links_to_them(service):
|
|
from tht.memory.review import ReviewResponse, apply, prepare
|
|
s, _ = service
|
|
existing = s.save(card())["id"]
|
|
proposals = [
|
|
{"id": "existing", "source_seqs": [10], "reason": "Definition", "card": card().model_dump()},
|
|
{"id": "rule", "source_seqs": [11], "reason": "Correct grain",
|
|
"card": card(family="sql_rule", links=[{"target_id": "proposal:existing", "meaning": "Definition"}]).model_dump()},
|
|
]
|
|
snapshot = review_snapshot(s, proposals)
|
|
summary = prepare(s, snapshot)
|
|
assert "existing" not in {p["id"] for p in summary["items"]}
|
|
rule = next(p for p in summary["items"] if p["id"] == "rule")
|
|
assert rule["card"]["links"][0]["target_id"] == existing
|
|
apply(s, snapshot, ReviewResponse.model_validate({"summary_id": summary["summary_id"],
|
|
"items": [{"id": "rule", "card": rule["card"]}]}))
|
|
assert s.list(CardQuery())["total"] == 2
|
|
assert s.get(existing)["origin"] == "manual"
|
|
|
|
|
|
def test_review_explicit_update_preserves_manual_identity_and_origin(service):
|
|
from tht.memory.review import ReviewResponse, apply, prepare
|
|
s, _ = service
|
|
existing = s.save(card())["id"]
|
|
proposal = {"id": "update", "source_seqs": [10], "reason": "Reviewer correction",
|
|
"target_id": existing, "target_revision": s.get(existing)["revision"],
|
|
"card": card(detail="Corrected definition").model_dump()}
|
|
snapshot = review_snapshot(s, [proposal])
|
|
summary = prepare(s, snapshot)
|
|
apply(s, snapshot, ReviewResponse.model_validate({"summary_id": summary["summary_id"],
|
|
"items": [{"id": "update", "card": proposal["card"]}]}))
|
|
assert s.list(CardQuery())["total"] == 1
|
|
assert s.get(existing)["detail"] == "Corrected definition"
|
|
assert s.get(existing)["origin"] == "manual"
|
|
|
|
|
|
def test_table_cleanup_removes_dependent_links_but_not_global_cards(service):
|
|
from tht.memory.cleanup import CleanupRequest, cleanup
|
|
s, _ = service
|
|
removed = s.save(card(dependencies=[{"database": "dwh", "schema_name": "sales", "table": "orders"}]))["id"]
|
|
global_card = s.save(card(subject="General", links=[{"target_id": removed, "meaning": "Example"}]))["id"]
|
|
request = CleanupRequest(sync_id="table-sync", database="dwh", schema_name="sales", removed_tables=["orders"])
|
|
assert cleanup(s, request) == {"deleted": 1, "indexed": True}
|
|
assert s.get(global_card)["links"] == []
|
|
assert cleanup(s, request)["deleted"] == 1
|
|
with pytest.raises(MemoryConflict):
|
|
cleanup(s, request.model_copy(update={"removed_tables": ["another"]}))
|
|
|
|
|
|
def test_rules_cli_exposes_only_applicable_families_without_approving_them(service, monkeypatch):
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app, memory_cmd, vector_cmd
|
|
s, vectors = service
|
|
snapshot = review_snapshot(s)
|
|
snapshot.decisions[:] = [d for d in snapshot.decisions
|
|
if not (d.type == "phase_approved" and int(d.subject.split(":")[1]) > 3)]
|
|
original_decisions = list(snapshot.decisions)
|
|
dependency = [{"database": "dwh", "schema_name": "sales"}]
|
|
rule = s.save(card(family="sql_rule", dependencies=dependency))["id"]
|
|
error = s.save(card(family="explained_error", rationale="Approved explanation", dependencies=dependency))["id"]
|
|
s.save(card(family="sql_rule", dependencies=[{"database": "other"}]))
|
|
s.save(card())
|
|
cfg = SimpleNamespace(database=SimpleNamespace(database="dwh", db_schema="sales"), embeddings=None)
|
|
monkeypatch.setattr(memory_cmd, "_load_config_or_exit", lambda _: cfg)
|
|
monkeypatch.setattr(memory_cmd, "memory_service", lambda _: s)
|
|
monkeypatch.setattr(memory_cmd, "load_snapshot_or_exit", lambda *_: snapshot)
|
|
monkeypatch.setattr(s, "close", lambda: None)
|
|
monkeypatch.setattr(vector_cmd, "open_searcher", lambda _: vectors)
|
|
monkeypatch.setattr(vector_cmd, "make_embedder", lambda _: Embeddings())
|
|
result = CliRunner().invoke(app, ["memory", "rules", "Order", "--session", snapshot.manifest.id, "--json"])
|
|
assert result.exit_code == 0, result.output
|
|
payload = json.loads(result.output)
|
|
assert {c["id"] for c in payload} == {rule, error}
|
|
assert all(c["consultative"] for c in payload)
|
|
assert snapshot.decisions == original_decisions
|
|
|
|
|
|
def test_undocumented_failures_are_not_reusable_proposal_sources(service):
|
|
from tht.memory.review import validate_proposals
|
|
s, _ = service
|
|
snapshot = review_snapshot(s)
|
|
proposal = {"id": "error", "source_seqs": [9], "reason": "Failure",
|
|
"card": card(family="explained_error", rationale="Timeout").model_dump()}
|
|
with pytest.raises(MemoryConflict, match="approved source"):
|
|
validate_proposals(snapshot, [proposal])
|
|
|
|
|
|
def test_physical_cleanup_uses_exact_dependencies_and_recovers_failed_vector_deletions(service):
|
|
from tht.memory.cleanup import CleanupRequest, cleanup
|
|
s, vectors = service
|
|
dep = {"database": "dwh", "schema_name": "sales", "table": "orders", "column": "id"}
|
|
removed = s.save(card(dependencies=[dep]))["id"]
|
|
surviving = [s.save(card(dependencies=dependencies))["id"] for dependencies in [[],
|
|
[{**dep, "database": "other"}], [{**dep, "schema_name": "audit"}], [{**dep, "column": "amount"}]]]
|
|
request = CleanupRequest(sync_id="sync", database="dwh", schema_name="sales",
|
|
removed_columns=[{"table": "orders", "column": "id"}])
|
|
vectors.fail = True
|
|
result = cleanup(s, request)
|
|
assert result == {"deleted": 1, "indexed": False}
|
|
with pytest.raises(MemoryNotFound): s.get(removed)
|
|
assert {c["id"] for c in s.list(CardQuery())["items"]} == set(surviving)
|
|
vectors.fail = False
|
|
assert cleanup(s, request) == {"deleted": 1, "indexed": True}
|
|
assert removed not in vectors.records
|
|
|
|
|
|
@pytest.mark.l2
|
|
def test_real_model_generates_correct_join_from_a_retrieved_rule(service, database_url, tmp_path):
|
|
"""Opt-in generation check on synthetic PostgreSQL data; no real DWH rows are sent."""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from tht.memory.retrieval import RecallScope
|
|
|
|
installation = os.environ.get("THT_MEMORY_L2_INSTALLATION")
|
|
container = os.environ.get("THT_MEMORY_L2_CORE")
|
|
if not installation or not container:
|
|
pytest.skip("Set THT_MEMORY_L2_INSTALLATION and THT_MEMORY_L2_CORE")
|
|
catalog = yaml.safe_load(Path(installation).read_text())["modelCatalog"]
|
|
provider_id, model = catalog["defaults"]["session"].split("/", 1)
|
|
provider = catalog["providers"][provider_id]
|
|
s, vectors = service
|
|
schema = "memory_case_" + uuid4().hex
|
|
engine = create_engine(database_url)
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.execute(text(f"CREATE SCHEMA {schema}"))
|
|
connection.execute(text(f"CREATE TABLE {schema}.orders (order_id int, financial_year int)"))
|
|
connection.execute(text(f"CREATE TABLE {schema}.lines (order_id int, financial_year int, amount int)"))
|
|
connection.execute(text(f"INSERT INTO {schema}.orders VALUES (1,2024),(1,2025),(2,2025)"))
|
|
connection.execute(text(f"INSERT INTO {schema}.lines VALUES (1,2024,900),(1,2025,100),(2,2025,20)"))
|
|
identity = s.save(card(family="sql_rule", subject="Order key across years", scope="Sales",
|
|
detail="Order IDs repeat across financial years. Join orders to lines on both order_id and financial_year.",
|
|
rationale="Reviewer approved this join correction",
|
|
dependencies=[{"database": "fixture", "schema_name": schema, "table": "orders"}]))["id"]
|
|
s.save(card(family="sql_rule", subject="Unrelated reporting rule",
|
|
detail="Join on order_id alone.", dependencies=[{"database": "another"}]))
|
|
candidates = s.retrieve("Total order amount in 2025", searcher=vectors, embedder=Embeddings(),
|
|
family="sql_rule", scope=RecallScope(database="fixture", schema_name=schema))
|
|
assert [c.card.id for c in candidates] == [identity]
|
|
prompt = (
|
|
f"Generate one read-only PostgreSQL SELECT. Return JSON with exactly one key sql. "
|
|
f"Schema: {schema}.orders(order_id int,financial_year int); "
|
|
f"{schema}.lines(order_id int,financial_year int,amount int). "
|
|
"Question: total line amount for orders in financial year 2025. "
|
|
"Use a join between orders and lines. Return only one numeric aggregate column. "
|
|
"The reviewer has approved this retrieved Memory rule for the current question: "
|
|
+ candidates[0].card.detail
|
|
)
|
|
request = {"model": provider["metadataGeneration"]["litellmProvider"] + "/" + model,
|
|
"api_base": provider["endpoint"]["baseUrl"],
|
|
"messages": [{"role": "user", "content": prompt}]}
|
|
# Resolve the installation secret inside core; never return it to the test runner.
|
|
program = (
|
|
"import sys,json,os; from dotenv import dotenv_values; "
|
|
"from tht.internal.litellm_completion import handle_request; "
|
|
"p=json.load(sys.stdin); k=p.pop('key_env'); "
|
|
"key=os.environ.get(k) or dotenv_values(os.environ['THT_SECRETS_FILE']).get(k); "
|
|
"assert key, 'Configured model credential unavailable'; "
|
|
"p['api_key']=key; print(json.dumps(handle_request(p)))"
|
|
)
|
|
request["key_env"] = provider["authentication"]["apiKeyEnv"]
|
|
result = subprocess.run(["docker", "exec", "-i", container, "python", "-c", program],
|
|
input=json.dumps(request), capture_output=True, text=True, timeout=180, check=True)
|
|
outcome = json.loads(result.stdout)
|
|
assert outcome["ok"], outcome.get("error")
|
|
content = outcome["content"].strip()
|
|
if content.startswith("```"):
|
|
content = content.split("\n", 1)[1].rsplit("```", 1)[0].strip()
|
|
sql = json.loads(content)["sql"]
|
|
from tht.sqlcheck import validate_sql
|
|
assert validate_sql(sql).ok
|
|
with engine.connect() as connection, connection.begin():
|
|
connection.execute(text("SET TRANSACTION READ ONLY"))
|
|
connection.execute(text("SET LOCAL statement_timeout='5s'"))
|
|
rows = connection.execute(text(sql)).all()
|
|
assert rows == [(120,)]
|
|
(tmp_path / "generated.sql").write_text(sql)
|
|
print(f"Real model {provider_id}/{model}: retrieved approved rule, generated SQL, result 120")
|
|
finally:
|
|
with engine.begin() as connection:
|
|
connection.execute(text(f"DROP SCHEMA IF EXISTS {schema} CASCADE"))
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.mark.l2
|
|
def test_real_model_proposes_a_reviewable_persistent_archive_correction(service, tmp_path):
|
|
"""Opt-in model check on synthetic knowledge; the test reviewer chooses Memory."""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from tht.archive_repair import RepairProposal, apply, prepare, target
|
|
|
|
installation = os.environ.get("THT_MEMORY_L2_INSTALLATION")
|
|
container = os.environ.get("THT_MEMORY_L2_CORE")
|
|
if not installation or not container:
|
|
pytest.skip("Set THT_MEMORY_L2_INSTALLATION and THT_MEMORY_L2_CORE")
|
|
s, vectors, snapshot, cfg, archive, _, initial, _ = repair_setup(service, tmp_path)
|
|
memory_id = initial["options"][0]["target_id"]
|
|
original = card(family="sql_rule", subject="Order business key",
|
|
detail="Join orders and lines on order_id alone.")
|
|
s.save(original, memory_id)
|
|
current = archive.get("evidence:order")
|
|
archive.save(current["unit"].model_copy(update={"payload": current["unit"].payload.model_copy(
|
|
update={"rule": "Order IDs repeat across financial years. Join on order_id and financial_year."})}),
|
|
expected_revision=current["revision"], actor="fixture curator")
|
|
archive.consolidate(actor="fixture curator", activate=lambda _: None)
|
|
targets = [{"archive": name, "target_id": identity,
|
|
**target(s, snapshot, cfg, name, identity)}
|
|
for name, identity in [("memory", memory_id), ("evidence", "evidence:order")]]
|
|
catalog = yaml.safe_load(Path(installation).read_text())["modelCatalog"]
|
|
provider_id, model = catalog["defaults"]["session"].split("/", 1)
|
|
provider = catalog["providers"][provider_id]
|
|
prompt = (
|
|
"This is an isolated acceptance case using synthetic data. Memory and Evidence "
|
|
"disagree about the business key used to join orders and lines. Prepare specific "
|
|
"alternatives for a human reviewer; do not decide for the reviewer. Return only JSON "
|
|
"matching this schema: " + json.dumps(RepairProposal.model_json_schema()) +
|
|
"\nReturn exactly two options: correct Memory to agree with Evidence, or correct "
|
|
"Evidence to agree with Memory. Each option must contain its complete resulting "
|
|
"content, preserving all unrelated fields, exact identity, revision, and source history. "
|
|
"Explain the effect on joins in the reason and labels. Current targets:\n" + json.dumps(targets)
|
|
)
|
|
request = {"model": provider["metadataGeneration"]["litellmProvider"] + "/" + model,
|
|
"api_base": provider["endpoint"]["baseUrl"], "messages": [{"role": "user", "content": prompt}],
|
|
"key_env": provider["authentication"]["apiKeyEnv"]}
|
|
program = (
|
|
"import sys,json,os; from dotenv import dotenv_values; "
|
|
"from tht.internal.litellm_completion import handle_request; "
|
|
"p=json.load(sys.stdin); k=p.pop('key_env'); "
|
|
"key=os.environ.get(k) or dotenv_values(os.environ['THT_SECRETS_FILE']).get(k); "
|
|
"assert key, 'Configured model credential unavailable'; "
|
|
"p['api_key']=key; print(json.dumps(handle_request(p)))"
|
|
)
|
|
response = subprocess.run(["docker", "exec", "-i", container, "python", "-c", program],
|
|
input=json.dumps(request), capture_output=True, text=True, timeout=180, check=True)
|
|
outcome = json.loads(response.stdout)
|
|
assert outcome["ok"], outcome.get("error")
|
|
content = outcome["content"].strip()
|
|
if content.startswith("```"):
|
|
content = content.split("\n", 1)[1].rsplit("```", 1)[0].strip()
|
|
proposal = RepairProposal.model_validate_json(content)
|
|
assert len(proposal.options) == 2
|
|
assert {o.archive for o in proposal.options} == {"memory", "evidence"}
|
|
prepared = prepare(s, snapshot, cfg, proposal)
|
|
selected = next(o for o in proposal.options if o.archive == "memory")
|
|
assert "financial_year" in selected.content["detail"]
|
|
evidence_before = archive.get("evidence:order")
|
|
applied = apply(s, snapshot, cfg, prepared["repair_id"], selected.id)
|
|
assert applied["indexed"] and applied["saved"]
|
|
assert archive.get("evidence:order") == evidence_before
|
|
recalled = s.retrieve("Order key", searcher=vectors, embedder=Embeddings(), family="sql_rule")
|
|
assert len(recalled) == 1 and "financial_year" in recalled[0].card.detail
|
|
(tmp_path / "model-repair.json").write_text(json.dumps(proposal.model_dump(mode="json"), indent=2))
|
|
print(f"Real model {provider_id}/{model}: two valid alternatives; selected correction saved and retrieved")
|
|
|
|
|
|
def test_hybrid_recall_with_configured_real_embedding_and_isolated_stores(database_url):
|
|
"""Opt in with the installation model volume, mounted read-only into a new Ollama.
|
|
|
|
Uses the production embedding client and Qdrant adapter, not prepared hit IDs.
|
|
PostgreSQL, Ollama process and Qdrant collection are isolated test resources.
|
|
"""
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from tht.config import EmbeddingsConfig
|
|
from tht.memory.retrieval import RecallScope, expand_and_rank
|
|
from tht.vectorstore.embeddings import OllamaEmbeddings
|
|
|
|
volume = os.environ.get("THT_MEMORY_TEST_OLLAMA_VOLUME")
|
|
if not volume:
|
|
pytest.skip("Set THT_MEMORY_TEST_OLLAMA_VOLUME to a provisioned installation model volume")
|
|
model = os.environ["THT_MEMORY_TEST_MODEL"]
|
|
dimensions = int(os.environ["THT_MEMORY_TEST_DIMENSIONS"])
|
|
services = yaml.safe_load((Path(__file__).resolve().parents[3] / "compose.yaml").read_text())["services"]
|
|
with (DockerContainer(services["embedding"]["image"])
|
|
.with_volume_mapping(volume, "/root/.ollama", mode="ro")
|
|
.with_exposed_ports(11434) as ollama,
|
|
DockerContainer(services["qdrant"]["image"]).with_exposed_ports(6333) as qdrant):
|
|
wait_for_logs(ollama, "Listening on")
|
|
wait_for_logs(qdrant, "Qdrant HTTP listening on 6333")
|
|
embedder = OllamaEmbeddings(EmbeddingsConfig(
|
|
base_url=f"http://{ollama.get_container_host_ip()}:{ollama.get_exposed_port(11434)}",
|
|
model=model, dimensions=dimensions, timeout=120,
|
|
))
|
|
url = f"http://{qdrant.get_container_host_ip()}:{qdrant.get_exposed_port(6333)}"
|
|
workspace = "hybrid-" + uuid4().hex
|
|
collections = {"reference": workspace+"-reference", "memory": workspace+"-memory"}
|
|
store = QdrantVectorStore(base_url=url, collections=collections,
|
|
workspace_id=workspace, expected_dimension=dimensions)
|
|
s = MemoryService(MemoryRepository(database_url, workspace), ADMIN, language="it",
|
|
store_factory=lambda: store, embedder_factory=lambda: embedder)
|
|
|
|
class Searcher:
|
|
def search(self, embedding, top_n, kinds, **kwargs):
|
|
return store.search(["memory"], embedding, limit=top_n, kinds=kinds, **kwargs)
|
|
|
|
def save(**value):
|
|
outcome = s.save(card(**value))
|
|
assert outcome["indexed"], outcome
|
|
return outcome["id"]
|
|
|
|
scope = RecallScope(database="dwh", schema_name="sales", table="commesse",
|
|
column="codice", scope="Commesse")
|
|
dependency = {"database": "dwh", "schema_name": "sales", "table": "commesse", "column": "codice"}
|
|
try:
|
|
general = save(subject="Contesto delle commesse", detail="Convenzioni del processo.",
|
|
scope="Commesse")
|
|
parent = save(subject="Convenzione del database", detail="Regola valida nel database.",
|
|
scope="Commesse", dependencies=[{"database": "dwh"}])
|
|
grain = save(subject="Granularità delle righe", detail="Una riga per commessa ed esercizio.",
|
|
scope="Commesse", dependencies=[dependency])
|
|
rule = save(subject="Chiave commessa K9R7", detail="Il codice commessa si ripete fra esercizi. "
|
|
"Usare codice ed esercizio insieme per evitare duplicazioni.",
|
|
scope="Commesse", concepts=["commessa"], dependencies=[dependency],
|
|
links=[{"target_id": grain, "meaning": "La chiave deve rispettare la granularità"}])
|
|
wrong_db = save(subject="Chiave commessa K9R7", detail="Regola di un altro database.",
|
|
scope="Commesse", dependencies=[{**dependency, "database": "other"}])
|
|
wrong_scope = save(subject="Chiave commessa K9R7", scope="Acquisti", dependencies=[dependency])
|
|
split = save(subject="Chiave commessa K9R7", scope="Commesse", dependencies=[
|
|
{**dependency, "column": "importo"}, {**dependency, "database": "other"},
|
|
])
|
|
# A directed link is not permission to enter another scope or use a different family.
|
|
value = {k: v for k, v in s.get(rule).items() if k in CardInput.model_fields}
|
|
value["links"] = [
|
|
{"target_id": grain, "meaning": "Grain"},
|
|
{"target_id": wrong_db, "meaning": "Related external rule"},
|
|
]
|
|
assert s.save(CardInput.model_validate(value), rule)["indexed"]
|
|
query = "K9R7: quale chiave evita duplicazioni fra esercizi delle commesse?"
|
|
vector = embedder.embed_query(query)
|
|
hits_by_mode = {}
|
|
for mode in ("dense", "bm25", "fused"):
|
|
hits = store.search(["memory"], vector, limit=10, kinds=["memory"],
|
|
query_text=query, query_language="italian", retrieval_mode=mode,
|
|
metadata_filter=scope.vector_filter("domain_clarification"))
|
|
hits_by_mode[mode] = hits
|
|
assert rule in {h.ref for h in hits}, (mode, hits)
|
|
assert not {wrong_db, wrong_scope, split} & {h.ref for h in hits}
|
|
assert hits_by_mode["bm25"][0].ref == rule
|
|
results = s.recall(query, searcher=Searcher(), embedder=embedder, scope=scope)
|
|
assert {r["id"] for r in results} == {rule, grain, general, parent}
|
|
# The actual lexical seed discovers its linked card without giving the
|
|
# expansion function that destination as a prepared search result.
|
|
with s.repository.operation() as repo:
|
|
linked = expand_and_rank(repo, hits_by_mode["bm25"][:1], scope=scope,
|
|
family="domain_clarification", excluded=set(), top=10)
|
|
assert next(r for r in linked if r.card.id == grain).path == (rule, grain)
|
|
assert s.delete(grain)["indexed"]
|
|
assert all(r["id"] != grain for r in s.recall(query, searcher=Searcher(),
|
|
embedder=embedder, scope=scope))
|
|
store.clear_reference()
|
|
assert all(r["indexed"] for r in s.rebuild())
|
|
assert {r["id"] for r in s.recall(query, searcher=Searcher(), embedder=embedder,
|
|
scope=scope)} == {rule, general, parent}
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_empty_archive_rules_skip_unavailable_embedding_and_vectors(service, monkeypatch):
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli import app, memory_cmd, vector_cmd
|
|
from tht.ports.vector import VectorStoreError
|
|
from tht.vectorstore.embeddings import EmbeddingsError
|
|
|
|
s, _ = service
|
|
snapshot = review_snapshot(s)
|
|
snapshot.decisions[:] = [d for d in snapshot.decisions
|
|
if not (d.type == "phase_approved" and int(d.subject.split(":")[1]) > 3)]
|
|
|
|
def unavailable_embedding(_):
|
|
raise EmbeddingsError("Embedding unavailable")
|
|
|
|
def unavailable_search(*args, **kwargs):
|
|
raise VectorStoreError("BM25 collection configuration mismatch")
|
|
|
|
cfg = SimpleNamespace(database=SimpleNamespace(database="dwh", db_schema="sales"), embeddings=None)
|
|
monkeypatch.setattr(memory_cmd, "_load_config_or_exit", lambda _: cfg)
|
|
monkeypatch.setattr(memory_cmd, "memory_service", lambda _: s)
|
|
monkeypatch.setattr(memory_cmd, "load_snapshot_or_exit", lambda *_: snapshot)
|
|
monkeypatch.setattr(s, "close", lambda: None)
|
|
monkeypatch.setattr(vector_cmd, "open_searcher", lambda _: SimpleNamespace(search=unavailable_search))
|
|
monkeypatch.setattr(vector_cmd, "make_embedder", lambda _: SimpleNamespace(embed_query=unavailable_embedding))
|
|
result = CliRunner().invoke(app, ["memory", "rules", "Order", "--session", snapshot.manifest.id, "--json"])
|
|
assert result.exit_code == 0, result.output
|
|
assert json.loads(result.output) == []
|