feat(harness): decisions.py ported + decision_retracted for step rollback (D15)

- ported from ChironeWp3 (22 DecisionType, append-only jsonl, monotonic seq)
- added decision_retracted type + retracts field for step-level rollback (D15):
  the retracted decision stays in the audit log, effective_decisions() (Task A5)
  will exclude it from the active view
- 5 tests: retract marker + monotonic seq + retracts default + empty session +
  literal includes retracted. All 13 harness tests pass.
This commit is contained in:
2026-06-26 22:30:16 +02:00
parent 4a1272fa0a
commit fbcd694ace
2 changed files with 151 additions and 0 deletions
+85
View File
@@ -0,0 +1,85 @@
from datetime import UTC, datetime
from pathlib import Path
from typing import Literal
from pydantic import BaseModel
DECISIONS_FILE = "review_decisions.jsonl"
# 22 tipi di ChironeWp3 (verified leggendo session/decisions.py) + 1 nuovo (D15):
# `decision_retracted` per il rollback a granularità step (ritira una decisione
# senza cancellarne la riga dal log di audit; effective_decisions la onora).
DecisionType = Literal[
"concept_clarified",
"question_rewritten",
"table_promoted",
"table_excluded",
"column_corrected",
"join_modified",
"evidence_accepted",
"evidence_rejected",
"ambiguity_open",
"memory_rejected",
"cte_approved",
"cte_corrected",
"cte_rejected",
"sql_revised",
"sql_approved",
"sql_rejected",
"phase_approved",
"phase_auto_approved",
"phase_reopened",
"phase_skipped",
"datamart_requested",
"datamart_declined",
# D15: marker di ritrazione. subject = "phase:N", retracts = decision_seq ritirata.
# Resta nel log di audit (append-only); effective_decisions() la esclude dalla vista.
"decision_retracted",
]
class DecisionRecord(BaseModel):
seq: int
ts: datetime
type: DecisionType
subject: str
detail: str = ""
rationale: str = ""
# D15: se type == "decision_retracted", indica quale seq viene ritirata.
retracts: int | None = None
def list_decisions(session_dir: Path) -> list[DecisionRecord]:
path = session_dir / DECISIONS_FILE
if not path.exists():
return []
return [
DecisionRecord.model_validate_json(line)
for line in path.read_text().splitlines()
if line.strip()
]
def append_decision(
session_dir: Path,
*,
type: str,
subject: str,
detail: str = "",
rationale: str = "",
retracts: int | None = None,
) -> DecisionRecord:
record = DecisionRecord(
seq=len(list_decisions(session_dir)) + 1,
ts=datetime.now(UTC),
type=type,
subject=subject,
detail=detail,
rationale=rationale,
retracts=retracts,
)
path = session_dir / DECISIONS_FILE
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a") as f:
f.write(record.model_dump_json() + "\n")
return record
+66
View File
@@ -0,0 +1,66 @@
from nsp.decisions import append_decision, list_decisions
def test_decision_retracted_in_audit_but_marked(tmp_path):
"""Una decisione ritirata resta nel log di audit (append-only) ma porta il marker."""
session = tmp_path / "s1"
session.mkdir()
d1 = append_decision(
session, type="table_promoted", subject="phase:4", detail="t1", rationale="r"
)
d2 = append_decision(
session,
type="decision_retracted",
subject="phase:4",
detail="wrong answer, retry",
rationale="user picked back",
retracts=d1.seq,
)
all_decisions = list_decisions(session)
assert len(all_decisions) == 2 # entrambe nel log di audit
assert all_decisions[0].seq == 1
assert all_decisions[0].type == "table_promoted"
assert all_decisions[1].seq == 2
assert all_decisions[1].type == "decision_retracted"
assert all_decisions[1].retracts == 1 # marker verso la decisione ritirata
def test_append_decision_assigns_monotonic_seq(tmp_path):
session = tmp_path / "s2"
session.mkdir()
d1 = append_decision(session, type="phase_approved", subject="phase:1")
d2 = append_decision(session, type="phase_approved", subject="phase:2")
d3 = append_decision(session, type="question_rewritten", subject="phase:3", detail="new q")
assert (d1.seq, d2.seq, d3.seq) == (1, 2, 3)
assert d3.detail == "new q"
# re-read from disk: seq preserved
loaded = list_decisions(session)
assert [d.seq for d in loaded] == [1, 2, 3]
def test_retracts_defaults_none(tmp_path):
"""Le decisioni normali hanno retracts=None (non serializzato come null rumoroso)."""
session = tmp_path / "s3"
session.mkdir()
d = append_decision(session, type="concept_clarified", subject="phase:1", detail="x")
assert d.retracts is None
# round-trip via JSON
loaded = list_decisions(session)
assert loaded[0].retracts is None
def test_empty_session_returns_empty_list(tmp_path):
session = tmp_path / "empty"
session.mkdir()
assert list_decisions(session) == []
def test_decision_type_literal_includes_retracted():
"""decision_retracted e' un tipo valido (pydantic lo accetta)."""
from nsp.decisions import DecisionRecord
from datetime import datetime, UTC
d = DecisionRecord(
seq=1, ts=datetime.now(UTC), type="decision_retracted",
subject="phase:4", retracts=1,
)
assert d.type == "decision_retracted"