Pipeline/backend/app/schemas/agent_sessions.py

150 lines
3.5 KiB
Python

"""Provider-neutral schemas for local agent session sources."""
from __future__ import annotations
from datetime import datetime
from typing import Any, Literal
from sqlmodel import SQLModel
AgentSessionSource = Literal["claude_code", "codex_cli", "openai_api"]
AgentSourceStatus = Literal["available", "unavailable", "unsupported"]
class TextBlock(SQLModel):
text: str
truncated: bool
class ThinkingBlock(SQLModel):
text: str
truncated: bool
class ToolUseBlock(SQLModel):
tool_use_id: str
tool_name: str
input: dict[str, Any]
input_truncated: bool
result: str | None = None
result_truncated: bool = False
is_error: bool = False
class SessionTokenUsageRead(SQLModel):
input: int
output: int
cache_read: int
cache_write: int
class SessionMessage(SQLModel):
uuid: str
role: str
timestamp: datetime | None = None
text_blocks: list[TextBlock]
thinking_blocks: list[ThinkingBlock]
tool_uses: list[ToolUseBlock]
model: str | None = None
tokens: SessionTokenUsageRead | None = None
class SessionMessagesResponse(SQLModel):
session_id: str
source: AgentSessionSource
source_status: AgentSourceStatus
source_path: str | None = None
last_scanned_at: datetime | None = None
messages: list[SessionMessage]
total: int
has_more: bool
class SessionTokensRead(SQLModel):
input: int
output: int
cache_read: int
cache_write: int
total: int
class AgentSessionRead(SQLModel):
session_id: str
source: AgentSessionSource
provider_label: str
project_dir: str
cwd: str | None = None
title: str | None = None
models: list[str]
tokens: SessionTokensRead
cost_usd: float
billing_source: str
message_count: int
first_message_at: datetime | None = None
last_message_at: datetime | None = None
is_active: bool
entrypoints: list[str]
git_branch: str | None = None
version: str | None = None
class AgentSessionStatsRead(SQLModel):
session_count: int
active_sessions: int
total_tokens: int
total_cost_usd: float
models: list[str]
class AgentSessionListResponse(SQLModel):
sessions: list[AgentSessionRead]
total: int
stats: AgentSessionStatsRead
source: AgentSessionSource
provider_label: str
source_status: AgentSourceStatus
source_path: str | None = None
last_scanned_at: datetime | None = None
unavailable_reason: str | None = None
setup_hint: str | None = None
class FileEntry(SQLModel):
path: str
count: int
class CommandEntry(SQLModel):
command: str
count: int
class ToolAnalyticsResponse(SQLModel):
tool_counts: dict[str, int]
top_files_read: list[FileEntry]
top_files_written: list[FileEntry]
top_commands: list[CommandEntry]
session_count: int
date_range_days: int
source: AgentSessionSource | None = None
source_status: AgentSourceStatus | None = None
source_path: str | None = None
last_scanned_at: datetime | None = None
class AgentSessionSourceRead(SQLModel):
source: AgentSessionSource
provider_label: str
source_status: AgentSourceStatus
source_path: str | None = None
session_count: int
last_activity_at: datetime | None = None
last_scanned_at: datetime | None = None
unavailable_reason: str | None = None
setup_hint: str | None = None
class AgentSessionSourcesResponse(SQLModel):
sources: list[AgentSessionSourceRead]
last_scanned_at: datetime