marcopanandClaude Fable 5
1ab0015460
Audit findings 5.1-5.3.
5.1 `phase reopen` now appends `phase_reopened` BEFORE the artifact
teardown: a crash between the two used to leave later-phase artifacts
deleted with the ledger still at the old phase (resume entered a phase
missing its artifacts). The inverse half-state — reopened with stale later
artifacts — is benign. Order locked by tests/test_phase_reopen_order.py.
5.2 New `tht decision add-batch --doc -`: N substantive decisions in ONE
atomic ledger write (meta types and cte_approved stay on `decision add`;
strictest min-phase enforced). reviewer_schema_linking now builds the
complete curation set and persists it with a single add-batch call — a
mid-loop failure can no longer leave the audit ledger half-written, and a
retry cannot duplicate the first K decisions.
5.3 The anti-bypass hook now also blocks BASH mutations of protected
state (`echo >> review_decisions.jsonl`, `sed -i` on the manifest,
`cat > tht-gate.js`, python open('w'), mv/rm/tee/…): FORBIDDEN only
covered tht subcommands and the write/edit hook only covered pi's own
tools. Read-only access (cat/grep/tail/ls) stays allowed.
Also: knownDecisionTypes is defensive — a workflow meta declaring NO
emits at all (older tht, minimal stubs) skips pre-validation instead of
rejecting every substantive type; with emits present, unknown types are
still rejected before the widget (new L1 test).
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
127 lines
4.5 KiB
Python
127 lines
4.5 KiB
Python
"""`decision add-batch`: N decisioni sostanziali in un'unica scrittura atomica.
|
|
|
|
Contratto: o TUTTE le decisioni entrano nel ledger o NESSUNA — un item invalido
|
|
rigetta l'intero batch (il retry del gate non può creare duplicati parziali).
|
|
"""
|
|
|
|
import json
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli.decision_cmd import decision_app
|
|
from tht.decisions import append_decision, append_decisions, list_decisions
|
|
from tht.phase import current_phase
|
|
|
|
|
|
def _walk_to_phase(session, target):
|
|
while current_phase(session) < target:
|
|
append_decision(session, type="phase_approved", subject=f"phase:{current_phase(session)}")
|
|
|
|
|
|
def _configure_command(monkeypatch, sessions):
|
|
import tht.cli.decision_cmd as mod
|
|
import tht.cli.session_cmd as session_mod
|
|
from tht.session.models import SessionManifest, SessionSnapshot
|
|
|
|
class _Repository:
|
|
def get(self, session_id):
|
|
return SessionSnapshot(
|
|
manifest=SessionManifest(
|
|
id=session_id, created_at="2026-01-01T00:00:00Z",
|
|
question="q", database="d", schema="s",
|
|
),
|
|
decisions=list_decisions(sessions / session_id),
|
|
)
|
|
|
|
def append_decisions(self, session_id, decisions):
|
|
return append_decisions(sessions / session_id, list(decisions))
|
|
|
|
class _Cfg:
|
|
pass
|
|
|
|
repository = _Repository()
|
|
monkeypatch.setattr(mod, "_load_config_or_exit", lambda _c: _Cfg())
|
|
monkeypatch.setattr(mod, "load_session_or_exit", lambda _cfg, _s: None)
|
|
monkeypatch.setattr(mod, "load_snapshot_or_exit", lambda _cfg, sid: repository.get(sid))
|
|
monkeypatch.setattr(mod, "session_repository", lambda _cfg: repository)
|
|
monkeypatch.setattr(session_mod, "load_snapshot_or_exit", lambda _cfg, sid: repository.get(sid))
|
|
|
|
|
|
SCHEMA_LINKING_BATCH = [
|
|
{"type": "table_promoted", "subject": "fact_impianto_pmk", "detail": "impianti PMK"},
|
|
{"type": "column_promoted", "subject": "fact_impianto_pmk.cod_paz"},
|
|
{"type": "column_excluded", "subject": "fact_impianto_pmk.note"},
|
|
{"type": "table_excluded", "subject": "dim_obsoleta", "rationale": "non pertinente"},
|
|
]
|
|
|
|
|
|
def test_add_batch_appends_all_decisions_in_one_write(tmp_path, monkeypatch):
|
|
session = tmp_path / "s1"
|
|
session.mkdir()
|
|
_walk_to_phase(session, 4)
|
|
_configure_command(monkeypatch, tmp_path)
|
|
|
|
result = CliRunner().invoke(
|
|
decision_app,
|
|
["add-batch", "--session", "s1", "--doc", "-"],
|
|
input=json.dumps(SCHEMA_LINKING_BATCH),
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
recorded = [d for d in list_decisions(session) if not d.type.startswith("phase_")]
|
|
assert [(d.type, d.subject) for d in recorded] == [
|
|
(item["type"], item["subject"]) for item in SCHEMA_LINKING_BATCH
|
|
]
|
|
|
|
|
|
def test_add_batch_rejects_the_whole_batch_when_one_item_is_invalid(tmp_path, monkeypatch):
|
|
session = tmp_path / "s1"
|
|
session.mkdir()
|
|
_walk_to_phase(session, 4)
|
|
_configure_command(monkeypatch, tmp_path)
|
|
before = len(list_decisions(session))
|
|
|
|
bad = SCHEMA_LINKING_BATCH + [{"type": "tipo_inesistente", "subject": "x"}]
|
|
result = CliRunner().invoke(
|
|
decision_app,
|
|
["add-batch", "--session", "s1", "--doc", "-"],
|
|
input=json.dumps(bad),
|
|
)
|
|
|
|
assert result.exit_code == 1
|
|
assert len(list_decisions(session)) == before
|
|
|
|
|
|
def test_add_batch_rejects_meta_and_cte_approved_types(tmp_path, monkeypatch):
|
|
session = tmp_path / "s1"
|
|
session.mkdir()
|
|
_walk_to_phase(session, 4)
|
|
_configure_command(monkeypatch, tmp_path)
|
|
before = len(list_decisions(session))
|
|
|
|
for forbidden in ("phase_approved", "decision_retracted", "cte_approved"):
|
|
result = CliRunner().invoke(
|
|
decision_app,
|
|
["add-batch", "--session", "s1", "--doc", "-"],
|
|
input=json.dumps([{"type": forbidden, "subject": "x"}]),
|
|
)
|
|
assert result.exit_code == 1, forbidden
|
|
assert len(list_decisions(session)) == before
|
|
|
|
|
|
def test_add_batch_enforces_the_strictest_min_phase(tmp_path, monkeypatch):
|
|
session = tmp_path / "s1"
|
|
session.mkdir()
|
|
_walk_to_phase(session, 2) # column_* richiede la fase di schema linking (4)
|
|
_configure_command(monkeypatch, tmp_path)
|
|
before = len(list_decisions(session))
|
|
|
|
result = CliRunner().invoke(
|
|
decision_app,
|
|
["add-batch", "--session", "s1", "--doc", "-"],
|
|
input=json.dumps([{"type": "column_promoted", "subject": "t.c"}]),
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert len(list_decisions(session)) == before
|