davebb/linear-coding-agent
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
bf790b63a002bc9daff539e433d468512c494200
linear-coding-agent/agent.py

262 lines
9.4 KiB
Python
Raw Blame History

"""
Agent Session Logic
===================
Core agent interaction functions for running autonomous coding sessions.
"""
import asyncio
from pathlib import Path
from typing import Optional
from claude_code_sdk import ClaudeSDKClient
from client import create_client
from progress import print_session_header, print_progress_summary, is_linear_initialized
from prompts import (
get_initializer_prompt,
get_initializer_bis_prompt,
get_coding_prompt,
copy_spec_to_project,
copy_new_spec_to_project,
)
# 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
- "done" if the project appears to be complete
- "error" if an error occurred
"""
print("Sending prompt to Claude Agent SDK...\n")
try:
project_done = False
# 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)
# Early guard: if the assistant declares the project feature-complete,
# we stop streaming and end the loop as "done" to avoid extra work.
if "feature-complete" in block.text.lower():
project_done = True
break
elif block_type == "ToolUseBlock" and hasattr(block, "name"):
if project_done:
break
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"):
if project_done:
# Ignore further tool results once we've determined the project is done
continue
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)
if project_done:
# Stop processing further streamed messages
break
print("\n" + "-" * 70 + "\n")
# Detect completion signals in the assistant's response to stop the loop
lower_text = response_text.lower()
if (
"feature-complete" in lower_text
or ("all" in lower_text and "issues" in lower_text and "completed" in lower_text)
):
return "done", response_text
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,
new_spec_filename: Optional[str] = None,
) -> 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)
"""
print("\n" + "=" * 70)
print(" AUTONOMOUS CODING AGENT DEMO")
print("=" * 70)
print(f"\nProject directory: {project_dir}")
print(f"Model: {model}")
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, continuation, or adding new specs
# We use .linear_project.json as the marker for initialization
is_first_run = not is_linear_initialized(project_dir)
use_initializer_bis = new_spec_filename is not None and not is_first_run
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 creating 50 Linear issues and setting up the project.")
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)
elif use_initializer_bis:
print("Adding new specifications - will use initializer bis agent")
print()
print("=" * 70)
print(f" NOTE: Adding new features from {new_spec_filename}")
print(" The agent will create new Linear issues for the additional features.")
print(" This may take several minutes. Watch for [Tool: ...] output.")
print("=" * 70)
print()
# Copy the new spec file into the project directory
copy_new_spec_to_project(project_dir, new_spec_filename)
print_progress_summary(project_dir)
else:
print("Continuing existing project (Linear initialized)")
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
is_initializer_session = is_first_run or (use_initializer_bis and iteration == 1)
is_bis_session = use_initializer_bis and iteration == 1
print_session_header(iteration, is_initializer_session, is_bis_session)
# Create client (fresh context)
client = create_client(project_dir, model)
# Choose prompt based on session type
if is_first_run:
prompt = get_initializer_prompt()
is_first_run = False # Only use initializer once
elif use_initializer_bis and iteration == 1:
prompt = get_initializer_bis_prompt()
use_initializer_bis = False # Only use initializer bis once
else:
prompt = get_coding_prompt()
# Run session with async context manager
async with client:
status, response = await run_agent_session(client, prompt, project_dir)
# Handle status
if status == "continue":
print(f"\nAgent will auto-continue in {AUTO_CONTINUE_DELAY_SECONDS}s...")
print_progress_summary(project_dir)
await asyncio.sleep(AUTO_CONTINUE_DELAY_SECONDS)
elif status == "error":
print("\nSession encountered an error")
print("Will retry with a fresh session...")
await asyncio.sleep(AUTO_CONTINUE_DELAY_SECONDS)
elif status == "done":
print("\nAgent reports that the project is complete (all issues done).")
print_progress_summary(project_dir)
break
# 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!")
Reference in New Issue View Git Blame Copy Permalink