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.
222 lines
7.9 KiB
Python
222 lines
7.9 KiB
Python
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 == []
|