ThothII/harness/tests/test_workflow_observable_contract.py

"""Executable baseline for persisted workflow behavior touched by the refactor."""

import hashlib
from dataclasses import asdict
from datetime import UTC, datetime

import pytest

from tht.decisions import DecisionRecord, append_decision
from tht.evidence import project_session
from tht.evidence.corpus.models import CanonicalDocument, CorpusManifest
from tht.evidence.corpus.store import CorpusStore
from tht.phase import current_phase, effective_decisions
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": [
                {"decision_exists": "memory_summary_reviewed"},
                {
                    "any": [
                        {"decision_exists": "datamart_requested"},
                        {"decision_exists": "datamart_declined"},
                    ]
                }
            ],
            "artifacts_out": [],
            "emits": [
                "datamart_requested",
                "datamart_declined",
                "memory_promoted",
                "memory_promotion_declined",
                "memory_summary_reviewed",
            ],
        },
    ]


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 = project_session(
        [],
        _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 = project_session(
        [
            _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