python: decode boolean-discriminated unions (#2123)

* python: decode boolean-discriminated unions

The Python generator captured a union discriminator's JSON Schema `const`
with `String()`, so a boolean const became the string "true"/"false" and the
emitted dispatcher matched `case "true":`. A JSON boolean decodes to Python
`True`, which never equals `"true"`, so every boolean-discriminated union
fell through to `raise ValueError`.

Two unions are affected. `sessions.list()` raised
`ValueError: Unknown SessionListEntry isRemote: False` for any non-empty
session list, and `QueuedCommandHandled.to_dict()` put the string `"true"` on
the wire where the schema declares `{"type": "boolean", "const": true}`.

Keep the const's JSON type through codegen and render it as a Python literal
(`True`/`False`), annotating the discriminator `ClassVar` as `bool`. This
mirrors how `go.ts` already models discriminator values. Regenerating changes
six lines of `python/copilot/generated/rpc.py`; no other language changes.

* python: strengthen sessions.list e2e discriminator coverage

Ensure the rpc sessions.list e2e test persists at least one session entry and asserts the matching session decodes to LocalSessionMetadataValue with is_remote=False, exercising the boolean discriminator path end-to-end.

Use an authed client token from GITHUB_TOKEN (default fakevalue) and enqueue a user turn before save/list so the entry is present without depending on full model completion.

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

* python: document intentional ExceptionGroup swallow in sessions.list e2e

CodeQL flagged the new `except ExceptionGroup: pass` in the sessions.list
teardown as an empty except clause with no explanation. Add the same
explanatory comment the four other identical teardowns in this file already
carry.

Also wrap the enqueue `session.send(...)` call, which was 104 characters and
failed the repo's 100-column ruff lint.

* python: wait for the saved session instead of a fixed sleep in sessions.list e2e

The sessions.list e2e test enqueued a turn, slept 200ms, then saved and listed
once. On the Windows runners the enqueued turn was not recorded yet when save
ran, so sessions.list came back empty and `assert len(listed.sessions) >= 1`
failed with `assert 0 >= 1`. Linux and macOS happened to win the race.

Replace the fixed sleep with the existing `wait_for_condition` harness helper,
re-saving on each attempt until the session actually appears in sessions.list.
All discriminator assertions are unchanged, so the boolean-discriminator path
this PR fixes is still exercised end-to-end.

`asyncio` was imported only for the removed sleep, so drop the import.

---------

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

Author: 3 people

python/copilot/generated/rpc.py

@@ -55,7 +55,7 @@
 )
 from copilot.session import PermissionHandler
 
-from .testharness import E2ETestContext
+from .testharness import E2ETestContext, wait_for_condition
 
 pytestmark = pytest.mark.asyncio(loop_scope="module")
 
@@ -282,63 +282,104 @@ async def test_should_add_secret_filter_values(self, ctx: E2ETestContext):
                 # error from anyio. We don't want it to fail the test.
                 pass
 
-    async def test_should_list_find_and_inspect_persisted_session_state(self, ctx: E2ETestContext):
+    async def test_should_list_find_and_inspect_persisted_session_state(
+        self, authed_ctx: E2ETestContext
+    ):
+        token = os.environ.get("GITHUB_TOKEN", "fakevalue")
+        await _configure_user(authed_ctx, token)
+        client = _make_authed_client(authed_ctx, token)
+
         session_id = str(uuid.uuid4())
-        working_directory = Path(ctx.work_dir) / f"server-rpc-list-{uuid.uuid4().hex}"
+        working_directory = Path(authed_ctx.work_dir) / f"server-rpc-list-{uuid.uuid4().hex}"
         working_directory.mkdir(parents=True, exist_ok=True)
         missing_task_id = f"missing-task-{uuid.uuid4().hex}"
         missing_session_id = str(uuid.uuid4())
