Phase B fixup: address CI e2e failures from runtime port + Kind/Type enum

Six classes of failures surfaced by the Python e2e CI:

1. `_make_subprocess_client(ctx, use_stdio=False)` in test_pending_work_resume_e2e.py
   and test_suspend_e2e.py was still creating an stdio `RuntimeConnection` regardless
   of the `use_stdio` flag (leftover from the bulk migration). The downstream
   `f"localhost:{server.runtime_port}"` then formatted `None` because stdio doesn't
   listen on a port, triggering `ValueError: Invalid port in cli_url: localhost:None`.
   Now branches on `use_stdio` and constructs `RuntimeConnection.tcp(...)` when False.

2. `.kind.value` / `.type.value` usages in e2e tests assumed the discriminator
   field was an `Enum`. After the Phase L codegen change it's a `ClassVar[str]`,
   so direct equality (`r.kind == "write"`, `attachment.type == "file"`) works
   and no `.value` is needed. Updated test_multi_client_e2e.py, test_permissions_e2e.py,
   test_tools_e2e.py, test_session_e2e.py.

3. `test_rpc_tasks_and_handlers_e2e.py` was constructing the now-deleted
   `PermissionDecision`/`PermissionDecisionApproveForIonApproval` merged
   blob types. Rewrote each call to use the proper per-variant constructors
   (`PermissionDecisionReject(feedback=...)`,
   `PermissionDecisionApprovePermanently(domain=...)`,
   `PermissionDecisionApproveForSession(approval=PermissionDecisionApproveForSessionApprovalCustomTool(...))`,
   `PermissionDecisionApproveForLocation(location_key=..., approval=PermissionDecisionApproveForLocationApprovalCustomTool(...))`),
   and `found_task.type == TaskInfoType.AGENT` to `isinstance(found_task, TaskAgentInfo)`.

4. Codegen: quicktype merges structurally-identical types and synthesizes a
   fuzzy class name (`PermissionDecisionApproveForIonApproval` is the merge
   of `PermissionDecisionApproveFor{Session,Location}Approval`). Extended
   `acronymCandidates` to try the `Session→Ion` / `Location→Ion` substitutions
   so the post-pass can resolve the fuzzy name, and added an explicit cleanup
   step that deletes the now-orphan blob after the union rewrites complete.
   Updated generated rpc.py reflects this.

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

Author: SteveSandersonMS

python/copilot/generated/rpc.py

