diff --git a/backend/agent/coding_agent.py b/backend/agent/coding_agent.py index e69de29..65307ad 100644 --- a/backend/agent/coding_agent.py +++ b/backend/agent/coding_agent.py @@ -0,0 +1,490 @@ +""" +Coding Agent +============ +Autonomous AI coding agent based on the Plan→Act→Observe→Fix→Done loop. +Structure follows ex05_coding_agent_solution.py from the course. + +Instead of a blocking CLI loop (input()), the CodingAgent class exposes +step-by-step methods so Streamlit can drive the loop via session_state: + + agent.start_task(task) # initialise + action = agent.propose_next_action() # ask LLM → returns action, does NOT execute + result = agent.approve() # execute pending action + agent.reject(feedback) # skip action, inject user feedback +""" + +import ast +import inspect +import json +import os +import subprocess +import sys +from pathlib import Path + +import requests +from dotenv import load_dotenv + +load_dotenv() + +# ── Workspace ──────────────────────────────────────────────────────────────── +# Two levels up from backend/agent/ → project root → workspace/ +WORKSPACE = Path(__file__).resolve().parents[2] / "workspace" +WORKSPACE.mkdir(exist_ok=True) + +# ── Agent limits ───────────────────────────────────────────────────────────── +MAX_ITERATIONS = 100 +MAX_RESULT_LENGTH = 10_000 +MAX_HISTORY_CHARS = 80_000 + + +# ═════════════════════════════════════════════════════════════════════════════ +# PART A – TOOL FUNCTIONS +# ═════════════════════════════════════════════════════════════════════════════ +# +# Each tool is a plain Python function decorated with @register_tool. +# The decorator adds the function to TOOL_REGISTRY so the dispatcher +# can call it by name at runtime. + +TOOL_REGISTRY: dict[str, callable] = {} + + +def register_tool(func): + """Decorator – adds a function to the global tool registry.""" + TOOL_REGISTRY[func.__name__] = func + return func + + +@register_tool +def read_file(path: str) -> str: + """Read a .py or .txt file from the workspace and return its contents.""" + target = (WORKSPACE / path).resolve() + if not str(target).startswith(str(WORKSPACE.resolve())): + return "ERROR: path is outside the workspace." + if not target.exists(): + return f"ERROR: file '{path}' not found." + if target.suffix not in (".py", ".txt"): + return f"ERROR: can only read .py and .txt files, got '{target.suffix}'." + return target.read_text() + + +@register_tool +def write_file(path: str, content: str) -> str: + """Write content to a .py or .txt file in the workspace.""" + target = (WORKSPACE / path).resolve() + if not str(target).startswith(str(WORKSPACE.resolve())): + return "ERROR: path is outside the workspace." + if target.suffix not in (".py", ".txt"): + return f"ERROR: can only write .py and .txt files, got '{target.suffix}'." + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content) + return f"OK: wrote {len(content)} chars to {path}." + + +@register_tool +def list_files(file_glob: str = "*") -> str: + """List files in the workspace matching the glob pattern.""" + found = sorted(WORKSPACE.glob(file_glob)) + found = [f.relative_to(WORKSPACE) for f in found if f.is_file()] + if not found: + return f"No files matching '{file_glob}' in workspace." + return "\n".join(str(f) for f in found) + + +@register_tool +def grep_search(pattern: str, file_glob: str = "*.py") -> str: + """Search for a pattern in workspace files and return matching lines with line numbers.""" + matches = [] + for filepath in sorted(WORKSPACE.glob(file_glob)): + if filepath.suffix not in (".py", ".txt"): + continue + try: + lines = filepath.read_text().splitlines() + except Exception: + continue + for i, line in enumerate(lines, 1): + if pattern in line: + rel = filepath.relative_to(WORKSPACE) + matches.append(f"{rel}:{i}: {line}") + if not matches: + return f"No matches for '{pattern}' in {file_glob}." + return "\n".join(matches) + + +@register_tool +def run_python(path: str) -> str: + """Execute a Python file in the workspace and return stdout and stderr.""" + target = (WORKSPACE / path).resolve() + if not str(target).startswith(str(WORKSPACE.resolve())): + return "ERROR: path is outside the workspace." + if not target.exists(): + return f"ERROR: file '{path}' not found." + result = subprocess.run( + [sys.executable, str(target)], + capture_output=True, text=True, timeout=30, + cwd=str(WORKSPACE), + ) + output = "" + if result.stdout: + output += f"STDOUT:\n{result.stdout}" + if result.stderr: + output += f"STDERR:\n{result.stderr}" + output += f"\nExit code: {result.returncode}" + return output.strip() + + +@register_tool +def validate_python(path: str) -> str: + """Check whether a Python file has valid syntax using ast.parse.""" + target = (WORKSPACE / path).resolve() + if not str(target).startswith(str(WORKSPACE.resolve())): + return "ERROR: path is outside the workspace." + if not target.exists(): + return f"ERROR: file '{path}' not found." + source = target.read_text() + try: + ast.parse(source) + return "OK: syntax is valid." + except SyntaxError as e: + return f"SYNTAX ERROR: {e}" + + +@register_tool +def done(summary: str) -> str: + """Signal that the agent has finished its task.""" + return f"DONE: {summary}" + + +# ═════════════════════════════════════════════════════════════════════════════ +# PART B – TOOL DISPATCHER +# ═════════════════════════════════════════════════════════════════════════════ + +def build_tool_description() -> str: + """Auto-generate tool descriptions from function signatures and docstrings.""" + lines = [] + for name, func in TOOL_REGISTRY.items(): + sig = inspect.signature(func) + params = [] + for pname, param in sig.parameters.items(): + if param.default is inspect.Parameter.empty: + params.append(f'"{pname}": ""') + else: + params.append(f'"{pname}": ""') + param_str = ", ".join(params) + doc = (func.__doc__ or "").strip().split("\n")[0] + lines.append(f" - {name}({{{param_str}}}): {doc}") + return "\n".join(lines) + + +def dispatch_tool(tool_name: str, arguments: dict) -> str: + """Call a tool by name with the given arguments.""" + if tool_name not in TOOL_REGISTRY: + return f"ERROR: unknown tool '{tool_name}'. Available: {list(TOOL_REGISTRY.keys())}" + func = TOOL_REGISTRY[tool_name] + try: + return func(**arguments) + except TypeError as e: + return f"ERROR calling {tool_name}: {e}" + except Exception as e: + return f"ERROR in {tool_name}: {type(e).__name__}: {e}" + + +# ═════════════════════════════════════════════════════════════════════════════ +# PART C – SYSTEM PROMPT +# ═════════════════════════════════════════════════════════════════════════════ + +SYSTEM_PROMPT = f"""\ +You are a coding agent that helps users with Python programming tasks. +You work inside a workspace directory and have access to tools. + + +You can: + - Read .py and .txt files from the workspace + - Write .py and .txt files to the workspace + - List files in the workspace + - Search for patterns in files using grep + - Execute Python files and see their output + - Validate Python syntax using ast.parse + - Signal completion when the task is done + + + +{build_tool_description()} + + + +For every user request, follow this workflow: + 1. PLAN: Think about what steps are needed. List them in "thought". + 2. ACT: Choose ONE tool to call for the current step. + 3. OBSERVE: You will receive the tool output. Analyse it carefully. + 4. REPLAN: If the result was unexpected, revise your plan in "thought". + 5. REPEAT: Go back to step 2 if more work is needed. + 6. DONE: Call the "done" tool when the task is complete. + + + +You MUST respond with a JSON object every time: +{{ + "thought": "", + "tool": "", + "arguments": {{ }} +}} + +Example: +{{ + "thought": "I need to write fibonacci.py first, then validate and run it.", + "tool": "write_file", + "arguments": {{"path": "fibonacci.py", "content": "def fib(n): ..."}} +}} + + + + - Always plan before acting. + - Call exactly ONE tool per response. + - After writing code, ALWAYS validate it with validate_python. + - After validation passes, run it with run_python to verify correctness. + - If an error occurs, analyse it and try to fix it (up to 3 retries). + - Stay within the workspace directory. + - When the task is fully complete, call the "done" tool. + - If you receive a , acknowledge it and adjust your plan. + - If you receive a tag, revise your plan before choosing the next tool. + +""" + + +# ═════════════════════════════════════════════════════════════════════════════ +# PART D – CODING AGENT CLASS +# ═════════════════════════════════════════════════════════════════════════════ + +def truncate_result(result: str) -> str: + """Truncate a tool result that exceeds MAX_RESULT_LENGTH.""" + if len(result) <= MAX_RESULT_LENGTH: + return result + half = MAX_RESULT_LENGTH // 2 + return ( + result[:half] + + f"\n\n... [TRUNCATED – {len(result)} chars total] ...\n\n" + + result[-half:] + ) + + +def trim_messages(messages: list) -> list: + """Drop old messages when history exceeds MAX_HISTORY_CHARS. + Always keeps the system prompt (index 0) and original task (index 1). + """ + total = sum(len(m["content"]) for m in messages) + if total <= MAX_HISTORY_CHARS: + return messages + + head = messages[:2] + tail = messages[2:] + original_task = messages[1]["content"] if len(messages) > 1 else "" + + while tail and sum(len(m["content"]) for m in head + tail) > MAX_HISTORY_CHARS: + tail.pop(0) + + reminder = { + "role": "user", + "content": ( + "Earlier conversation history was trimmed to fit the context window. " + f"REMINDER – your original task was:\n{original_task}\n" + "Continue working towards completing this task. " + "Do NOT start over or redo work already completed." + ), + } + return head + [reminder] + tail + + +def _strip_code_fences(text: str) -> str: + """Remove markdown code fences (```json ... ```) from a string.""" + text = text.strip() + if text.startswith("```"): + lines = text.split("\n") + end = -1 if lines[-1].strip() == "```" else len(lines) + text = "\n".join(lines[1:end]) + return text.strip() + + +class CodingAgent: + """Step-by-step coding agent for use in Streamlit. + + Usage: + agent = CodingAgent() + agent.start_task("Write a fibonacci function") + + action = agent.propose_next_action() + # → {"thought": "...", "tool": "write_file", "arguments": {...}} + # Show action in UI, wait for user input + + result = agent.approve() + # → {"tool": "write_file", "result": "OK: ...", "is_done": False} + + agent.reject("Please use recursion instead") + # → agent replans on next propose_next_action() + """ + + def __init__(self): + self.messages: list = [] + self.pending_action: dict | None = None + self.is_done: bool = False + self.iteration: int = 0 + self._setup_api() + + def _setup_api(self): + """Load API configuration from environment variables.""" + self.api_url = f"http://{os.getenv('HOST')}:{os.getenv('PORT')}/v1/chat/completions" + self.api_key = os.getenv("API_KEY") + self.model = os.getenv("MODEL") + + def _call_api(self, messages: list) -> str: + """Make a raw API call and return the response content string.""" + headers = {"Content-Type": "application/json"} + if self.api_key and self.api_key != "EMPTY": + headers["Authorization"] = f"Bearer {self.api_key}" + + payload = { + "model": self.model, + "messages": messages, + "temperature": 0.2, + "max_tokens": 4096, + "stream": False, + } + + response = requests.post(self.api_url, headers=headers, json=payload, timeout=60) + + if response.status_code != 200: + raise Exception(f"API Error {response.status_code}: {response.text}") + + data = response.json() + if "choices" in data and len(data["choices"]) > 0: + return data["choices"][0]["message"]["content"] + raise Exception("Invalid API response format") + + # ── Public interface ────────────────────────────────────────────────────── + + def start_task(self, task: str) -> None: + """Initialise the agent with a new task. Resets all state.""" + self.messages = [ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": task}, + ] + self.pending_action = None + self.is_done = False + self.iteration = 0 + + def propose_next_action(self) -> dict: + """Ask the LLM what to do next. + + Returns the parsed action dict without executing anything. + The action is stored internally as pending_action until + approve() or reject() is called. + + Returns: + {"thought": str, "tool": str, "arguments": dict} + """ + if self.is_done: + return {"thought": "Task already completed.", "tool": "done", "arguments": {}} + + if self.iteration >= MAX_ITERATIONS: + return {"thought": "Max iterations reached.", "tool": "done", + "arguments": {"summary": "Stopped: max iterations reached."}} + + self.iteration += 1 + self.messages = trim_messages(self.messages) + + try: + raw = self._call_api(self.messages) + raw = _strip_code_fences(raw) + action = json.loads(raw) + except json.JSONDecodeError: + action = { + "thought": "Could not parse LLM response as JSON.", + "tool": "done", + "arguments": {"summary": "Stopped: JSON parse error."}, + } + raw = json.dumps(action) + except Exception as e: + action = { + "thought": f"API call failed: {e}", + "tool": "done", + "arguments": {"summary": f"Stopped: {e}"}, + } + raw = json.dumps(action) + + self.pending_action = {"raw": raw, "action": action} + return action + + def approve(self) -> dict: + """Execute the pending action and return the result. + + Returns: + {"tool": str, "arguments": dict, "result": str, "is_done": bool} + """ + if not self.pending_action: + raise Exception("No pending action. Call propose_next_action() first.") + + raw = self.pending_action["raw"] + action = self.pending_action["action"] + tool_name = action.get("tool", "") + arguments = action.get("arguments", {}) + + # Append the assistant message to history + self.messages.append({"role": "assistant", "content": raw}) + self.pending_action = None + + # Handle completion + if tool_name == "done": + self.is_done = True + return { + "tool": "done", + "arguments": arguments, + "result": arguments.get("summary", "Task completed."), + "is_done": True, + } + + # Execute the tool + result = dispatch_tool(tool_name, arguments) + result = truncate_result(result) + + # Build feedback – nudge agent to replan on errors + feedback = f'\n{result}\n' + if result.startswith("ERROR") or result.startswith("SYNTAX ERROR"): + feedback += ( + "\n\nThe tool returned an error. " + "Re-examine your plan: what went wrong and what should you do differently? " + "State your revised plan in your next thought." + ) + + self.messages.append({"role": "user", "content": feedback}) + + return { + "tool": tool_name, + "arguments": arguments, + "result": result, + "is_done": False, + } + + def reject(self, feedback: str) -> None: + """Reject the pending action and inject user feedback. + + The pending action is NOT executed. On the next call to + propose_next_action() the agent will replan based on the feedback. + """ + if not self.pending_action: + return + + raw = self.pending_action["raw"] + action = self.pending_action["action"] + tool_name = action.get("tool", "unknown") + + self.messages.append({"role": "assistant", "content": raw}) + self.messages.append({ + "role": "user", + "content": ( + f"{feedback}\n" + "The user has given you guidance BEFORE you executed " + f"your proposed action ({tool_name}). Do NOT proceed with that action. " + "Revise your plan to incorporate their feedback and state your " + "updated plan in your next thought." + ), + }) + self.pending_action = None diff --git a/backend/agent/coding_agent_example.py b/backend/agent/coding_agent_example.py deleted file mode 100644 index 8954d12..0000000 --- a/backend/agent/coding_agent_example.py +++ /dev/null @@ -1,605 +0,0 @@ -""" -Exercise 5b -- Build a Basic AI Coding Agent (Guided Version) -============================================================== -AISE501 . Prompting in Coding . Spring Semester 2026 - -This is a GUIDED version of Exercise 5 with more scaffolding. -It teaches the same concepts but reduces boilerplate so you can -focus on the key insight: how an LLM uses tools. - -The key insight ---------------- -An LLM cannot run code or read files by itself. But we can give it -"superpowers" through a simple trick: - - 1. TELL the LLM (via the system prompt) what tools exist. - 2. ASK the LLM to respond with JSON saying which tool to call. - 3. PARSE the JSON, call the real Python function, and - 4. FEED the result back into the conversation as a new message. - -This is how ALL AI coding agents work (Claude Code, Cursor, Copilot). -The LLM never actually "runs" code — it just asks us to run it! - -What is already provided ------------------------- -To let you focus on the interesting parts, the following are PRE-BUILT: - - All 7 tool functions (Part A) — read_file, grep_search, etc. - - The tool dispatcher (Part B) — maps tool names to functions. - - Helper functions: truncate_result, trim_messages, ask_human. - -What you need to build (the interesting parts) ------------------------------------------------ - Part C The SYSTEM PROMPT that teaches the LLM about its tools (TODOs 1-2). - Part D The AGENT LOOP that connects the LLM to the tools (TODOs 3-6). - Part E The INTERACTIVE CHAT interface (TODOs 7-8). - -Think of it like wiring a robot: - - Part A+B are the robot's HANDS (already built). - - Part C is the robot's INSTRUCTION MANUAL (you write it). - - Part D is the robot's BRAIN LOOP (you wire it). - - Part E is the ON SWITCH (you connect it). - -The conversation flow ---------------------- -Here is exactly what happens in one iteration of the agent loop: - - ┌─────────────────────────────────────────────────────────────┐ - │ messages = [ │ - │ {"role": "system", "content": ""}, │ - │ {"role": "user", "content": "Fix the bug in app.py"},│ - │ ] │ - └──────────────────────────┬──────────────────────────────────┘ - │ - ┌─────────▼─────────┐ - │ LLM generates │ - │ JSON response │ - └─────────┬─────────┘ - │ - ┌──────────────▼──────────────┐ - │ {"thought": "I should...", │ - │ "tool": "read_file", │ - │ "arguments": { │ - │ "path": "app.py" │ - │ }} │ - └──────────────┬──────────────┘ - │ - ┌─────────▼─────────┐ - │ You parse JSON, │ - │ call read_file() │ - └─────────┬─────────┘ - │ - ┌──────────────▼──────────────────┐ - │ Append to messages: │ - │ {"role":"assistant", "content":..}│ - │ {"role":"user", "content": │ - │ "file contents │ - │ "} │ - └──────────────┬──────────────────┘ - │ - ┌─────────▼─────────┐ - │ Next iteration: │ - │ LLM sees result, │ - │ picks next tool │ - └───────────────────┘ -""" - -import ast -import json -import subprocess -import sys -from pathlib import Path - -from server_utils import ( - chat, - chat_json, - get_client, - print_messages, - print_separator, - strip_code_fences, -) - -client = get_client() - -# ── Agent Configuration ────────────────────────────────────────────────────── -WORKSPACE = Path(__file__).parent / "workspace" -WORKSPACE.mkdir(exist_ok=True) - -MAX_ITERATIONS = 50 -MAX_RESULT_LENGTH = 8000 -MAX_HISTORY_CHARS = 60000 - - -# ═══════════════════════════════════════════════════════════════════════════════ -# PART A -- TOOL FUNCTIONS (pre-built) -# ═══════════════════════════════════════════════════════════════════════════════ -# -# These are the tools the agent can use. Each is a normal Python function. -# The LLM will never call these directly — it will OUTPUT JSON saying -# "please call read_file with path='app.py'", and OUR CODE will call it. - - -def read_file(path: str) -> str: - """Read a .txt or .py file from the workspace and return its contents.""" - target = (WORKSPACE / path).resolve() - if not str(target).startswith(str(WORKSPACE.resolve())): - return "ERROR: path is outside the workspace." - if not target.exists(): - return f"ERROR: file '{path}' not found." - if target.suffix not in (".py", ".txt"): - return f"ERROR: can only read .py and .txt files, got '{target.suffix}'." - return target.read_text() - - -def grep_search(pattern: str, file_glob: str = "*.py") -> str: - """Search for a pattern in workspace files matching the glob.""" - matches = [] - for filepath in sorted(WORKSPACE.glob(file_glob)): - if filepath.suffix not in (".py", ".txt"): - continue - try: - lines = filepath.read_text().splitlines() - except Exception: - continue - for i, line in enumerate(lines, 1): - if pattern in line: - rel = filepath.relative_to(WORKSPACE) - matches.append(f"{rel}:{i}: {line}") - if not matches: - return f"No matches for '{pattern}' in {file_glob}." - return "\n".join(matches) - - -def list_files(file_glob: str = "*") -> str: - """List files in the workspace matching the glob pattern.""" - found = sorted(WORKSPACE.glob(file_glob)) - found = [f.relative_to(WORKSPACE) for f in found if f.is_file()] - if not found: - return f"No files matching '{file_glob}' in workspace." - return "\n".join(str(f) for f in found) - - -def write_file(path: str, content: str) -> str: - """Write content to a .py or .txt file in the workspace.""" - target = (WORKSPACE / path).resolve() - if not str(target).startswith(str(WORKSPACE.resolve())): - return "ERROR: path is outside the workspace." - if target.suffix not in (".py", ".txt"): - return f"ERROR: can only write .py and .txt files, got '{target.suffix}'." - target.parent.mkdir(parents=True, exist_ok=True) - target.write_text(content) - return f"OK: wrote {len(content)} chars to {path}." - - -def run_python(path: str) -> str: - """Execute a Python file in the workspace and return stdout + stderr.""" - target = (WORKSPACE / path).resolve() - if not str(target).startswith(str(WORKSPACE.resolve())): - return "ERROR: path is outside the workspace." - if not target.exists(): - return f"ERROR: file '{path}' not found." - result = subprocess.run( - [sys.executable, str(target)], - capture_output=True, - text=True, - timeout=30, - cwd=str(WORKSPACE), - ) - output = "" - if result.stdout: - output += f"STDOUT:\n{result.stdout}" - if result.stderr: - output += f"STDERR:\n{result.stderr}" - output += f"\nExit code: {result.returncode}" - return output.strip() - - -def validate_python(path: str) -> str: - """Check whether a Python file has valid syntax using ast.parse.""" - target = (WORKSPACE / path).resolve() - if not str(target).startswith(str(WORKSPACE.resolve())): - return "ERROR: path is outside the workspace." - if not target.exists(): - return f"ERROR: file '{path}' not found." - source = target.read_text() - try: - ast.parse(source) - return "OK: syntax is valid." - except SyntaxError as e: - return f"SYNTAX ERROR: {e}" - - -def done(summary: str) -> str: - """Signal that the agent has finished its task.""" - return f"DONE: {summary}" - - -# ═══════════════════════════════════════════════════════════════════════════════ -# PART B -- TOOL DISPATCHER (pre-built) -# ═══════════════════════════════════════════════════════════════════════════════ -# -# This is the bridge between the LLM's JSON output and Python function calls. -# -# When the LLM says: {"tool": "read_file", "arguments": {"path": "app.py"}} -# The dispatcher does: TOOL_FUNCTIONS["read_file"](path="app.py") -# -# The **arguments syntax means "unpack the dict as keyword arguments": -# {"path": "app.py"} → read_file(path="app.py") - -TOOL_FUNCTIONS = { - "read_file": read_file, - "grep_search": grep_search, - "list_files": list_files, - "write_file": write_file, - "run_python": run_python, - "validate_python": validate_python, - "done": done, -} - - -def dispatch_tool(tool_name: str, arguments: dict) -> str: - """Look up a tool by name and call it with the given arguments. - - Example: - dispatch_tool("read_file", {"path": "app.py"}) - → calls read_file(path="app.py") - → returns the file contents as a string - """ - if tool_name not in TOOL_FUNCTIONS: - return f"ERROR: unknown tool '{tool_name}'. Available: {list(TOOL_FUNCTIONS.keys())}" - func = TOOL_FUNCTIONS[tool_name] - try: - return func(**arguments) - except TypeError as e: - return f"ERROR calling {tool_name}: {e}" - except Exception as e: - return f"ERROR in {tool_name}: {type(e).__name__}: {e}" - - -# ═══════════════════════════════════════════════════════════════════════════════ -# PART C -- SYSTEM PROMPT (TODOs 1-2) -# ═══════════════════════════════════════════════════════════════════════════════ -# -# The system prompt is the MOST IMPORTANT part of the agent. It is the only -# way the LLM knows what tools it has and how to use them. -# -# Think about it: the LLM is just a text model. It has no built-in ability -# to read files or run code. The system prompt is where we TELL it: -# "You have these tools. When you want to use one, output this JSON format. -# I (the code) will parse your JSON, run the tool, and give you the result." -# -# The LLM then "plays along" — it outputs JSON that LOOKS LIKE a tool call, -# and our agent loop code makes it ACTUALLY happen. - -# TODO 1: Complete the TOOL_DESCRIPTIONS string below. -# This text will be embedded in the system prompt inside a section. -# The LLM needs to know: -# - The name of each tool (must match the keys in TOOL_FUNCTIONS above!) -# - What arguments each tool takes -# - What each tool does -# -# Four tools are already described for you as examples. -# Add the missing three: write_file, run_python, validate_python. -# -# Follow the same format: -# - tool_name({"param": ""}): What the tool does. - -TOOL_DESCRIPTIONS = """\ - - read_file({"path": ""}): Read a .py or .txt file from the workspace. - - grep_search({"pattern": "", "file_glob": ""}): Search for a pattern in files. - - list_files({"file_glob": ""}): List files matching the pattern. - - write_file({"path": "", "content": ""}): Write a .py or .txt file to the workspace. - - run_python({"path": ""}): Execute a Python file and return stdout + stderr. - - validate_python({"path": ""}): Check Python file syntax and return result. - - done({"summary": ""}): Signal that you are finished. -""" - -# TODO 2: Complete the system prompt. -# The structure is provided — fill in the and sections. -# -# For , describe these steps: -# 1. PLAN: Think about what steps are needed. List them in "thought". -# 2. ACT: Choose ONE tool to call. -# 3. OBSERVE: Analyse the tool's output carefully. -# 4. REPLAN: If the result was unexpected, revise your plan. -# 5. REPEAT: Go back to ACT if more work is needed. -# 6. DONE: Call the "done" tool when the task is complete. -# -# For , include at least: -# - Always plan before acting. -# - Call exactly ONE tool per response. -# - After writing code, always validate and run it. -# - If an error occurs, try to fix it (up to 3 retries). -# - Stay within the workspace directory. -# - When finished, call the "done" tool. -# -# IMPORTANT: The JSON example uses {{ and }} because this is an f-string. -# In an f-string, {{ produces a literal { in the output. -# So {{"thought": "..."}} becomes {"thought": "..."} when printed. - -SYSTEM_PROMPT = f"""\ -You are a coding agent that helps users with Python programming tasks. -You work inside a workspace directory and have access to tools. - - -Available tools: -{TOOL_DESCRIPTIONS} - - - -To accomplish a task, follow this workflow: - 1. PLAN: Think about what steps are needed. List them in "thought". - 2. ACT: Choose ONE tool to call. - 3. OBSERVE: Analyse the tool's output carefully. - 4. REPLAN: If the result was unexpected, revise your plan. - 5. REPEAT: Go back to ACT if more work is needed. - 6. DONE: Call the "done" tool when the task is complete. - - - -You MUST respond with a JSON object every time. The format is: -{{{{ - "thought": "", - "tool": "", - "arguments": {{{{ }}}} -}}}} - -Example — to read a file: -{{{{ - "thought": "I need to read app.py to understand the code.", - "tool": "read_file", - "arguments": {{{{"path": "app.py"}}}} -}}}} - -Example — to signal completion: -{{{{ - "thought": "I have fixed all the bugs and verified the code runs.", - "tool": "done", - "arguments": {{{{"summary": "Fixed 3 bugs in app.py and verified all tests pass."}}}} -}}}} - - - -Rules for operating as a coding agent: -- Always plan before acting. Your "thought" should explain your reasoning and strategy. -- Call exactly ONE tool per response. Do not try to call multiple tools. -- Always validate Python code after writing it using validate_python. -- Always run Python code after validation to verify it works using run_python. -- If an error occurs, analyze it carefully and retry up to 3 times. -- Stay within the workspace directory. Never try to access files outside it. -- When the task is complete, immediately call the "done" tool with a summary of what was accomplished. -- If a human provides feedback in tags, acknowledge it and adjust your plan accordingly. - -""" - - -# ═══════════════════════════════════════════════════════════════════════════════ -# PART D -- AGENT LOOP (TODOs 3-6) -# ═══════════════════════════════════════════════════════════════════════════════ -# -# This is where everything comes together. The agent loop: -# -# 1. Sends messages to the LLM (including the system prompt with tools). -# 2. The LLM responds with JSON like: {"tool": "read_file", "arguments": {"path": "app.py"}} -# 3. We parse that JSON and call the real Python function. -# 4. We put the result back into the conversation as a new message. -# 5. We send the updated conversation to the LLM again. -# 6. The LLM sees the result and decides what to do next. -# 7. Repeat until the LLM calls "done" or we hit the iteration limit. - - -def truncate_result(result: str) -> str: - """Truncate a tool result if it exceeds MAX_RESULT_LENGTH.""" - if len(result) <= MAX_RESULT_LENGTH: - return result - half = MAX_RESULT_LENGTH // 2 - return ( - result[:half] - + f"\n\n... [TRUNCATED — {len(result)} chars total, showing first and last {half}] ...\n\n" - + result[-half:] - ) - - -def trim_messages(messages: list) -> list: - """Trim older messages if total character count exceeds MAX_HISTORY_CHARS.""" - total = sum(len(m["content"]) for m in messages) - if total <= MAX_HISTORY_CHARS: - return messages - head = messages[:2] - tail = messages[2:] - original_task = messages[1]["content"] if len(messages) > 1 else "" - while tail and sum(len(m["content"]) for m in head + tail) > MAX_HISTORY_CHARS: - tail.pop(0) - reminder = { - "role": "user", - "content": ( - "Earlier conversation history was trimmed. " - f"REMINDER — your original task was:\n{original_task}\n" - "Continue from where you left off." - ), - } - return head + [reminder] + tail - - -def ask_human() -> str: - """Ask the user to approve, redirect, or stop before each action.""" - try: - reply = input( - "\n [Enter]=continue, or type a comment (stop to abort): " - ).strip() - return reply - except (EOFError, KeyboardInterrupt): - return "stop" - - -def agent_loop(user_task: str) -> None: - """Run the agent loop: plan -> user review -> act -> observe -> repeat. - - Study this function carefully — it IS the agent. Everything else is - just support. The loop implements this cycle: - - LLM produces JSON → we parse it → we call the tool → - we feed the result back → LLM produces next JSON → ... - """ - - # TODO 3: Initialise the message list. - # Create a list with two messages: - # 1. {"role": "system", "content": SYSTEM_PROMPT} - # 2. {"role": "user", "content": user_task} - # - # The system message teaches the LLM about its tools. - # The user message is the task to accomplish. - messages = [ - {"role": "system", "content": SYSTEM_PROMPT}, - {"role": "user", "content": user_task}, - ] - - for iteration in range(1, MAX_ITERATIONS + 1): - print_separator(f"Agent Iteration {iteration}") - messages = trim_messages(messages) - - # TODO 4: Get the LLM's next action. - try: - raw = chat_json(client, messages, temperature=0.2, max_tokens=4096) - action = json.loads(raw) - thought = action.get("thought", "") - tool_name = action.get("tool", "") - arguments = action.get("arguments", {}) - except json.JSONDecodeError as e: - print(f"Failed to parse JSON response: {e}") - messages.append({"role": "assistant", "content": raw}) - messages.append( - { - "role": "user", - "content": "Please respond with valid JSON in the specified format.", - } - ) - continue - - # Print the agent's plan - print(f"\nThought: {thought}") - print(f"Tool: {tool_name}") - print(f"Arguments: {arguments}") - - # TODO 5: Human-in-the-loop — let the user review before execution. - if tool_name == "done": - print( - f"\n✓ Agent proposed completion: {arguments.get('summary', 'Task completed')}" - ) - user_input = ask_human() - if user_input.lower() in {"stop", "abort", "cancel"}: - print("Aborted by user.") - return - elif user_input: - messages.append({"role": "assistant", "content": raw}) - messages.append( - { - "role": "user", - "content": f"{user_input}\nPlease revise your plan based on this feedback.", - } - ) - continue - # If user approved (pressed Enter), fall through to execute - else: - user_input = ask_human() - if user_input.lower() in {"stop", "abort", "cancel"}: - print("Agent aborted by user.") - return - elif user_input: - # User provided feedback - don't execute, ask to revise - messages.append({"role": "assistant", "content": raw}) - messages.append( - { - "role": "user", - "content": f"{user_input}\nPlease revise your plan based on this feedback.", - } - ) - continue - - # TODO 6: Execute the tool and feed the result back. - if tool_name == "done": - print(f"\n✓ Agent completed: {arguments.get('summary', 'Task completed')}") - return - - # Call the tool - result = dispatch_tool(tool_name, arguments) - result = truncate_result(result) - - # Append the assistant's response and tool result to the conversation - messages.append({"role": "assistant", "content": raw}) - messages.append( - { - "role": "user", - "content": f'\n{result}\n', - } - ) - - # Print result for debugging - print( - f"\nResult: {result[:200]}..." - if len(result) > 200 - else f"\nResult: {result}" - ) - - print_separator("Agent stopped (max iterations reached)") - - -# ═══════════════════════════════════════════════════════════════════════════════ -# PART E -- INTERACTIVE CHAT (TODOs 7-8) -# ═══════════════════════════════════════════════════════════════════════════════ - -# TODO 7: Implement the input loop. -# - Read input with: user_input = input("You> ").strip() -# - Handle EOFError and KeyboardInterrupt (Ctrl+C) -# - Skip empty input -# - Exit on "quit" or "exit" -# - Otherwise call agent_loop(user_input) - - -def interactive_chat(): - """Run an interactive chat loop where the user gives tasks to the agent.""" - print_separator("AI Coding Agent -- Interactive Mode") - print("Type your task and press Enter. Type 'quit' or 'exit' to stop.") - print(f"Workspace: {WORKSPACE.resolve()}\n") - - # Show what files are in the workspace - files = [f for f in sorted(WORKSPACE.glob("*")) if f.is_file()] - if files: - print("Files in workspace:") - for f in files: - print(f" {f.name}") - else: - print("Workspace is empty.") - print() - - # TODO 7: Implement the input loop. - while True: - try: - user_input = input("You> ").strip() - except (EOFError, KeyboardInterrupt): - print("\nExiting...") - return - - # Skip empty input - if not user_input: - continue - - # Check for exit commands - if user_input.lower() in {"quit", "exit"}: - print("Exiting interactive chat.") - return - - # Run the agent with the user's task - agent_loop(user_input) - print() - - -# ═══════════════════════════════════════════════════════════════════════════════ -# MAIN -# ═══════════════════════════════════════════════════════════════════════════════ - -if __name__ == "__main__": - source = Path(__file__).parent / "analyze_me.py" - dest = WORKSPACE / "analyze_me.py" - if not dest.exists(): - dest.write_text(source.read_text()) - interactive_chat() diff --git a/backend/agent/tools.py b/backend/agent/tools.py deleted file mode 100644 index e69de29..0000000 diff --git a/run_agent.py b/run_agent.py new file mode 100644 index 0000000..4e5e26e --- /dev/null +++ b/run_agent.py @@ -0,0 +1,78 @@ +""" +Temporäres Test-Script für den CodingAgent – kann danach gelöscht werden. + +Ausführen: + python run_agent.py + +Steuerung: + Enter → Aktion ausführen (approve) + Text + Enter → Feedback geben (reject + replan) + stop → Abbrechen +""" + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) + +from backend.agent.coding_agent import CodingAgent, WORKSPACE + + +def run(): + print("\n" + "=" * 60) + print(" CodingAgent – Interaktiver Test") + print("=" * 60) + print(f" Workspace: {WORKSPACE}") + print(" [Enter] = Aktion ausführen | Text = Feedback | 'stop' = Abbruch") + print("=" * 60 + "\n") + + task = input("Aufgabe eingeben: ").strip() + if not task: + print("Keine Aufgabe eingegeben. Beende.") + return + + agent = CodingAgent() + agent.start_task(task) + print(f"\nAgent gestartet für: '{task}'\n") + + step = 0 + while not agent.is_done: + step += 1 + print(f"\n{'─' * 60}") + print(f" Schritt {step} – Agent überlegt...") + + action = agent.propose_next_action() + + print(f"\n Thought : {action.get('thought', '')}") + print(f" Tool : {action.get('tool', '')}") + print(f" Arguments: {action.get('arguments', {})}") + print() + + user_input = input(" [Enter]=ausführen | Text=Feedback | stop=Abbruch: ").strip() + + if user_input.lower() in ("stop", "abort"): + print("\nAbgebrochen.") + break + + if user_input: + agent.reject(user_input) + print(f" → Feedback injiziert. Agent plant neu.\n") + continue + + result = agent.approve() + + print(f"\n Resultat ({result['tool']}):") + print(f" {result['result'][:300]}{'...' if len(result['result']) > 300 else ''}") + + if result["is_done"]: + print("\n" + "=" * 60) + print(" FERTIG!") + print(f" {result['result']}") + print("=" * 60) + + +if __name__ == "__main__": + try: + run() + except KeyboardInterrupt: + print("\n\nUnterbrochen.") diff --git a/tests/test_coding_agent.py b/tests/test_coding_agent.py new file mode 100644 index 0000000..6d903e3 --- /dev/null +++ b/tests/test_coding_agent.py @@ -0,0 +1,573 @@ +""" +Tests for CodingAgent (backend/agent/coding_agent.py) + +Structure: + TestHelpers – truncate_result, trim_messages, _strip_code_fences + TestDispatcher – dispatch_tool routing + TestTools – tool functions (read_file, write_file, …) using tmp workspace + TestCodingAgentInit – __init__ and start_task + TestProposeNextAction – propose_next_action with mocked API + TestApprove – approve with mocked API + real tool execution + TestReject – reject injects feedback correctly + TestFullLoop – integration: real API, skipped if unreachable +""" + +import json +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from backend.agent.coding_agent import ( + MAX_HISTORY_CHARS, + MAX_RESULT_LENGTH, + CodingAgent, + _strip_code_fences, + dispatch_tool, + done, + grep_search, + list_files, + read_file, + run_python, + truncate_result, + trim_messages, + validate_python, + write_file, +) + + +# ───────────────────────────────────────────────────────────────────────────── +# Helpers +# ───────────────────────────────────────────────────────────────────────────── + +def _make_api_response(content: str, status_code: int = 200): + """Build a mock requests.Response that returns *content* as the AI message.""" + mock = MagicMock() + mock.status_code = status_code + mock.json.return_value = { + "choices": [{"message": {"role": "assistant", "content": content}}] + } + mock.text = "error body" + return mock + + +def _agent_action_json(tool: str, thought: str = "thinking...", **arguments) -> str: + return json.dumps({"thought": thought, "tool": tool, "arguments": arguments}) + + +# ═════════════════════════════════════════════════════════════════════════════ +# TestHelpers +# ═════════════════════════════════════════════════════════════════════════════ + +class TestTruncateResult: + + def test_short_result_unchanged(self): + assert truncate_result("hello") == "hello" + + def test_long_result_is_truncated(self): + long = "x" * (MAX_RESULT_LENGTH + 100) + result = truncate_result(long) + assert len(result) < len(long) + assert "TRUNCATED" in result + + def test_exact_limit_not_truncated(self): + text = "a" * MAX_RESULT_LENGTH + assert truncate_result(text) == text + + def test_truncated_keeps_start_and_end(self): + text = "START" + "x" * MAX_RESULT_LENGTH + "END" + result = truncate_result(text) + assert "START" in result + assert "END" in result + + +class TestTrimMessages: + + def _make_messages(self, n_extra: int, chars_each: int = 100) -> list: + msgs = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "original task"}, + ] + for i in range(n_extra): + msgs.append({"role": "assistant", "content": "x" * chars_each}) + msgs.append({"role": "user", "content": "y" * chars_each}) + return msgs + + def test_short_history_unchanged(self): + msgs = self._make_messages(2) + assert trim_messages(msgs) == msgs + + def test_long_history_is_trimmed(self): + msgs = self._make_messages(n_extra=500, chars_each=200) + original_total = sum(len(m["content"]) for m in msgs) + trimmed = trim_messages(msgs) + trimmed_total = sum(len(m["content"]) for m in trimmed) + # Must be significantly shorter than the original + # (slightly above MAX_HISTORY_CHARS is acceptable due to the injected reminder message) + assert trimmed_total < original_total + assert len(trimmed) < len(msgs) + + def test_system_message_always_kept(self): + msgs = self._make_messages(n_extra=500, chars_each=200) + trimmed = trim_messages(msgs) + assert trimmed[0]["role"] == "system" + + def test_original_task_always_kept(self): + msgs = self._make_messages(n_extra=500, chars_each=200) + trimmed = trim_messages(msgs) + assert trimmed[1]["content"] == "original task" + + def test_reminder_injected_when_trimmed(self): + msgs = self._make_messages(n_extra=500, chars_each=200) + trimmed = trim_messages(msgs) + contents = [m["content"] for m in trimmed] + assert any("system_note" in c for c in contents) + + +class TestStripCodeFences: + + def test_plain_text_unchanged(self): + assert _strip_code_fences("hello") == "hello" + + def test_removes_json_fence(self): + text = "```json\n{\"key\": 1}\n```" + assert _strip_code_fences(text) == '{"key": 1}' + + def test_removes_plain_fence(self): + text = "```\nhello\n```" + assert _strip_code_fences(text) == "hello" + + def test_strips_whitespace(self): + assert _strip_code_fences(" hello ") == "hello" + + +# ═════════════════════════════════════════════════════════════════════════════ +# TestDispatcher +# ═════════════════════════════════════════════════════════════════════════════ + +class TestDispatcher: + + def test_unknown_tool_returns_error(self): + result = dispatch_tool("nonexistent_tool", {}) + assert "ERROR" in result + assert "nonexistent_tool" in result + + def test_done_tool_dispatched(self): + result = dispatch_tool("done", {"summary": "finished"}) + assert "finished" in result + + def test_wrong_arguments_returns_error(self): + result = dispatch_tool("read_file", {"wrong_param": "x"}) + assert "ERROR" in result + + +# ═════════════════════════════════════════════════════════════════════════════ +# TestTools (patched WORKSPACE → tmp_path) +# ═════════════════════════════════════════════════════════════════════════════ + +class TestWriteFile: + + def test_write_creates_file(self, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = write_file("hello.py", "print('hi')") + assert result.startswith("OK:") + assert (tmp_path / "hello.py").read_text() == "print('hi')" + + def test_write_outside_workspace_blocked(self, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = write_file("../evil.py", "bad") + assert "ERROR" in result + + def test_write_unsupported_extension_blocked(self, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = write_file("script.sh", "echo hi") + assert "ERROR" in result + + +class TestReadFile: + + def test_read_existing_file(self, tmp_path): + (tmp_path / "data.txt").write_text("hello world") + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = read_file("data.txt") + assert result == "hello world" + + def test_read_nonexistent_file(self, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = read_file("ghost.py") + assert "ERROR" in result + + def test_read_outside_workspace_blocked(self, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = read_file("../secret.py") + assert "ERROR" in result + + def test_read_unsupported_extension(self, tmp_path): + (tmp_path / "data.csv").write_text("a,b") + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = read_file("data.csv") + assert "ERROR" in result + + +class TestListFiles: + + def test_empty_workspace(self, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = list_files() + assert "No files" in result + + def test_lists_existing_files(self, tmp_path): + (tmp_path / "a.py").touch() + (tmp_path / "b.txt").touch() + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = list_files() + assert "a.py" in result + assert "b.txt" in result + + def test_glob_filter(self, tmp_path): + (tmp_path / "a.py").touch() + (tmp_path / "b.txt").touch() + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = list_files("*.py") + assert "a.py" in result + assert "b.txt" not in result + + +class TestGrepSearch: + + def test_finds_pattern(self, tmp_path): + (tmp_path / "code.py").write_text("def hello():\n pass\n") + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = grep_search("def hello") + assert "code.py" in result + assert "def hello" in result + + def test_no_match_returns_message(self, tmp_path): + (tmp_path / "code.py").write_text("x = 1\n") + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = grep_search("nonexistent_pattern") + assert "No matches" in result + + def test_returns_line_number(self, tmp_path): + (tmp_path / "code.py").write_text("x = 1\ndef foo():\n pass\n") + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = grep_search("def foo") + assert ":2:" in result + + +class TestValidatePython: + + def test_valid_syntax(self, tmp_path): + (tmp_path / "good.py").write_text("def f(x):\n return x * 2\n") + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = validate_python("good.py") + assert result == "OK: syntax is valid." + + def test_invalid_syntax(self, tmp_path): + (tmp_path / "bad.py").write_text("def f(x)\n return x\n") + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = validate_python("bad.py") + assert "SYNTAX ERROR" in result + + def test_file_not_found(self, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = validate_python("ghost.py") + assert "ERROR" in result + + +class TestRunPython: + + def test_successful_execution(self, tmp_path): + (tmp_path / "hello.py").write_text("print('hello world')\n") + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = run_python("hello.py") + assert "hello world" in result + assert "Exit code: 0" in result + + def test_runtime_error_captured(self, tmp_path): + (tmp_path / "bad.py").write_text("raise ValueError('oops')\n") + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = run_python("bad.py") + assert "ValueError" in result + assert "Exit code: 1" in result + + def test_file_not_found(self, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + result = run_python("ghost.py") + assert "ERROR" in result + + +# ═════════════════════════════════════════════════════════════════════════════ +# TestCodingAgentInit +# ═════════════════════════════════════════════════════════════════════════════ + +class TestCodingAgentInit: + + def test_initial_state_is_clean(self): + agent = CodingAgent() + assert agent.messages == [] + assert agent.pending_action is None + assert agent.is_done is False + assert agent.iteration == 0 + + def test_api_url_is_set(self): + agent = CodingAgent() + assert agent.api_url.startswith("http://") + assert "/v1/chat/completions" in agent.api_url + + def test_start_task_sets_messages(self): + agent = CodingAgent() + agent.start_task("Write fibonacci.py") + assert len(agent.messages) == 2 + assert agent.messages[0]["role"] == "system" + assert agent.messages[1]["role"] == "user" + assert "fibonacci" in agent.messages[1]["content"] + + def test_start_task_resets_state(self): + agent = CodingAgent() + agent.is_done = True + agent.iteration = 5 + agent.start_task("New task") + assert agent.is_done is False + assert agent.iteration == 0 + assert agent.pending_action is None + + def test_start_task_twice_resets_messages(self): + agent = CodingAgent() + agent.start_task("First task") + agent.start_task("Second task") + assert "Second task" in agent.messages[1]["content"] + assert len(agent.messages) == 2 + + +# ═════════════════════════════════════════════════════════════════════════════ +# TestProposeNextAction (mocked API) +# ═════════════════════════════════════════════════════════════════════════════ + +class TestProposeNextAction: + + @pytest.fixture + def agent(self): + a = CodingAgent() + a.start_task("Write hello.py") + return a + + def _mock_api(self, agent, tool="list_files", thought="planning", **args): + payload = _agent_action_json(tool, thought, **args) + agent._call_api = MagicMock(return_value=payload) + + def test_returns_dict_with_required_keys(self, agent): + self._mock_api(agent) + action = agent.propose_next_action() + assert "thought" in action + assert "tool" in action + assert "arguments" in action + + def test_increments_iteration(self, agent): + self._mock_api(agent) + agent.propose_next_action() + assert agent.iteration == 1 + + def test_stores_pending_action(self, agent): + self._mock_api(agent) + agent.propose_next_action() + assert agent.pending_action is not None + + def test_returns_correct_tool(self, agent): + self._mock_api(agent, tool="list_files") + action = agent.propose_next_action() + assert action["tool"] == "list_files" + + def test_handles_json_parse_error_gracefully(self, agent): + agent._call_api = MagicMock(return_value="this is not json {{") + action = agent.propose_next_action() + assert action["tool"] == "done" + + def test_handles_api_exception_gracefully(self, agent): + agent._call_api = MagicMock(side_effect=Exception("connection refused")) + action = agent.propose_next_action() + assert action["tool"] == "done" + + def test_strips_code_fences_from_response(self, agent): + payload = "```json\n" + _agent_action_json("list_files", "thinking") + "\n```" + agent._call_api = MagicMock(return_value=payload) + action = agent.propose_next_action() + assert action["tool"] == "list_files" + + def test_already_done_returns_done_action(self, agent): + agent.is_done = True + action = agent.propose_next_action() + assert action["tool"] == "done" + + +# ═════════════════════════════════════════════════════════════════════════════ +# TestApprove (mocked API + real tool execution via tmp_path) +# ═════════════════════════════════════════════════════════════════════════════ + +class TestApprove: + + @pytest.fixture + def agent(self): + a = CodingAgent() + a.start_task("Do something") + return a + + def _set_pending(self, agent, tool: str, **arguments): + raw = _agent_action_json(tool, "thought", **arguments) + agent.pending_action = { + "raw": raw, + "action": {"thought": "thought", "tool": tool, "arguments": arguments}, + } + + def test_approve_without_pending_raises(self, agent): + with pytest.raises(Exception): + agent.approve() + + def test_approve_done_sets_is_done(self, agent): + self._set_pending(agent, "done", summary="all done") + result = agent.approve() + assert result["is_done"] is True + assert agent.is_done is True + + def test_approve_done_returns_summary(self, agent): + self._set_pending(agent, "done", summary="finished successfully") + result = agent.approve() + assert "finished successfully" in result["result"] + + def test_approve_clears_pending_action(self, agent): + self._set_pending(agent, "done", summary="x") + agent.approve() + assert agent.pending_action is None + + def test_approve_appends_assistant_message(self, agent): + self._set_pending(agent, "done", summary="x") + before = len(agent.messages) + agent.approve() + assert len(agent.messages) > before + + def test_approve_tool_result_appended_to_messages(self, agent, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + self._set_pending(agent, "list_files") + agent.approve() + tool_results = [m for m in agent.messages if "tool_result" in m["content"]] + assert len(tool_results) == 1 + + def test_approve_error_result_adds_replan_tag(self, agent, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + self._set_pending(agent, "read_file", path="nonexistent.py") + agent.approve() + last_msg = agent.messages[-1]["content"] + assert "replan" in last_msg + + def test_approve_returns_tool_name_in_result(self, agent, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + self._set_pending(agent, "list_files") + result = agent.approve() + assert result["tool"] == "list_files" + assert result["is_done"] is False + + +# ═════════════════════════════════════════════════════════════════════════════ +# TestReject +# ═════════════════════════════════════════════════════════════════════════════ + +class TestReject: + + @pytest.fixture + def agent(self): + a = CodingAgent() + a.start_task("Do something") + return a + + def _set_pending(self, agent, tool="write_file"): + raw = _agent_action_json(tool, "thought") + agent.pending_action = { + "raw": raw, + "action": {"thought": "thought", "tool": tool, "arguments": {}}, + } + + def test_reject_clears_pending_action(self, agent): + self._set_pending(agent) + agent.reject("Use a different approach") + assert agent.pending_action is None + + def test_reject_appends_human_message(self, agent): + self._set_pending(agent) + agent.reject("Do it differently") + user_msgs = [m for m in agent.messages if m["role"] == "user"] + assert any("Do it differently" in m["content"] for m in user_msgs) + + def test_reject_adds_replan_tag(self, agent): + self._set_pending(agent) + agent.reject("feedback") + last = agent.messages[-1]["content"] + assert "replan" in last.lower() + + def test_reject_without_pending_does_not_crash(self, agent): + agent.pending_action = None + agent.reject("no pending action") # should not raise + + def test_reject_does_not_execute_tool(self, agent, tmp_path): + self._set_pending(agent, "write_file") + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + agent.reject("Do not write anything") + assert not list(tmp_path.glob("*")) # no files created + + +# ═════════════════════════════════════════════════════════════════════════════ +# TestFullLoop (integration – real API, skipped if unreachable) +# ═════════════════════════════════════════════════════════════════════════════ + +class TestFullLoop: + """End-to-end test: agent runs a real task against the live API. + Skipped automatically if the API is not reachable. + """ + + MAX_STEPS = 15 # safety limit for the test loop + + def _run_until_done(self, agent) -> list: + """Drive the agent loop until done or MAX_STEPS reached.""" + steps = [] + for _ in range(self.MAX_STEPS): + action = agent.propose_next_action() + result = agent.approve() + steps.append(result) + if result["is_done"]: + break + return steps + + def test_agent_completes_hello_world_task(self, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + agent = CodingAgent() + try: + agent.start_task( + "Write a Python file called hello.py that prints 'Hello World'. " + "Validate it and run it." + ) + steps = self._run_until_done(agent) + except Exception as e: + pytest.skip(f"API not reachable: {e}") + + assert agent.is_done, "Agent did not reach done state" + tools_used = [s["tool"] for s in steps] + assert "write_file" in tools_used + assert "done" in tools_used + + def test_agent_creates_file_on_disk(self, tmp_path): + with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): + agent = CodingAgent() + try: + agent.start_task("Write a file called output.txt containing the text 'test passed'.") + self._run_until_done(agent) + except Exception as e: + pytest.skip(f"API not reachable: {e}") + + py_files = list(tmp_path.glob("*.txt")) + list(tmp_path.glob("*.py")) + assert len(py_files) > 0, "Agent did not create any file" + + +if __name__ == "__main__": + pytest.main([__file__, "-v", "-s"])