import ipaddress import re import uuid from typing import Annotated, Literal from fastapi import APIRouter, Cookie, Header, Request from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from sqlalchemy import delete, func, select from .auth import AuthDep, SessionDep, _audit, _require_csrf from .authorization import authorize, permissions_for from .catalog import PERMISSIONS from .compute_provider import ComputeProviderDep from .computer_catalog import DEFAULT_COMPUTER_TEMPLATE_KEY from .computer_provisioning import ComputerProvisioningService from .computer_tools import AGENT_COMPUTER_TOOLS from .config import get_settings from .errors import AppError from .models import ( Agent, AgentDelegation, AgentPermission, AgentSchedule, AgentTool, AgentVersion, AIModelProfile, Computer, ComputerNetworkPolicy, ComputerOperation, ComputerProfile, ComputerSession, ComputerTemplate, DeviceSession, Execution, WorkspaceComputerLimit, ) from .private_computers import ( build_dedicated_computer, build_private_computer, computer_template_for_key, dedicated_computer_for_agent, private_computer_for_agent, ) from .security import canonical_payload_hash from .tool_runtime import CORE_TOOLS router = APIRouter(prefix="/api/v1/agents", tags=["agents"]) class ScheduleInput(BaseModel): timezone: str = Field(min_length=1, max_length=80) working_days: list[Literal["mon", "tue", "wed", "thu", "fri", "sat", "sun"]] = Field( min_length=1, max_length=7 ) start_minute: int = Field(ge=0, le=1439) end_minute: int = Field(ge=1, le=1440) enabled: bool = True @model_validator(mode="after") def validate_schedule(self): if self.start_minute >= self.end_minute: raise ValueError("Schedule start must be before end") if len(self.working_days) != len(set(self.working_days)): raise ValueError("Working days must be unique") return self class ComputerNetworkInput(BaseModel): model_config = ConfigDict(extra="forbid") internet_access: bool = True lan_access: bool = False platform_api_access: bool = True other_agent_networks: bool = False allowed_domains: list[str] = Field(default_factory=list, max_length=200) blocked_domains: list[str] = Field(default_factory=list, max_length=200) @model_validator(mode="after") def validate_domains(self): allowed = [item.strip().lower().rstrip(".") for item in self.allowed_domains] blocked = [item.strip().lower().rstrip(".") for item in self.blocked_domains] for item in allowed + blocked: if ( not item or len(item) > 253 or "://" in item or "/" in item or "@" in item ): raise ValueError("Network policy entries must be DNS hostnames") try: address = ipaddress.ip_address(item) except ValueError: address = None if address is not None and not address.is_global: raise ValueError("Private and special-purpose addresses are not allowed") if len(set(allowed)) != len(allowed) or len(set(blocked)) != len(blocked): raise ValueError("Network policy domains must be unique") if set(allowed) & set(blocked): raise ValueError("A domain cannot be both allowed and blocked") self.allowed_domains = allowed self.blocked_domains = blocked return self class AgentComputerInput(BaseModel): create_dedicated: bool = True os_family: Literal["linux", "windows"] = "linux" template_key: str = Field(default=DEFAULT_COMPUTER_TEMPLATE_KEY, min_length=2, max_length=80) cpu_cores: int = Field(default=2, ge=1, le=64) memory_mb: int = Field(default=4096, ge=512, le=524288) disk_gb: int = Field(default=40, ge=10, le=16384) browser: Literal["chromium", "chrome", "edge", "firefox"] = "chromium" persistent_disk: bool = True system_privilege: Literal[ "restricted", "standard_user", "administrator", "full_system" ] = "standard_user" computer_autonomy: int = Field(default=1, ge=0, le=5) start_policy: Literal[ "always_running", "start_when_needed", "scheduled", "manual" ] = "start_when_needed" network_policy: ComputerNetworkInput = Field(default_factory=ComputerNetworkInput) @field_validator("template_key") @classmethod def normalize_template_key(cls, value: str) -> str: value = value.strip().lower() if not re.fullmatch(r"[a-z][a-z0-9-]+", value): raise ValueError("Computer template key is invalid") return value class AgentCreateInput(BaseModel): name: str = Field(min_length=1, max_length=160) description: str = Field(default="", max_length=1000) purpose: str = Field(default="", max_length=4000) system_instructions: str = Field(default="", max_length=20_000) personality: str = Field(default="", max_length=4000) autonomy_level: int = Field(default=1, ge=0, le=5) preferred_model: str | None = Field(default=None, min_length=1, max_length=160) fallback_model: str | None = Field(default=None, min_length=1, max_length=160) manager_agent_id: uuid.UUID | None = None escalation_agent_id: uuid.UUID | None = None budget_limit_minor: int | None = Field(default=None, ge=0) budget_currency: str = Field(default="USD", pattern=r"^[A-Z]{3}$") execution_limit: int | None = Field(default=None, ge=1) tool_names: list[str] = Field(default_factory=list, max_length=100) permission_keys: list[str] = Field(default_factory=list, max_length=200) schedule: ScheduleInput | None = None computer: AgentComputerInput = Field(default_factory=AgentComputerInput) @field_validator("name", "description", "purpose", "system_instructions", "personality") @classmethod def strip_text(cls, value: str) -> str: return value.strip() @model_validator(mode="after") def validate_collections(self): if not self.name: raise ValueError("Agent name is required") if len(self.tool_names) != len(set(self.tool_names)): raise ValueError("Tool names must be unique") if len(self.permission_keys) != len(set(self.permission_keys)): raise ValueError("Permission keys must be unique") if self.preferred_model and self.preferred_model == self.fallback_model: raise ValueError("Fallback model must differ from preferred model") return self class AgentUpdateInput(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=160) description: str | None = Field(default=None, max_length=1000) purpose: str | None = Field(default=None, max_length=4000) system_instructions: str | None = Field(default=None, max_length=20_000) personality: str | None = Field(default=None, max_length=4000) autonomy_level: int | None = Field(default=None, ge=0, le=5) preferred_model: str | None = Field(default=None, min_length=1, max_length=160) fallback_model: str | None = Field(default=None, min_length=1, max_length=160) manager_agent_id: uuid.UUID | None = None escalation_agent_id: uuid.UUID | None = None budget_limit_minor: int | None = Field(default=None, ge=0) budget_currency: str | None = Field(default=None, pattern=r"^[A-Z]{3}$") execution_limit: int | None = Field(default=None, ge=1) tool_names: list[str] | None = Field(default=None, max_length=100) permission_keys: list[str] | None = Field(default=None, max_length=200) schedule: ScheduleInput | None = None @field_validator("name", "description", "purpose", "system_instructions", "personality") @classmethod def strip_optional_text(cls, value: str | None) -> str | None: return value.strip() if value is not None else None @model_validator(mode="after") def validate_collections(self): for field in ("name", "description", "purpose", "autonomy_level", "budget_currency"): if field in self.model_fields_set and getattr(self, field) is None: raise ValueError(f"{field} cannot be null") if self.name is not None and not self.name: raise ValueError("Agent name is required") if self.tool_names is not None and len(self.tool_names) != len(set(self.tool_names)): raise ValueError("Tool names must be unique") if self.permission_keys is not None and len(self.permission_keys) != len( set(self.permission_keys) ): raise ValueError("Permission keys must be unique") return self class AgentStatusInput(BaseModel): status: Literal["active", "disabled", "archived"] class DuplicateAgentInput(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=160) class DelegationInput(BaseModel): target_agent_id: uuid.UUID instruction: str = Field(min_length=1, max_length=20_000) permission_scope: list[str] = Field(default_factory=list, max_length=200) @field_validator("instruction") @classmethod def strip_instruction(cls, value: str) -> str: value = value.strip() if not value: raise ValueError("Delegation instruction is required") return value async def _require_mutation_csrf( session: SessionDep, auth: AuthDep, cookie_token: str | None, header_token: str | None, ) -> None: _require_csrf(await session.get(DeviceSession, auth.session_id), cookie_token, header_token) async def _load_agent( session: SessionDep, workspace_id: uuid.UUID, agent_id: uuid.UUID, *, lock: bool = False ) -> Agent: statement = select(Agent).where( Agent.id == agent_id, Agent.workspace_id == workspace_id, ) if lock: statement = statement.with_for_update() agent = await session.scalar(statement) if not agent: raise AppError("AGENT_NOT_FOUND", "The agent is unavailable.", 404) return agent async def _latest_version( session: SessionDep, workspace_id: uuid.UUID, agent_id: uuid.UUID ) -> AgentVersion: version = await session.scalar( select(AgentVersion) .where( AgentVersion.workspace_id == workspace_id, AgentVersion.agent_id == agent_id, ) .order_by(AgentVersion.version_number.desc()) .limit(1) ) if not version: raise AppError("AGENT_VERSION_MISSING", "The agent configuration is unavailable.", 503) return version async def _components( session: SessionDep, workspace_id: uuid.UUID, agent_id: uuid.UUID ) -> tuple[list[str], list[str], AgentSchedule | None]: tools = list((await session.scalars( select(AgentTool.tool_name).where( AgentTool.workspace_id == workspace_id, AgentTool.agent_id == agent_id ).order_by(AgentTool.tool_name) )).all()) permissions = list((await session.scalars( select(AgentPermission.permission_key).where( AgentPermission.workspace_id == workspace_id, AgentPermission.agent_id == agent_id, ).order_by(AgentPermission.permission_key) )).all()) schedule = await session.scalar(select(AgentSchedule).where( AgentSchedule.workspace_id == workspace_id, AgentSchedule.agent_id == agent_id )) return tools, permissions, schedule def _schedule_data(schedule: AgentSchedule | None): if not schedule: return None return { "timezone": schedule.timezone, "working_days": schedule.working_days, "start_minute": schedule.start_minute, "end_minute": schedule.end_minute, "enabled": schedule.enabled, } async def _agent_data(session: SessionDep, agent: Agent, *, detail: bool = False) -> dict: computer = await session.scalar(select(Computer).where( Computer.workspace_id == agent.workspace_id, Computer.id == agent.computer_id, )) if agent.computer_id else None browser_profile = await session.scalar(select(ComputerProfile).where( ComputerProfile.workspace_id == agent.workspace_id, ComputerProfile.agent_id == agent.id, )) data = { "id": str(agent.id), "name": agent.name, "description": agent.description, "purpose": agent.purpose, "status": agent.status, "autonomy_level": agent.autonomy_level, "computer_required": agent.computer_required, "system_privilege": agent.system_privilege, "computer_autonomy": agent.computer_autonomy, "preferred_model": agent.preferred_model, "fallback_model": agent.fallback_model, "manager_agent_id": str(agent.manager_agent_id) if agent.manager_agent_id else None, "escalation_agent_id": ( str(agent.escalation_agent_id) if agent.escalation_agent_id else None ), "budget_limit_minor": agent.budget_limit_minor, "budget_currency": agent.budget_currency, "execution_limit": agent.execution_limit, "version": agent.version, "computer": ({ "id": str(computer.id), "name": computer.name, "status": computer.status, "os_family": computer.os_family, "os_distribution": computer.os_distribution, "os_version": computer.os_version, "cpu_cores": computer.cpu_cores, "memory_mb": computer.memory_mb, "disk_gb": computer.disk_gb, "browser": computer.browser, "persistent_disk": computer.persistent_disk, "start_policy": computer.start_policy, "browser_profile_id": str(browser_profile.id) if browser_profile else None, "retention_days": browser_profile.retention_days if browser_profile else None, } if computer else None), "created_at": agent.created_at, "updated_at": agent.updated_at, } if detail: version = await _latest_version(session, agent.workspace_id, agent.id) tools, permission_keys, schedule = await _components( session, agent.workspace_id, agent.id ) data.update({ "system_instructions": version.system_instructions, "personality": version.personality, "tool_names": tools, "permission_keys": permission_keys, "schedule": _schedule_data(schedule), }) return data async def _validate_relationships( session: SessionDep, *, workspace_id: uuid.UUID, agent_id: uuid.UUID, manager_agent_id: uuid.UUID | None, escalation_agent_id: uuid.UUID | None, ) -> None: for related_id in {manager_agent_id, escalation_agent_id} - {None}: if related_id == agent_id: raise AppError("AGENT_RELATIONSHIP_CYCLE", "An agent cannot reference itself.", 409) await _load_agent(session, workspace_id, related_id) cursor = manager_agent_id visited = {agent_id} while cursor: if cursor in visited: raise AppError("AGENT_RELATIONSHIP_CYCLE", "The manager hierarchy contains a cycle.", 409) visited.add(cursor) manager = await _load_agent(session, workspace_id, cursor) cursor = manager.manager_agent_id def _validate_capabilities( tool_names: list[str], permission_keys: list[str], actor_permissions: set[str] ) -> None: unknown_permissions = set(permission_keys) - PERMISSIONS.keys() if unknown_permissions: raise AppError("AGENT_PERMISSION_UNKNOWN", "An agent permission is unknown.", 422) if not set(permission_keys) <= actor_permissions: raise AppError( "AGENT_PERMISSION_ESCALATION", "An agent cannot receive permissions you do not have.", 403 ) known_tools = CORE_TOOLS | AGENT_COMPUTER_TOOLS unknown_tools = set(tool_names) - known_tools.keys() if unknown_tools: raise AppError("AGENT_TOOL_UNKNOWN", "An agent tool is not registered.", 422) required = {known_tools[name].permission for name in tool_names} if not required <= set(permission_keys): raise AppError( "AGENT_TOOL_PERMISSION_MISSING", "Every agent tool requires its matching agent permission.", 422, ) async def _validate_model_profiles( session: SessionDep, workspace_id: uuid.UUID, preferred: str | None, fallback: str | None, ) -> None: selected = {item for item in (preferred, fallback) if item} if not selected: return configured = set((await session.scalars(select(AIModelProfile.profile_key).where( AIModelProfile.workspace_id == workspace_id, AIModelProfile.profile_key.in_(selected), AIModelProfile.enabled.is_(True), ))).all()) if configured != selected: raise AppError( "AGENT_MODEL_PROFILE_INVALID", "Agent models must reference enabled workspace model profiles.", 422, ) def _snapshot(agent: Agent, tool_names: list[str], permission_keys: list[str], schedule) -> dict: return { "name": agent.name, "description": agent.description, "purpose": agent.purpose, "status": agent.status, "autonomy_level": agent.autonomy_level, "computer_required": agent.computer_required, "computer_id": str(agent.computer_id) if agent.computer_id else None, "system_privilege": agent.system_privilege, "computer_autonomy": agent.computer_autonomy, "computer_network_policy": agent.computer_network_policy, "computer_file_policy": agent.computer_file_policy, "preferred_model": agent.preferred_model, "fallback_model": agent.fallback_model, "manager_agent_id": str(agent.manager_agent_id) if agent.manager_agent_id else None, "escalation_agent_id": str(agent.escalation_agent_id) if agent.escalation_agent_id else None, "budget_limit_minor": agent.budget_limit_minor, "budget_currency": agent.budget_currency, "execution_limit": agent.execution_limit, "tool_names": sorted(tool_names), "permission_keys": sorted(permission_keys), "schedule": schedule.model_dump(mode="json") if isinstance(schedule, ScheduleInput) else _schedule_data(schedule), } async def _persist_components( session: SessionDep, agent: Agent, *, tool_names: list[str], permission_keys: list[str], schedule: ScheduleInput | None, ) -> None: await session.execute(delete(AgentTool).where( AgentTool.workspace_id == agent.workspace_id, AgentTool.agent_id == agent.id )) await session.execute(delete(AgentPermission).where( AgentPermission.workspace_id == agent.workspace_id, AgentPermission.agent_id == agent.id, )) await session.execute(delete(AgentSchedule).where( AgentSchedule.workspace_id == agent.workspace_id, AgentSchedule.agent_id == agent.id )) session.add_all([ AgentTool(workspace_id=agent.workspace_id, agent_id=agent.id, tool_name=name, constraints={}) for name in tool_names ]) session.add_all([ AgentPermission(workspace_id=agent.workspace_id, agent_id=agent.id, permission_key=key) for key in permission_keys ]) if schedule: session.add(AgentSchedule( workspace_id=agent.workspace_id, agent_id=agent.id, timezone=schedule.timezone, working_days=list(schedule.working_days), start_minute=schedule.start_minute, end_minute=schedule.end_minute, enabled=schedule.enabled, )) async def _assert_name_available( session: SessionDep, workspace_id: uuid.UUID, name: str, exclude_id: uuid.UUID | None = None ) -> None: statement = select(Agent.id).where(Agent.workspace_id == workspace_id, Agent.name == name) if exclude_id: statement = statement.where(Agent.id != exclude_id) if await session.scalar(statement): raise AppError("AGENT_NAME_CONFLICT", "An agent with this name already exists.", 409) @router.post("", status_code=201) async def create_agent( payload: AgentCreateInput, request: Request, auth: AuthDep, session: SessionDep, provider: ComputeProviderDep, hayva_csrf: Annotated[str | None, Cookie()] = None, x_csrf_token: Annotated[str | None, Header()] = None, ): await authorize(session, workspace_id=auth.workspace_id, user_id=auth.user_id, permission="agents.create") await _require_mutation_csrf(session, auth, hayva_csrf, x_csrf_token) granted = await permissions_for(session, workspace_id=auth.workspace_id, user_id=auth.user_id) _validate_capabilities(payload.tool_names, payload.permission_keys, granted) await _validate_model_profiles( session, auth.workspace_id, payload.preferred_model, payload.fallback_model ) agent_id = uuid.uuid4() await _assert_name_available(session, auth.workspace_id, payload.name) await _validate_relationships( session, workspace_id=auth.workspace_id, agent_id=agent_id, manager_agent_id=payload.manager_agent_id, escalation_agent_id=payload.escalation_agent_id, ) computer: Computer | None = None network_policy: ComputerNetworkPolicy | None = None template: ComputerTemplate | None = None provisioning_operation: ComputerOperation | None = None if payload.computer.create_dedicated: await authorize( session, workspace_id=auth.workspace_id, user_id=auth.user_id, permission="computer.control", ) if payload.computer.system_privilege in {"administrator", "full_system"}: await authorize( session, workspace_id=auth.workspace_id, user_id=auth.user_id, permission="approvals.approve", ) limits = await session.scalar(select(WorkspaceComputerLimit).where( WorkspaceComputerLimit.workspace_id == auth.workspace_id, ).with_for_update()) if not limits: raise AppError("COMPUTER_LIMITS_MISSING", "Computer resource limits are missing.", 409) existing_count = await session.scalar(select(func.count(Computer.id)).where( Computer.workspace_id == auth.workspace_id, Computer.status != "destroyed", )) if (existing_count or 0) >= limits.maximum_computers: raise AppError( "COMPUTER_LIMIT_REACHED", "The workspace computer limit has been reached.", 409 ) template = await computer_template_for_key( session, workspace_id=auth.workspace_id, template_key=payload.computer.template_key, ) if template.os_family != payload.computer.os_family: raise AppError( "COMPUTER_TEMPLATE_OS_MISMATCH", "The selected OS template does not match the requested OS family.", 422, ) if ( payload.computer.cpu_cores < template.minimum_cpu or payload.computer.memory_mb < template.minimum_memory_mb or payload.computer.disk_gb < template.minimum_disk_gb ): raise AppError( "COMPUTER_TEMPLATE_MINIMUM_NOT_MET", "The requested computer resources are below the OS template minimum.", 422, ) if ( payload.computer.cpu_cores > limits.maximum_cpu_per_computer or payload.computer.memory_mb > limits.maximum_memory_mb or payload.computer.disk_gb > limits.maximum_disk_gb ): raise AppError( "COMPUTER_RESOURCE_LIMIT_EXCEEDED", "The requested computer resources exceed workspace limits.", 409, ) computer, network_policy = build_dedicated_computer( workspace_id=auth.workspace_id, created_by_user_id=auth.user_id, agent_name=payload.name, template=template, compute_provider=get_settings().compute_provider, cpu_cores=payload.computer.cpu_cores, memory_mb=payload.computer.memory_mb, disk_gb=payload.computer.disk_gb, browser=payload.computer.browser, persistent_disk=payload.computer.persistent_disk, system_privilege=payload.computer.system_privilege, start_policy=payload.computer.start_policy, network_policy=payload.computer.network_policy.model_dump(), ) session.add_all([computer, network_policy]) await session.flush() agent = Agent( id=agent_id, workspace_id=auth.workspace_id, created_by_user_id=auth.user_id, name=payload.name, description=payload.description, purpose=payload.purpose, status="disabled", autonomy_level=payload.autonomy_level, computer_required=payload.computer.create_dedicated, computer_id=computer.id if computer else None, system_privilege=payload.computer.system_privilege, computer_autonomy=payload.computer.computer_autonomy, computer_network_policy=payload.computer.network_policy.model_dump(), computer_file_policy={"scope": "guest_only", "host_access": False}, preferred_model=payload.preferred_model, fallback_model=payload.fallback_model, manager_agent_id=payload.manager_agent_id, escalation_agent_id=payload.escalation_agent_id, budget_limit_minor=payload.budget_limit_minor, budget_currency=payload.budget_currency, execution_limit=payload.execution_limit, version=1, ) session.add(agent) await session.flush() browser_profile = None if computer: browser_profile = build_private_computer( workspace_id=auth.workspace_id, agent_id=agent.id, computer_id=computer.id, created_by_user_id=auth.user_id, agent_name=agent.name, active=False, ) session.add(browser_profile) provisioning_operation = ComputerOperation( id=uuid.uuid4(), workspace_id=auth.workspace_id, computer_id=computer.id, requested_by_user_id=auth.user_id, operation="create", risk="medium", idempotency_key=f"agent-create:{agent.id}", request_hash=canonical_payload_hash({ "agent_id": str(agent.id), "computer_id": str(computer.id), "template_key": template.template_key, }), request_redacted={"template_key": template.template_key}, status="pending", ) session.add(provisioning_operation) await _persist_components( session, agent, tool_names=payload.tool_names, permission_keys=payload.permission_keys, schedule=payload.schedule, ) session.add(AgentVersion( id=uuid.uuid4(), workspace_id=auth.workspace_id, agent_id=agent.id, version_number=1, system_instructions=payload.system_instructions, personality=payload.personality, config_snapshot=_snapshot( agent, payload.tool_names, payload.permission_keys, payload.schedule ), created_by_user_id=auth.user_id, )) await _audit( session, workspace_id=auth.workspace_id, actor_id=auth.user_id, event_type="agent.created", request_id=request.state.request_id, data={ "agent_id": str(agent.id), "computer_id": str(computer.id) if computer else None, "computer_profile_id": str(browser_profile.id) if browser_profile else None, "autonomy_level": agent.autonomy_level, }, ) if computer and provisioning_operation: await _audit( session, workspace_id=auth.workspace_id, actor_id=auth.user_id, event_type="computer.operation_requested", request_id=request.state.request_id, data={ "agent_id": str(agent.id), "computer_id": str(computer.id), "operation_id": str(provisioning_operation.id), "operation": "create", "risk": provisioning_operation.risk, }, ) await session.commit() if computer and template and network_policy and provisioning_operation: service = ComputerProvisioningService(provider) await service.create_computer( computer, template, network_policy, provisioning_operation ) await _audit( session, workspace_id=auth.workspace_id, actor_id=auth.user_id, event_type=f"computer.operation_{provisioning_operation.status}", request_id=request.state.request_id, data={ "agent_id": str(agent.id), "computer_id": str(computer.id), "operation_id": str(provisioning_operation.id), "operation": "create", "status": provisioning_operation.status, "error_code": provisioning_operation.error_code, }, ) await session.commit() await session.refresh(agent) return {"success": True, "agent": await _agent_data(session, agent, detail=True)} @router.get("") async def list_agents(auth: AuthDep, session: SessionDep): await authorize(session, workspace_id=auth.workspace_id, user_id=auth.user_id, permission="agents.read") agents = list((await session.scalars( select(Agent).where(Agent.workspace_id == auth.workspace_id) .order_by(Agent.created_at, Agent.name).limit(200) )).all()) return {"success": True, "agents": [await _agent_data(session, agent) for agent in agents]} @router.get("/{agent_id}") async def get_agent(agent_id: uuid.UUID, auth: AuthDep, session: SessionDep): await authorize(session, workspace_id=auth.workspace_id, user_id=auth.user_id, permission="agents.read") agent = await _load_agent(session, auth.workspace_id, agent_id) return {"success": True, "agent": await _agent_data(session, agent, detail=True)} @router.patch("/{agent_id}") async def update_agent( agent_id: uuid.UUID, payload: AgentUpdateInput, request: Request, auth: AuthDep, session: SessionDep, hayva_csrf: Annotated[str | None, Cookie()] = None, x_csrf_token: Annotated[str | None, Header()] = None, ): await authorize(session, workspace_id=auth.workspace_id, user_id=auth.user_id, permission="agents.modify") await _require_mutation_csrf(session, auth, hayva_csrf, x_csrf_token) agent = await _load_agent(session, auth.workspace_id, agent_id, lock=True) current_version = await _latest_version(session, auth.workspace_id, agent.id) current_tools, current_permissions, current_schedule = await _components( session, auth.workspace_id, agent.id ) fields = payload.model_fields_set tool_names = payload.tool_names if "tool_names" in fields else current_tools permission_keys = ( payload.permission_keys if "permission_keys" in fields else current_permissions ) schedule = payload.schedule if "schedule" in fields else ( ScheduleInput.model_validate(_schedule_data(current_schedule)) if current_schedule else None ) granted = await permissions_for(session, workspace_id=auth.workspace_id, user_id=auth.user_id) _validate_capabilities(tool_names or [], permission_keys or [], granted) manager_id = payload.manager_agent_id if "manager_agent_id" in fields else agent.manager_agent_id escalation_id = ( payload.escalation_agent_id if "escalation_agent_id" in fields else agent.escalation_agent_id ) await _validate_relationships( session, workspace_id=auth.workspace_id, agent_id=agent.id, manager_agent_id=manager_id, escalation_agent_id=escalation_id, ) if payload.name is not None: await _assert_name_available(session, auth.workspace_id, payload.name, agent.id) for field in ( "name", "description", "purpose", "autonomy_level", "preferred_model", "fallback_model", "budget_limit_minor", "budget_currency", "execution_limit", ): if field in fields: setattr(agent, field, getattr(payload, field)) if agent.preferred_model and agent.preferred_model == agent.fallback_model: raise AppError("AGENT_MODEL_FALLBACK_INVALID", "Fallback model must differ.", 422) await _validate_model_profiles( session, auth.workspace_id, agent.preferred_model, agent.fallback_model ) agent.manager_agent_id = manager_id agent.escalation_agent_id = escalation_id if agent.autonomy_level == 0: agent.status = "disabled" browser_profile = None if agent.computer_id: await dedicated_computer_for_agent( session, workspace_id=auth.workspace_id, agent_id=agent.id, lock=True ) browser_profile = await private_computer_for_agent( session, workspace_id=auth.workspace_id, agent_id=agent.id, lock=True ) if agent.status != "active" and browser_profile and browser_profile.status == "active": browser_profile.status = "disabled" browser_profile.version += 1 agent.version += 1 await _persist_components( session, agent, tool_names=tool_names or [], permission_keys=permission_keys or [], schedule=schedule, ) session.add(AgentVersion( id=uuid.uuid4(), workspace_id=auth.workspace_id, agent_id=agent.id, version_number=agent.version, system_instructions=( payload.system_instructions if "system_instructions" in fields else current_version.system_instructions ) or "", personality=( payload.personality if "personality" in fields else current_version.personality ) or "", config_snapshot=_snapshot(agent, tool_names or [], permission_keys or [], schedule), created_by_user_id=auth.user_id, )) await _audit( session, workspace_id=auth.workspace_id, actor_id=auth.user_id, event_type="agent.updated", request_id=request.state.request_id, data={"agent_id": str(agent.id), "version": agent.version}, ) await session.commit() await session.refresh(agent) return {"success": True, "agent": await _agent_data(session, agent, detail=True)} @router.post("/{agent_id}/status") async def set_agent_status( agent_id: uuid.UUID, payload: AgentStatusInput, request: Request, auth: AuthDep, session: SessionDep, hayva_csrf: Annotated[str | None, Cookie()] = None, x_csrf_token: Annotated[str | None, Header()] = None, ): await authorize(session, workspace_id=auth.workspace_id, user_id=auth.user_id, permission="agents.modify") await _require_mutation_csrf(session, auth, hayva_csrf, x_csrf_token) agent = await _load_agent(session, auth.workspace_id, agent_id, lock=True) dedicated = None browser_profile = None if agent.computer_id: dedicated = await dedicated_computer_for_agent( session, workspace_id=auth.workspace_id, agent_id=agent.id, lock=True ) browser_profile = await private_computer_for_agent( session, workspace_id=auth.workspace_id, agent_id=agent.id, lock=True ) if payload.status == "active" and agent.autonomy_level == 0: raise AppError("AGENT_AUTONOMY_DISABLED", "A disabled-autonomy agent cannot be enabled.", 409) if payload.status == "active" and not dedicated: raise AppError( "AGENT_DEDICATED_COMPUTER_REQUIRED", "A runtime agent cannot be enabled without a dedicated computer.", 409, ) if payload.status == "active" and ( not dedicated.provider_resource_id or dedicated.status not in {"running", "idle", "stopped", "suspended"} ): raise AppError( "AGENT_DEDICATED_COMPUTER_UNAVAILABLE", "Provision and verify a stable dedicated computer before enabling this agent.", 409, ) if payload.status == "active" and browser_profile and browser_profile.status == "revoked": raise AppError( "AGENT_PRIVATE_COMPUTER_REVOKED", "A revoked private computer cannot be reactivated.", 409, ) if agent.status != payload.status: agent.status = payload.status if browser_profile: browser_profile.status = "active" if payload.status == "active" else "disabled" browser_profile.version += 1 agent.version += 1 latest = await _latest_version(session, auth.workspace_id, agent.id) tools, permission_keys, schedule = await _components(session, auth.workspace_id, agent.id) session.add(AgentVersion( id=uuid.uuid4(), workspace_id=auth.workspace_id, agent_id=agent.id, version_number=agent.version, system_instructions=latest.system_instructions, personality=latest.personality, config_snapshot=_snapshot(agent, tools, permission_keys, schedule), created_by_user_id=auth.user_id, )) await _audit( session, workspace_id=auth.workspace_id, actor_id=auth.user_id, event_type=f"agent.{payload.status}", request_id=request.state.request_id, data={"agent_id": str(agent.id), "version": agent.version}, ) await session.commit() await session.refresh(agent) return {"success": True, "agent": await _agent_data(session, agent, detail=True)} @router.post("/{agent_id}/duplicate", status_code=201) async def duplicate_agent( agent_id: uuid.UUID, payload: DuplicateAgentInput, request: Request, auth: AuthDep, session: SessionDep, provider: ComputeProviderDep, hayva_csrf: Annotated[str | None, Cookie()] = None, x_csrf_token: Annotated[str | None, Header()] = None, ): await authorize(session, workspace_id=auth.workspace_id, user_id=auth.user_id, permission="agents.create") await _require_mutation_csrf(session, auth, hayva_csrf, x_csrf_token) source = await _load_agent(session, auth.workspace_id, agent_id) latest = await _latest_version(session, auth.workspace_id, source.id) tools, permission_keys, schedule = await _components(session, auth.workspace_id, source.id) name = (payload.name or f"{source.name} Copy").strip() source_computer = ( await dedicated_computer_for_agent( session, workspace_id=auth.workspace_id, agent_id=source.id ) if source.computer_required else None ) source_template_key = DEFAULT_COMPUTER_TEMPLATE_KEY if source_computer: source_template_key = await session.scalar(select(ComputerTemplate.template_key).where( ComputerTemplate.workspace_id == auth.workspace_id, ComputerTemplate.id == source_computer.template_id, )) or DEFAULT_COMPUTER_TEMPLATE_KEY create_payload = AgentCreateInput( name=name, description=source.description, purpose=source.purpose, system_instructions=latest.system_instructions, personality=latest.personality, autonomy_level=source.autonomy_level, preferred_model=source.preferred_model, fallback_model=source.fallback_model, manager_agent_id=source.manager_agent_id, escalation_agent_id=source.escalation_agent_id, budget_limit_minor=source.budget_limit_minor, budget_currency=source.budget_currency, execution_limit=source.execution_limit, tool_names=tools, permission_keys=permission_keys, schedule=ScheduleInput.model_validate(_schedule_data(schedule)) if schedule else None, computer=AgentComputerInput( create_dedicated=source.computer_required, os_family=source_computer.os_family if source_computer else "linux", template_key=source_template_key, cpu_cores=source_computer.cpu_cores if source_computer else 2, memory_mb=source_computer.memory_mb if source_computer else 4096, disk_gb=source_computer.disk_gb if source_computer else 40, browser=source_computer.browser if source_computer else "chromium", persistent_disk=(source_computer.persistent_disk if source_computer else True), system_privilege=source.system_privilege, computer_autonomy=source.computer_autonomy, start_policy=(source_computer.start_policy if source_computer else "start_when_needed"), network_policy=ComputerNetworkInput.model_validate(source.computer_network_policy), ), ) return await create_agent( create_payload, request, auth, session, provider, hayva_csrf, x_csrf_token ) @router.delete("/{agent_id}") async def delete_agent( agent_id: uuid.UUID, request: Request, auth: AuthDep, session: SessionDep, hayva_csrf: Annotated[str | None, Cookie()] = None, x_csrf_token: Annotated[str | None, Header()] = None, ): await authorize(session, workspace_id=auth.workspace_id, user_id=auth.user_id, permission="agents.delete") await _require_mutation_csrf(session, auth, hayva_csrf, x_csrf_token) agent = await _load_agent(session, auth.workspace_id, agent_id, lock=True) if agent.status != "archived": raise AppError("AGENT_DELETE_REQUIRES_ARCHIVE", "Archive the agent before deletion.", 409) if agent.computer_id: dedicated = await dedicated_computer_for_agent( session, workspace_id=auth.workspace_id, agent_id=agent.id, lock=True ) if dedicated.status != "destroyed": raise AppError( "AGENT_DELETE_REQUIRES_COMPUTER_DESTROY", "Destroy and verify the dedicated computer before deleting its agent.", 409, ) dependent = await session.scalar(select(func.count(Agent.id)).where( Agent.workspace_id == auth.workspace_id, (Agent.manager_agent_id == agent.id) | (Agent.escalation_agent_id == agent.id), )) delegation = await session.scalar(select(func.count(AgentDelegation.id)).where( AgentDelegation.workspace_id == auth.workspace_id, (AgentDelegation.source_agent_id == agent.id) | (AgentDelegation.target_agent_id == agent.id), )) execution = await session.scalar(select(func.count(Execution.id)).where( Execution.workspace_id == auth.workspace_id, Execution.agent_id == agent.id )) computer_session = await session.scalar(select(func.count(ComputerSession.id)).where( ComputerSession.workspace_id == auth.workspace_id, ComputerSession.agent_id == agent.id, )) if dependent or delegation or execution or computer_session: raise AppError("AGENT_DELETE_REFERENCED", "The agent is still referenced.", 409) await _audit( session, workspace_id=auth.workspace_id, actor_id=auth.user_id, event_type="agent.deleted", request_id=request.state.request_id, data={"agent_id": str(agent.id), "last_version": agent.version}, ) for model in (AgentSchedule, AgentTool, AgentPermission, AgentVersion): await session.execute(delete(model).where( model.workspace_id == auth.workspace_id, model.agent_id == agent.id )) await session.execute(delete(ComputerProfile).where( ComputerProfile.workspace_id == auth.workspace_id, ComputerProfile.agent_id == agent.id, )) await session.delete(agent) await session.commit() return {"success": True, "deleted_agent_id": str(agent_id)} @router.post("/{source_agent_id}/delegations", status_code=201) async def create_delegation( source_agent_id: uuid.UUID, payload: DelegationInput, request: Request, auth: AuthDep, session: SessionDep, hayva_csrf: Annotated[str | None, Cookie()] = None, x_csrf_token: Annotated[str | None, Header()] = None, ): await authorize(session, workspace_id=auth.workspace_id, user_id=auth.user_id, permission="agents.modify") await _require_mutation_csrf(session, auth, hayva_csrf, x_csrf_token) source = await _load_agent(session, auth.workspace_id, source_agent_id) target = await _load_agent(session, auth.workspace_id, payload.target_agent_id) if source.id == target.id: raise AppError("DELEGATION_SELF", "An agent cannot delegate to itself.", 409) if source.status != "active" or target.status != "active": raise AppError("DELEGATION_AGENT_DISABLED", "Both agents must be active.", 409) actor_permissions = await permissions_for( session, workspace_id=auth.workspace_id, user_id=auth.user_id ) source_permissions = set((await session.scalars(select(AgentPermission.permission_key).where( AgentPermission.workspace_id == auth.workspace_id, AgentPermission.agent_id == source.id, ))).all()) target_permissions = set((await session.scalars(select(AgentPermission.permission_key).where( AgentPermission.workspace_id == auth.workspace_id, AgentPermission.agent_id == target.id, ))).all()) maximum_scope = source_permissions & target_permissions & actor_permissions requested_scope = set(payload.permission_scope) if payload.permission_scope else maximum_scope if not requested_scope or not requested_scope <= maximum_scope: raise AppError( "DELEGATION_SCOPE_INVALID", "Delegation scope must be a non-empty permission intersection.", 403, ) delegation = AgentDelegation( id=uuid.uuid4(), workspace_id=auth.workspace_id, source_agent_id=source.id, target_agent_id=target.id, requested_by_user_id=auth.user_id, instruction=payload.instruction, instruction_hash=canonical_payload_hash({"instruction": payload.instruction}), permission_scope=sorted(requested_scope), status="requested", ) session.add(delegation) await _audit( session, workspace_id=auth.workspace_id, actor_id=auth.user_id, event_type="agent.delegation_requested", request_id=request.state.request_id, data={"delegation_id": str(delegation.id), "source_agent_id": str(source.id), "target_agent_id": str(target.id), "permission_scope": sorted(requested_scope)}, ) await session.commit() return {"success": True, "delegation": { "id": str(delegation.id), "source_agent_id": str(source.id), "target_agent_id": str(target.id), "status": delegation.status, "permission_scope": delegation.permission_scope, "created_at": delegation.created_at, }}