types.py

"""
Type definitions for the Copilot SDK
"""

from __future__ import annotations

from collections.abc import Awaitable, Callable
from dataclasses import KW_ONLY, dataclass, field
from typing import Any, Literal, NotRequired, Required, TypedDict

# Import generated SessionEvent types
from .generated.session_events import (
    PermissionRequest,
    SessionEvent,
)

# SessionEvent is now imported from generated types
# It provides proper type discrimination for all event types

# Valid reasoning effort levels for models that support it
ReasoningEffort = Literal["low", "medium", "high", "xhigh"]

# Connection state
ConnectionState = Literal["disconnected", "connecting", "connected", "error"]

# Log level type
LogLevel = Literal["none", "error", "warning", "info", "debug", "all"]


# Selection range for text attachments
class SelectionRange(TypedDict):
    line: int
    character: int


class Selection(TypedDict):
    start: SelectionRange
    end: SelectionRange


# Attachment types - discriminated union based on 'type' field
class FileAttachment(TypedDict):
    """File attachment."""

    type: Literal["file"]
    path: str
    displayName: NotRequired[str]


class DirectoryAttachment(TypedDict):
    """Directory attachment."""

    type: Literal["directory"]
    path: str
    displayName: NotRequired[str]


class SelectionAttachment(TypedDict):
    """Selection attachment with text from a file."""

    type: Literal["selection"]
    filePath: str
    displayName: str
    selection: NotRequired[Selection]
    text: NotRequired[str]


class BlobAttachment(TypedDict):
    """Inline base64-encoded content attachment (e.g. images)."""

    type: Literal["blob"]
    data: str
    """Base64-encoded content"""
    mimeType: str
    """MIME type of the inline data"""
    displayName: NotRequired[str]


# Attachment type - union of all attachment types
Attachment = FileAttachment | DirectoryAttachment | SelectionAttachment | BlobAttachment


# Configuration for OpenTelemetry integration with the Copilot CLI.
class TelemetryConfig(TypedDict, total=False):
    """Configuration for OpenTelemetry integration with the Copilot CLI."""

    otlp_endpoint: str
    """OTLP HTTP endpoint URL for trace/metric export. Sets OTEL_EXPORTER_OTLP_ENDPOINT."""
    file_path: str
    """File path for JSON-lines trace output. Sets COPILOT_OTEL_FILE_EXPORTER_PATH."""
    exporter_type: str
    """Exporter backend type: "otlp-http" or "file". Sets COPILOT_OTEL_EXPORTER_TYPE."""
    source_name: str
    """Instrumentation scope name. Sets COPILOT_OTEL_SOURCE_NAME."""
    capture_content: bool
    """Whether to capture message content. Sets OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT."""  # noqa: E501


# Configuration for CopilotClient connection modes


@dataclass
class SubprocessConfig:
    """Config for spawning a local Copilot CLI subprocess.

    Example:
        >>> config = SubprocessConfig(github_token="ghp_...")
        >>> client = CopilotClient(config)

        >>> # Custom CLI path with TCP transport
        >>> config = SubprocessConfig(
        ...     cli_path="/usr/local/bin/copilot",
        ...     use_stdio=False,
        ...     log_level="debug",
        ... )
    """

    cli_path: str | None = None
    """Path to the Copilot CLI executable. ``None`` uses the bundled binary."""

    cli_args: list[str] = field(default_factory=list)
    """Extra arguments passed to the CLI executable (inserted before SDK-managed args)."""

    _: KW_ONLY

    cwd: str | None = None
    """Working directory for the CLI process. ``None`` uses the current directory."""

    use_stdio: bool = True
    """Use stdio transport (``True``, default) or TCP (``False``)."""

    port: int = 0
    """TCP port for the CLI server (only when ``use_stdio=False``). 0 means random."""

    log_level: LogLevel = "info"
    """Log level for the CLI process."""

    env: dict[str, str] | None = None
    """Environment variables for the CLI process. ``None`` inherits the current env."""

    github_token: str | None = None
    """GitHub token for authentication. Takes priority over other auth methods."""

    use_logged_in_user: bool | None = None
    """Use the logged-in user for authentication.

    ``None`` (default) resolves to ``True`` unless ``github_token`` is set.
    """

    telemetry: TelemetryConfig | None = None
    """OpenTelemetry configuration. Providing this enables telemetry — no separate flag needed."""


