diff --git a/harness/tests/test_session_coherence_smoke.py b/harness/tests/test_session_coherence_smoke.py new file mode 100644 index 00000000..ef6f908c --- /dev/null +++ b/harness/tests/test_session_coherence_smoke.py @@ -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