electron-rare/Aperant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(planner): enforce implementation_plan schema (issue #884) (#912)

Browse Source 
* fix(planner): enforce implementation_plan schema

* fix(logs): sync planning->coding phase to source

* style(backend): ruff format

* fix(progress): backfill empty description from title

* fix(planner): prevent post-processing during planning

* fix(planner): normalize phase_id and reuse alias mapping

* fix(planner): address review suggestions

* fix(progress): prevent phase field overrides

* test(planner): avoid hardcoded planner.md path

* fix(auto-fix): normalize phase_id and depends_on

* fix(planner): harden plan normalization edge cases

* fix(auto-fix): handle file I/O errors
This commit is contained in:
Umaru
2026-01-11 21:07:25 +08:00
committed by GitHub
parent c4e08aeec2
commit 29d28bf0a2
8 changed files with 744 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/agents/coder.py
+70 -5
View File
@@ -136,6 +136,25 @@ async def run_autonomous_agent(
# Track which phase we're in for logging
current_log_phase = LogPhase.CODING
is_planning_phase = False
planning_retry_context: str | None = None
planning_validation_failures = 0
max_planning_validation_retries = 3
def _validate_and_fix_implementation_plan() -> tuple[bool, list[str]]:
from spec.validate_pkg import SpecValidator, auto_fix_plan
spec_validator = SpecValidator(spec_dir)
result = spec_validator.validate_implementation_plan()
if result.valid:
return True, []
fixed = auto_fix_plan(spec_dir)
if fixed:
result = spec_validator.validate_implementation_plan()
if result.valid:
return True, []
return False, result.errors
if first_run:
print_status(
@@ -223,8 +242,8 @@ async def run_autonomous_agent(
print("To continue, run the script again without --max-iterations")
break
# Get the next subtask to work on
next_subtask = get_next_subtask(spec_dir)
# Get the next subtask to work on (planner sessions shouldn't bind to a subtask)
next_subtask = None if first_run else get_next_subtask(spec_dir)
subtask_id = next_subtask.get("id") if next_subtask else None
phase_name = next_subtask.get("phase_name") if next_subtask else None
@@ -275,6 +294,8 @@ async def run_autonomous_agent(
# Generate appropriate prompt
if first_run:
prompt = generate_planner_prompt(spec_dir, project_dir)
if planning_retry_context:
prompt += "\n\n" + planning_retry_context
# Retrieve Graphiti memory context for planning phase
# This gives the planner knowledge of previous patterns, gotchas, and insights
@@ -311,6 +332,10 @@ async def run_autonomous_agent(
task_logger.start_phase(
LogPhase.CODING, "Starting implementation..."
)
# In worktree mode, the UI prefers planning logs from the main spec dir.
# Ensure the planning->coding transition is immediately reflected there.
if sync_spec_to_source(spec_dir, source_spec_dir):
print_status("Phase transition synced to main project", "success")
if not next_subtask:
print("No pending subtasks found - build may be complete!")
@@ -369,8 +394,46 @@ async def run_autonomous_agent(
client, prompt, spec_dir, verbose, phase=current_log_phase
)
plan_validated = False
if is_planning_phase and status != "error":
valid, errors = _validate_and_fix_implementation_plan()
if valid:
plan_validated = True
planning_retry_context = None
else:
planning_validation_failures += 1
if planning_validation_failures >= max_planning_validation_retries:
print_status(
"implementation_plan.json validation failed too many times",
"error",
)
for err in errors:
print(f" - {err}")
status_manager.update(state=BuildState.ERROR)
return
print_status(
"implementation_plan.json invalid - retrying planner", "warning"
)
for err in errors:
print(f" - {err}")
planning_retry_context = (
"## IMPLEMENTATION PLAN VALIDATION ERRORS\n\n"
"The previous `implementation_plan.json` is INVALID.\n"
"You MUST rewrite it to match the required schema:\n"
"- Top-level: `feature`, `workflow_type`, `phases`\n"
"- Each phase: `id` (or `phase`) and `name`, and `subtasks`\n"
"- Each subtask: `id`, `description`, `status` (use `pending` for not started)\n\n"
"Validation errors:\n" + "\n".join(f"- {e}" for e in errors)
)
# Stay in planning mode for the next iteration
first_run = True
status = "continue"
# === POST-SESSION PROCESSING (100% reliable) ===
if subtask_id and not first_run:
# Only run post-session processing for coding sessions.
if subtask_id and current_log_phase == LogPhase.CODING:
linear_is_enabled = (
linear_task is not None and linear_task.task_id is not None
)
@@ -408,7 +471,7 @@ async def run_autonomous_agent(
attempt_count=attempt_count,
)
print_status("Linear notified of stuck subtask", "info")
elif is_planning_phase and source_spec_dir:
elif plan_validated and source_spec_dir:
# After planning phase, sync the newly created implementation plan back to source
if sync_spec_to_source(spec_dir, source_spec_dir):
print_status("Implementation plan synced to main project", "success")
@@ -442,7 +505,9 @@ async def run_autonomous_agent(
print_progress_summary(spec_dir)
# Update state back to building
status_manager.update(state=BuildState.BUILDING)
status_manager.update(
state=BuildState.PLANNING if is_planning_phase else BuildState.BUILDING
)
# Show next subtask info
next_subtask = get_next_subtask(spec_dir)
apps/backend/core/plan_normalization.py
+50
View File
@@ -0,0 +1,50 @@
"""
Implementation Plan Normalization Utilities
===========================================
Small helpers for normalizing common LLM/legacy field variants in
implementation_plan.json without changing status semantics.
"""
from typing import Any
def normalize_subtask_aliases(subtask: dict[str, Any]) -> tuple[dict[str, Any], bool]:
"""Normalize common subtask field aliases.
- If `id` is missing and `subtask_id` exists, copy it into `id` as a string.
- If `description` is missing/empty and `title` is a non-empty string, copy it
into `description`.
"""
normalized = dict(subtask)
changed = False
id_value = normalized.get("id")
id_missing = (
"id" not in normalized
or id_value is None
or (isinstance(id_value, str) and not id_value.strip())
)
if id_missing and "subtask_id" in normalized:
subtask_id = normalized.get("subtask_id")
if subtask_id is not None:
subtask_id_str = str(subtask_id).strip()
if subtask_id_str:
normalized["id"] = subtask_id_str
changed = True
description_value = normalized.get("description")
description_missing = (
"description" not in normalized
or description_value is None
or (isinstance(description_value, str) and not description_value.strip())
)
title = normalized.get("title")
if description_missing and isinstance(title, str):
title_str = title.strip()
if title_str:
normalized["description"] = title_str
changed = True
return normalized, changed
apps/backend/core/progress.py
+31 -12
View File
@@ -11,6 +11,7 @@ Enhanced with colored output, icons, and better visual formatting.
import json
from pathlib import Path
from core.plan_normalization import normalize_subtask_aliases
from ui import (
Icons,
bold,
@@ -379,7 +380,7 @@ def get_current_phase(spec_dir: Path) -> dict | None:
plan = json.load(f)
for phase in plan.get("phases", []):
subtasks = phase.get("subtasks", [])
subtasks = phase.get("subtasks", phase.get("chunks", []))
# Phase is current if it has incomplete subtasks and dependencies are met
has_incomplete = any(s.get("status") != "completed" for s in subtasks)
if has_incomplete:
@@ -415,24 +416,39 @@ def get_next_subtask(spec_dir: Path) -> dict | None:
return None
try:
with open(plan_file) as f:
with open(plan_file, encoding="utf-8") as f:
plan = json.load(f)
phases = plan.get("phases", [])
# Build a map of phase completion
phase_complete = {}
for phase in phases:
phase_id = phase.get("id") or phase.get("phase")
subtasks = phase.get("subtasks", [])
phase_complete[phase_id] = all(
phase_complete: dict[str, bool] = {}
for i, phase in enumerate(phases):
phase_id_value = phase.get("id")
phase_id_raw = (
phase_id_value if phase_id_value is not None else phase.get("phase")
)
phase_id_key = (
str(phase_id_raw) if phase_id_raw is not None else f"unknown:{i}"
)
subtasks = phase.get("subtasks", phase.get("chunks", []))
phase_complete[phase_id_key] = all(
s.get("status") == "completed" for s in subtasks
)
# Find next available subtask
for phase in phases:
phase_id = phase.get("id") or phase.get("phase")
depends_on = phase.get("depends_on", [])
phase_id_value = phase.get("id")
phase_id = (
phase_id_value if phase_id_value is not None else phase.get("phase")
)
depends_on_raw = phase.get("depends_on", [])
if isinstance(depends_on_raw, list):
depends_on = [str(d) for d in depends_on_raw if d is not None]
elif depends_on_raw is None:
depends_on = []
else:
depends_on = [str(depends_on_raw)]
# Check if dependencies are satisfied
deps_satisfied = all(phase_complete.get(dep, False) for dep in depends_on)
@@ -440,13 +456,16 @@ def get_next_subtask(spec_dir: Path) -> dict | None:
continue
# Find first pending subtask in this phase
for subtask in phase.get("subtasks", []):
if subtask.get("status") == "pending":
for subtask in phase.get("subtasks", phase.get("chunks", [])):
status = subtask.get("status", "pending")
if status in {"pending", "not_started", "not started"}:
subtask_out, _changed = normalize_subtask_aliases(subtask)
subtask_out["status"] = "pending"
return {
**subtask_out,
"phase_id": phase_id,
"phase_name": phase.get("name"),
"phase_num": phase.get("phase"),
**subtask,
}
return None
apps/backend/prompts_pkg/prompt_generator.py
+15 -5
View File
@@ -253,12 +253,22 @@ def generate_planner_prompt(spec_dir: Path, project_dir: Path | None = None) ->
Returns:
Planner prompt string
"""
# Load the full planner prompt from file
prompts_dir = Path(__file__).parent / "prompts"
planner_file = prompts_dir / "planner.md"
# Load the full planner prompt from file.
candidate_dirs = [
Path(__file__).parent.parent / "prompts", # current layout
Path(__file__).parent / "prompts", # legacy fallback (if any)
]
planner_file = next(
(
(candidate_dir / "planner.md")
for candidate_dir in candidate_dirs
if (candidate_dir / "planner.md").exists()
),
None,
)
if planner_file.exists():
prompt = planner_file.read_text()
if planner_file:
prompt = planner_file.read_text(encoding="utf-8")
else:
prompt = (
"Read spec.md and create implementation_plan.json with phases and subtasks."
apps/backend/prompts_pkg/prompts.py
+1 -1
View File
@@ -394,7 +394,7 @@ def is_first_run(spec_dir: Path) -> bool:
return True
try:
with open(plan_file) as f:
with open(plan_file, encoding="utf-8") as f:
plan = json.load(f)
# Check if there are any phases with subtasks
apps/backend/spec/validate_pkg/auto_fix.py
+106 -6
View File
@@ -8,6 +8,29 @@ Automated fixes for common implementation plan issues.
import json
from pathlib import Path
from core.plan_normalization import normalize_subtask_aliases
def _normalize_status(value: object) -> str:
"""Normalize common status variants to schema-compliant values."""
if not isinstance(value, str):
return "pending"
normalized = value.strip().lower()
if normalized in {"pending", "in_progress", "completed", "blocked", "failed"}:
return normalized
# Common non-standard variants produced by LLMs or legacy tooling
if normalized in {"not_started", "not started", "todo", "to_do", "backlog"}:
return "pending"
if normalized in {"in-progress", "inprogress", "working"}:
return "in_progress"
if normalized in {"done", "complete", "completed_successfully"}:
return "completed"
# Unknown values fall back to pending to prevent deadlocks in execution
return "pending"
def auto_fix_plan(spec_dir: Path) -> bool:
"""Attempt to auto-fix common implementation_plan.json issues.
@@ -24,16 +47,33 @@ def auto_fix_plan(spec_dir: Path) -> bool:
return False
try:
with open(plan_file) as f:
with open(plan_file, encoding="utf-8") as f:
plan = json.load(f)
except json.JSONDecodeError:
except (OSError, json.JSONDecodeError):
return False
fixed = False
# Support older/simple plans that use top-level "subtasks" (or "chunks")
if "phases" not in plan and (
isinstance(plan.get("subtasks"), list) or isinstance(plan.get("chunks"), list)
):
subtasks = plan.get("subtasks") or plan.get("chunks") or []
plan["phases"] = [
{
"id": "1",
"phase": 1,
"name": "Phase 1",
"subtasks": subtasks,
}
]
plan.pop("subtasks", None)
plan.pop("chunks", None)
fixed = True
# Fix missing top-level fields
if "feature" not in plan:
plan["feature"] = "Unnamed Feature"
plan["feature"] = plan.get("title") or plan.get("spec_id") or "Unnamed Feature"
fixed = True
if "workflow_type" not in plan:
@@ -46,20 +86,72 @@ def auto_fix_plan(spec_dir: Path) -> bool:
# Fix phases
for i, phase in enumerate(plan.get("phases", [])):
# Normalize common phase field aliases
if "name" not in phase and "title" in phase:
phase["name"] = phase.get("title")
fixed = True
if "phase" not in phase and "phase_id" in phase:
phase_id = phase.get("phase_id")
phase_id_str = str(phase_id).strip() if phase_id is not None else ""
phase_num: int | None = None
if isinstance(phase_id, int) and not isinstance(phase_id, bool):
phase_num = phase_id
elif (
isinstance(phase_id, float)
and not isinstance(phase_id, bool)
and phase_id.is_integer()
):
phase_num = int(phase_id)
elif isinstance(phase_id, str) and phase_id_str.isdigit():
phase_num = int(phase_id_str)
if phase_num is not None:
if "id" not in phase:
phase["id"] = str(phase_num)
fixed = True
phase["phase"] = phase_num
fixed = True
elif "id" not in phase and phase_id is not None:
phase["id"] = phase_id_str
fixed = True
if "phase" not in phase:
phase["phase"] = i + 1
fixed = True
depends_on_raw = phase.get("depends_on", [])
if isinstance(depends_on_raw, list):
normalized_depends_on = [
str(d).strip() for d in depends_on_raw if d is not None
]
elif depends_on_raw is None:
normalized_depends_on = []
else:
normalized_depends_on = [str(depends_on_raw).strip()]
if normalized_depends_on != depends_on_raw:
phase["depends_on"] = normalized_depends_on
fixed = True
if "name" not in phase:
phase["name"] = f"Phase {i + 1}"
fixed = True
if "subtasks" not in phase:
phase["subtasks"] = []
phase["subtasks"] = phase.get("chunks", [])
fixed = True
elif "chunks" in phase and not phase.get("subtasks"):
# If subtasks exists but is empty, fall back to chunks if present
phase["subtasks"] = phase.get("chunks", [])
fixed = True
# Fix subtasks
for j, subtask in enumerate(phase.get("subtasks", [])):
normalized, changed = normalize_subtask_aliases(subtask)
if changed:
subtask.update(normalized)
fixed = True
if "id" not in subtask:
subtask["id"] = f"subtask-{i + 1}-{j + 1}"
fixed = True
@@ -71,10 +163,18 @@ def auto_fix_plan(spec_dir: Path) -> bool:
if "status" not in subtask:
subtask["status"] = "pending"
fixed = True
else:
normalized_status = _normalize_status(subtask.get("status"))
if subtask.get("status") != normalized_status:
subtask["status"] = normalized_status
fixed = True
if fixed:
with open(plan_file, "w") as f:
json.dump(plan, f, indent=2)
try:
with open(plan_file, "w", encoding="utf-8") as f:
json.dump(plan, f, indent=2, ensure_ascii=False)
except OSError:
return False
print(f"Auto-fixed: {plan_file}")
return fixed
tests/test_agent_architecture.py
+44 -20
View File
@@ -30,7 +30,9 @@ class TestNoExternalParallelism:
def test_no_coordinator_module(self):
"""No external coordinator module should exist."""
coordinator_path = Path(__file__).parent.parent / "apps" / "backend" / "coordinator.py"
coordinator_path = (
Path(__file__).parent.parent / "apps" / "backend" / "coordinator.py"
)
assert not coordinator_path.exists(), (
"coordinator.py should not exist. Parallel orchestration is handled "
"internally by the agent using Claude Code's Task tool."
@@ -38,7 +40,9 @@ class TestNoExternalParallelism:
def test_no_task_tool_module(self):
"""No task_tool wrapper module should exist."""
task_tool_path = Path(__file__).parent.parent / "apps" / "backend" / "task_tool.py"
task_tool_path = (
Path(__file__).parent.parent / "apps" / "backend" / "task_tool.py"
)
assert not task_tool_path.exists(), (
"task_tool.py should not exist. The agent spawns subagents directly "
"using Claude Code's built-in Task tool."
@@ -46,7 +50,9 @@ class TestNoExternalParallelism:
def test_no_subtask_worker_config(self):
"""No external subtask worker agent config should exist."""
worker_config = Path(__file__).parent.parent / ".claude" / "agents" / "subtask-worker.md"
worker_config = (
Path(__file__).parent.parent / ".claude" / "agents" / "subtask-worker.md"
)
assert not worker_config.exists(), (
"subtask-worker.md should not exist. Subagents use Claude Code's "
"built-in agent types, not custom configs."
@@ -59,7 +65,7 @@ class TestCLIInterface:
def test_no_parallel_flag(self):
"""CLI should not have --parallel argument."""
run_py_path = Path(__file__).parent.parent / "apps" / "backend" / "run.py"
content = run_py_path.read_text()
content = run_py_path.read_text(encoding="utf-8")
# Check that --parallel is not defined as an argument
assert '"--parallel"' not in content, (
@@ -74,7 +80,7 @@ class TestCLIInterface:
def test_no_parallel_examples_in_docs(self):
"""CLI documentation should not mention parallel mode."""
run_py_path = Path(__file__).parent.parent / "apps" / "backend" / "run.py"
content = run_py_path.read_text()
content = run_py_path.read_text(encoding="utf-8")
# The docstring should not have --parallel examples
assert "--parallel" not in content[:2000], (
@@ -125,8 +131,10 @@ class TestAgentPrompt:
def test_mentions_subagents(self):
"""Agent prompt mentions subagent capability."""
coder_prompt_path = Path(__file__).parent.parent / "apps" / "backend" / "prompts" / "coder.md"
content = coder_prompt_path.read_text()
coder_prompt_path = (
Path(__file__).parent.parent / "apps" / "backend" / "prompts" / "coder.md"
)
content = coder_prompt_path.read_text(encoding="utf-8")
assert "subagent" in content.lower(), (
"Agent prompt should document subagent capability for parallel work."
@@ -134,12 +142,16 @@ class TestAgentPrompt:
def test_mentions_parallel_capability(self):
"""Agent prompt mentions parallel/concurrent capability."""
coder_prompt_path = Path(__file__).parent.parent / "apps" / "backend" / "prompts" / "coder.md"
content = coder_prompt_path.read_text()
coder_prompt_path = (
Path(__file__).parent.parent / "apps" / "backend" / "prompts" / "coder.md"
)
content = coder_prompt_path.read_text(encoding="utf-8")
has_task_tool = "task tool" in content.lower() or "Task tool" in content
has_parallel = "parallel" in content.lower()
has_concurrent = "concurrent" in content.lower() or "simultaneously" in content.lower()
has_concurrent = (
"concurrent" in content.lower() or "simultaneously" in content.lower()
)
assert has_task_tool or has_parallel or has_concurrent, (
"Agent prompt should mention parallel/concurrent work capability."
@@ -159,7 +171,7 @@ class TestModuleIntegrity:
def test_run_module_valid_syntax(self):
"""Run module has valid Python syntax."""
run_py_path = Path(__file__).parent.parent / "apps" / "backend" / "run.py"
content = run_py_path.read_text()
content = run_py_path.read_text(encoding="utf-8")
try:
ast.parse(content)
@@ -170,7 +182,7 @@ class TestModuleIntegrity:
"""Core modules don't import coordinator."""
for filename in ["run.py", "core/agent.py"]:
filepath = Path(__file__).parent.parent / "apps" / "backend" / filename
content = filepath.read_text()
content = filepath.read_text(encoding="utf-8")
assert "from coordinator import" not in content, (
f"{filename} should not import coordinator"
@@ -183,7 +195,7 @@ class TestModuleIntegrity:
"""Core modules don't import task_tool."""
for filename in ["run.py", "core/agent.py"]:
filepath = Path(__file__).parent.parent / "apps" / "backend" / filename
content = filepath.read_text()
content = filepath.read_text(encoding="utf-8")
assert "from task_tool import" not in content, (
f"{filename} should not import task_tool"
@@ -230,7 +242,9 @@ class TestElectronToolScoping:
# Must pass is_electron=True for Electron tools to be included
# This is the new phase-aware behavior
qa_tools = get_allowed_tools("qa_reviewer", project_capabilities={"is_electron": True})
qa_tools = get_allowed_tools(
"qa_reviewer", project_capabilities={"is_electron": True}
)
# At least one Electron tool should be present
has_electron = any("electron" in tool.lower() for tool in qa_tools)
@@ -250,7 +264,9 @@ class TestElectronToolScoping:
from auto_claude_tools import ELECTRON_TOOLS, get_allowed_tools
# Must pass is_electron=True for Electron tools to be included
qa_fixer_tools = get_allowed_tools("qa_fixer", project_capabilities={"is_electron": True})
qa_fixer_tools = get_allowed_tools(
"qa_fixer", project_capabilities={"is_electron": True}
)
has_electron = any("electron" in tool.lower() for tool in qa_fixer_tools)
assert has_electron, (
@@ -268,7 +284,9 @@ class TestElectronToolScoping:
from auto_claude_tools import get_allowed_tools
# Even with is_electron=True, coder should not get Electron tools
coder_tools = get_allowed_tools("coder", project_capabilities={"is_electron": True})
coder_tools = get_allowed_tools(
"coder", project_capabilities={"is_electron": True}
)
has_electron = any("electron" in tool.lower() for tool in coder_tools)
assert not has_electron, (
@@ -283,7 +301,9 @@ class TestElectronToolScoping:
from auto_claude_tools import get_allowed_tools
# Even with is_electron=True, planner should not get Electron tools
planner_tools = get_allowed_tools("planner", project_capabilities={"is_electron": True})
planner_tools = get_allowed_tools(
"planner", project_capabilities={"is_electron": True}
)
has_electron = any("electron" in tool.lower() for tool in planner_tools)
assert not has_electron, (
@@ -299,7 +319,9 @@ class TestElectronToolScoping:
for agent_type in ["planner", "coder", "qa_reviewer", "qa_fixer"]:
# Even with is_electron=True, no tools without env var
tools = get_allowed_tools(agent_type, project_capabilities={"is_electron": True})
tools = get_allowed_tools(
agent_type, project_capabilities={"is_electron": True}
)
has_electron = any("electron" in tool.lower() for tool in tools)
assert not has_electron, (
f"{agent_type} should NOT have Electron tools when ELECTRON_MCP_ENABLED is not set"
@@ -311,8 +333,10 @@ class TestSubtaskTerminology:
def test_progress_uses_subtask_terminology(self):
"""Progress module uses subtask terminology."""
progress_path = Path(__file__).parent.parent / "apps" / "backend" / "core" / "progress.py"
content = progress_path.read_text()
progress_path = (
Path(__file__).parent.parent / "apps" / "backend" / "core" / "progress.py"
)
content = progress_path.read_text(encoding="utf-8")
assert "subtask" in content.lower(), (
"core/progress.py should use subtask terminology"
tests/test_issue_884_plan_schema.py
+427
View File
@@ -0,0 +1,427 @@
#!/usr/bin/env python3
"""
Regression tests for issue #884.
The planner may generate a non-standard implementation_plan.json schema
(`not_started`, `phase_id`, `subtask_id`, `title`, etc.) which can cause
execution to get stuck because no "pending" subtasks are detected.
"""
import importlib
import json
from pathlib import Path
import pytest
from core.progress import get_next_subtask
from prompt_generator import generate_planner_prompt
from spec.validate_pkg import SpecValidator, auto_fix_plan
def _write_plan(path: Path, data: dict) -> None:
path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
def test_generate_planner_prompt_loads_repo_planner_md(spec_dir: Path):
prompt = generate_planner_prompt(spec_dir, project_dir=spec_dir.parent)
prompt_generator = importlib.import_module(generate_planner_prompt.__module__)
assert prompt_generator.__file__ is not None
candidate_dirs = [
Path(prompt_generator.__file__).parent.parent / "prompts", # current layout
Path(prompt_generator.__file__).parent / "prompts", # legacy fallback (if any)
]
planner_file = next(
(
(candidate_dir / "planner.md")
for candidate_dir in candidate_dirs
if (candidate_dir / "planner.md").exists()
),
None,
)
assert planner_file is not None
planner_md = planner_file.read_text(encoding="utf-8").strip()
assert planner_md in prompt
def test_get_next_subtask_accepts_not_started_and_alias_fields(spec_dir: Path):
plan = {
"spec_id": "002-add-upstream-connection-test",
"phases": [
{
"phase_id": "1",
"title": "Research & Design",
"status": "not_started",
"subtasks": [
{
"subtask_id": "1.1",
"title": "Research provider-specific test endpoints",
"status": "not_started",
}
],
}
],
}
_write_plan(spec_dir / "implementation_plan.json", plan)
next_task = get_next_subtask(spec_dir)
assert next_task is not None
assert next_task.get("id") == "1.1"
assert next_task.get("description") == "Research provider-specific test endpoints"
assert next_task.get("status") == "pending"
def test_get_next_subtask_populates_description_from_title_when_empty(spec_dir: Path):
plan = {
"spec_id": "002-add-upstream-connection-test",
"phases": [
{
"phase_id": "1",
"title": "Research & Design",
"status": "not_started",
"subtasks": [
{
"subtask_id": "1.1",
"title": "Research provider-specific test endpoints",
"description": "",
"status": "not_started",
}
],
}
],
}
_write_plan(spec_dir / "implementation_plan.json", plan)
next_task = get_next_subtask(spec_dir)
assert next_task is not None
assert next_task.get("id") == "1.1"
assert next_task.get("description") == "Research provider-specific test endpoints"
assert next_task.get("status") == "pending"
def test_get_next_subtask_handles_depends_on_with_mixed_id_types(spec_dir: Path):
plan = {
"feature": "Test feature",
"workflow_type": "feature",
"phases": [
{
"phase": 1,
"name": "Phase 1",
"subtasks": [
{"id": "1.1", "description": "Done", "status": "completed"},
],
},
{
"phase": 2,
"name": "Phase 2",
"depends_on": ["1"],
"subtasks": [
{"id": "2.1", "description": "Next", "status": "pending"},
],
},
],
}
_write_plan(spec_dir / "implementation_plan.json", plan)
next_task = get_next_subtask(spec_dir)
assert next_task is not None
assert next_task.get("id") == "2.1"
def test_get_next_subtask_phase_fields_override_malformed_subtask_phase_fields(
spec_dir: Path,
):
plan = {
"feature": "Test feature",
"workflow_type": "feature",
"phases": [
{
"id": "phase-1",
"name": "Phase 1",
"phase": 1,
"subtasks": [
{
"id": "1.1",
"description": "Do thing",
"status": "pending",
"phase_id": "bad-phase",
"phase_name": "Bad Phase",
"phase_num": 999,
}
],
}
],
}
_write_plan(spec_dir / "implementation_plan.json", plan)
next_task = get_next_subtask(spec_dir)
assert next_task is not None
assert next_task.get("id") == "1.1"
assert next_task.get("phase_id") == "phase-1"
assert next_task.get("phase_name") == "Phase 1"
assert next_task.get("phase_num") == 1
def test_auto_fix_plan_normalizes_nonstandard_schema_and_validates(spec_dir: Path):
plan = {
"spec_id": "002-add-upstream-connection-test",
"phases": [
{
"phase_id": "1",
"title": "Research & Design",
"status": "not_started",
"subtasks": [
{
"subtask_id": "1.1",
"title": "Research provider-specific test endpoints",
"description": "Research lightweight API endpoints for each provider",
"status": "not_started",
"files_to_modify": [],
"notes": "",
}
],
}
],
}
plan_path = spec_dir / "implementation_plan.json"
_write_plan(plan_path, plan)
fixed = auto_fix_plan(spec_dir)
assert fixed is True
loaded = json.loads(plan_path.read_text(encoding="utf-8"))
assert loaded.get("feature")
assert loaded.get("workflow_type")
assert loaded.get("phases")
assert loaded["phases"][0].get("name") == "Research & Design"
subtask = loaded["phases"][0]["subtasks"][0]
assert subtask.get("id") == "1.1"
assert subtask.get("description")
assert subtask.get("status") == "pending"
result = SpecValidator(spec_dir).validate_implementation_plan()
assert result.valid is True
def test_auto_fix_plan_normalizes_numeric_phase_ids_for_depends_on_validation(
spec_dir: Path,
):
plan = {
"feature": "Test feature",
"workflow_type": "feature",
"phases": [
{
"phase_id": "1",
"title": "Phase 1",
"subtasks": [
{"id": "1.1", "description": "Done", "status": "completed"}
],
},
{
"phase_id": "2",
"title": "Phase 2",
"depends_on": ["1"],
"subtasks": [{"id": "2.1", "description": "Next", "status": "pending"}],
},
],
}
plan_path = spec_dir / "implementation_plan.json"
_write_plan(plan_path, plan)
fixed = auto_fix_plan(spec_dir)
assert fixed is True
loaded = json.loads(plan_path.read_text(encoding="utf-8"))
assert loaded["phases"][0]["id"] == "1"
assert loaded["phases"][0]["phase"] == 1
assert SpecValidator(spec_dir).validate_implementation_plan().valid is True
def test_auto_fix_plan_sets_phase_from_numeric_phase_id_even_with_existing_id(
spec_dir: Path,
):
plan = {
"feature": "Test feature",
"workflow_type": "feature",
"phases": [
{
"id": "phase-foo",
"phase_id": 2,
"name": "Phase Foo",
"subtasks": [
{"id": "2.1", "description": "Do thing", "status": "pending"},
],
}
],
}
plan_path = spec_dir / "implementation_plan.json"
_write_plan(plan_path, plan)
fixed = auto_fix_plan(spec_dir)
assert fixed is True
loaded = json.loads(plan_path.read_text(encoding="utf-8"))
assert loaded["phases"][0]["id"] == "phase-foo"
assert loaded["phases"][0]["phase"] == 2
assert SpecValidator(spec_dir).validate_implementation_plan().valid is True
@pytest.mark.asyncio
async def test_planner_session_does_not_trigger_post_session_processing_on_retry(
temp_git_repo: Path, monkeypatch: pytest.MonkeyPatch
):
"""
Regression: planner retries shouldn't trigger coder-only post-session processing.
Even if a (malformed) implementation plan already contains something that would
normally be detected as a pending subtask, planner sessions must not execute the
coding post-processing pipeline.
"""
from agents.coder import run_autonomous_agent
from task_logger import LogPhase
spec_dir = temp_git_repo / ".auto-claude" / "specs" / "001-test"
spec_dir.mkdir(parents=True, exist_ok=True)
(spec_dir / "spec.md").write_text("# Test spec\n", encoding="utf-8")
class DummyClient:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return False
def fake_create_client(*_args, **_kwargs):
return DummyClient()
async def fake_get_graphiti_context(*_args, **_kwargs):
return None
def fake_get_next_subtask(_spec_dir: Path):
# This would have caused post-session processing to run during planning
# prior to the regression fix.
return {"id": "1.1", "description": "Should not be processed in planning"}
async def fake_post_session_processing(*_args, **_kwargs):
raise AssertionError("post_session_processing must not run during planning")
async def fake_run_agent_session(
_client,
_message: str,
_spec_dir: Path,
_verbose: bool = False,
phase: LogPhase = LogPhase.CODING,
) -> tuple[str, str]:
assert phase == LogPhase.PLANNING
return "error", "planner failed"
monkeypatch.setattr("agents.coder.create_client", fake_create_client)
monkeypatch.setattr("agents.coder.get_graphiti_context", fake_get_graphiti_context)
monkeypatch.setattr("agents.coder.get_next_subtask", fake_get_next_subtask)
monkeypatch.setattr(
"agents.coder.post_session_processing", fake_post_session_processing
)
monkeypatch.setattr("agents.coder.run_agent_session", fake_run_agent_session)
monkeypatch.setattr("agents.coder.AUTO_CONTINUE_DELAY_SECONDS", 0)
monkeypatch.setattr("agents.coder.load_subtask_context", lambda *_a, **_k: {})
await run_autonomous_agent(
project_dir=temp_git_repo,
spec_dir=spec_dir,
model="test-model",
max_iterations=1,
verbose=False,
)
@pytest.mark.asyncio
async def test_worktree_planning_to_coding_sync_updates_source_phase_status(
temp_git_repo: Path, monkeypatch: pytest.MonkeyPatch
):
"""
In worktree mode, planning logs are preferred from the main spec dir.
Ensure planning is marked completed in the source spec BEFORE the first coding session starts.
"""
from agents.coder import run_autonomous_agent
from task_logger import LogPhase
worktree_spec_dir = temp_git_repo / ".worktrees" / "001-test" / "specs" / "001-test"
source_spec_dir = temp_git_repo / ".auto-claude" / "specs" / "001-test"
worktree_spec_dir.mkdir(parents=True, exist_ok=True)
source_spec_dir.mkdir(parents=True, exist_ok=True)
for d in (worktree_spec_dir, source_spec_dir):
(d / "spec.md").write_text("# Test spec\n", encoding="utf-8")
class DummyClient:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return False
def fake_create_client(*_args, **_kwargs):
return DummyClient()
async def fake_get_graphiti_context(*_args, **_kwargs):
return None
async def fake_post_session_processing(*_args, **_kwargs):
return True
async def fake_run_agent_session(
_client,
_message: str,
spec_dir: Path,
_verbose: bool = False,
phase: LogPhase = LogPhase.CODING,
) -> tuple[str, str]:
if phase == LogPhase.PLANNING:
plan = {
"feature": "Test feature",
"workflow_type": "feature",
"phases": [
{
"id": "1",
"name": "Phase 1",
"subtasks": [
{
"id": "1.1",
"description": "Do thing",
"status": "pending",
}
],
}
],
}
(spec_dir / "implementation_plan.json").write_text(
json.dumps(plan, indent=2),
encoding="utf-8",
)
return "continue", "planned"
# First coding session should see planning already completed in source spec logs
# Note: task_logs.json is created/synced by run_autonomous_agent; absence indicates a bug.
logs = json.loads(
(source_spec_dir / "task_logs.json").read_text(encoding="utf-8")
)
assert logs["phases"]["planning"]["status"] == "completed"
assert logs["phases"]["coding"]["status"] == "active"
return "complete", "done"
monkeypatch.setattr("agents.coder.create_client", fake_create_client)
monkeypatch.setattr("agents.coder.get_graphiti_context", fake_get_graphiti_context)
monkeypatch.setattr(
"agents.coder.post_session_processing", fake_post_session_processing
)
monkeypatch.setattr("agents.coder.run_agent_session", fake_run_agent_session)
monkeypatch.setattr("agents.coder.AUTO_CONTINUE_DELAY_SECONDS", 0)
monkeypatch.setattr("agents.coder.load_subtask_context", lambda *_a, **_k: {})
await run_autonomous_agent(
project_dir=temp_git_repo,
spec_dir=worktree_spec_dir,
model="test-model",
max_iterations=2,
verbose=False,
source_spec_dir=source_spec_dir,
)