@dataclass
class ExternalServerConfig:
    """Config for connecting to an existing Copilot CLI server over TCP.

    Example:
        >>> config = ExternalServerConfig(url="localhost:3000")
        >>> client = CopilotClient(config)
    """

    url: str
    """Server URL. Supports ``"host:port"``, ``"http://host:port"``, or just ``"port"``."""


ToolResultType = Literal["success", "failure", "rejected", "denied"]


@dataclass
class ToolBinaryResult:
    """Binary content returned by a tool."""

    data: str = ""
    mime_type: str = ""
    type: str = ""
    description: str = ""


@dataclass
class ToolResult:
    """Result of a tool invocation."""

    text_result_for_llm: str = ""
    result_type: ToolResultType = "success"
    error: str | None = None
    binary_results_for_llm: list[ToolBinaryResult] | None = None
    session_log: str | None = None
    tool_telemetry: dict[str, Any] | None = None


@dataclass
class ToolInvocation:
    """Context passed to a tool handler when invoked."""

    session_id: str = ""
    tool_call_id: str = ""
    tool_name: str = ""
    arguments: Any = None


ToolHandler = Callable[[ToolInvocation], ToolResult | Awaitable[ToolResult]]


@dataclass
class Tool:
    name: str
    description: str
    handler: ToolHandler
    parameters: dict[str, Any] | None = None
    overrides_built_in_tool: bool = False
    skip_permission: bool = False


# System message configuration (discriminated union)
# Use SystemMessageAppendConfig for default behavior,
# SystemMessageReplaceConfig for full control,
# or SystemMessageCustomizeConfig for section-level overrides.

# Known system prompt section identifiers for the "customize" mode.
SystemPromptSection = Literal[
    "identity",
    "tone",
    "tool_efficiency",
    "environment_context",
    "code_change_rules",
    "guidelines",
    "safety",
    "tool_instructions",
    "custom_instructions",
    "last_instructions",
]

SYSTEM_PROMPT_SECTIONS: dict[SystemPromptSection, str] = {
    "identity": "Agent identity preamble and mode statement",
    "tone": "Response style, conciseness rules, output formatting preferences",
    "tool_efficiency": "Tool usage patterns, parallel calling, batching guidelines",
    "environment_context": "CWD, OS, git root, directory listing, available tools",
    "code_change_rules": "Coding rules, linting/testing, ecosystem tools, style",
    "guidelines": "Tips, behavioral best practices, behavioral guidelines",
    "safety": "Environment limitations, prohibited actions, security policies",
    "tool_instructions": "Per-tool usage instructions",
    "custom_instructions": "Repository and organization custom instructions",
    "last_instructions": (
        "End-of-prompt instructions: parallel tool calling, persistence, task completion"
    ),
}


SectionTransformFn = Callable[[str], str | Awaitable[str]]
"""Transform callback: receives current section content, returns new content."""

SectionOverrideAction = Literal["replace", "remove", "append", "prepend"] | SectionTransformFn
"""Override action: a string literal for static overrides, or a callback for transforms."""


class SectionOverride(TypedDict, total=False):
    """Override operation for a single system prompt section."""

    action: Required[SectionOverrideAction]
    content: NotRequired[str]


class SystemMessageAppendConfig(TypedDict, total=False):
    """
    Append mode: Use CLI foundation with optional appended content.
    """

    mode: NotRequired[Literal["append"]]
    content: NotRequired[str]


class SystemMessageReplaceConfig(TypedDict):
    """
    Replace mode: Use caller-provided system message entirely.
    Removes all SDK guardrails including security restrictions.
    """

    mode: Literal["replace"]
    content: str


class SystemMessageCustomizeConfig(TypedDict, total=False):
    """
    Customize mode: Override individual sections of the system prompt.
    Keeps the SDK-managed prompt structure while allowing targeted modifications.
    """

    mode: Required[Literal["customize"]]
    sections: NotRequired[dict[SystemPromptSection, SectionOverride]]
    content: NotRequired[str]


