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

* 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

apps/backend/agents/coder.py

@@ -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
+        # Get the next subtask to work on (planner sessions shouldn't bind to a subtask)
-        next_subtask = get_next_subtask(spec_dir)
+        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

@@ -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

@@ -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 = {}
+        phase_complete: dict[str, bool] = {}
-        for phase in phases:
+        for i, phase in enumerate(phases):
-            phase_id = phase.get("id") or phase.get("phase")
+            phase_id_value = phase.get("id")
-            subtasks = phase.get("subtasks", [])
+            phase_id_raw = (
-            phase_complete[phase_id] = all(
+                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")
+            phase_id_value = phase.get("id")
-            depends_on = phase.get("depends_on", [])
+            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", []):
+            for subtask in phase.get("subtasks", phase.get("chunks", [])):
-                if subtask.get("status") == "pending":
+                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

@@ -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
+    # Load the full planner prompt from file.
-    prompts_dir = Path(__file__).parent / "prompts"
+    candidate_dirs = [
-    planner_file = prompts_dir / "planner.md"
+        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():
+    if planner_file:
-        prompt = planner_file.read_text()
+        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

@@ -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

@@ -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:
+        try:
-            json.dump(plan, f, indent=2)
+            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

@@ -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"
+        coder_prompt_path = (
-        content = coder_prompt_path.read_text()
+            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"
+        coder_prompt_path = (
-        content = coder_prompt_path.read_text()
+            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"
+        progress_path = (
-        content = progress_path.read_text()
+            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

@@ -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,
+    )