cinnabarflower/mnemo_cards
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

subtasks

Browse source
This commit is contained in:
Dmitry 2025-11-21 02:56:30 +03:00
parent f9484d962d
commit 83e45287cc
4 changed files with 308 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.forgejo/workflows/agent-development.yml
View file

@ -44,7 +44,7 @@ jobs:
echo "FLUTTER_ROOT=/opt/flutter" >> "$GITHUB_ENV"
echo "/opt/flutter/bin" >> "$GITHUB_PATH"
/opt/flutter/bin/flutter --version
- name: Install Cursor CLI
run: |
# bash should already be installed in previous step

84
tools/agent/agent_orchestrator.py
View file

@ -7,11 +7,12 @@ import subprocess
import time
from pathlib import Path
from datetime import datetime, timezone
from typing import Optional
from typing import Optional, List
from config import AgentConfig, get_test_command, get_lint_command
from task_manager import TaskManager, GlobalLock, Task
from cursor_cli_wrapper import CursorCLI, CursorResultStatus
from planning_agent import PlanningAgent
class AgentOrchestrator:
@ -107,13 +108,36 @@ class AgentOrchestrator:
if retry_count >= self.config.max_retries:
print(f"\n❌ Task {next_task.id} failed after {retry_count} retries")
self.task_manager.mark_task_failed(
next_task.id,
f"Failed after {retry_count} retries"
# Try to break down the task into subtasks
print(f"\n🔀 Attempting to break down task {next_task.id} into subtasks...")
planning_agent = PlanningAgent(self.config)
failure_reason = f"Failed after {retry_count} retry attempts. Task may be too complex."
breakdown_success = planning_agent.break_down_task(
next_task,
failure_reason
)
# Create GitHub issue
self._create_issue_for_failed_task(next_task)
if breakdown_success:
print(f"\n✅ Task {next_task.id} successfully broken down into subtasks")
print(" Subtasks have been added to task_list.json")
# Mark original task as failed
self.task_manager.mark_task_failed(
next_task.id,
f"Failed after {retry_count} retries. Broken down into subtasks."
)
else:
print(f"\n⚠️ Failed to break down task {next_task.id}")
print(" Marking task as failed without subtasks")
self.task_manager.mark_task_failed(
next_task.id,
f"Failed after {retry_count} retries"
)
# Create GitHub issue
self._create_issue_for_failed_task(next_task)
break
else:
print(f"\n⚠️ Task {next_task.id} failed, retrying ({retry_count}/{self.config.max_retries})")
@ -189,8 +213,30 @@ class AgentOrchestrator:
print(f"\n📊 Agent completed:")
print(f" - Files created: {len(result.files_created)}")
print(f" - Files modified: {len(result.files_modified)}")
print(f" - Files read: {len(result.files_read)}")
print(f" - Tool calls: {result.tool_calls}")
# Check if any files were actually changed
if len(result.files_created) == 0 and len(result.files_modified) == 0:
print("\n⚠️ WARNING: Agent completed but no files were created or modified!")
print(f" Agent read {len(result.files_read)} file(s) but made no changes.")
print(" This might indicate:")
print(" - Task was already completed")
print(" - Agent only analyzed files without making changes")
print(" - Agent encountered an issue but didn't report it")
# Check git status to see if there are any uncommitted changes
git_changes = self._check_git_changes()
if git_changes:
print(f"\n However, git shows {len(git_changes)} uncommitted change(s):")
for change in git_changes[:5]: # Show first 5
print(f" - {change}")
if len(git_changes) > 5:
print(f" ... and {len(git_changes) - 5} more")
else:
print("\n Git confirms: no changes detected in working directory.")
print(" Task may need to be retried or task description clarified.")
# Pull latest changes before committing
print("\n🔄 Pulling latest changes...")
self._git_pull()
@ -331,6 +377,32 @@ Changes:
print(f"❌ Test output:\n{result.stdout}")
return False
def _check_git_changes(self) -> List[str]:
"""Check for uncommitted changes in git."""
# Check status for modified, added, deleted files
result = subprocess.run(
["git", "status", "--porcelain"],
cwd=self.config.project_root,
capture_output=True,
text=True
)
if result.returncode != 0:
return []
changes = []
for line in result.stdout.strip().split('\n'):
if line.strip():
# Format: "XY filename" where X is staging area, Y is working tree
# We care about files that are modified (M), added (A), deleted (D), or renamed (R)
status = line[:2]
filename = line[3:].strip()
if any(c in status for c in ['M', 'A', 'D', 'R']):
changes.append(filename)
return changes
def _git_pull(self) -> bool:
"""Pull latest changes from remote."""
result = subprocess.run(

14
tools/agent/cursor_cli_wrapper.py
View file

@ -329,17 +329,21 @@ class CursorCLI:
if "success" in result:
success = result["success"]
path = tool_call["writeToolCall"].get("args", {}).get("path", "unknown")
lines = success.get("linesCreated", 0)
lines_created = success.get("linesCreated", 0)
lines_modified = success.get("linesModified", 0)
if lines > 0:
if lines_created > 0:
files_created.append(path)
if self.verbose:
print(f" ✅ Created {lines} lines")
else:
print(f" ✅ Created {lines_created} lines")
elif lines_modified > 0:
files_modified.append(path)
if self.verbose:
lines_modified = success.get("linesModified", 0)
print(f" ✅ Modified {lines_modified} lines")
else:
# File was written but not actually changed
if self.verbose:
print(f" ⚠️ Written but no changes detected (file unchanged)")
elif "readToolCall" in tool_call:
result = tool_call["readToolCall"].get("result", {})

220
tools/agent/planning_agent.py
View file

@ -9,6 +9,7 @@ from typing import Dict, Any, List
from config import AgentConfig
from cursor_cli_wrapper import CursorCLI, CursorResultStatus
from task_manager import Task
class PlanningAgent:
@ -301,6 +302,225 @@ Component: {self.config.component}
print("✅ Pushed to remote")
else:
print("⚠️ No changes to commit (task list unchanged)")
def break_down_task(self, task: Task, failure_reason: str = "") -> bool:
"""
Break down a failed task into smaller subtasks.
Args:
task: The task that failed
failure_reason: Reason why the task failed
Returns:
True if subtasks were successfully created, False otherwise
"""
print(f"\n🔀 Breaking down task {task.id} into subtasks...")
print(f" Task: {task.title}")
print(f" Reason: {failure_reason}")
try:
# Build prompt for breaking down the task
prompt = self._build_breakdown_prompt(task, failure_reason)
# Run cursor agent to break down the task
print("\n🤖 Running Cursor Agent to break down task...")
result = self.cursor_cli.run_agent(
task_description=prompt,
force=True,
max_iterations=3
)
if result.status != CursorResultStatus.SUCCESS:
print(f"❌ Failed to break down task: {result.error}")
return False
# Load current task list before breakdown
if not self.config.task_list_path.exists():
print(f"❌ Task list not found: {self.config.task_list_path}")
return False
# Get original task list for comparison
original_task_list = self._load_and_validate_task_list()
if not original_task_list:
print("❌ Invalid task list")
return False
original_tasks = original_task_list.get("tasks", [])
original_task_ids = {t.get("id") for t in original_tasks}
original_count = len(original_tasks)
# Reload to get updated task list after agent execution
task_list = self._load_and_validate_task_list()
if not task_list:
print("❌ Failed to reload task list after breakdown")
return False
updated_tasks = task_list.get("tasks", [])
new_count = len(updated_tasks)
# Find subtasks - they should have IDs containing the original task ID
# or be new tasks added after the breakdown
# Look for tasks with IDs like "TASK-001-SUB-1" or similar patterns
subtasks = [
t for t in updated_tasks
if (task.id in t.get("id", "") and t.get("id") != task.id) or
(t.get("id", "").startswith(task.id + "-") or
t.get("id", "").startswith(task.id + "_"))
]
# If no subtasks found by ID pattern, check if task count increased
if len(subtasks) == 0:
if new_count > original_count:
# New tasks were added, assume they are subtasks
# Get tasks that weren't in the original list
subtasks = [t for t in updated_tasks if t.get("id") not in original_task_ids]
if len(subtasks) == 0:
print("⚠️ No subtasks found. Agent may not have created them.")
print(" Checking if task was updated instead...")
# Check if the original task was modified
original_task = next((t for t in updated_tasks if t.get("id") == task.id), None)
if original_task:
print(f" Task {task.id} still exists in task list")
return False
else:
print(f" Task {task.id} was removed but no subtasks found")
return False
print(f"\n✅ Task broken down into {len(subtasks)} subtask(s):")
for subtask in subtasks:
print(f" - {subtask['id']}: {subtask['title']} ({subtask.get('estimated_hours', 0)}h)")
# Mark original task as failed (if it still exists)
original_task = next((t for t in updated_tasks if t.get("id") == task.id), None)
if original_task:
original_task["status"] = "failed"
original_task["description"] += f"\n\n[FAILED] {failure_reason}\n[REPLACED BY] {', '.join([s['id'] for s in subtasks])}"
self._save_task_list(task_list)
# Commit the changes
self._commit_task_list()
return True
except Exception as e:
print(f"\n❌ Error breaking down task: {e}")
import traceback
traceback.print_exc()
return False
def _build_breakdown_prompt(self, task: Task, failure_reason: str) -> str:
"""Build prompt for breaking down a task into subtasks."""
# Load planning prompt template
prompt_file = self.config.prompts_dir / "planning_prompt.md"
if prompt_file.exists():
with open(prompt_file, 'r') as f:
template = f.read()
else:
template = "You are an AI planning agent. Break down complex tasks into smaller subtasks.\n\n"
# Load current task list for context
context = ""
if self.config.task_list_path.exists():
with open(self.config.task_list_path, 'r') as f:
current_tasks = json.load(f)
context += f"### Current task_list.json:\n```json\n{json.dumps(current_tasks, indent=2)}\n```\n\n"
prompt = f"""{template}
## TASK BREAKDOWN REQUEST
A task has failed after multiple attempts and needs to be broken down into smaller, more manageable subtasks.
### Failed Task:
```json
{json.dumps(task.to_dict(), indent=2)}
```
### Failure Reason:
{failure_reason if failure_reason else "Task was too complex and could not be completed after multiple retry attempts."}
### Current Context:
{context}
### Your Task:
1. **Analyze the failed task** - Understand why it might have been too complex
2. **Break it down** into 3-5 smaller, focused subtasks
3. **Each subtask should be:**
- 1-4 hours of work (preferably 2-3 hours)
- Independently testable and completable
- Have clear acceptance criteria
- Specify files to modify
4. **Set dependencies:**
- Subtasks should NOT depend on the original task ID (since it failed)
- Set dependencies between subtasks if there's a logical order
- If subtasks are independent, they can have empty dependencies
5. **Update task_list.json:**
- Add the new subtasks to the tasks array
- You can either:
a) Mark the original task as "failed" and add subtasks
b) Replace the original task with subtasks
- Keep all other tasks unchanged
### Subtask Structure:
Each subtask should follow this format:
```json
{{
"id": "{task.id}-SUB-1",
"title": "Clear, focused title for this subtask",
"priority": "{task.priority}",
"status": "pending",
"estimated_hours": 2.5,
"description": "Detailed description of what this subtask accomplishes",
"acceptance_criteria": [
"Specific, testable criterion 1",
"Specific, testable criterion 2",
"Comprehensive test coverage"
],
"dependencies": [],
"files_to_modify": ["path/to/specific/file.dart"],
"component": "{task.component}"
}}
```
### Guidelines:
- **Keep subtasks small**: 1-4 hours each, preferably 2-3 hours
- **Make them independent**: Each subtask should be completable on its own
- **Clear dependencies**: If subtasks depend on each other, set dependencies
- **Preserve priority**: Use the same priority as the original task
- **Specific files**: Each subtask should modify specific files, not all files from original task
- **Testable**: Each subtask should have clear acceptance criteria
### Example Breakdown:
If original task was: "Implement user authentication system (8 hours)"
Good breakdown:
- SUB-1: Create authentication models and DTOs (2h)
- SUB-2: Implement authentication service layer (2h)
- SUB-3: Create authentication API endpoints (2h)
- SUB-4: Add authentication UI components (2h)
Now break down task `{task.id}` into subtasks and update the task_list.json file!
"""
return prompt
def _save_task_list(self, task_list: Dict[str, Any]) -> None:
"""Save task list to file."""
with open(self.config.task_list_path, 'w') as f:
json.dump(task_list, f, indent=2)
def main():