127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
"""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) and (
|
|
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),
|
|
)
|