@@ -279,7 +279,7 @@ async def test_one_client_approves_permission_and_both_see_the_result(
         assert len(c1_perm_completed) > 0
         assert len(c2_perm_completed) > 0
         for event in c1_perm_completed + c2_perm_completed:
-            assert event.data.result.kind.value == "approved"
+            assert event.data.result.kind == "approved"
 
         await session2.disconnect()
 
@@ -328,7 +328,7 @@ async def test_one_client_rejects_permission_and_both_see_the_result(
         assert len(c1_perm_completed) > 0
         assert len(c2_perm_completed) > 0
         for event in c1_perm_completed + c2_perm_completed:
-            assert event.data.result.kind.value == "denied-interactively-by-user"
+            assert event.data.result.kind == "denied-interactively-by-user"
 
         await session2.disconnect()
 

python/e2e/test_multi_client_e2e.py

@@ -36,9 +36,15 @@ def _make_subprocess_client(ctx: E2ETestContext, *, use_stdio: bool = True) -> C
     github_token = (
         "fake-token-for-e2e-tests" if os.environ.get("GITHUB_ACTIONS") == "true" else None
     )
+    if use_stdio:
+        connection = RuntimeConnection.stdio(path=ctx.cli_path)
+    else:
+        connection = RuntimeConnection.tcp(
+            path=ctx.cli_path, connection_token="py-tcp-shared-test-token"
+        )
     return CopilotClient(
         CopilotClientOptions(
-            connection=RuntimeConnection.stdio(path=ctx.cli_path),
+            connection=connection,
             working_directory=ctx.work_dir,
             env=ctx.get_env(),
             github_token=github_token,

python/e2e/test_pending_work_resume_e2e.py

@@ -46,7 +46,7 @@ def on_permission_request(
         assert len(permission_requests) > 0
 
         # Should include write permission request
-        write_requests = [req for req in permission_requests if req.kind.value == "write"]
+        write_requests = [req for req in permission_requests if req.kind == "write"]
         assert len(write_requests) > 0
 
         await session.disconnect()

python/e2e/test_permissions_e2e.py

@@ -12,14 +12,15 @@
 import pytest
 
 from copilot.generated.rpc import (
-    ApprovalKind,
     CommandsHandlePendingCommandRequest,
     HandlePendingToolCallRequest,
-    PermissionDecision,
-    PermissionDecisionApproveForIonApproval,
-    PermissionDecisionKind,
+    PermissionDecisionApproveForLocation,
+    PermissionDecisionApproveForLocationApprovalCustomTool,
+    PermissionDecisionApproveForSession,
+    PermissionDecisionApproveForSessionApprovalCustomTool,
+    PermissionDecisionApprovePermanently,
+    PermissionDecisionReject,
     PermissionDecisionRequest,
-    TaskInfoType,
     TasksCancelRequest,
     TasksPromoteToBackgroundRequest,
     TasksRemoveRequest,
@@ -137,32 +138,24 @@ async def test_should_return_expected_results_for_missing_pending_handler_reques
             permission = await session.rpc.permissions.handle_pending_permission_request(
                 PermissionDecisionRequest(
                     request_id="missing-permission-request",
-                    result=PermissionDecision(
-                        kind=PermissionDecisionKind.REJECT,
-                        feedback="not approved",
-                    ),
+                    result=PermissionDecisionReject(feedback="not approved"),
                 )
             )
             assert permission.success is False
 
             permanent = await session.rpc.permissions.handle_pending_permission_request(
                 PermissionDecisionRequest(
                     request_id="missing-permanent-permission-request",
-                    result=PermissionDecision(
-                        kind=PermissionDecisionKind.APPROVE_PERMANENTLY,
-                        domain="example.com",
-                    ),
+                    result=PermissionDecisionApprovePermanently(domain="example.com"),
                 )
             )
             assert permanent.success is False
 
             session_approval = await session.rpc.permissions.handle_pending_permission_request(
                 PermissionDecisionRequest(
                     request_id="missing-session-approval-request",
-                    result=PermissionDecision(
-                        kind=PermissionDecisionKind.APPROVE_FOR_SESSION,
-                        approval=PermissionDecisionApproveForIonApproval(
-                            kind=ApprovalKind.CUSTOM_TOOL,
+                    result=PermissionDecisionApproveForSession(
+                        approval=PermissionDecisionApproveForSessionApprovalCustomTool(
                             tool_name="missing-tool",
                         ),
                     ),
@@ -173,11 +166,9 @@ async def test_should_return_expected_results_for_missing_pending_handler_reques
             location_approval = await session.rpc.permissions.handle_pending_permission_request(
                 PermissionDecisionRequest(
                     request_id="missing-location-approval-request",
-                    result=PermissionDecision(
-                        kind=PermissionDecisionKind.APPROVE_FOR_LOCATION,
+                    result=PermissionDecisionApproveForLocation(
                         location_key="missing-location",
-                        approval=PermissionDecisionApproveForIonApproval(
-                            kind=ApprovalKind.CUSTOM_TOOL,
+                        approval=PermissionDecisionApproveForLocationApprovalCustomTool(
                             tool_name="missing-tool",
                         ),
                     ),
@@ -215,7 +206,7 @@ async def test_should_report_implemented_error_for_invalid_task_agent_model(
 
     async def test_should_start_background_agent_and_report_task_details(self, ctx: E2ETestContext):
         """Start a background agent task and verify task details then remove it."""
-        from copilot.generated.rpc import TaskInfoExecutionMode, TaskInfoStatus
+        from copilot.generated.rpc import TaskAgentInfo, TaskInfoExecutionMode, TaskInfoStatus
 
         session = await ctx.client.create_session(
             on_permission_request=PermissionHandler.approve_all,
@@ -248,7 +239,7 @@ async def test_should_start_background_agent_and_report_task_details(self, ctx:
             )
             assert found_task.id == task_id
             assert found_task.description == "SDK background agent coverage"
-            assert found_task.type == TaskInfoType.AGENT
+            assert isinstance(found_task, TaskAgentInfo)
             assert found_task.agent_type == "general-purpose"
             assert found_task.execution_mode == TaskInfoExecutionMode.BACKGROUND
             assert found_task.prompt == "Reply with TASK_AGENT_DONE exactly."

python/e2e/test_rpc_tasks_and_handlers_e2e.py

@@ -701,7 +701,7 @@ async def test_should_send_with_file_attachment(self, ctx: E2ETestContext):
         attachments = user_messages[-1].data.attachments
         assert attachments is not None and len(attachments) == 1
         attachment = attachments[0]
-        assert attachment.type.value == "file"
+        assert attachment.type == "file"
         assert attachment.display_name == "attached-file.txt"
         assert attachment.path == file_path
         assert attachment.line_range is not None
@@ -739,7 +739,7 @@ async def test_should_send_with_directory_attachment(self, ctx: E2ETestContext):
         attachments = user_messages[-1].data.attachments
         assert attachments is not None and len(attachments) == 1
         attachment = attachments[0]
-        assert attachment.type.value == "directory"
+        assert attachment.type == "directory"
         assert attachment.display_name == "attached-directory"
         assert attachment.path == directory_path
 
@@ -778,7 +778,7 @@ async def test_should_send_with_selection_attachment(self, ctx: E2ETestContext):
         attachments = user_messages[-1].data.attachments
         assert attachments is not None and len(attachments) == 1
         attachment = attachments[0]
-        assert attachment.type.value == "selection"
+        assert attachment.type == "selection"
         assert attachment.display_name == "selected-file.cs"
         assert attachment.file_path == file_path
         assert attachment.text == 'string Value = "SELECTION_SENTINEL";'

python/e2e/test_session_e2e.py

@@ -30,9 +30,15 @@ def _make_subprocess_client(ctx: E2ETestContext, *, use_stdio: bool = True) -> C
     github_token = (
         "fake-token-for-e2e-tests" if os.environ.get("GITHUB_ACTIONS") == "true" else None
     )
+    if use_stdio:
+        connection = RuntimeConnection.stdio(path=ctx.cli_path)
+    else:
+        connection = RuntimeConnection.tcp(
+            path=ctx.cli_path, connection_token="py-tcp-shared-test-token"
+        )
     return CopilotClient(
         CopilotClientOptions(
-            connection=RuntimeConnection.stdio(path=ctx.cli_path),
+            connection=connection,
             working_directory=ctx.work_dir,
             env=ctx.get_env(),
             github_token=github_token,

python/e2e/test_suspend_e2e.py

@@ -203,7 +203,7 @@ def on_permission_request(request, invocation):
         assert "HELLO" in assistant_message.data.content
 
         # Should have received a custom-tool permission request
-        custom_tool_requests = [r for r in permission_requests if r.kind.value == "custom-tool"]
+        custom_tool_requests = [r for r in permission_requests if r.kind == "custom-tool"]
         assert len(custom_tool_requests) > 0
         assert custom_tool_requests[0].tool_name == "encrypt_string"
 

python/e2e/test_tools_e2e.py

@@ -359,6 +359,14 @@ function postProcessRefBasedDiscriminatedUnionsForPython(
             [/Id\b/g, "ID"],
             [/Llm/g, "LLM"],
             [/Cli/g, "CLI"],
+            // quicktype merges structurally-identical unions and synthesizes a
+            // single class name using a common substring. For our schema this
+            // surfaces as e.g. `PermissionDecisionApproveForIonApproval` (merge
+            // of `PermissionDecisionApproveFor{Session,Location}Approval` —
+            // common substring is `Ion`). Try that substitution so we find
+            // and delete the merged blob during union post-processing.
+            [/Session/g, "Ion"],
+            [/Location/g, "Ion"],
         ];
         const results = new Set<string>([name]);
         for (const [pattern, replacement] of substitutions) {
@@ -451,6 +459,47 @@ function postProcessRefBasedDiscriminatedUnionsForPython(
     }
 
     code = applyUnionRewritesToPython(code, resolved);
+
+    // Clean up quicktype-merged blobs that get left behind when quicktype merges
+    // multiple structurally-identical types into one with a synthesized fuzzy
+    // name (e.g. ``PermissionDecisionApproveForIonApproval`` for the merge of
+    // ``PermissionDecisionApproveFor{Session,Location}Approval``). These
+    // blobs are unused after we emit the proper per-variant + union shape.
+    const orphanFuzzyMergedNames = ["PermissionDecisionApproveForIonApproval"];
+    for (const orphanName of orphanFuzzyMergedNames) {
+        const lines = code.split("\n");
+        let classStart = -1;
+        for (let i = 0; i < lines.length; i++) {
+            if (lines[i] === `class ${orphanName}:` || lines[i].startsWith(`class ${orphanName}(`)) {
+                classStart = i;
+                break;
+            }
+        }
+        if (classStart < 0) continue;
+        let blockStart = classStart;
+        while (
+            blockStart > 0 &&
+            (lines[blockStart - 1] === "@dataclass" || /^# /.test(lines[blockStart - 1]))
+        ) {
+            blockStart--;
+        }
+        let blockEnd = classStart + 1;
+        while (blockEnd < lines.length) {
+            const ln = lines[blockEnd];
+            if (
+                /^class \w/.test(ln) ||
+                /^def \w/.test(ln) ||
+                ln === "@dataclass" ||
+                /^# (?:Experimental|Deprecated|Internal):/.test(ln)
+            ) {
+                break;
+            }
+            blockEnd++;
+        }
+        lines.splice(blockStart, blockEnd - blockStart);
+        code = lines.join("\n");
+    }
+
     return { code, unions: resolved };
 }