From bf7e850e6bc954b43b4f876d2e3e4f5baf82b26b Mon Sep 17 00:00:00 2001 From: mptyl Date: Tue, 7 Jul 2026 13:26:59 +0200 Subject: [PATCH] feat(memory): filter gate-declined candidates from promotion preview Co-Authored-By: Claude Fable 5 --- harness/tests/test_memory_promotion.py | 40 +++++++++++++++++++++++++- harness/tht/memory.py | 33 +++++++++++++++++---- 2 files changed, 67 insertions(+), 6 deletions(-) diff --git a/harness/tests/test_memory_promotion.py b/harness/tests/test_memory_promotion.py index acde0b81..1bd867fd 100644 --- a/harness/tests/test_memory_promotion.py +++ b/harness/tests/test_memory_promotion.py @@ -8,7 +8,9 @@ la rende un passo del workflow. Questi test fissano il contratto harness-side: """ from datetime import datetime -from tht.decisions import DecisionRecord +from tht.decisions import DecisionRecord, append_decision +from tht.memory import declined_promotion_seqs, reusable_promotions +from tht.session.models import SessionManifest from tht.workflow import load_workflow @@ -24,3 +26,39 @@ 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] diff --git a/harness/tht/memory.py b/harness/tht/memory.py index 23b24f4d..e081b478 100644 --- a/harness/tht/memory.py +++ b/harness/tht/memory.py @@ -42,6 +42,23 @@ def decided_memory_ids(decisions: list[DecisionRecord]) -> set[str]: return out +_DECLINED_SEQ_RE = re.compile(r"\bseq:(\d+)\b") + + +def declined_promotion_seqs(decisions: list[DecisionRecord]) -> set[int]: + """decision_seq dei candidati che il reviewer ha rifiutato al gate di promozione + (F8, `memory_promotion_declined` con detail "seq:"): una riapertura del gate + non deve riproporli. I promossi sono gia' dedupati dal registro.""" + out: set[int] = set() + for d in decisions: + if d.type != "memory_promotion_declined": + continue + m = _DECLINED_SEQ_RE.search(d.detail or "") + if m: + out.add(int(m.group(1))) + return out + + def load_registry(registry_path: Path) -> list[MemoryRecord]: if not registry_path.exists(): return [] @@ -112,7 +129,7 @@ def _default_concepts(decision: DecisionRecord) -> list[str]: return [] -def _question_context(decisions: list[DecisionRecord], manifest: SessionManifest) -> str: +def question_context(decisions: list[DecisionRecord], manifest: SessionManifest) -> str: rewritten = [d for d in decisions if d.type == "question_rewritten"] return rewritten[-1].detail if rewritten else manifest.question @@ -139,7 +156,7 @@ def _compute_promotions( selected = decisions if seqs is None else [d for d in decisions if d.seq in seqs] already = {(r.session_id, r.decision_seq) for r in existing} n = _next_id_num(existing) - context = _question_context(decisions, manifest) + context = question_context(decisions, manifest) out: list[MemoryRecord] = [] for d in selected: if (manifest.id, d.seq) in already: @@ -172,12 +189,18 @@ def promote( def reusable_promotions( session_dir: Path, manifest: SessionManifest, registry_path: Path ) -> list[MemoryRecord]: - """Candidati riusabili (tipi in REUSABLE_TYPES) non ancora promossi, SENZA - cap: il chiamante applica MAX_PROMOTION_CANDIDATES e segnala il troncamento.""" + """Candidati riusabili (tipi in REUSABLE_TYPES) non ancora promossi ne' rifiutati + al gate, SENZA cap: il chiamante applica MAX_PROMOTION_CANDIDATES e segnala il + troncamento.""" + from tht.phase import effective_decisions + cand = _compute_promotions( session_dir, manifest, seqs=None, existing=load_registry(registry_path) ) - return [c for c in cand if c.type in REUSABLE_TYPES] + declined = declined_promotion_seqs(effective_decisions(session_dir)) + return [ + c for c in cand if c.type in REUSABLE_TYPES and c.decision_seq not in declined + ] def preview_promotions(