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

Merge branch 'logger' of https://gitea.fhgr.ch/meulilivio/AISE1_Project into func_improvments

Browse Source
This commit is contained in:
Livio Meuli 2026-05-25 15:00:23 +02:00
commit 3891007fb2
20 changed files with 1257 additions and 223 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -50,3 +50,6 @@ data/raw/
# Workspace
workspace/
# Logs
logs/

76
backend/agent/coding_agent.py
View File

@ -18,9 +18,6 @@ import os
import re
from pathlib import Path
import asyncio
# REVIEW: pprint is only used for a single debug print in build_all_tool_description();
# replace with a plain print() call and remove this import.
import pprint
import requests
from dotenv import load_dotenv
@ -28,13 +25,14 @@ from dotenv import load_dotenv
#from mcp_server_adapter import MCPToolAdapter # Import from current directory for easier testing without package structure
from backend.agent.mcp_server_adapter import MCPToolAdapter
from backend.managers.debug_logger import get_logger
logger = get_logger(__name__)
# ── mcp server initialization ────────────────────────────────────────────────────────────────
adapter = MCPToolAdapter()
# REVIEW: debug print — remove before shipping.
print("MCPToolAdapter created. Listing all tools from servers...")
logger.info("MCPToolAdapter created. Listing all tools from servers...")
asyncio.run(adapter.initialize_all_servers())
# REVIEW: debug print — remove before shipping.
print("listed tools from all servers")
logger.info("Listed tools from all servers")
load_dotenv()
@ -64,13 +62,10 @@ def build_all_tool_description() -> str:
``"- <tool_name>: <description>"``.
"""
all_tools = adapter.get_all_tools()
# REVIEW: debug print — remove before shipping.
print(f"Building tool description for {len(all_tools)} tools.")
logger.info("Building tool description for %s tools.", str(len(all_tools)))
descriptions = []
for tool in all_tools:
# REVIEW: debug print via pprint — remove before shipping; replace pprint import with plain print if kept.
pprint.pprint(f"{tool}")
descriptions.append(f"- {tool['tool_name']}: {tool['tool_description']}")
return "\n".join(descriptions)
@ -97,23 +92,23 @@ async def dispatch_tool(tool_name: str, arguments: dict) -> str:
return f"DONE: {summary}"
try:
print(f"Trying to call tool '{tool_name}' with arguments: {arguments}")
print(f"Trying to call tool '{tool_name}' in dispatch_tool through MCPToolAdapter...")
logger.info("Calling tool '%s' in dispatch_tool through MCPToolAdapter...", tool_name)
result = await adapter.call_tool(tool_name, arguments)
# REVIEW: debug print — remove before shipping.
print(f"Raw result from tool '{tool_name}': {result}")
logger.info(f"Result from tool '%s' recieved", tool_name)
if result.isError:
# MCP servers signal tool-level errors via the isError flag rather
# than raising exceptions, so we surface them explicitly.
texts = [block.text for block in result.content if block.type == "text"]
logger.warning("Result from '%s' is Error", tool_name)
return f"Tool error: {' '.join(texts)}"
texts = [block.text for block in result.content if block.type == "text"]
return "\n".join(texts)
except Exception as e:
logger.exception(f"Error calling tool '%s' with argument: %s", tool_name, arguments)
return f"Error calling tool '{tool_name}': {e}"
# ═════════════════════════════════════════════════════════════════════════════
@ -187,6 +182,7 @@ Example:
def truncate_result(result: str) -> str:
"""Truncate a tool result that exceeds MAX_RESULT_LENGTH."""
logger.info("Result has been truncated")
if len(result) <= MAX_RESULT_LENGTH:
return result
half = MAX_RESULT_LENGTH // 2
@ -210,6 +206,8 @@ def trim_messages(messages: list) -> list:
Returns:
Gekürzte History mit maximal MAX_HISTORY_CHARS Zeichen, immer mit Head + Reminder + Tail.
"""
logger.info("Message is being trimmed")
total = sum(len(m["content"]) for m in messages)
if total <= MAX_HISTORY_CHARS:
return messages
@ -395,17 +393,8 @@ class CodingAgent:
self.model = os.getenv("MODEL")
def _call_api(self, messages: list) -> str:
"""Send the conversation history to the LLM and return its reply text.
Args:
messages: Full conversation history as a list of {role, content} dicts.
Returns:
The raw string content of the assistant's reply.
Raises:
Exception: On non-200 HTTP status or an unexpected response schema.
"""
"""Make a raw API call and return the response content string."""
headers = {"Content-Type": "application/json"}
if self.api_key and self.api_key != "EMPTY":
headers["Authorization"] = f"Bearer {self.api_key}"
@ -420,15 +409,30 @@ class CodingAgent:
"stream": False,
}
response = requests.post(self.api_url, headers=headers, json=payload, timeout=60)
if response.status_code != 200:
raise Exception(f"API Error {response.status_code}: {response.text}")
try:
response = requests.post(
self.api_url,
headers=headers,
json=payload,
timeout=60)
response.raise_for_status()
logger.info("LLM API response requested")
if response.status_code != 200:
logger.error("API Error %s: %s", response.status_code, response.text)
raise Exception(f"API Error {response.status_code}: {response.text}")
except requests.RequestException as exc:
logger.exception("API Error; HTTP-Fehler: %s", exc)
raise Exception(f"HTTP-Fehler: {exc}") from exc
data = response.json()
if "choices" in data and len(data["choices"]) > 0:
logger.info("valid API output, data returned")
return data["choices"][0]["message"]["content"]
logger.error("Invalid API response format")
raise Exception("Invalid API response format")
# ── Public interface ──────────────────────────────────────────────────────
@ -441,6 +445,7 @@ class CodingAgent:
self.pending_action = None
self.is_done = False
self.iteration = 0
logger.info("New ask initialized")
async def propose_next_action(self) -> dict:
"""Ask the LLM what to do next.
@ -467,6 +472,7 @@ class CodingAgent:
raw = _strip_code_fences(raw)
cleaned = extract_json(raw)
action = json.loads(cleaned)
logger.info("Propose next action successfull")
except json.JSONDecodeError:
action = {
"thought": "Could not parse LLM response as JSON.",
@ -474,6 +480,7 @@ class CodingAgent:
"arguments": {"summary": "Stopped: JSON parse error."},
}
raw = json.dumps(action)
logger.critical("Parsing API response into valid JASON failed in Step 'propose_next_action'")
except Exception as e:
action = {
"thought": f"API call failed: {e}",
@ -481,6 +488,7 @@ class CodingAgent:
"arguments": {"summary": f"Stopped: {e}"},
}
raw = json.dumps(action)
logger.critical("API call faliled in Step %s: %s", self.iteration, e)
self.pending_action = {"raw": raw, "action": action}
return action
@ -503,6 +511,8 @@ class CodingAgent:
self.messages.append({"role": "assistant", "content": raw})
self.pending_action = None
logger.info("Messages prepared after approval")
# Handle completion
if tool_name == "done":
self.is_done = True
@ -516,6 +526,7 @@ class CodingAgent:
# Execute the tool
result = await dispatch_tool(tool_name, arguments)
result = truncate_result(result)
logger.info("Tool called and result truncated")
# 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
@ -527,6 +538,7 @@ class CodingAgent:
"Re-examine your plan: what went wrong and what should you do differently? "
"State your revised plan in your next thought.</replan>"
)
logger.warning("Error Message in the tool result, replan-feedback will appended")
self.messages.append({"role": "user", "content": feedback})
@ -554,6 +566,7 @@ class CodingAgent:
"address their question accordingly.</replan>"
),
})
logger.info("Follow-up message appended.")
def reject(self, feedback: str) -> None:
"""Reject the pending action and inject user feedback.
@ -580,6 +593,7 @@ class CodingAgent:
),
})
self.pending_action = None
logger.info("Rejection message appended.")
# REVIEW: dead code — this module is always imported, never run as a script.
# The __main__ guard below is unreachable in normal use. Move this to run_agent.py or delete it.

61
backend/agent/mcp_server_adapter.py
View File

