fix: handle empty/greenfield projects in spec creation (#1426)

apps/backend/prompts/spec_writer.md

@@ -24,7 +24,7 @@ You MUST create `spec.md` with ALL required sections (see template below).
 ## PHASE 0: LOAD ALL CONTEXT (MANDATORY)
 
 ```bash
-# Read all input files
+# Read all input files (some may not exist for greenfield/empty projects)
 cat project_index.json
 cat requirements.json
 cat context.json
@@ -35,6 +35,12 @@ Extract from these files:
 - **From requirements.json**: Task description, workflow type, services, acceptance criteria
 - **From context.json**: Files to modify, files to reference, patterns
 
+**IMPORTANT**: If any input file is missing, empty, or shows 0 files, this is likely a **greenfield/new project**. Adapt accordingly:
+- Skip sections that reference existing code (e.g., "Files to Modify", "Patterns to Follow")
+- Instead, focus on files to CREATE and the initial project structure
+- Define the tech stack, dependencies, and setup instructions from scratch
+- Use industry best practices as patterns rather than referencing existing code
+
 ---
 
 ## PHASE 1: ANALYZE CONTEXT

apps/backend/spec/phases/spec_phases.py

@@ -6,18 +6,54 @@
 """
 
 import json
-from typing import TYPE_CHECKING
+from pathlib import Path
 
 from .. import validator, writer
+from ..discovery import get_project_index_stats
 from .models import MAX_RETRIES, PhaseResult
 
-if TYPE_CHECKING:
-    pass
+
+def _is_greenfield_project(spec_dir: Path) -> bool:
+    """Check if the project is empty/greenfield (0 discovered files)."""
+    stats = get_project_index_stats(spec_dir)
+    if not stats:
+        return False  # Can't determine - don't assume greenfield
+    return stats.get("file_count", 0) == 0
+
+
+def _greenfield_context() -> str:
+    """Return additional context for greenfield/empty projects."""
+    return """
+**GREENFIELD PROJECT**: This is an empty or new project with no existing code.
+There are no existing files to reference or modify. You are creating everything from scratch.
+
+Adapt your approach:
+- Do NOT reference existing files, patterns, or code structures
+- Focus on what needs to be CREATED, not modified
+- Define the initial project structure, files, and directories
+- Specify the tech stack, frameworks, and dependencies to install
+- Provide setup instructions for the new project
+- For "Files to Modify" and "Files to Reference" sections, list files to CREATE instead
+- For "Patterns to Follow", describe industry best practices rather than existing code
+"""
 
 
 class SpecPhaseMixin:
     """Mixin for spec writing and critique phase methods."""
 
+    def _check_and_log_greenfield(self) -> bool:
+        """Check if the project is greenfield and log if so.
+
+        Returns:
+            True if the project is greenfield (no existing files).
+        """
+        is_greenfield = _is_greenfield_project(self.spec_dir)
+        if is_greenfield:
+            self.ui.print_status(
+                "Greenfield project detected - adapting spec for new project", "info"
+            )
+        return is_greenfield
+
     async def phase_quick_spec(self) -> PhaseResult:
         """Quick spec for simple tasks - combines context and spec in one step."""
         spec_file = self.spec_dir / "spec.md"
@@ -29,6 +65,8 @@ async def phase_quick_spec(self) -> PhaseResult:
                 "quick_spec", True, [str(spec_file), str(plan_file)], [], 0
             )
 
+        is_greenfield = self._check_and_log_greenfield()
+
         errors = []
         for attempt in range(MAX_RETRIES):
             self.ui.print_status(
@@ -42,7 +80,7 @@ async def phase_quick_spec(self) -> PhaseResult:
 
 This is a SIMPLE task. Create a minimal spec and implementation plan directly.
 No research or extensive analysis needed.
-
+{_greenfield_context() if is_greenfield else ""}
 Create:
 1. A concise spec.md with just the essential sections
 2. A simple implementation_plan.json with 1-2 subtasks
@@ -80,6 +118,9 @@ async def phase_spec_writing(self) -> PhaseResult:
                 "spec.md exists but has issues, regenerating...", "warning"
             )
 
+        is_greenfield = self._check_and_log_greenfield()
+        greenfield_ctx = _greenfield_context() if is_greenfield else ""
+
         errors = []
         for attempt in range(MAX_RETRIES):
             self.ui.print_status(
@@ -88,6 +129,7 @@ async def phase_spec_writing(self) -> PhaseResult:
 
             success, output = await self.run_agent_fn(
                 "spec_writer.md",
+                additional_context=greenfield_ctx,
                 phase_name="spec_writing",
             )
 

apps/backend/spec/pipeline/orchestrator.py

@@ -6,6 +6,7 @@
 """
 
 import json
+import types
 from collections.abc import Callable
 from pathlib import Path
 
@@ -18,6 +19,7 @@
 from task_logger import (
     LogEntryType,
     LogPhase,
+    TaskLogger,
     get_task_logger,
 )
 from ui import (
@@ -238,6 +240,9 @@ async def run(self, interactive: bool = True, auto_approve: bool = False) -> boo
         task_logger.start_phase(LogPhase.PLANNING, "Starting spec creation process")
         TaskEventEmitter.from_spec_dir(self.spec_dir).emit("PLANNING_STARTED")
 
+        # Track whether we've already ended the planning phase (to avoid double-end)
+        self._planning_phase_ended = False
+
         try:
             return await self._run_phases(interactive, auto_approve, task_logger, ui)
         except Exception as e:
@@ -251,22 +256,24 @@ async def run(self, interactive: bool = True, auto_approve: bool = False) -> boo
                 )
             except Exception:
                 pass  # Don't mask the original error
-            try:
-                task_logger.end_phase(
-                    LogPhase.PLANNING,
-                    success=False,
-                    message=f"Spec creation crashed: {e}",
-                )
-            except Exception:
-                pass  # Best effort - don't mask the original error when logging fails
+            if not self._planning_phase_ended:
+                self._planning_phase_ended = True
+                try:
+                    task_logger.end_phase(
+                        LogPhase.PLANNING,
+                        success=False,
+                        message=f"Spec creation crashed: {e}",
+                    )
+                except Exception:
+                    pass  # Best effort - don't mask the original error when logging fails
             raise
 
     async def _run_phases(
         self,
         interactive: bool,
         auto_approve: bool,
-        task_logger,
-        ui,
+        task_logger: TaskLogger,
+        ui: types.ModuleType,
     ) -> bool:
         """Internal method that runs all spec creation phases.
 
@@ -327,6 +334,7 @@ def run_phase(name: str, phase_fn: Callable) -> phases.PhaseResult:
         results.append(result)
         if not result.success:
             print_status("Discovery failed", "error")
+            self._planning_phase_ended = True
             task_logger.end_phase(
                 LogPhase.PLANNING, success=False, message="Discovery failed"
             )
@@ -342,6 +350,7 @@ def run_phase(name: str, phase_fn: Callable) -> phases.PhaseResult:
         results.append(result)
         if not result.success:
             print_status("Requirements gathering failed", "error")
+            self._planning_phase_ended = True
             task_logger.end_phase(
                 LogPhase.PLANNING,
                 success=False,
@@ -380,6 +389,7 @@ def run_phase(name: str, phase_fn: Callable) -> phases.PhaseResult:
         results.append(result)
         if not result.success:
             print_status("Complexity assessment failed", "error")
+            self._planning_phase_ended = True
             task_logger.end_phase(
                 LogPhase.PLANNING, success=False, message="Complexity assessment failed"
             )
@@ -442,6 +452,7 @@ def run_phase(name: str, phase_fn: Callable) -> phases.PhaseResult:
                     f"Phase '{phase_name}' failed: {'; '.join(result.errors)}",
                     LogEntryType.ERROR,
                 )
+                self._planning_phase_ended = True
                 task_logger.end_phase(
                     LogPhase.PLANNING,
                     success=False,
@@ -456,6 +467,7 @@ def run_phase(name: str, phase_fn: Callable) -> phases.PhaseResult:
         self._print_completion_summary(results, phases_executed)
 
         # End planning phase successfully
+        self._planning_phase_ended = True
         task_logger.end_phase(
             LogPhase.PLANNING, success=True, message="Spec creation complete"
         )
@@ -729,9 +741,8 @@ def _run_review_checkpoint(self, auto_approve: bool) -> bool:
                 print_status("Build will not proceed without approval.", "warning")
                 return False
 
-        except SystemExit as e:
-            if e.code != 0:
-                return False
+        except SystemExit:
+            # Review checkpoint may call sys.exit(); treat any exit as unapproved
             return False
         except KeyboardInterrupt:
             print()