96 lines
4.1 KiB
Python
96 lines
4.1 KiB
Python
"""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
|