""" AUTO-GENERATED FILE - DO NOT EDIT Generated from: copilot-core/session-events.schema.json Generated by: scripts/generate-session-types.ts Generated at: 2026-02-06T20:38:23.376Z To update these types: 1. Update the schema in copilot-agent-runtime 2. Run: npm run generate:session-types """ from dataclasses import dataclass from typing import Any, Optional, List, Dict, Union, TypeVar, Type, cast, Callable from enum import Enum from datetime import datetime from uuid import UUID import dateutil.parser T = TypeVar("T") EnumT = TypeVar("EnumT", bound=Enum) def from_float(x: Any) -> float: assert isinstance(x, (float, int)) and not isinstance(x, bool) return float(x) def to_float(x: Any) -> float: assert isinstance(x, (int, float)) return x def to_class(c: Type[T], x: Any) -> dict: assert isinstance(x, c) return cast(Any, x).to_dict() def from_str(x: Any) -> str: assert isinstance(x, str) return x def from_none(x: Any) -> Any: assert x is None return x def from_union(fs, x): for f in fs: try: return f(x) except: pass assert False def to_enum(c: Type[EnumT], x: Any) -> EnumT: assert isinstance(x, c) return x.value def from_list(f: Callable[[Any], T], x: Any) -> List[T]: assert isinstance(x, list) return [f(y) for y in x] def from_dict(f: Callable[[Any], T], x: Any) -> Dict[str, T]: assert isinstance(x, dict) return { k: f(v) for (k, v) in x.items() } def from_bool(x: Any) -> bool: assert isinstance(x, bool) return x def from_datetime(x: Any) -> datetime: return dateutil.parser.parse(x) def from_int(x: Any) -> int: assert isinstance(x, int) and not isinstance(x, bool) return x @dataclass class End: character: float line: float @staticmethod def from_dict(obj: Any) -> 'End': assert isinstance(obj, dict) character = from_float(obj.get("character")) line = from_float(obj.get("line")) return End(character, line) def to_dict(self) -> dict: result: dict = {} result["character"] = to_float(self.character) result["line"] = to_float(self.line) return result @dataclass class Start: character: float line: float @staticmethod def from_dict(obj: Any) -> 'Start': assert isinstance(obj, dict) character = from_float(obj.get("character")) line = from_float(obj.get("line")) return Start(character, line) def to_dict(self) -> dict: result: dict = {} result["character"] = to_float(self.character) result["line"] = to_float(self.line) return result @dataclass class Selection: end: End start: Start @staticmethod def from_dict(obj: Any) -> 'Selection': assert isinstance(obj, dict) end = End.from_dict(obj.get("end")) start = Start.from_dict(obj.get("start")) return Selection(end, start) def to_dict(self) -> dict: result: dict = {} result["end"] = to_class(End, self.end) result["start"] = to_class(Start, self.start) return result class AttachmentType(Enum): DIRECTORY = "directory" FILE = "file" SELECTION = "selection" @dataclass class Attachment: display_name: str type: AttachmentType path: Optional[str] = None file_path: Optional[str] = None selection: Optional[Selection] = None text: Optional[str] = None @staticmethod def from_dict(obj: Any) -> 'Attachment': assert isinstance(obj, dict) display_name = from_str(obj.get("displayName")) type = AttachmentType(obj.get("type")) path = from_union([from_str, from_none], obj.get("path")) file_path = from_union([from_str, from_none], obj.get("filePath")) selection = from_union([Selection.from_dict, from_none], obj.get("selection")) text = from_union([from_str, from_none], obj.get("text")) return Attachment(display_name, type, path, file_path, selection, text) def to_dict(self) -> dict: result: dict = {} result["displayName"] = from_str(self.display_name) result["type"] = to_enum(AttachmentType, self.type) if self.path is not None: result["path"] = from_union([from_str, from_none], self.path) if self.file_path is not None: result["filePath"] = from_union([from_str, from_none], self.file_path) if self.selection is not None: result["selection"] = from_union([lambda x: to_class(Selection, x), from_none], self.selection) if self.text is not None: result["text"] = from_union([from_str, from_none], self.text) return result @dataclass class CodeChanges: files_modified: List[str] lines_added: float lines_removed: float @staticmethod def from_dict(obj: Any) -> 'CodeChanges': assert isinstance(obj, dict) files_modified = from_list(from_str, obj.get("filesModified")) lines_added = from_float(obj.get("linesAdded")) lines_removed = from_float(obj.get("linesRemoved")) return CodeChanges(files_modified, lines_added, lines_removed) def to_dict(self) -> dict: result: dict = {} result["filesModified"] = from_list(from_str, self.files_modified) result["linesAdded"] = to_float(self.lines_added) result["linesRemoved"] = to_float(self.lines_removed) return result @dataclass class CompactionTokensUsed: cached_input: float input: float output: float @staticmethod def from_dict(obj: Any) -> 'CompactionTokensUsed': assert isinstance(obj, dict) cached_input = from_float(obj.get("cachedInput")) input = from_float(obj.get("input")) output = from_float(obj.get("output")) return CompactionTokensUsed(cached_input, input, output) def to_dict(self) -> dict: result: dict = {} result["cachedInput"] = to_float(self.cached_input) result["input"] = to_float(self.input) result["output"] = to_float(self.output) return result @dataclass class ContextClass: cwd: str branch: Optional[str] = None git_root: Optional[str] = None repository: Optional[str] = None @staticmethod def from_dict(obj: Any) -> 'ContextClass': assert isinstance(obj, dict) cwd = from_str(obj.get("cwd")) branch = from_union([from_str, from_none], obj.get("branch")) git_root = from_union([from_str, from_none], obj.get("gitRoot")) repository = from_union([from_str, from_none], obj.get("repository")) return ContextClass(cwd, branch, git_root, repository) def to_dict(self) -> dict: result: dict = {} result["cwd"] = from_str(self.cwd) if self.branch is not None: result["branch"] = from_union([from_str, from_none], self.branch) if self.git_root is not None: result["gitRoot"] = from_union([from_str, from_none], self.git_root) if self.repository is not None: result["repository"] = from_union([from_str, from_none], self.repository) return result @dataclass class ErrorClass: message: str code: Optional[str] = None stack: Optional[str] = None @staticmethod def from_dict(obj: Any) -> 'ErrorClass': assert isinstance(obj, dict) message = from_str(obj.get("message")) code = from_union([from_str, from_none], obj.get("code")) stack = from_union([from_str, from_none], obj.get("stack")) return ErrorClass(message, code, stack) def to_dict(self) -> dict: result: dict = {} result["message"] = from_str(self.message) if self.code is not None: result["code"] = from_union([from_str, from_none], self.code) if self.stack is not None: result["stack"] = from_union([from_str, from_none], self.stack) return result @dataclass class Metadata: prompt_version: Optional[str] = None variables: Optional[Dict[str, Any]] = None @staticmethod def from_dict(obj: Any) -> 'Metadata': assert isinstance(obj, dict) prompt_version = from_union([from_str, from_none], obj.get("promptVersion")) variables = from_union([lambda x: from_dict(lambda x: x, x), from_none], obj.get("variables")) return Metadata(prompt_version, variables) def to_dict(self) -> dict: result: dict = {} if self.prompt_version is not None: result["promptVersion"] = from_union([from_str, from_none], self.prompt_version) if self.variables is not None: result["variables"] = from_union([lambda x: from_dict(lambda x: x, x), from_none], self.variables) return result @dataclass class Requests: cost: float count: float @staticmethod def from_dict(obj: Any) -> 'Requests': assert isinstance(obj, dict) cost = from_float(obj.get("cost")) count = from_float(obj.get("count")) return Requests(cost, count) def to_dict(self) -> dict: result: dict = {} result["cost"] = to_float(self.cost) result["count"] = to_float(self.count) return result @dataclass class Usage: cache_read_tokens: float cache_write_tokens: float input_tokens: float output_tokens: float @staticmethod def from_dict(obj: Any) -> 'Usage': assert isinstance(obj, dict) cache_read_tokens = from_float(obj.get("cacheReadTokens")) cache_write_tokens = from_float(obj.get("cacheWriteTokens")) input_tokens = from_float(obj.get("inputTokens")) output_tokens = from_float(obj.get("outputTokens")) return Usage(cache_read_tokens, cache_write_tokens, input_tokens, output_tokens) def to_dict(self) -> dict: result: dict = {} result["cacheReadTokens"] = to_float(self.cache_read_tokens) result["cacheWriteTokens"] = to_float(self.cache_write_tokens) result["inputTokens"] = to_float(self.input_tokens) result["outputTokens"] = to_float(self.output_tokens) return result @dataclass class ModelMetric: requests: Requests usage: Usage @staticmethod def from_dict(obj: Any) -> 'ModelMetric': assert isinstance(obj, dict) requests = Requests.from_dict(obj.get("requests")) usage = Usage.from_dict(obj.get("usage")) return ModelMetric(requests, usage) def to_dict(self) -> dict: result: dict = {} result["requests"] = to_class(Requests, self.requests) result["usage"] = to_class(Usage, self.usage) return result @dataclass class QuotaSnapshot: entitlement_requests: float is_unlimited_entitlement: bool overage: float overage_allowed_with_exhausted_quota: bool remaining_percentage: float usage_allowed_with_exhausted_quota: bool used_requests: float reset_date: Optional[datetime] = None @staticmethod def from_dict(obj: Any) -> 'QuotaSnapshot': assert isinstance(obj, dict) entitlement_requests = from_float(obj.get("entitlementRequests")) is_unlimited_entitlement = from_bool(obj.get("isUnlimitedEntitlement")) overage = from_float(obj.get("overage")) overage_allowed_with_exhausted_quota = from_bool(obj.get("overageAllowedWithExhaustedQuota")) remaining_percentage = from_float(obj.get("remainingPercentage")) usage_allowed_with_exhausted_quota = from_bool(obj.get("usageAllowedWithExhaustedQuota")) used_requests = from_float(obj.get("usedRequests")) reset_date = from_union([from_datetime, from_none], obj.get("resetDate")) return QuotaSnapshot(entitlement_requests, is_unlimited_entitlement, overage, overage_allowed_with_exhausted_quota, remaining_percentage, usage_allowed_with_exhausted_quota, used_requests, reset_date) def to_dict(self) -> dict: result: dict = {} result["entitlementRequests"] = to_float(self.entitlement_requests) result["isUnlimitedEntitlement"] = from_bool(self.is_unlimited_entitlement) result["overage"] = to_float(self.overage) result["overageAllowedWithExhaustedQuota"] = from_bool(self.overage_allowed_with_exhausted_quota) result["remainingPercentage"] = to_float(self.remaining_percentage) result["usageAllowedWithExhaustedQuota"] = from_bool(self.usage_allowed_with_exhausted_quota) result["usedRequests"] = to_float(self.used_requests) if self.reset_date is not None: result["resetDate"] = from_union([lambda x: x.isoformat(), from_none], self.reset_date) return result @dataclass class Repository: name: str owner: str branch: Optional[str] = None @staticmethod def from_dict(obj: Any) -> 'Repository': assert isinstance(obj, dict) name = from_str(obj.get("name")) owner = from_str(obj.get("owner")) branch = from_union([from_str, from_none], obj.get("branch")) return Repository(name, owner, branch) def to_dict(self) -> dict: result: dict = {} result["name"] = from_str(self.name) result["owner"] = from_str(self.owner) if self.branch is not None: result["branch"] = from_union([from_str, from_none], self.branch) return result @dataclass class Result: content: str detailed_content: Optional[str] = None @staticmethod def from_dict(obj: Any) -> 'Result': assert isinstance(obj, dict) content = from_str(obj.get("content")) detailed_content = from_union([from_str, from_none], obj.get("detailedContent")) return Result(content, detailed_content) def to_dict(self) -> dict: result: dict = {} result["content"] = from_str(self.content) if self.detailed_content is not None: result["detailedContent"] = from_union([from_str, from_none], self.detailed_content) return result class Role(Enum): DEVELOPER = "developer" SYSTEM = "system" class ShutdownType(Enum): ERROR = "error" ROUTINE = "routine" class SourceType(Enum): LOCAL = "local" REMOTE = "remote" class ToolRequestType(Enum): CUSTOM = "custom" FUNCTION = "function" @dataclass class ToolRequest: name: str tool_call_id: str arguments: Any = None type: Optional[ToolRequestType] = None @staticmethod def from_dict(obj: Any) -> 'ToolRequest': assert isinstance(obj, dict) name = from_str(obj.get("name")) tool_call_id = from_str(obj.get("toolCallId")) arguments = obj.get("arguments") type = from_union([ToolRequestType, from_none], obj.get("type")) return ToolRequest(name, tool_call_id, arguments, type) def to_dict(self) -> dict: result: dict = {} result["name"] = from_str(self.name) result["toolCallId"] = from_str(self.tool_call_id) if self.arguments is not None: result["arguments"] = self.arguments if self.type is not None: result["type"] = from_union([lambda x: to_enum(ToolRequestType, x), from_none], self.type) return result @dataclass class Data: context: Optional[Union[ContextClass, str]] = None copilot_version: Optional[str] = None producer: Optional[str] = None selected_model: Optional[str] = None session_id: Optional[str] = None start_time: Optional[datetime] = None version: Optional[float] = None event_count: Optional[float] = None resume_time: Optional[datetime] = None error_type: Optional[str] = None message: Optional[str] = None provider_call_id: Optional[str] = None stack: Optional[str] = None status_code: Optional[int] = None info_type: Optional[str] = None new_model: Optional[str] = None previous_model: Optional[str] = None handoff_time: Optional[datetime] = None remote_session_id: Optional[str] = None repository: Optional[Repository] = None source_type: Optional[SourceType] = None summary: Optional[str] = None messages_removed_during_truncation: Optional[float] = None performed_by: Optional[str] = None post_truncation_messages_length: Optional[float] = None post_truncation_tokens_in_messages: Optional[float] = None pre_truncation_messages_length: Optional[float] = None pre_truncation_tokens_in_messages: Optional[float] = None token_limit: Optional[float] = None tokens_removed_during_truncation: Optional[float] = None events_removed: Optional[float] = None up_to_event_id: Optional[str] = None code_changes: Optional[CodeChanges] = None current_model: Optional[str] = None error_reason: Optional[str] = None model_metrics: Optional[Dict[str, ModelMetric]] = None session_start_time: Optional[float] = None shutdown_type: Optional[ShutdownType] = None total_api_duration_ms: Optional[float] = None total_premium_requests: Optional[float] = None current_tokens: Optional[float] = None messages_length: Optional[float] = None checkpoint_number: Optional[float] = None checkpoint_path: Optional[str] = None compaction_tokens_used: Optional[CompactionTokensUsed] = None error: Optional[Union[ErrorClass, str]] = None messages_removed: Optional[float] = None post_compaction_tokens: Optional[float] = None pre_compaction_messages_length: Optional[float] = None pre_compaction_tokens: Optional[float] = None request_id: Optional[str] = None success: Optional[bool] = None summary_content: Optional[str] = None tokens_removed: Optional[float] = None attachments: Optional[List[Attachment]] = None content: Optional[str] = None source: Optional[str] = None transformed_content: Optional[str] = None turn_id: Optional[str] = None intent: Optional[str] = None reasoning_id: Optional[str] = None delta_content: Optional[str] = None encrypted_content: Optional[str] = None message_id: Optional[str] = None parent_tool_call_id: Optional[str] = None reasoning_opaque: Optional[str] = None reasoning_text: Optional[str] = None tool_requests: Optional[List[ToolRequest]] = None total_response_size_bytes: Optional[float] = None api_call_id: Optional[str] = None cache_read_tokens: Optional[float] = None cache_write_tokens: Optional[float] = None cost: Optional[float] = None duration: Optional[float] = None initiator: Optional[str] = None input_tokens: Optional[float] = None model: Optional[str] = None output_tokens: Optional[float] = None quota_snapshots: Optional[Dict[str, QuotaSnapshot]] = None reason: Optional[str] = None arguments: Any = None tool_call_id: Optional[str] = None tool_name: Optional[str] = None mcp_server_name: Optional[str] = None mcp_tool_name: Optional[str] = None partial_output: Optional[str] = None progress_message: Optional[str] = None is_user_requested: Optional[bool] = None result: Optional[Result] = None tool_telemetry: Optional[Dict[str, Any]] = None allowed_tools: Optional[List[str]] = None name: Optional[str] = None path: Optional[str] = None agent_description: Optional[str] = None agent_display_name: Optional[str] = None agent_name: Optional[str] = None tools: Optional[List[str]] = None hook_invocation_id: Optional[str] = None hook_type: Optional[str] = None input: Any = None output: Any = None metadata: Optional[Metadata] = None role: Optional[Role] = None @staticmethod def from_dict(obj: Any) -> 'Data': assert isinstance(obj, dict) context = from_union([ContextClass.from_dict, from_str, from_none], obj.get("context")) copilot_version = from_union([from_str, from_none], obj.get("copilotVersion")) producer = from_union([from_str, from_none], obj.get("producer")) selected_model = from_union([from_str, from_none], obj.get("selectedModel")) session_id = from_union([from_str, from_none], obj.get("sessionId")) start_time = from_union([from_datetime, from_none], obj.get("startTime")) version = from_union([from_float, from_none], obj.get("version")) event_count = from_union([from_float, from_none], obj.get("eventCount")) resume_time = from_union([from_datetime, from_none], obj.get("resumeTime")) error_type = from_union([from_str, from_none], obj.get("errorType")) message = from_union([from_str, from_none], obj.get("message")) provider_call_id = from_union([from_str, from_none], obj.get("providerCallId")) stack = from_union([from_str, from_none], obj.get("stack")) status_code = from_union([from_int, from_none], obj.get("statusCode")) info_type = from_union([from_str, from_none], obj.get("infoType")) new_model = from_union([from_str, from_none], obj.get("newModel")) previous_model = from_union([from_str, from_none], obj.get("previousModel")) handoff_time = from_union([from_datetime, from_none], obj.get("handoffTime")) remote_session_id = from_union([from_str, from_none], obj.get("remoteSessionId")) repository = from_union([Repository.from_dict, from_none], obj.get("repository")) source_type = from_union([SourceType, from_none], obj.get("sourceType")) summary = from_union([from_str, from_none], obj.get("summary")) messages_removed_during_truncation = from_union([from_float, from_none], obj.get("messagesRemovedDuringTruncation")) performed_by = from_union([from_str, from_none], obj.get("performedBy")) post_truncation_messages_length = from_union([from_float, from_none], obj.get("postTruncationMessagesLength")) post_truncation_tokens_in_messages = from_union([from_float, from_none], obj.get("postTruncationTokensInMessages")) pre_truncation_messages_length = from_union([from_float, from_none], obj.get("preTruncationMessagesLength")) pre_truncation_tokens_in_messages = from_union([from_float, from_none], obj.get("preTruncationTokensInMessages")) token_limit = from_union([from_float, from_none], obj.get("tokenLimit")) tokens_removed_during_truncation = from_union([from_float, from_none], obj.get("tokensRemovedDuringTruncation")) events_removed = from_union([from_float, from_none], obj.get("eventsRemoved")) up_to_event_id = from_union([from_str, from_none], obj.get("upToEventId")) code_changes = from_union([CodeChanges.from_dict, from_none], obj.get("codeChanges")) current_model = from_union([from_str, from_none], obj.get("currentModel")) error_reason = from_union([from_str, from_none], obj.get("errorReason")) model_metrics = from_union([lambda x: from_dict(ModelMetric.from_dict, x), from_none], obj.get("modelMetrics")) session_start_time = from_union([from_float, from_none], obj.get("sessionStartTime")) shutdown_type = from_union([ShutdownType, from_none], obj.get("shutdownType")) total_api_duration_ms = from_union([from_float, from_none], obj.get("totalApiDurationMs")) total_premium_requests = from_union([from_float, from_none], obj.get("totalPremiumRequests")) current_tokens = from_union([from_float, from_none], obj.get("currentTokens")) messages_length = from_union([from_float, from_none], obj.get("messagesLength")) checkpoint_number = from_union([from_float, from_none], obj.get("checkpointNumber")) checkpoint_path = from_union([from_str, from_none], obj.get("checkpointPath")) compaction_tokens_used = from_union([CompactionTokensUsed.from_dict, from_none], obj.get("compactionTokensUsed")) error = from_union([ErrorClass.from_dict, from_str, from_none], obj.get("error")) messages_removed = from_union([from_float, from_none], obj.get("messagesRemoved")) post_compaction_tokens = from_union([from_float, from_none], obj.get("postCompactionTokens")) pre_compaction_messages_length = from_union([from_float, from_none], obj.get("preCompactionMessagesLength")) pre_compaction_tokens = from_union([from_float, from_none], obj.get("preCompactionTokens")) request_id = from_union([from_str, from_none], obj.get("requestId")) success = from_union([from_bool, from_none], obj.get("success")) summary_content = from_union([from_str, from_none], obj.get("summaryContent")) tokens_removed = from_union([from_float, from_none], obj.get("tokensRemoved")) attachments = from_union([lambda x: from_list(Attachment.from_dict, x), from_none], obj.get("attachments")) content = from_union([from_str, from_none], obj.get("content")) source = from_union([from_str, from_none], obj.get("source")) transformed_content = from_union([from_str, from_none], obj.get("transformedContent")) turn_id = from_union([from_str, from_none], obj.get("turnId")) intent = from_union([from_str, from_none], obj.get("intent")) reasoning_id = from_union([from_str, from_none], obj.get("reasoningId")) delta_content = from_union([from_str, from_none], obj.get("deltaContent")) encrypted_content = from_union([from_str, from_none], obj.get("encryptedContent")) message_id = from_union([from_str, from_none], obj.get("messageId")) parent_tool_call_id = from_union([from_str, from_none], obj.get("parentToolCallId")) reasoning_opaque = from_union([from_str, from_none], obj.get("reasoningOpaque")) reasoning_text = from_union([from_str, from_none], obj.get("reasoningText")) tool_requests = from_union([lambda x: from_list(ToolRequest.from_dict, x), from_none], obj.get("toolRequests")) total_response_size_bytes = from_union([from_float, from_none], obj.get("totalResponseSizeBytes")) api_call_id = from_union([from_str, from_none], obj.get("apiCallId")) cache_read_tokens = from_union([from_float, from_none], obj.get("cacheReadTokens")) cache_write_tokens = from_union([from_float, from_none], obj.get("cacheWriteTokens")) cost = from_union([from_float, from_none], obj.get("cost")) duration = from_union([from_float, from_none], obj.get("duration")) initiator = from_union([from_str, from_none], obj.get("initiator")) input_tokens = from_union([from_float, from_none], obj.get("inputTokens")) model = from_union([from_str, from_none], obj.get("model")) output_tokens = from_union([from_float, from_none], obj.get("outputTokens")) quota_snapshots = from_union([lambda x: from_dict(QuotaSnapshot.from_dict, x), from_none], obj.get("quotaSnapshots")) reason = from_union([from_str, from_none], obj.get("reason")) arguments = obj.get("arguments") tool_call_id = from_union([from_str, from_none], obj.get("toolCallId")) tool_name = from_union([from_str, from_none], obj.get("toolName")) mcp_server_name = from_union([from_str, from_none], obj.get("mcpServerName")) mcp_tool_name = from_union([from_str, from_none], obj.get("mcpToolName")) partial_output = from_union([from_str, from_none], obj.get("partialOutput")) progress_message = from_union([from_str, from_none], obj.get("progressMessage")) is_user_requested = from_union([from_bool, from_none], obj.get("isUserRequested")) result = from_union([Result.from_dict, from_none], obj.get("result")) tool_telemetry = from_union([lambda x: from_dict(lambda x: x, x), from_none], obj.get("toolTelemetry")) allowed_tools = from_union([lambda x: from_list(from_str, x), from_none], obj.get("allowedTools")) name = from_union([from_str, from_none], obj.get("name")) path = from_union([from_str, from_none], obj.get("path")) agent_description = from_union([from_str, from_none], obj.get("agentDescription")) agent_display_name = from_union([from_str, from_none], obj.get("agentDisplayName")) agent_name = from_union([from_str, from_none], obj.get("agentName")) tools = from_union([lambda x: from_list(from_str, x), from_none], obj.get("tools")) hook_invocation_id = from_union([from_str, from_none], obj.get("hookInvocationId")) hook_type = from_union([from_str, from_none], obj.get("hookType")) input = obj.get("input") output = obj.get("output") metadata = from_union([Metadata.from_dict, from_none], obj.get("metadata")) role = from_union([Role, from_none], obj.get("role")) return Data(context, copilot_version, producer, selected_model, session_id, start_time, version, event_count, resume_time, error_type, message, provider_call_id, stack, status_code, info_type, new_model, previous_model, handoff_time, remote_session_id, repository, source_type, summary, messages_removed_during_truncation, performed_by, post_truncation_messages_length, post_truncation_tokens_in_messages, pre_truncation_messages_length, pre_truncation_tokens_in_messages, token_limit, tokens_removed_during_truncation, events_removed, up_to_event_id, code_changes, current_model, error_reason, model_metrics, session_start_time, shutdown_type, total_api_duration_ms, total_premium_requests, current_tokens, messages_length, checkpoint_number, checkpoint_path, compaction_tokens_used, error, messages_removed, post_compaction_tokens, pre_compaction_messages_length, pre_compaction_tokens, request_id, success, summary_content, tokens_removed, attachments, content, source, transformed_content, turn_id, intent, reasoning_id, delta_content, encrypted_content, message_id, parent_tool_call_id, reasoning_opaque, reasoning_text, tool_requests, total_response_size_bytes, api_call_id, cache_read_tokens, cache_write_tokens, cost, duration, initiator, input_tokens, model, output_tokens, quota_snapshots, reason, arguments, tool_call_id, tool_name, mcp_server_name, mcp_tool_name, partial_output, progress_message, is_user_requested, result, tool_telemetry, allowed_tools, name, path, agent_description, agent_display_name, agent_name, tools, hook_invocation_id, hook_type, input, output, metadata, role) def to_dict(self) -> dict: result: dict = {} if self.context is not None: result["context"] = from_union([lambda x: to_class(ContextClass, x), from_str, from_none], self.context) if self.copilot_version is not None: result["copilotVersion"] = from_union([from_str, from_none], self.copilot_version) if self.producer is not None: result["producer"] = from_union([from_str, from_none], self.producer) if self.selected_model is not None: result["selectedModel"] = from_union([from_str, from_none], self.selected_model) if self.session_id is not None: result["sessionId"] = from_union([from_str, from_none], self.session_id) if self.start_time is not None: result["startTime"] = from_union([lambda x: x.isoformat(), from_none], self.start_time) if self.version is not None: result["version"] = from_union([to_float, from_none], self.version) if self.event_count is not None: result["eventCount"] = from_union([to_float, from_none], self.event_count) if self.resume_time is not None: result["resumeTime"] = from_union([lambda x: x.isoformat(), from_none], self.resume_time) if self.error_type is not None: result["errorType"] = from_union([from_str, from_none], self.error_type) if self.message is not None: result["message"] = from_union([from_str, from_none], self.message) if self.provider_call_id is not None: result["providerCallId"] = from_union([from_str, from_none], self.provider_call_id) if self.stack is not None: result["stack"] = from_union([from_str, from_none], self.stack) if self.status_code is not None: result["statusCode"] = from_union([from_int, from_none], self.status_code) if self.info_type is not None: result["infoType"] = from_union([from_str, from_none], self.info_type) if self.new_model is not None: result["newModel"] = from_union([from_str, from_none], self.new_model) if self.previous_model is not None: result["previousModel"] = from_union([from_str, from_none], self.previous_model) if self.handoff_time is not None: result["handoffTime"] = from_union([lambda x: x.isoformat(), from_none], self.handoff_time) if self.remote_session_id is not None: result["remoteSessionId"] = from_union([from_str, from_none], self.remote_session_id) if self.repository is not None: result["repository"] = from_union([lambda x: to_class(Repository, x), from_none], self.repository) if self.source_type is not None: result["sourceType"] = from_union([lambda x: to_enum(SourceType, x), from_none], self.source_type) if self.summary is not None: result["summary"] = from_union([from_str, from_none], self.summary) if self.messages_removed_during_truncation is not None: result["messagesRemovedDuringTruncation"] = from_union([to_float, from_none], self.messages_removed_during_truncation) if self.performed_by is not None: result["performedBy"] = from_union([from_str, from_none], self.performed_by) if self.post_truncation_messages_length is not None: result["postTruncationMessagesLength"] = from_union([to_float, from_none], self.post_truncation_messages_length) if self.post_truncation_tokens_in_messages is not None: result["postTruncationTokensInMessages"] = from_union([to_float, from_none], self.post_truncation_tokens_in_messages) if self.pre_truncation_messages_length is not None: result["preTruncationMessagesLength"] = from_union([to_float, from_none], self.pre_truncation_messages_length) if self.pre_truncation_tokens_in_messages is not None: result["preTruncationTokensInMessages"] = from_union([to_float, from_none], self.pre_truncation_tokens_in_messages) if self.token_limit is not None: result["tokenLimit"] = from_union([to_float, from_none], self.token_limit) if self.tokens_removed_during_truncation is not None: result["tokensRemovedDuringTruncation"] = from_union([to_float, from_none], self.tokens_removed_during_truncation) if self.events_removed is not None: result["eventsRemoved"] = from_union([to_float, from_none], self.events_removed) if self.up_to_event_id is not None: result["upToEventId"] = from_union([from_str, from_none], self.up_to_event_id) if self.code_changes is not None: result["codeChanges"] = from_union([lambda x: to_class(CodeChanges, x), from_none], self.code_changes) if self.current_model is not None: result["currentModel"] = from_union([from_str, from_none], self.current_model) if self.error_reason is not None: result["errorReason"] = from_union([from_str, from_none], self.error_reason) if self.model_metrics is not None: result["modelMetrics"] = from_union([lambda x: from_dict(lambda x: to_class(ModelMetric, x), x), from_none], self.model_metrics) if self.session_start_time is not None: result["sessionStartTime"] = from_union([to_float, from_none], self.session_start_time) if self.shutdown_type is not None: result["shutdownType"] = from_union([lambda x: to_enum(ShutdownType, x), from_none], self.shutdown_type) if self.total_api_duration_ms is not None: result["totalApiDurationMs"] = from_union([to_float, from_none], self.total_api_duration_ms) if self.total_premium_requests is not None: result["totalPremiumRequests"] = from_union([to_float, from_none], self.total_premium_requests) if self.current_tokens is not None: result["currentTokens"] = from_union([to_float, from_none], self.current_tokens) if self.messages_length is not None: result["messagesLength"] = from_union([to_float, from_none], self.messages_length) if self.checkpoint_number is not None: result["checkpointNumber"] = from_union([to_float, from_none], self.checkpoint_number) if self.checkpoint_path is not None: result["checkpointPath"] = from_union([from_str, from_none], self.checkpoint_path) if self.compaction_tokens_used is not None: result["compactionTokensUsed"] = from_union([lambda x: to_class(CompactionTokensUsed, x), from_none], self.compaction_tokens_used) if self.error is not None: result["error"] = from_union([lambda x: to_class(ErrorClass, x), from_str, from_none], self.error) if self.messages_removed is not None: result["messagesRemoved"] = from_union([to_float, from_none], self.messages_removed) if self.post_compaction_tokens is not None: result["postCompactionTokens"] = from_union([to_float, from_none], self.post_compaction_tokens) if self.pre_compaction_messages_length is not None: result["preCompactionMessagesLength"] = from_union([to_float, from_none], self.pre_compaction_messages_length) if self.pre_compaction_tokens is not None: result["preCompactionTokens"] = from_union([to_float, from_none], self.pre_compaction_tokens) if self.request_id is not None: result["requestId"] = from_union([from_str, from_none], self.request_id) if self.success is not None: result["success"] = from_union([from_bool, from_none], self.success) if self.summary_content is not None: result["summaryContent"] = from_union([from_str, from_none], self.summary_content) if self.tokens_removed is not None: result["tokensRemoved"] = from_union([to_float, from_none], self.tokens_removed) if self.attachments is not None: result["attachments"] = from_union([lambda x: from_list(lambda x: to_class(Attachment, x), x), from_none], self.attachments) if self.content is not None: result["content"] = from_union([from_str, from_none], self.content) if self.source is not None: result["source"] = from_union([from_str, from_none], self.source) if self.transformed_content is not None: result["transformedContent"] = from_union([from_str, from_none], self.transformed_content) if self.turn_id is not None: result["turnId"] = from_union([from_str, from_none], self.turn_id) if self.intent is not None: result["intent"] = from_union([from_str, from_none], self.intent) if self.reasoning_id is not None: result["reasoningId"] = from_union([from_str, from_none], self.reasoning_id) if self.delta_content is not None: result["deltaContent"] = from_union([from_str, from_none], self.delta_content) if self.encrypted_content is not None: result["encryptedContent"] = from_union([from_str, from_none], self.encrypted_content) if self.message_id is not None: result["messageId"] = from_union([from_str, from_none], self.message_id) if self.parent_tool_call_id is not None: result["parentToolCallId"] = from_union([from_str, from_none], self.parent_tool_call_id) if self.reasoning_opaque is not None: result["reasoningOpaque"] = from_union([from_str, from_none], self.reasoning_opaque) if self.reasoning_text is not None: result["reasoningText"] = from_union([from_str, from_none], self.reasoning_text) if self.tool_requests is not None: result["toolRequests"] = from_union([lambda x: from_list(lambda x: to_class(ToolRequest, x), x), from_none], self.tool_requests) if self.total_response_size_bytes is not None: result["totalResponseSizeBytes"] = from_union([to_float, from_none], self.total_response_size_bytes) if self.api_call_id is not None: result["apiCallId"] = from_union([from_str, from_none], self.api_call_id) if self.cache_read_tokens is not None: result["cacheReadTokens"] = from_union([to_float, from_none], self.cache_read_tokens) if self.cache_write_tokens is not None: result["cacheWriteTokens"] = from_union([to_float, from_none], self.cache_write_tokens) if self.cost is not None: result["cost"] = from_union([to_float, from_none], self.cost) if self.duration is not None: result["duration"] = from_union([to_float, from_none], self.duration) if self.initiator is not None: result["initiator"] = from_union([from_str, from_none], self.initiator) if self.input_tokens is not None: result["inputTokens"] = from_union([to_float, from_none], self.input_tokens) if self.model is not None: result["model"] = from_union([from_str, from_none], self.model) if self.output_tokens is not None: result["outputTokens"] = from_union([to_float, from_none], self.output_tokens) if self.quota_snapshots is not None: result["quotaSnapshots"] = from_union([lambda x: from_dict(lambda x: to_class(QuotaSnapshot, x), x), from_none], self.quota_snapshots) if self.reason is not None: result["reason"] = from_union([from_str, from_none], self.reason) if self.arguments is not None: result["arguments"] = self.arguments if self.tool_call_id is not None: result["toolCallId"] = from_union([from_str, from_none], self.tool_call_id) if self.tool_name is not None: result["toolName"] = from_union([from_str, from_none], self.tool_name) if self.mcp_server_name is not None: result["mcpServerName"] = from_union([from_str, from_none], self.mcp_server_name) if self.mcp_tool_name is not None: result["mcpToolName"] = from_union([from_str, from_none], self.mcp_tool_name) if self.partial_output is not None: result["partialOutput"] = from_union([from_str, from_none], self.partial_output) if self.progress_message is not None: result["progressMessage"] = from_union([from_str, from_none], self.progress_message) if self.is_user_requested is not None: result["isUserRequested"] = from_union([from_bool, from_none], self.is_user_requested) if self.result is not None: result["result"] = from_union([lambda x: to_class(Result, x), from_none], self.result) if self.tool_telemetry is not None: result["toolTelemetry"] = from_union([lambda x: from_dict(lambda x: x, x), from_none], self.tool_telemetry) if self.allowed_tools is not None: result["allowedTools"] = from_union([lambda x: from_list(from_str, x), from_none], self.allowed_tools) if self.name is not None: result["name"] = from_union([from_str, from_none], self.name) if self.path is not None: result["path"] = from_union([from_str, from_none], self.path) if self.agent_description is not None: result["agentDescription"] = from_union([from_str, from_none], self.agent_description) if self.agent_display_name is not None: result["agentDisplayName"] = from_union([from_str, from_none], self.agent_display_name) if self.agent_name is not None: result["agentName"] = from_union([from_str, from_none], self.agent_name) if self.tools is not None: result["tools"] = from_union([lambda x: from_list(from_str, x), from_none], self.tools) if self.hook_invocation_id is not None: result["hookInvocationId"] = from_union([from_str, from_none], self.hook_invocation_id) if self.hook_type is not None: result["hookType"] = from_union([from_str, from_none], self.hook_type) if self.input is not None: result["input"] = self.input if self.output is not None: result["output"] = self.output if self.metadata is not None: result["metadata"] = from_union([lambda x: to_class(Metadata, x), from_none], self.metadata) if self.role is not None: result["role"] = from_union([lambda x: to_enum(Role, x), from_none], self.role) return result class SessionEventType(Enum): ABORT = "abort" ASSISTANT_INTENT = "assistant.intent" ASSISTANT_MESSAGE = "assistant.message" ASSISTANT_MESSAGE_DELTA = "assistant.message_delta" ASSISTANT_REASONING = "assistant.reasoning" ASSISTANT_REASONING_DELTA = "assistant.reasoning_delta" ASSISTANT_TURN_END = "assistant.turn_end" ASSISTANT_TURN_START = "assistant.turn_start" ASSISTANT_USAGE = "assistant.usage" HOOK_END = "hook.end" HOOK_START = "hook.start" PENDING_MESSAGES_MODIFIED = "pending_messages.modified" SESSION_COMPACTION_COMPLETE = "session.compaction_complete" SESSION_COMPACTION_START = "session.compaction_start" SESSION_ERROR = "session.error" SESSION_HANDOFF = "session.handoff" SESSION_IDLE = "session.idle" SESSION_INFO = "session.info" SESSION_MODEL_CHANGE = "session.model_change" SESSION_RESUME = "session.resume" SESSION_SHUTDOWN = "session.shutdown" SESSION_SNAPSHOT_REWIND = "session.snapshot_rewind" SESSION_START = "session.start" SESSION_TRUNCATION = "session.truncation" SESSION_USAGE_INFO = "session.usage_info" SKILL_INVOKED = "skill.invoked" SUBAGENT_COMPLETED = "subagent.completed" SUBAGENT_FAILED = "subagent.failed" SUBAGENT_SELECTED = "subagent.selected" SUBAGENT_STARTED = "subagent.started" SYSTEM_MESSAGE = "system.message" TOOL_EXECUTION_COMPLETE = "tool.execution_complete" TOOL_EXECUTION_PARTIAL_RESULT = "tool.execution_partial_result" TOOL_EXECUTION_PROGRESS = "tool.execution_progress" TOOL_EXECUTION_START = "tool.execution_start" TOOL_USER_REQUESTED = "tool.user_requested" USER_MESSAGE = "user.message" # UNKNOWN is used for forward compatibility - new event types from the server # will map to this value instead of raising an error UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "SessionEventType": """Handle unknown event types gracefully for forward compatibility.""" return cls.UNKNOWN @dataclass class SessionEvent: data: Data id: UUID timestamp: datetime type: SessionEventType ephemeral: Optional[bool] = None parent_id: Optional[UUID] = None @staticmethod def from_dict(obj: Any) -> 'SessionEvent': assert isinstance(obj, dict) data = Data.from_dict(obj.get("data")) id = UUID(obj.get("id")) timestamp = from_datetime(obj.get("timestamp")) type = SessionEventType(obj.get("type")) ephemeral = from_union([from_bool, from_none], obj.get("ephemeral")) parent_id = from_union([from_none, lambda x: UUID(x)], obj.get("parentId")) return SessionEvent(data, id, timestamp, type, ephemeral, parent_id) def to_dict(self) -> dict: result: dict = {} result["data"] = to_class(Data, self.data) result["id"] = str(self.id) result["timestamp"] = self.timestamp.isoformat() result["type"] = to_enum(SessionEventType, self.type) if self.ephemeral is not None: result["ephemeral"] = from_union([from_bool, from_none], self.ephemeral) result["parentId"] = from_union([from_none, lambda x: str(x)], self.parent_id) return result def session_event_from_dict(s: Any) -> SessionEvent: return SessionEvent.from_dict(s) def session_event_to_dict(x: SessionEvent) -> Any: return to_class(SessionEvent, x)