meulilivio/AISE1_Project_Irina_Livio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'Agent Logik' (#5) from feature/backend_implementation into main

Browse Source 
Reviewed-on: meulilivio/AISE1_Project#5
This commit is contained in:
Livio Meuli 2026-04-09 16:45:50 +02:00
commit 1cc7d96e3a
5 changed files with 1141 additions and 605 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

490
backend/agent/coding_agent.py
View File

@ -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}": "<value>"')
else:
params.append(f'"{pname}": "<optional, default={param.default!r}>"')
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.
<capabilities>
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
</capabilities>
<tools>
{build_tool_description()}
</tools>
<workflow>
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.
</workflow>
<response_format>
You MUST respond with a JSON object every time:
{{
"thought": "<your reasoning about what to do next>",
"tool": "<tool name>",
"arguments": {{ <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): ..."}}
}}
</response_format>
<rules>
- 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 <human_message>, acknowledge it and adjust your plan.
- If you receive a <replan> tag, revise your plan before choosing the next tool.
</rules>
"""
# ═════════════════════════════════════════════════════════════════════════════
# 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": (
"<system_note>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.</system_note>"
),
}
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'<tool_result tool="{tool_name}">\n{result}\n</tool_result>'
if result.startswith("ERROR") or result.startswith("SYNTAX ERROR"):
feedback += (
"\n\n<replan>The 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.</replan>"
)
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"<human_message>{feedback}</human_message>\n"
"<replan>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.</replan>"
),
})
self.pending_action = None

605
backend/agent/coding_agent_example.py
View File

@ -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": "<system prompt>"}, │
│ {"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": │
│ "<tool_result>file contents │
│ </tool_result>"} │
└──────────────┬──────────────────┘
│
┌─────────▼─────────┐
│ 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 <tools> 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": "<description>"}): What the tool does.
TOOL_DESCRIPTIONS = """\
- read_file({"path": "<relative path>"}): Read a .py or .txt file from the workspace.
- grep_search({"pattern": "<text>", "file_glob": "<glob, default='*.py'>"}): Search for a pattern in files.
- list_files({"file_glob": "<glob, default='*'>"}): List files matching the pattern.
- write_file({"path": "<relative path>", "content": "<file content>"}): Write a .py or .txt file to the workspace.
- run_python({"path": "<relative path>"}): Execute a Python file and return stdout + stderr.
- validate_python({"path": "<relative path>"}): Check Python file syntax and return result.
- done({"summary": "<what you accomplished>"}): Signal that you are finished.
"""
# TODO 2: Complete the system prompt.
# The structure is provided — fill in the <workflow> and <rules> sections.
#
# For <workflow>, 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 <rules>, 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.
<tools>
Available tools:
{TOOL_DESCRIPTIONS}
</tools>
<workflow>
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.
</workflow>
<response_format>
You MUST respond with a JSON object every time. The format is:
{{{{
"thought": "<your reasoning about what to do next>",
"tool": "<tool name from the list above>",
"arguments": {{{{ <arguments for the tool> }}}}
}}}}
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."}}}}
}}}}
</response_format>
<rules>
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 <human_message> tags, acknowledge it and adjust your plan accordingly.
</rules>
"""
# ═══════════════════════════════════════════════════════════════════════════════
# 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": (
"<system_note>Earlier conversation history was trimmed. "
f"REMINDER — your original task was:\n{original_task}\n"
"Continue from where you left off.</system_note>"
),
}
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"<human_message>{user_input}</human_message>\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"<human_message>{user_input}</human_message>\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'<tool_result tool="{tool_name}">\n{result}\n</tool_result>',
}
)
# 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()

0
backend/agent/tools.py
View File

78
run_agent.py Normal file
View File

@ -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.")

573
tests/test_coding_agent.py Normal file
View File

@ -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"])