# Union type - use one based on your needs
SystemMessageConfig = (
    SystemMessageAppendConfig | SystemMessageReplaceConfig | SystemMessageCustomizeConfig
)


# Permission result types

PermissionRequestResultKind = Literal[
    "approved",
    "denied-by-rules",
    "denied-by-content-exclusion-policy",
    "denied-no-approval-rule-and-could-not-request-from-user",
    "denied-interactively-by-user",
    "no-result",
]


@dataclass
class PermissionRequestResult:
    """Result of a permission request."""

    kind: PermissionRequestResultKind = "denied-no-approval-rule-and-could-not-request-from-user"
    rules: list[Any] | None = None
    feedback: str | None = None
    message: str | None = None
    path: str | None = None


_PermissionHandlerFn = Callable[
    [PermissionRequest, dict[str, str]],
    PermissionRequestResult | Awaitable[PermissionRequestResult],
]


class PermissionHandler:
    @staticmethod
    def approve_all(
        request: PermissionRequest, invocation: dict[str, str]
    ) -> PermissionRequestResult:
        return PermissionRequestResult(kind="approved")


# ============================================================================
# User Input Request Types
# ============================================================================


class UserInputRequest(TypedDict, total=False):
    """Request for user input from the agent (enables ask_user tool)"""

    question: str
    choices: list[str]
    allowFreeform: bool


class UserInputResponse(TypedDict):
    """Response to a user input request"""

    answer: str
    wasFreeform: bool


UserInputHandler = Callable[
    [UserInputRequest, dict[str, str]],
    UserInputResponse | Awaitable[UserInputResponse],
]


# ============================================================================
# Hook Types
# ============================================================================


class BaseHookInput(TypedDict):
    """Base interface for all hook inputs"""

    timestamp: int
    cwd: str


class PreToolUseHookInput(TypedDict):
    """Input for pre-tool-use hook"""

    timestamp: int
    cwd: str
    toolName: str
    toolArgs: Any


class PreToolUseHookOutput(TypedDict, total=False):
    """Output for pre-tool-use hook"""

    permissionDecision: Literal["allow", "deny", "ask"]
    permissionDecisionReason: str
    modifiedArgs: Any
    additionalContext: str
    suppressOutput: bool


PreToolUseHandler = Callable[
    [PreToolUseHookInput, dict[str, str]],
    PreToolUseHookOutput | None | Awaitable[PreToolUseHookOutput | None],
]


class PostToolUseHookInput(TypedDict):
    """Input for post-tool-use hook"""

    timestamp: int
    cwd: str
    toolName: str
    toolArgs: Any
    toolResult: Any


class PostToolUseHookOutput(TypedDict, total=False):
    """Output for post-tool-use hook"""

    modifiedResult: Any
    additionalContext: str
    suppressOutput: bool


PostToolUseHandler = Callable[
    [PostToolUseHookInput, dict[str, str]],
    PostToolUseHookOutput | None | Awaitable[PostToolUseHookOutput | None],
]


class UserPromptSubmittedHookInput(TypedDict):
    """Input for user-prompt-submitted hook"""

    timestamp: int
    cwd: str
    prompt: str


class UserPromptSubmittedHookOutput(TypedDict, total=False):
    """Output for user-prompt-submitted hook"""

    modifiedPrompt: str
    additionalContext: str
    suppressOutput: bool


UserPromptSubmittedHandler = Callable[
    [UserPromptSubmittedHookInput, dict[str, str]],
    UserPromptSubmittedHookOutput | None | Awaitable[UserPromptSubmittedHookOutput | None],
]


class SessionStartHookInput(TypedDict):
    """Input for session-start hook"""

    timestamp: int
    cwd: str
    source: Literal["startup", "resume", "new"]
    initialPrompt: NotRequired[str]


class SessionStartHookOutput(TypedDict, total=False):
    """Output for session-start hook"""

    additionalContext: str
    modifiedConfig: dict[str, Any]


SessionStartHandler = Callable[
    [SessionStartHookInput, dict[str, str]],
    SessionStartHookOutput | None | Awaitable[SessionStartHookOutput | None],
]


