Phase D + E: lifecycle polymorphic union + datetime timestamps

Mirrors PR #1357 Phases D + E (TypeScript SDK API review).

Phase D — Session lifecycle event polymorphic hierarchy:

* Split the flat `SessionLifecycleEvent` dataclass into a `SessionLifecycleEventBase`
  + five concrete variant dataclasses (`SessionCreatedEvent`,
  `SessionDeletedEvent`, `SessionUpdatedEvent`, `SessionForegroundEvent`,
  `SessionBackgroundEvent`). `SessionLifecycleEvent` is now a `Union` type alias
  over the five variants; pattern-match on the variant class to branch on the
  event kind. Each variant exposes `type: ClassVar[Literal["..."]]`, so
  existing `event.type == "session.created"` consumer code continues to work.
* `SessionLifecycleEvent.from_dict` becomes a module-private
  `_session_lifecycle_event_from_dict` factory that switches on the wire
  `type` and constructs the matching variant.

Phase D / E — Timestamps as `datetime`:

* `SessionMetadata.startTime` / `modifiedTime` now `datetime` (previously
  ISO-8601 `str`). `to_dict` round-trips back to ISO strings.
* `SessionLifecycleEventMetadata.startTime` / `modifiedTime` likewise.
* All `Pre/PostToolUseHookInput`, `PreMcpToolCallHookInput`,
  `UserPromptSubmittedHookInput`, `SessionStartHookInput`,
  `SessionEndHookInput`, `ErrorOccurredHookInput` declare
  `timestamp: datetime`. `CopilotSession._handle_hooks_invoke` converts the
  wire epoch-milliseconds integer into a timezone-aware `datetime` at the
  dispatch boundary (same place the existing `cwd → workingDirectory` remap
  lives).

Tests updated in lock-step: `test_session_e2e.py` asserts the new `datetime`
types on `SessionMetadata`.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: SteveSandersonMS

python/copilot/client.py

@@ -30,7 +30,7 @@
 from datetime import UTC, datetime
 from pathlib import Path
 from types import TracebackType
-from typing import Any, Literal, TypedDict, cast, overload
+from typing import Any, ClassVar, Literal, TypedDict, cast, overload
 
 from ._diagnostics import log_timing
 from ._jsonrpc import JsonRpcClient, JsonRpcError, ProcessExitedError
