279 lines
7.9 KiB
Python
279 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
import re
|
|
|
|
TASK_HEADING_RE = re.compile(r"^##\s+(TASK-\d+)\s+[-–—]\s+(.+)$", re.MULTILINE)
|
|
STATUS_RE = re.compile(r"^Status:\s*(.+)$", re.MULTILINE)
|
|
|
|
GOAL_RE = re.compile(r"^Goal:\s*(.+)", re.MULTILINE)
|
|
AC_RE = re.compile(r"^- (.+)$", re.MULTILINE)
|
|
GAP_RE = re.compile(r"^Implementation Gap:\s*(.+)", re.MULTILINE)
|
|
|
|
|
|
def _indefinite_article(word: str) -> str:
|
|
"""Return 'a' or 'an' based on the first letter of word."""
|
|
if not word:
|
|
return "an"
|
|
first = word[0].lower()
|
|
if first in "aeiou":
|
|
return "an"
|
|
return "a"
|
|
|
|
|
|
@dataclass
|
|
class Task:
|
|
task_id: str
|
|
title: str
|
|
status: str
|
|
body: str
|
|
|
|
def role(self, root: Path) -> tuple[str, str]:
|
|
if "Role:" not in self.body:
|
|
return ("Implementation Agent", _indefinite_article("Implementation Agent"))
|
|
parts = self.body.split("Role:", 1)
|
|
value = parts[1].splitlines()[0].strip()
|
|
if value:
|
|
return (value, _indefinite_article(value))
|
|
return ("Implementation Agent", _indefinite_article("Implementation Agent"))
|
|
|
|
def goal(self, root: Path) -> str:
|
|
match = GOAL_RE.search(self.body)
|
|
if match:
|
|
return match.group(1).strip()
|
|
return ""
|
|
|
|
def ac_lines(self) -> list[str]:
|
|
return AC_RE.findall(self.body)
|
|
|
|
def implementation_gap(self, root: Path) -> str | None:
|
|
match = GAP_RE.search(self.body)
|
|
if match and match.group(1).strip():
|
|
return match.group(1).strip()
|
|
return None
|
|
|
|
|
|
def read_tasks(root: Path) -> list[Task]:
|
|
tasks_path = root / "TASKS.md"
|
|
if not tasks_path.exists():
|
|
return []
|
|
|
|
text = tasks_path.read_text(encoding="utf-8")
|
|
matches = list(TASK_HEADING_RE.finditer(text))
|
|
tasks: list[Task] = []
|
|
|
|
for index, match in enumerate(matches):
|
|
start = match.start()
|
|
end = matches[index + 1].start() if index + 1 < len(matches) else len(text)
|
|
block = text[start:end]
|
|
status_match = STATUS_RE.search(block)
|
|
status = status_match.group(1).strip() if status_match else "Unknown"
|
|
tasks.append(Task(match.group(1), match.group(2).strip(), status, block))
|
|
|
|
return tasks
|
|
|
|
|
|
def get_next_task(root: Path) -> Task | None:
|
|
for task in read_tasks(root):
|
|
if task.status.lower() == "todo":
|
|
return task
|
|
return None
|
|
|
|
|
|
def update_task_status(root: Path, task_id: str, new_status: str) -> bool:
|
|
tasks_path = root / "TASKS.md"
|
|
if not tasks_path.exists():
|
|
return False
|
|
|
|
text = tasks_path.read_text(encoding="utf-8")
|
|
matches = list(TASK_HEADING_RE.finditer(text))
|
|
|
|
for index, match in enumerate(matches):
|
|
if match.group(1) != task_id:
|
|
continue
|
|
|
|
start = match.start()
|
|
end = matches[index + 1].start() if index + 1 < len(matches) else len(text)
|
|
block = text[start:end]
|
|
if STATUS_RE.search(block):
|
|
new_block = STATUS_RE.sub(f"Status: {new_status}", block, count=1)
|
|
else:
|
|
lines = block.splitlines()
|
|
lines.insert(1, f"Status: {new_status}")
|
|
new_block = "\n".join(lines) + "\n"
|
|
|
|
tasks_path.write_text(text[:start] + new_block + text[end:], encoding="utf-8")
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
CONTEXT_FILES_TO_READ = [
|
|
"README.md",
|
|
"TASKS.md",
|
|
"PROJECT_STATE.md",
|
|
"AGENT_HANDOFF.md",
|
|
"context/agent-guidelines.md",
|
|
]
|
|
|
|
TEST_CMD_SECTION_RE = re.compile(
|
|
r"^##\s*Test Commands\s*\n([\s\S]*?)(?=^##|\Z)",
|
|
re.MULTILINE,
|
|
)
|
|
BASH_BLOCK_RE = re.compile(r"```bash\n(.*?)```", re.DOTALL)
|
|
|
|
|
|
def _read_file_safe(root: Path, relative: str) -> str:
|
|
path = root / relative
|
|
if path.exists():
|
|
return path.read_text(encoding="utf-8")
|
|
return f"# {relative} — not found"
|
|
|
|
|
|
def _extract_test_commands_from_claude(root: Path) -> list[str]:
|
|
"""Extract bash test commands from CLAUDE.md Test Commands section."""
|
|
|
|
def _looks_like_command(line: str) -> bool:
|
|
"""Heuristic: keep lines that look like shell commands."""
|
|
if not line.strip():
|
|
return False
|
|
lower = line.lower()
|
|
# Skip markdown, prose, headings
|
|
if any(lower.startswith(p) for p in ("# ", "- ", "if ", "do not", "for ", "use ")) or lower in (
|
|
"validation commands",
|
|
):
|
|
return False
|
|
# Skip lines that look like sentences (contain spaces followed by lowercase words)
|
|
# but keep things like `source .venv/bin/activate` and `pip install ...`
|
|
parts = line.split()
|
|
if len(parts) <= 1:
|
|
return True
|
|
# If the first word is a known shell builtin / command prefix, accept it
|
|
known_prefixes = ("source", "cd", "ls", "cp", "mv", "rm", "mkdir", "echo", "grep",
|
|
"git", "pip", "python", "pytest", "rdb", "cat", "head", "tail",
|
|
"sed", "awk", "find", "install")
|
|
return parts[0].lower() in known_prefixes
|
|
|
|
claude_path = root / "CLAUDE.md"
|
|
if not claude_path.exists():
|
|
return []
|
|
|
|
text = claude_path.read_text(encoding="utf-8")
|
|
section_match = TEST_CMD_SECTION_RE.search(text)
|
|
if not section_match:
|
|
return []
|
|
|
|
section_text = section_match.group(1)
|
|
blocks = BASH_BLOCK_RE.findall(section_text)
|
|
seen: set[str] = set()
|
|
commands: list[str] = []
|
|
for block in blocks:
|
|
for line in block.strip().splitlines():
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
if _looks_like_command(stripped):
|
|
if stripped not in seen:
|
|
seen.add(stripped)
|
|
commands.append(stripped)
|
|
return commands
|
|
|
|
|
|
def generate_agent_prompt(root: Path) -> str:
|
|
task = get_next_task(root)
|
|
if not task:
|
|
return "No Todo task found."
|
|
|
|
role_text, article = task.role(root)
|
|
|
|
sections = [
|
|
f"You are {article} {role_text.lower()}.",
|
|
"",
|
|
"---",
|
|
"",
|
|
"# Read these files first",
|
|
"",
|
|
f"""- README.md
|
|
- TASKS.md
|
|
- PROJECT_STATE.md
|
|
- AGENT_HANDOFF.md""",
|
|
"",
|
|
"- context/agent-guidelines.md",
|
|
"",
|
|
"---",
|
|
"",
|
|
"# Task",
|
|
"",
|
|
f"## {task.task_id} — {task.title}",
|
|
"",
|
|
f"Status: {task.status}",
|
|
"",
|
|
f"Goal:\n{task.goal(root)}",
|
|
]
|
|
|
|
gap = task.implementation_gap(root)
|
|
if gap:
|
|
sections.extend([
|
|
"",
|
|
"Implementation Gap:",
|
|
gap,
|
|
])
|
|
|
|
sections.extend([
|
|
"",
|
|
"Acceptance Criteria:",
|
|
])
|
|
|
|
for line in task.ac_lines():
|
|
sections.append(f"- {line}")
|
|
|
|
sections.extend([
|
|
"",
|
|
"---",
|
|
"",
|
|
"# Constraints",
|
|
"",
|
|
"- Inspect the codebase first, then edit.",
|
|
"- Do not repeatedly reread unchanged files.",
|
|
"- Limit your work to ONE small implementation step only. Do not combine tasks or features.",
|
|
"- Make the smallest useful change possible.",
|
|
])
|
|
|
|
test_cmds = _extract_test_commands_from_claude(root)
|
|
if test_cmds:
|
|
sections.extend([
|
|
"",
|
|
"---",
|
|
"",
|
|
"# Test Commands (from CLAUDE.md)",
|
|
"",
|
|
"Use these existing test commands:",
|
|
("```bash\n" + "\n".join(test_cmds) + "\n```\n"),
|
|
])
|
|
|
|
sections.extend([
|
|
"",
|
|
"---",
|
|
"",
|
|
"# Validation",
|
|
"",
|
|
"- Run `rdb status` to confirm stage and task state.",
|
|
"- Run tests: `python -m pytest`.",
|
|
"- Verify acceptance criteria are met.",
|
|
"",
|
|
"---",
|
|
"",
|
|
"# Reporting",
|
|
"",
|
|
"When complete, provide:",
|
|
"- Summary of changes",
|
|
"- Files modified",
|
|
"- Test results",
|
|
"- Validation notes",
|
|
"- Update TASKS.md status to Done if all criteria pass",
|
|
"",
|
|
])
|
|
|
|
return "\n".join(sections)
|