class SessionEndHookInput(TypedDict):
    """Input for session-end hook"""

    timestamp: int
    cwd: str
    reason: Literal["complete", "error", "abort", "timeout", "user_exit"]
    finalMessage: NotRequired[str]
    error: NotRequired[str]


class SessionEndHookOutput(TypedDict, total=False):
    """Output for session-end hook"""

    suppressOutput: bool
    cleanupActions: list[str]
    sessionSummary: str


SessionEndHandler = Callable[
    [SessionEndHookInput, dict[str, str]],
    SessionEndHookOutput | None | Awaitable[SessionEndHookOutput | None],
]


class ErrorOccurredHookInput(TypedDict):
    """Input for error-occurred hook"""

    timestamp: int
    cwd: str
    error: str
    errorContext: Literal["model_call", "tool_execution", "system", "user_input"]
    recoverable: bool


class ErrorOccurredHookOutput(TypedDict, total=False):
    """Output for error-occurred hook"""

    suppressOutput: bool
    errorHandling: Literal["retry", "skip", "abort"]
    retryCount: int
    userNotification: str


ErrorOccurredHandler = Callable[
    [ErrorOccurredHookInput, dict[str, str]],
    ErrorOccurredHookOutput | None | Awaitable[ErrorOccurredHookOutput | None],
]


class SessionHooks(TypedDict, total=False):
    """Configuration for session hooks"""

    on_pre_tool_use: PreToolUseHandler
    on_post_tool_use: PostToolUseHandler
    on_user_prompt_submitted: UserPromptSubmittedHandler
    on_session_start: SessionStartHandler
    on_session_end: SessionEndHandler
    on_error_occurred: ErrorOccurredHandler


# ============================================================================
# MCP Server Configuration Types
# ============================================================================


class MCPLocalServerConfig(TypedDict, total=False):
    """Configuration for a local/stdio MCP server."""

    tools: list[str]  # List of tools to include. [] means none. "*" means all.
    type: NotRequired[Literal["local", "stdio"]]  # Server type
    timeout: NotRequired[int]  # Timeout in milliseconds
    command: str  # Command to run
    args: list[str]  # Command arguments
    env: NotRequired[dict[str, str]]  # Environment variables
    cwd: NotRequired[str]  # Working directory


class MCPRemoteServerConfig(TypedDict, total=False):
    """Configuration for a remote MCP server (HTTP or SSE)."""

    tools: list[str]  # List of tools to include. [] means none. "*" means all.
    type: Literal["http", "sse"]  # Server type
    timeout: NotRequired[int]  # Timeout in milliseconds
    url: str  # URL of the remote server
    headers: NotRequired[dict[str, str]]  # HTTP headers


MCPServerConfig = MCPLocalServerConfig | MCPRemoteServerConfig


# ============================================================================
# Custom Agent Configuration Types
# ============================================================================


class CustomAgentConfig(TypedDict, total=False):
    """Configuration for a custom agent."""

    name: str  # Unique name of the custom agent
    display_name: NotRequired[str]  # Display name for UI purposes
    description: NotRequired[str]  # Description of what the agent does
    # List of tool names the agent can use
    tools: NotRequired[list[str] | None]
    prompt: str  # The prompt content for the agent
    # MCP servers specific to agent
    mcp_servers: NotRequired[dict[str, MCPServerConfig]]
    infer: NotRequired[bool]  # Whether agent is available for model inference


class InfiniteSessionConfig(TypedDict, total=False):
    """
    Configuration for infinite sessions with automatic context compaction
    and workspace persistence.

    When enabled, sessions automatically manage context window limits through
    background compaction and persist state to a workspace directory.
    """

    # Whether infinite sessions are enabled (default: True)
    enabled: bool
    # Context utilization threshold (0.0-1.0) at which background compaction starts.
    # Compaction runs asynchronously, allowing the session to continue processing.
    # Default: 0.80
    background_compaction_threshold: float
    # Context utilization threshold (0.0-1.0) at which the session blocks until
    # compaction completes. This prevents context overflow when compaction hasn't
    # finished in time. Default: 0.95
    buffer_exhaustion_threshold: float