@ -19,6 +19,8 @@ from pathlib import Path
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from backend.managers.debug_logger import get_logger
logger = get_logger(__name__)
class MCPToolAdapter:
"""Discovers and dispatches MCP tools from one or more stdio-based MCP servers.
@ -52,30 +54,25 @@ class MCPToolAdapter:
"""
path = Path(__file__).parent / self.config_path
if not path.exists():
print(f"Config file not found: {path}")
logger.warning("Config file not found: %s", path)
return {}
try:
with open(path, 'r') as f:
return json.load(f)
config_file = json.load(f)
logger.info("MCP-Server config loaded successfully")
return config_file
except json.JSONDecodeError as e:
print(f"Error decoding JSON config: {e}")
logger.critical("Error decoding JSON from server config: %s", e)
return {}
async def initialize_all_servers(self):
"""Connect to every configured MCP server and register their tools.
Opens a short-lived stdio connection to each server, calls list_tools(),
and stores each discovered tool in ``self.tool_registry``. Servers that
fail to connect are skipped with a warning so a single broken server does
not prevent the others from loading.
"""
print("Initializing MCP sessions...")
"""Lädt die Konfiguration und fragt alle Server ab, um die Tools zu registrieren."""
config = self._load_config()
print(f"Loaded config for servers: {list(config.keys())}")
logger.info("Loaded config for servers: %s", list(config.keys()))
for server_name, params in config.items():
print(f"Testing connection to {server_name}...")
logger.info("Initializing connection to %s", server_name)
self.servers[server_name] = params
server_script = str(Path(__file__).parent / params["args"][0])
@ -95,16 +92,15 @@ class MCPToolAdapter:
try:
async with stdio_client(server_params) as (read_stream, write_stream):
print(f"Connected to {server_name}. Initializing session...")
logger.info("Connected to %s. Initializing session...", server_name)
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
print(f"Session initialized for {server_name}. Requesting tools...")
logger.info("Session initialized for %s. Requesting tools...", server_name)
result = await session.list_tools()
# REVIEW: debug print — remove before shipping.
print(f"Tools received from {server_name}: {result}")
tools = result.tools
# REVIEW: duplicate print — identical message already printed inside the `async with` block above.
print(f"Tools received from {server_name}: {result}")
logger.info(f"Tools received from %s: %s Tools", server_name, str(len(tools)))
for tool in tools:
# Build a human-readable parameter description for the system prompt.
@ -127,12 +123,10 @@ class MCPToolAdapter:
"tool_description": t_definition
})
print(f"Registered tool '{tool.name}' from {server_name}.")
print(f"Session for {server_name} ready. {len(tools)} tools found.")
logger.info("Registered tool '%s' from %s.", tool.name, server_name)
except Exception as e:
print(f"Failed to initialize {server_name}: {e}")
logger.exception("Failed to initialize %s: %s", server_name, str(e))
def get_all_tools(self) -> List[Dict[str, Any]]:
"""Return the full list of registered tools across all servers.
@ -160,7 +154,7 @@ class MCPToolAdapter:
tool_entry = next((t for t in self.tool_registry if t["tool_name"] == tool_name), None)
if not tool_entry:
print(f"Tool '{tool_name}' not found in MCP adapter registry.")
logger.warning("Tool '%s' not found in MCP adapter registry.", tool_name)
return f"Error: Tool '{tool_name}' not found in registry."
server_name = tool_entry["server"]
@ -183,32 +177,15 @@ class MCPToolAdapter:
async with stdio_client(server_params) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
logger.info("Session successfully initialized, calling tool '%s' on server '%s", tool_name, server_name)
result = await session.call_tool(tool_name, arguments)
return result
except Exception as e:
logger.exception("Error calling tool '%s' on server '%s': %s", tool_name, server_name, str(e))
return f"Error calling tool '{tool_name}' on server '{server_name}': {str(e)}"
return f"Error: Session for server '{server_name}' not active."
async def shutdown_all_sessions(self):
"""Close all open server connections gracefully.
Note: This method references ``self.exit_stack`` which is not currently
populated (connections are opened per-call). It is kept as a placeholder
for a future persistent-connection implementation.
"""
# REVIEW: self.exit_stack is never assigned in __init__ — calling this method will
# always raise AttributeError. Either remove this method or initialise exit_stack
# in __init__ as an empty dict.
for server_name, (transport_gen, session) in self.exit_stack.items():
try:
await session.__aexit__(None, None, None)
await transport_gen.__aexit__(None, None, None)
print(f"Session for {server_name} shut down.")
except Exception as e:
print(f"Error during shutdown of {server_name}: {e}")
def main():
"""Debug Function for Tool-Registry"""
adapter = MCPToolAdapter()
asyncio.run(adapter.initialize_all_servers())
print("All servers initialized. Registered tools:")

5
backend/agent/mcp_server_config.json
View File

@ -5,10 +5,7 @@
"WebSearchServer": {
"command": "py",
"args": ["servers/mcp_server_web_search.py"],
"env": {
"DDGS_API_KEY": "your_ddgs_api_key_here"
}
"args": ["servers/mcp_server_web_search.py"]
},
"CodeExecutionServer": {

43
backend/agent/servers/mcp_server_code_execution.py
View File

@ -24,8 +24,11 @@ from pyflakes.reporter import Reporter # For linting Code
from mcp.server.fastmcp import FastMCP
from pathlib import Path
#from backend.managers.debug_logger import get_logger
#logger = get_logger(__name__)
# ── Configuration ────────────────────────────────────────────────────────────
EXEC_TIMEOUT = 45 # seconds before killing the subprocess
EXEC_TIMEOUT = 15 # seconds before killing the subprocess
MAX_OUTPUT_LENGTH = 3000 # max characters of stdout+stderr to return
# ── Create the MCP server ────────────────────────────────────────────────────
@ -64,8 +67,9 @@ FORBIDDEN_SEQUENCES = ["../", "..\\", "/etc/", "/dev/",
"C:\\Windows", "C:\\Program Files", "C:\\Users",
"compile(", "__import__", "os.", "sys.", "subprocess."]
ALLOWED_PACKAGES = ["pygame", "numpy", "pandas"]
"""
Pre-installed Packages in Sandbox: "pygame", "numpy", "pandas"
"""
# ── Static Analysis ────────────────────────────────────────────────────
def check_code_safety(code: str) -> str | None:
"""
@ -78,10 +82,13 @@ def check_code_safety(code: str) -> str | None:
str or None
Error message if forbidden code found, None if safe.
"""
#logger.info("Checking code safety.")
try:
tree = ast.parse(code)
#logger.info("Code has valid Syntax")
except SyntaxError as e:
#logger.exception("SyntaxError: %s", e)
return f"SyntaxError: {e}"
for node in ast.walk(tree):
@ -89,6 +96,7 @@ def check_code_safety(code: str) -> str | None:
for alias in node.names:
top_level_module = alias.name.split('.')[0]
if top_level_module in BLOCKED_IMPORTS:
#logger.warning("Blocked import '%s'", alias.name)
return (f"Blocked import: Import of '{alias.name}' is not allowed."
f"line {node.lineno}")
@ -96,16 +104,19 @@ def check_code_safety(code: str) -> str | None:
if node.module:
top_level = node.module.split(".")[0]
if top_level in BLOCKED_IMPORTS:
#logger.warning("Blocked import from '%s'", alias.name)
return (f"Blocked import: Import from '{node.module}' is not allowed."
f"(module '{top_level}' is blocked) line {node.lineno}")
elif isinstance(node, ast.Call):
if isinstance(node.func, ast.Name):
if node.func.id in BLOCKED_BUILTINS:
#logger.warning("Blocked ubiltin '%s'", node.func.id)
return f"Blocked builtin: Use of builtin '{node.func.id}' is not allowed."
for seq in FORBIDDEN_SEQUENCES:
if seq in code:
#logger.warning("Suspect path sequence '%s' detected.", seq)
return f"Blocked: Suspect path sequence '{seq}' detected."
return None # No violations found
@ -120,11 +131,16 @@ def analyse_structure(code: str) -> str:
Returns:
A summary of the code's structure, including functions, classes, and imports.
"""
#logger.info("Tool analyse_structure is being executed on MCP code execution server")
try:
tree = ast.parse(code)
#logger.info("Code tree parsed successfully")
except SyntaxError as e:
#logger.warning("Syntax Error in provided code. Line %s : %s", e.lineno, e.msg)
return f"Syntax Error: Invalid Python code provided. Line {e.lineno}: {e.msg}"
except Exception as e:
#logger.exception("Error parsing code: %s", str(e))
return f"Error parsing code: {str(e)}"
analysis = {
@ -180,6 +196,7 @@ def analyse_structure(code: str) -> str:
lines.append(f" - def {func['name']}({args_str})")
if not any([analysis["imports"], analysis["classes"], analysis["functions"]]):
#logger.info("Code analysis successfull but no top-level items found")
return "Analysis complete: No top-level imports, classes, or functions found."
return "\n".join(lines)
@ -196,6 +213,8 @@ def lint_code(code: str) -> str:
Returns:
A report of linting issues or a success message if the code is clean.
"""
#logger.info("Tool lint_code is being executed on MCP code execution server")
error_buffer = io.StringIO()
warning_buffer = io.StringIO()
@ -203,7 +222,9 @@ def lint_code(code: str) -> str:
try:
check(code, filename="<agent_code>", reporter=reporter)
#logger.info("Linting successfull")
except Exception as e:
#logger.exception("Critical error during linting: %s", str(e))
return f"Critical error during linting: {str(e)}"
errors = error_buffer.getvalue().strip()
@ -211,6 +232,7 @@ def lint_code(code: str) -> str:
# Ergebnis-String zusammenbauen
if not errors and not warnings:
#logger.info("No issues found")
return "Linting complete: No issues found. The code is syntactically sound."
report = ["--- Linting Report ---"]
@ -225,6 +247,7 @@ def lint_code(code: str) -> str:
report.append("\nAdvice: Please fix these issues before attempting to execute the code.")
#logger.info("There are issues with provided code. Check Report: %s", "\n".join(report))
return "\n".join(report)
@mcp.tool()
@ -244,6 +267,7 @@ def run_python_sandboxed(code: str) -> str:
Returns:
Combined stdout+stderr, or an error message in str format.
"""
#logger.info("Tool run_python_sandboxed is being executed on MCP code execution server")
# Reject code that references blocked modules or builtins before spawning a process.
static_safety = check_code_safety(code)
@ -266,12 +290,16 @@ def run_python_sandboxed(code: str) -> str:
if not output.strip():
return "Code executed successfully (no output)."
#logger.info("Code ran successfully")
return output
except subprocess.TimeoutExpired:
#logger.warning("Code execution exceeded time limit of %s seconds", EXEC_TIMEOUT)
return f"Error: Code execution exceeded time limit of {EXEC_TIMEOUT} seconds and was terminated."
except Exception as e:
#logger.exception("Error during code execution: %s", str(e))
return f"Error during code execution: {e}"
@mcp.tool()
@ -290,17 +318,22 @@ def python_code_validation(code: str) -> str:
the first violation found. Returns None implicitly when the code is
both syntactically valid and safe (no safety concerns found).
"""
#logger.info("Tool python_code_validation is being executed on MCP code execution server")
try:
ast.parse(code)
#logger.info("Ast parsing successfull")
except SyntaxError as e:
#logger.warning("Syntax Error while ast parsing code: %s", str(e))
return f"SyntaxError: {e}"
try:
static_analysis_result = check_code_safety(code)
if static_analysis_result:
# Inform the agent that the code would be rejected by run_python_code_sandboxed.
#logger.info("Code safety issues detected")
return f"Valid Syntax, but with safety concerns: {static_analysis_result}; code execution is not allowed."
except Exception as e:
#logger.exception("Error during code safety analysis: %e", str(e))
return f"Error during code safety analysis: {e}"
return "Code is valid and can be executed in the sandbox"

58
backend/agent/servers/mcp_server_file_search.py
View File

@ -16,6 +16,9 @@ Exposes the following MCP tools:
from pathlib import Path
from mcp.server.fastmcp import FastMCP
#from backend.managers.debug_logger import get_logger
#logger = get_logger(__name__)
# ── Configuration ────────────────────────────────────────────────────────────
# Navigate four levels up from servers/ to the project root, then into workspace/.
project_dir = Path(__file__).resolve().parent.parent.parent.parent
@ -32,10 +35,12 @@ def _safe_path(requested: str) -> Path:
"""Resolve and validate a path is inside ALLOWED_DIR."""
resolved = (ALLOWED_DIR / requested).resolve()
if not str(resolved).startswith(str(ALLOWED_DIR)):
#logger.warning("Access denied: '%s' resolves outside allowed directory.", str(requested))
raise ValueError(
f"Access denied: '{requested}' resolves outside "
f"the allowed directory '{ALLOWED_DIR}'"
)
#logger.info("Requested path is safe")
return resolved
@ -47,6 +52,8 @@ def list_files() -> str:
Returns a newline-separated list of relative file paths.
"""
#logger.info("Tool list_files is being executed on MCP file search server")
files = sorted(
f.relative_to(ALLOWED_DIR)
for f in ALLOWED_DIR.rglob("*")
@ -67,6 +74,7 @@ def get_file_tree(dir_path: str=ALLOWED_DIR) -> str:
Returns:
A string representing the directory structure, similar to 'tree' command output.
"""
#logger.info("Tool get_file_tree is being executed on MCP file search server")
try:
safe_dir = _safe_path(dir_path)
# REVIEW: unreachable code — _safe_path() always returns a Path object (never None/falsy)
@ -74,13 +82,16 @@ def get_file_tree(dir_path: str=ALLOWED_DIR) -> str:
if not safe_dir:
return f"Error: Invalid directory path '{dir_path}'."
elif not safe_dir.exists():
#logger.warning("Directory '%s' does not exist.", dir_path)
return f"Error: Directory '{dir_path}' does not exist."
elif not safe_dir.is_dir():
#logger.warning("'%s' is not a valid directory", dir_path)
return f"Error: '{dir_path}' is not a valid directory within the allowed path."
except ValueError as e:
#logger.exception("Error while checking directory and its path: %s", str(e))
return f"Error: {e}"
#logger.info("Generating file tree.")
def _tree(dir_path: Path, prefix="") -> str:
"""Recursively build a tree string for the given directory.
@ -110,11 +121,7 @@ def get_file_tree(dir_path: str=ALLOWED_DIR) -> str:
lines.append(_tree(entry, prefix + extension))
return "\n".join(lines)
# REVIEW: redundant — get_file_tree() passes `dir_path` (original str argument) to `_tree()`
# rather than the validated `safe_dir` (resolved Path). If dir_path is a relative string,
# the inner `_tree()` will call `dir_path.iterdir()` on a str, causing an AttributeError.
# Should pass `safe_dir` instead.
return _tree(dir_path)
return _tree(Path(dir_path))
@mcp.tool()
@ -127,6 +134,8 @@ def search_files(query: str) -> str:
Returns:
A formatted string of search results, or a message if no matches found.
"""
#logger.info("Tool search_files is being executed on MCP file search server")
query_lower = query.lower()
results = []
@ -144,11 +153,17 @@ def search_files(query: str) -> str:
if query_lower in line.lower():
snippet = line.strip()[:100]
results.append(f"[content] {rel}:{i} -- {snippet}")
except (UnicodeDecodeError, PermissionError):
except UnicodeDecodeError as e:
#logger.warning("Decode error in file/folder '%s': %s", f, e)
pass
except PermissionError as e:
#logger.warning("Permission error in file/folder: '%s': %s", f, e)
pass
if not results:
#logger.info("No matches found for user query")
return f"No matches found for '{query}'."
#logger.info("Result returned, limited to 30 matches.")
return "\n".join(results[:30]) # limit to 30 matches
@ -162,23 +177,33 @@ def read_file(path: str) -> str:
Returns:
The file content as a string, or an error message if the file cannot be read.
"""
#logger.info("Tool read_file is being executed on MCP file search server")
try:
resolved = _safe_path(path)
except ValueError as e:
return f"Error: {e}"
if not resolved.exists():
#logger.warning("File '%s' does not exist.", path)
return f"Error: File '{path}' does not exist."
if not resolved.is_file():
#logger.warning("'%s' is not a valid file.", path)
return f"Error: '{path}' is not a file."
try:
return resolved.read_text(encoding="utf-8")
text = resolved.read_text(encoding="utf-8")
#logger.info("File read successfully.")
return text
except UnicodeDecodeError:
#logger.warning("'%s' is not a text file (binary content).", path)
return f"Error: '{path}' is not a text file (binary content)."
except PermissionError:
#logger.warning(f"Permission denied when trying to read '%s'.", path)
return f"Error: Permission denied when trying to read '{path}'."
except Exception as e:
#logger.exception("Error reading file '%s': %s", path, e)
return f"Error reading file '{path}': {e}"
@mcp.tool()
@ -193,6 +218,7 @@ def write_new_file(path: str, content: str) -> str:
Returns:
A success or error message.
"""
#logger.info("Tool write_new_file is being executed on MCP file search server")
try:
resolved = _safe_path(path)
@ -200,26 +226,30 @@ def write_new_file(path: str, content: str) -> str:
return f"Error: {e}"
if resolved.exists():
#logger.warning("Requested file path '%s' already exists, overwriting not allowed.", path)
return (f"ERROR: File '{path}' already exists."
f"Overwriting is not allowed with this tool."
f"Use a different path or filename to create a new file.")
if resolved.suffix not in ALLOWED_FILE_TYPES:
#logger.warning("File type not allowed: %s", resolved.suffix)
return f"ERROR: can only write {', '.join(ALLOWED_FILE_TYPES)} types, got '{resolved.suffix}'."
try:
resolved.parent.mkdir(parents=True, exist_ok=True)
resolved.write_text(content, encoding="utf-8")
#logger.info("File written successfully.")
return f"OK: wrote {len(content)} chars to {path}."
except FileNotFoundError as e:
print(f"FileNotFoundError for {path}: {e}")
#logger.warning("FileNotFoundError for '%s': %s", path, e)
return f"Error: {e}"
except PermissionError as e:
print(f"PermissionError for {path}: {e}")
#logger.warning("PermissionError for '%s': %s", path, e)
return f"Error: {e}"
except Exception as e:
#logger.exception("Error writing file: %s", e)
return f"Error: {e}"
@ -233,25 +263,31 @@ def create_new_directory(path: str) -> str:
Returns:
A success or error message.
"""
#logger.info("Tool create_new_directory is being executed on MCP file search server")
try:
resolved = _safe_path(path)
except ValueError as e:
return f"Error: {e}"
if resolved.exists():
#logger.warning("Requested path '%s' already exists, overwriting not allowed.", path)
return f"Error: File '{path}' already exists."
# REVIEW: redundant — `resolved.suffix != None` is always True (Path.suffix always returns str);
# the None check is unnecessary. Simplify to `if resolved.suffix != "":`.
if resolved.suffix != None and resolved.suffix != "":
#logger.warning("Can only create directories, got '%s'.", resolved.suffix)
return f"Error: can only create directories, got '{resolved.suffix}'."
try:
resolved.parent.mkdir(parents=True, exist_ok=True)
resolved.mkdir()
#logger.info("Directory '%s' created successfully.", path)
return f"OK: created empty directory at {path}."
except Exception as e:
return f"Error creating dictionary file '{path}': {e}"
#logger.exception("Error creating directory '%s': %s", path, e)
return f"Error creating directory '{path}': {e}"
# ── Run the server ───────────────────────────────────────────────────────────

68
backend/agent/servers/mcp_server_web_search.py
View File

@ -9,12 +9,28 @@ schemes and private/loopback IP ranges to prevent SSRF vulnerabilities.
"""
from urllib.parse import urlparse
import requests
from bs4 import BeautifulSoup
from ddgs import DDGS
from mcp.server.fastmcp import FastMCP
#from backend.managers.debug_logger import get_logger
#logger = get_logger(__name__)
# ── Configuration ────────────────────────────────────────────────────────────
MAX_PAGE_LENGTH = 4000 # max characters to return from a fetched page
REQUEST_TIMEOUT = 10 # seconds
# ── Bolcked prefixes & Hosts ────────────────────────────────────────────────────
PRIVATE_PREFIXES = [
"10.", "172.16.", "172.17.", "172.18.", "172.19.",
"172.20.", "172.21.", "172.22.", "172.23.", "172.24.",
"172.25.", "172.26.", "172.27.", "172.28.", "172.29.",
"172.30.", "172.31.", "192.168.",
]
BLOCKED_HOSTS = ["localhost", "127.0.0.1", "0.0.0.0", "169.254.169.254"]
# ── Create the MCP server ────────────────────────────────────────────────────
mcp = FastMCP("WebSearchServer")
@ -39,29 +55,23 @@ def _validate_url(url: str) -> str:
ValueError: If the URL fails any of the security checks.
"""
parsed = urlparse(url)
#logger.info("Validateing URL")
if parsed.scheme not in ("http", "https"):
#logger.warning("Blocked scheme '%s'. Only http and https are allowed.", parsed.scheme)
raise ValueError(
f"Blocked scheme '{parsed.scheme}'. Only http and https are allowed."
)
hostname = parsed.hostname or ""
# Block well-known loopback and cloud-metadata addresses.
blocked_hosts = {"localhost", "127.0.0.1", "0.0.0.0", "169.254.169.254"}
if hostname in blocked_hosts:
if hostname in BLOCKED_HOSTS:
#logger.warning("Blocked internal host: %s", hostname)
raise ValueError(f"Blocked internal host: {hostname}")
# Block all private RFC-1918 ranges by checking the string prefix.
# This is a best-effort check; a full implementation would resolve DNS first.
private_prefixes = (
"10.", "172.16.", "172.17.", "172.18.", "172.19.",
"172.20.", "172.21.", "172.22.", "172.23.", "172.24.",
"172.25.", "172.26.", "172.27.", "172.28.", "172.29.",
"172.30.", "172.31.", "192.168.",
)
for prefix in private_prefixes:
for prefix in PRIVATE_PREFIXES:
if hostname.startswith(prefix):
#logger.warning("Blocked private IP range: %s", hostname)
raise ValueError(f"Blocked private IP range: {hostname}")
return url
@ -79,12 +89,16 @@ def web_search(query: str, max_results: int = 5) -> str:
Returns:
A formatted string of search results, or a message if no matches found.
"""
#logger.info("Tool web_search is being executed on MCP web search server")
try:
from ddgs import DDGS
results = DDGS().text(query, max_results=max_results)
if not results:
#logger.info("DDGS API call successful, no web search results found.")
return f"No results found for: {query}"
#logger.info("DDGS API call successfull, web search results returned.")
formatted = []
for r in results:
@ -96,6 +110,7 @@ def web_search(query: str, max_results: int = 5) -> str:
return "\n---\n".join(formatted)
except Exception as e:
#logger.exception("DDGS API call failed, web search error: %s", e)
return f"Search error: {e}"
@ -108,24 +123,36 @@ def fetch_page(url: str) -> str:
Returns:
The text content of the fetched page, or an error message.
"""
#logger.info("Tool fetch_page is being executed on MCP web search server")
try:
url = _validate_url(url)
except ValueError as e:
return f"URL blocked: {e}"
try:
import requests
from bs4 import BeautifulSoup
response = requests.get(
url,
timeout=REQUEST_TIMEOUT,
headers={"User-Agent": "Mozilla/5.0 (Lightweight Web Search MCP Server)"},
)
response.raise_for_status()
if response.status_code != 200:
#logger.warning("HTTP error %s while fetching %s", response.status_code, url)
return f"HTTP error {response.status_code} fetching {url}"
#logger.info("DDGS API call successfull")
except requests.RequestException as e:
#logger.warning("HTTP-Fehler: %s", e)
return f"HTTP-Fehler: {e}"
except Exception as e:
#logger.exception("Error fetching page: %s", e)
return f"Error fetching page: {e}"
try:
soup = BeautifulSoup(response.text, "html.parser")
# Remove boilerplate elements that add noise without informational value.
@ -134,13 +161,16 @@ def fetch_page(url: str) -> str:
text = soup.get_text(separator="\n", strip=True)
#logger.info("HTML parsing with BeautifulSoup successfull")
if len(text) > MAX_PAGE_LENGTH:
text = text[:MAX_PAGE_LENGTH] + "\n\n[... truncated ...]"
return text if text else "Page fetched but no text content found."
except Exception as e:
return f"Error fetching page: {e}"
#logger.exception("Error parsing HTML: %s", e)
return f"Error parsing html: {e}"
# ── Run the server ───────────────────────────────────────────────────────────

71
backend/managers/_debug_logger.py Normal file
View File

@ -0,0 +1,71 @@
from datetime import datetime
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):
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 = []
def format_debug_output(self, output: dict) -> str:
"""Format an ExecutionEngine result dict into a human-readable string.
Args:
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)
if __name__ == "__main__":
logger = DebugLogger()
print(logger.get_logs())

76
backend/managers/chat_manager.py
View File

@ -5,6 +5,9 @@ from dotenv import load_dotenv
import requests
import json
from backend.managers.debug_logger import get_logger
logger = get_logger(__name__)
load_dotenv()
@ -40,6 +43,7 @@ class ChatManager:
# REVIEW: dead code — clear_history() is never called anywhere in the codebase.
def clear_history(self) -> None:
"""Wipe the conversation history (starts a fresh chat)."""
logger.info("Chat history was cleared")
self.chat_history = []
def send_message(self, user_message: str) -> str:
@ -51,25 +55,27 @@ class ChatManager:
# Add user message to history
self.add_message("user", user_message)
logger.info("Sending message to LLM API")
# Prepare request to OpenAI-compatible API
headers = {
"Content-Type": "application/json",
}
# Add API key if available
if self.api_key and self.api_key != "EMPTY":
headers["Authorization"] = f"Bearer {self.api_key}"
# Full history is sent so the model has multi-turn conversation context
payload = {
"model": self.model,
"messages": self.chat_history,
"temperature": 0.7,
"max_tokens": 2000,
"stream": False,
}
try:
# Prepare request to OpenAI-compatible API
headers = {
"Content-Type": "application/json",
}
# Add API key if available
if self.api_key and self.api_key != "EMPTY":
headers["Authorization"] = f"Bearer {self.api_key}"
# Full history is sent so the model has multi-turn conversation context
payload = {
"model": self.model,
"messages": self.chat_history,
"temperature": 0.7,
"max_tokens": self.max_tokens,
"stream": False,
}
# Make API request
response = requests.post(
self.api_url, headers=headers, json=payload, timeout=30
@ -77,9 +83,24 @@ class ChatManager:
# Check if request was successful
if response.status_code != 200:
error_msg = f"API Error {response.status_code}: {response.text}"
raise Exception(error_msg)
logger.warning("API HTTP status error %s: %s", response.status_code, response.text)
raise Exception(f"API Error {response.status_code}")
logger.info("Response recieved from API")
except requests.exceptions.Timeout as e:
error_msg = f"Timeout Error: {str(e)}"
self.add_message("assistant", f"Error: {error_msg}")
logger.exception("LLM API timeout: %s", e)
raise RuntimeError("LLM API timeout") from e
except requests.exceptions.RequestException as e:
error_msg = f"Connection Error: {str(e)}"
self.add_message("assistant", f"Error: {error_msg}")
logger.exception("LLM API connection failed: %s", e)
raise RuntimeError("LLM API connection failed") from e
try:
# Parse response
response_data = response.json()
@ -90,23 +111,24 @@ class ChatManager:
# Add AI response to history
self.add_message("assistant", ai_message)
logger.info("Assistant response generated")
return ai_message
else:
logger.warning("Invalid API response format: %s", response_data)
raise Exception("Invalid API response format")
except requests.exceptions.RequestException as e:
error_msg = f"Connection Error: {str(e)}"
# Add error message to history so user sees it
self.add_message("assistant", f"Error: {error_msg}")
raise Exception(error_msg)
except json.JSONDecodeError as e:
error_msg = f"JSON Decode Error: {str(e)}"
self.add_message("assistant", f"Error: {error_msg}")
logger.exception("JSON Decode Error: %s", e)
raise Exception(error_msg)
except Exception as e:
error_msg = f"Error: {str(e)}"
self.add_message("assistant", f"Error: {error_msg}")
raise Exception(error_msg)
logger.exception("JSON parsing and message formatting failed: %s", e)
raise RuntimeError("JSON parsing and message formatting failed") from e
# REVIEW: dead code — get_chat_display() is never called anywhere in the codebase.
# The UI renders st.session_state.chat_history directly. This method also does the

127
backend/managers/debug_logger.py
View File

@ -1,67 +1,94 @@
from datetime import datetime
"""
Central logging setup for the application.
- Provides a unified logger via get_logger(__name__)
- Writes all logs to a central rotating log file (logs/app.log)
- Writes errors separately to logs/errors.log
- Automatically includes the module name in each log entry
- Supports standard logging levels: DEBUG, INFO, WARNING, ERROR, CRITICAL
Usage:
from backend.managers.debug_logger import get_logger
logger = get_logger(__name__)
logger.info("Service started")
logger.debug("Debug details")
logger.error("Something went wrong")
try:
...
except Exception:
logger.exception("Unexpected error")
Logging levels (use consistently):
DEBUG: Detailed technical info for developers (variables, flow, internal state).
INFO: Normal application events (start/stop, successful operations, key milestones).
WARNING: Something unexpected happened, but the program continues normally.
ERROR: A specific operation failed, but the application is still running.
CRITICAL: A severe failure that may stop the application or make it unusable.
EXCEPTION: Same as ERROR, but used inside an `except` block and includes stacktrace
(via logger.exception()).
"""
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
BASE_DIR = Path(__file__).resolve().parent.parent
LOG_DIR = BASE_DIR / "logs"
LOG_DIR.mkdir(exist_ok=True)
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.
"""
_initialized = False
def __init__(self):
self.logs: list[dict] = []
@classmethod
def setup(cls):
# prevents multiple setup
if cls._initialized:
return
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"),
})
formatter = logging.Formatter(
"%(asctime)s [%(levelname)s] [%(name)s: Line %(lineno)d] %(message)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"),
})
# Main log file
file_handler = RotatingFileHandler(
LOG_DIR / "app.log",
maxBytes=5_000_000,
backupCount=5,
encoding="utf-8"
)
def get_logs(self) -> list[dict]:
"""Return a copy of all collected log entries."""
return list(self.logs)
file_handler.setFormatter(formatter)
def clear(self) -> None:
"""Reset the log — call before each new execution."""
self.logs = []
# Separate Error-Log
error_handler = RotatingFileHandler(
LOG_DIR / "errors.log",
maxBytes=5_000_000,
backupCount=3,
encoding="utf-8"
)
# REVIEW: dead code — format_debug_output() is never called anywhere in the codebase.
def format_debug_output(self, output: dict) -> str:
"""Format an ExecutionEngine result dict into a human-readable string.
error_handler.setLevel(logging.ERROR)
error_handler.setFormatter(formatter)
Args:
output: dict with keys 'stdout', 'stderr', and 'rc'.
root_logger = logging.getLogger()
Returns:
A formatted string ready for display in the UI.
"""
lines = []
root_logger.setLevel(logging.DEBUG)
status = "SUCCESS" if output.get("rc") == 0 else "FAILED"
lines.append(f"[{status}] Exit code: {output.get('rc')}")
root_logger.addHandler(file_handler)
root_logger.addHandler(error_handler)
#root_logger.propagate = False
if output.get("stdout"):
lines.append("\n--- stdout ---")
lines.append(output["stdout"].rstrip())
cls._initialized = True
if output.get("stderr"):
lines.append("\n--- stderr ---")
lines.append(output["stderr"].rstrip())
@classmethod
def get_logger(cls, name: str):
cls.setup()
return logging.getLogger(name)
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)
# praktische shortcut function
def get_logger(name: str):
return DebugLogger.get_logger(name)

