mirror of
https://github.com/leonvanzyl/autocoder.git
synced 2026-01-30 06:12:06 +00:00
9c07dd72db
Applied Fixes
More flexible string matching: Changed from response.lower().strip().startswith("limit reached") to "limit reached" in response.lower() to handle cases where the message has prefix text or variations in whitespace.
Improved regex pattern: Updated to r"(?i)\bresets(?:\s+at)?\s+(\d+)(?::(\d+))?\s*(am|pm)\s*\(([^)]+)\)" which now handles:
Optional "at" after "resets" (e.g., "resets at 3pm" or "resets 3pm")
Flexible whitespace around components
Word boundaries to prevent partial matches
Timezone sanitization: Added .strip() to tz_name = match.group(4).strip() to remove any leading/trailing whitespace that could cause ZoneInfo() to fail.
Safety clamp: Added delay_seconds = min(delta.total_seconds(), 24 * 60 * 60) to ensure the delay never exceeds 24 hours, preventing the agent from being stuck waiting for extremely long periods.
287 lines
10 KiB
Python
287 lines
10 KiB
Python
"""
|
|
Agent Session Logic
|
|
===================
|
|
|
|
Core agent interaction functions for running autonomous coding sessions.
|
|
"""
|
|
|
|
import asyncio
|
|
import io
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from claude_agent_sdk import ClaudeSDKClient
|
|
|
|
# Fix Windows console encoding for Unicode characters (emoji, etc.)
|
|
# Without this, print() crashes when Claude outputs emoji like ✅
|
|
if sys.platform == "win32":
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
|
|
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
|
|
|
|
from client import create_client
|
|
from progress import has_features, print_progress_summary, print_session_header
|
|
from prompts import (
|
|
copy_spec_to_project,
|
|
get_coding_prompt,
|
|
get_coding_prompt_yolo,
|
|
get_initializer_prompt,
|
|
)
|
|
|
|
# Configuration
|
|
AUTO_CONTINUE_DELAY_SECONDS = 3
|
|
|
|
|
|
async def run_agent_session(
|
|
client: ClaudeSDKClient,
|
|
message: str,
|
|
project_dir: Path,
|
|
) -> tuple[str, str]:
|
|
"""
|
|
Run a single agent session using Claude Agent SDK.
|
|
|
|
Args:
|
|
client: Claude SDK client
|
|
message: The prompt to send
|
|
project_dir: Project directory path
|
|
|
|
Returns:
|
|
(status, response_text) where status is:
|
|
- "continue" if agent should continue working
|
|
- "error" if an error occurred
|
|
"""
|
|
print("Sending prompt to Claude Agent SDK...\n")
|
|
|
|
try:
|
|
# Send the query
|
|
await client.query(message)
|
|
|
|
# Collect response text and show tool use
|
|
response_text = ""
|
|
async for msg in client.receive_response():
|
|
msg_type = type(msg).__name__
|
|
|
|
# Handle AssistantMessage (text and tool use)
|
|
if msg_type == "AssistantMessage" and hasattr(msg, "content"):
|
|
for block in msg.content:
|
|
block_type = type(block).__name__
|
|
|
|
if block_type == "TextBlock" and hasattr(block, "text"):
|
|
response_text += block.text
|
|
print(block.text, end="", flush=True)
|
|
elif block_type == "ToolUseBlock" and hasattr(block, "name"):
|
|
print(f"\n[Tool: {block.name}]", flush=True)
|
|
if hasattr(block, "input"):
|
|
input_str = str(block.input)
|
|
if len(input_str) > 200:
|
|
print(f" Input: {input_str[:200]}...", flush=True)
|
|
else:
|
|
print(f" Input: {input_str}", flush=True)
|
|
|
|
# Handle UserMessage (tool results)
|
|
elif msg_type == "UserMessage" and hasattr(msg, "content"):
|
|
for block in msg.content:
|
|
block_type = type(block).__name__
|
|
|
|
if block_type == "ToolResultBlock":
|
|
result_content = getattr(block, "content", "")
|
|
is_error = getattr(block, "is_error", False)
|
|
|
|
# Check if command was blocked by security hook
|
|
if "blocked" in str(result_content).lower():
|
|
print(f" [BLOCKED] {result_content}", flush=True)
|
|
elif is_error:
|
|
# Show errors (truncated)
|
|
error_str = str(result_content)[:500]
|
|
print(f" [Error] {error_str}", flush=True)
|
|
else:
|
|
# Tool succeeded - just show brief confirmation
|
|
print(" [Done]", flush=True)
|
|
|
|
print("\n" + "-" * 70 + "\n")
|
|
return "continue", response_text
|
|
|
|
except Exception as e:
|
|
print(f"Error during agent session: {e}")
|
|
return "error", str(e)
|
|
|
|
|
|
async def run_autonomous_agent(
|
|
project_dir: Path,
|
|
model: str,
|
|
max_iterations: Optional[int] = None,
|
|
yolo_mode: bool = False,
|
|
) -> None:
|
|
"""
|
|
Run the autonomous agent loop.
|
|
|
|
Args:
|
|
project_dir: Directory for the project
|
|
model: Claude model to use
|
|
max_iterations: Maximum number of iterations (None for unlimited)
|
|
yolo_mode: If True, skip browser testing and use YOLO prompt
|
|
"""
|
|
print("\n" + "=" * 70)
|
|
print(" AUTONOMOUS CODING AGENT DEMO")
|
|
print("=" * 70)
|
|
print(f"\nProject directory: {project_dir}")
|
|
print(f"Model: {model}")
|
|
if yolo_mode:
|
|
print("Mode: YOLO (testing disabled)")
|
|
else:
|
|
print("Mode: Standard (full testing)")
|
|
if max_iterations:
|
|
print(f"Max iterations: {max_iterations}")
|
|
else:
|
|
print("Max iterations: Unlimited (will run until completion)")
|
|
print()
|
|
|
|
# Create project directory
|
|
project_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Check if this is a fresh start or continuation
|
|
# Uses has_features() which checks if the database actually has features,
|
|
# not just if the file exists (empty db should still trigger initializer)
|
|
is_first_run = not has_features(project_dir)
|
|
|
|
if is_first_run:
|
|
print("Fresh start - will use initializer agent")
|
|
print()
|
|
print("=" * 70)
|
|
print(" NOTE: First session takes 10-20+ minutes!")
|
|
print(" The agent is generating 200 detailed test cases.")
|
|
print(" This may appear to hang - it's working. Watch for [Tool: ...] output.")
|
|
print("=" * 70)
|
|
print()
|
|
# Copy the app spec into the project directory for the agent to read
|
|
copy_spec_to_project(project_dir)
|
|
else:
|
|
print("Continuing existing project")
|
|
print_progress_summary(project_dir)
|
|
|
|
# Main loop
|
|
iteration = 0
|
|
|
|
while True:
|
|
iteration += 1
|
|
|
|
# Check max iterations
|
|
if max_iterations and iteration > max_iterations:
|
|
print(f"\nReached max iterations ({max_iterations})")
|
|
print("To continue, run the script again without --max-iterations")
|
|
break
|
|
|
|
# Print session header
|
|
print_session_header(iteration, is_first_run)
|
|
|
|
# Create client (fresh context)
|
|
client = create_client(project_dir, model, yolo_mode=yolo_mode)
|
|
|
|
# Choose prompt based on session type
|
|
# Pass project_dir to enable project-specific prompts
|
|
if is_first_run:
|
|
prompt = get_initializer_prompt(project_dir)
|
|
is_first_run = False # Only use initializer once
|
|
else:
|
|
# Use YOLO prompt if in YOLO mode
|
|
if yolo_mode:
|
|
prompt = get_coding_prompt_yolo(project_dir)
|
|
else:
|
|
prompt = get_coding_prompt(project_dir)
|
|
|
|
# Run session with async context manager
|
|
async with client:
|
|
status, response = await run_agent_session(client, prompt, project_dir)
|
|
|
|
# Handle status
|
|
if status == "continue":
|
|
delay_seconds = AUTO_CONTINUE_DELAY_SECONDS
|
|
target_time_str = None
|
|
|
|
if "limit reached" in response.lower():
|
|
print("Claude Agent SDK indicated limit reached.")
|
|
|
|
# Try to parse reset time from response
|
|
match = re.search(
|
|
r"(?i)\bresets(?:\s+at)?\s+(\d+)(?::(\d+))?\s*(am|pm)\s*\(([^)]+)\)",
|
|
response,
|
|
)
|
|
if match:
|
|
hour = int(match.group(1))
|
|
minute = int(match.group(2)) if match.group(2) else 0
|
|
period = match.group(3).lower()
|
|
tz_name = match.group(4).strip()
|
|
|
|
# Convert to 24-hour format
|
|
if period == "pm" and hour != 12:
|
|
hour += 12
|
|
elif period == "am" and hour == 12:
|
|
hour = 0
|
|
|
|
try:
|
|
tz = ZoneInfo(tz_name)
|
|
now = datetime.now(tz)
|
|
target = now.replace(
|
|
hour=hour, minute=minute, second=0, microsecond=0
|
|
)
|
|
|
|
# If target time has already passed today, wait until tomorrow
|
|
if target <= now:
|
|
target += timedelta(days=1)
|
|
|
|
delta = target - now
|
|
delay_seconds = min(
|
|
delta.total_seconds(), 24 * 60 * 60
|
|
) # Clamp to 24 hours max
|
|
target_time_str = target.strftime("%B %d, %Y at %I:%M %p %Z")
|
|
|
|
except Exception as e:
|
|
print(f"Error parsing reset time: {e}, using default delay")
|
|
|
|
if target_time_str:
|
|
print(
|
|
f"\nClaude Code Limit Reached. Agent will auto-continue in {delay_seconds:.0f}s ({target_time_str})...",
|
|
flush=True,
|
|
)
|
|
else:
|
|
print(
|
|
f"\nAgent will auto-continue in {delay_seconds:.0f}s...", flush=True
|
|
)
|
|
|
|
sys.stdout.flush() # this should allow the pause to be displayed before sleeping
|
|
print_progress_summary(project_dir)
|
|
await asyncio.sleep(delay_seconds)
|
|
|
|
elif status == "error":
|
|
print("\nSession encountered an error")
|
|
print("Will retry with a fresh session...")
|
|
await asyncio.sleep(AUTO_CONTINUE_DELAY_SECONDS)
|
|
|
|
# Small delay between sessions
|
|
if max_iterations is None or iteration < max_iterations:
|
|
print("\nPreparing next session...\n")
|
|
await asyncio.sleep(1)
|
|
|
|
# Final summary
|
|
print("\n" + "=" * 70)
|
|
print(" SESSION COMPLETE")
|
|
print("=" * 70)
|
|
print(f"\nProject directory: {project_dir}")
|
|
print_progress_summary(project_dir)
|
|
|
|
# Print instructions for running the generated application
|
|
print("\n" + "-" * 70)
|
|
print(" TO RUN THE GENERATED APPLICATION:")
|
|
print("-" * 70)
|
|
print(f"\n cd {project_dir.resolve()}")
|
|
print(" ./init.sh # Run the setup script")
|
|
print(" # Or manually:")
|
|
print(" npm install && npm run dev")
|
|
print("\n Then open http://localhost:3000 (or check init.sh for the URL)")
|
|
print("-" * 70)
|
|
|
|
print("\nDone!")
|