""" AUTO-GENERATED FILE - DO NOT EDIT Generated from: @github/copilot/session-events.schema.json Generated by: scripts/generate-session-types.ts Generated at: 2026-01-22T04:11:05.267Z To update these types: 1. Update the schema in copilot-agent-runtime 2. Run: npm run generate:session-types """ from enum import Enum from dataclasses import dataclass from typing import Any, Optional, Dict, Union, List, TypeVar, Type, Callable, cast from datetime import datetime from uuid import UUID import dateutil.parser T = TypeVar("T") EnumT = TypeVar("EnumT", bound=Enum) def from_str(x: Any) -> str: assert isinstance(x, str) return x def to_enum(c: Type[EnumT], x: Any) -> EnumT: assert isinstance(x, c) return x.value 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 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 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_list(f: Callable[[Any], T], x: Any) -> List[T]: assert isinstance(x, list) return [f(y) for y in x] def to_class(c: Type[T], x: Any) -> dict: assert isinstance(x, c) return cast(Any, x).to_dict() class AttachmentType(Enum): DIRECTORY = "directory" FILE = "file" @dataclass class Attachment: display_name: str path: str type: AttachmentType @staticmethod def from_dict(obj: Any) -> 'Attachment': assert isinstance(obj, dict) display_name = from_str(obj.get("displayName")) path = from_str(obj.get("path")) type = AttachmentType(obj.get("type")) return Attachment(display_name, path, type) def to_dict(self) -> dict: result: dict = {} result["displayName"] = from_str(self.display_name) result["path"] = from_str(self.path) result["type"] = to_enum(AttachmentType, self.type) 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 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 @staticmethod def from_dict(obj: Any) -> 'Result': assert isinstance(obj, dict) content = from_str(obj.get("content")) return Result(content) def to_dict(self) -> dict: result: dict = {} result["content"] = from_str(self.content) return result class Role(Enum): DEVELOPER = "developer" SYSTEM = "system" 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 stack: Optional[str] = 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 current_tokens: Optional[float] = None messages_length: Optional[float] = 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 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 message_id: Optional[str] = None parent_tool_call_id: 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 provider_call_id: Optional[str] = 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 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 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 name: Optional[str] = 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")) stack = from_union([from_str, from_none], obj.get("stack")) 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")) current_tokens = from_union([from_float, from_none], obj.get("currentTokens")) messages_length = from_union([from_float, from_none], obj.get("messagesLength")) 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")) 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")) message_id = from_union([from_str, from_none], obj.get("messageId")) parent_tool_call_id = from_union([from_str, from_none], obj.get("parentToolCallId")) 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")) provider_call_id = from_union([from_str, from_none], obj.get("providerCallId")) 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")) 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")) 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")) name = from_union([from_str, from_none], obj.get("name")) 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, stack, 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, current_tokens, messages_length, compaction_tokens_used, error, messages_removed, post_compaction_tokens, pre_compaction_messages_length, pre_compaction_tokens, success, summary_content, tokens_removed, attachments, content, source, transformed_content, turn_id, intent, reasoning_id, delta_content, message_id, parent_tool_call_id, tool_requests, total_response_size_bytes, api_call_id, cache_read_tokens, cache_write_tokens, cost, duration, initiator, input_tokens, model, output_tokens, provider_call_id, quota_snapshots, reason, arguments, tool_call_id, tool_name, partial_output, progress_message, is_user_requested, result, tool_telemetry, agent_description, agent_display_name, agent_name, tools, hook_invocation_id, hook_type, input, output, metadata, name, 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.stack is not None: result["stack"] = from_union([from_str, from_none], self.stack) 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.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.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.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.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.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.provider_call_id is not None: result["providerCallId"] = from_union([from_str, from_none], self.provider_call_id) 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.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.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.name is not None: result["name"] = from_union([from_str, from_none], self.name) 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_START = "session.start" SESSION_TRUNCATION = "session.truncation" SESSION_USAGE_INFO = "session.usage_info" 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)