# Azure-specific provider options
class AzureProviderOptions(TypedDict, total=False):
    """Azure-specific provider configuration"""

    api_version: str  # Azure API version. Defaults to "2024-10-21".


# Configuration for a custom API provider
class ProviderConfig(TypedDict, total=False):
    """Configuration for a custom API provider"""

    type: Literal["openai", "azure", "anthropic"]
    wire_api: Literal["completions", "responses"]
    base_url: str
    api_key: str
    # Bearer token for authentication. Sets the Authorization header directly.
    # Use this for services requiring bearer token auth instead of API key.
    # Takes precedence over api_key when both are set.
    bearer_token: str
    azure: AzureProviderOptions  # Azure-specific options


# Event handler type
SessionEventHandler = Callable[[SessionEvent], None]


# Response from ping
@dataclass
class PingResponse:
    """Response from ping"""

    message: str  # Echo message with "pong: " prefix
    timestamp: int  # Server timestamp in milliseconds
    protocolVersion: int  # Protocol version for SDK compatibility

    @staticmethod
    def from_dict(obj: Any) -> PingResponse:
        assert isinstance(obj, dict)
        message = obj.get("message")
        timestamp = obj.get("timestamp")
        protocolVersion = obj.get("protocolVersion")
        if message is None or timestamp is None or protocolVersion is None:
            raise ValueError(
                f"Missing required fields in PingResponse: message={message}, "
                f"timestamp={timestamp}, protocolVersion={protocolVersion}"
            )
        return PingResponse(str(message), int(timestamp), int(protocolVersion))

    def to_dict(self) -> dict:
        result: dict = {}
        result["message"] = self.message
        result["timestamp"] = self.timestamp
        result["protocolVersion"] = self.protocolVersion
        return result


# Error information from client stop
@dataclass
class StopError(Exception):
    """Error that occurred during client stop cleanup."""

    message: str  # Error message describing what failed during cleanup

    def __post_init__(self) -> None:
        Exception.__init__(self, self.message)

    @staticmethod
    def from_dict(obj: Any) -> StopError:
        assert isinstance(obj, dict)
        message = obj.get("message")
        if message is None:
            raise ValueError("Missing required field 'message' in StopError")
        return StopError(str(message))

    def to_dict(self) -> dict:
        result: dict = {}
        result["message"] = self.message
        return result


# Response from status.get
@dataclass
class GetStatusResponse:
    """Response from status.get"""

    version: str  # Package version (e.g., "1.0.0")
    protocolVersion: int  # Protocol version for SDK compatibility

    @staticmethod
    def from_dict(obj: Any) -> GetStatusResponse:
        assert isinstance(obj, dict)
        version = obj.get("version")
        protocolVersion = obj.get("protocolVersion")
        if version is None or protocolVersion is None:
            raise ValueError(
                f"Missing required fields in GetStatusResponse: version={version}, "
                f"protocolVersion={protocolVersion}"
            )
        return GetStatusResponse(str(version), int(protocolVersion))

    def to_dict(self) -> dict:
        result: dict = {}
        result["version"] = self.version
        result["protocolVersion"] = self.protocolVersion
        return result


# Response from auth.getStatus
@dataclass
class GetAuthStatusResponse:
    """Response from auth.getStatus"""

    isAuthenticated: bool  # Whether the user is authenticated
    authType: str | None = None  # Authentication type
    host: str | None = None  # GitHub host URL
    login: str | None = None  # User login name
    statusMessage: str | None = None  # Human-readable status message

    @staticmethod
    def from_dict(obj: Any) -> GetAuthStatusResponse:
        assert isinstance(obj, dict)
        isAuthenticated = obj.get("isAuthenticated")
        if isAuthenticated is None:
            raise ValueError("Missing required field 'isAuthenticated' in GetAuthStatusResponse")
        authType = obj.get("authType")
        host = obj.get("host")
        login = obj.get("login")
        statusMessage = obj.get("statusMessage")
        return GetAuthStatusResponse(
            isAuthenticated=bool(isAuthenticated),
            authType=authType,
            host=host,
            login=login,
            statusMessage=statusMessage,
        )

    def to_dict(self) -> dict:
        result: dict = {}
        result["isAuthenticated"] = self.isAuthenticated
        if self.authType is not None:
            result["authType"] = self.authType
        if self.host is not None:
            result["host"] = self.host
        if self.login is not None:
            result["login"] = self.login
        if self.statusMessage is not None:
            result["statusMessage"] = self.statusMessage
        return result


