diff --git a/codeframe/core/agent.py b/codeframe/core/agent.py index 80e820ad..f11372f7 100644 --- a/codeframe/core/agent.py +++ b/codeframe/core/agent.py @@ -254,6 +254,7 @@ def to_dict(self) -> dict: MAX_CONSECUTIVE_FAILURES = 3 MAX_STEP_RETRIES = 2 MAX_SELF_CORRECTION_ATTEMPTS = 2 +MAX_CONSECUTIVE_VERIFICATION_FAILURES = 3 # TRUE requirements ambiguity - create blocker immediately # These are situations where the agent genuinely cannot proceed without human input @@ -567,6 +568,7 @@ def _execute_plan(self) -> None: ) consecutive_failures = 0 + consecutive_verification_failures = 0 self._debug_log( f"Starting plan execution with {len(self.state.plan.steps)} steps", @@ -620,24 +622,47 @@ def _execute_plan(self) -> None: # Run incremental verification for file changes if step.type in {StepType.FILE_CREATE, StepType.FILE_EDIT}: gate_result = self._run_incremental_verification() - if gate_result and not gate_result.passed: + if gate_result and gate_result.passed: + consecutive_verification_failures = 0 + elif gate_result and not gate_result.passed: # Try to fix lint issues automatically (works for style, not syntax) if not self._try_auto_fix(gate_result): # Auto-fix failed - need to self-correct the code + # Extract detailed error info from gate result + failed_checks = [ + c for c in gate_result.checks + if c.status != GateStatus.PASSED + ] + failed_check_names = [c.name for c in failed_checks] + + # Build detailed error string with actual output + error_details = [] + for check in failed_checks: + if check.output: + error_details.append( + f"[{check.name}] {check.output[:500]}" + ) + error_detail_str = ( + "\n".join(error_details) + if error_details + else "No details available" + ) + self._emit_event("verification_failed", { "step": step.index, - "error": "Code verification failed after file change", + "error": f"Verification failed: {failed_check_names}", + "gates": failed_check_names, + "error_count": len(failed_checks), + "error_details": error_detail_str[:1000], }) - # Trigger self-correction for the verification failure - failed_checks = [ - c.name for c in gate_result.checks - if c.status != GateStatus.PASSED - ] failed_result = StepResult( step=step, status=ExecutionStatus.FAILED, - error=f"Verification failed: {failed_checks}", + error=( + f"Verification failed: {failed_check_names}" + f"\n{error_detail_str}" + ), ) # Try self-correction to fix the code @@ -668,15 +693,41 @@ def _execute_plan(self) -> None: current_result = corrected_result - if not self_correction_succeeded: + if self_correction_succeeded: + consecutive_verification_failures = 0 + else: # Couldn't fix the verification error + consecutive_verification_failures += 1 consecutive_failures += 1 + if consecutive_verification_failures >= MAX_CONSECUTIVE_VERIFICATION_FAILURES: + self._debug_log( + f"ABORTING: Too many consecutive verification failures ({consecutive_verification_failures})", + level="ERROR", + always=True, + ) + self._emit_event("execution_aborted", { + "reason": f"Too many consecutive verification failures ({consecutive_verification_failures})", + "step": step.index, + }) + # Force blocker creation — bypass LLM classification + # since this is a definitive abort, not a tactical decision + error_msg = current_result.error if current_result else "Repeated verification failures" + blocker = blockers.create( + workspace=self.workspace, + question=f"Agent aborted: {consecutive_verification_failures} consecutive verification failures at step {step.index} ({step.description}). Last error: {error_msg[:500]}", + task_id=self.state.task_id, + ) + self.state.status = AgentStatus.BLOCKED + self.state.blocker = BlockerInfo( + reason="Too many consecutive verification failures", + question=blocker.question, + context=f"Step {step.index}: {step.description}", + ) + return if consecutive_failures >= MAX_CONSECUTIVE_FAILURES: - # Create blocker asking for help with the code self._create_blocker_from_failure(step, current_result) return - # Otherwise, we continue to next step with broken file - # (not ideal, but prevents infinite loop) + # Otherwise, continue to next step with broken file self.state.current_step += 1 @@ -821,7 +872,7 @@ def _run_incremental_verification(self) -> Optional[GateResult]: result = run_gates( self.workspace, gates=["ruff"], - verbose=False, + verbose=True, ) self.state.gate_results.append(result) return result diff --git a/codeframe/core/executor.py b/codeframe/core/executor.py index 70eb302b..da18347d 100644 --- a/codeframe/core/executor.py +++ b/codeframe/core/executor.py @@ -287,15 +287,53 @@ def _execute_file_create( step: PlanStep, context: TaskContext, ) -> StepResult: - """Create a new file with generated content.""" + """Create a new file with generated content. + + If the file already exists, falls back gracefully: + - Identical content: returns SUCCESS (no-op) + - Different content: falls back to edit behavior + """ file_path = self.repo_path / step.target - # Check if file already exists + # Check if file already exists -- fall back gracefully if file_path.exists(): + existing_content = file_path.read_text(encoding="utf-8") + + # Generate the content we would write + new_content = self._generate_file_content(step, context) + + # If content is identical, no-op success + if existing_content.strip() == new_content.strip(): + return StepResult( + step=step, + status=ExecutionStatus.SUCCESS, + output=f"File already exists with correct content: {step.target}", + ) + + # Content differs -- fall back to edit behavior + if self.dry_run: + return StepResult( + step=step, + status=ExecutionStatus.SUCCESS, + output=f"[DRY RUN] Would edit existing file: {step.target}", + ) + + file_path.write_text(new_content, encoding="utf-8") + + change = FileChange( + path=step.target, + operation="edit", + original_content=existing_content, + new_content=new_content, + timestamp=datetime.now(timezone.utc), + ) + self.changes.append(change) + return StepResult( step=step, - status=ExecutionStatus.FAILED, - error=f"File already exists: {step.target}", + status=ExecutionStatus.SUCCESS, + output=f"Updated existing file: {step.target}", + file_changes=[change], ) # Generate file content using LLM diff --git a/codeframe/core/gates.py b/codeframe/core/gates.py index d7f12f5d..49a28cf4 100644 --- a/codeframe/core/gates.py +++ b/codeframe/core/gates.py @@ -6,13 +6,14 @@ This module is headless - no FastAPI or HTTP dependencies. """ +import re import subprocess import shutil from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from pathlib import Path -from typing import Optional +from typing import Any, Optional from codeframe.core.workspace import Workspace from codeframe.core import events @@ -23,6 +24,34 @@ def _utc_now() -> datetime: return datetime.now(timezone.utc) +_RUFF_ERROR_PATTERN = re.compile(r'^(.+?):(\d+):(\d+): ([A-Z]+\d+) (.+)$') + + +def _parse_ruff_errors(output: str) -> list[dict[str, Any]]: + """Parse ruff output into structured error dicts. + + Parses lines matching the pattern: path/file.py:10:5: E501 Line too long + + Args: + output: Raw ruff stdout/stderr output. + + Returns: + List of dicts with keys: file, line, col, code, message. + """ + errors = [] + for line in output.splitlines(): + match = _RUFF_ERROR_PATTERN.match(line.strip()) + if match: + errors.append({ + "file": match.group(1), + "line": int(match.group(2)), + "col": int(match.group(3)), + "code": match.group(4), + "message": match.group(5), + }) + return errors + + class GateStatus(str, Enum): """Status of a gate check.""" @@ -42,6 +71,7 @@ class GateCheck: exit_code: Process exit code (if run) output: Captured stdout/stderr duration_ms: How long the check took + detailed_errors: Structured error list parsed from tool output """ name: str @@ -49,6 +79,7 @@ class GateCheck: exit_code: Optional[int] = None output: str = "" duration_ms: int = 0 + detailed_errors: Optional[list[dict[str, Any]]] = None @dataclass @@ -87,6 +118,38 @@ def summary(self) -> str: return ", ".join(parts) if parts else "no checks run" + def get_error_summary(self) -> str: + """Format all errors into a readable multi-line string. + + Returns: + Newline-separated string of all structured errors from failed checks, + or empty string if no errors. + """ + lines = [] + for check in self.checks: + if check.detailed_errors: + for err in check.detailed_errors: + lines.append( + f"{err['file']}:{err['line']}:{err['col']}: " + f"{err['code']} {err['message']}" + ) + return "\n".join(lines) + + def get_errors_by_file(self) -> dict[str, list[str]]: + """Group error messages by file path. + + Returns: + Dict mapping file paths to lists of formatted error strings. + """ + by_file: dict[str, list[str]] = {} + for check in self.checks: + if check.detailed_errors: + for err in check.detailed_errors: + file_path = err["file"] + msg = f"{err['code']} {err['message']} (line {err['line']})" + by_file.setdefault(file_path, []).append(msg) + return by_file + def run( workspace: Workspace, @@ -317,7 +380,7 @@ def _run_ruff(repo_path: Path, verbose: bool = False) -> GateCheck: if result.stderr: output += "\n" + result.stderr - return GateCheck( + check = GateCheck( name="ruff", status=GateStatus.PASSED if result.returncode == 0 else GateStatus.FAILED, exit_code=result.returncode, @@ -325,6 +388,12 @@ def _run_ruff(repo_path: Path, verbose: bool = False) -> GateCheck: duration_ms=duration_ms, ) + # Parse detailed errors for failed checks + if check.status == GateStatus.FAILED: + check.detailed_errors = _parse_ruff_errors(output) + + return check + except subprocess.TimeoutExpired: return GateCheck( name="ruff", diff --git a/codeframe/core/planner.py b/codeframe/core/planner.py index 9706e7e5..64cc261a 100644 --- a/codeframe/core/planner.py +++ b/codeframe/core/planner.py @@ -228,7 +228,7 @@ def to_markdown(self) -> str: 4. Be specific about what files to modify and what changes to make 5. Consider edge cases and potential issues 6. Keep the plan achievable - don't over-engineer -7. If a file exists that you need to modify, use file_edit not file_create +7. IMPORTANT: Before choosing file_create, verify the file does not already exist in the repository structure. If it exists, use file_edit instead. Never use file_create for files listed in the Repository Structure section. 8. Run tests after implementation to verify correctness Return ONLY the JSON object, no additional text.""" @@ -334,6 +334,17 @@ def _build_prompt(self, context: TaskContext) -> str: sections.append(f" - {f.path}") sections.append("") + # Existing files warning for planner + if context.file_tree: + sections.append("## Existing Files Warning") + sections.append( + "The following files already exist in the workspace. " + "Use file_edit (NOT file_create) for these:" + ) + for f_info in context.file_tree[:50]: + sections.append(f" - {f_info.path}") + sections.append("") + # Loaded file contents if context.loaded_files: sections.append("## Relevant Source Files") diff --git a/tests/core/test_agent.py b/tests/core/test_agent.py index c7a8bf27..1c292eab 100644 --- a/tests/core/test_agent.py +++ b/tests/core/test_agent.py @@ -2,6 +2,7 @@ import pytest import json +import inspect from pathlib import Path from datetime import datetime, timezone from unittest.mock import MagicMock, patch @@ -373,3 +374,60 @@ def test_verifying_to_completed(self): state = AgentState(status=AgentStatus.VERIFYING) state.status = AgentStatus.COMPLETED assert state.status == AgentStatus.COMPLETED + + +class TestVerificationRecovery: + """Tests for verification recovery and early abort.""" + + def test_max_consecutive_verification_failures_constant_exists(self): + """New constant for verification-specific failure tracking exists.""" + from codeframe.core.agent import MAX_CONSECUTIVE_VERIFICATION_FAILURES + assert MAX_CONSECUTIVE_VERIFICATION_FAILURES == 3 + + def test_incremental_verification_tracks_failures_separately(self): + """Verification failures tracked separately from step execution failures.""" + # The _execute_plan method must have a separate counter for verification failures + # distinct from the existing consecutive_failures counter for step execution. + # We verify by inspecting the source for the new variable name. + source = inspect.getsource(Agent._execute_plan) + assert "consecutive_verification_failures" in source + + def test_verification_failed_event_includes_details(self): + """verification_failed events include gate name, error count, and error details.""" + # The enhanced verification_failed event should include structured gate info + # rather than just a generic error string. + source = inspect.getsource(Agent._execute_plan) + assert '"gates"' in source or "'gates'" in source + assert '"error_count"' in source or "'error_count'" in source + assert '"error_details"' in source or "'error_details'" in source + + def test_incremental_verification_uses_verbose(self): + """_run_incremental_verification captures full error details via verbose=True.""" + source = inspect.getsource(Agent._run_incremental_verification) + assert "verbose=True" in source + + def test_verification_counter_resets_on_clean_pass(self): + """consecutive_verification_failures resets when incremental verification passes.""" + # The code must reset the counter on a clean pass, not just on self-correction success. + # This prevents premature abort after: fail → pass → fail → fail (counter would be 3 + # without the reset, but should be 2). + source = inspect.getsource(Agent._execute_plan) + # Find the clean-pass reset: gate_result.passed → reset counter + assert "gate_result.passed" in source + # The reset must appear in the passed branch, before the failure branch + passed_idx = source.index("gate_result.passed") + reset_after_pass = source.index( + "consecutive_verification_failures = 0", passed_idx + ) + assert reset_after_pass > passed_idx + + def test_abort_forces_blocker_creation(self): + """Abort path creates blocker directly, bypassing LLM classification.""" + # The abort path must call blockers.create directly rather than + # _create_blocker_from_failure which can silently return for + # RESOLVE_AUTONOMOUSLY or TECHNICAL_FIX classifications. + source = inspect.getsource(Agent._execute_plan) + abort_idx = source.index("execution_aborted") + # After the abort event, blockers.create must be called directly + blocker_create_idx = source.index("blockers.create", abort_idx) + assert blocker_create_idx > abort_idx diff --git a/tests/core/test_executor.py b/tests/core/test_executor.py index 3ca628ef..5c7ff11e 100644 --- a/tests/core/test_executor.py +++ b/tests/core/test_executor.py @@ -156,16 +156,24 @@ def test_file_create(self, tmp_path, mock_provider, sample_context): assert (tmp_path / "main.py").exists() assert len(result.file_changes) == 1 - def test_file_create_fails_if_exists(self, tmp_path, mock_provider, sample_context): - """File create fails if file already exists.""" + def test_file_create_falls_back_when_exists(self, tmp_path, mock_provider, sample_context): + """File create falls back to edit when file exists with different content.""" (tmp_path / "existing.py").write_text("# existing") + mock_provider.set_response_handler( + lambda msgs: LLMResponse(content="# Updated content\nprint('updated')") + ) executor = Executor(mock_provider, tmp_path) step = PlanStep(1, StepType.FILE_CREATE, "Create", "existing.py") result = executor.execute_step(step, sample_context) - assert result.status == ExecutionStatus.FAILED - assert "already exists" in result.error + assert result.status == ExecutionStatus.SUCCESS + assert "existing.py" in result.output + content = (tmp_path / "existing.py").read_text() + assert "Updated content" in content + assert len(result.file_changes) == 1 + assert result.file_changes[0].original_content == "# existing" + assert result.file_changes[0].operation == "edit" def test_file_create_nested(self, tmp_path, mock_provider, sample_context): """Can create file in nested directory.""" @@ -224,6 +232,87 @@ def test_file_delete_already_gone(self, tmp_path, mock_provider, sample_context) assert result.status == ExecutionStatus.SUCCESS +class TestFileCreateConflictHandling: + """Tests for file_create fallback when file already exists.""" + + @pytest.fixture + def mock_provider(self): + provider = MockProvider() + provider.set_response_handler( + lambda msgs: LLMResponse(content="# Generated code\nprint('hello')") + ) + return provider + + @pytest.fixture + def sample_context(self): + task = Task( + id="t1", workspace_id="w1", prd_id=None, + title="Test task", description="Test", + status=TaskStatus.IN_PROGRESS, + priority=0, + created_at=_utc_now(), + updated_at=_utc_now(), + ) + return TaskContext(task=task) + + def test_file_create_falls_back_to_edit_when_content_differs( + self, tmp_path, mock_provider, sample_context + ): + """file_create falls back to file_edit when file exists with different content.""" + (tmp_path / "existing.py").write_text("# old content") + mock_provider.set_response_handler( + lambda msgs: LLMResponse(content="# Updated content\nprint('updated')") + ) + executor = Executor(mock_provider, tmp_path) + step = PlanStep(1, StepType.FILE_CREATE, "Create", "existing.py") + + result = executor.execute_step(step, sample_context) + + assert result.status == ExecutionStatus.SUCCESS + assert "existing.py" in result.output + # File should have new content via edit fallback + content = (tmp_path / "existing.py").read_text() + assert "Updated content" in content + # Should have a file change recorded with original content + assert len(result.file_changes) == 1 + assert result.file_changes[0].original_content == "# old content" + assert result.file_changes[0].operation == "edit" + + def test_file_create_succeeds_when_identical_content( + self, tmp_path, mock_provider, sample_context + ): + """file_create returns SUCCESS when file exists with identical content.""" + existing_content = "# Generated code\nprint('hello')" + (tmp_path / "same.py").write_text(existing_content) + mock_provider.set_response_handler( + lambda msgs: LLMResponse(content=existing_content) + ) + executor = Executor(mock_provider, tmp_path) + step = PlanStep(1, StepType.FILE_CREATE, "Create", "same.py") + + result = executor.execute_step(step, sample_context) + + assert result.status == ExecutionStatus.SUCCESS + assert "already exists" in result.output.lower() + # Content should remain unchanged + assert (tmp_path / "same.py").read_text() == existing_content + + def test_file_create_dry_run_with_existing_file( + self, tmp_path, mock_provider, sample_context + ): + """file_create dry run still works when file exists.""" + (tmp_path / "existing.py").write_text("# old") + executor = Executor(mock_provider, tmp_path, dry_run=True) + step = PlanStep(1, StepType.FILE_CREATE, "Create", "existing.py") + + result = executor.execute_step(step, sample_context) + + assert result.status == ExecutionStatus.SUCCESS + assert "DRY RUN" in result.output + # Original content untouched + assert (tmp_path / "existing.py").read_text() == "# old" + + class TestExecutorDryRun: """Tests for dry run mode.""" diff --git a/tests/core/test_gates_observability.py b/tests/core/test_gates_observability.py new file mode 100644 index 00000000..182ab40d --- /dev/null +++ b/tests/core/test_gates_observability.py @@ -0,0 +1,163 @@ +"""Tests for gate observability enhancements.""" + +import pytest +from codeframe.core.gates import GateCheck, GateResult, GateStatus, _parse_ruff_errors + + +class TestParseRuffErrors: + """Tests for ruff output parsing into structured errors.""" + + def test_parse_single_error(self): + """Parses a single ruff error line.""" + output = "src/main.py:10:5: E501 Line too long (120 > 79)" + errors = _parse_ruff_errors(output) + assert len(errors) == 1 + assert errors[0]["file"] == "src/main.py" + assert errors[0]["line"] == 10 + assert errors[0]["col"] == 5 + assert errors[0]["code"] == "E501" + assert "Line too long" in errors[0]["message"] + + def test_parse_multiple_errors(self): + """Parses multiple ruff error lines.""" + output = ( + "src/main.py:10:5: E501 Line too long\n" + "src/utils.py:25:1: F401 `os` imported but unused\n" + "src/main.py:3:1: I001 Import block is un-sorted\n" + ) + errors = _parse_ruff_errors(output) + assert len(errors) == 3 + assert errors[1]["file"] == "src/utils.py" + assert errors[1]["code"] == "F401" + + def test_parse_empty_output(self): + """Returns empty list for empty output.""" + assert _parse_ruff_errors("") == [] + assert _parse_ruff_errors("All checks passed!") == [] + + def test_parse_mixed_output(self): + """Handles output with non-error lines mixed in.""" + output = ( + "Found 2 errors.\n" + "src/main.py:10:5: E501 Line too long\n" + "[*] 1 fixable with `ruff check --fix`.\n" + "src/utils.py:1:1: F401 unused import\n" + ) + errors = _parse_ruff_errors(output) + assert len(errors) == 2 + + def test_parse_multi_letter_rule_codes(self): + """Parses ruff codes with multi-letter prefixes like ANN, PLR, SIM.""" + output = ( + "src/api.py:5:1: ANN401 Dynamically typed expressions not allowed\n" + "src/utils.py:12:5: PLR2004 Magic value used in comparison\n" + "src/main.py:8:1: SIM118 Use `key in dict` instead of `key in dict.keys()`\n" + "src/config.py:3:1: UP035 `typing.Dict` is deprecated, use `dict` instead\n" + ) + errors = _parse_ruff_errors(output) + assert len(errors) == 4 + assert errors[0]["code"] == "ANN401" + assert errors[1]["code"] == "PLR2004" + assert errors[2]["code"] == "SIM118" + assert errors[3]["code"] == "UP035" + + +class TestGateCheckDetailedErrors: + """Tests for detailed_errors field on GateCheck.""" + + def test_gatecheck_has_detailed_errors_field(self): + """GateCheck has optional detailed_errors field.""" + check = GateCheck( + name="ruff", + status=GateStatus.FAILED, + output="src/main.py:1:1: F401 unused", + detailed_errors=[ + { + "file": "src/main.py", + "line": 1, + "col": 1, + "code": "F401", + "message": "unused", + } + ], + ) + assert check.detailed_errors is not None + assert len(check.detailed_errors) == 1 + + def test_gatecheck_detailed_errors_default_none(self): + """detailed_errors defaults to None.""" + check = GateCheck(name="ruff", status=GateStatus.PASSED) + assert check.detailed_errors is None + + +class TestGateResultErrorMethods: + """Tests for GateResult error summary and grouping methods.""" + + @pytest.fixture + def failed_gate_result(self): + """GateResult with failed ruff check and detailed errors.""" + check = GateCheck( + name="ruff", + status=GateStatus.FAILED, + exit_code=1, + output=( + "src/main.py:10:5: E501 Line too long\n" + "src/utils.py:1:1: F401 unused import" + ), + detailed_errors=[ + { + "file": "src/main.py", + "line": 10, + "col": 5, + "code": "E501", + "message": "Line too long", + }, + { + "file": "src/utils.py", + "line": 1, + "col": 1, + "code": "F401", + "message": "unused import", + }, + ], + ) + return GateResult( + passed=False, + checks=[check], + ) + + def test_get_error_summary(self, failed_gate_result): + """get_error_summary returns formatted string of all errors.""" + summary = failed_gate_result.get_error_summary() + assert "E501" in summary + assert "F401" in summary + assert "src/main.py" in summary + + def test_get_errors_by_file(self, failed_gate_result): + """get_errors_by_file groups errors by file path.""" + by_file = failed_gate_result.get_errors_by_file() + assert "src/main.py" in by_file + assert "src/utils.py" in by_file + assert len(by_file["src/main.py"]) == 1 + assert "E501" in by_file["src/main.py"][0] + + def test_get_error_summary_no_errors(self): + """get_error_summary handles no errors gracefully.""" + result = GateResult( + passed=True, + checks=[ + GateCheck(name="ruff", status=GateStatus.PASSED), + ], + ) + summary = result.get_error_summary() + assert summary == "" or "no errors" in summary.lower() + + def test_get_errors_by_file_no_errors(self): + """get_errors_by_file returns empty dict when no errors.""" + result = GateResult( + passed=True, + checks=[ + GateCheck(name="ruff", status=GateStatus.PASSED), + ], + ) + assert result.get_errors_by_file() == {} diff --git a/tests/core/test_planner.py b/tests/core/test_planner.py index 0eac968f..ece41cb0 100644 --- a/tests/core/test_planner.py +++ b/tests/core/test_planner.py @@ -12,7 +12,7 @@ Complexity, PLANNING_SYSTEM_PROMPT, ) -from codeframe.core.context import TaskContext +from codeframe.core.context import TaskContext, FileInfo from codeframe.core.tasks import Task, TaskStatus from codeframe.core.prd import PrdRecord from codeframe.adapters.llm import MockProvider, LLMResponse, Purpose @@ -374,3 +374,45 @@ def test_system_prompt_used(self, mock_provider): planner.create_plan(context) assert mock_provider.last_call["system"] == PLANNING_SYSTEM_PROMPT + + +class TestPlannerExistingFilesContext: + """Tests for workspace-aware planning with existing files.""" + + def test_build_prompt_includes_existing_files_section(self): + """Planner prompt includes existing files from file_tree.""" + provider = MockProvider() + provider.set_response_handler( + lambda msgs: LLMResponse( + content=json.dumps({ + "summary": "test", + "steps": [], + "files_to_create": [], + "files_to_modify": [], + "estimated_complexity": "low", + "considerations": [], + }) + ) + ) + planner = Planner(provider) + + task = Task( + id="t1", workspace_id="w1", prd_id=None, + title="Add feature", description="Add a new feature", + status=TaskStatus.IN_PROGRESS, + priority=0, + created_at=_utc_now(), + updated_at=_utc_now(), + ) + context = TaskContext( + task=task, + file_tree=[ + FileInfo(path="src/main.py", size_bytes=100, extension=".py"), + FileInfo(path="src/utils.py", size_bytes=200, extension=".py"), + FileInfo(path="tests/test_main.py", size_bytes=150, extension=".py"), + ], + ) + + prompt = planner._build_prompt(context) + assert "Existing Files" in prompt or "existing" in prompt.lower() + assert "file_edit" in prompt