feat(memory): filter gate-declined candidates from promotion preview

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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]

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:<n>"): 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(