116 lines
4.6 KiB
Python
116 lines
4.6 KiB
Python
"""L1: gate di promozione memorie (F8) — tipi di decisione + filtro candidati.
|
|
|
|
La promozione era solo CLI facoltativa (mai innescata): il gate reviewer_memory_promote
|
|
la rende un passo del workflow. Questi test fissano il contratto harness-side:
|
|
- i due nuovi decision type esistono e sono ammessi dalla Fase 8 (workflow.yaml emits)
|
|
- i candidati rifiutati al gate (memory_promotion_declined, detail "seq:<n>") non
|
|
vengono riproposti da reusable_promotions/preview_promotions.
|
|
"""
|
|
from datetime import datetime
|
|
|
|
from tht.decisions import DecisionRecord, append_decision
|
|
from tht.memory import (
|
|
MemoryRecord,
|
|
declined_promotion_seqs,
|
|
memory_vector_records,
|
|
promote,
|
|
reusable_promotions,
|
|
)
|
|
from tht.session.models import SessionManifest
|
|
from tht.workflow import load_workflow
|
|
|
|
|
|
def test_promotion_decision_types_are_valid():
|
|
for t in ("memory_promoted", "memory_promotion_declined"):
|
|
d = DecisionRecord(
|
|
seq=1, ts=datetime(2026, 1, 1), type=t, subject="fact_x", detail="seq:5"
|
|
)
|
|
assert d.type == t
|
|
|
|
|
|
def test_promotion_decision_types_min_phase_is_f8():
|
|
wf = load_workflow()
|
|
assert wf.decision_min_phase("memory_promoted") == 8
|
|
assert wf.decision_min_phase("memory_promotion_declined") == 8
|
|
|
|
|
|
def _manifest() -> SessionManifest:
|
|
return SessionManifest(
|
|
id="s1", created_at=datetime(2026, 1, 1), question="domanda originale",
|
|
database="db", schema="public",
|
|
)
|
|
|
|
|
|
def test_declined_promotion_seqs_parses_seq_detail():
|
|
d = DecisionRecord(
|
|
seq=9, ts=datetime(2026, 1, 1), type="memory_promotion_declined",
|
|
subject="fact_x", detail="seq:5",
|
|
)
|
|
assert declined_promotion_seqs([d]) == {5}
|
|
|
|
|
|
def test_declined_promotion_seqs_ignores_malformed_and_other_types():
|
|
ds = [
|
|
DecisionRecord(seq=1, ts=datetime(2026, 1, 1),
|
|
type="memory_promotion_declined", subject="x", detail=""),
|
|
DecisionRecord(seq=2, ts=datetime(2026, 1, 1),
|
|
type="table_promoted", subject="x", detail="seq:3"),
|
|
]
|
|
assert declined_promotion_seqs(ds) == set()
|
|
|
|
|
|
def test_reusable_promotions_exclude_declined(tmp_path):
|
|
append_decision(tmp_path, type="table_promoted", subject="fact_a",
|
|
detail="tab principale", rationale="scelta reviewer") # seq 1
|
|
append_decision(tmp_path, type="concept_clarified", subject="attivo",
|
|
detail="flag_attivo = TRUE") # seq 2
|
|
append_decision(tmp_path, type="memory_promotion_declined",
|
|
subject="fact_a", detail="seq:1") # seq 3
|
|
cand = reusable_promotions(tmp_path, _manifest(), tmp_path / "registry.jsonl")
|
|
assert [c.decision_seq for c in cand] == [2]
|
|
|
|
|
|
def test_reusable_promotions_deduplicate_identical_memory_content(tmp_path):
|
|
append_decision(tmp_path, type="concept_clarified", subject="paziente attivo",
|
|
detail="flag_attivo = TRUE", rationale="scelta reviewer")
|
|
append_decision(tmp_path, type="concept_clarified", subject="paziente attivo",
|
|
detail="flag_attivo = TRUE", rationale="scelta reviewer")
|
|
|
|
cand = reusable_promotions(tmp_path, _manifest(), tmp_path / "registry.jsonl")
|
|
|
|
assert [c.decision_seq for c in cand] == [1]
|
|
|
|
|
|
def test_only_concept_clarified_is_proposed_or_promoted(tmp_path):
|
|
append_decision(tmp_path, type="table_promoted", subject="fact_a",
|
|
detail="tabella principale")
|
|
append_decision(tmp_path, type="table_excluded", subject="fact_b",
|
|
detail="tabella non pertinente")
|
|
append_decision(tmp_path, type="concept_clarified", subject="paziente attivo",
|
|
detail="flag_attivo = TRUE")
|
|
registry = tmp_path / "registry.jsonl"
|
|
|
|
candidates = reusable_promotions(tmp_path, _manifest(), registry)
|
|
promoted = promote(tmp_path, _manifest(), seqs=[1, 2, 3], registry_path=registry)
|
|
|
|
assert [(c.decision_seq, c.type) for c in candidates] == [(3, "concept_clarified")]
|
|
assert [(c.decision_seq, c.type) for c in promoted] == [(3, "concept_clarified")]
|
|
|
|
|
|
def test_legacy_table_records_are_not_published_as_memory_vectors():
|
|
records = [
|
|
MemoryRecord(
|
|
id="mem-0001", ts=datetime(2026, 1, 1), session_id="s1",
|
|
decision_seq=1, type="table_promoted", subject="fact_a",
|
|
),
|
|
MemoryRecord(
|
|
id="mem-0002", ts=datetime(2026, 1, 1), session_id="s1",
|
|
decision_seq=2, type="concept_clarified", subject="paziente attivo",
|
|
detail="flag_attivo = TRUE",
|
|
),
|
|
]
|
|
|
|
vectors = memory_vector_records(records)
|
|
|
|
assert [record.ref for record in vectors] == ["mem-0002"]
|