"""Whether agent chats exist here at all, and what they may spend. An administrator's half of the feature. The other half -- which machines, whose credentials -- belongs to whoever owns them and lives at `/agents`. Nothing here is about isolation, because there is none to configure: commands run on a host somebody chose, and its containment is that host's. The settings are budgets, and the two lists that decide what a mode asks about. """ from __future__ import annotations import logging from fastapi import APIRouter, Form, Request, Response, status from fastapi.responses import RedirectResponse from sqlalchemy import func, select from lembas.api.deps import AdminUser, Db from lembas.db.models import SshProfile from lembas.services import settings_store from lembas.services.agent import policy from lembas.services.agent import ssh as ssh_service from lembas.services.agent import terminal as terminal_service from lembas.web.templating import render log = logging.getLogger(__name__) router = APIRouter(prefix="/admin/agents", tags=["admin-agents"]) def _lines(text: str) -> list[str]: """One pattern per line, blanks dropped.""" return [line.strip() for line in (text or "").splitlines() if line.strip()] @router.get("") async def agents_page(request: Request, db: Db, user: AdminUser, saved: bool = False): values = settings_store.agents(db) return render( request, "admin/agents.html", { "values": values, "allow_text": "\n".join(values.get("allow_default") or []), "deny_text": "\n".join(values.get("deny_default") or []), "problem": ssh_service.available(), "profile_count": db.scalar(select(func.count()).select_from(SshProfile)) or 0, "terminal_count": terminal_service.count(), "modes": [(m, policy.MODE_LABELS[m], policy.MODE_HINTS[m]) for m in policy.MODES], "saved": saved, }, ) @router.post("") async def save_agents( db: Db, user: AdminUser, enabled: bool = Form(False), default_timeout: int = Form(60), max_timeout: int = Form(600), max_output_bytes: int = Form(64 * 1024), max_steps: int = Form(200), max_wall_seconds: int = Form(900), max_total_output_bytes: int = Form(1024 * 1024), max_completion_tokens: int = Form(200_000), approval_timeout: int = Form(900), allow_default: str = Form(""), deny_default: str = Form(""), ask_free_text: bool = Form(False), terminal_enabled: bool = Form(False), terminal_idle_timeout: int = Form(1800), terminal_max_sessions: int = Form(20), terminal_max_per_user: int = Form(3), terminal_integration: bool = Form(False), index_enabled: bool = Form(False), index_chars: int = Form(2000), instructions_enabled: bool = Form(False), instructions_chars: int = Form(4000), ) -> Response: settings_store.update( db, { "enabled": enabled, # Clamped here as well as on read. A number with no bound is a way # to break the instance from a form, which is the same reasoning # the search settings carry. "default_timeout": min(max(default_timeout, 1), 3600), "max_timeout": min(max(max_timeout, 1), 3600), "max_output_bytes": min(max(max_output_bytes, 1024), 1024 * 1024), "max_steps": min(max(max_steps, 1), 1000), "max_wall_seconds": min(max(max_wall_seconds, 30), 7200), "max_total_output_bytes": min(max(max_total_output_bytes, 4096), 8 * 1024 * 1024), # Floor of 0, not 1: zero is how "no ceiling" is said. "max_completion_tokens": min(max(max_completion_tokens, 0), 5_000_000), "approval_timeout": min(max(approval_timeout, 60), 3600), "allow_default": _lines(allow_default), "deny_default": _lines(deny_default), "ask_free_text": ask_free_text, "terminal_enabled": terminal_enabled, "terminal_idle_timeout": min(max(terminal_idle_timeout, 60), 86400), "terminal_max_sessions": min(max(terminal_max_sessions, 1), 500), "terminal_max_per_user": min(max(terminal_max_per_user, 1), 50), "terminal_integration": terminal_integration, "index_enabled": index_enabled, # Zero is kept rather than clamped up: it means "list the # directory for the file picker but put none of it in the # prompt", which nothing else can say. "index_chars": min(max(index_chars, 0), 20_000), "instructions_enabled": instructions_enabled, "instructions_chars": min(max(instructions_chars, 0), 20_000), }, key=settings_store.AGENTS, ) log.info("agent execution %s by %s", "enabled" if enabled else "disabled", user.email) return RedirectResponse("/admin/agents?saved=1", status_code=status.HTTP_303_SEE_OTHER)