@@ -812,8 +812,8 @@ 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
+    startTime: datetime  # Timestamp when session was created
+    modifiedTime: datetime  # 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
@@ -835,8 +835,8 @@ def from_dict(obj: Any) -> SessionMetadata:
         context = SessionContext.from_dict(context_dict) if context_dict else None
         return SessionMetadata(
             sessionId=str(sessionId),
-            startTime=str(startTime),
-            modifiedTime=str(modifiedTime),
+            startTime=_parse_session_timestamp(startTime),
+            modifiedTime=_parse_session_timestamp(modifiedTime),
             isRemote=bool(isRemote),
             summary=summary,
             context=context,
@@ -845,8 +845,8 @@ def from_dict(obj: Any) -> SessionMetadata:
     def to_dict(self) -> dict:
         result: dict = {}
         result["sessionId"] = self.sessionId
-        result["startTime"] = self.startTime
-        result["modifiedTime"] = self.modifiedTime
+        result["startTime"] = self.startTime.isoformat()
+        result["modifiedTime"] = self.modifiedTime.isoformat()
         result["isRemote"] = self.isRemote
         if self.summary is not None:
             result["summary"] = self.summary
@@ -855,6 +855,18 @@ def to_dict(self) -> dict:
         return result
 
 
+def _parse_session_timestamp(value: Any) -> datetime:
+    """Parse a wire-format timestamp into ``datetime``.
+
+    Accepts either an ISO-8601 string (server-sent JSON) or an existing
+    ``datetime`` (round-tripped from a previous parse). Returns the value
+    as-is if it's already a ``datetime``.
+    """
+    if isinstance(value, datetime):
+        return value
+    return from_datetime(value)
+
+
 # ============================================================================
 # Session Lifecycle Types (for TUI+server mode)
 # ============================================================================
@@ -872,37 +884,95 @@ def to_dict(self) -> dict:
 class SessionLifecycleEventMetadata:
     """Metadata for session lifecycle events."""
 
-    startTime: str
-    modifiedTime: str
+    startTime: datetime
+    modifiedTime: datetime
     summary: str | None = None
 
     @staticmethod
     def from_dict(data: dict) -> SessionLifecycleEventMetadata:
         return SessionLifecycleEventMetadata(
-            startTime=data.get("startTime", ""),
-            modifiedTime=data.get("modifiedTime", ""),
+            startTime=_parse_session_timestamp(data.get("startTime", "")),
+            modifiedTime=_parse_session_timestamp(data.get("modifiedTime", "")),
             summary=data.get("summary"),
         )
 
 
 @dataclass
-class SessionLifecycleEvent:
-    """Session lifecycle event notification."""
+class SessionLifecycleEventBase:
+    """Base for session lifecycle event variants.
+
+    Construct concrete variants directly (e.g. :class:`SessionCreatedEvent`,
+    :class:`SessionDeletedEvent`); pattern-match on the variant class to
+    branch on the event kind.
+    """
 
-    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,
-        )
+
+@dataclass
+class SessionCreatedEvent(SessionLifecycleEventBase):
+    """Emitted when a session is created."""
+
+    type: ClassVar[Literal["session.created"]] = "session.created"
+
+
+@dataclass
+class SessionDeletedEvent(SessionLifecycleEventBase):
+    """Emitted when a session is deleted."""
+
+    type: ClassVar[Literal["session.deleted"]] = "session.deleted"
+
+
+@dataclass
+class SessionUpdatedEvent(SessionLifecycleEventBase):
+    """Emitted when a session is updated (summary/title/etc. changed)."""
+
+    type: ClassVar[Literal["session.updated"]] = "session.updated"
+
+
+@dataclass
+class SessionForegroundEvent(SessionLifecycleEventBase):
+    """Emitted when a session moves to the foreground (TUI+server mode)."""
+
+    type: ClassVar[Literal["session.foreground"]] = "session.foreground"
+
+
+@dataclass
+class SessionBackgroundEvent(SessionLifecycleEventBase):
+    """Emitted when a session moves to the background (TUI+server mode)."""
+
+    type: ClassVar[Literal["session.background"]] = "session.background"
+
+
+SessionLifecycleEvent = (
+    SessionCreatedEvent
+    | SessionDeletedEvent
+    | SessionUpdatedEvent
+    | SessionForegroundEvent
+    | SessionBackgroundEvent
+)
+
+
+def _session_lifecycle_event_from_dict(data: dict) -> SessionLifecycleEvent:
+    """Construct the correct :class:`SessionLifecycleEvent` variant from a wire dict."""
+    metadata = None
+    if "metadata" in data and data["metadata"]:
+        metadata = SessionLifecycleEventMetadata.from_dict(data["metadata"])
+    session_id = data.get("sessionId", "")
+    match data.get("type"):
+        case "session.created":
+            return SessionCreatedEvent(sessionId=session_id, metadata=metadata)
+        case "session.deleted":
+            return SessionDeletedEvent(sessionId=session_id, metadata=metadata)
+        case "session.foreground":
+            return SessionForegroundEvent(sessionId=session_id, metadata=metadata)
+        case "session.background":
+            return SessionBackgroundEvent(sessionId=session_id, metadata=metadata)
+        case _:
+            # Default to ``session.updated`` for unknown event types so consumers
+            # keep working across server upgrades.
+            return SessionUpdatedEvent(sessionId=session_id, metadata=metadata)
 
 
 SessionLifecycleHandler = Callable[[SessionLifecycleEvent], None]
@@ -2922,7 +2992,7 @@ def handle_notification(method: str, params: dict):
                     session._dispatch_event(event)
             elif method == "session.lifecycle":
                 # Handle session lifecycle events
-                lifecycle_event = SessionLifecycleEvent.from_dict(params)
+                lifecycle_event = _session_lifecycle_event_from_dict(params)
                 self._dispatch_lifecycle_event(lifecycle_event)
 
         self._client.set_notification_handler(handle_notification)
@@ -3047,7 +3117,7 @@ def handle_notification(method: str, params: dict):
                     session._dispatch_event(event)
             elif method == "session.lifecycle":
                 # Handle session lifecycle events
-                lifecycle_event = SessionLifecycleEvent.from_dict(params)
+                lifecycle_event = _session_lifecycle_event_from_dict(params)
                 self._dispatch_lifecycle_event(lifecycle_event)
 
         self._client.set_notification_handler(handle_notification)

python/copilot/session.py

@@ -18,6 +18,7 @@
 import time
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
+from datetime import UTC, datetime
 from types import TracebackType
 from typing import TYPE_CHECKING, Any, Literal, NotRequired, Required, TypedDict, cast
 
@@ -627,7 +628,7 @@ class PreToolUseHookInput(TypedDict):
     """Input for pre-tool-use hook"""
 
     sessionId: str
-    timestamp: int
+    timestamp: datetime
     workingDirectory: str
     toolName: str
     toolArgs: Any
@@ -653,7 +654,7 @@ class PreMcpToolCallHookInput(TypedDict):
     """Input for pre-MCP-tool-call hook"""
 
     sessionId: str