# Model capabilities
@dataclass
class ModelVisionLimits:
    """Vision-specific limits"""

    supported_media_types: list[str] | None = None
    max_prompt_images: int | None = None
    max_prompt_image_size: int | None = None

    @staticmethod
    def from_dict(obj: Any) -> ModelVisionLimits:
        assert isinstance(obj, dict)
        supported_media_types = obj.get("supported_media_types")
        max_prompt_images = obj.get("max_prompt_images")
        max_prompt_image_size = obj.get("max_prompt_image_size")
        return ModelVisionLimits(
            supported_media_types=supported_media_types,
            max_prompt_images=max_prompt_images,
            max_prompt_image_size=max_prompt_image_size,
        )

    def to_dict(self) -> dict:
        result: dict = {}
        if self.supported_media_types is not None:
            result["supported_media_types"] = self.supported_media_types
        if self.max_prompt_images is not None:
            result["max_prompt_images"] = self.max_prompt_images
        if self.max_prompt_image_size is not None:
            result["max_prompt_image_size"] = self.max_prompt_image_size
        return result


@dataclass
class ModelLimits:
    """Model limits"""

    max_prompt_tokens: int | None = None
    max_context_window_tokens: int | None = None
    vision: ModelVisionLimits | None = None

    @staticmethod
    def from_dict(obj: Any) -> ModelLimits:
        assert isinstance(obj, dict)
        max_prompt_tokens = obj.get("max_prompt_tokens")
        max_context_window_tokens = obj.get("max_context_window_tokens")
        vision_dict = obj.get("vision")
        vision = ModelVisionLimits.from_dict(vision_dict) if vision_dict else None
        return ModelLimits(
            max_prompt_tokens=max_prompt_tokens,
            max_context_window_tokens=max_context_window_tokens,
            vision=vision,
        )

    def to_dict(self) -> dict:
        result: dict = {}
        if self.max_prompt_tokens is not None:
            result["max_prompt_tokens"] = self.max_prompt_tokens
        if self.max_context_window_tokens is not None:
            result["max_context_window_tokens"] = self.max_context_window_tokens
        if self.vision is not None:
            result["vision"] = self.vision.to_dict()
        return result


@dataclass
class ModelSupports:
    """Model support flags"""

    vision: bool
    reasoning_effort: bool = False  # Whether this model supports reasoning effort

    @staticmethod
    def from_dict(obj: Any) -> ModelSupports:
        assert isinstance(obj, dict)
        vision = obj.get("vision")
        if vision is None:
            raise ValueError("Missing required field 'vision' in ModelSupports")
        reasoning_effort = obj.get("reasoningEffort", False)
        return ModelSupports(vision=bool(vision), reasoning_effort=bool(reasoning_effort))

    def to_dict(self) -> dict:
        result: dict = {}
        result["vision"] = self.vision
        result["reasoningEffort"] = self.reasoning_effort
        return result


@dataclass
class ModelCapabilities:
    """Model capabilities and limits"""

    supports: ModelSupports
    limits: ModelLimits

    @staticmethod
    def from_dict(obj: Any) -> ModelCapabilities:
        assert isinstance(obj, dict)
        supports_dict = obj.get("supports")
        limits_dict = obj.get("limits")
        if supports_dict is None or limits_dict is None:
            raise ValueError(
                f"Missing required fields in ModelCapabilities: supports={supports_dict}, "
                f"limits={limits_dict}"
            )
        supports = ModelSupports.from_dict(supports_dict)
        limits = ModelLimits.from_dict(limits_dict)
        return ModelCapabilities(supports=supports, limits=limits)

    def to_dict(self) -> dict:
        result: dict = {}
        result["supports"] = self.supports.to_dict()
        result["limits"] = self.limits.to_dict()
        return result


