from aiohttp import web from pathlib import Path import asyncio import json from ..auto_approver import AutoApprover, AutoApproverConfig from ..chat_io_buffer import ChatIOBuffer from ..entry.entry_factory import EntryFactory from ..iteration_parser import IterationParser from ..web_agent import WebAgent from ..working_memory import WorkingMemory class Api: def __init__( self, work_dir: Path, app: web.Application, agent: WebAgent, io_buffer: ChatIOBuffer, working_memory: WorkingMemory, auto_approver: AutoApprover ): self._work_dir = work_dir self._app = app self._agent = agent self._working_memory = working_memory self._io_buffer = io_buffer self._auto_approver = auto_approver self._init_routes() def _init_routes(self): """Initialize REST API and WebSocket routes.""" self._app.router.add_get("/api/llms", self._get_llms) self._app.router.add_get("/api/llms/active", self._get_active_llm) self._app.router.add_post("/api/llms/active", self._set_active_llm) self._app.router.add_post("/api/inference", self._run_inference) self._app.router.add_post("/api/inference/stop", self._stop_inference) self._app.router.add_get("/api/response", self._get_response) self._app.router.add_post("/api/response", self._set_response) self._app.router.add_post("/api/response/approve", self._approve_response) self._app.router.add_post("/api/chat/message", self._add_chat_message) self._app.router.add_delete("/api/chat/message/{id}", self._delete_chat_message) self._app.router.add_delete("/api/chat", self._clear_chat) self._app.router.add_get("/api/auto_approver", self._get_auto_approver_config) self._app.router.add_post("/api/auto_approver", self._set_auto_approver_config) self._app.router.add_get("/api/memory", self._get_memory) self._app.router.add_post("/api/memory/entry", self._create_entry) self._app.router.add_put("/api/memory/entry/{id}", self._save_entry) self._app.router.add_delete("/api/memory/entry/{id}", self._delete_entry) self._app.router.add_post("/api/memory/entry/{id}/reset", self._reset_entry) self._app.router.add_post("/api/memory/entry/{id}/update", self._update_entry) self._app.router.add_post("/api/memory/load_iteration", self._load_iteration) self._app.router.add_get("/api/context_info", self._get_context_info) async def _get_llms(self, request: web.Request) -> web.Response: return web.Response( text=json.dumps( [{"name": name} for name in self._agent.llms] ), content_type="application/json" ) async def _get_active_llm(self, request: web.Request) -> web.Response: return web.Response( text=json.dumps( {"active_llm": self._agent.active_llm} ), content_type="application/json" ) async def _set_active_llm(self, request: web.Request) -> web.Response: try: data = await request.json() self._agent.active_llm = data["active_llm"] return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _run_inference(self, request: web.Request) -> web.Response: try: await asyncio.get_event_loop().run_in_executor(None, self._agent.run_inference) return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _stop_inference(self, request: web.Request) -> web.Response: """Stop inference on specified LLM.""" try: self._agent.stop_inference() return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _get_response(self, request: web.Request) -> web.Response: """Get current buffer content""" return web.Response( text=json.dumps({ "response": self._agent.response_buffer.get_text(), }), content_type="application/json" ) async def _set_response(self, request: web.Request) -> web.Response: """Edit the shared response buffer""" try: data = await request.json() text = data.get("response") self._agent._response_buffer.set_text(text) return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _approve_response(self, request: web.Request) -> web.Response: """Approve current buffer content""" try: self._agent.approve_response() return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _add_chat_message(self, request: web.Request) -> web.Response: """Add a chat message""" try: data = await request.json() message = data.get("message") self._io_buffer.add_user_message(message) return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _delete_chat_message(self, request: web.Request) -> web.Response: """Delete a chat message""" try: message_id = request.match_info["id"] self._io_buffer.delete_message(message_id) return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _clear_chat(self, request: web.Request) -> web.Response: """Clear chat messages""" try: self._io_buffer.clear() return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _get_auto_approver_config(self, request: web.Request) -> web.Response: """Get current auto approver configuration.""" return web.Response( text=json.dumps(self._auto_approver.config), content_type="application/json" ) async def _set_auto_approver_config(self, request: web.Request) -> web.Response: """Update auto approver configuration.""" try: data = await request.json() self._auto_approver.config = AutoApproverConfig(**data) return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _set_context_enabled(self, request: web.Request) -> web.Response: """Set context auto-approval enabled state.""" try: data = await request.json() enabled = data.get("enabled") self._auto_approver.context_enabled = enabled or False return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _set_response_enabled(self, request: web.Request) -> web.Response: """Set response auto-approval enabled state.""" try: data = await request.json() enabled = data.get("enabled") self._auto_approver.response_enabled = enabled or False return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _set_context_timeout(self, request: web.Request) -> web.Response: """Set context auto-approval timeout.""" try: data = await request.json() timeout = data.get("timeout") if timeout is None: return web.Response(status=400, text="Missing timeout parameter") self._auto_approver.context_timeout = float(timeout) return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _set_response_timeout(self, request: web.Request) -> web.Response: """Set response auto-approval timeout.""" try: data = await request.json() timeout = data.get("timeout") if timeout is None: return web.Response(status=400, text="Missing timeout parameter") self._auto_approver.response_timeout = float(timeout) return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _get_memory(self, request: web.Request) -> web.Response: """Get complete working memory state.""" entries = self._working_memory.get_entries() return web.Response( text=json.dumps({ "type": "memory_state", "entries": [e.serialize() for e in entries] }), content_type="application/json" ) async def _create_entry(self, request: web.Request) -> web.Response: """Create a new entry in working memory.""" try: data = await request.json() entry = EntryFactory.create_entry(data, self._work_dir, self._io_buffer) self._working_memory.add_entry(entry) return web.Response( text=json.dumps({"id": entry.id}), content_type="application/json" ) except Exception as e: return web.Response(status=400, text=str(e)) async def _save_entry(self, request: web.Request) -> web.Response: """Update properties of an existing entry.""" try: entry_id = request.match_info["id"] data = await request.json() entry = self._working_memory.get_entry(entry_id) if not entry: return web.Response(status=404, text="Entry not found") EntryFactory.update_entry(entry, data) entry.notify_change() return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _delete_entry(self, request: web.Request) -> web.Response: """Delete an entry from working memory.""" entry_id = request.match_info["id"] self._working_memory.remove_entry(entry_id) return web.Response(status=200) async def _reset_entry(self, request: web.Request) -> web.Response: """Reset an entry's state.""" entry_id = request.match_info["id"] entry = self._working_memory.get_entry(entry_id) if not entry: return web.Response(status=404, text="Entry not found") entry.reset() return web.Response(status=200) async def _update_entry(self, request: web.Request) -> web.Response: """Update an entry's state.""" try: entry_id = request.match_info["id"] entry = self._working_memory.get_entry(entry_id) if not entry: return web.Response(status=404, text="Entry not found") entry.update() entry.notify_change() return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _load_iteration(self, request: web.Request) -> web.Response: """Load entries from iteration XML content into working memory""" try: data = await request.json() content = data.get("content") if not content: return web.Response(status=400, text="Missing content in request body") (_context, response, entries) = IterationParser.parse_iteration(content, self._work_dir, self._io_buffer) for entry in entries: self._working_memory.add_entry(entry) self._agent.response_buffer.set_text(response) return web.Response(status=200) except Exception as e: return web.Response(status=400, text=str(e)) async def _get_context_info(self, request: web.Request) -> web.Response: """Get current context info.""" return web.Response( text=json.dumps(self._agent.context_info), content_type="application/json" )