From ce465ac0681e461d425e8fa93bfdbb4c4d246953 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Filho?= Date: Tue, 12 May 2026 09:01:50 -0300 Subject: [PATCH 1/7] feat: add gate-driven pipeline advancement --- scripts/advance-pipeline.py | 412 ++++++++++++++++++++++++++++++++++++ 1 file changed, 412 insertions(+) create mode 100644 scripts/advance-pipeline.py diff --git a/scripts/advance-pipeline.py b/scripts/advance-pipeline.py new file mode 100644 index 0000000..8a453e3 --- /dev/null +++ b/scripts/advance-pipeline.py @@ -0,0 +1,412 @@ +#!/usr/bin/env python3 +"""Advance one SDD project phase when the current gate evidence passes.""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +EXIT_SUCCESS = 0 +EXIT_USER_ERROR = 1 +EXIT_RUNTIME_ERROR = 2 + +PHASE_SEQUENCE = [ + "RESEARCH", + "DISCUSS", + "SPEC", + "PLAN", + "EXECUTE", + "VERIFY", + "REVIEW", + "RELEASE", + "ARCHIVE", +] + + +class PipelineError(Exception): + """User-correctable pipeline input error.""" + + +class GateValidationError(PipelineError): + """Gate evidence failed validation.""" + + def __init__(self, failures: list[str]) -> None: + self.failures = failures + super().__init__("gate validation failed") + + +def parse_args(argv: list[str]) -> argparse.Namespace: + repo_root = Path(__file__).resolve().parents[1] + parser = argparse.ArgumentParser( + description="Advance a project when templates/gates/.yaml passes." + ) + parser.add_argument( + "--project-root", + type=Path, + default=Path.cwd(), + help="Project root containing STATE.json and STATUS.md", + ) + parser.add_argument( + "--framework-root", + type=Path, + default=repo_root, + help="SDD Framework root containing templates/gates", + ) + parser.add_argument( + "--now", + default=None, + help="Override current UTC time for deterministic tests", + ) + parser.add_argument( + "--format", + choices=("tty", "json"), + default="json", + help="Output format", + ) + return parser.parse_args(argv) + + +def parse_utc(value: str) -> datetime: + if value.endswith("Z"): + value = value[:-1] + "+00:00" + parsed = datetime.fromisoformat(value) + if parsed.tzinfo is None: + raise ValueError("timestamp must include timezone") + return parsed.astimezone(timezone.utc) + + +def iso_z(value: datetime) -> str: + return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def load_json(path: Path, kind: str) -> Any: + try: + return json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise PipelineError(f"{kind} not found: {path}") from exc + except json.JSONDecodeError as exc: + raise PipelineError(f"invalid {kind}: {exc.msg}") from exc + except OSError as exc: + raise PipelineError(f"could not read {kind}: {exc}") from exc + + +def write_json(path: Path, data: dict[str, Any]) -> None: + try: + path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + except OSError as exc: + raise PipelineError(f"could not write {path}: {exc}") from exc + + +def current_phase(state: dict[str, Any]) -> str: + phase = state.get("phase") + if not isinstance(phase, str) or not phase: + raise PipelineError("STATE.json phase must be a non-empty string") + return phase + + +def gate_path(framework_root: Path, phase: str) -> Path: + return framework_root / "templates" / "gates" / f"{phase.lower()}.yaml" + + +def evidence_path(project_root: Path, phase: str) -> Path: + return project_root / ".sdd" / "gates" / f"{phase.lower()}.json" + + +def normalize_evidence(payload: Any) -> dict[str, Any]: + if not isinstance(payload, dict): + raise PipelineError("gate evidence must be a JSON object") + checks = payload.get("checks", payload) + if not isinstance(checks, dict): + raise PipelineError("gate evidence checks must be a JSON object") + return checks + + +def validate_check(name: str, spec: dict[str, Any], value: Any) -> list[str]: + failures = [] + check_type = spec.get("type") + + if check_type == "boolean": + if not isinstance(value, bool): + return [f"{name} must be a boolean"] + if spec.get("required") is True and value is not True: + failures.append(f"{name} must be true") + return failures + + if check_type == "enum": + values = spec.get("values") + if not isinstance(values, list) or value not in values: + failures.append(f"{name} must be one of {values!r}") + return failures + + if check_type == "integer": + if isinstance(value, bool) or not isinstance(value, int): + return [f"{name} must be an integer"] + minimum = spec.get("minimum") + maximum = spec.get("maximum") + if isinstance(minimum, int) and value < minimum: + failures.append(f"{name} must be >= {minimum}") + if isinstance(maximum, int) and value > maximum: + failures.append(f"{name} must be <= {maximum}") + return failures + + return [f"{name} has unsupported gate type {check_type!r}"] + + +def validate_gate(gate: Any, checks: dict[str, Any]) -> tuple[str, list[str]]: + if not isinstance(gate, dict): + raise PipelineError("gate file must be a JSON object") + gate_checks = gate.get("checks") + if not isinstance(gate_checks, dict) or not gate_checks: + raise PipelineError("gate file must contain a non-empty checks object") + + failures = [] + for name, raw_spec in gate_checks.items(): + if not isinstance(raw_spec, dict): + failures.append(f"{name} gate definition must be an object") + continue + if name not in checks: + if raw_spec.get("required") is True: + failures.append(f"{name} is required") + continue + failures.extend(validate_check(name, raw_spec, checks[name])) + + next_phase = checks.get("next_phase") + if isinstance(next_phase, str) and next_phase == "BLOCKED": + failures.append("next_phase is BLOCKED") + + if failures: + raise GateValidationError(failures) + + if isinstance(next_phase, str): + return next_phase, [] + return default_next_phase(str(gate.get("phase", ""))), [] + + +def default_next_phase(phase: str) -> str: + try: + index = PHASE_SEQUENCE.index(phase) + except ValueError as exc: + raise PipelineError(f"unknown phase: {phase}") from exc + if index == len(PHASE_SEQUENCE) - 1: + return "ARCHIVE" + return PHASE_SEQUENCE[index + 1] + + +def update_status( + status_path: Path, + *, + old_phase: str, + new_phase: str, + next_step: str, + now: datetime, + agent: str, +) -> None: + try: + lines = status_path.read_text(encoding="utf-8").splitlines() + except FileNotFoundError as exc: + raise PipelineError(f"STATUS.md not found: {status_path}") from exc + except OSError as exc: + raise PipelineError(f"could not read STATUS.md: {exc}") from exc + + updated = [] + for line in lines: + if line.startswith("- Phase:"): + updated.append(f"- Phase: {new_phase}") + elif line.startswith("- Next step:"): + updated.append(f"- Next step: {next_step}") + elif line.startswith("- Blockers:"): + updated.append("- Blockers: none") + elif line.startswith("- Updated at:"): + updated.append(f"- Updated at: {iso_z(now)}") + else: + updated.append(line) + + history_row = ( + f"| {now.date().isoformat()} | {new_phase} | " + f"Advanced from {old_phase} to {new_phase}. | {agent} |" + ) + insert_at = find_history_insert_at(updated) + updated.insert(insert_at, history_row) + + try: + status_path.write_text("\n".join(updated).rstrip() + "\n", encoding="utf-8") + except OSError as exc: + raise PipelineError(f"could not write STATUS.md: {exc}") from exc + + +def find_history_insert_at(lines: list[str]) -> int: + try: + history_index = lines.index("## History") + except ValueError: + return len(lines) + + index = history_index + 1 + while index < len(lines): + if index > history_index + 1 and lines[index].startswith("## "): + return index + index += 1 + return len(lines) + + +def apply_pass( + project_root: Path, + state: dict[str, Any], + *, + old_phase: str, + new_phase: str, + checks: dict[str, Any], + now: datetime, +) -> dict[str, Any]: + next_step = str(checks.get("next_step") or f"Begin {new_phase} phase.") + agent = "advance-pipeline" + state.update( + { + "phase": new_phase, + "next_step": next_step, + "last_update": iso_z(now), + "last_agent": agent, + "retry_count": 0, + "last_error": None, + } + ) + write_json(project_root / "STATE.json", state) + update_status( + project_root / "STATUS.md", + old_phase=old_phase, + new_phase=new_phase, + next_step=next_step, + now=now, + agent=agent, + ) + return state + + +def apply_failure( + project_root: Path, + state: dict[str, Any], + *, + failures: list[str], + now: datetime, +) -> dict[str, Any]: + retry_count = int(state.get("retry_count", 0)) + 1 + next_step = f"Fix gate evidence: {failures[0]}" + state.update( + { + "next_step": next_step, + "last_update": iso_z(now), + "last_agent": "advance-pipeline", + "retry_count": retry_count, + "last_error": "; ".join(failures), + } + ) + if retry_count >= 3: + state["phase"] = "BLOCKED" + write_json(project_root / "STATE.json", state) + return state + + +def result_payload( + *, + project_root: Path, + phase: str | None, + gate_result: str, + failing_checks: list[str], + next_step: str | None, + state_changes: dict[str, Any], +) -> dict[str, Any]: + return { + "project_root": str(project_root), + "phase": phase, + "gate_result": gate_result, + "failing_checks": failing_checks, + "next_step": next_step, + "state_changes": state_changes, + } + + +def emit(payload: dict[str, Any], output_format: str) -> None: + if output_format == "json": + print(json.dumps(payload, indent=2)) + return + print(f"{payload['gate_result']}: {payload.get('next_step') or ''}") + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv or sys.argv[1:]) + project_root = args.project_root.resolve() + framework_root = args.framework_root.resolve() + + try: + now = parse_utc(args.now) if args.now else datetime.now(timezone.utc) + state = load_json(project_root / "STATE.json", "STATE.json") + if not isinstance(state, dict): + raise PipelineError("STATE.json must be a JSON object") + phase = current_phase(state) + if phase == "BLOCKED": + payload = result_payload( + project_root=project_root, + phase=phase, + gate_result="blocked", + failing_checks=["project is already BLOCKED"], + next_step=state.get("next_step"), + state_changes={}, + ) + emit(payload, args.format) + return EXIT_USER_ERROR + + gate = load_json(gate_path(framework_root, phase), f"{phase} gate") + evidence = normalize_evidence(load_json(evidence_path(project_root, phase), f"{phase} gate evidence")) + new_phase, _ = validate_gate(gate, evidence) + updated_state = apply_pass( + project_root, + state, + old_phase=phase, + new_phase=new_phase, + checks=evidence, + now=now, + ) + payload = result_payload( + project_root=project_root, + phase=new_phase, + gate_result="pass", + failing_checks=[], + next_step=updated_state.get("next_step"), + state_changes={"phase": new_phase, "retry_count": 0}, + ) + emit(payload, args.format) + return EXIT_SUCCESS + except GateValidationError as exc: + updated_state = apply_failure(project_root, state, failures=exc.failures, now=now) + gate_result = "blocked" if updated_state.get("phase") == "BLOCKED" else "fail" + payload = result_payload( + project_root=project_root, + phase=updated_state.get("phase"), + gate_result=gate_result, + failing_checks=exc.failures, + next_step=updated_state.get("next_step"), + state_changes={ + "phase": updated_state.get("phase"), + "retry_count": updated_state.get("retry_count"), + }, + ) + emit(payload, args.format) + return EXIT_USER_ERROR + except (PipelineError, ValueError) as exc: + payload = result_payload( + project_root=project_root, + phase=None, + gate_result="blocked", + failing_checks=[str(exc)], + next_step=None, + state_changes={}, + ) + emit(payload, args.format) + return EXIT_USER_ERROR + + +if __name__ == "__main__": + raise SystemExit(main()) From e7974b0b3a5b8726c116da4c11f93931591d7e76 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Filho?= Date: Tue, 12 May 2026 09:02:08 -0300 Subject: [PATCH 2/7] test: add greenfield pipeline fixture --- examples/greenfield-cli-tool/STATE.json | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) create mode 100644 examples/greenfield-cli-tool/STATE.json diff --git a/examples/greenfield-cli-tool/STATE.json b/examples/greenfield-cli-tool/STATE.json new file mode 100644 index 0000000..55c0a3d --- /dev/null +++ b/examples/greenfield-cli-tool/STATE.json @@ -0,0 +1,21 @@ +{ + "phase": "RESEARCH", + "next_step": "Review research evidence and advance to SPEC.", + "spec": "openspec/project.md", + "plan": ".sdd/plans/README.md", + "branch": "main", + "worktree": ".", + "last_commit": "0000000", + "last_agent": "human", + "last_update": "2026-05-12T10:00:00Z", + "lock": { + "agent": "none", + "started_at": "1970-01-01T00:00:00Z", + "ttl_minutes": 1 + }, + "retry_count": 0, + "last_error": null, + "flavor": "software", + "spec_engine": "manual", + "handoff_backend": "local" +} From 0f3c3855aa9e3cba2856ecc8c4cb443d29da3452 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Filho?= Date: Tue, 12 May 2026 09:02:22 -0300 Subject: [PATCH 3/7] test: add greenfield status fixture --- examples/greenfield-cli-tool/STATUS.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) create mode 100644 examples/greenfield-cli-tool/STATUS.md diff --git a/examples/greenfield-cli-tool/STATUS.md b/examples/greenfield-cli-tool/STATUS.md new file mode 100644 index 0000000..007ba59 --- /dev/null +++ b/examples/greenfield-cli-tool/STATUS.md @@ -0,0 +1,19 @@ +# Greenfield CLI Tool Status + +## Current state + +- Phase: RESEARCH +- Next step: Review research evidence and advance to SPEC. +- Blockers: none +- Updated at: 2026-05-12T10:00:00Z +- Owner: human + +## History + +| Date | Phase | Change | Agent/operator | +|------|-------|--------|----------------| +| 2026-05-12 | RESEARCH | Initial research fixture created. | human | + +## Notes + +Smoke fixture for `scripts/advance-pipeline.py`. From 65e14a8923cae311078150e10773bcdc74b97231 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Filho?= Date: Tue, 12 May 2026 09:02:31 -0300 Subject: [PATCH 4/7] test: add greenfield research gate evidence --- examples/greenfield-cli-tool/.sdd/gates/research.json | 10 ++++++++++ 1 file changed, 10 insertions(+) create mode 100644 examples/greenfield-cli-tool/.sdd/gates/research.json diff --git a/examples/greenfield-cli-tool/.sdd/gates/research.json b/examples/greenfield-cli-tool/.sdd/gates/research.json new file mode 100644 index 0000000..b383844 --- /dev/null +++ b/examples/greenfield-cli-tool/.sdd/gates/research.json @@ -0,0 +1,10 @@ +{ + "checks": { + "research_doc_exists": true, + "approaches_compared": 2, + "citations_present": true, + "tradeoffs_recorded": true, + "next_phase": "SPEC", + "next_step": "Draft the CLI tool spec from the accepted research direction." + } +} From 82dc43fd2571269ac68de2302bbe4c6e40d8eb0f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Filho?= Date: Tue, 12 May 2026 09:02:48 -0300 Subject: [PATCH 5/7] test: add greenfield research artifact --- .../docs/research/2026-05-12-cli-tool.md | 29 +++++++++++++++++++ 1 file changed, 29 insertions(+) create mode 100644 examples/greenfield-cli-tool/docs/research/2026-05-12-cli-tool.md diff --git a/examples/greenfield-cli-tool/docs/research/2026-05-12-cli-tool.md b/examples/greenfield-cli-tool/docs/research/2026-05-12-cli-tool.md new file mode 100644 index 0000000..0c47d41 --- /dev/null +++ b/examples/greenfield-cli-tool/docs/research/2026-05-12-cli-tool.md @@ -0,0 +1,29 @@ +# Greenfield CLI Tool Research + +## Problem + +Build a tiny CLI project fixture that can exercise the SDD Framework phase transition from RESEARCH to SPEC without external services. + +## Approaches Compared + +### Python argparse CLI + +Uses only the Python standard library, which keeps the fixture portable and aligned with the framework's no-runtime-dependency rule for scripts. + +### Click-based CLI + +Provides a richer command authoring model, but adds a dependency that is unnecessary for this fixture. + +## Decision + +Use a standard-library Python CLI fixture for the smoke path. + +## Tradeoffs + +- Standard-library code keeps the example easy to run in CI. +- The example is intentionally minimal and does not represent a full application scaffold. + +## Citations + +- Python argparse documentation: https://docs.python.org/3/library/argparse.html +- SDD Framework PLAN.md: ../../../../PLAN.md From 30b7b5923043f50d536dadac8338a94ebb0d0fbc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Filho?= Date: Tue, 12 May 2026 09:03:39 -0300 Subject: [PATCH 6/7] test: cover gate-driven pipeline advancement --- tests/scripts/test_advance_pipeline.py | 119 +++++++++++++++++++++++++ 1 file changed, 119 insertions(+) create mode 100644 tests/scripts/test_advance_pipeline.py diff --git a/tests/scripts/test_advance_pipeline.py b/tests/scripts/test_advance_pipeline.py new file mode 100644 index 0000000..2a4e0b5 --- /dev/null +++ b/tests/scripts/test_advance_pipeline.py @@ -0,0 +1,119 @@ +import json +import shutil +import subprocess +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT = REPO_ROOT / "scripts" / "advance-pipeline.py" +EXAMPLE = REPO_ROOT / "examples" / "greenfield-cli-tool" +NOW = "2026-05-12T12:00:00Z" + + +def run_advance(project_root: Path) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [ + sys.executable, + str(SCRIPT), + "--project-root", + str(project_root), + "--framework-root", + str(REPO_ROOT), + "--now", + NOW, + "--format", + "json", + ], + cwd=REPO_ROOT, + text=True, + capture_output=True, + check=False, + ) + + +def copy_example(tmp_path: Path) -> Path: + project_root = tmp_path / "greenfield-cli-tool" + shutil.copytree(EXAMPLE, project_root) + return project_root + + +def read_state(project_root: Path) -> dict: + return json.loads((project_root / "STATE.json").read_text(encoding="utf-8")) + + +def test_advance_pipeline_passes_research_gate_and_updates_state(tmp_path: Path) -> None: + project_root = copy_example(tmp_path) + + result = run_advance(project_root) + + assert result.returncode == 0, result.stdout + result.stderr + payload = json.loads(result.stdout) + assert payload["gate_result"] == "pass" + assert payload["phase"] == "SPEC" + assert payload["state_changes"] == {"phase": "SPEC", "retry_count": 0} + + state = read_state(project_root) + assert state["phase"] == "SPEC" + assert state["retry_count"] == 0 + assert state["last_error"] is None + assert state["last_update"] == NOW + assert state["last_agent"] == "advance-pipeline" + assert state["next_step"] == "Draft the CLI tool spec from the accepted research direction." + + +def test_advance_pipeline_updates_status_history(tmp_path: Path) -> None: + project_root = copy_example(tmp_path) + + result = run_advance(project_root) + + assert result.returncode == 0, result.stdout + result.stderr + status = (project_root / "STATUS.md").read_text(encoding="utf-8") + assert "- Phase: SPEC" in status + assert "- Updated at: 2026-05-12T12:00:00Z" in status + assert "| 2026-05-12 | SPEC | Advanced from RESEARCH to SPEC. | advance-pipeline |" in status + + +def test_advance_pipeline_fails_and_increments_retry_count(tmp_path: Path) -> None: + project_root = copy_example(tmp_path) + evidence_path = project_root / ".sdd" / "gates" / "research.json" + evidence = json.loads(evidence_path.read_text(encoding="utf-8")) + evidence["checks"]["citations_present"] = False + evidence_path.write_text(json.dumps(evidence, indent=2) + "\n", encoding="utf-8") + + result = run_advance(project_root) + + assert result.returncode == 1 + payload = json.loads(result.stdout) + assert payload["gate_result"] == "fail" + assert payload["failing_checks"] == ["citations_present must be true"] + + state = read_state(project_root) + assert state["phase"] == "RESEARCH" + assert state["retry_count"] == 1 + assert state["last_error"] == "citations_present must be true" + assert state["next_step"] == "Fix gate evidence: citations_present must be true" + + +def test_advance_pipeline_blocks_after_third_failed_retry(tmp_path: Path) -> None: + project_root = copy_example(tmp_path) + state = read_state(project_root) + state["retry_count"] = 2 + (project_root / "STATE.json").write_text( + json.dumps(state, indent=2) + "\n", + encoding="utf-8", + ) + evidence_path = project_root / ".sdd" / "gates" / "research.json" + evidence = json.loads(evidence_path.read_text(encoding="utf-8")) + evidence["checks"]["research_doc_exists"] = False + evidence_path.write_text(json.dumps(evidence, indent=2) + "\n", encoding="utf-8") + + result = run_advance(project_root) + + assert result.returncode == 1 + payload = json.loads(result.stdout) + assert payload["gate_result"] == "blocked" + + state = read_state(project_root) + assert state["phase"] == "BLOCKED" + assert state["retry_count"] == 3 + assert state["last_error"] == "research_doc_exists must be true" From 1d7060f404d6f956e7d22995b9a05d7e73683356 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Filho?= Date: Tue, 12 May 2026 09:04:28 -0300 Subject: [PATCH 7/7] docs: record gate advancement smoke path --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3734d8b..6ec4cec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- `scripts/advance-pipeline.py` gate consumption with STATE/STATUS updates, retry handling, and BLOCKED escalation after repeated gate failures. +- Greenfield CLI smoke fixture and tests covering RESEARCH to SPEC advancement through `examples/greenfield-cli-tool`. - Framework-owned templates for status, state, agent instructions, ADRs, manual-engine specs, handoff schema, and cron prompt skeletons. - Machine-verifiable gate templates for all nine pipeline phases. - `scripts/validate-templates.py` and template tests for the M1 template set.