-
-        session = await ctx.client.create_session(
-            session_id=session_id,
-            working_directory=str(working_directory),
-            on_permission_request=PermissionHandler.approve_all,
-        )
+        session = None
         try:
-            await session.log("SERVER_RPC_LIST_READY")
-            save = await ctx.client.rpc.sessions.save(SessionsSaveRequest(session_id=session_id))
-            assert save is not None
-
-            listed = await ctx.client.rpc.sessions.list(
-                SessionsListRequest(
-                    filter=SessionListFilter(cwd=str(working_directory)),
-                    metadata_limit=0,
+            await client.start()
+            session = await client.create_session(
+                session_id=session_id,
+                working_directory=str(working_directory),
+                on_permission_request=PermissionHandler.approve_all,
+            )
+
+            await session.send(
+                "Record a turn for sessions.list discriminator coverage", mode="enqueue"
+            )
+
+            listed = None
+
+            async def session_is_listed() -> bool:
+                nonlocal listed
+                # Re-save on every attempt: on slower runners the enqueued turn is not
+                # necessarily recorded yet when the first save runs, so a single save
+                # followed by a fixed sleep races the CLI's own persistence.
+                save = await client.rpc.sessions.save(SessionsSaveRequest(session_id=session_id))
+                assert save is not None
+                listed = await client.rpc.sessions.list(
+                    SessionsListRequest(
+                        filter=SessionListFilter(cwd=str(working_directory)),
+                        metadata_limit=0,
+                    )
                 )
+                return any(item.session_id == session_id for item in listed.sessions or [])
+
+            await wait_for_condition(
+                session_is_listed,
+                timeout=60.0,
+                timeout_message=(
+                    "Timed out waiting for the saved session to be returned by sessions.list."
+                ),
             )
+
+            assert listed is not None
             assert listed.sessions is not None
+            assert len(listed.sessions) >= 1
+            matching = [item for item in listed.sessions if item.session_id == session_id]
+            assert len(matching) == 1
+            assert isinstance(matching[0], LocalSessionMetadataValue)
+            assert matching[0].is_remote is False
             assert all(
                 item.context is None
                 or os.path.normcase(os.path.abspath(item.context.cwd))
                 == os.path.normcase(os.path.abspath(str(working_directory)))
                 for item in listed.sessions
             )
 
-            by_prefix = await ctx.client.rpc.sessions.find_by_prefix(
+            by_prefix = await client.rpc.sessions.find_by_prefix(
                 SessionsFindByPrefixRequest(prefix=session_id[:8])
             )
             assert by_prefix.session_id in (None, session_id)
 
-            by_task = await ctx.client.rpc.sessions.find_by_task_id(
+            by_task = await client.rpc.sessions.find_by_task_id(
                 SessionsFindByTaskIDRequest(task_id=missing_task_id)
             )
             assert by_task.session_id is None
 
-            last_for_context = await ctx.client.rpc.sessions.get_last_for_context(
+            last_for_context = await client.rpc.sessions.get_last_for_context(
                 SessionsGetLastForContextRequest(context=SessionContext(cwd=str(working_directory)))
             )
             assert last_for_context.session_id in (None, session_id)
 
-            sizes = await ctx.client.rpc.sessions.get_sizes()
+            sizes = await client.rpc.sessions.get_sizes()
             assert sizes.sizes is not None
             if session_id in sizes.sizes:
                 assert sizes.sizes[session_id] >= 0
 
-            in_use = await ctx.client.rpc.sessions.check_in_use(
+            in_use = await client.rpc.sessions.check_in_use(
                 SessionsCheckInUseRequest(session_ids=[session_id, missing_session_id])
             )
             assert missing_session_id not in in_use.in_use
         finally:
-            await session.disconnect()
+            if session is not None:
+                await session.disconnect()
+            try:
+                await client.stop()
+            except ExceptionGroup:
+                # Intentional: shutting down the per-test client can race the
+                # CLI's own teardown and surface as an aggregated cancellation
+                # error from anyio. We don't want it to fail the test.
+                pass
 
     async def test_should_enrich_basic_session_metadata(self, ctx: E2ETestContext):
         session_id = str(uuid.uuid4())

python/e2e/test_rpc_server_e2e.py

@@ -1,12 +1,21 @@
 """Tests for generated RPC method behavior."""
 
+import json
 from unittest.mock import AsyncMock
 
 import pytest
 
 from copilot.rpc import (
     CommandsApi,
     CommandsInvokeRequest,
+    CommandsRespondToQueuedCommandRequest,
+    LocalSessionMetadataValue,
+    QueuedCommandHandled,
+    QueuedCommandNotHandled,
+    RemoteControlStatusOff,
+    RemoteControlStatusResult,
+    RemoteSessionMetadataValue,
+    SessionList,
     SlashCommandTextResult,
 )
 
@@ -22,3 +31,77 @@ async def test_commands_invoke_deserializes_slash_command_result():
     assert isinstance(result, SlashCommandTextResult)
     assert result.text == "hello"
     assert result.markdown is True
+
+
+def test_remote_control_status_deserializes_string_discriminated_union():
+    result = RemoteControlStatusResult.from_dict({"status": {"state": "off"}})
+
+    assert isinstance(result.status, RemoteControlStatusOff)
+    assert result.status.state == "off"
+    assert result.status.to_dict() == {"state": "off"}
+
+
+def test_session_list_deserializes_boolean_discriminated_entries():
+    payload = {
+        "sessions": [
+            {
+                "sessionId": "example-local",
+                "startTime": "2026-07-26T10:00:00.000Z",
+                "modifiedTime": "2026-07-26T10:05:00.000Z",
+                "isRemote": False,
+            },
+            {
+                "sessionId": "example-remote",
+                "startTime": "2026-07-26T11:00:00.000Z",
+                "modifiedTime": "2026-07-26T11:05:00.000Z",
+                "isRemote": True,
+                "remoteSessionIds": ["example-remote"],
+                "repository": {"owner": "github", "name": "copilot-sdk", "branch": "main"},
+            },
+        ]
+    }
+
+    result = SessionList.from_dict(payload)
+
+    local, remote = result.sessions
+    assert isinstance(local, LocalSessionMetadataValue)
+    assert local.session_id == "example-local"
+    assert local.is_remote is False
+    assert isinstance(remote, RemoteSessionMetadataValue)
+    assert remote.session_id == "example-remote"
+    assert remote.is_remote is True
+    assert remote.repository.owner == "github"
+
+
+@pytest.mark.parametrize(
+    ("handled", "expected_type"),
+    [(True, QueuedCommandHandled), (False, QueuedCommandNotHandled)],
+)
+def test_queued_command_result_deserializes_boolean_discriminator(handled, expected_type):
+    request = CommandsRespondToQueuedCommandRequest.from_dict(
+        {"requestId": "example-request", "result": {"handled": handled}}
+    )
+
+    assert isinstance(request.result, expected_type)
+
+
+@pytest.mark.parametrize(
+    ("variant", "expected_handled", "expected_json"),
+    [
+        (QueuedCommandHandled(), True, '{"handled": true}'),
+        (QueuedCommandNotHandled(), False, '{"handled": false}'),
+    ],
+)
+def test_queued_command_result_serializes_boolean_discriminator(
+    variant, expected_handled, expected_json
+):
+    encoded = variant.to_dict()
+
+    assert encoded["handled"] is expected_handled
+    assert json.dumps(encoded) == expected_json
+
+    request = CommandsRespondToQueuedCommandRequest(request_id="example-request", result=variant)
+    round_tripped = CommandsRespondToQueuedCommandRequest.from_dict(request.to_dict())
+
+    assert request.to_dict()["result"]["handled"] is expected_handled
+    assert isinstance(round_tripped.result, type(variant))