9
backend/managers/execution_engine.py
View File

@ -8,6 +8,9 @@ processes from blocking the UI indefinitely.
import subprocess
from pathlib import Path
from backend.managers.debug_logger import get_logger
logger = get_logger(__name__)
# Maximum time (seconds) a subprocess is allowed to run before being killed.
RUN_TIMEOUT = 30
@ -48,6 +51,8 @@ class ExecutionEngine:
]
else:
return {"stdout": "", "stderr": f"Unsupported file type: {suffix}", "rc": 1}
logger.info("Running file %s with suffix %s", active_file.name, suffix)
try:
proc = subprocess.run(
@ -57,12 +62,16 @@ class ExecutionEngine:
text=True,
timeout=RUN_TIMEOUT,
)
logger.info("File ran successfully.")
return {"stdout": proc.stdout, "stderr": proc.stderr, "rc": proc.returncode}
except subprocess.TimeoutExpired:
logger.warning("Time out afte %s s", RUN_TIMEOUT)
return {"stdout": "", "stderr": f"Timed out after {RUN_TIMEOUT}s", "rc": -1}
except FileNotFoundError as e:
# Raised when the interpreter/compiler binary is not found on PATH
logger.warning("Interpreter/compiler binary is not found on PATH: %s", e)
return {"stdout": "", "stderr": str(e), "rc": -1}
except Exception as e:
logger.exception("Error while running %s: %s", active_file.name, e)
return {"stdout": "", "stderr": str(e), "rc": -1}

