import json import uuid from contextlib import nullcontext from datetime import UTC, datetime from types import SimpleNamespace from typer.testing import CliRunner from tht.cli import app from tht.decisions import DecisionInput from tht.memory import MemoryRecord, recall_memories from tht.memory.models import Card from tht.memory.service import MemoryService from tht.phase import current_phase from tht.session.filesystem_repository import FilesystemSessionRepository from tht.session.models import PrincipalContext, SessionManifest def _memory(id_: str, type_: str = "concept_clarified") -> MemoryRecord: return MemoryRecord( id=id_, ts=datetime(2026, 8, 24, tzinfo=UTC), session_id="source-session", decision_seq=int(id_.split("-")[1]), type=type_, subject=f"subject {id_}", detail=f"detail {id_}", rationale=f"rationale {id_}", question_context="source question", concepts=[f"concept {id_}"], ) class Embedder: def __init__(self): self.questions = [] def embed_query(self, question): self.questions.append(question) return [0.1, 0.2] class Searcher: def __init__(self, hits): self.hits = hits self.calls = [] def search(self, embedding, *, top_n, kinds, **kwargs): self.calls.append((embedding, top_n, kinds)) return self.hits def test_recall_preserves_rank_and_public_payload_for_reusable_memories(): records = [_memory("mem-0001"), _memory("mem-0002", "table_promoted"), _memory("mem-0003")] searcher = Searcher([ SimpleNamespace(ref="mem-0003", similarity=0.93456), SimpleNamespace(ref="mem-0002", similarity=0.92345), SimpleNamespace(ref="orphan", similarity=0.91234), SimpleNamespace(ref="mem-0001", similarity=0.87654), ]) embedder = Embedder() results = recall_memories( "active patients", records=records, decisions=[], searcher=searcher, embedder=embedder, top=5, ) assert [result["id"] for result in results] == ["mem-0003", "mem-0001"] assert results[0] == { "id": "mem-0003", "type": "concept_clarified", "subject": "subject mem-0003", "detail": "detail mem-0003", "rationale": "rationale mem-0003", "question_context": "source question", "tables": [], "concepts": ["concept mem-0003"], "session_id": "source-session", "score": 0.9346, } assert embedder.questions == ["active patients"] assert searcher.calls == [([0.1, 0.2], 5, ["memory"])] def _workspace_config(tmp_path): config = tmp_path / "workspace.yaml" config.write_text( f""" runtime_identity: workspace_id: psd-clinical workspace_revision: {'a' * 40} dwh: type: postgres_direct connection: {{database: analytics, schema: mart, user: reader, password: secret}} vectors: type: qdrant base_url: http://qdrant:6333 collection: psd-clinical roots: sessions: {tmp_path / 'sessions'} artifacts: {tmp_path / 'artifacts'} indexes: {tmp_path / 'indexes'} embeddings: provider: ollama_internal base_url: http://embedding:11434 model: qwen3-embedding:0.6b dim: 1024 """ ) return config def test_recall_cli_reconstructs_applied_and_rejected_memory_from_persisted_f2_session( tmp_path, monkeypatch ): monkeypatch.setenv("THT_HOME", str(tmp_path / "home")) repository = FilesystemSessionRepository( tmp_path / "home", "psd-clinical", PrincipalContext(issuer="local", subject="reviewer"), root=tmp_path / "sessions", ) session_id = str(uuid.uuid4()) repository.create(SessionManifest( id=session_id, created_at=datetime(2026, 8, 24, tzinfo=UTC), question="active patients", database="analytics", schema="mart", )) repository.append_decisions(session_id, [ DecisionInput(type="phase_approved", subject="phase:1"), DecisionInput( type="concept_clarified", subject="active patient", rationale="Applied from mem-0003", ), DecisionInput( type="memory_rejected", subject="mem-0001", rationale="Not relevant to the resumed question", ), ]) records = [_memory("mem-0001"), _memory("mem-0003")] searcher = Searcher([ SimpleNamespace(ref="mem-0003", similarity=0.9, metadata={"memory_revision": "r", "memory_format": 2}), SimpleNamespace(ref="mem-0001", similarity=0.8, metadata={"memory_revision": "r", "memory_format": 2}), ]) embedder = Embedder() cards = {r.id: Card(id=r.id, family="domain_clarification", subject=r.subject, detail=r.detail, scope="psd-clinical", workspace_id="psd-clinical", origin="workflow", created_at=r.ts, updated_at=r.ts, revision="r", indexed=True) for r in records} archive = SimpleNamespace(list=lambda query: {"total": len(cards)}, get=lambda identity: cards[identity], close=lambda: None) archive.operation = lambda: nullcontext(archive) service = MemoryService(archive, PrincipalContext(issuer="local", subject="reviewer"), store_factory=lambda: None, embedder_factory=Embedder) monkeypatch.setattr("tht.cli.memory_cmd.memory_service", lambda cfg: service) monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", lambda cfg: searcher) monkeypatch.setattr("tht.cli.vector_cmd.make_embedder", lambda cfg: embedder) response = CliRunner().invoke( app, [ "memory", "search", "active patients", "--session", session_id, "--json", "-c", str(_workspace_config(tmp_path)), ], ) assert response.exit_code == 0, response.output assert json.loads(response.stdout) == [] assert current_phase(repository.get(session_id)) == 2 assert embedder.questions == ["active patients"] assert searcher.calls == [([0.1, 0.2], 20, ["memory"])] def test_empty_archive_recall_does_not_require_vector_projection(tmp_path, monkeypatch): """A fresh installation can have an empty dense-only Memory collection.""" from tht.ports.vector import VectorStoreError class UnavailableSearcher: def search(self, *args, **kwargs): raise VectorStoreError("BM25 collection configuration mismatch") archive = SimpleNamespace(list=lambda query: {"items": [], "total": 0}, close=lambda: None) service = MemoryService(archive, PrincipalContext(issuer="local", subject="reviewer"), store_factory=lambda: None, embedder_factory=Embedder) embedder = Embedder() monkeypatch.setattr("tht.cli.memory_cmd.memory_service", lambda cfg: service) monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", lambda cfg: UnavailableSearcher()) monkeypatch.setattr("tht.cli.vector_cmd.make_embedder", lambda cfg: embedder) for command in ("search", "solved-search"): response = CliRunner().invoke(app, ["memory", command, "active patients", "--json", "-c", str(_workspace_config(tmp_path))]) assert response.exit_code == 0, response.output assert json.loads(response.stdout) == [] assert embedder.questions == [] def test_recall_does_not_hide_an_unavailable_authoritative_archive(): import pytest from tht.memory.models import MemoryUnavailable def unavailable(query): raise MemoryUnavailable("Memory archive is unavailable") archive = SimpleNamespace(list=unavailable) service = MemoryService(archive, PrincipalContext(issuer="local", subject="reviewer"), store_factory=lambda: None, embedder_factory=Embedder) embedder = Embedder() with pytest.raises(MemoryUnavailable): service.recall("active patients", searcher=Searcher([]), embedder=embedder) assert embedder.questions == []