-    timestamp: int
+    timestamp: datetime
     workingDirectory: str
     serverName: str
     toolName: str
@@ -684,7 +685,7 @@ class PostToolUseHookInput(TypedDict):
     """Input for post-tool-use hook"""
 
     sessionId: str
-    timestamp: int
+    timestamp: datetime
     workingDirectory: str
     toolName: str
     toolArgs: Any
@@ -709,7 +710,7 @@ class UserPromptSubmittedHookInput(TypedDict):
     """Input for user-prompt-submitted hook"""
 
     sessionId: str
-    timestamp: int
+    timestamp: datetime
     workingDirectory: str
     prompt: str
 
@@ -732,7 +733,7 @@ class SessionStartHookInput(TypedDict):
     """Input for session-start hook"""
 
     sessionId: str
-    timestamp: int
+    timestamp: datetime
     workingDirectory: str
     source: Literal["startup", "resume", "new"]
     initialPrompt: NotRequired[str]
@@ -755,7 +756,7 @@ class SessionEndHookInput(TypedDict):
     """Input for session-end hook"""
 
     sessionId: str
-    timestamp: int
+    timestamp: datetime
     workingDirectory: str
     reason: Literal["complete", "error", "abort", "timeout", "user_exit"]
     finalMessage: NotRequired[str]
@@ -780,7 +781,7 @@ class ErrorOccurredHookInput(TypedDict):
     """Input for error-occurred hook"""
 
     sessionId: str
-    timestamp: int
+    timestamp: datetime
     workingDirectory: str
     error: str
     errorContext: Literal["model_call", "tool_execution", "system", "user_input"]
@@ -2229,9 +2230,18 @@ async def _handle_hooks_invoke(self, hook_type: str, input_data: Any) -> Any:
 
         try:
             handler_start = time.perf_counter()
-            # Remap wire key "cwd" to public API key "workingDirectory"
-            if "cwd" in input_data:
-                input_data = {**input_data, "workingDirectory": input_data.pop("cwd")}
+            # Normalize input from the wire format:
+            # - Remap wire key "cwd" to public API key "workingDirectory".
+            # - Convert "timestamp" from epoch milliseconds to ``datetime`` so
+            #   hook handlers see a timezone-aware ``datetime`` rather than a
+            #   raw integer (matches TS PR #1357 Phase E).
+            transformed = dict(input_data)
+            if "cwd" in transformed:
+                transformed["workingDirectory"] = transformed.pop("cwd")
+            timestamp = transformed.get("timestamp")
+            if isinstance(timestamp, (int, float)):
+                transformed["timestamp"] = datetime.fromtimestamp(timestamp / 1000, tz=UTC)
+            input_data = transformed
             result = handler(input_data, {"session_id": self.session_id})
             if inspect.isawaitable(result):
                 result = await result

python/e2e/test_session_e2e.py

@@ -2,6 +2,7 @@
 
 import base64
 import os
+from datetime import datetime
 
 import pytest
 
@@ -306,8 +307,8 @@ async def test_should_list_sessions(self, ctx: E2ETestContext):
             assert hasattr(session_data, "isRemote")
             # summary is optional
             assert isinstance(session_data.sessionId, str)
-            assert isinstance(session_data.startTime, str)
-            assert isinstance(session_data.modifiedTime, str)
+            assert isinstance(session_data.startTime, datetime)
+            assert isinstance(session_data.modifiedTime, datetime)
             assert isinstance(session_data.isRemote, bool)
 
         # Verify context field is present
@@ -365,8 +366,8 @@ async def test_should_get_session_metadata(self, ctx: E2ETestContext):
         metadata = await ctx.client.get_session_metadata(session.session_id)
         assert metadata is not None
         assert metadata.sessionId == session.session_id
-        assert isinstance(metadata.startTime, str)
-        assert isinstance(metadata.modifiedTime, str)
+        assert isinstance(metadata.startTime, datetime)
+        assert isinstance(metadata.modifiedTime, datetime)
         assert isinstance(metadata.isRemote, bool)
 
         # Verify context field is present
@@ -854,8 +855,8 @@ async def test_should_get_session_metadata_by_id(self, ctx: E2ETestContext):
             await asyncio.sleep(0.1)
         assert metadata is not None
         assert metadata.sessionId == session.session_id
-        assert isinstance(metadata.startTime, str) and metadata.startTime
-        assert isinstance(metadata.modifiedTime, str) and metadata.modifiedTime
+        assert isinstance(metadata.startTime, datetime)
+        assert isinstance(metadata.modifiedTime, datetime)
 
         not_found = await ctx.client.get_session_metadata("non-existent-session-id")
         assert not_found is None