@dataclass
class ModelPolicy:
    """Model policy state"""

    state: str  # "enabled", "disabled", or "unconfigured"
    terms: str

    @staticmethod
    def from_dict(obj: Any) -> ModelPolicy:
        assert isinstance(obj, dict)
        state = obj.get("state")
        terms = obj.get("terms")
        if state is None or terms is None:
            raise ValueError(
                f"Missing required fields in ModelPolicy: state={state}, terms={terms}"
            )
        return ModelPolicy(state=str(state), terms=str(terms))

    def to_dict(self) -> dict:
        result: dict = {}
        result["state"] = self.state
        result["terms"] = self.terms
        return result


@dataclass
class ModelBilling:
    """Model billing information"""

    multiplier: float

    @staticmethod
    def from_dict(obj: Any) -> ModelBilling:
        assert isinstance(obj, dict)
        multiplier = obj.get("multiplier")
        if multiplier is None:
            raise ValueError("Missing required field 'multiplier' in ModelBilling")
        return ModelBilling(multiplier=float(multiplier))

    def to_dict(self) -> dict:
        result: dict = {}
        result["multiplier"] = self.multiplier
        return result


@dataclass
class ModelInfo:
    """Information about an available model"""

    id: str  # Model identifier (e.g., "claude-sonnet-4.5")
    name: str  # Display name
    capabilities: ModelCapabilities  # Model capabilities and limits
    policy: ModelPolicy | None = None  # Policy state
    billing: ModelBilling | None = None  # Billing information
    # Supported reasoning effort levels (only present if model supports reasoning effort)
    supported_reasoning_efforts: list[str] | None = None
    # Default reasoning effort level (only present if model supports reasoning effort)
    default_reasoning_effort: str | None = None

    @staticmethod
    def from_dict(obj: Any) -> ModelInfo:
        assert isinstance(obj, dict)
        id = obj.get("id")
        name = obj.get("name")
        capabilities_dict = obj.get("capabilities")
        if id is None or name is None or capabilities_dict is None:
            raise ValueError(
                f"Missing required fields in ModelInfo: id={id}, name={name}, "
                f"capabilities={capabilities_dict}"
            )
        capabilities = ModelCapabilities.from_dict(capabilities_dict)
        policy_dict = obj.get("policy")
        policy = ModelPolicy.from_dict(policy_dict) if policy_dict else None
        billing_dict = obj.get("billing")
        billing = ModelBilling.from_dict(billing_dict) if billing_dict else None
        supported_reasoning_efforts = obj.get("supportedReasoningEfforts")
        default_reasoning_effort = obj.get("defaultReasoningEffort")
        return ModelInfo(
            id=str(id),
            name=str(name),
            capabilities=capabilities,
            policy=policy,
            billing=billing,
            supported_reasoning_efforts=supported_reasoning_efforts,
            default_reasoning_effort=default_reasoning_effort,
        )

    def to_dict(self) -> dict:
        result: dict = {}
        result["id"] = self.id
        result["name"] = self.name
        result["capabilities"] = self.capabilities.to_dict()
        if self.policy is not None:
            result["policy"] = self.policy.to_dict()
        if self.billing is not None:
            result["billing"] = self.billing.to_dict()
        if self.supported_reasoning_efforts is not None:
            result["supportedReasoningEfforts"] = self.supported_reasoning_efforts
        if self.default_reasoning_effort is not None:
            result["defaultReasoningEffort"] = self.default_reasoning_effort
        return result


@dataclass
class SessionContext:
    """Working directory context for a session"""

    cwd: str  # Working directory where the session was created
    gitRoot: str | None = None  # Git repository root (if in a git repo)
    repository: str | None = None  # GitHub repository in "owner/repo" format
    branch: str | None = None  # Current git branch

    @staticmethod
    def from_dict(obj: Any) -> SessionContext:
        assert isinstance(obj, dict)
        cwd = obj.get("cwd")
        if cwd is None:
            raise ValueError("Missing required field 'cwd' in SessionContext")
        return SessionContext(
            cwd=str(cwd),
            gitRoot=obj.get("gitRoot"),
            repository=obj.get("repository"),
            branch=obj.get("branch"),
        )

    def to_dict(self) -> dict:
        result: dict = {"cwd": self.cwd}
        if self.gitRoot is not None:
            result["gitRoot"] = self.gitRoot
        if self.repository is not None:
            result["repository"] = self.repository
        if self.branch is not None:
            result["branch"] = self.branch
        return result


