test(harness): L1 session-coherence smoke -- full walk + rollback (D1)

Pure-logic smoke (no LLM, no DB) that builds a synthetic ledger by hand and asserts
the Phase-A substrate stays coherent: full F1->F8 walk reaches terminal phase
(max+1); rollback truncates the effective view (stale phase-7 decision excluded
after reopen to F4) and resets current_phase; teardown deletes artifacts beyond the
target while preserving the target phase's; re-approve after rollback advances
correctly; taskdoc stays under byte budget across all phases; decision_retraction
excludes the retracted seq + the marker itself from effective_decisions.

This is the CI-runnable coherence net for the L2 session test (which exercises the
LLM->gate loop that L1 cannot).
This commit is contained in:
2026-06-26 23:14:57 +02:00
parent 60b6e38096
commit d5c0fffc2a
@@ -0,0 +1,106 @@
"""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, taskdoc byte budget).
"""
from pathlib import Path
from nsp.decisions import append_decision
from nsp.phase import current_phase, effective_decisions
from nsp.taskdoc import generate_task_doc
from nsp.teardown import teardown_to_phase
from nsp.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_task_doc_stays_under_byte_budget_across_phases(tmp_path):
s = tmp_path / "sess"
s.mkdir()
(s / "question.md").write_text("# Domanda\ndammi i pazienti con ablazione nel 2025")
(s / "schema_linking.json").write_text('{"candidates":[{"kind":"table","name":"fct_ricoveri"}]}')
for ph in range(1, 9):
doc = generate_task_doc(session_dir=s, phase=ph, promoted_tables=["fct_ricoveri"])
assert doc.byte_budget_ok, f"phase {ph} task doc over byte budget"
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