feat(harness): Workflow.schema_linking_phase() (Onda 1.1, TDD)
Aggiunge il metodo che i cmd CLI useranno al posto della vecchia costante SCHEMA_LINKING_PHASE (drift fix Onda 1). Ritorna il num della fase il cui artifacts_out contiene schema_linking.json, default 5 se nessuna la dichiara. L1: 4 test (fase reale F4, posizione arbitraria, default 5, artefatti multipli). Suite: 157 passed.
This commit is contained in:
@@ -0,0 +1,53 @@
|
||||
"""L1: Workflow.schema_linking_phase() — the phase that produces schema_linking.json.
|
||||
|
||||
Used by the CLI drift fix (the old SCHEMA_LINKING_PHASE constant is gone; commands
|
||||
now ask the Workflow for it). Returns the num of the phase whose artifacts_out
|
||||
contains schema_linking.json, defaulting to 5 if no phase declares it.
|
||||
"""
|
||||
from tht.workflow import PhaseSpec, Workflow
|
||||
|
||||
|
||||
def _wf(phases: list[PhaseSpec]) -> Workflow:
|
||||
return Workflow(schema_version=1, phases=phases, _decision_min_map={})
|
||||
|
||||
|
||||
def test_schema_linking_phase_returns_num_of_phase_with_artifact():
|
||||
# the real workflow: F4 schema_linking produces schema_linking.json
|
||||
from tht.workflow import load_workflow
|
||||
|
||||
wf = load_workflow()
|
||||
assert wf.schema_linking_phase() == 4
|
||||
|
||||
|
||||
def test_schema_linking_phase_finds_it_wherever_it_is():
|
||||
# synthetic: schema_linking.json declared on phase 3
|
||||
phases = [
|
||||
PhaseSpec(id="F1", num=1, name="a", advance="kind:phase", prerequisites=[]),
|
||||
PhaseSpec(id="F2", num=2, name="b", advance="kind:phase", prerequisites=[]),
|
||||
PhaseSpec(
|
||||
id="F3", num=3, name="c", advance="kind:phase",
|
||||
prerequisites=[], artifacts_out=["schema_linking.json"],
|
||||
),
|
||||
]
|
||||
assert _wf(phases).schema_linking_phase() == 3
|
||||
|
||||
|
||||
def test_schema_linking_phase_defaults_to_5_when_no_artifact_declared():
|
||||
# defensive: if no phase declares schema_linking.json, fall back to 5 (the
|
||||
# conventional schema-linking phase) rather than raising.
|
||||
phases = [
|
||||
PhaseSpec(id="F1", num=1, name="a", advance="kind:phase", prerequisites=[]),
|
||||
PhaseSpec(id="F2", num=2, name="b", advance="kind:phase", prerequisites=[]),
|
||||
]
|
||||
assert _wf(phases).schema_linking_phase() == 5
|
||||
|
||||
|
||||
def test_schema_linking_phase_with_multiple_artifacts_finds_the_right_one():
|
||||
# a phase can produce multiple artifacts; schema_linking.json among them
|
||||
phases = [
|
||||
PhaseSpec(
|
||||
id="F5", num=5, name="sintesi", advance="kind:phase",
|
||||
prerequisites=[], artifacts_out=["cte_plan.json", "schema_linking.json"],
|
||||
),
|
||||
]
|
||||
assert _wf(phases).schema_linking_phase() == 5
|
||||
@@ -50,6 +50,15 @@ class Workflow:
|
||||
reference it (via decision_exists / decision_subject_exists). Defaults to 1."""
|
||||
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.
|
||||
|
||||
Reference in New Issue
Block a user