ThothII/harness/tht/workflow.py

"""Reads workflow.yaml -- the SINGLE source of workflow truth (spec F2, §5.3).

phase.py, the gate (tht-gate.js), and the skill all read from here.
No more duplicated constants (the JS/Python drift bug in the reference implementation -- PHASE_NAMES
truncated to 7 in JS -- is structurally impossible because there is one source).

Edit workflow.yaml to change the workflow: add/reorder/merge/skip phases.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import yaml

_WF_PATH = Path(__file__).resolve().parent.parent / "workflow.yaml"


@dataclass
class PhaseSpec:
    id: str
    num: int
    name: str
    advance: str
    prerequisites: list[Any]
    artifacts_out: list[str] = field(default_factory=list)
    emits: list[str] = field(default_factory=list)


@dataclass
class Workflow:
    schema_version: int
    phases: list[PhaseSpec]
    _decision_min_map: dict[str, int] = field(default_factory=dict)

    @property
    def max_phase(self) -> int:
        return len(self.phases)

    def phase_by_num(self, n: int) -> PhaseSpec:
        return self.phases[n - 1]

    def phase_name(self, n: int) -> str:
        if 1 <= n <= self.max_phase:
            return self.phase_by_num(n).name
        return "?"

    def decision_min_phase(self, decision_type: str) -> int:
        """A decision type's min phase = the earliest phase that EMITS it (workflow.yaml
        `emits`), or that references it in a prerequisite. Defaults to 1 (meta types like
        phase_approved/decision_retracted are not phase-gated)."""
        return self._decision_min_map.get(decision_type, 1)

    def schema_linking_phase(self) -> int:
        """The phase that produces schema_linking.json (artifacts_out). Default 5
        (the conventional schema-linking phase) if no phase declares it. Used by the
        CLI drift fix in place of the old SCHEMA_LINKING_PHASE constant."""
        for p in self.phases:
            if "schema_linking.json" in p.artifacts_out:
                return p.num
        return 5


def _collect_decision_mins(phases: list[PhaseSpec]) -> dict[str, int]:
    """Scan prerequisites for decision_exists / decision_subject_exists mentions.

    Supports both forms:
      - decision_exists: <type>            (scalar)
      - decision_exists: [<type>, ...]     (list, first element is the type)
      - decision_subject_exists: [<type>, <subject>]  (list, first element is the type)
    """
    mins: dict[str, int] = {}

    def scan(node: Any, phase_num: int) -> None:
        if isinstance(node, dict):
            for key, value in node.items():
                if key in ("decision_exists", "decision_subject_exists"):
                    if isinstance(value, list) and value:
                        dtype = value[0]
                    elif isinstance(value, str):
                        dtype = value
                    else:
                        continue
                    if isinstance(dtype, str):
                        if dtype not in mins or phase_num < mins[dtype]:
                            mins[dtype] = phase_num
                else:
                    scan(value, phase_num)
        elif isinstance(node, list):
            for item in node:
                scan(item, phase_num)

    for p in phases:
        scan(p.prerequisites, p.num)
        # `emits`: la fase dichiara i decision type che produce -> fonte primaria del
        # min phase (copre i tipi non citati nei prerequisites, es. table_promoted,
        # value_grounded, concept_formula_*). Vince la fase piu' bassa.
        for dtype in p.emits:
            if isinstance(dtype, str) and (dtype not in mins or p.num < mins[dtype]):
                mins[dtype] = p.num
    return mins


def load_workflow(path: Path | str = _WF_PATH) -> Workflow:
    path = Path(path)
    raw = yaml.safe_load(path.read_text())
    phases: list[PhaseSpec] = []
    for i, p in enumerate(raw["phases"], start=1):
        phases.append(
            PhaseSpec(
                id=p["id"],
                num=i,
                name=p["name"],
                advance=p["advance"],
                prerequisites=p.get("prerequisites", []),
                artifacts_out=p.get("artifacts_out", []),
                emits=p.get("emits", []),
            )
        )
    return Workflow(
        schema_version=raw.get("schema_version", 1),
        phases=phases,
        _decision_min_map=_collect_decision_mins(phases),
    )