From a91b9d879688cda84c3ba745f9a2b98fae4abb35 Mon Sep 17 00:00:00 2001 From: Sollan Systems Date: Mon, 13 Jul 2026 15:09:01 -0400 Subject: [PATCH] feat(kernel): loop-engineer/plan@1 Loop Plan IR + loop plan-lint CLI (#49) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - schemas/plan.schema.json: plan@1 — goal/acceptance_criteria/tasks/ terminal_state_mapping required; 8 task kinds (agent tool gate approval join subloop human terminal); capability-based model_policy (#56): read/reason/write/verify -> fast_low_cost/deep_reasoning/code_generation/ independent_review; completion_policy reuses the shared all_required normalizer - loop/plan.py: validate_plan() with jsonschema mode + hand-rolled structural fallback (type-checked required core surface in both modes); cross-field rules in both modes: task-id/criterion-id uniqueness, dangling depends_on/join_on refs, invalid_dependency_entry for non-string entries, dependency-graph acyclicity (iterative DFS), per-kind required fields, approval_gates referential integrity - CLI: flat `loop plan-lint [--mode basic|strict|release] ` reusing the doctor --mode contract; strict/release fail loud without jsonschema (exit 2, no traceback) - goldens: examples/plans/coverage-repair.plan.json (all 8 kinds) + invalid/ negatives; ci.yml plan-lint smoke (positive + 2 negatives) - docs: reference/repo-os-contract.md §15 (scope boundary: plan@1 is standalone in v1, not yet read by loop doctor) Suite: extras 586 passed / 15 skipped; pyyaml-only 563 / 38. Lane: claudex gpt-5.6-terra, accepted on attempt 2 after one productive repair (basic-mode type-check parity gap found by fresh sonnet review, reproduced by the governor, closed with both-mode regression tests). Receipts: cx_s3_plan_a1 (repair_requested), cx_s3_plan_a2 (accepted). Closes #49 Claude-Session: https://claude.ai/code/session_01EJ8zA8Cbi4o2amawpj8bZW --- .github/workflows/ci.yml | 12 + examples/plans/coverage-repair.plan.json | 32 +++ .../plans/invalid/cyclic-dependency.plan.json | 11 + examples/plans/invalid/missing-goal.plan.json | 6 + loop/__init__.py | 4 + loop/__main__.py | 41 ++- loop/plan.py | 267 ++++++++++++++++++ reference/repo-os-contract.md | 36 +++ schemas/plan.schema.json | 23 ++ scripts/test_loop_cli.py | 59 ++++ scripts/test_plan_schema.py | 225 +++++++++++++++ scripts/test_wheel_selfcontained.py | 11 + 12 files changed, 714 insertions(+), 13 deletions(-) create mode 100644 examples/plans/coverage-repair.plan.json create mode 100644 examples/plans/invalid/cyclic-dependency.plan.json create mode 100644 examples/plans/invalid/missing-goal.plan.json create mode 100644 loop/plan.py create mode 100644 schemas/plan.schema.json create mode 100644 scripts/test_plan_schema.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 75ed8c3..d0697f4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -51,6 +51,18 @@ jobs: python -B -m loop doctor --mode release examples/coverage-repair python -B -m loop inspect examples/coverage-repair + - name: Plan lint smoke test + run: | + python -B -m loop plan-lint --mode release examples/plans/coverage-repair.plan.json + if python -B -m loop plan-lint --mode release examples/plans/invalid/cyclic-dependency.plan.json; then + echo "::error::cyclic-dependency.plan.json unexpectedly passed plan-lint" + exit 1 + fi + if python -B -m loop plan-lint --mode release examples/plans/invalid/missing-goal.plan.json; then + echo "::error::missing-goal.plan.json unexpectedly passed plan-lint" + exit 1 + fi + recipe-langgraph: name: recipe (langgraph) runs-on: ubuntu-latest diff --git a/examples/plans/coverage-repair.plan.json b/examples/plans/coverage-repair.plan.json new file mode 100644 index 0000000..6273aca --- /dev/null +++ b/examples/plans/coverage-repair.plan.json @@ -0,0 +1,32 @@ +{ + "schema": "loop-engineer/plan@1", + "goal": "Eliminate the known flaky-test cluster in the coverage-repair example without weakening any existing assertion.", + "constraints": ["no new third-party test dependencies", "must not increase full-suite runtime by more than 10%"], + "acceptance_criteria": [ + { "id": "AC1", "description": "All previously flaky tests pass 20/20 consecutive runs" }, + { "id": "AC2", "description": "No regression in the existing suite" } + ], + "tasks": [ + { "id": "collect-logs", "kind": "agent", "title": "Collect recent CI failure logs", "depends_on": [], "role": "read", "verify": "scripts/verify-fast" }, + { "id": "diagnose-flakiness", "kind": "agent", "title": "Diagnose the flaky-test cluster", "depends_on": ["collect-logs"], "role": "reason", "verify": "scripts/verify-fast" }, + { "id": "run-flaky-detector", "kind": "tool", "title": "Run the flaky-test detector", "depends_on": ["diagnose-flakiness"], "tool_name": "pytest-flaky-detector", "verify": "scripts/verify-fast" }, + { "id": "repair-branch-a", "kind": "agent", "title": "Repair candidate A", "depends_on": ["run-flaky-detector"], "role": "write", "verify": "scripts/verify-full" }, + { "id": "repair-branch-b", "kind": "agent", "title": "Repair candidate B", "depends_on": ["run-flaky-detector"], "role": "write", "verify": "scripts/verify-full" }, + { "id": "join-repairs", "kind": "join", "title": "Join repair candidates", "depends_on": ["repair-branch-a", "repair-branch-b"], "join_on": ["repair-branch-a", "repair-branch-b"] }, + { "id": "verify-fast-gate", "kind": "gate", "title": "Fast deterministic gate", "depends_on": ["join-repairs"], "verify": "scripts/verify-fast" }, + { "id": "independent-review", "kind": "agent", "title": "Independent adversarial review", "depends_on": ["verify-fast-gate"], "role": "verify", "verify": "scripts/verify-safety" }, + { "id": "ship-approval", "kind": "approval", "title": "Release sign-off", "depends_on": ["independent-review"], "approval_gate": "release-sign-off" }, + { "id": "manual-changelog-entry", "kind": "human", "title": "Add CHANGELOG entry", "depends_on": ["ship-approval"], "instructions": "Add a one-line CHANGELOG entry summarizing the flaky-test fix once ship-approval is granted." }, + { "id": "regression-sweep", "kind": "subloop", "title": "Full regression sweep", "depends_on": ["manual-changelog-entry"], "subloop_ref": "loop-engineer/plan@1:regression-sweep-v1" }, + { "id": "succeeded", "kind": "terminal", "title": "Succeeded", "depends_on": ["regression-sweep"], "terminal_state": "Succeeded" } + ], + "completion_policy": { "mode": "all_required" }, + "terminal_state_mapping": { + "success": "Succeeded", "budget_exhausted": "FailedBudget", "verifier_veto": "FailedSafety", "unresolvable_flake": "FailedUnverifiable", "external_ci_outage": "FailedBlocked", "spec_ambiguous": "FailedSpecGap", "operator_abort": "AbortedByHuman" + }, + "approval_gates": ["release-sign-off"], + "verifiers": ["scripts/verify-fast", "scripts/verify-full", "scripts/verify-safety"], + "budgets": { "max_iterations": 12, "wall_clock_minutes": 90 }, + "defaults": { "retry": { "max_attempts": 2, "backoff": "exponential", "timeout_seconds": 600 } }, + "model_policy": { "read": "fast_low_cost", "reason": "deep_reasoning", "write": "code_generation", "verify": "independent_review" } +} diff --git a/examples/plans/invalid/cyclic-dependency.plan.json b/examples/plans/invalid/cyclic-dependency.plan.json new file mode 100644 index 0000000..ff9065a --- /dev/null +++ b/examples/plans/invalid/cyclic-dependency.plan.json @@ -0,0 +1,11 @@ +{ + "schema": "loop-engineer/plan@1", + "goal": "Deliberately invalid fixture: exercises loop/plan.py's cycle detector.", + "acceptance_criteria": [{ "id": "AC1", "description": "unreachable — this plan is intentionally cyclic" }], + "tasks": [ + { "id": "task-a", "kind": "tool", "title": "A", "depends_on": ["task-c"], "tool_name": "noop", "verify": "true" }, + { "id": "task-b", "kind": "tool", "title": "B", "depends_on": ["task-a"], "tool_name": "noop", "verify": "true" }, + { "id": "task-c", "kind": "tool", "title": "C", "depends_on": ["task-b"], "tool_name": "noop", "verify": "true" } + ], + "terminal_state_mapping": { "success": "Succeeded", "failure": "FailedUnverifiable" } +} diff --git a/examples/plans/invalid/missing-goal.plan.json b/examples/plans/invalid/missing-goal.plan.json new file mode 100644 index 0000000..8c67b99 --- /dev/null +++ b/examples/plans/invalid/missing-goal.plan.json @@ -0,0 +1,6 @@ +{ + "schema": "loop-engineer/plan@1", + "acceptance_criteria": [{ "id": "AC1", "description": "n/a" }], + "tasks": [{ "id": "only-task", "kind": "human", "title": "Manual step", "depends_on": [], "instructions": "n/a" }], + "terminal_state_mapping": { "success": "Succeeded" } +} diff --git a/loop/__init__.py b/loop/__init__.py index 5247905..26f95cf 100644 --- a/loop/__init__.py +++ b/loop/__init__.py @@ -7,12 +7,16 @@ from .paths import LoopPaths, resolve_loop_paths from .contract import TERMINAL_STATES, VALIDATION_MODES, doctor_report, validate_contract +from .plan import PLAN_SCHEMA_ID, TASK_KINDS, validate_plan __all__ = [ "LoopPaths", + "PLAN_SCHEMA_ID", + "TASK_KINDS", "TERMINAL_STATES", "VALIDATION_MODES", "doctor_report", "resolve_loop_paths", "validate_contract", + "validate_plan", ] diff --git a/loop/__main__.py b/loop/__main__.py index 67bf991..a8f7fad 100644 --- a/loop/__main__.py +++ b/loop/__main__.py @@ -6,22 +6,24 @@ from pathlib import Path from .contract import VALIDATION_MODES, ValidationModeError, doctor_report +from .plan import validate_plan _PROG = "python3 -m loop" -_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect", "metrics") +_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect", "metrics", "plan-lint") # Read commands operate on an EXISTING contract dir; scaffold CREATES one, so it # is exempt from the "target must exist" guard. -_READ_COMMANDS = ("doctor", "validate", "verify", "inspect", "metrics") +_READ_COMMANDS = ("doctor", "validate", "verify", "inspect", "metrics", "plan-lint") -_USAGE = f"usage: {_PROG} " +_USAGE = f"usage: {_PROG} " _HELP = f"""{_PROG} — validate, inspect, and measure a portable repo-OS loop contract. {_USAGE} {_PROG} metrics [--baseline] {_PROG} doctor|validate|verify [--mode basic|strict|release] + {_PROG} plan-lint [--mode basic|strict|release] commands: scaffold Write a fresh, doctor-clean loop contract into . @@ -34,14 +36,18 @@ real .loop/ evidence (RUNLOG, verify bundles, held-out gate, repair records) and emit a JSON scorecard. With --baseline, write a checked-in baseline scorecard — refused unless the run is gate-backed. + plan-lint Validate a loop-engineer/plan@1 Loop Plan IR document: task-kind + fields, dependency-graph acyclicity, and the terminal-state + mapping. --mode selects validation strength, same as doctor. arguments: - A workspace root or its .loop/ directory. + A workspace root or its .loop/ directory (all commands except plan-lint). + A single loop-engineer/plan@1 JSON file (plan-lint only). options: --mode {{basic,strict,release}} - (doctor/validate/verify only) basic forces structural checks; - strict/release require jsonschema. Default: auto-detect. + (doctor/validate/verify/plan-lint only) basic forces structural + checks; strict/release require jsonschema. Default: auto-detect. --baseline (metrics only) write docs/metrics-baseline.json over a gate-backed run; exits non-zero and writes nothing otherwise. -h, --help Show this help and exit. @@ -155,7 +161,7 @@ def main(argv: list[str] | None = None) -> int: return 2 mode = None - if command in {"doctor", "validate", "verify"}: + if command in {"doctor", "validate", "verify", "plan-lint"}: try: mode, argv = _extract_mode_flag(argv) except ValueError as exc: @@ -175,12 +181,14 @@ def main(argv: list[str] | None = None) -> int: target = Path(argv[0]) if command in _READ_COMMANDS and not target.exists(): - print( - f"{command}: target path does not exist: {target}\n" - f" pass an existing workspace root or its .loop/ directory " - f"(run `{_PROG} scaffold {target}` to create a new contract).", - file=sys.stderr, - ) + if command == "plan-lint": + hint = "pass an existing loop-engineer/plan@1 JSON file" + else: + hint = ( + f"pass an existing workspace root or its .loop/ directory " + f"(run `{_PROG} scaffold {target}` to create a new contract)" + ) + print(f"{command}: target path does not exist: {target}\n {hint}.", file=sys.stderr) return 2 if command == "scaffold": @@ -201,6 +209,13 @@ def main(argv: list[str] | None = None) -> int: print(f"{command}: {exc}", file=sys.stderr) return 2 + if command == "plan-lint": + try: + return _print_json(validate_plan(target, mode=mode)) + except ValidationModeError as exc: + print(f"{command}: {exc}", file=sys.stderr) + return 2 + # command == "inspect": keep the historical inspector script as the scoring # UI over the same contract artifacts; import lazily to avoid making # scripts/ a package. diff --git a/loop/plan.py b/loop/plan.py new file mode 100644 index 0000000..b6f9630 --- /dev/null +++ b/loop/plan.py @@ -0,0 +1,267 @@ +"""loop-engineer/plan@1 — the Loop Plan IR (ADR 0001). + +Kernel-side schema + lint only: this module never dispatches an agent, tool, +or model provider, and it is not yet wired into `loop doctor` / a scaffolded +workspace's `.loop/` tree (see reference/repo-os-contract.md #15 for the +documented scope boundary). It exists so a plan document can be authored and +validated on its own, ahead of the execution-runtime milestone that will give +it an on-disk home. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from .completion import CompletionPolicyError, normalize_completion_policy +from .contract import TERMINAL_STATES, ContractIssue, _resolve_requested_mode, _schemas_dir + +PLAN_SCHEMA_ID = "loop-engineer/plan@1" +TASK_KINDS = ("agent", "tool", "gate", "approval", "join", "subloop", "human", "terminal") +MODEL_POLICY_ROLES = ("read", "reason", "write", "verify") +MODEL_CAPABILITIES = ("fast_low_cost", "deep_reasoning", "code_generation", "independent_review") + +_KIND_REQUIRED_FIELDS: dict[str, tuple[str, ...]] = { + "agent": ("role", "verify"), "tool": ("tool_name", "verify"), "gate": ("verify",), + "approval": ("approval_gate",), "join": ("join_on",), "subloop": ("subloop_ref",), + "human": ("instructions",), "terminal": ("terminal_state",), +} + + +def _read_plan_json(path: Path, issues: list[dict]) -> dict[str, Any] | None: + if not path.exists(): + issues.append(ContractIssue("missing_file", f"missing plan file: {path}", path)) + return None + if path.is_dir(): + issues.append(ContractIssue("invalid_target", f"plan-lint target is a directory, not a file: {path}", path)) + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + issues.append(ContractIssue("invalid_json", f"{path.name}: {exc}", path)) + return None + except UnicodeDecodeError as exc: + issues.append(ContractIssue("invalid_encoding", f"{path.name}: not valid UTF-8: {exc}", path)) + return None + if not isinstance(data, dict): + issues.append(ContractIssue("invalid_json", f"{path.name}: expected object", path)) + return None + return data + + +def _load_plan_schema() -> dict[str, Any]: + return json.loads((_schemas_dir() / "plan.schema.json").read_text(encoding="utf-8")) + + +def _jsonschema_validate_plan(data: dict[str, Any], path: Path, issues: list[dict]) -> None: + import jsonschema # type: ignore + validator = jsonschema.Draft202012Validator(_load_plan_schema()) + for err in validator.iter_errors(data): + location = "/".join(str(p) for p in err.absolute_path) or "" + issues.append(ContractIssue("schema_violation", f"{path.name}: {location}: {err.message}", path)) + + +def _structural_validate_plan(data: dict[str, Any], path: Path, issues: list[dict]) -> None: + if data.get("schema") != PLAN_SCHEMA_ID: + issues.append(ContractIssue("schema_mismatch", f"{path.name}: expected schema {PLAN_SCHEMA_ID!r}, got {data.get('schema')!r}", path)) + if not isinstance(data.get("goal"), str) or not data["goal"].strip(): + issues.append(ContractIssue("invalid_plan", "goal must be a non-empty string", path)) + criteria = data.get("acceptance_criteria") + if not isinstance(criteria, list) or not criteria: + issues.append(ContractIssue("invalid_plan", "acceptance_criteria must be a non-empty array", path)) + else: + for index, item in enumerate(criteria, start=1): + if not isinstance(item, dict): + issues.append(ContractIssue("invalid_plan", f"acceptance criterion #{index} must be an object", path)) + continue + criterion_id = item.get("id") + criterion_label = repr(criterion_id) if criterion_id is not None else f"#{index}" + if not isinstance(criterion_id, str) or not criterion_id: + issues.append(ContractIssue("invalid_plan", f"acceptance criterion {criterion_label} id must be a non-empty string", path)) + description = item.get("description") + if not isinstance(description, str) or not description: + issues.append(ContractIssue("invalid_plan", f"acceptance criterion {criterion_label} description must be a non-empty string", path)) + tasks = data.get("tasks") + if not isinstance(tasks, list) or not tasks: + issues.append(ContractIssue("invalid_plan", "tasks must be a non-empty array", path)) + else: + for index, task in enumerate(tasks, start=1): + if not isinstance(task, dict): + issues.append(ContractIssue("invalid_task", "task must be an object", path)) + continue + task_id = task.get("id") + task_label = repr(task_id) if task_id is not None else f"#{index}" + for key in ("id", "kind", "title", "depends_on"): + if key not in task: + issues.append(ContractIssue("invalid_task", f"task {task_label} missing {key!r}", path)) + if not isinstance(task_id, str) or not task_id: + issues.append(ContractIssue("invalid_task", f"task {task_label} id must be a non-empty string", path)) + title = task.get("title") + if not isinstance(title, str) or not title: + issues.append(ContractIssue("invalid_task", f"task {task_label} title must be a non-empty string", path)) + kind = task.get("kind") + if not isinstance(kind, str): + issues.append(ContractIssue("invalid_task", f"task {task_label} kind must be a string", path)) + elif kind not in TASK_KINDS: + issues.append(ContractIssue("invalid_task_kind", f"task {task_label} has unknown kind {kind!r}", path)) + if not isinstance(task.get("depends_on"), list): + issues.append(ContractIssue("invalid_task", f"task {task_label} depends_on must be an array", path)) + mapping = data.get("terminal_state_mapping") + if not isinstance(mapping, dict) or not mapping: + issues.append(ContractIssue("invalid_plan", "terminal_state_mapping must be a non-empty object", path)) + else: + for key, value in mapping.items(): + if value not in TERMINAL_STATES: + issues.append(ContractIssue("invalid_terminal_state", f"terminal_state_mapping[{key!r}] = {value!r} is not a canonical terminal state", path)) + model_policy = data.get("model_policy") + if model_policy is not None: + if not isinstance(model_policy, dict): + issues.append(ContractIssue("invalid_plan", "model_policy must be an object", path)) + else: + for role, capability in model_policy.items(): + if role not in MODEL_POLICY_ROLES: + issues.append(ContractIssue("invalid_model_policy", f"unknown model_policy role {role!r}", path)) + if capability not in MODEL_CAPABILITIES: + issues.append(ContractIssue("invalid_model_policy", f"unknown model_policy capability {capability!r} for role {role!r}", path)) + try: + normalize_completion_policy(data.get("completion_policy")) + except CompletionPolicyError as exc: + issues.append(ContractIssue("invalid_plan", f"completion_policy is invalid: {exc}", path)) + + +def _check_task_kind_fields(tasks: list[Any], path: Path, issues: list[dict]) -> None: + for task in tasks: + if not isinstance(task, dict): + continue + kind = task.get("kind") + if kind not in _KIND_REQUIRED_FIELDS: + continue + task_id = task.get("id", "") + for field in _KIND_REQUIRED_FIELDS[kind]: + if not task.get(field): + issues.append(ContractIssue("missing_kind_field", f"task {task_id!r} (kind={kind!r}) missing required field {field!r}", path)) + if kind == "agent" and task.get("role") is not None and task["role"] not in MODEL_POLICY_ROLES: + issues.append(ContractIssue("invalid_task_role", f"task {task_id!r} has unknown role {task['role']!r}", path)) + if kind == "terminal" and task.get("terminal_state") is not None and task["terminal_state"] not in TERMINAL_STATES: + issues.append(ContractIssue("invalid_terminal_state", f"task {task_id!r} terminal_state {task['terminal_state']!r} is not canonical", path)) + if kind == "join": + join_on = task.get("join_on") + if isinstance(join_on, list) and len(join_on) < 2: + issues.append(ContractIssue("invalid_join", f"task {task_id!r} join_on needs at least 2 upstream task ids", path)) + + +def _check_task_ids_and_dependencies(tasks: list[Any], path: Path, issues: list[dict]) -> None: + ids: list[str] = [] + seen: set[str] = set() + edges: dict[str, list[str]] = {} + for task in tasks: + if not isinstance(task, dict): + continue + task_id = task.get("id") + task_label = repr(task_id) if task_id is not None else "" + depends_on = task.get("depends_on") + if isinstance(depends_on, list): + for dependency in depends_on: + if not isinstance(dependency, str): + issues.append(ContractIssue("invalid_dependency_entry", f"task {task_label} depends_on contains non-string entry {dependency!r}", path)) + join_on = task.get("join_on") + if isinstance(join_on, list): + for dependency in join_on: + if not isinstance(dependency, str): + issues.append(ContractIssue("invalid_dependency_entry", f"task {task_label} join_on contains non-string entry {dependency!r}", path)) + if not isinstance(task_id, str) or not task_id: + continue + if task_id in seen: + issues.append(ContractIssue("duplicate_task_id", f"duplicate task id {task_id!r}", path)) + seen.add(task_id) + ids.append(task_id) + edges[task_id] = [d for d in depends_on if isinstance(d, str)] if isinstance(depends_on, list) else [] + known = set(ids) + for task in tasks: + if not isinstance(task, dict): + continue + task_id = task.get("id") + if not isinstance(task_id, str): + continue + for dep in edges.get(task_id, []): + if dep not in known: + issues.append(ContractIssue("unknown_dependency", f"task {task_id!r} depends_on unknown task {dep!r}", path)) + join_on = task.get("join_on") + if isinstance(join_on, list): + for dep in join_on: + if isinstance(dep, str) and dep not in known: + issues.append(ContractIssue("unknown_dependency", f"task {task_id!r} join_on unknown task {dep!r}", path)) + WHITE, GRAY, BLACK = 0, 1, 2 + color = {task_id: WHITE for task_id in ids} + cyclic: set[str] = set() + for start in ids: + if color[start] != WHITE: + continue + stack = [(start, iter(edges.get(start, [])))] + color[start] = GRAY + while stack: + node, it = stack[-1] + advanced = False + for dep in it: + if dep not in known: + continue + if color[dep] == GRAY: + cyclic.add(node) + cyclic.add(dep) + elif color[dep] == WHITE: + color[dep] = GRAY + stack.append((dep, iter(edges.get(dep, [])))) + advanced = True + break + if not advanced: + color[node] = BLACK + stack.pop() + if cyclic: + issues.append(ContractIssue("cyclic_dependency", "dependency graph has a cycle among tasks: " + ", ".join(sorted(cyclic)), path)) + + +def _check_acceptance_criteria_ids(criteria: list[Any], path: Path, issues: list[dict]) -> None: + seen: set[str] = set() + for item in criteria: + if not isinstance(item, dict): + continue + crit_id = item.get("id") + if isinstance(crit_id, str) and crit_id: + if crit_id in seen: + issues.append(ContractIssue("duplicate_criterion_id", f"duplicate acceptance_criteria id {crit_id!r}", path)) + seen.add(crit_id) + + +def _check_approval_gates(data: dict[str, Any], tasks: list[Any], path: Path, issues: list[dict]) -> None: + declared = data.get("approval_gates") + declared_set = set(declared) if isinstance(declared, list) else set() + approval_tasks = [t for t in tasks if isinstance(t, dict) and t.get("kind") == "approval"] + if approval_tasks and not declared_set: + issues.append(ContractIssue("missing_approval_gates", "plan has approval tasks but declares no top-level approval_gates", path)) + return + for task in approval_tasks: + gate = task.get("approval_gate") + if isinstance(gate, str) and gate not in declared_set: + issues.append(ContractIssue("unknown_approval_gate", f"task {task.get('id')!r} references undeclared approval_gate {gate!r}", path)) + + +def validate_plan(target: str | Path, *, mode: str | None = None) -> dict[str, Any]: + requested_mode, resolved_mode = _resolve_requested_mode(mode) + path = Path(target) + issues: list[dict] = [] + data = _read_plan_json(path, issues) + if data is not None: + if resolved_mode == "jsonschema": + _jsonschema_validate_plan(data, path, issues) + else: + _structural_validate_plan(data, path, issues) + tasks = data.get("tasks") + if isinstance(tasks, list): + _check_task_kind_fields(tasks, path, issues) + _check_task_ids_and_dependencies(tasks, path, issues) + _check_approval_gates(data, tasks, path, issues) + criteria = data.get("acceptance_criteria") + if isinstance(criteria, list): + _check_acceptance_criteria_ids(criteria, path, issues) + return {"ok": not issues, "path": str(path), "validation_mode": resolved_mode, "requested_mode": requested_mode, "schemas_checked": [PLAN_SCHEMA_ID], "issues": issues} diff --git a/reference/repo-os-contract.md b/reference/repo-os-contract.md index 7719079..437887f 100644 --- a/reference/repo-os-contract.md +++ b/reference/repo-os-contract.md @@ -544,6 +544,42 @@ land silently. --- +## 15. `loop-engineer/plan@1` — the Loop Plan IR + +`schemas/plan.schema.json` defines a canonical, validated description of a +goal, its tasks, and its policies — the document a future execution runtime +interprets (ADR 0001). It is authored and linted as a **standalone JSON +file**, validated by `loop plan-lint ` / `loop.plan.validate_plan()`. + +**Scope boundary:** unlike manifest/state/tasks/terminal (§11), plan@1 is +**not yet** an artifact `loop doctor` reads from a scaffolded workspace — +it has no `.loop/`-relative home today. The execution-runtime milestone +that materializes a plan into a live `TASKS.json` will make that call. + +**Task kinds:** `agent | tool | gate | approval | join | subloop | human | +terminal` — each carries a common `id`/`kind`/`title`/`depends_on` base +plus kind-specific required fields (`loop/plan.py::_KIND_REQUIRED_FIELDS`). + +**Capability-based model policy** (issue #56, ADR 0001 consequence 5): an +optional top-level `model_policy` maps roles (`read`/`reason`/`write`/ +`verify`) to capabilities (`fast_low_cost`/`deep_reasoning`/ +`code_generation`/`independent_review`) — never a vendor model name. An +`agent`-kind task declares a `role`; a provider profile resolves the +capability to an actual model **outside** the portable contract, recorded +to a receipt for reproducibility, not to the plan. + +**Cross-field rules JSON Schema cannot express** (enforced by +`loop/plan.py`, in both validation modes): task-id and +acceptance-criteria-id uniqueness, dangling `depends_on`/`join_on` +references, dependency-graph acyclicity, per-kind required fields, and +`approval_gates` referential integrity. + +Golden examples: `examples/plans/coverage-repair.plan.json` (valid, all 8 +kinds); `examples/plans/invalid/` (deliberately broken fixtures used by +the negative tests). + +--- + Sources: "Designing a Loop Engineer Skill for Frontier Agent Workflows" (2026), synthesizing Anthropic guidance on long-running agent harnesses (anthropic.com, 2025), OpenAI Agents/Codex guidance, Google Conductor, and arXiv PreFlect (2602.07187), SWE-Marathon (2606.07682), Web Agents diff --git a/schemas/plan.schema.json b/schemas/plan.schema.json new file mode 100644 index 0000000..4c5044a --- /dev/null +++ b/schemas/plan.schema.json @@ -0,0 +1,23 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "loop-engineer/plan@1", + "title": "Loop Engineer Plan @1", + "description": "The Loop Plan IR (ADR 0001): a canonical, validated description of a goal, its tasks, and its policies that an execution runtime interprets. Kernel-validated only in v1 — not yet an artifact `loop doctor` reads from a scaffolded workspace; see reference/repo-os-contract.md #15. Cross-field rules JSON Schema cannot express (dependency-graph acyclicity, task-id/criterion-id uniqueness, per-kind required fields, terminal-state-mapping vocabulary, approval-gate referential integrity) are enforced by loop/plan.py in both validation modes.", + "type": "object", + "required": ["schema", "goal", "acceptance_criteria", "tasks", "terminal_state_mapping"], + "properties": { + "schema": { "const": "loop-engineer/plan@1" }, + "goal": { "type": "string", "minLength": 1 }, + "constraints": { "type": "array", "items": { "type": "string", "minLength": 1 } }, + "acceptance_criteria": { "type": "array", "minItems": 1, "items": { "type": "object", "required": ["id", "description"], "properties": { "id": { "type": "string", "minLength": 1 }, "description": { "type": "string", "minLength": 1 } }, "additionalProperties": true } }, + "tasks": { "type": "array", "minItems": 1, "items": { "type": "object", "required": ["id", "kind", "title", "depends_on"], "properties": { "id": { "type": "string", "minLength": 1 }, "kind": { "enum": ["agent", "tool", "gate", "approval", "join", "subloop", "human", "terminal"] }, "title": { "type": "string", "minLength": 1 }, "depends_on": { "type": "array", "items": { "type": "string", "minLength": 1 } }, "role": { "enum": ["read", "reason", "write", "verify"] }, "verify": { "type": "string", "minLength": 1 }, "tool_name": { "type": "string", "minLength": 1 }, "approval_gate": { "type": "string", "minLength": 1 }, "join_on": { "type": "array", "minItems": 2, "items": { "type": "string", "minLength": 1 } }, "subloop_ref": { "type": "string", "minLength": 1 }, "instructions": { "type": "string", "minLength": 1 }, "terminal_state": { "enum": ["Succeeded", "FailedUnverifiable", "FailedBlocked", "FailedBudget", "FailedSafety", "FailedSpecGap", "AbortedByHuman"] }, "retry": { "type": "object", "properties": { "max_attempts": { "type": "integer", "minimum": 0 }, "backoff": { "enum": ["none", "fixed", "exponential"] }, "timeout_seconds": { "type": "number", "exclusiveMinimum": 0 } }, "additionalProperties": true } }, "additionalProperties": true } }, + "completion_policy": { "type": ["object", "null"], "required": ["mode"], "properties": { "mode": { "const": "all_required" } }, "additionalProperties": false, "default": { "mode": "all_required" } }, + "terminal_state_mapping": { "type": "object", "minProperties": 1, "propertyNames": { "type": "string", "minLength": 1 }, "additionalProperties": { "enum": ["Succeeded", "FailedUnverifiable", "FailedBlocked", "FailedBudget", "FailedSafety", "FailedSpecGap", "AbortedByHuman"] } }, + "approval_gates": { "type": "array", "items": { "type": "string", "minLength": 1 } }, + "verifiers": { "type": "array", "items": { "type": "string", "minLength": 1 } }, + "budgets": { "type": "object" }, + "defaults": { "type": "object", "properties": { "retry": { "type": "object", "properties": { "max_attempts": { "type": "integer", "minimum": 0 }, "backoff": { "enum": ["none", "fixed", "exponential"] }, "timeout_seconds": { "type": "number", "exclusiveMinimum": 0 } }, "additionalProperties": true } }, "additionalProperties": true }, + "model_policy": { "type": "object", "properties": { "read": { "enum": ["fast_low_cost", "deep_reasoning", "code_generation", "independent_review"] }, "reason": { "enum": ["fast_low_cost", "deep_reasoning", "code_generation", "independent_review"] }, "write": { "enum": ["fast_low_cost", "deep_reasoning", "code_generation", "independent_review"] }, "verify": { "enum": ["fast_low_cost", "deep_reasoning", "code_generation", "independent_review"] } }, "additionalProperties": false } + }, + "additionalProperties": true +} diff --git a/scripts/test_loop_cli.py b/scripts/test_loop_cli.py index f83f3d9..ed53f3a 100644 --- a/scripts/test_loop_cli.py +++ b/scripts/test_loop_cli.py @@ -240,3 +240,62 @@ def test_mode_is_not_consumed_by_other_commands(tmp_path): ) assert result.returncode == 0 assert (tmp_path / "--mode").is_dir() + + +def test_help_lists_plan_lint_command(): + assert "plan-lint" in _run("--help").stdout + + +def test_help_documents_plan_lint_mode_flag(): + assert "plan-lint [--mode basic|strict|release] " in _run("--help").stdout + + +def test_plan_lint_missing_target_argument_prints_usage_and_exits_nonzero(): + result = _run("plan-lint") + assert result.returncode != 0 + assert "usage" in result.stderr.lower() + assert "Traceback" not in result.stderr + + +def test_plan_lint_nonexistent_file_gives_distinct_actionable_error(tmp_path): + missing = tmp_path / "does-not-exist.plan.json" + result = _run("plan-lint", str(missing)) + assert result.returncode != 0 + assert "loop-engineer/plan@1 JSON file" in result.stderr + assert "scaffold" not in result.stderr + + +def test_plan_lint_valid_golden_example_exits_zero(): + import importlib.util + + mode = "release" if importlib.util.find_spec("jsonschema") is not None else "basic" + result = _run("plan-lint", "--mode", mode, "examples/plans/coverage-repair.plan.json") + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout)["ok"] is True + + +def test_plan_lint_cyclic_example_exits_nonzero(): + result = _run("plan-lint", "examples/plans/invalid/cyclic-dependency.plan.json") + assert result.returncode == 1 + report = json.loads(result.stdout) + assert report["ok"] is False + assert any(issue["code"] == "cyclic_dependency" for issue in report["issues"]) + + +def test_plan_lint_basic_mode_also_catches_cycle(): + result = _run("plan-lint", "--mode", "basic", "examples/plans/invalid/cyclic-dependency.plan.json") + assert result.returncode == 1 + assert any(issue["code"] == "cyclic_dependency" for issue in json.loads(result.stdout)["issues"]) + + +def test_plan_lint_reports_plan_schema_id(): + result = _run("plan-lint", "--mode", "basic", "examples/plans/coverage-repair.plan.json") + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout)["schemas_checked"] == ["loop-engineer/plan@1"] + + +def test_plan_lint_accepts_mode_flag_like_doctor(): + result = _run("plan-lint", "--mode", "bogus", "examples/plans/coverage-repair.plan.json") + assert result.returncode == 2 + assert "usage" in result.stderr.lower() + assert "Traceback" not in result.stderr diff --git a/scripts/test_plan_schema.py b/scripts/test_plan_schema.py new file mode 100644 index 0000000..7fc5150 --- /dev/null +++ b/scripts/test_plan_schema.py @@ -0,0 +1,225 @@ +"""Contract tests for the standalone loop-engineer/plan@1 Plan IR.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from loop import plan +from loop.contract import ValidationModeError + + +ROOT = Path(__file__).resolve().parent.parent +GOLDEN = ROOT / "examples" / "plans" / "coverage-repair.plan.json" +CYCLIC = ROOT / "examples" / "plans" / "invalid" / "cyclic-dependency.plan.json" +MISSING_GOAL = ROOT / "examples" / "plans" / "invalid" / "missing-goal.plan.json" + + +def _issues() -> list[dict]: + return [] + + +def _core_type_parity_plan() -> dict: + return { + "schema": plan.PLAN_SCHEMA_ID, + "goal": "validate required core types", + "acceptance_criteria": [{"id": 7, "description": 8}], + "tasks": [{"id": 123, "kind": None, "title": 456, "depends_on": ["a", 5, None]}], + "terminal_state_mapping": {"done": "Succeeded"}, + } + + +def test_plan_schema_has_expected_id(): + schema = json.loads((ROOT / "schemas" / "plan.schema.json").read_text(encoding="utf-8")) + assert schema["$id"] == "loop-engineer/plan@1" + + +def test_task_kinds_match_issue_vocabulary(): + assert plan.TASK_KINDS == ("agent", "tool", "gate", "approval", "join", "subloop", "human", "terminal") + + +def test_model_policy_vocabulary_matches_issue_56(): + assert plan.MODEL_POLICY_ROLES == ("read", "reason", "write", "verify") + assert plan.MODEL_CAPABILITIES == ("fast_low_cost", "deep_reasoning", "code_generation", "independent_review") + + +def test_valid_golden_example_passes_jsonschema_mode(): + pytest.importorskip("jsonschema") + report = plan.validate_plan(GOLDEN, mode="release") + assert report["ok"] is True + assert report["issues"] == [] + + +def test_valid_golden_example_passes_structural_fallback(monkeypatch): + monkeypatch.setitem(sys.modules, "jsonschema", None) + assert plan.validate_plan(GOLDEN, mode="basic")["ok"] is True + + +def test_cyclic_example_fails_both_modes_with_cyclic_dependency_code(monkeypatch): + pytest.importorskip("jsonschema") + for mode in ("release", "basic"): + if mode == "basic": + monkeypatch.setitem(sys.modules, "jsonschema", None) + report = plan.validate_plan(CYCLIC, mode=mode) + assert report["ok"] is False + assert any(issue["code"] == "cyclic_dependency" for issue in report["issues"]) + + +def test_missing_goal_example_fails_both_modes(monkeypatch): + pytest.importorskip("jsonschema") + strict = plan.validate_plan(MISSING_GOAL, mode="release") + assert any(issue["code"] == "schema_violation" and "goal" in issue["message"] for issue in strict["issues"]) + monkeypatch.setitem(sys.modules, "jsonschema", None) + basic = plan.validate_plan(MISSING_GOAL, mode="basic") + assert any(issue["code"] == "invalid_plan" for issue in basic["issues"]) + + +def test_required_core_type_errors_fail_in_both_modes_and_cli(tmp_path, monkeypatch): + pytest.importorskip("jsonschema") + source = tmp_path / "core-type-errors.json" + source.write_text(json.dumps(_core_type_parity_plan()), encoding="utf-8") + + for mode in ("release", "basic"): + if mode == "basic": + monkeypatch.setitem(sys.modules, "jsonschema", None) + report = plan.validate_plan(source, mode=mode) + assert report["ok"] is False + result = subprocess.run( + [sys.executable, "-B", "-m", "loop", "plan-lint", "--mode", mode, str(source)], + cwd=ROOT, + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 1 + + +def test_non_string_dependency_entries_are_reported_in_both_modes(tmp_path, monkeypatch): + pytest.importorskip("jsonschema") + source = tmp_path / "invalid-dependency-entries.json" + source.write_text(json.dumps({ + "schema": plan.PLAN_SCHEMA_ID, + "goal": "report invalid dependency entries", + "acceptance_criteria": [{"id": "AC1", "description": "Invalid dependency entries are rejected."}], + "tasks": [ + {"id": "a", "kind": "human", "title": "First", "depends_on": ["b", 5, None], "instructions": "Do work."}, + {"id": "b", "kind": "join", "title": "Join", "depends_on": [], "join_on": ["a", 5, None]}, + ], + "terminal_state_mapping": {"done": "Succeeded"}, + }), encoding="utf-8") + + for mode in ("release", "basic"): + if mode == "basic": + monkeypatch.setitem(sys.modules, "jsonschema", None) + report = plan.validate_plan(source, mode=mode) + assert report["ok"] is False + assert sum(issue["code"] == "invalid_dependency_entry" for issue in report["issues"]) == 4 + + +def test_cycle_detection_unit_minimal(): + issues = _issues() + plan._check_task_ids_and_dependencies([ + {"id": "a", "depends_on": ["c"]}, {"id": "b", "depends_on": ["a"]}, {"id": "c", "depends_on": ["b"]} + ], GOLDEN, issues) + assert any(issue["code"] == "cyclic_dependency" for issue in issues) + + +def test_duplicate_task_id_detected(): + issues = _issues() + plan._check_task_ids_and_dependencies([{"id": "a", "depends_on": []}, {"id": "a", "depends_on": []}], GOLDEN, issues) + assert any(issue["code"] == "duplicate_task_id" for issue in issues) + + +def test_unknown_dependency_detected(): + issues = _issues() + plan._check_task_ids_and_dependencies([{"id": "a", "depends_on": ["missing"]}], GOLDEN, issues) + assert any(issue["code"] == "unknown_dependency" for issue in issues) + + +def test_duplicate_criterion_id_detected(): + issues = _issues() + plan._check_acceptance_criteria_ids([{"id": "AC1"}, {"id": "AC1"}], GOLDEN, issues) + assert any(issue["code"] == "duplicate_criterion_id" for issue in issues) + + +@pytest.mark.parametrize("kind", plan.TASK_KINDS) +def test_each_kind_requires_its_field(kind): + issues = _issues() + plan._check_task_kind_fields([{"id": "task", "kind": kind}], GOLDEN, issues) + assert any(issue["code"] == "missing_kind_field" for issue in issues) + + +def test_join_requires_at_least_two_upstream_tasks(): + issues = _issues() + plan._check_task_kind_fields([{"id": "join", "kind": "join", "join_on": ["a"]}], GOLDEN, issues) + assert any(issue["code"] == "invalid_join" for issue in issues) + + +def test_terminal_task_state_must_be_canonical(): + issues = _issues() + plan._check_task_kind_fields([{"id": "end", "kind": "terminal", "terminal_state": "Done"}], GOLDEN, issues) + assert any(issue["code"] == "invalid_terminal_state" for issue in issues) + + +def test_terminal_state_mapping_values_must_be_canonical(): + data = {"schema": plan.PLAN_SCHEMA_ID, "goal": "x", "acceptance_criteria": [{"id": "a", "description": "x"}], "tasks": [{"id": "a", "kind": "human", "title": "x", "depends_on": [], "instructions": "x"}], "terminal_state_mapping": {"x": "Finished"}} + path = GOLDEN.parent / "temporary-not-used.json" + issues = _issues() + plan._structural_validate_plan(data, path, issues) + assert any(issue["code"] == "invalid_terminal_state" for issue in issues) + + +def test_model_policy_rejects_unknown_role_and_capability(): + for policy in ({"orchestrate": "fast_low_cost"}, {"read": "gpt-5.5"}): + issues = _issues() + plan._structural_validate_plan({"schema": plan.PLAN_SCHEMA_ID, "goal": "x", "acceptance_criteria": [{"id": "a", "description": "x"}], "tasks": [{"id": "a", "kind": "human", "title": "x", "depends_on": [], "instructions": "x"}], "terminal_state_mapping": {"x": "Succeeded"}, "model_policy": policy}, GOLDEN, issues) + assert any(issue["code"] == "invalid_model_policy" for issue in issues) + + +def test_approval_task_without_declared_gates_flagged(): + issues = _issues() + plan._check_approval_gates({}, [{"id": "a", "kind": "approval", "approval_gate": "gate"}], GOLDEN, issues) + assert any(issue["code"] == "missing_approval_gates" for issue in issues) + + +def test_approval_task_references_undeclared_gate(): + issues = _issues() + plan._check_approval_gates({"approval_gates": ["a"]}, [{"id": "b", "kind": "approval", "approval_gate": "b"}], GOLDEN, issues) + assert any(issue["code"] == "unknown_approval_gate" for issue in issues) + + +def test_completion_policy_reuses_shared_normalizer(): + issues = _issues() + plan._structural_validate_plan({"schema": plan.PLAN_SCHEMA_ID, "goal": "x", "acceptance_criteria": [{"id": "a", "description": "x"}], "tasks": [{"id": "a", "kind": "human", "title": "x", "depends_on": [], "instructions": "x"}], "terminal_state_mapping": {"x": "Succeeded"}, "completion_policy": {"mode": "bogus"}}, GOLDEN, issues) + assert any(issue["code"] == "invalid_plan" and "completion_policy" in issue["message"] for issue in issues) + + +def test_strict_mode_without_jsonschema_raises(monkeypatch): + monkeypatch.setitem(sys.modules, "jsonschema", None) + with pytest.raises(ValidationModeError): + plan.validate_plan(GOLDEN, mode="strict") + + +def test_missing_file_reports_missing_file_issue(tmp_path): + report = plan.validate_plan(tmp_path / "missing.json", mode="basic") + assert any(issue["code"] == "missing_file" for issue in report["issues"]) + + +def test_directory_target_reports_invalid_target(tmp_path): + report = plan.validate_plan(tmp_path, mode="basic") + assert any(issue["code"] == "invalid_target" for issue in report["issues"]) + + +def test_malformed_json_reports_invalid_json(tmp_path): + source = tmp_path / "malformed.json" + source.write_text("{", encoding="utf-8") + report = plan.validate_plan(source, mode="basic") + assert any(issue["code"] == "invalid_json" for issue in report["issues"]) + + +def test_schemas_checked_names_plan_schema_id(): + assert plan.validate_plan(GOLDEN, mode="basic")["schemas_checked"] == ["loop-engineer/plan@1"] diff --git a/scripts/test_wheel_selfcontained.py b/scripts/test_wheel_selfcontained.py index e59ff37..d56936c 100644 --- a/scripts/test_wheel_selfcontained.py +++ b/scripts/test_wheel_selfcontained.py @@ -92,3 +92,14 @@ def test_both_console_scripts_are_installed(wheel_env, tmp_path): proc = subprocess.run([str(exe), "--version"], cwd=tmp_path, capture_output=True, text=True) assert proc.returncode == 0, f"{name}: {proc.stderr}" assert proc.stdout.strip() + + +def test_plan_lint_from_wheel_only(wheel_env, tmp_path): + # No jsonschema in this venv (pip wheel --no-deps): proves plan-lint's + # structural-fallback mode genuinely runs from a repo-checkout-free install. + plan_file = REPO_ROOT / "examples" / "plans" / "coverage-repair.plan.json" + result = _run(wheel_env, ["plan-lint", str(plan_file)], cwd=tmp_path) + assert result.returncode == 0, result.stdout + result.stderr + report = json.loads(result.stdout) + assert report["ok"] is True + assert report["validation_mode"] == "structural-fallback"