"""L1: session-coherence smoke -- full walk + rollback, pure logic (no LLM, no DB). The CI-runnable proxy for 'full session coherence': build a synthetic ledger by hand (decisions appended directly, not via an LLM) representing a full F1->F8 walk, then a rollback F6->F4 + re-derive, and assert the invariants the D1-the-L2-version would check. This is the coherence net for the Phase-A substrate (phase fold, effective view, teardown). """ from pathlib import Path from tht.decisions import append_decision from tht.phase import current_phase, effective_decisions from tht.teardown import teardown_to_phase from tht.workflow import load_workflow def _approve(session: Path, phase: int) -> None: append_decision(session, type="phase_approved", subject=f"phase:{phase}") def test_full_walk_reaches_terminal_phase(tmp_path): s = tmp_path / "sess" s.mkdir() wf = load_workflow() for n in range(1, wf.max_phase + 1): _approve(s, n) # terminal: all phases approved -> current_phase == max_phase + 1 assert current_phase(s) == wf.max_phase + 1 def test_rollback_truncates_effective_view_and_current_phase(tmp_path): s = tmp_path / "sess" s.mkdir() for n in range(1, 7): # approve F1..F6 _approve(s, n) # a substantive decision at phase 4 (preserved in the ledger) append_decision(s, type="table_promoted", subject="phase:4", detail="dim_pazienti") assert current_phase(s) == 7 # rollback to F4 append_decision(s, type="phase_reopened", subject="phase:4") assert current_phase(s) == 4 # decisions strictly after the reopen target are stale in the effective view. # The table_promoted was at phase:4 (<= current 4), so it survives; but a later # substantive decision appended AFTER the reopen for a phase > 4 must be excluded. append_decision(s, type="sql_approved", subject="phase:7") # stale: phase 7 > 4 eff_types = [d.type for d in effective_decisions(s)] assert "sql_approved" not in eff_types # rolled-back decision is stale assert "table_promoted" in eff_types # phase:4 decision is still effective def test_teardown_deletes_artifacts_beyond_rollback_target(tmp_path): s = tmp_path / "sess" s.mkdir() # artifacts as if F4..F7 had produced them (s / "schema_linking.json").write_text("{}") # F4 (s / "cte_plan.json").write_text("[]") # F6 (s / "sql_final.sql").write_text("SELECT 1") # F7 report = teardown_to_phase(s, target_phase=4) assert (s / "schema_linking.json").exists() # F4 preserved (target is 4) assert not (s / "cte_plan.json").exists() # F6 deleted assert not (s / "sql_final.sql").exists() # F7 deleted assert "cte_plan.json" in report.deleted_files assert "sql_final.sql" in report.deleted_files def test_re_approve_after_rollback_advances_correctly(tmp_path): s = tmp_path / "sess" s.mkdir() for n in range(1, 4): _approve(s, n) append_decision(s, type="phase_reopened", subject="phase:2") # back to F2 assert current_phase(s) == 2 _approve(s, 2) assert current_phase(s) == 3 _approve(s, 3) assert current_phase(s) == 4 def test_decision_retraction_excluded_from_effective(tmp_path): s = tmp_path / "sess" s.mkdir() # walk to phase 4 in order so current_phase reaches 4 (the fold only advances on # an approval for the CURRENT phase, n == cur). for n in range(1, 5): _approve(s, n) append_decision(s, type="table_promoted", subject="phase:4", detail="t1") append_decision(s, type="table_excluded", subject="phase:4", detail="t2") # retract the table_excluded (seq 6) -- step-level rollback, D15 append_decision(s, type="decision_retracted", subject="phase:4", retracts=6) assert current_phase(s) == 5 # phase 4 approved -> cursor 5 eff_types = [d.type for d in effective_decisions(s)] # the retracted table_excluded and the retraction marker are both gone assert eff_types.count("table_excluded") == 0 assert eff_types.count("decision_retracted") == 0 assert "table_promoted" in eff_types # unaffected by the unrelated retraction