marcopan
4a1272fa0a
- workflow.yaml: 8-phase definition, data-driven prerequisites, no hardcoded ladder - workflow.py: load_workflow() reader; max_phase=len(phases), decision_min_phase derived from prerequisites scan (no duplication) - 6 tests: phase count, decision_min_phase derivation, name lookup (incl. F8 presence -- the JS drift bug structurally impossible now), artifacts_out, advance strategy, schema_version. All 8 harness tests pass.
108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
"""Reads workflow.yaml -- the SINGLE source of workflow truth (spec F2, §5.3).
|
|
|
|
phase.py, the gate (nsp-gate.js), and the skill all read from here.
|
|
No more duplicated constants (the JS/Python drift bug in ChironeWp3 -- 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)
|
|
|
|
|
|
@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 whose prerequisites
|
|
reference it (via decision_exists / decision_subject_exists). Defaults to 1."""
|
|
return self._decision_min_map.get(decision_type, 1)
|
|
|
|
|
|
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)
|
|
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", []),
|
|
)
|
|
)
|
|
return Workflow(
|
|
schema_version=raw.get("schema_version", 1),
|
|
phases=phases,
|
|
_decision_min_map=_collect_decision_mins(phases),
|
|
)
|