File size: 3,320 Bytes
67b16c6 6155b26 67b16c6 79b2fcc ff8c636 67b16c6 82bad21 67b16c6 0611031 67b16c6 092f909 67b16c6 27da720 77324b8 67b16c6 d7f2a7c 67b16c6 79b2fcc 27da720 bf564b6 d9d9785 6155b26 77324b8 6155b26 67b16c6 77324b8 67b16c6 79b2fcc 754345f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Pydantic models for API requests and responses."""
from enum import Enum
from typing import Any
from pydantic import BaseModel, Field
class OpType(str, Enum):
"""Operation types matching agent/core/agent_loop.py."""
USER_INPUT = "user_input"
EXEC_APPROVAL = "exec_approval"
INTERRUPT = "interrupt"
UNDO = "undo"
COMPACT = "compact"
SHUTDOWN = "shutdown"
class Operation(BaseModel):
"""Operation to be submitted to the agent."""
op_type: OpType
data: dict[str, Any] | None = None
class Submission(BaseModel):
"""Submission wrapper with ID and operation."""
id: str
operation: Operation
class ToolApproval(BaseModel):
"""Approval decision for a single tool call."""
tool_call_id: str
approved: bool
feedback: str | None = None
edited_script: str | None = None
namespace: str | None = None
class ApprovalRequest(BaseModel):
"""Request to approve/reject tool calls."""
session_id: str
approvals: list[ToolApproval]
class SubmitRequest(BaseModel):
"""Request to submit user input."""
session_id: str
# Cap text size to prevent context-bloat / cost-amplification: a malicious
# or runaway client could otherwise attach megabytes that then ride along
# in every subsequent turn until /api/compact is called.
text: str = Field(..., min_length=1, max_length=100_000)
class TruncateRequest(BaseModel):
"""Request to truncate conversation history to before a specific user message."""
user_message_index: int
class SessionResponse(BaseModel):
"""Response when creating a new session."""
session_id: str
ready: bool = True
model: str | None = None
class PendingApprovalTool(BaseModel):
"""A tool waiting for user approval."""
tool: str
tool_call_id: str
arguments: dict[str, Any] = {}
class SessionAutoApprovalInfo(BaseModel):
"""Per-session auto-approval budget state."""
enabled: bool = False
cost_cap_usd: float | None = None
estimated_spend_usd: float = 0.0
remaining_usd: float | None = None
class SessionInfo(BaseModel):
"""Session metadata."""
session_id: str
created_at: str
is_active: bool
is_processing: bool = False
message_count: int
user_id: str = "dev"
pending_approval: list[PendingApprovalTool] | None = None
model: str | None = None
title: str | None = None
notification_destinations: list[str] = Field(default_factory=list)
auto_approval: SessionAutoApprovalInfo = Field(
default_factory=SessionAutoApprovalInfo
)
class SessionNotificationsRequest(BaseModel):
"""Replace the session's auto-notification destinations."""
destinations: list[str]
class SessionYoloRequest(BaseModel):
"""Update a session's auto-approval policy."""
enabled: bool
cost_cap_usd: float | None = Field(default=None, ge=0)
class HealthResponse(BaseModel):
"""Health check response."""
status: str = "ok"
active_sessions: int = 0
max_sessions: int = 0
class LLMHealthResponse(BaseModel):
"""LLM provider health check response."""
status: str # "ok" | "error"
model: str
error: str | None = None
error_type: str | None = (
None # "auth" | "credits" | "rate_limit" | "network" | "unknown"
)
|