60
backend/managers/file_manager.py
View File

@ -7,8 +7,10 @@ touching the filesystem, preventing path-traversal attacks.
import streamlit as st
from pathlib import Path
# REVIEW: dead code — module-level WORKSPACE constant is never used anywhere in this file or
# the rest of the codebase. FileManager.__init__ creates the workspace via self.base_path.mkdir().
from backend.managers.debug_logger import get_logger
logger = get_logger(__name__)
# The workspace folder is created at module load so it always exists.
WORKSPACE = Path("workspace")
WORKSPACE.mkdir(exist_ok=True)
@ -38,12 +40,16 @@ class FileManager:
Returns:
bool: True if folder was created successfully, False otherwise.
"""
logger.info("Creating folder at %s named %s", relative_path, name)
if not name:
logger.warning("Invalid folder name")
st.error(f"Invalid folder name: {name}")
return False
# Slashes in the name would silently create nested paths — reject them.
if "/" in name or "\\" in name:
logger.warning("'/' or '\\' in foldername not allowed")
st.error(f"Invalid folder name (no slashes allowed): {name}")
return False
@ -62,11 +68,14 @@ class FileManager:
try:
folder_path.mkdir(exist_ok=False)
logger.info("Folder created successfully.")
return True
except FileExistsError:
logger.warning("Folder already exists.")
st.warning(f"Folder already exists: {relative_path}")
return False
except Exception as e:
logger.exception("Error creating folder %s: %s", relative_path, str(e))
st.error(f"Error creating folder {relative_path}: {str(e)}")
return False
@ -81,14 +90,18 @@ class FileManager:
name (str): The name of the new file to create (should not contain slashes).
Returns:
bool: True if file was created successfully, False otherwise.
"""
"""
logger.info("Creating file at %s named %s", relative_path, name)
if not name or name.strip() == "" :
logger.warning("Invalid folder name")
st.error(f"Invalid file name: {name}")
return False
name = Path(name)
if not name.suffix:
name = name.with_suffix(".txt") # Default to .txt if no extension provided
logger.info("No suffix was provided, creating .txt file")
if relative_path:
relative_path = Path(relative_path)
@ -104,11 +117,14 @@ class FileManager:
try:
file_path.touch(exist_ok=False)
logger.info("File created successfully.")
return True
except FileExistsError:
logger.warning("Folder already exists.")
st.warning(f"File already exists: {relative_path}")
return False
except Exception as e:
logger.exception("Error creating file %s: %s", relative_path, str(e))
st.error(f"Error creating file {relative_path}: {str(e)}")
return False
@ -123,29 +139,37 @@ class FileManager:
Returns:
str: The content of the file, or an empty string if there was an error.
"""
logger.info("Reading file at %s.", relative_path)
file_path = (relative_path).resolve()
if not file_path.exists():
st.error(f"File not found: {relative_path}")
logger.warning("Filepath does not exist.")
return ""
if not file_path.is_file():
st.error(f"Path is not a file: {relative_path}")
logger.warning("Path is not a file.")
return ""
# Ensure the resolved path is still inside the workspace (prevents path traversal).
if not str(file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}")
logger.warning("Access denied. File ist outside WORKSPACE")
return ""
try:
with open(file_path, "r") as f:
return f.read()
content = f.read()
logger.info("File read successfully.")
return content
except FileNotFoundError:
# REVIEW: unreachable code — FileNotFoundError cannot be raised here because
# `file_path.exists()` is already checked above and returns "" on failure.
st.error(f"File not found: {relative_path}")
logger.warning("File not found")
return ""
except Exception as e:
st.error(f"Error reading file {relative_path}: {str(e)}")
logger.exception("Error reading file at %s: %s", relative_path, e)
return ""
def save_file(self, relative_path: str, content: str) -> bool:
@ -160,19 +184,24 @@ class FileManager:
Returns:
bool: True if save was successful, False otherwise.
"""
logger.info("Saving file at %s.", relative_path)
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())):
st.error(f"Access denied: {relative_path}")
logger.warning("Access denied. File outside WORKSPACE.")
return False
try:
with open(file_path, "w") as f:
f.write(content)
logger.info("File written successfully.")
return True
except Exception as e:
st.error(f"Error saving file {relative_path}: {str(e)}")
logger.exception("Error saving file %s: %s", relative_path, e)
return False
def rename_file(self, old_relative_path: str, new_name: str) -> bool:
@ -186,8 +215,11 @@ class FileManager:
Returns:
bool: True if rename was successful, False otherwise.
"""
logger.info("Rename file at %s to %s.", old_relative_path, new_name)
if not new_name or new_name.strip() == "":
st.error(f"Invalid file name: {new_name}")
logger.warning("New Name is empty.")
return False
file_type = Path(old_relative_path).suffix
@ -203,13 +235,20 @@ class FileManager:
# 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())):
st.error(f"Access denied: {old_relative_path}")
logger.warning("Access denied, file outside WORKSPACE.")
return False
try:
old_file_path.rename(new_file_path)
logger.info("Renamed successfully.")
return True
except FileNotFoundError:
st.error(f"File not found: {old_relative_path}")
logger.warning("Original file not found.")
return False
except Exception as e:
st.error(f"Error renaming file {old_relative_path} to {new_name}: {str(e)}")
logger.exception("Error deleting folder %s to %s: %s", old_relative_path, new_name, str(e))
return False
@ -222,23 +261,29 @@ class FileManager:
Returns:
bool: True if deletion was successful, False otherwise.
"""
logger.info("Deleting folder %s.", relative_path)
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())):
st.error(f"Access denied: {relative_path}")
logger.warning("Access denied, folder outside WORKSPACE.")
return False
if not folder_path.exists():
st.error(f"Folder not found: {relative_path}")
logger.warning("Folder path not found.")
return False
try:
import shutil
shutil.rmtree(folder_path)
logger.info("Folder deleted successfully.")
return True
except Exception as e:
st.error(f"Error deleting folder {relative_path}: {str(e)}")
logger.exception("Error deleting folder %s: %s", relative_path, str(e))
return False
def delete_file(self, relative_path: str) -> bool:
@ -250,21 +295,26 @@ class FileManager:
Returns:
bool: True if deletion was successful, False otherwise.
"""
logger.info("Deleting file %s.", relative_path)
file_path = Path(relative_path)
abs_file_path = (Path(self.base_path) / file_path).resolve()
if not str(abs_file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}")
logger.warning("Access denied, file outside WORKSPACE.")
return False
try:
abs_file_path.unlink()
logger.info("File deleted successfully.")
return True
except FileNotFoundError:
st.error(f"File not found: {relative_path}")
logger.warning("File not found")
return False
except Exception as e:
st.error(f"Error deleting file {relative_path}: {str(e)}")
logger.exception("Error deleting folder %s: %s", relative_path, str(e))
return False
def get_file_tree(self):
@ -276,6 +326,8 @@ class FileManager:
Returns:
dict: A nested dictionary representing the file tree.
"""
logger.info("Getting file tree ...")
def build_tree(path: Path):
tree = {}

5
backend/managers/system_prompter.py
View File

@ -1,5 +1,8 @@
"""Builds the system prompt that is sent to the AI at the start of each chat session."""
from backend.managers.debug_logger import get_logger
logger = get_logger(__name__)
# Prevents very large files from flooding the context window with tokens.
MAX_FILE_CHARS = 4000
@ -27,6 +30,7 @@ class SystemPrompter:
Returns:
A ready-to-use system prompt string.
"""
logger.info("Generating system prompt.")
base = (
"You are an expert code assistant integrated into a lightweight code editor. "
"Help the user with code suggestions, debugging, explanations, and improvements. "
@ -36,6 +40,7 @@ class SystemPrompter:
prompt = base
if file_context:
logger.info("Appending file context.")
name = file_context.get("name", "unknown")
content = file_context.get("content", "")

5
frontend/app.py
View File

@ -17,6 +17,9 @@ from pathlib import Path
# where streamlit is launched from.
sys.path.insert(0, str(Path(__file__).parent.parent))
from backend.managers.debug_logger import get_logger
logger = get_logger(__name__)
from frontend.sidebar import render_sidebar
from frontend.editor import render_editor
from frontend.chat import render_chat
@ -53,8 +56,10 @@ def main():
# Switch between the two main views based on the sidebar radio button
if st.session_state.get("radio_interface_options") == "Code Editor":
logger.info("Editor mode")
render_editor()
elif st.session_state.get("radio_interface_options") == "Chat with AI Assistant":
logger.info("Chat/Agent mode")
render_chat()

89
frontend/chat.py
View File

@ -6,9 +6,13 @@ from pathlib import Path
import streamlit as st
from backend.managers.chat_manager import ChatManager
from backend.managers.system_prompter import SystemPrompter
from backend.managers.search_manager import SearchManager
from backend.agent.coding_agent import CodingAgent
from backend.managers.debug_logger import get_logger
logger = get_logger(__name__)
import asyncio
import json
# ── Agent Mode helpers ────────────────────────────────────────────────────────
@ -41,7 +45,7 @@ def _start_agent(task: str):
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
logger.info("Starting coding agent.")
agent = CodingAgent()
agent.start_task(task)
action = _run_async(agent.propose_next_action())
@ -57,6 +61,7 @@ def _approve_action():
pending = st.session_state.agent_pending_action
result = _run_async(agent.approve())
logger.info("Approve action and propose next step.")
# Append a record to the log so the user can review every completed step.
st.session_state.agent_log.append({
@ -81,6 +86,7 @@ def _reject_action(feedback: str):
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().
"""
logger.info("Rejecting proposed action.")
agent = st.session_state.coding_agent
agent.reject(feedback or "Please try a different approach.")
next_action = _run_async(agent.propose_next_action())
@ -90,6 +96,7 @@ def _reject_action(feedback: str):
def _followup_agent(question: str):
"""Continue a finished task by injecting a follow-up question and resuming the loop."""
logger.info("Asking follow up question")
agent = st.session_state.coding_agent
agent.follow_up(question)
action = _run_async(agent.propose_next_action())
@ -99,6 +106,7 @@ def _followup_agent(question: str):
def _reset_agent():
"""Clear all agent state and return to the idle (task input) screen."""
logger.info("Resetting Agent")
st.session_state.coding_agent = None
st.session_state.agent_status = "idle"
st.session_state.agent_log = []
@ -107,6 +115,62 @@ def _reset_agent():
# ── Agent Mode UI ─────────────────────────────────────────────────────────────
def _render_arguments(args: dict):
if not args:
return
with st.expander("📦 Arguments", expanded=False):
if args.get("path"):
st.markdown("##### 📁 Path")
st.code(args["path"])
if args.get("dir_path"):
st.markdown("##### 🌳 Directory")
st.code(args["dir_path"])
if args.get("query"):
st.markdown("##### 🔎 Query")
st.code(args["query"])
if args.get("url"):
st.markdown("##### 🌐 URL")
st.code(args["url"])
if args.get("content"):
st.markdown("##### 📝 Content")
st.code(args["content"])
if args.get("code"):
st.markdown("##### 🐍 Python Code")
st.code(args["code"], language="python")
if args.get("max_results") is not None:
st.markdown("##### 📊 Max Results")
st.code(str(args["max_results"]))
known_keys = {
"path",
"dir_path",
"query",
"content",
"url",
"code",
"max_results",
}
extra_args = {
k: v for k, v in args.items()
if k not in known_keys
}
if extra_args:
st.markdown("##### ⚙️ Other")
st.code(
json.dumps(extra_args, indent=2),
language="json"
)
def render_agent_mode():
"""Render the step-by-step agent UI.
@ -115,6 +179,7 @@ def render_agent_mode():
- "waiting_approval" → show proposed action, Approve / Reject / Abort
- "done" → success message, follow-up input, New Task button
"""
logger.info("Agent mode.")
# The toggle must always render so Streamlit keeps agent_mode=True in session_state.
st.toggle("Agent Mode", key="agent_mode")
@ -129,8 +194,8 @@ def render_agent_mode():
with st.chat_message("assistant"):
st.markdown(f"**Step {i + 1} — `{step['tool']}`**")
st.caption(f"Thought: {step['thought']}")
if step.get("arguments"):
st.json(step["arguments"])
if step.get("arguments"):
_render_arguments(step["arguments"])
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"):
@ -149,9 +214,6 @@ def render_agent_mode():
placeholder="e.g. Write a function that sorts a list and saves it to sorted.py",
)
if st.button("Start Agent", type="primary", use_container_width=True):
# REVIEW: commented-out code — remove if not needed.
#loop = asyncio.new_event_loop()
#asyncio.set_event_loop(loop)
if task.strip():
with st.spinner("Agent is thinking..."):
_start_agent(task.strip())
@ -169,15 +231,7 @@ def render_agent_mode():
args = pending.get("arguments", {})
if args:
# Show file content separately as a code block for readability;
# other arguments are displayed as JSON.
if "content" in args:
display_args = {k: v for k, v in args.items() if k != "content"}
if display_args:
st.json(display_args)
st.code(args["content"], language="python")
else:
st.json(args)
_render_arguments(args)
feedback = st.text_input(
"Rejection feedback (optional):",
@ -322,8 +376,7 @@ def render_normal_chat():
full context throughout the session. If search results are active when the
user sends a message, they are prepended to that message as a context block.
"""
_render_search_panel()
logger.info("Chat mode")
# Replay the conversation history as chat bubbles (skip system messages).
for message in st.session_state.chat_history:
if message["role"] == "system":

20
frontend/editor.py
View File

@ -8,9 +8,8 @@ from pathlib import Path
from backend.managers.file_manager import FileManager
from backend.managers.execution_engine import ExecutionEngine
# REVIEW: DebugLogger is imported and used to log execution steps, but its output is never
# surfaced in the UI — DebugLogger writes to an in-memory buffer that nothing reads or renders.
from backend.managers.debug_logger import DebugLogger
from backend.managers.debug_logger import get_logger
logger = get_logger(__name__)
# Maps file extensions to Ace editor language modes for syntax highlighting.
LANG_MAP = {
@ -55,8 +54,10 @@ def _rename_dialog(file_path: str):
if st.session_state.active_file == file_path:
st.session_state.active_file = new_file_path
st.rerun()
logger.info("Rename file %s to %s successfull", file_path, new_name )
else:
st.error("Rename failed. Check that the file still exists.")
logger.warning("Rename failed.")
st.error("Rename failed. Check that the file %s still exists.", file_path)
with col2:
if st.button("Cancel", use_container_width=True):
st.rerun()
@ -89,8 +90,10 @@ def _delete_dialog(abs_file_path: str):
if st.session_state.open_files else None
)
st.rerun()
logger.info("Deleting file %s successfull.", abs_file_path)
else:
st.error("Delete failed. Check that the file still exists.")
logger.warning("Deleting file %s failed.", abs_file_path)
with col2:
if st.button("Cancel", use_container_width=True):
st.rerun()
@ -112,9 +115,8 @@ def run_active_file():
return
execution_engine = ExecutionEngine()
debug_logger = DebugLogger()
debug_logger.clear()
debug_logger.log(f"Executing code from {active_file}...")
logger.info("Executing code from %s...", active_file)
# ast check — only for Python files
if Path(active_file).suffix == ".py":
@ -131,9 +133,9 @@ def run_active_file():
output = execution_engine.run_code(Path(active_file))
if output["rc"] == 0:
debug_logger.log("Execution completed successfully.")
logger.info("Execution completed successfully.")
else:
debug_logger.log_error(f"Execution failed with exit code {output['rc']}.")
logger.error("Execution failed with exit code %s.", output['rc'])
result = {
"stdout": output["stdout"],

5
requirements.txt
View File

@ -25,5 +25,6 @@ python-dotenv>=1.0.0
#For code editor functionality
streamlit-ace>=0.1.0
#Whitelisted Imports from Agent-Sandbox
pygame
#MCP-Code execution tools
pyflakes>=0.1.0
pygame>=0.1.0

2
tests/test_debug_logger.py
View File

@ -6,7 +6,7 @@ import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from backend.managers.debug_logger import DebugLogger
from backend.managers._debug_logger import DebugLogger
# ── Fixtures ──────────────────────────────────────────────────────────────────

428
tests/test_execution_engine.py
View File

@ -0,0 +1,428 @@
import pytest
import subprocess
from pathlib import Path
from unittest.mock import Mock, patch
from backend.managers.execution_engine import ExecutionEngine
# =========================================================
# FIXTURE
# =========================================================
@pytest.fixture()
def engine():
return ExecutionEngine()
# =========================================================
# BASIC TESTS (1–10)
# =========================================================
# ---------------------------------------------------------
# 1. Python-Datei wird korrekt ausgeführt
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_python_file_success(mock_run, engine, tmp_path):
file = tmp_path / "test.py"
file.write_text("print('hello')")
mock_run.return_value = Mock(
stdout="hello\n",
stderr="",
returncode=0
)
result = engine.run_code(file)
assert result["rc"] == 0
assert "hello" in result["stdout"]
# ---------------------------------------------------------
# 2. Python-Datei mit Fehler
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_python_file_error(mock_run, engine, tmp_path):
file = tmp_path / "broken.py"
file.write_text("1/0")
mock_run.return_value = Mock(
stdout="",
stderr="ZeroDivisionError",
returncode=1
)
result = engine.run_code(file)
assert result["rc"] == 1
assert "ZeroDivisionError" in result["stderr"]
# ---------------------------------------------------------
# 3. LaTeX-Datei wird kompiliert
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_tex_file_success(mock_run, engine, tmp_path):
file = tmp_path / "doc.tex"
file.write_text("\\documentclass{article}")
mock_run.return_value = Mock(
stdout="PDF created",
stderr="",
returncode=0
)
result = engine.run_code(file)
assert result["rc"] == 0
assert "PDF created" in result["stdout"]
# ---------------------------------------------------------
# 4. Unsupported File Type
# ---------------------------------------------------------
def test_run_unsupported_file(engine, tmp_path):
file = tmp_path / "test.js"
file.write_text("console.log('x')")
result = engine.run_code(file)
assert result["rc"] == 1
assert "Unsupported file type" in result["stderr"]
# ---------------------------------------------------------
# 5. Timeout wird behandelt
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_timeout(mock_run, engine, tmp_path):
file = tmp_path / "slow.py"
file.write_text("while True: pass")
mock_run.side_effect = subprocess.TimeoutExpired(
cmd=["py"],
timeout=30
)
result = engine.run_code(file)
assert result["rc"] == -1
assert "Timed out" in result["stderr"]
# ---------------------------------------------------------
# 6. Fehlender Interpreter
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_missing_interpreter(mock_run, engine, tmp_path):
file = tmp_path / "test.py"
file.write_text("print(1)")
mock_run.side_effect = FileNotFoundError("py not found")
result = engine.run_code(file)
assert result["rc"] == -1
assert "py not found" in result["stderr"]
# ---------------------------------------------------------
# 7. Allgemeine Exception
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_general_exception(mock_run, engine, tmp_path):
file = tmp_path / "test.py"
file.write_text("print(1)")
mock_run.side_effect = RuntimeError("unexpected")
result = engine.run_code(file)
assert result["rc"] == -1
assert "unexpected" in result["stderr"]
# ---------------------------------------------------------
# 8. subprocess.run wird mit cwd ausgeführt
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_uses_correct_cwd(mock_run, engine, tmp_path):
folder = tmp_path / "project"
folder.mkdir()
file = folder / "main.py"
file.write_text("print(1)")
mock_run.return_value = Mock(
stdout="",
stderr="",
returncode=0
)
engine.run_code(file)
_, kwargs = mock_run.call_args
assert kwargs["cwd"] == folder.resolve()
# ---------------------------------------------------------
# 9. subprocess.run nutzt capture_output
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_capture_output_enabled(mock_run, engine, tmp_path):
file = tmp_path / "test.py"
file.write_text("print(1)")
mock_run.return_value = Mock(
stdout="",
stderr="",
returncode=0
)
engine.run_code(file)
_, kwargs = mock_run.call_args
assert kwargs["capture_output"] is True
# ---------------------------------------------------------
# 10. subprocess.run nutzt text=True
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_text_mode_enabled(mock_run, engine, tmp_path):
file = tmp_path / "test.py"
file.write_text("print(1)")
mock_run.return_value = Mock(
stdout="",
stderr="",
returncode=0
)
engine.run_code(file)
_, kwargs = mock_run.call_args
assert kwargs["text"] is True
# =========================================================
# EDGE CASE TESTS (11–20)
# =========================================================
# ---------------------------------------------------------
# 11. Unicode Output
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_unicode_output(mock_run, engine, tmp_path):
file = tmp_path / "unicode.py"
file.write_text("print('🔥 Grüezi 世界')")
mock_run.return_value = Mock(
stdout="🔥 Grüezi 世界\n",
stderr="",
returncode=0
)
result = engine.run_code(file)
assert "🔥 Grüezi 世界" in result["stdout"]
# ---------------------------------------------------------
# 12. Leerer stdout/stderr
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_empty_output(mock_run, engine, tmp_path):
file = tmp_path / "empty.py"
file.write_text("x = 1")
mock_run.return_value = Mock(
stdout="",
stderr="",
returncode=0
)
result = engine.run_code(file)
assert result["stdout"] == ""
assert result["stderr"] == ""
# ---------------------------------------------------------
# 13. Sehr langer stdout
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_large_output(mock_run, engine, tmp_path):
file = tmp_path / "large.py"
file.write_text("print('A')")
mock_run.return_value = Mock(
stdout="A" * 100000,
stderr="",
returncode=0
)
result = engine.run_code(file)
assert len(result["stdout"]) == 100000
# ---------------------------------------------------------
# 14. Dateiname mit Leerzeichen
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_filename_with_spaces(mock_run, engine, tmp_path):
file = tmp_path / "my script.py"
file.write_text("print(1)")
mock_run.return_value = Mock(
stdout="ok",
stderr="",
returncode=0
)
engine.run_code(file)
args, _ = mock_run.call_args
assert "my script.py" in args[0]
# ---------------------------------------------------------
# 15. Dateiname mit Unicode
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_unicode_filename(mock_run, engine, tmp_path):
file = tmp_path / "🔥_test.py"
file.write_text("print(1)")
mock_run.return_value = Mock(
stdout="ok",
stderr="",
returncode=0
)
result = engine.run_code(file)
assert result["rc"] == 0
# ---------------------------------------------------------
# 16. .tex nutzt pdflatex
# ---------------------------------------------------------
@patch("subprocess.run")
def test_tex_uses_pdflatex(mock_run, engine, tmp_path):
file = tmp_path / "doc.tex"
file.write_text("x")
mock_run.return_value = Mock(
stdout="",
stderr="",
returncode=0
)
engine.run_code(file)
args, _ = mock_run.call_args
assert args[0][0] == "pdflatex"
# ---------------------------------------------------------
# 17. .py nutzt py Interpreter
# ---------------------------------------------------------
@patch("subprocess.run")
def test_python_uses_py_interpreter(mock_run, engine, tmp_path):
file = tmp_path / "main.py"
file.write_text("print(1)")
mock_run.return_value = Mock(
stdout="",
stderr="",
returncode=0
)
engine.run_code(file)
args, _ = mock_run.call_args
assert args[0][0] == "py"
# ---------------------------------------------------------
# 18. Relative Pfade funktionieren
# ---------------------------------------------------------
@patch("subprocess.run")
def test_relative_paths(mock_run, engine, tmp_path):
sub = tmp_path / "src"
sub.mkdir()
file = sub / "main.py"
file.write_text("print(1)")
mock_run.return_value = Mock(
stdout="ok",
stderr="",
returncode=0
)
result = engine.run_code(file)
assert result["rc"] == 0
# ---------------------------------------------------------
# 19. Großgeschriebenes Suffix blockiert
# ---------------------------------------------------------
def test_uppercase_suffix_not_supported(engine, tmp_path):
file = tmp_path / "SCRIPT.PY"
file.write_text("print(1)")
result = engine.run_code(file)
assert result["rc"] == 1
# ---------------------------------------------------------
# 20. Leere Datei ausführen
# ---------------------------------------------------------
@patch("subprocess.run")
def test_run_empty_file(mock_run, engine, tmp_path):
file = tmp_path / "empty.py"
file.write_text("")
mock_run.return_value = Mock(
stdout="",
stderr="",
returncode=0
)
result = engine.run_code(file)
assert result["rc"] == 0

269
tests/test_mcp_server_code_execution.py
View File

@ -0,0 +1,269 @@
import pytest
from unittest.mock import Mock, patch
import subprocess
from backend.agent.servers import mcp_server_code_execution as server
# =========================================================
# BASIC TESTS (1–10)
# =========================================================
# ---------------------------------------------------------
# 1. Erlaubter Code besteht Safety Check
# ---------------------------------------------------------
def test_check_code_safety_valid():
code = "print('hello')"
result = server.check_code_safety(code)
assert result is None
# ---------------------------------------------------------
# 2. Blockierter Import wird erkannt
# ---------------------------------------------------------
def test_check_code_safety_blocked_import():
code = "import os"
result = server.check_code_safety(code)
assert "Blocked import" in result
# ---------------------------------------------------------
# 3. Blockierter Builtin wird erkannt
# ---------------------------------------------------------
def test_check_code_safety_blocked_builtin():
code = "eval('2+2')"
result = server.check_code_safety(code)
assert "Blocked builtin" in result
# ---------------------------------------------------------
# 4. analyse_structure erkennt Funktionen
# ---------------------------------------------------------
def test_analyse_structure_function():
code = """
def hello(name):
return name
"""
result = server.analyse_structure(code)
assert "def hello(name)" in result
# ---------------------------------------------------------
# 5. analyse_structure erkennt Klassen
# ---------------------------------------------------------
def test_analyse_structure_class():
code = """
class User:
def login(self):
pass
"""
result = server.analyse_structure(code)
assert "class User" in result
assert "method: login" in result
# ---------------------------------------------------------
# 6. lint_code erkennt Undefined Variable
# ---------------------------------------------------------
def test_lint_code_undefined_variable():
code = "print(x)"
result = server.lint_code(code)
assert "undefined name 'x'" in result.lower()
# ---------------------------------------------------------
# 7. lint_code erkennt sauberen Code
# ---------------------------------------------------------
def test_lint_code_clean():
code = """
x = 1
print(x)
"""
result = server.lint_code(code)
assert "No issues found" in result
# ---------------------------------------------------------
# 8. python_code_validation validiert sicheren Code
# ---------------------------------------------------------
def test_python_code_validation_safe():
code = "print('safe')"
result = server.python_code_validation(code)
assert "can be executed" in result
# ---------------------------------------------------------
# 9. run_python_sandboxed führt Code aus
# ---------------------------------------------------------
def test_run_python_sandboxed_success():
code = "print('hello world')"
result = server.run_python_sandboxed(code)
assert "hello world" in result
# ---------------------------------------------------------
# 10. run_python_sandboxed ohne Output
# ---------------------------------------------------------
def test_run_python_sandboxed_no_output():
code = "x = 5"
result = server.run_python_sandboxed(code)
assert "no output" in result.lower()
# =========================================================
# EDGE CASE TESTS (11–20)
# =========================================================
# ---------------------------------------------------------
# 11. Syntaxfehler erkennen
# ---------------------------------------------------------
def test_check_code_safety_syntax_error():
code = "def broken("
result = server.check_code_safety(code)
assert "SyntaxError" in result
# ---------------------------------------------------------
# 12. ImportFrom blockieren
# ---------------------------------------------------------
def test_check_code_safety_import_from():
code = "from os import path"
result = server.check_code_safety(code)
assert "Blocked import" in result
# ---------------------------------------------------------
# 13. Gefährliche Path-Sequenzen erkennen
# ---------------------------------------------------------
def test_check_code_safety_path_traversal():
code = "print('../etc/passwd')"
result = server.check_code_safety(code)
assert "Suspect path sequence" in result
# ---------------------------------------------------------
# 14. __import__ erkennen
# ---------------------------------------------------------
def test_check_code_safety_import_escape():
code = "__import__('os')"
result = server.check_code_safety(code)
assert "Blocked" in result
# ---------------------------------------------------------
# 15. subprocess Escape erkennen
# ---------------------------------------------------------
def test_check_code_safety_subprocess_escape():
code = "subprocess.run(['ls'])"
result = server.check_code_safety(code)
assert "Suspect path sequence" in result
# ---------------------------------------------------------
# 16. Endlosschleife Timeout
# ---------------------------------------------------------
def test_run_python_sandboxed_timeout():
code = """
while True:
pass
"""
result = server.run_python_sandboxed(code)
assert "time limit" in result.lower()
# ---------------------------------------------------------
# 17. Sehr großer Output wird gekürzt
# ---------------------------------------------------------
def test_run_python_sandboxed_large_output():
code = "print('A' * 10000)"
result = server.run_python_sandboxed(code)
assert "truncated" in result.lower()
# ---------------------------------------------------------
# 18. Unicode Output funktioniert
# ---------------------------------------------------------
def test_run_python_sandboxed_unicode():
code = "print('🔥 Grüezi 世界')"
result = server.run_python_sandboxed(code)
assert "🔥 Grüezi 世界" in result
# ---------------------------------------------------------
# 19. analyse_structure bei leerem Code
# ---------------------------------------------------------
def test_analyse_structure_empty():
code = ""
result = server.analyse_structure(code)
assert "No top-level imports" in result
# ---------------------------------------------------------
# 20. Sandbox behandelt Runtime Errors
# ---------------------------------------------------------
def test_run_python_sandboxed_runtime_error():
code = "1 / 0"
result = server.run_python_sandboxed(code)
assert "ZeroDivisionError" in result