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

Merge branch 'main' of https://gitea.fhgr.ch/meulilivio/AISE1_Project into MCP-setup

Browse Source
This commit is contained in:
Irina Rueegg 2026-05-21 16:13:40 +02:00
commit a65dd047ba
22 changed files with 886 additions and 571 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
backend/agent/coding_agent.py
View File

@ -178,9 +178,12 @@ def trim_messages(messages: list) -> list:
tail = messages[2:] tail = messages[2:]
original_task = messages[1]["content"] if len(messages) > 1 else "" original_task = messages[1]["content"] if len(messages) > 1 else ""
# Drop the oldest messages first (index 2 onwards) until we are under the limit.
# The system prompt (0) and original task (1) are never dropped.
while tail and sum(len(m["content"]) for m in head + tail) > MAX_HISTORY_CHARS: while tail and sum(len(m["content"]) for m in head + tail) > MAX_HISTORY_CHARS:
tail.pop(0) tail.pop(0)
# Inject a reminder so the agent doesn't lose track of its goal after trimming.
reminder = { reminder = {
"role": "user", "role": "user",
"content": ( "content": (
@ -337,7 +340,7 @@ class CodingAgent:
payload = { payload = {
"model": self.model, "model": self.model,
"messages": messages, "messages": messages,
"temperature": 0.2, "temperature": 0.2, # low temperature → deterministic, more reliable tool calls
"max_tokens": 4096, "max_tokens": 4096,
"stream": False, "stream": False,
} }
@ -439,7 +442,9 @@ class CodingAgent:
result = await dispatch_tool(tool_name, arguments) result = await dispatch_tool(tool_name, arguments)
result = truncate_result(result) result = truncate_result(result)
# Build feedback – nudge agent to replan on errors # Wrap the tool output in an XML tag so the LLM can easily find it.
# Append a <replan> tag on errors to force the agent to reconsider
# its plan rather than blindly retrying the same failing action.
feedback = f'<tool_result tool="{tool_name}">\n{result}\n</tool_result>' feedback = f'<tool_result tool="{tool_name}">\n{result}\n</tool_result>'
if result.startswith("ERROR") or result.startswith("SYNTAX ERROR"): if result.startswith("ERROR") or result.startswith("SYNTAX ERROR"):
feedback += ( feedback += (

21
backend/managers/chat_manager.py
View File

@ -1,4 +1,4 @@
"""Chat Manager - Handles chat history and AI communication""" """Manages the chat history and communication with the AI model API."""
import os import os
from dotenv import load_dotenv from dotenv import load_dotenv
@ -9,6 +9,13 @@ load_dotenv()
class ChatManager: class ChatManager:
"""Handles sending messages and maintaining conversation history.
Connects to an OpenAI-compatible REST endpoint configured via environment
variables. All messages (user, assistant, system) are kept in memory so
the full conversation is sent with every request.
"""
def __init__(self): def __init__(self):
self.api_host = os.getenv("HOST") self.api_host = os.getenv("HOST")
self.api_port = os.getenv("PORT") self.api_port = os.getenv("PORT")
@ -22,15 +29,23 @@ class ChatManager:
self.chat_history = [] self.chat_history = []
def add_message(self, role: str, content: str) -> None: def add_message(self, role: str, content: str) -> None:
"""Append a single message to the conversation history."""
self.chat_history.append({"role": role, "content": content}) self.chat_history.append({"role": role, "content": content})
def get_history(self) -> list: def get_history(self) -> list:
return self.chat_history """Return a copy of the conversation history."""
return list(self.chat_history)
def clear_history(self) -> None: def clear_history(self) -> None:
"""Wipe the conversation history (starts a fresh chat)."""
self.chat_history = [] self.chat_history = []
def send_message(self, user_message: str) -> str: def send_message(self, user_message: str) -> str:
"""Send a user message to the AI and return its reply.
Adds the user message to history, calls the API with the full history
as context, and appends the AI reply to history before returning it.
"""
# Add user message to history # Add user message to history
self.add_message("user", user_message) self.add_message("user", user_message)
@ -44,6 +59,7 @@ class ChatManager:
if self.api_key and self.api_key != "EMPTY": if self.api_key and self.api_key != "EMPTY":
headers["Authorization"] = f"Bearer {self.api_key}" headers["Authorization"] = f"Bearer {self.api_key}"
# Full history is sent so the model has multi-turn conversation context
payload = { payload = {
"model": self.model, "model": self.model,
"messages": self.chat_history, "messages": self.chat_history,
@ -91,6 +107,7 @@ class ChatManager:
raise Exception(error_msg) raise Exception(error_msg)
def get_chat_display(self) -> list: def get_chat_display(self) -> list:
"""Return a copy of the history suitable for display in the UI."""
return [ return [
{"role": msg["role"], "content": msg["content"]} {"role": msg["role"], "content": msg["content"]}
for msg in self.chat_history for msg in self.chat_history

69
backend/managers/debug_logger.py
View File

@ -1,9 +1,66 @@
from datetime import datetime
class DebugLogger: class DebugLogger:
"""In-memory logger for code execution events.
Collects timestamped INFO and ERROR entries during a single run.
Call clear() before each new execution to start fresh.
"""
def __init__(self): def __init__(self):
self.logs: list[dict] = []
def log(self, message: str) -> None:
"""Append a general info message."""
self.logs.append({
"level": "INFO",
"message": message,
"timestamp": datetime.now().strftime("%H:%M:%S"),
})
def log_error(self, error_message: str) -> None:
"""Append an error message."""
self.logs.append({
"level": "ERROR",
"message": error_message,
"timestamp": datetime.now().strftime("%H:%M:%S"),
})
def get_logs(self) -> list[dict]:
"""Return a copy of all collected log entries."""
return list(self.logs)
def clear(self) -> None:
"""Reset the log — call before each new execution."""
self.logs = [] self.logs = []
def log(self, message): def format_debug_output(self, output: dict) -> str:
self.logs.append(message) """Format an ExecutionEngine result dict into a human-readable string.
def get_logs(self): Args:
return self.logs output: dict with keys 'stdout', 'stderr', and 'rc'.
Returns:
A formatted string ready for display in the UI.
"""
lines = []
status = "SUCCESS" if output.get("rc") == 0 else "FAILED"
lines.append(f"[{status}] Exit code: {output.get('rc')}")
if output.get("stdout"):
lines.append("\n--- stdout ---")
lines.append(output["stdout"].rstrip())
if output.get("stderr"):
lines.append("\n--- stderr ---")
lines.append(output["stderr"].rstrip())
if not output.get("stdout") and not output.get("stderr"):
lines.append("No output produced.")
for entry in self.logs:
lines.append(f"[{entry['timestamp']}] [{entry['level']}] {entry['message']}")
return "\n".join(lines)

32
backend/managers/execution_engine.py
View File

@ -1,19 +1,38 @@
import subprocess import subprocess
from pathlib import Path from pathlib import Path
RUN_TIMEOUT = 30 # seconds # Maximum time (seconds) a subprocess is allowed to run before being killed.
RUN_TIMEOUT = 30
class ExecutionEngine: class ExecutionEngine:
"""Runs files from the editor in a subprocess and returns the output.
Currently supports Python (.py) and LaTeX (.tex) files.
Returns a dict with keys: stdout, stderr, rc (return code).
"""
def __init__(self): def __init__(self):
pass pass
def run_code(self, active_file: Path) -> dict: def run_code(self, active_file: Path) -> dict:
"""Execute the given file and return its output.
Args:
active_file: Absolute path to the file that should be run.
Returns:
{"stdout": str, "stderr": str, "rc": int}
rc == 0 means success, anything else is an error.
"""
suffix = active_file.suffix suffix = active_file.suffix
current_dir = active_file.parent.resolve() current_dir = active_file.parent.resolve()
# Build the shell command depending on file type
if suffix == ".py": if suffix == ".py":
cmd = ["py", active_file.name] cmd = ["py", active_file.name]
elif suffix == ".tex": elif suffix == ".tex":
# pdflatex in non-interactive mode so it never waits for input
cmd = [ cmd = [
"pdflatex", "pdflatex",
"-interaction=nonstopmode", "-interaction=nonstopmode",
@ -22,20 +41,21 @@ class ExecutionEngine:
] ]
else: else:
return {"stdout": "", "stderr": f"Unsupported file type: {suffix}", "rc": 1} return {"stdout": "", "stderr": f"Unsupported file type: {suffix}", "rc": 1}
try: try:
proc = subprocess.run( proc = subprocess.run(
cmd, cmd,
cwd=current_dir, cwd=current_dir, # run inside the file's own directory
capture_output=True, capture_output=True,
text=True, text=True,
timeout=RUN_TIMEOUT, timeout=RUN_TIMEOUT,
) )
return {"stdout": proc.stdout, "stderr": proc.stderr, "rc": proc.returncode} return {"stdout": proc.stdout, "stderr": proc.stderr, "rc": proc.returncode}
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
return {"stdout": "", "stderr": f"Timed out after {RUN_TIMEOUT}s", "rc": -1} return {"stdout": "", "stderr": f"Timed out after {RUN_TIMEOUT}s", "rc": -1}
except FileNotFoundError as e: except FileNotFoundError as e:
# Raised when the interpreter/compiler binary is not found on PATH
return {"stdout": "", "stderr": str(e), "rc": -1} return {"stdout": "", "stderr": str(e), "rc": -1}
except Exception as e: except Exception as e:
return {"stdout": "", "stderr": str(e), "rc": -1} return {"stdout": "", "stderr": str(e), "rc": -1}

60
backend/managers/file_manager.py
View File

@ -1,6 +1,13 @@
"""Manages all file and folder operations inside the workspace directory.
Every method validates that the target path stays inside the workspace before
touching the filesystem, preventing path-traversal attacks.
"""
import streamlit as st import streamlit as st
from pathlib import Path from pathlib import Path
# The workspace folder is created at module load so it always exists.
WORKSPACE = Path("workspace") WORKSPACE = Path("workspace")
WORKSPACE.mkdir(exist_ok=True) WORKSPACE.mkdir(exist_ok=True)
@ -24,11 +31,12 @@ class FileManager:
if not name: if not name:
st.error(f"Invalid folder name: {name}") st.error(f"Invalid folder name: {name}")
return False return False
# Slashes in the name would silently create nested paths — reject them.
if "/" in name or "\\" in name: if "/" in name or "\\" in name:
st.error(f"Invalid folder name (no slashes allowed): {name}") st.error(f"Invalid folder name (no slashes allowed): {name}")
return False return False
name = Path(name) name = Path(name)
if relative_path: if relative_path:
relative_path = Path(relative_path) relative_path = Path(relative_path)
@ -37,10 +45,11 @@ class FileManager:
folder_path = (self.base_path / relative_path / name).resolve() folder_path = (self.base_path / relative_path / name).resolve()
# Ensure the resolved path is still inside the workspace (prevents path traversal).
if not str(folder_path).startswith(str(self.base_path.resolve())): if not str(folder_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}") st.error(f"Access denied: {relative_path}")
return False return False
try: try:
folder_path.mkdir(exist_ok=False) folder_path.mkdir(exist_ok=False)
return True return True
@ -66,11 +75,11 @@ class FileManager:
if not name or name.strip() == "" : if not name or name.strip() == "" :
st.error(f"Invalid file name: {name}") st.error(f"Invalid file name: {name}")
return False return False
name = Path(name) name = Path(name)
if not name.suffix: if not name.suffix:
name = name.with_suffix(".txt") # Default to .txt if no extension provided name = name.with_suffix(".txt") # Default to .txt if no extension provided
if relative_path: if relative_path:
relative_path = Path(relative_path) relative_path = Path(relative_path)
else: else:
@ -78,10 +87,11 @@ class FileManager:
file_path = (self.base_path / relative_path / name).resolve() file_path = (self.base_path / relative_path / name).resolve()
# Ensure the resolved path is still inside the workspace (prevents path traversal).
if not str(file_path).startswith(str(self.base_path.resolve())): if not str(file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}") st.error(f"Access denied: {relative_path}")
return False return False
try: try:
file_path.touch(exist_ok=False) file_path.touch(exist_ok=False)
return True return True
@ -95,10 +105,11 @@ class FileManager:
def read_file(self, relative_path: Path) -> str: def read_file(self, relative_path: Path) -> str:
""" """
Reads the content of a file. Reads the content of a file.
The relative_path should be the path to the file relative to the base path. Accepts an absolute Path object (as stored in st.session_state.open_files).
The path is validated to ensure it stays inside the workspace.
Args: Args:
relative_path (str): The relative path (without base path) to the file to read, including the file name relative_path (Path): Absolute path to the file to read.
Returns: Returns:
str: The content of the file, or an empty string if there was an error. str: The content of the file, or an empty string if there was an error.
""" """
@ -110,10 +121,11 @@ class FileManager:
if not file_path.is_file(): if not file_path.is_file():
st.error(f"Path is not a file: {relative_path}") st.error(f"Path is not a file: {relative_path}")
return "" return ""
# Ensure the resolved path is still inside the workspace (prevents path traversal).
if not str(file_path).startswith(str(self.base_path.resolve())): if not str(file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}") st.error(f"Access denied: {relative_path}")
return "" return ""
try: try:
with open(file_path, "r") as f: with open(file_path, "r") as f:
return f.read() return f.read()
@ -126,21 +138,23 @@ class FileManager:
def save_file(self, relative_path: str, content: str) -> bool: def save_file(self, relative_path: str, content: str) -> bool:
""" """
Saves content to a file. Saves content to a file.
The relative_path should be the path to the file relative to the base path. Accepts an absolute path string (as stored in st.session_state.open_files).
The path is validated to ensure it stays inside the workspace.
Args: Args:
relative_path (str): The relative path(without base path) to the file to save, including the file name relative_path (str): Absolute path to the file to save, including the file name.
content (str): The content to write to the file content (str): The content to write to the file.
Returns: Returns:
bool: True if save was successful, False otherwise. bool: True if save was successful, False otherwise.
""" """
file_path = (Path(relative_path)).resolve() file_path = (Path(relative_path)).resolve()
# Ensure the resolved path is still inside the workspace (prevents path traversal).
if not str(file_path).startswith(str(self.base_path.resolve())): if not str(file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}") st.error(f"Access denied: {relative_path}")
return False return False
try: try:
with open(file_path, "w") as f: with open(file_path, "w") as f:
f.write(content) f.write(content)
@ -148,7 +162,7 @@ class FileManager:
except Exception as e: except Exception as e:
st.error(f"Error saving file {relative_path}: {str(e)}") st.error(f"Error saving file {relative_path}: {str(e)}")
return False return False
def rename_file(self, old_relative_path: str, new_name: str) -> bool: def rename_file(self, old_relative_path: str, new_name: str) -> bool:
""" """
Renames a file while keeping the same extension. Renames a file while keeping the same extension.
@ -163,20 +177,22 @@ class FileManager:
if not new_name or new_name.strip() == "": if not new_name or new_name.strip() == "":
st.error(f"Invalid file name: {new_name}") st.error(f"Invalid file name: {new_name}")
return False return False
file_type = Path(old_relative_path).suffix file_type = Path(old_relative_path).suffix
new_name = Path(new_name) new_name = Path(new_name)
# Force the original extension so the file type cannot be changed by renaming.
if not Path(new_name).suffix == file_type: if not Path(new_name).suffix == file_type:
new_name = Path(new_name).with_suffix(file_type) # Ensure the file extension remains the same new_name = Path(new_name).with_suffix(file_type) # Ensure the file extension remains the same
old_file_path = (Path(self.base_path / old_relative_path)).resolve() old_file_path = (Path(self.base_path / old_relative_path)).resolve()
new_file_path = old_file_path.parent / new_name new_file_path = old_file_path.parent / new_name
# Both old and new paths must stay inside the workspace.
if not str(old_file_path).startswith(str(self.base_path.resolve())) or not str(new_file_path).startswith(str(self.base_path.resolve())): if not str(old_file_path).startswith(str(self.base_path.resolve())) or not str(new_file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {old_relative_path}") st.error(f"Access denied: {old_relative_path}")
return False return False
try: try:
old_file_path.rename(new_file_path) old_file_path.rename(new_file_path)
return True return True
@ -196,6 +212,7 @@ class FileManager:
""" """
folder_path = (self.base_path / relative_path).resolve() folder_path = (self.base_path / relative_path).resolve()
# Ensure the resolved path is still inside the workspace (prevents path traversal).
if not str(folder_path).startswith(str(self.base_path.resolve())): if not str(folder_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}") st.error(f"Access denied: {relative_path}")
return False return False
@ -223,7 +240,6 @@ class FileManager:
""" """
file_path = Path(relative_path) file_path = Path(relative_path)
abs_file_path = (Path(self.base_path) / file_path).resolve() abs_file_path = (Path(self.base_path) / file_path).resolve()
print(f"Absolute file path resolved to: {abs_file_path}") # Debugging info
if not str(abs_file_path).startswith(str(self.base_path.resolve())): if not str(abs_file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}") st.error(f"Access denied: {relative_path}")
@ -254,11 +270,11 @@ class FileManager:
for item in sorted(path.iterdir()): for item in sorted(path.iterdir()):
if item.is_dir(): if item.is_dir():
tree[item.name] = build_tree(item) tree[item.name] = build_tree(item) # recurse into sub-folders
else: else:
tree[item.name] = None tree[item.name] = None # leaf node for files
return tree return tree
return build_tree(self.base_path) return build_tree(self.base_path)
if __name__ == "__main__": if __name__ == "__main__":
FileManager() FileManager()

20
backend/managers/system_prompter.py
View File

@ -1,18 +1,26 @@
"""System Prompter - Builds system prompts with optional file context""" """Builds the system prompt that is sent to the AI at the start of each chat session."""
MAX_FILE_CHARS = 4000 # Limit file context to avoid token overflow # Prevents very large files from flooding the context window with tokens.
MAX_FILE_CHARS = 4000
class SystemPrompter: class SystemPrompter:
"""Generates system prompts for the chat assistant.
When a file is open in the editor it can be embedded in the prompt so the
AI has direct context of the code the user is currently working on.
"""
@staticmethod @staticmethod
def generate_prompt(file_context: dict | None = None) -> str: def generate_prompt(file_context: dict | None = None) -> str:
"""Build a system prompt, optionally embedding a file's content. """Build a system prompt, optionally embedding a file's content.
Args: Args:
file_context: dict with keys 'name' and 'content', or None. file_context: dict with keys 'name' (filename) and 'content' (raw text),
or None if no file should be included.
Returns: Returns:
A system prompt string. A ready-to-use system prompt string.
""" """
base = ( base = (
"You are an expert code assistant integrated into a lightweight code editor. " "You are an expert code assistant integrated into a lightweight code editor. "
@ -23,9 +31,11 @@ class SystemPrompter:
if file_context: if file_context:
name = file_context.get("name", "unknown") name = file_context.get("name", "unknown")
content = file_context.get("content", "") content = file_context.get("content", "")
# Truncate large files to avoid exceeding token limits
# Truncate large files to avoid exceeding the model's token limit
if len(content) > MAX_FILE_CHARS: if len(content) > MAX_FILE_CHARS:
content = content[:MAX_FILE_CHARS] + "\n... [truncated]" content = content[:MAX_FILE_CHARS] + "\n... [truncated]"
file_section = ( file_section = (
f"\n\nThe user currently has the following file open in the editor:\n" f"\n\nThe user currently has the following file open in the editor:\n"
f"<file name=\"{name}\">\n" f"<file name=\"{name}\">\n"

21
frontend/app.py
View File

@ -1,8 +1,20 @@
"""Entry point for the Streamlit app.
Runs with: streamlit run frontend/app.py
Responsibilities:
- Configure the page layout
- Inject global CSS tweaks
- Render the sidebar (navigation + file explorer)
- Delegate to the correct view (Chat or Code Editor) based on the radio selection
"""
import streamlit as st import streamlit as st
import sys import sys
from pathlib import Path from pathlib import Path
# Add project root to Python path for imports # Add the project root to sys.path so backend imports work regardless of
# where streamlit is launched from.
sys.path.insert(0, str(Path(__file__).parent.parent)) sys.path.insert(0, str(Path(__file__).parent.parent))
from frontend.sidebar import render_sidebar from frontend.sidebar import render_sidebar
@ -10,12 +22,16 @@ from frontend.editor import render_editor
from frontend.chat import render_chat from frontend.chat import render_chat
from frontend.state import init_state from frontend.state import init_state
# Initialise all session-state keys before any widget is rendered
init_state() init_state()
def main(): def main():
st.set_page_config(page_title="Lightweight code editor", layout="wide") st.set_page_config(page_title="Lightweight code editor", layout="wide")
# Small spacing corrections applied globally:
# - Reduce the default top padding of the main content area
# - Pull the sidebar content up so the logo sits at the very top
st.markdown( st.markdown(
""" """
<style> <style>
@ -28,14 +44,17 @@ def main():
st.title("Lightweight code editor") st.title("Lightweight code editor")
# Re-run init_state to cover any keys that might have been missed on cold start
init_state() init_state()
render_sidebar() render_sidebar()
# Switch between the two main views based on the sidebar radio button
if st.session_state.get("radio_interface_options") == "Code Editor": if st.session_state.get("radio_interface_options") == "Code Editor":
render_editor() render_editor()
elif st.session_state.get("radio_interface_options") == "Chat with AI Assistant": elif st.session_state.get("radio_interface_options") == "Chat with AI Assistant":
render_chat() render_chat()
if __name__ == "__main__": if __name__ == "__main__":
main() main()

51
frontend/chat.py
View File

@ -1,3 +1,5 @@
"""Chat view — renders both the normal chat interface and the Coding Agent mode."""
import streamlit as st import streamlit as st
from backend.managers.chat_manager import ChatManager from backend.managers.chat_manager import ChatManager
from backend.managers.system_prompter import SystemPrompter from backend.managers.system_prompter import SystemPrompter
@ -16,8 +18,10 @@ def _run_async(coro):
return loop.run_until_complete(coro) return loop.run_until_complete(coro)
def _start_agent(task: str): def _start_agent(task: str):
"""Initialise a fresh CodingAgent, start the task, """Create a new CodingAgent, feed it the task, and propose the first action.
and propose the first action.""" Stores the agent and its state in session_state so Streamlit can reference
them across reruns without losing progress.
"""
from backend.agent.coding_agent import CodingAgent from backend.agent.coding_agent import CodingAgent
agent = CodingAgent() agent = CodingAgent()
agent.start_task(task) agent.start_task(task)
@ -29,13 +33,13 @@ def _start_agent(task: str):
def _approve_action(): def _approve_action():
"""Execute the pending action, """Execute the pending action, log it, then immediately propose the next step."""
append it to the log, then propose the next step."""
agent = st.session_state.coding_agent agent = st.session_state.coding_agent
pending = st.session_state.agent_pending_action pending = st.session_state.agent_pending_action
result = _run_async(agent.approve()) result = _run_async(agent.approve())
# Append a record to the log so the user can review every completed step.
st.session_state.agent_log.append({ st.session_state.agent_log.append({
"thought": pending.get("thought", ""), "thought": pending.get("thought", ""),
"tool": result["tool"], "tool": result["tool"],
@ -44,6 +48,7 @@ def _approve_action():
}) })
if result["is_done"]: if result["is_done"]:
# Agent called the "done" tool — task is fully complete.
st.session_state.agent_status = "done" st.session_state.agent_status = "done"
st.session_state.agent_pending_action = None st.session_state.agent_pending_action = None
else: else:
@ -53,7 +58,10 @@ def _approve_action():
def _reject_action(feedback: str): def _reject_action(feedback: str):
"""Reject the pending action with optional feedback, then replan.""" """Reject the pending action with feedback so the agent replans.
The pending action is discarded; the agent receives the user's feedback and
proposes a different approach on the next call to propose_next_action().
"""
agent = st.session_state.coding_agent agent = st.session_state.coding_agent
agent.reject(feedback or "Please try a different approach.") agent.reject(feedback or "Please try a different approach.")
next_action = _run_async(agent.propose_next_action()) next_action = _run_async(agent.propose_next_action())
@ -62,7 +70,7 @@ def _reject_action(feedback: str):
def _followup_agent(question: str): def _followup_agent(question: str):
"""Inject a follow-up question into the finished agent and resume the loop.""" """Continue a finished task by injecting a follow-up question and resuming the loop."""
agent = st.session_state.coding_agent agent = st.session_state.coding_agent
agent.follow_up(question) agent.follow_up(question)
action = _run_async(agent.propose_next_action()) action = _run_async(agent.propose_next_action())
@ -71,7 +79,7 @@ def _followup_agent(question: str):
def _reset_agent(): def _reset_agent():
"""Reset all agent state back to idle.""" """Clear all agent state and return to the idle (task input) screen."""
st.session_state.coding_agent = None st.session_state.coding_agent = None
st.session_state.agent_status = "idle" st.session_state.agent_status = "idle"
st.session_state.agent_log = [] st.session_state.agent_log = []
@ -81,13 +89,21 @@ def _reset_agent():
# ── Agent Mode UI ───────────────────────────────────────────────────────────── # ── Agent Mode UI ─────────────────────────────────────────────────────────────
def render_agent_mode(): def render_agent_mode():
# Toggle must always be rendered so Streamlit keeps agent_mode=True in session_state """Render the step-by-step agent UI.
Three distinct screens based on agent_status:
- "idle" → task description input + Start button
- "waiting_approval" → show proposed action, Approve / Reject / Abort
- "done" → success message, follow-up input, New Task button
"""
# The toggle must always render so Streamlit keeps agent_mode=True in session_state.
st.toggle("Agent Mode", key="agent_mode") st.toggle("Agent Mode", key="agent_mode")
agent_status = st.session_state.get("agent_status", "idle") agent_status = st.session_state.get("agent_status", "idle")
agent_log = st.session_state.get("agent_log", []) agent_log = st.session_state.get("agent_log", [])
# ── Agent Log ──────────────────────────────────────────────────────────── # ── Agent Log ────────────────────────────────────────────────────────────
# Collapsed by default so it doesn't clutter the UI during active tasks.
if agent_log: if agent_log:
with st.expander(f"Agent Log — {len(agent_log)} step(s) completed", expanded=False): with st.expander(f"Agent Log — {len(agent_log)} step(s) completed", expanded=False):
for i, step in enumerate(agent_log): for i, step in enumerate(agent_log):
@ -97,6 +113,7 @@ def render_agent_mode():
if step.get("arguments"): if step.get("arguments"):
st.json(step["arguments"]) st.json(step["arguments"])
result_text = step.get("result", "") result_text = step.get("result", "")
# Colour the result based on whether the tool succeeded or failed.
if result_text.startswith("ERROR") or result_text.startswith("SYNTAX ERROR"): if result_text.startswith("ERROR") or result_text.startswith("SYNTAX ERROR"):
st.error(result_text) st.error(result_text)
elif result_text.startswith("OK") or result_text.startswith("DONE"): elif result_text.startswith("OK") or result_text.startswith("DONE"):
@ -132,6 +149,8 @@ def render_agent_mode():
args = pending.get("arguments", {}) args = pending.get("arguments", {})
if args: if args:
# Show file content separately as a code block for readability;
# other arguments are displayed as JSON.
if "content" in args: if "content" in args:
display_args = {k: v for k, v in args.items() if k != "content"} display_args = {k: v for k, v in args.items() if k != "content"}
if display_args: if display_args:
@ -194,7 +213,13 @@ def render_agent_mode():
# ── Normal Chat ─────────────────────────────────────────────────────────────── # ── Normal Chat ───────────────────────────────────────────────────────────────
def render_normal_chat(): def render_normal_chat():
# Chat history as bubbles """Render the standard multi-turn chat interface.
On the first message the system prompt is injected into the history.
Each subsequent message appends to the same conversation so the AI retains
full context throughout the session.
"""
# Replay the conversation history as chat bubbles (skip system messages).
for message in st.session_state.chat_history: for message in st.session_state.chat_history:
role = message["role"] role = message["role"]
if role == "system": if role == "system":
@ -207,16 +232,17 @@ def render_normal_chat():
if user_input: if user_input:
chat_manager = st.session_state.chat_manager chat_manager = st.session_state.chat_manager
# Inject system prompt on the first message # On the very first user message, prepend the system prompt so the AI
# knows it is a code assistant embedded in an editor.
if not chat_manager.get_history(): if not chat_manager.get_history():
system_prompt = SystemPrompter.generate_prompt() system_prompt = SystemPrompter.generate_prompt()
chat_manager.add_message("system", system_prompt) chat_manager.add_message("system", system_prompt)
# Show user message immediately without waiting for response # Show user message immediately without waiting for response.
with st.chat_message("user"): with st.chat_message("user"):
st.markdown(user_input) st.markdown(user_input)
# Show response with spinner while API is called # Call the AI and show its response with a spinner while waiting.
with st.chat_message("assistant"): with st.chat_message("assistant"):
with st.spinner("Thinking..."): with st.spinner("Thinking..."):
try: try:
@ -239,6 +265,7 @@ def render_normal_chat():
# ── Entry point ─────────────────────────────────────────────────────────────── # ── Entry point ───────────────────────────────────────────────────────────────
def render_chat(): def render_chat():
"""Top-level chat view — switches between Agent Mode and normal chat."""
if st.session_state.get("agent_mode", False): if st.session_state.get("agent_mode", False):
st.subheader("Coding Agent") st.subheader("Coding Agent")
render_agent_mode() render_agent_mode()

57
frontend/editor.py
View File

@ -1,3 +1,5 @@
"""Code Editor view — renders the Ace editor, file tabs, and execution output."""
import streamlit as st import streamlit as st
import streamlit_ace as st_ace import streamlit_ace as st_ace
from pathlib import Path from pathlib import Path
@ -6,6 +8,7 @@ from backend.managers.file_manager import FileManager
from backend.managers.execution_engine import ExecutionEngine from backend.managers.execution_engine import ExecutionEngine
from backend.managers.debug_logger import DebugLogger from backend.managers.debug_logger import DebugLogger
# Maps file extensions to Ace editor language modes for syntax highlighting.
LANG_MAP = { LANG_MAP = {
".py": "python", ".tex": "latex", ".js": "javascript", ".py": "python", ".tex": "latex", ".js": "javascript",
".html": "html", ".css": "css", ".sh": "bash", ".html": "html", ".css": "css", ".sh": "bash",
@ -18,6 +21,10 @@ LANG_MAP = {
@st.dialog("Rename File") @st.dialog("Rename File")
def _rename_dialog(file_path: str): def _rename_dialog(file_path: str):
"""Dialog for renaming the given file.
Updates open_files, files_content and active_file in session_state so
all tabs and the editor reference the new path immediately.
"""
fm = FileManager() fm = FileManager()
st.write(f"Current name: **{Path(file_path).name}**") st.write(f"Current name: **{Path(file_path).name}**")
new_name = st.text_input("New name:", value=Path(file_path).stem) new_name = st.text_input("New name:", value=Path(file_path).stem)
@ -34,10 +41,13 @@ def _rename_dialog(file_path: str):
if fm.rename_file(rel_path, new_name.strip()): if fm.rename_file(rel_path, new_name.strip()):
ext = Path(file_path).suffix ext = Path(file_path).suffix
new_file_path = str(Path(file_path).parent / (Path(new_name.strip()).stem + ext)) new_file_path = str(Path(file_path).parent / (Path(new_name.strip()).stem + ext))
# Update the open-files list in place so the tab order is preserved.
i = st.session_state.open_files.index(file_path) i = st.session_state.open_files.index(file_path)
st.session_state.open_files[i] = new_file_path st.session_state.open_files[i] = new_file_path
# Transfer cached editor content to the new path key.
st.session_state.files_content[new_file_path] = \ st.session_state.files_content[new_file_path] = \
st.session_state.files_content.pop(file_path) st.session_state.files_content.pop(file_path)
# Update active_file if the renamed file was the active one.
if st.session_state.active_file == file_path: if st.session_state.active_file == file_path:
st.session_state.active_file = new_file_path st.session_state.active_file = new_file_path
st.rerun() st.rerun()
@ -75,22 +85,30 @@ def _delete_dialog(abs_file_path: str):
def run_active_file(): def run_active_file():
"""Execute the currently active file and store the result in session_state.
Returns the execution result dict {stdout, stderr, return_code}, or None
if no active file is set.
"""
active_file = st.session_state.active_file active_file = st.session_state.active_file
if not active_file: if not active_file:
st.warning("No active file to run.") st.warning("No active file to run.")
return return
execution_engine = ExecutionEngine() execution_engine = ExecutionEngine()
debug_logger = DebugLogger() debug_logger = DebugLogger()
debug_logger.clear()
debug_logger.log(f"Executing code from {active_file}...") debug_logger.log(f"Executing code from {active_file}...")
with st.spinner(f"Running {Path(active_file).name}..."): with st.spinner(f"Running {Path(active_file).name}..."):
output = execution_engine.run_code(Path(active_file)) output = execution_engine.run_code(Path(active_file))
debug_logger.log("Execution completed.") if output["rc"] == 0:
debug_logger.log("Execution completed successfully.")
else:
debug_logger.log_error(f"Execution failed with exit code {output['rc']}.")
st.session_state.code_execution_output = { st.session_state.code_execution_output = {
"stdout": output["stdout"], "stdout": output["stdout"],
"stderr": output["stderr"], "stderr": output["stderr"],
@ -100,6 +118,7 @@ def run_active_file():
return result return result
def render_editor(): def render_editor():
"""Render the full Code Editor view with tabs, Ace editor, and run output."""
st.subheader("Code Editor") st.subheader("Code Editor")
if not st.session_state.open_files: if not st.session_state.open_files:
@ -109,16 +128,19 @@ def render_editor():
fm = FileManager() fm = FileManager()
# ── Tab bar via st.tabs() ───────────────────────────────────────────────── # ── Tab bar via st.tabs() ─────────────────────────────────────────────────
# Build one tab per open file, named by the file's basename.
tab_names = [Path(f).name for f in st.session_state.open_files] tab_names = [Path(f).name for f in st.session_state.open_files]
tabs = st.tabs(tab_names) tabs = st.tabs(tab_names)
for idx, file_path in enumerate(st.session_state.open_files): for idx, file_path in enumerate(st.session_state.open_files):
with tabs[idx]: with tabs[idx]:
# Load file content from disk on first open; afterwards use the cached version.
if file_path not in st.session_state.files_content: if file_path not in st.session_state.files_content:
st.session_state.files_content[file_path] = fm.read_file(Path(file_path)) st.session_state.files_content[file_path] = fm.read_file(Path(file_path))
file_language = LANG_MAP.get(Path(file_path).suffix, "text") file_language = LANG_MAP.get(Path(file_path).suffix, "text")
# Ace editor widget — auto_update sends content to Python on each keystroke.
code = st_ace.st_ace( code = st_ace.st_ace(
value=st.session_state.files_content[file_path], value=st.session_state.files_content[file_path],
language=file_language, language=file_language,
@ -128,6 +150,7 @@ def render_editor():
height=400, height=400,
) )
# Keep the in-memory cache in sync with what the editor currently shows.
if code != st.session_state.files_content[file_path]: if code != st.session_state.files_content[file_path]:
st.session_state.files_content[file_path] = code st.session_state.files_content[file_path] = code
@ -141,6 +164,7 @@ def render_editor():
if st.button("Close File", key=f"close_{file_path}"): if st.button("Close File", key=f"close_{file_path}"):
st.session_state.open_files.remove(file_path) st.session_state.open_files.remove(file_path)
st.session_state.files_content.pop(file_path, None) st.session_state.files_content.pop(file_path, None)
# Switch active_file to the next available tab.
st.session_state.active_file = ( st.session_state.active_file = (
st.session_state.open_files[0] st.session_state.open_files[0]
if st.session_state.open_files else None if st.session_state.open_files else None
@ -159,33 +183,34 @@ def render_editor():
result = run_active_file() result = run_active_file()
if not result: if not result:
st.stop() st.stop()
st.subheader("Execution Output") st.subheader("Execution Output")
# Green on exit code 0 (success), red on anything else (error/crash).
if result["return_code"] == 0: if result["return_code"] == 0:
st.success(f"Exit code: {result['return_code']}") st.success(f"Exit code: {result['return_code']}")
else: else:
st.error(f"Exit code: {result['return_code']}") st.error(f"Exit code: {result['return_code']}")
if result["stdout"]: if result["stdout"]:
st.text_area( st.text_area(
"Standard Output", "Standard Output",
value=result["stdout"], value=result["stdout"],
height=200, height=200,
disabled=True, disabled=True,
key="run_stdout") key="run_stdout")
if result["stderr"]: if result["stderr"]:
st.text_area( st.text_area(
"Standard Error", "Standard Error",
value=result["stderr"], value=result["stderr"],
height=200, height=200,
disabled=True, disabled=True,
key="run_stderr") key="run_stderr")
if not result["stdout"] and not result["stderr"]: if not result["stdout"] and not result["stderr"]:
st.info("No output produced by the code execution.") st.info("No output produced by the code execution.")
if __name__ == "__main__": if __name__ == "__main__":
render_editor() render_editor()

40
frontend/sidebar.py
View File

@ -1,10 +1,15 @@
"""Sidebar — navigation radio, logo, and the workspace file explorer."""
import streamlit as st import streamlit as st
from streamlit_arborist import tree_view from streamlit_arborist import tree_view
from pathlib import Path from pathlib import Path
from backend.managers.file_manager import FileManager from backend.managers.file_manager import FileManager
# Shared FileManager instance for all sidebar operations.
fm = FileManager() fm = FileManager()
# Maps file extensions (and special keys "folder"/"default") to display emojis
# shown next to each entry in the file tree.
SUFFIX_MAP = { SUFFIX_MAP = {
".py": "🐍", # Python ".py": "🐍", # Python
".js": "🟨", # JavaScript (Gelbes Quadrat/Logo) ".js": "🟨", # JavaScript (Gelbes Quadrat/Logo)
@ -29,11 +34,13 @@ SUFFIX_MAP = {
@st.dialog("Delete Folder") @st.dialog("Delete Folder")
def _delete_folder_dialog(folder_rel: str, folder_name: str): def _delete_folder_dialog(folder_rel: str, folder_name: str):
"""Confirmation dialog before permanently deleting a folder and its contents."""
st.warning(f"Delete **{folder_name}** and all its contents? This cannot be undone.") st.warning(f"Delete **{folder_name}** and all its contents? This cannot be undone.")
col1, col2 = st.columns(2) col1, col2 = st.columns(2)
with col1: with col1:
if st.button("Delete", type="primary", use_container_width=True): if st.button("Delete", type="primary", use_container_width=True):
if fm.delete_folder(folder_rel): if fm.delete_folder(folder_rel):
# Clear the selected-folder state so the action bar disappears.
st.session_state.selected_folder = None st.session_state.selected_folder = None
st.session_state.selected_folder_rel = None st.session_state.selected_folder_rel = None
st.rerun() st.rerun()
@ -187,19 +194,33 @@ def _delete_file_dialog(relative_file_path: str, file_name: str):
# ── File tree ───────────────────────────────────────────────────────────────── # ── File tree ─────────────────────────────────────────────────────────────────
def build_arborist_tree(tree, parent_path=Path()): def build_arborist_tree(tree, parent_path=Path()):
"""Convert the FileManager dict tree into the node format expected by streamlit-arborist.
Folders become nodes with a "children" list; files become leaf nodes
with an emoji prefix derived from their extension.
Args:
tree: Nested dict from FileManager.get_file_tree()
parent_path: Accumulates the relative path while recursing.
Returns:
List of node dicts accepted by tree_view().
"""
nodes = [] nodes = []
for name, content in sorted(tree.items()): for name, content in sorted(tree.items()):
full_path = parent_path / name full_path = parent_path / name
node_id = str(full_path.as_posix()) node_id = str(full_path.as_posix()) # forward-slash IDs work cross-platform
if isinstance(content, dict): if isinstance(content, dict):
# Directory — recurse to build child nodes.
nodes.append({ nodes.append({
"id": node_id, "id": node_id,
"name": f"{name}", "name": f"{name}",
"children": build_arborist_tree(content, full_path) "children": build_arborist_tree(content, full_path)
}) })
else: else:
# File — pick an emoji based on extension, fall back to default.
suffix = Path(name).suffix suffix = Path(name).suffix
icon = SUFFIX_MAP.get(suffix, SUFFIX_MAP["default"]) icon = SUFFIX_MAP.get(suffix, SUFFIX_MAP["default"])
@ -213,6 +234,7 @@ def build_arborist_tree(tree, parent_path=Path()):
def render_filetree_arborist(tree): def render_filetree_arborist(tree):
"""Render the interactive file tree and return the currently selected node dict."""
data = build_arborist_tree(tree) data = build_arborist_tree(tree)
selected = tree_view( selected = tree_view(
@ -220,7 +242,7 @@ def render_filetree_arborist(tree):
icons={"open": "📂", "closed": "📁"}, icons={"open": "📂", "closed": "📁"},
height=200, height=200,
selection=None, selection=None,
select_internal_nodes=True, select_internal_nodes=True, # allow clicking folder names, not just files
open_by_default=False open_by_default=False
) )
@ -230,6 +252,12 @@ def render_filetree_arborist(tree):
# ── Sidebar ─────────────────────────────────────────────────────────────────── # ── Sidebar ───────────────────────────────────────────────────────────────────
def render_sidebar(): def render_sidebar():
"""Render the full sidebar: navigation radio and workspace file explorer.
Tree click handling:
- Clicking a file → appended to open_files, set as active_file
- Clicking a folder → stored in selected_folder so the action bar appears
"""
st.sidebar.title("Navigation") st.sidebar.title("Navigation")
navigation_section = st.sidebar.container() navigation_section = st.sidebar.container()
@ -259,11 +287,14 @@ def render_sidebar():
if selected: if selected:
selected_path = selected.get("id") selected_path = selected.get("id")
# Only react when the user clicks a *different* node to
# avoid re-running on every Streamlit rerender.
if st.session_state.last_selected != selected_path: if st.session_state.last_selected != selected_path:
st.session_state.last_selected = selected_path st.session_state.last_selected = selected_path
abs_path = fm.base_path / selected_path abs_path = fm.base_path / selected_path
if abs_path.is_file(): if abs_path.is_file():
# Open the file in the editor.
st.session_state.selected_folder = None st.session_state.selected_folder = None
st.session_state.selected_folder_rel = None st.session_state.selected_folder_rel = None
file_str = str(abs_path) file_str = str(abs_path)
@ -273,11 +304,13 @@ def render_sidebar():
st.rerun() st.rerun()
elif abs_path.is_dir(): elif abs_path.is_dir():
# Select the folder so its action bar appears below.
st.session_state.selected_folder = str(abs_path) st.session_state.selected_folder = str(abs_path)
st.session_state.selected_folder_rel = selected_path st.session_state.selected_folder_rel = selected_path
st.rerun() st.rerun()
# Folder actions — rendered outside the selection block so they persist across reruns # Folder action bar — rendered unconditionally outside the selection
# block so it persists across reruns even when no new click happens.
if st.session_state.get("selected_folder"): if st.session_state.get("selected_folder"):
folder_name = Path(st.session_state.selected_folder).name folder_name = Path(st.session_state.selected_folder).name
folder_rel = st.session_state.selected_folder_rel folder_rel = st.session_state.selected_folder_rel
@ -303,6 +336,7 @@ def render_sidebar():
_delete_file_dialog(file_rel, active_file_name) _delete_file_dialog(file_rel, active_file_name)
with add_more: with add_more:
# Popover for workspace-root actions (not tied to any selected folder).
with st.popover("⚙️ Explorer Options", key="popover_options", use_container_width=True): with st.popover("⚙️ Explorer Options", key="popover_options", use_container_width=True):
if st.button("Add File", key="btn_add_file", use_container_width=True): if st.button("Add File", key="btn_add_file", use_container_width=True):
_add_file_dialog("") _add_file_dialog("")

23
frontend/state.py
View File

@ -1,12 +1,21 @@
"""Centralised session-state initialisation for the Streamlit app.
All keys used throughout the app are declared here with their default values.
Calling init_state() at the top of app.py ensures every key exists before any
page tries to read it, preventing KeyError on the first load.
"""
import streamlit as st import streamlit as st
from backend.managers.chat_manager import ChatManager from backend.managers.chat_manager import ChatManager
def init_state(): def init_state():
# Sidebar state initialization # Sidebar state initialization
# last_selected tracks the previously clicked tree node to detect new clicks
if "last_selected" not in st.session_state: if "last_selected" not in st.session_state:
st.session_state.last_selected = None st.session_state.last_selected = None
# Absolute path and workspace-relative path of the currently highlighted folder
if "selected_folder" not in st.session_state: if "selected_folder" not in st.session_state:
st.session_state.selected_folder = None st.session_state.selected_folder = None
@ -14,27 +23,32 @@ def init_state():
st.session_state.selected_folder_rel = None st.session_state.selected_folder_rel = None
# Chat manager (persists across reruns) # Chat manager (persists across reruns)
# ChatManager keeps the full conversation history in memory across reruns
if "chat_manager" not in st.session_state: if "chat_manager" not in st.session_state:
st.session_state.chat_manager = ChatManager() st.session_state.chat_manager = ChatManager()
# Editor state initialization # Editor state initialization
# List of absolute file paths that are currently open as tabs
if "open_files" not in st.session_state: if "open_files" not in st.session_state:
"""A list of currently open file paths - absolute paths only. The order determines the tab order in the UI. """A list of currently open file paths - absolute paths only. The order determines the tab order in the UI.
Format: [ "path/to/file1.py", "path/to/file2.js", ... ] Format: [ "path/to/file1.py", "path/to/file2.js", ... ]
""" """
st.session_state.open_files = [] st.session_state.open_files = []
# Dict mapping file path → current editor content (may be unsaved)
if "files_content" not in st.session_state: if "files_content" not in st.session_state:
"""A dictionary mapping file paths to their current content in the editor. """A dictionary mapping file paths to their current content in the editor.
Format: { "path/to/file.py": "file content as string", ... } Format: { "path/to/file.py": "file content as string", ... }
""" """
st.session_state.files_content = {} st.session_state.files_content = {}
# Absolute path of the file whose tab is currently active
if "active_file" not in st.session_state: if "active_file" not in st.session_state:
"""The currently active file in the editor (absolute path in string e.g. "/workspace/path/to/file.py"). """The currently active file in the editor (absolute path in string e.g. "/workspace/path/to/file.py").
Should be one of the paths in open_files or None if no file is open.""" Should be one of the paths in open_files or None if no file is open."""
st.session_state.active_file = None st.session_state.active_file = None
# Index of the active tab (used by st.tabs)
if "active_tab" not in st.session_state: if "active_tab" not in st.session_state:
st.session_state.active_tab = 0 st.session_state.active_tab = 0
@ -44,26 +58,33 @@ def init_state():
if "code_suggestions" not in st.session_state: if "code_suggestions" not in st.session_state:
st.session_state.code_suggestions = [] st.session_state.code_suggestions = []
# Output dict from the last code run: {stdout, stderr, return_code}
if "code_execution_output" not in st.session_state: if "code_execution_output" not in st.session_state:
st.session_state.code_execution_output = "" st.session_state.code_execution_output = ""
# Chat state initialization # Chat state initialization
# Flat list of {"role": ..., "content": ...} dicts shown as chat bubbles
if "chat_history" not in st.session_state: if "chat_history" not in st.session_state:
st.session_state.chat_history = [] st.session_state.chat_history = []
# Agent Mode state # Agent Mode state
# Whether the UI is currently in Agent Mode (vs normal chat)
if "agent_mode" not in st.session_state: if "agent_mode" not in st.session_state:
st.session_state.agent_mode = False st.session_state.agent_mode = False
# The live CodingAgent instance while a task is running
if "coding_agent" not in st.session_state: if "coding_agent" not in st.session_state:
st.session_state.coding_agent = None st.session_state.coding_agent = None
# Current status of the agent: "idle" | "waiting_approval" | "done"
if "agent_status" not in st.session_state: if "agent_status" not in st.session_state:
st.session_state.agent_status = "idle" st.session_state.agent_status = "idle"
# List of completed steps shown in the collapsible Agent Log
if "agent_log" not in st.session_state: if "agent_log" not in st.session_state:
st.session_state.agent_log = [] st.session_state.agent_log = []
# The action the agent proposed but has not yet been approved or rejected
if "agent_pending_action" not in st.session_state: if "agent_pending_action" not in st.session_state:
st.session_state.agent_pending_action = None st.session_state.agent_pending_action = None

1
requirements.txt
View File

@ -17,6 +17,7 @@ pandas>=2.0.0
# Testing # Testing
pytest>=7.0.0 pytest>=7.0.0
pytest-cov>=4.0.0 pytest-cov>=4.0.0
pytest-asyncio>=0.23.0
# Development & Utilities # Development & Utilities
python-dotenv>=1.0.0 python-dotenv>=1.0.0

24
tests/conftest.py Normal file
View File

@ -0,0 +1,24 @@
"""Shared pytest configuration — runs before any test module is imported.
Patches MCPToolAdapter at the sys.modules level so that importing
backend.agent.coding_agent never tries to start real MCP subprocess servers.
"""
import sys
from unittest.mock import AsyncMock, MagicMock
# Build a fake adapter instance whose async methods return immediately.
_mock_adapter = MagicMock()
_mock_adapter.initialize_all_servers = AsyncMock(return_value=None)
_mock_adapter.get_all_tools = MagicMock(return_value=[])
_mock_adapter.call_tool = AsyncMock(return_value=MagicMock(isError=False, content=[]))
# Wrap as a class mock: calling MCPToolAdapter() returns _mock_adapter.
_mock_adapter_cls = MagicMock(return_value=_mock_adapter)
# Inject before any test imports coding_agent so the module-level
# asyncio.run(adapter.initialize_all_servers()) uses the mock.
sys.modules.setdefault(
"backend.agent.mcp_server_adapter",
MagicMock(MCPToolAdapter=_mock_adapter_cls),
)

297
tests/test_chat_manager.py
View File

@ -1,242 +1,157 @@
"""Test script for ChatManager - Pytest compatible tests""" """Tests for ChatManager (backend/managers/chat_manager.py)."""
import sys import sys
from pathlib import Path from pathlib import Path
import pytest import pytest
import requests
from unittest.mock import patch, MagicMock from unittest.mock import patch, MagicMock
# Add project root to Python path
sys.path.insert(0, str(Path(__file__).parent.parent)) sys.path.insert(0, str(Path(__file__).parent.parent))
from backend.managers.chat_manager import ChatManager from backend.managers.chat_manager import ChatManager
class TestChatManager: # ── Helpers ──────────────────────────────────────────────────────────────────
"""Test suite for ChatManager functionality."""
def _mock_ok(content="AI reply"):
"""Return a mocked 200 response with a single assistant choice."""
mock = MagicMock()
mock.status_code = 200
mock.json.return_value = {
"choices": [{"message": {"role": "assistant", "content": content}}]
}
mock.text = ""
return mock
# ── History management ────────────────────────────────────────────────────────
class TestHistory:
"""Tests for add_message and clear_history."""
@pytest.fixture @pytest.fixture
def chat_manager(self): def cm(self):
return ChatManager() return ChatManager()
def test_initialization(self, chat_manager): def test_add_message_appends_correct_entry(self, cm):
"""Test that ChatManager initializes correctly.""" cm.add_message("user", "Hello")
assert chat_manager.api_url is not None assert cm.chat_history == [{"role": "user", "content": "Hello"}]
assert chat_manager.model is not None
assert chat_manager.chat_history == []
def test_add_message(self, chat_manager): def test_add_multiple_messages_preserves_order(self, cm):
"""Test adding messages to chat history.""" cm.add_message("user", "Hi")
chat_manager.add_message("user", "Hello") cm.add_message("assistant", "Hello!")
assert len(chat_manager.chat_history) == 1 assert cm.chat_history[0]["role"] == "user"
assert chat_manager.chat_history[0]["role"] == "user" assert cm.chat_history[1]["role"] == "assistant"
assert chat_manager.chat_history[0]["content"] == "Hello"
def test_get_history(self, chat_manager): def test_clear_history_empties_list(self, cm):
"""Test retrieving chat history.""" cm.add_message("user", "Hi")
chat_manager.add_message("user", "Hello") cm.clear_history()
chat_manager.add_message("assistant", "Hi there!") assert cm.chat_history == []
history = chat_manager.get_history()
assert len(history) == 2
assert history[0]["role"] == "user"
assert history[1]["role"] == "assistant"
def test_clear_history(self, chat_manager):
"""Test clearing chat history."""
chat_manager.add_message("user", "Hello")
assert len(chat_manager.chat_history) == 1
chat_manager.clear_history()
assert len(chat_manager.chat_history) == 0
def test_send_message_integration(self, chat_manager):
"""
Integration test for sending message to AI.
This test actually communicates with the API.
"""
try:
# Send a simple test message
response = chat_manager.send_message("Hello, what is 2+2?")
# Verify response is not empty
assert isinstance(response, str)
assert len(response) > 0
# Verify message was added to history
assert len(chat_manager.chat_history) == 2 # user + assistant
assert chat_manager.chat_history[0]["role"] == "user"
assert chat_manager.chat_history[1]["role"] == "assistant"
print(f"API Test Passed")
print(f"Response: {response}")
except Exception as e:
# If API is not reachable, mark as skipped
pytest.skip(f"API not reachable: {str(e)}")
def test_multiple_messages(self, chat_manager):
"""Test sending multiple messages in a conversation."""
try:
# Send first message
response1 = chat_manager.send_message("What is your name?")
assert len(response1) > 0
# Send follow-up message
response2 = chat_manager.send_message("Tell me more")
assert len(response2) > 0
# Verify full conversation is in history
assert len(chat_manager.chat_history) == 4 # 2 user + 2 assistant
print(f"Conversation Test Passed")
print(f"Messages: {len(chat_manager.chat_history)}")
except Exception as e:
pytest.skip(f"API not reachable: {str(e)}")
class TestChatManagerSendMessage: # ── send_message (mocked HTTP) ────────────────────────────────────────────────
"""Unit tests for send_message using mocked HTTP requests."""
class TestSendMessage:
"""Tests for send_message: history updates, HTTP payload, error handling, and auth headers."""
@pytest.fixture @pytest.fixture
def chat_manager(self): def cm(self):
return ChatManager() return ChatManager()
def _mock_response(self, content="AI reply", status_code=200): def test_user_message_added_to_history(self, cm):
mock = MagicMock() with patch("requests.post", return_value=_mock_ok()):
mock.status_code = status_code cm.send_message("Hello")
mock.json.return_value = { assert cm.chat_history[0] == {"role": "user", "content": "Hello"}
"choices": [{"message": {"role": "assistant", "content": content}}]
}
mock.text = "error text"
return mock
def test_send_message_adds_user_message_to_history(self, chat_manager): def test_assistant_reply_added_to_history(self, cm):
with patch("requests.post", return_value=self._mock_response()): with patch("requests.post", return_value=_mock_ok("Hi there")):
chat_manager.send_message("Hello") cm.send_message("Hello")
assert chat_manager.chat_history[0] == {"role": "user", "content": "Hello"} assert cm.chat_history[1] == {"role": "assistant", "content": "Hi there"}
def test_send_message_adds_assistant_response_to_history(self, chat_manager): def test_returns_assistant_content_string(self, cm):
with patch("requests.post", return_value=self._mock_response("Hi there")): with patch("requests.post", return_value=_mock_ok("Answer")):
chat_manager.send_message("Hello") result = cm.send_message("Question")
assert chat_manager.chat_history[1] == {"role": "assistant", "content": "Hi there"} assert result == "Answer"
def test_send_message_returns_ai_content(self, chat_manager): def test_full_history_sent_in_request_payload(self, cm):
with patch("requests.post", return_value=self._mock_response("Answer")): """All prior messages must be forwarded so the model has conversation context."""
response = chat_manager.send_message("Question") cm.add_message("system", "You are helpful.")
assert response == "Answer" with patch("requests.post", return_value=_mock_ok()) as mock_post:
cm.send_message("Hello")
payload = mock_post.call_args.kwargs["json"]
assert payload["messages"][0]["role"] == "system"
assert payload["messages"][1]["role"] == "user"
def test_send_message_history_grows_with_each_call(self, chat_manager): def test_connection_error_raises_and_adds_error_to_history(self, cm):
with patch("requests.post", return_value=self._mock_response()):
chat_manager.send_message("First")
chat_manager.send_message("Second")
assert len(chat_manager.chat_history) == 4 # 2 user + 2 assistant
def test_send_message_connection_error_raises(self, chat_manager):
import requests
with patch("requests.post", side_effect=requests.exceptions.ConnectionError("refused")): with patch("requests.post", side_effect=requests.exceptions.ConnectionError("refused")):
with pytest.raises(Exception, match="Connection Error"): with pytest.raises(Exception, match="Connection Error"):
chat_manager.send_message("Hello") cm.send_message("Hello")
assert any("Error" in msg["content"] for msg in cm.chat_history)
def test_send_message_api_error_status_raises(self, chat_manager): def test_api_error_status_raises(self, cm):
mock = self._mock_response(status_code=500) mock = MagicMock()
mock.status_code = 500
mock.text = "Internal Server Error"
with patch("requests.post", return_value=mock): with patch("requests.post", return_value=mock):
with pytest.raises(Exception, match="API Error 500"): with pytest.raises(Exception, match="API Error 500"):
chat_manager.send_message("Hello") cm.send_message("Hello")
def test_send_message_empty_choices_raises(self, chat_manager): def test_timeout_raises(self, cm):
with patch("requests.post", side_effect=requests.exceptions.Timeout()):
with pytest.raises(Exception):
cm.send_message("Hello")
def test_empty_choices_raises(self, cm):
mock = MagicMock() mock = MagicMock()
mock.status_code = 200 mock.status_code = 200
mock.json.return_value = {"choices": []} mock.json.return_value = {"choices": []}
with patch("requests.post", return_value=mock): with patch("requests.post", return_value=mock):
with pytest.raises(Exception, match="Invalid API response format"): with pytest.raises(Exception, match="Invalid API response format"):
chat_manager.send_message("Hello") cm.send_message("Hello")
def test_send_message_missing_choices_key_raises(self, chat_manager): def test_api_key_included_in_header_when_set(self, cm):
cm.api_key = "test-key-123"
with patch("requests.post", return_value=_mock_ok()) as mock_post:
cm.send_message("Hello")
headers = mock_post.call_args.kwargs["headers"]
assert headers.get("Authorization") == "Bearer test-key-123"
def test_api_key_excluded_from_header_when_empty_sentinel(self, cm):
# "EMPTY" is the sentinel string the UI writes when the user leaves the key field blank.
cm.api_key = "EMPTY"
with patch("requests.post", return_value=_mock_ok()) as mock_post:
cm.send_message("Hello")
headers = mock_post.call_args.kwargs["headers"]
assert "Authorization" not in headers
def test_json_decode_error_raises(self, cm):
import json
mock = MagicMock() mock = MagicMock()
mock.status_code = 200 mock.status_code = 200
mock.json.return_value = {} mock.json.side_effect = json.JSONDecodeError("bad json", "", 0)
with patch("requests.post", return_value=mock): with patch("requests.post", return_value=mock):
with pytest.raises(Exception): with pytest.raises(Exception, match="JSON Decode Error"):
chat_manager.send_message("Hello") cm.send_message("Hello")
def test_send_message_timeout_raises(self, chat_manager):
import requests
with patch("requests.post", side_effect=requests.exceptions.Timeout()):
with pytest.raises(Exception):
chat_manager.send_message("Hello")
class TestChatManagerGetChatDisplay: # ── get_chat_display ──────────────────────────────────────────────────────────
"""Tests for get_chat_display()."""
class TestGetChatDisplay:
"""Tests for get_chat_display: correct shape and ordering."""
@pytest.fixture @pytest.fixture
def chat_manager(self): def cm(self):
return ChatManager() return ChatManager()
def test_empty_history_returns_empty_list(self, chat_manager): def test_display_has_role_and_content_keys(self, cm):
assert chat_manager.get_chat_display() == [] cm.add_message("user", "Hello")
entry = cm.get_chat_display()[0]
assert "role" in entry
assert "content" in entry
def test_display_contains_role_and_content_keys(self, chat_manager): def test_display_preserves_message_order(self, cm):
chat_manager.add_message("user", "Hello") cm.add_message("user", "First")
display = chat_manager.get_chat_display() cm.add_message("assistant", "Second")
assert "role" in display[0] display = cm.get_chat_display()
assert "content" in display[0]
def test_display_preserves_message_order(self, chat_manager):
chat_manager.add_message("user", "First")
chat_manager.add_message("assistant", "Second")
display = chat_manager.get_chat_display()
assert display[0]["role"] == "user" assert display[0]["role"] == "user"
assert display[1]["role"] == "assistant" assert display[1]["role"] == "assistant"
def test_display_matches_history(self, chat_manager):
chat_manager.add_message("user", "Hi")
chat_manager.add_message("assistant", "Hello!")
assert chat_manager.get_chat_display() == chat_manager.get_history()
def test_system_message_included_in_display(self, chat_manager):
chat_manager.add_message("system", "You are a helper.")
display = chat_manager.get_chat_display()
assert display[0]["role"] == "system"
def test_chat_manager_demo():
"""Demo test - Shows interactive chat (can be run manually)."""
print("\n" + "=" * 60)
print("ChatManager Demo - Interactive Test")
print("=" * 60 + "\n")
chat_manager = ChatManager()
print(f"Connected to API: {chat_manager.api_url}")
print(f"Model: {chat_manager.model}\n")
# Demo conversation
test_messages = ["Hello! What can you do?", "Tell me a joke", "What is Python?"]
print("Starting conversation...\n")
for message in test_messages:
print(f"User: {message}")
try:
response = chat_manager.send_message(message)
print(f"Assistant: {response}\n")
except Exception as e:
print(f"Error: {str(e)}\n")
pytest.skip(f"API not reachable: {str(e)}")
# Display full chat history
print("=" * 60)
print("Chat History:")
print("=" * 60)
for msg in chat_manager.get_history():
print(f"{msg['role'].upper()}: {msg['content']}\n")
if __name__ == "__main__":
# Run with: pytest tests/test_chat_manager.py -v -s
pytest.main([__file__, "-v", "-s"])

358
tests/test_coding_agent.py
View File

@ -3,13 +3,16 @@ Tests for CodingAgent (backend/agent/coding_agent.py)
Structure: Structure:
TestHelpers – truncate_result, trim_messages, _strip_code_fences 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 TestCodingAgentInit – __init__ and start_task
TestProposeNextAction – propose_next_action with mocked API TestProposeNextAction – propose_next_action with mocked API
TestApprove – approve with mocked API + real tool execution TestApprove – approve with mocked API + real tool execution
TestReject – reject injects feedback correctly TestReject – reject injects feedback correctly
TestFullLoop – integration: real API, skipped if unreachable TestFullLoop – integration: real API, skipped if unreachable
Note: TestTools (write_file, read_file, etc.) and TestDispatcher were removed
because those tool functions are now MCP server tools, not standalone functions
in coding_agent.py. They will be tested via test_mcp_server_*.py once the MCP
servers are finalised.
""" """
import json import json
@ -18,24 +21,18 @@ from pathlib import Path
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
import pytest import pytest
import pytest_asyncio
sys.path.insert(0, str(Path(__file__).parent.parent)) sys.path.insert(0, str(Path(__file__).parent.parent))
from backend.agent.coding_agent import ( from backend.agent.coding_agent import (
MAX_HISTORY_CHARS, MAX_HISTORY_CHARS,
MAX_ITERATIONS,
MAX_RESULT_LENGTH, MAX_RESULT_LENGTH,
CodingAgent, CodingAgent,
_strip_code_fences, _strip_code_fences,
dispatch_tool,
done,
grep_search,
list_files,
read_file,
run_python,
truncate_result, truncate_result,
trim_messages, trim_messages,
validate_python,
write_file,
) )
@ -55,6 +52,8 @@ def _make_api_response(content: str, status_code: int = 200):
def _agent_action_json(tool: str, thought: str = "thinking...", **arguments) -> str: def _agent_action_json(tool: str, thought: str = "thinking...", **arguments) -> str:
"""Return a JSON string in the exact format the agent expects from the LLM:
{"thought": "...", "tool": "<name>", "arguments": {...}}."""
return json.dumps({"thought": thought, "tool": tool, "arguments": arguments}) return json.dumps({"thought": thought, "tool": tool, "arguments": arguments})
@ -63,6 +62,7 @@ def _agent_action_json(tool: str, thought: str = "thinking...", **arguments) ->
# ═════════════════════════════════════════════════════════════════════════════ # ═════════════════════════════════════════════════════════════════════════════
class TestTruncateResult: class TestTruncateResult:
"""Tests for truncate_result(): ensures long tool outputs are capped before entering the message history."""
def test_short_result_unchanged(self): def test_short_result_unchanged(self):
assert truncate_result("hello") == "hello" assert truncate_result("hello") == "hello"
@ -73,10 +73,6 @@ class TestTruncateResult:
assert len(result) < len(long) assert len(result) < len(long)
assert "TRUNCATED" in result 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): def test_truncated_keeps_start_and_end(self):
text = "START" + "x" * MAX_RESULT_LENGTH + "END" text = "START" + "x" * MAX_RESULT_LENGTH + "END"
result = truncate_result(text) result = truncate_result(text)
@ -85,8 +81,11 @@ class TestTruncateResult:
class TestTrimMessages: class TestTrimMessages:
"""Tests for trim_messages(): keeps system + original task, drops old turns when history grows too large."""
def _make_messages(self, n_extra: int, chars_each: int = 100) -> list: def _make_messages(self, n_extra: int, chars_each: int = 100) -> list:
"""Build a message list with a fixed system + user header followed by
n_extra assistant/user pairs, each pair consuming 2*chars_each characters."""
msgs = [ msgs = [
{"role": "system", "content": "sys"}, {"role": "system", "content": "sys"},
{"role": "user", "content": "original task"}, {"role": "user", "content": "original task"},
@ -105,8 +104,6 @@ class TestTrimMessages:
original_total = sum(len(m["content"]) for m in msgs) original_total = sum(len(m["content"]) for m in msgs)
trimmed = trim_messages(msgs) trimmed = trim_messages(msgs)
trimmed_total = sum(len(m["content"]) for m in trimmed) 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 trimmed_total < original_total
assert len(trimmed) < len(msgs) assert len(trimmed) < len(msgs)
@ -121,6 +118,8 @@ class TestTrimMessages:
assert trimmed[1]["content"] == "original task" assert trimmed[1]["content"] == "original task"
def test_reminder_injected_when_trimmed(self): def test_reminder_injected_when_trimmed(self):
# trim_messages inserts a "system_note" message so the agent knows that
# earlier turns were dropped and it should not reference missing context.
msgs = self._make_messages(n_extra=500, chars_each=200) msgs = self._make_messages(n_extra=500, chars_each=200)
trimmed = trim_messages(msgs) trimmed = trim_messages(msgs)
contents = [m["content"] for m in trimmed] contents = [m["content"] for m in trimmed]
@ -128,6 +127,7 @@ class TestTrimMessages:
class TestStripCodeFences: class TestStripCodeFences:
"""Tests for _strip_code_fences(): the LLM sometimes wraps its JSON in markdown fences — this strips them."""
def test_plain_text_unchanged(self): def test_plain_text_unchanged(self):
assert _strip_code_fences("hello") == "hello" assert _strip_code_fences("hello") == "hello"
@ -140,171 +140,13 @@ class TestStripCodeFences:
text = "```\nhello\n```" text = "```\nhello\n```"
assert _strip_code_fences(text) == "hello" 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 # TestCodingAgentInit
# ═════════════════════════════════════════════════════════════════════════════ # ═════════════════════════════════════════════════════════════════════════════
class TestCodingAgentInit: class TestCodingAgentInit:
"""Tests for CodingAgent.__init__ and start_task(): state is clean before and after task setup."""
def test_initial_state_is_clean(self): def test_initial_state_is_clean(self):
agent = CodingAgent() agent = CodingAgent()
@ -313,11 +155,6 @@ class TestCodingAgentInit:
assert agent.is_done is False assert agent.is_done is False
assert agent.iteration == 0 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): def test_start_task_sets_messages(self):
agent = CodingAgent() agent = CodingAgent()
agent.start_task("Write fibonacci.py") agent.start_task("Write fibonacci.py")
@ -348,6 +185,7 @@ class TestCodingAgentInit:
# ═════════════════════════════════════════════════════════════════════════════ # ═════════════════════════════════════════════════════════════════════════════
class TestProposeNextAction: class TestProposeNextAction:
"""Tests for propose_next_action(): API is mocked so no real HTTP calls are made."""
@pytest.fixture @pytest.fixture
def agent(self): def agent(self):
@ -359,55 +197,72 @@ class TestProposeNextAction:
payload = _agent_action_json(tool, thought, **args) payload = _agent_action_json(tool, thought, **args)
agent._call_api = MagicMock(return_value=payload) agent._call_api = MagicMock(return_value=payload)
def test_returns_dict_with_required_keys(self, agent): @pytest.mark.asyncio
async def test_returns_dict_with_required_keys(self, agent):
self._mock_api(agent) self._mock_api(agent)
action = agent.propose_next_action() action = await agent.propose_next_action()
assert "thought" in action assert "thought" in action
assert "tool" in action assert "tool" in action
assert "arguments" in action assert "arguments" in action
def test_increments_iteration(self, agent): @pytest.mark.asyncio
async def test_increments_iteration(self, agent):
self._mock_api(agent) self._mock_api(agent)
agent.propose_next_action() await agent.propose_next_action()
assert agent.iteration == 1 assert agent.iteration == 1
def test_stores_pending_action(self, agent): @pytest.mark.asyncio
async def test_stores_pending_action(self, agent):
self._mock_api(agent) self._mock_api(agent)
agent.propose_next_action() await agent.propose_next_action()
assert agent.pending_action is not None assert agent.pending_action is not None
def test_returns_correct_tool(self, agent): @pytest.mark.asyncio
async def test_returns_correct_tool(self, agent):
self._mock_api(agent, tool="list_files") self._mock_api(agent, tool="list_files")
action = agent.propose_next_action() action = await agent.propose_next_action()
assert action["tool"] == "list_files" assert action["tool"] == "list_files"
def test_handles_json_parse_error_gracefully(self, agent): @pytest.mark.asyncio
async def test_handles_json_parse_error_gracefully(self, agent):
agent._call_api = MagicMock(return_value="this is not json {{") agent._call_api = MagicMock(return_value="this is not json {{")
action = agent.propose_next_action() action = await agent.propose_next_action()
assert action["tool"] == "done" assert action["tool"] == "done"
def test_handles_api_exception_gracefully(self, agent): @pytest.mark.asyncio
async def test_handles_api_exception_gracefully(self, agent):
agent._call_api = MagicMock(side_effect=Exception("connection refused")) agent._call_api = MagicMock(side_effect=Exception("connection refused"))
action = agent.propose_next_action() action = await agent.propose_next_action()
assert action["tool"] == "done" assert action["tool"] == "done"
def test_strips_code_fences_from_response(self, agent): @pytest.mark.asyncio
async def test_strips_code_fences_from_response(self, agent):
payload = "```json\n" + _agent_action_json("list_files", "thinking") + "\n```" payload = "```json\n" + _agent_action_json("list_files", "thinking") + "\n```"
agent._call_api = MagicMock(return_value=payload) agent._call_api = MagicMock(return_value=payload)
action = agent.propose_next_action() action = await agent.propose_next_action()
assert action["tool"] == "list_files" assert action["tool"] == "list_files"
def test_already_done_returns_done_action(self, agent): @pytest.mark.asyncio
async def test_already_done_returns_done_action(self, agent):
agent.is_done = True agent.is_done = True
action = agent.propose_next_action() action = await agent.propose_next_action()
assert action["tool"] == "done" assert action["tool"] == "done"
@pytest.mark.asyncio
async def test_max_iterations_returns_done_without_api_call(self, agent):
agent.iteration = MAX_ITERATIONS
agent._call_api = MagicMock(side_effect=AssertionError("API must not be called"))
action = await agent.propose_next_action()
assert action["tool"] == "done"
agent._call_api.assert_not_called()
# ═════════════════════════════════════════════════════════════════════════════ # ═════════════════════════════════════════════════════════════════════════════
# TestApprove (mocked API + real tool execution via tmp_path) # TestApprove (mocked API + mocked dispatch_tool)
# ═════════════════════════════════════════════════════════════════════════════ # ═════════════════════════════════════════════════════════════════════════════
class TestApprove: class TestApprove:
"""Tests for approve(): dispatch_tool is mocked so no filesystem or subprocess side-effects occur."""
@pytest.fixture @pytest.fixture
def agent(self): def agent(self):
@ -416,56 +271,68 @@ class TestApprove:
return a return a
def _set_pending(self, agent, tool: str, **arguments): def _set_pending(self, agent, tool: str, **arguments):
"""Inject a pending_action into the agent as if propose_next_action() had just run.
'raw' holds the original JSON string; 'action' holds the parsed dict."""
raw = _agent_action_json(tool, "thought", **arguments) raw = _agent_action_json(tool, "thought", **arguments)
agent.pending_action = { agent.pending_action = {
"raw": raw, "raw": raw,
"action": {"thought": "thought", "tool": tool, "arguments": arguments}, "action": {"thought": "thought", "tool": tool, "arguments": arguments},
} }
def test_approve_without_pending_raises(self, agent): @pytest.mark.asyncio
async def test_approve_without_pending_raises(self, agent):
with pytest.raises(Exception): with pytest.raises(Exception):
agent.approve() await agent.approve()
def test_approve_done_sets_is_done(self, agent): @pytest.mark.asyncio
async def test_approve_done_sets_is_done(self, agent):
self._set_pending(agent, "done", summary="all done") self._set_pending(agent, "done", summary="all done")
result = agent.approve() result = await agent.approve()
assert result["is_done"] is True assert result["is_done"] is True
assert agent.is_done is True assert agent.is_done is True
def test_approve_done_returns_summary(self, agent): @pytest.mark.asyncio
async def test_approve_done_returns_summary(self, agent):
self._set_pending(agent, "done", summary="finished successfully") self._set_pending(agent, "done", summary="finished successfully")
result = agent.approve() result = await agent.approve()
assert "finished successfully" in result["result"] assert "finished successfully" in result["result"]
def test_approve_clears_pending_action(self, agent): @pytest.mark.asyncio
async def test_approve_clears_pending_action(self, agent):
self._set_pending(agent, "done", summary="x") self._set_pending(agent, "done", summary="x")
agent.approve() await agent.approve()
assert agent.pending_action is None assert agent.pending_action is None
def test_approve_appends_assistant_message(self, agent): @pytest.mark.asyncio
async def test_approve_appends_assistant_message(self, agent):
self._set_pending(agent, "done", summary="x") self._set_pending(agent, "done", summary="x")
before = len(agent.messages) before = len(agent.messages)
agent.approve() await agent.approve()
assert len(agent.messages) > before assert len(agent.messages) > before
def test_approve_tool_result_appended_to_messages(self, agent, tmp_path): @pytest.mark.asyncio
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): async def test_approve_tool_result_appended_to_messages(self, agent):
with patch("backend.agent.coding_agent.dispatch_tool", return_value="file list"):
self._set_pending(agent, "list_files") self._set_pending(agent, "list_files")
agent.approve() await agent.approve()
tool_results = [m for m in agent.messages if "tool_result" in m["content"]] tool_results = [m for m in agent.messages if "tool_result" in m["content"]]
assert len(tool_results) == 1 assert len(tool_results) == 1
def test_approve_error_result_adds_replan_tag(self, agent, tmp_path): @pytest.mark.asyncio
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): async def test_approve_error_result_adds_replan_tag(self, agent):
# When a tool returns an error, approve() adds a "replan" tag to the message
# so the LLM knows the last action failed and must choose a different approach.
with patch("backend.agent.coding_agent.dispatch_tool", return_value="ERROR: file not found"):
self._set_pending(agent, "read_file", path="nonexistent.py") self._set_pending(agent, "read_file", path="nonexistent.py")
agent.approve() await agent.approve()
last_msg = agent.messages[-1]["content"] last_msg = agent.messages[-1]["content"]
assert "replan" in last_msg assert "replan" in last_msg
def test_approve_returns_tool_name_in_result(self, agent, tmp_path): @pytest.mark.asyncio
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): async def test_approve_returns_tool_name_in_result(self, agent):
with patch("backend.agent.coding_agent.dispatch_tool", return_value="(empty)"):
self._set_pending(agent, "list_files") self._set_pending(agent, "list_files")
result = agent.approve() result = await agent.approve()
assert result["tool"] == "list_files" assert result["tool"] == "list_files"
assert result["is_done"] is False assert result["is_done"] is False
@ -475,6 +342,7 @@ class TestApprove:
# ═════════════════════════════════════════════════════════════════════════════ # ═════════════════════════════════════════════════════════════════════════════
class TestReject: class TestReject:
"""Tests for reject(): user feedback is injected into the history and the pending action is discarded."""
@pytest.fixture @pytest.fixture
def agent(self): def agent(self):
@ -508,66 +376,12 @@ class TestReject:
def test_reject_without_pending_does_not_crash(self, agent): def test_reject_without_pending_does_not_crash(self, agent):
agent.pending_action = None agent.pending_action = None
agent.reject("no pending action") # should not raise agent.reject("no pending action")
def test_reject_does_not_execute_tool(self, agent, tmp_path): def test_reject_does_not_execute_tool(self, agent, tmp_path):
self._set_pending(agent, "write_file") self._set_pending(agent, "write_file")
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path): with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
agent.reject("Do not write anything") agent.reject("Do not write anything")
assert not list(tmp_path.glob("*")) # no files created assert not list(tmp_path.glob("*"))
# ═════════════════════════════════════════════════════════════════════════════
# 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"])

110
tests/test_debug_logger.py Normal file
View File

@ -0,0 +1,110 @@
"""Tests for DebugLogger (backend/managers/debug_logger.py)."""
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from backend.managers.debug_logger import DebugLogger
# ── Fixtures ──────────────────────────────────────────────────────────────────
@pytest.fixture
def logger():
"""Return a fresh DebugLogger for each test."""
return DebugLogger()
# ── log() ─────────────────────────────────────────────────────────────────────
class TestLog:
"""Tests for log(): each call appends an INFO-level entry with message."""
def test_log_appends_entry(self, logger):
logger.log("started")
assert len(logger.logs) == 1
def test_log_sets_level_info(self, logger):
logger.log("started")
assert logger.logs[0]["level"] == "INFO"
def test_log_stores_message(self, logger):
logger.log("executing file.py")
assert logger.logs[0]["message"] == "executing file.py"
# ── log_error() ───────────────────────────────────────────────────────────────
class TestLogError:
"""Tests for log_error(): level is ERROR, not INFO."""
def test_log_error_sets_level_error(self, logger):
logger.log_error("something broke")
assert logger.logs[0]["level"] == "ERROR"
def test_log_and_log_error_are_distinct_levels(self, logger):
logger.log("info message")
logger.log_error("error message")
assert logger.logs[0]["level"] == "INFO"
assert logger.logs[1]["level"] == "ERROR"
# ── get_logs() ────────────────────────────────────────────────────────────────
class TestGetLogs:
"""Tests for get_logs()."""
def test_get_logs_returns_all_entries(self, logger):
logger.log("first")
logger.log_error("second")
assert len(logger.get_logs()) == 2
# ── clear() ───────────────────────────────────────────────────────────────────
class TestClear:
"""Tests for clear()."""
def test_clear_removes_all_entries(self, logger):
logger.log("first")
logger.log_error("second")
logger.clear()
assert logger.logs == []
# ── format_debug_output() ─────────────────────────────────────────────────────
class TestFormatDebugOutput:
"""Tests for format_debug_output(): renders rc, stdout, stderr, and log entries."""
def test_success_exit_code_shows_success(self, logger):
result = logger.format_debug_output({"rc": 0, "stdout": "", "stderr": ""})
assert "[SUCCESS]" in result
def test_nonzero_exit_code_shows_failed(self, logger):
result = logger.format_debug_output({"rc": 1, "stdout": "", "stderr": ""})
assert "[FAILED]" in result
def test_stdout_included_when_present(self, logger):
result = logger.format_debug_output({"rc": 0, "stdout": "Hello", "stderr": ""})
assert "Hello" in result
assert "stdout" in result
def test_stderr_included_when_present(self, logger):
result = logger.format_debug_output({"rc": 1, "stdout": "", "stderr": "NameError"})
assert "NameError" in result
assert "stderr" in result
def test_log_entries_appended_to_output(self, logger):
logger.log("Executing file.py...")
logger.log_error("exit code 1")
result = logger.format_debug_output({"rc": 1, "stdout": "", "stderr": ""})
assert "Executing file.py..." in result
assert "exit code 1" in result
def test_missing_keys_do_not_raise(self, logger):
# Defensive: format_debug_output uses .get() so missing keys are safe
result = logger.format_debug_output({})
assert isinstance(result, str)

220
tests/test_file_manager.py
View File

@ -0,0 +1,220 @@
"""Tests for FileManager (backend/managers/file_manager.py)."""
import sys
from pathlib import Path
import pytest
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).parent.parent))
from backend.managers.file_manager import FileManager
# ── Fixtures ──────────────────────────────────────────────────────────────────
@pytest.fixture(autouse=True)
def mock_streamlit():
"""Suppress all st.error / st.warning calls — they require a running Streamlit app."""
with patch("backend.managers.file_manager.st"):
yield
@pytest.fixture
def fm(tmp_path):
"""Return a FileManager whose workspace is an isolated pytest temp directory."""
return FileManager(base_path=tmp_path)
# ── create_folder ─────────────────────────────────────────────────────────────
class TestCreateFolder:
"""Tests for create_folder(): name validation, path-traversal protection, and nested creation."""
def test_creates_folder_successfully(self, fm, tmp_path):
result = fm.create_folder("", "myfolder")
assert result is True
assert (tmp_path / "myfolder").is_dir()
def test_empty_name_returns_false(self, fm):
assert fm.create_folder("", "") is False
def test_slash_in_name_returns_false(self, fm):
assert fm.create_folder("", "a/b") is False
def test_backslash_in_name_returns_false(self, fm):
assert fm.create_folder("", "a\\b") is False
def test_duplicate_folder_returns_false(self, fm, tmp_path):
(tmp_path / "existing").mkdir()
assert fm.create_folder("", "existing") is False
def test_path_traversal_returns_false(self, fm):
assert fm.create_folder("../../", "evil") is False
def test_nested_folder_created_inside_base(self, fm, tmp_path):
(tmp_path / "sub").mkdir()
result = fm.create_folder("sub", "child")
assert result is True
assert (tmp_path / "sub" / "child").is_dir()
# ── create_file ───────────────────────────────────────────────────────────────
class TestCreateFile:
"""Tests for create_file(): name validation, auto .txt extension, and path-traversal protection."""
def test_creates_file_successfully(self, fm, tmp_path):
result = fm.create_file("", "test.py")
assert result is True
assert (tmp_path / "test.py").is_file()
def test_empty_name_returns_false(self, fm):
assert fm.create_file("", "") is False
def test_whitespace_only_name_returns_false(self, fm):
assert fm.create_file("", " ") is False
def test_no_extension_defaults_to_txt(self, fm, tmp_path):
fm.create_file("", "notes")
assert (tmp_path / "notes.txt").is_file()
def test_duplicate_file_returns_false(self, fm, tmp_path):
(tmp_path / "existing.py").touch()
assert fm.create_file("", "existing.py") is False
def test_path_traversal_returns_false(self, fm):
assert fm.create_file("../../", "evil.py") is False
# ── read_file ─────────────────────────────────────────────────────────────────
class TestReadFile:
"""Tests for read_file(): accepts an absolute Path, validates workspace boundary, returns content or ""."""
def test_reads_file_content(self, fm, tmp_path):
f = tmp_path / "hello.py"
f.write_text("print('hello')")
assert fm.read_file(f) == "print('hello')"
def test_nonexistent_file_returns_empty_string(self, fm, tmp_path):
assert fm.read_file(tmp_path / "ghost.py") == ""
def test_file_outside_workspace_returns_empty_string(self, fm, tmp_path):
outside = tmp_path.parent / "outside.py"
outside.write_text("secret")
assert fm.read_file(outside) == ""
# ── save_file ─────────────────────────────────────────────────────────────────
class TestSaveFile:
"""Tests for save_file(): accepts an absolute path string, overwrites content, and blocks path traversal."""
def test_saves_content_to_file(self, fm, tmp_path):
f = tmp_path / "output.py"
f.touch()
result = fm.save_file(str(f), "x = 1")
assert result is True
assert f.read_text() == "x = 1"
def test_overwrites_existing_content(self, fm, tmp_path):
f = tmp_path / "script.py"
f.write_text("old content")
fm.save_file(str(f), "new content")
assert f.read_text() == "new content"
def test_path_traversal_returns_false(self, fm, tmp_path):
outside = str(tmp_path.parent / "evil.py")
assert fm.save_file(outside, "bad") is False
# ── rename_file ───────────────────────────────────────────────────────────────
class TestRenameFile:
"""Tests for rename_file(): renames by stem only — the original extension is always preserved."""
def test_renames_file_successfully(self, fm, tmp_path):
(tmp_path / "old.py").touch()
result = fm.rename_file("old.py", "new")
assert result is True
assert (tmp_path / "new.py").exists()
assert not (tmp_path / "old.py").exists()
def test_preserves_original_extension(self, fm, tmp_path):
# Even if the caller passes a different extension (.txt), rename_file
# silently replaces it with the original (.py) to prevent accidental type changes.
(tmp_path / "script.py").touch()
fm.rename_file("script.py", "renamed.txt")
assert (tmp_path / "renamed.py").exists()
def test_empty_new_name_returns_false(self, fm, tmp_path):
(tmp_path / "file.py").touch()
assert fm.rename_file("file.py", "") is False
def test_nonexistent_file_returns_false(self, fm):
assert fm.rename_file("ghost.py", "new_name") is False
def test_path_traversal_returns_false(self, fm):
assert fm.rename_file("../../evil.py", "new_name") is False
# ── delete_file ───────────────────────────────────────────────────────────────
class TestDeleteFile:
"""Tests for delete_file(): accepts a relative path, validates workspace boundary, removes the file."""
def test_deletes_file_successfully(self, fm, tmp_path):
f = tmp_path / "todelete.py"
f.touch()
result = fm.delete_file("todelete.py")
assert result is True
assert not f.exists()
def test_nonexistent_file_returns_false(self, fm):
assert fm.delete_file("ghost.py") is False
def test_path_traversal_returns_false(self, fm):
assert fm.delete_file("../../evil.py") is False
# ── delete_folder ─────────────────────────────────────────────────────────────
class TestDeleteFolder:
"""Tests for delete_folder(): recursively removes a folder and all its contents."""
def test_deletes_folder_and_contents(self, fm, tmp_path):
sub = tmp_path / "todelete"
sub.mkdir()
(sub / "file.py").touch()
result = fm.delete_folder("todelete")
assert result is True
assert not sub.exists()
def test_path_traversal_returns_false(self, fm):
assert fm.delete_folder("../../") is False
# ── get_file_tree ─────────────────────────────────────────────────────────────
class TestGetFileTree:
"""Tests for get_file_tree(): returns a nested dict where files map to None and dirs map to dicts."""
def test_empty_workspace_returns_empty_dict(self, fm):
assert fm.get_file_tree() == {}
def test_file_is_represented_as_none(self, fm, tmp_path):
(tmp_path / "main.py").touch()
tree = fm.get_file_tree()
assert tree["main.py"] is None
def test_directory_is_represented_as_dict(self, fm, tmp_path):
(tmp_path / "src").mkdir()
tree = fm.get_file_tree()
assert isinstance(tree["src"], dict)
def test_nested_structure_is_correct(self, fm, tmp_path):
(tmp_path / "src").mkdir()
(tmp_path / "src" / "app.py").touch()
tree = fm.get_file_tree()
assert tree["src"]["app.py"] is None

6
tests/test_main.py
View File

@ -1,6 +0,0 @@
"""Test cases for the main module."""
def test_placeholder():
"""Placeholder test."""
assert True

0
tests/test_mcp_server_code_execution.py Normal file
View File

0
tests/test_mcp_server_file_search.py Normal file
View File

0
tests/test_mcp_server_web_search.py Normal file
View File

38
tests/test_system_prompter.py
View File

@ -12,23 +12,11 @@ from backend.managers.system_prompter import SystemPrompter, MAX_FILE_CHARS
class TestSystemPrompterBasePrompt: class TestSystemPrompterBasePrompt:
"""Tests for generate_prompt() without file context.""" """Tests for generate_prompt() without file context."""
def test_returns_non_empty_string(self):
prompt = SystemPrompter.generate_prompt()
assert isinstance(prompt, str)
assert len(prompt) > 0
def test_describes_code_assistant(self):
prompt = SystemPrompter.generate_prompt()
assert "code assistant" in prompt.lower()
def test_contains_no_file_xml_tag(self): def test_contains_no_file_xml_tag(self):
prompt = SystemPrompter.generate_prompt() prompt = SystemPrompter.generate_prompt()
assert "<file" not in prompt assert "<file" not in prompt
assert "<code>" not in prompt assert "<code>" not in prompt
def test_none_equals_no_argument(self):
assert SystemPrompter.generate_prompt(file_context=None) == SystemPrompter.generate_prompt()
class TestSystemPrompterWithFileContext: class TestSystemPrompterWithFileContext:
"""Tests for generate_prompt() with file_context provided.""" """Tests for generate_prompt() with file_context provided."""
@ -49,15 +37,6 @@ class TestSystemPrompterWithFileContext:
prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": "pass"}) prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": "pass"})
assert "<code>" in prompt assert "<code>" in prompt
def test_with_context_is_longer_than_base(self):
base = SystemPrompter.generate_prompt()
with_ctx = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": "x=1"})
assert len(with_ctx) > len(base)
def test_missing_name_key_uses_unknown(self):
prompt = SystemPrompter.generate_prompt(file_context={"content": "some code"})
assert "unknown" in prompt
def test_missing_content_key_does_not_raise(self): def test_missing_content_key_does_not_raise(self):
prompt = SystemPrompter.generate_prompt(file_context={"name": "empty.py"}) prompt = SystemPrompter.generate_prompt(file_context={"name": "empty.py"})
assert "empty.py" in prompt assert "empty.py" in prompt
@ -77,12 +56,19 @@ class TestSystemPrompterTruncation:
assert "[truncated]" not in prompt assert "[truncated]" not in prompt
assert content in prompt assert content in prompt
def test_file_exactly_at_limit_is_not_truncated(self):
content = "x" * MAX_FILE_CHARS
prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": content})
assert "[truncated]" not in prompt
def test_file_one_over_limit_is_truncated(self): def test_file_one_over_limit_is_truncated(self):
content = "x" * (MAX_FILE_CHARS + 1) content = "x" * (MAX_FILE_CHARS + 1)
prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": content}) prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": content})
assert "[truncated]" in prompt assert "[truncated]" in prompt
class TestSystemPrompterSpecialCharacters:
"""Tests that XML special characters in file content are handled without breaking the prompt."""
def test_xml_tags_in_content_are_preserved_literally(self):
# User code often contains HTML or XML. The prompt builder must embed it
# verbatim — escaping or stripping tags would corrupt the file content.
prompt = SystemPrompter.generate_prompt(
file_context={"name": "template.html", "content": "<div>hello</div>"}
)
assert "<div>hello</div>" in prompt