@dataclass
class SessionListFilter:
    """Filter options for listing sessions"""

    cwd: str | None = None  # Filter by exact cwd match
    gitRoot: str | None = None  # Filter by git root
    repository: str | None = None  # Filter by repository (owner/repo format)
    branch: str | None = None  # Filter by branch

    def to_dict(self) -> dict:
        result: dict = {}
        if self.cwd is not None:
            result["cwd"] = self.cwd
        if self.gitRoot is not None:
            result["gitRoot"] = self.gitRoot
        if self.repository is not None:
            result["repository"] = self.repository
        if self.branch is not None:
            result["branch"] = self.branch
        return result


@dataclass
class SessionMetadata:
    """Metadata about a session"""

    sessionId: str  # Session identifier
    startTime: str  # ISO 8601 timestamp when session was created
    modifiedTime: str  # ISO 8601 timestamp when session was last modified
    isRemote: bool  # Whether the session is remote
    summary: str | None = None  # Optional summary of the session
    context: SessionContext | None = None  # Working directory context

    @staticmethod
    def from_dict(obj: Any) -> SessionMetadata:
        assert isinstance(obj, dict)
        sessionId = obj.get("sessionId")
        startTime = obj.get("startTime")
        modifiedTime = obj.get("modifiedTime")
        isRemote = obj.get("isRemote")
        if sessionId is None or startTime is None or modifiedTime is None or isRemote is None:
            raise ValueError(
                f"Missing required fields in SessionMetadata: sessionId={sessionId}, "
                f"startTime={startTime}, modifiedTime={modifiedTime}, isRemote={isRemote}"
            )
        summary = obj.get("summary")
        context_dict = obj.get("context")
        context = SessionContext.from_dict(context_dict) if context_dict else None
        return SessionMetadata(
            sessionId=str(sessionId),
            startTime=str(startTime),
            modifiedTime=str(modifiedTime),
            isRemote=bool(isRemote),
            summary=summary,
            context=context,
        )

    def to_dict(self) -> dict:
        result: dict = {}
        result["sessionId"] = self.sessionId
        result["startTime"] = self.startTime
        result["modifiedTime"] = self.modifiedTime
        result["isRemote"] = self.isRemote
        if self.summary is not None:
            result["summary"] = self.summary
        if self.context is not None:
            result["context"] = self.context.to_dict()
        return result


# Session Lifecycle Types (for TUI+server mode)

SessionLifecycleEventType = Literal[
    "session.created",
    "session.deleted",
    "session.updated",
    "session.foreground",
    "session.background",
]


@dataclass
class SessionLifecycleEventMetadata:
    """Metadata for session lifecycle events."""

    startTime: str
    modifiedTime: str
    summary: str | None = None

    @staticmethod
    def from_dict(data: dict) -> SessionLifecycleEventMetadata:
        return SessionLifecycleEventMetadata(
            startTime=data.get("startTime", ""),
            modifiedTime=data.get("modifiedTime", ""),
            summary=data.get("summary"),
        )


@dataclass
class SessionLifecycleEvent:
    """Session lifecycle event notification."""

    type: SessionLifecycleEventType
    sessionId: str
    metadata: SessionLifecycleEventMetadata | None = None

    @staticmethod
    def from_dict(data: dict) -> SessionLifecycleEvent:
        metadata = None
        if "metadata" in data and data["metadata"]:
            metadata = SessionLifecycleEventMetadata.from_dict(data["metadata"])
        return SessionLifecycleEvent(
            type=data.get("type", "session.updated"),
            sessionId=data.get("sessionId", ""),
            metadata=metadata,
        )


# Handler types for session lifecycle events
SessionLifecycleHandler = Callable[[SessionLifecycleEvent], None]