|
#!/usr/bin/env python3 |
|
"""Generate goal-loop planning artifacts from a structured contract JSON.""" |
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import csv |
|
import json |
|
from datetime import datetime, timezone |
|
from pathlib import Path |
|
from typing import Any |
|
|
|
|
|
TRACKER_COLUMNS = [ |
|
"Feature ID", |
|
"Name", |
|
"User Story", |
|
"Expected Behaviour", |
|
"Test Command", |
|
"Status", |
|
"Severity", |
|
"Attempts", |
|
"Last Checked", |
|
"Notes", |
|
] |
|
|
|
STATUS_VALUES = {"Not Started", "Implementing", "Testing", "Pass", "Fail", "Blocked"} |
|
SEVERITY_VALUES = {"P0", "P1", "P2", "P3"} |
|
|
|
|
|
def as_list(value: Any) -> list[str]: |
|
if value is None or value == "": |
|
return [] |
|
if isinstance(value, list): |
|
return [str(item) for item in value if str(item).strip()] |
|
return [str(value)] |
|
|
|
|
|
def as_dict(value: Any) -> dict[str, Any]: |
|
return value if isinstance(value, dict) else {} |
|
|
|
|
|
def bullet(items: list[str], empty: str = "None specified.") -> str: |
|
if not items: |
|
return f"- {empty}" |
|
return "\n".join(f"- {item}" for item in items) |
|
|
|
|
|
def value_list(mapping: dict[str, Any], keys: list[str]) -> list[str]: |
|
values: list[str] = [] |
|
for key in keys: |
|
for item in as_list(mapping.get(key)): |
|
values.append(item) |
|
return values |
|
|
|
|
|
def normalize_task(raw: dict[str, Any], index: int) -> dict[str, Any]: |
|
task_id = str(raw.get("id") or raw.get("feature_id") or f"F{index:03d}") |
|
status = str(raw.get("status") or "Not Started") |
|
severity = str(raw.get("severity") or "P1") |
|
if status not in STATUS_VALUES: |
|
status = "Not Started" |
|
if severity not in SEVERITY_VALUES: |
|
severity = "P1" |
|
|
|
return { |
|
"id": task_id, |
|
"name": str(raw.get("name") or f"Task {index}"), |
|
"user_story": str(raw.get("user_story") or raw.get("story") or ""), |
|
"expected_behavior": str(raw.get("expected_behavior") or raw.get("expected_behaviour") or ""), |
|
"acceptance_criteria": as_list(raw.get("acceptance_criteria")), |
|
"test_command": str(raw.get("test_command") or ""), |
|
"severity": severity, |
|
"status": status, |
|
"attempts": int(raw.get("attempts") or 0), |
|
"last_checked": str(raw.get("last_checked") or ""), |
|
"notes": str(raw.get("notes") or ""), |
|
} |
|
|
|
|
|
def normalize_contract(raw: dict[str, Any]) -> dict[str, Any]: |
|
evidence = as_dict(raw.get("evidence")) |
|
constraints = as_dict(raw.get("constraints")) |
|
iteration_policy = as_dict(raw.get("iteration_policy")) |
|
budgets = as_dict(raw.get("budgets")) |
|
tasks_raw = raw.get("tasks") if isinstance(raw.get("tasks"), list) else [] |
|
|
|
objective = str(raw.get("objective") or raw.get("outcome") or "Complete the requested coding goal.") |
|
tasks = [normalize_task(task if isinstance(task, dict) else {}, i) for i, task in enumerate(tasks_raw, 1)] |
|
if not tasks: |
|
tasks = [ |
|
normalize_task( |
|
{ |
|
"id": "F001", |
|
"name": "Implement goal", |
|
"user_story": objective, |
|
"expected_behavior": "The requested outcome is implemented and verified.", |
|
"test_command": ", ".join(as_list(evidence.get("commands"))), |
|
"severity": "P0", |
|
}, |
|
1, |
|
) |
|
] |
|
|
|
return { |
|
"objective": objective, |
|
"done_when": as_list(raw.get("done_when") or raw.get("success_criteria")), |
|
"evidence": { |
|
"commands": as_list(evidence.get("commands")), |
|
"benchmarks": as_list(evidence.get("benchmarks")), |
|
"manual_checks": as_list(evidence.get("manual_checks")), |
|
"artifacts": as_list(evidence.get("artifacts")), |
|
}, |
|
"constraints": { |
|
"editable_paths": as_list(constraints.get("editable_paths")), |
|
"read_only_paths": as_list(constraints.get("read_only_paths")), |
|
"do_not_change": as_list(constraints.get("do_not_change")), |
|
"external_limits": as_list(constraints.get("external_limits")), |
|
"required_tools": as_list(constraints.get("required_tools") or constraints.get("tools")), |
|
}, |
|
"iteration_policy": { |
|
"priority_order": str(iteration_policy.get("priority_order") or "highest severity first, then plan order"), |
|
"commit_policy": str(iteration_policy.get("commit_policy") or "commit small coherent diffs when permitted"), |
|
"test_cadence": str(iteration_policy.get("test_cadence") or "run task tests after each change and broader checks before final"), |
|
"subagents": str(iteration_policy.get("subagents") or "do not use subagents unless explicitly allowed"), |
|
"max_attempts_per_task": int(iteration_policy.get("max_attempts_per_task") or 3), |
|
}, |
|
"budgets": { |
|
"time": str(budgets.get("time") or "unspecified"), |
|
"tokens": str(budgets.get("tokens") or "unspecified"), |
|
"cost": str(budgets.get("cost") or "unspecified"), |
|
"branch": str(budgets.get("branch") or "use existing branch unless user requests scratch branch"), |
|
}, |
|
"pause_conditions": as_list(raw.get("pause_conditions")), |
|
"tasks": tasks, |
|
"assumptions": as_list(raw.get("assumptions")), |
|
"open_questions": as_list(raw.get("open_questions")), |
|
} |
|
|
|
|
|
def render_spec(contract: dict[str, Any]) -> str: |
|
constraints = contract["constraints"] |
|
evidence = contract["evidence"] |
|
budgets = contract["budgets"] |
|
policy = contract["iteration_policy"] |
|
created = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") |
|
|
|
lines = [ |
|
"# SPEC", |
|
"", |
|
f"Generated: {created}", |
|
"", |
|
"## Objective", |
|
"", |
|
contract["objective"], |
|
"", |
|
"## Done When", |
|
"", |
|
bullet(contract["done_when"]), |
|
"", |
|
"## Evidence", |
|
"", |
|
"### Commands", |
|
bullet(evidence["commands"]), |
|
"", |
|
"### Benchmarks", |
|
bullet(evidence["benchmarks"]), |
|
"", |
|
"### Manual Checks", |
|
bullet(evidence["manual_checks"]), |
|
"", |
|
"### Required Artifacts", |
|
bullet(evidence["artifacts"]), |
|
"", |
|
"## Constraints", |
|
"", |
|
"### Editable Paths", |
|
bullet(constraints["editable_paths"]), |
|
"", |
|
"### Read-only Paths", |
|
bullet(constraints["read_only_paths"]), |
|
"", |
|
"### Do Not Change", |
|
bullet(constraints["do_not_change"]), |
|
"", |
|
"### External Limits", |
|
bullet(constraints["external_limits"]), |
|
"", |
|
"### Required Tools", |
|
bullet(constraints["required_tools"]), |
|
"", |
|
"## Iteration Policy", |
|
"", |
|
f"- Priority order: {policy['priority_order']}", |
|
f"- Commit policy: {policy['commit_policy']}", |
|
f"- Test cadence: {policy['test_cadence']}", |
|
f"- Subagents: {policy['subagents']}", |
|
f"- Max attempts per task: {policy['max_attempts_per_task']}", |
|
"", |
|
"## Budgets", |
|
"", |
|
f"- Time: {budgets['time']}", |
|
f"- Tokens: {budgets['tokens']}", |
|
f"- Cost: {budgets['cost']}", |
|
f"- Branch: {budgets['branch']}", |
|
"", |
|
"## Pause Conditions", |
|
"", |
|
bullet(contract["pause_conditions"]), |
|
"", |
|
"## Assumptions", |
|
"", |
|
bullet(contract["assumptions"]), |
|
"", |
|
"## Open Questions", |
|
"", |
|
bullet(contract["open_questions"]), |
|
"", |
|
] |
|
return "\n".join(lines) |
|
|
|
|
|
def render_plan(contract: dict[str, Any]) -> str: |
|
lines = [ |
|
"# Plan", |
|
"", |
|
f"Objective: {contract['objective']}", |
|
"", |
|
"## Task List", |
|
"", |
|
] |
|
for task in contract["tasks"]: |
|
criteria = bullet(task["acceptance_criteria"], "No separate acceptance criteria specified.") |
|
lines.extend( |
|
[ |
|
f"### {task['id']}: {task['name']}", |
|
"", |
|
f"- Severity: {task['severity']}", |
|
f"- Status: {task['status']}", |
|
f"- User story: {task['user_story'] or 'Not specified.'}", |
|
f"- Expected behavior: {task['expected_behavior'] or 'Not specified.'}", |
|
f"- Test command: `{task['test_command'] or 'Not specified.'}`", |
|
"", |
|
"Acceptance criteria:", |
|
criteria, |
|
"", |
|
] |
|
) |
|
return "\n".join(lines) |
|
|
|
|
|
def render_goal_prompt(contract: dict[str, Any]) -> str: |
|
evidence = value_list(contract["evidence"], ["commands", "benchmarks", "manual_checks", "artifacts"]) |
|
constraints = value_list( |
|
contract["constraints"], |
|
["editable_paths", "read_only_paths", "do_not_change", "external_limits", "required_tools"], |
|
) |
|
policy = contract["iteration_policy"] |
|
limits = [ |
|
f"{policy['max_attempts_per_task']} attempts per task", |
|
f"time budget: {contract['budgets']['time']}", |
|
f"token budget: {contract['budgets']['tokens']}", |
|
f"cost budget: {contract['budgets']['cost']}", |
|
] |
|
|
|
return ( |
|
f"/goal {contract['objective']} verified by {', '.join(evidence) or 'the agreed evidence'} " |
|
f"while preserving {', '.join(constraints) or 'the agreed constraints'}. " |
|
f"Use the required tools and resources from SPEC.md. Between iterations, prioritize {policy['priority_order']}; " |
|
f"{policy['test_cadence']}; {policy['commit_policy']}; subagents: {policy['subagents']}. " |
|
f"If blocked after {', '.join(limits)}, report the blocker, update goal_tracker.csv, and stop.\n" |
|
) |
|
|
|
|
|
def write_tracker(path: Path, contract: dict[str, Any]) -> None: |
|
with path.open("w", newline="", encoding="utf-8") as handle: |
|
writer = csv.DictWriter(handle, fieldnames=TRACKER_COLUMNS) |
|
writer.writeheader() |
|
for task in contract["tasks"]: |
|
writer.writerow( |
|
{ |
|
"Feature ID": task["id"], |
|
"Name": task["name"], |
|
"User Story": task["user_story"], |
|
"Expected Behaviour": task["expected_behavior"], |
|
"Test Command": task["test_command"], |
|
"Status": task["status"], |
|
"Severity": task["severity"], |
|
"Attempts": task["attempts"], |
|
"Last Checked": task["last_checked"], |
|
"Notes": task["notes"], |
|
} |
|
) |
|
|
|
|
|
def write_text(path: Path, content: str) -> None: |
|
path.write_text(content, encoding="utf-8") |
|
|
|
|
|
def ensure_writable(paths: list[Path], force: bool) -> None: |
|
existing = [str(path) for path in paths if path.exists()] |
|
if existing and not force: |
|
joined = "\n".join(f"- {item}" for item in existing) |
|
raise SystemExit(f"Refusing to overwrite existing files. Use --force to replace:\n{joined}") |
|
|
|
|
|
def main() -> int: |
|
parser = argparse.ArgumentParser(description=__doc__) |
|
parser.add_argument("--contract", required=True, type=Path, help="Path to goal_contract.json") |
|
parser.add_argument("--out", default=Path("."), type=Path, help="Output directory") |
|
parser.add_argument("--force", action="store_true", help="Overwrite generated files if they exist") |
|
args = parser.parse_args() |
|
|
|
raw = json.loads(args.contract.read_text(encoding="utf-8")) |
|
if not isinstance(raw, dict): |
|
raise SystemExit("Contract JSON must be an object.") |
|
|
|
contract = normalize_contract(raw) |
|
out_dir = args.out |
|
out_dir.mkdir(parents=True, exist_ok=True) |
|
|
|
outputs = { |
|
"SPEC.md": render_spec(contract), |
|
"Plan.md": render_plan(contract), |
|
"goal_prompt.md": render_goal_prompt(contract), |
|
"goal_contract.normalized.json": json.dumps(contract, indent=2, sort_keys=True) + "\n", |
|
} |
|
output_paths = [out_dir / name for name in outputs] |
|
output_paths.append(out_dir / "goal_tracker.csv") |
|
ensure_writable(output_paths, args.force) |
|
|
|
for name, content in outputs.items(): |
|
write_text(out_dir / name, content) |
|
write_tracker(out_dir / "goal_tracker.csv", contract) |
|
|
|
print(f"Generated {len(output_paths)} files in {out_dir}") |
|
for path in output_paths: |
|
print(path) |
|
return 0 |
|
|
|
|
|
if __name__ == "__main__": |
|
raise SystemExit(main()) |
