250 lines
7.9 KiB
Python
250 lines
7.9 KiB
Python
"""Executable baseline for persisted workflow behavior touched by the refactor."""
|
|
|
|
from dataclasses import asdict
|
|
from datetime import UTC, datetime
|
|
import hashlib
|
|
|
|
import pytest
|
|
|
|
from tht.corpus.models import CanonicalDocument, CorpusManifest
|
|
from tht.corpus.store import CorpusStore
|
|
from tht.decisions import DecisionRecord, append_decision
|
|
from tht.phase import current_phase, effective_decisions
|
|
from tht.session.artifacts import build_evidence_entries
|
|
from tht.session.models import Candidate, SchemaLinking
|
|
from tht.workflow import load_workflow
|
|
|
|
|
|
def test_workflow_definition_has_the_approved_semantic_contract():
|
|
workflow = load_workflow()
|
|
|
|
assert workflow.schema_version == 1
|
|
assert workflow.max_phase == 8
|
|
assert [asdict(phase) for phase in workflow.phases] == [
|
|
{
|
|
"id": "F1",
|
|
"num": 1,
|
|
"name": "chiarimento",
|
|
"advance": "kind:phase",
|
|
"prerequisites": [],
|
|
"artifacts_out": [],
|
|
"emits": ["concept_clarified", "ambiguity_open"],
|
|
},
|
|
{
|
|
"id": "F2",
|
|
"num": 2,
|
|
"name": "memoria",
|
|
"advance": "auto_if_empty",
|
|
"prerequisites": [],
|
|
"artifacts_out": [],
|
|
"emits": ["memory_rejected", "concept_clarified"],
|
|
},
|
|
{
|
|
"id": "F3",
|
|
"num": 3,
|
|
"name": "riscrittura",
|
|
"advance": "kind:phase",
|
|
"prerequisites": [{"decision_exists": "question_rewritten"}],
|
|
"artifacts_out": ["question.md"],
|
|
"emits": ["question_rewritten"],
|
|
},
|
|
{
|
|
"id": "F4",
|
|
"num": 4,
|
|
"name": "schema_linking",
|
|
"advance": "reviewer_decide",
|
|
"prerequisites": [],
|
|
"artifacts_out": ["schema_linking.json"],
|
|
"emits": [
|
|
"table_promoted",
|
|
"table_excluded",
|
|
"column_promoted",
|
|
"column_excluded",
|
|
"column_corrected",
|
|
"join_modified",
|
|
"evidence_accepted",
|
|
"evidence_rejected",
|
|
"value_grounded",
|
|
"concept_formula_approved",
|
|
"concept_formula_rejected",
|
|
],
|
|
},
|
|
{
|
|
"id": "F5",
|
|
"num": 5,
|
|
"name": "sintesi",
|
|
"advance": "kind:phase",
|
|
"prerequisites": [
|
|
{"file_validates": ["schema_linking.json", "SchemaLinking"]}
|
|
],
|
|
"artifacts_out": [],
|
|
"emits": [],
|
|
},
|
|
{
|
|
"id": "F6",
|
|
"num": 6,
|
|
"name": "cte",
|
|
"advance": "auto_if_empty_or_skipped",
|
|
"prerequisites": [
|
|
{
|
|
"any": [
|
|
{"decision_subject_exists": ["phase_skipped", "phase:6"]},
|
|
{"all_ctes_approved": True},
|
|
]
|
|
}
|
|
],
|
|
"artifacts_out": ["cte_plan.json", "ctes/", "cte_tests.json"],
|
|
"emits": ["cte_approved", "cte_corrected", "cte_rejected"],
|
|
},
|
|
{
|
|
"id": "F7",
|
|
"num": 7,
|
|
"name": "sql_finale",
|
|
"advance": "kind:phase",
|
|
"prerequisites": [{"decision_exists": "sql_approved"}],
|
|
"artifacts_out": ["sql_final.sql"],
|
|
"emits": ["sql_revised", "sql_approved", "sql_rejected"],
|
|
},
|
|
{
|
|
"id": "F8",
|
|
"num": 8,
|
|
"name": "datamart",
|
|
"advance": "reviewer_decide",
|
|
"prerequisites": [
|
|
{
|
|
"any": [
|
|
{"decision_exists": "datamart_requested"},
|
|
{"decision_exists": "datamart_declined"},
|
|
]
|
|
}
|
|
],
|
|
"artifacts_out": [],
|
|
"emits": [
|
|
"datamart_requested",
|
|
"datamart_declined",
|
|
"memory_promoted",
|
|
"memory_promotion_declined",
|
|
],
|
|
},
|
|
]
|
|
|
|
|
|
def _record(seq: int, type_: str, subject: str) -> DecisionRecord:
|
|
return DecisionRecord(
|
|
seq=seq,
|
|
ts=datetime(2026, 8, 24, tzinfo=UTC),
|
|
type=type_,
|
|
subject=subject,
|
|
)
|
|
|
|
|
|
def _linking(*evidence_ids: str, decision_seq: int = 17) -> SchemaLinking:
|
|
return SchemaLinking(
|
|
question="q",
|
|
candidates=[
|
|
Candidate(
|
|
kind="table",
|
|
name="fact_procedure",
|
|
evidence=list(evidence_ids),
|
|
decision="promoted",
|
|
decision_seq=decision_seq,
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def test_schema_linking_evidence_used_resolves_from_the_active_canonical_corpus(tmp_path):
|
|
content = "# Curated definition\n"
|
|
digest = hashlib.sha256(content.encode()).hexdigest()
|
|
document = CanonicalDocument(
|
|
document_id=f"doc:{digest}",
|
|
source_id="fs:evi-used",
|
|
source_uri="file:///curated/evi-used.md",
|
|
source_fingerprint=f"sha256:{'a' * 64}",
|
|
content_hash=f"sha256:{digest}",
|
|
content=content,
|
|
pipeline_version="evidence-v1",
|
|
metadata={"frontmatter": {"id": "evi-used"}},
|
|
)
|
|
store = CorpusStore(tmp_path / "corpus")
|
|
generation = store.stage(
|
|
CorpusManifest(documents=(document,)),
|
|
{document.document_id: content},
|
|
)
|
|
store.publish(generation)
|
|
|
|
entries = build_evidence_entries(
|
|
[],
|
|
_linking("evi-used"),
|
|
tmp_path / "artifacts" / "evidence",
|
|
)
|
|
|
|
assert entries == [{
|
|
"id": "evi-used",
|
|
"file": str(
|
|
tmp_path / "artifacts" / ".materialized-evidence" / f"{digest}.md"
|
|
),
|
|
"esito": "usata",
|
|
"decision_seq": 17,
|
|
}]
|
|
|
|
|
|
def test_legacy_evidence_without_a_canonical_corpus_keeps_used_and_reviewed_outcomes(tmp_path):
|
|
evidence_root = tmp_path / "artifacts" / "evidence"
|
|
evidence_root.mkdir(parents=True)
|
|
for evidence_id in ("evi-used", "evi-accepted", "evi-rejected"):
|
|
(evidence_root / f"{evidence_id}.md").write_text(f"# {evidence_id}\n")
|
|
|
|
entries = build_evidence_entries(
|
|
[
|
|
_record(21, "evidence_accepted", "evi-accepted"),
|
|
_record(22, "evidence_rejected", "evi-rejected"),
|
|
],
|
|
_linking("evi-used", "evi-accepted"),
|
|
evidence_root,
|
|
)
|
|
|
|
assert [(entry["id"], entry["esito"], entry["decision_seq"]) for entry in entries] == [
|
|
("evi-used", "usata", 17),
|
|
("evi-accepted", "accettata", 21),
|
|
("evi-rejected", "scartata", 22),
|
|
]
|
|
assert all(entry["file"].endswith(f"{entry['id']}.md") for entry in entries)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("approved_through", "phase_decisions", "expected_phase"),
|
|
[
|
|
(0, [("concept_clarified", "ablazione")], 1),
|
|
(1, [("concept_clarified", "paziente attivo")], 2),
|
|
(2, [("question_rewritten", "domanda")], 3),
|
|
(3, [("evidence_accepted", "evi-7")], 4),
|
|
(
|
|
7,
|
|
[
|
|
("datamart_declined", "phase:8"),
|
|
("memory_promotion_declined", "paziente attivo"),
|
|
],
|
|
8,
|
|
),
|
|
],
|
|
ids=["F1", "F2", "F3", "F4-Evidence", "F8"],
|
|
)
|
|
def test_resume_reconstructs_each_touched_open_phase(
|
|
tmp_path, approved_through, phase_decisions, expected_phase
|
|
):
|
|
session_dir = tmp_path / f"resume-f{expected_phase}"
|
|
session_dir.mkdir()
|
|
for phase in range(1, approved_through + 1):
|
|
append_decision(
|
|
session_dir,
|
|
type="phase_approved",
|
|
subject=f"phase:{phase}",
|
|
)
|
|
for decision_type, subject in phase_decisions:
|
|
append_decision(session_dir, type=decision_type, subject=subject)
|
|
|
|
assert current_phase(session_dir) == expected_phase
|
|
effective = {(decision.type, decision.subject) for decision in effective_decisions(session_dir)}
|
|
assert set(phase_decisions) <= effective
|