Stabilize Python and Java E2E harnesses

java/sdk/src/test/java/com/github/copilot/CapiProxy.java

@@ -57,16 +57,15 @@ public class CapiProxy implements AutoCloseable {
 
     private static final ObjectMapper MAPPER = new ObjectMapper();
     private static final Pattern LISTENING_PATTERN = Pattern.compile("Listening: (http://[^\\s]+)(?:\\s+(\\{.*\\}))?$");
+    private static final HttpClient HTTP_CLIENT = HttpClient.newHttpClient();
 
     private Process process;
     private String proxyUrl;
     private String connectProxyUrl;
     private String caFilePath;
-    private final HttpClient httpClient;
     private BufferedReader stdoutReader;
 
     public CapiProxy() {
-        this.httpClient = HttpClient.newHttpClient();
     }
 
     /**
@@ -212,7 +211,7 @@ public void configure(String filePath, String workDir, TestInfo testInfo) throws
         HttpRequest request = HttpRequest.newBuilder().uri(URI.create(proxyUrl + "/config"))
                 .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString(body)).build();
 
-        HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+        HttpResponse<String> response = HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString());
         if (response.statusCode() != 200) {
             throw new IOException("Proxy config failed with status " + response.statusCode() + ": " + response.body());
         }
@@ -234,7 +233,7 @@ public List<Map<String, Object>> getExchanges() throws IOException, InterruptedE
 
         HttpRequest request = HttpRequest.newBuilder().uri(URI.create(proxyUrl + "/exchanges")).GET().build();
 
-        HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+        HttpResponse<String> response = HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString());
         if (response.statusCode() != 200) {
             throw new IOException("Failed to get exchanges: " + response.statusCode());
         }
@@ -284,7 +283,7 @@ public void setCopilotUserByToken(String token, String login, String copilotPlan
         HttpRequest request = HttpRequest.newBuilder().uri(URI.create(proxyUrl + "/copilot-user-config"))
                 .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString(body)).build();
 
-        HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+        HttpResponse<String> response = HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString());
         if (response.statusCode() != 200) {
             throw new IOException(
                     "Failed to set copilot user config: " + response.statusCode() + ": " + response.body());
@@ -331,7 +330,7 @@ public void setCopilotUserByToken(String token, Map<String, Object> response)
         HttpRequest request = HttpRequest.newBuilder().uri(URI.create(proxyUrl + "/copilot-user-config"))
                 .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString(body)).build();
 
-        HttpResponse<String> response2 = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+        HttpResponse<String> response2 = HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString());
         if (response2.statusCode() != 200) {
             throw new IOException(
                     "Failed to set copilot user config: " + response2.statusCode() + ": " + response2.body());
@@ -376,7 +375,7 @@ public void stop(boolean skipWritingCache) throws IOException, InterruptedExcept
                 HttpRequest request = HttpRequest.newBuilder().uri(URI.create(stopUrl))
                         .POST(HttpRequest.BodyPublishers.noBody()).build();
 
-                httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+                HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString());
             } catch (Exception e) {
                 // Best effort - ignore errors
             }

java/sdk/src/test/java/com/github/copilot/CapiProxyTest.java

@@ -0,0 +1,21 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) Microsoft Corporation. All rights reserved.
+ *--------------------------------------------------------------------------------------------*/
+
+package com.github.copilot;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.lang.reflect.Modifier;
+
+import org.junit.jupiter.api.Test;
+
+class CapiProxyTest {
+
+    @Test
+    void proxyInstancesShareHttpClientResources() throws Exception {
+        var field = CapiProxy.class.getDeclaredField("HTTP_CLIENT");
+        assertTrue(Modifier.isStatic(field.getModifiers()),
+                "E2E contexts must not retain one HttpClient selector manager per test class");
+    }
+}

java/sdk/src/test/java/com/github/copilot/E2ETestContext.java

@@ -65,8 +65,8 @@ public class E2ETestContext implements AutoCloseable {
      */
     private static final String DEFAULT_GITHUB_TOKEN = "fake-token-for-e2e-tests";
     private static final Pattern SNAKE_CASE = Pattern.compile("[^a-zA-Z0-9]");
-    private static final Pattern USER_CONTENT_PATTERN = Pattern
-            .compile("^\\s+-\\s+role:\\s+user\\s*$\\s+content:\\s*(.+?)$", Pattern.MULTILINE);
+    private static final Pattern USER_ROLE_PATTERN = Pattern.compile("^(\\s*)-\\s+role:\\s+user\\s*$");
+    private static final Pattern CONTENT_PATTERN = Pattern.compile("^(\\s*)content:\\s*(.*)$");
 
     private final String cliPath;
     private final Path homeDir;
@@ -227,25 +227,109 @@ public List<String> getExpectedUserPrompts() {
             return List.of();
         }
         try {
-            String content = Files.readString(currentSnapshotFile);
-            List<String> prompts = new ArrayList<>();
-            Matcher matcher = USER_CONTENT_PATTERN.matcher(content);
-            while (matcher.find()) {
-                String prompt = matcher.group(1).trim();
-                // Remove quotes if present
-                if ((prompt.startsWith("\"") && prompt.endsWith("\""))
-                        || (prompt.startsWith("'") && prompt.endsWith("'"))) {
-                    prompt = prompt.substring(1, prompt.length() - 1);
+            return parseExpectedUserPrompts(Files.readString(currentSnapshotFile));
+        } catch (IOException e) {
+            LOG.warning("Failed to read snapshot file: " + e.getMessage());
+            return List.of();
+        }
+    }
+
+    static List<String> parseExpectedUserPrompts(String yaml) {
+        String[] lines = yaml.split("\\R", -1);
+        List<String> prompts = new ArrayList<>();
+
+        for (int i = 0; i < lines.length; i++) {
+            Matcher roleMatcher = USER_ROLE_PATTERN.matcher(lines[i]);
+            if (!roleMatcher.matches()) {
+                continue;
+            }
+
+            int roleIndent = roleMatcher.group(1).length();
+            for (i++; i < lines.length; i++) {
+                String line = lines[i];
+                if (!line.isBlank() && leadingWhitespace(line) <= roleIndent) {
+                    i--;
+                    break;
                 }
-                if (!prompts.contains(prompt)) {
+
+                Matcher contentMatcher = CONTENT_PATTERN.matcher(line);
+                if (!contentMatcher.matches()) {
+                    continue;
+                }
+
+                int contentIndent = contentMatcher.group(1).length();
+                String value = contentMatcher.group(2).trim();
+                List<String> continuation = new ArrayList<>();
+                while (i + 1 < lines.length
+                        && (lines[i + 1].isBlank() || leadingWhitespace(lines[i + 1]) > contentIndent)) {
+                    continuation.add(lines[++i]);
+                }
+
+                String prompt = isBlockScalar(value)
+                        ? parseBlockScalar(value.charAt(0), continuation)
+                        : parsePlainScalar(value, continuation);
+                if (!prompt.isEmpty() && !prompts.contains(prompt)) {
                     prompts.add(prompt);
                 }
+                break;
             }
-            return prompts;
-        } catch (IOException e) {
-            LOG.warning("Failed to read snapshot file: " + e.getMessage());
-            return List.of();
         }
+
+        return prompts;
+    }
+
+    private static String parsePlainScalar(String firstLine, List<String> continuation) {
+        StringBuilder value = new StringBuilder(unquote(firstLine));
+        for (String line : continuation) {
+            if (!line.isBlank()) {
+                if (!value.isEmpty()) {
+                    value.append(' ');
+                }
+                value.append(line.trim());
+            }
+        }
+        return value.toString();
+    }
+
+    private static String parseBlockScalar(char style, List<String> lines) {
+        int contentIndent = lines.stream().filter(line -> !line.isBlank()).mapToInt(E2ETestContext::leadingWhitespace)
+                .min().orElse(0);
+        StringBuilder value = new StringBuilder();
+        boolean previousWasContent = false;
+        for (String line : lines) {
+            String text = line.isBlank() ? "" : line.substring(Math.min(contentIndent, line.length()));
+            if (text.isEmpty()) {
+                value.append('\n');
+                previousWasContent = false;
+            } else {
+                if (previousWasContent) {
+                    value.append(style == '>' ? ' ' : '\n');
+                }
+                value.append(text);
+                previousWasContent = true;
+            }
+        }
+        return value.toString().stripTrailing();
+    }
+
+    private static boolean isBlockScalar(String value) {
+        return value.matches("[>|][+-]?");
+    }
+
+    private static String unquote(String value) {
+        if (value.length() >= 2 && ((value.startsWith("\"") && value.endsWith("\""))
+                || (value.startsWith("'") && value.endsWith("'")))) {
+            return value.substring(1, value.length() - 1);
+        }
+        return value;
+    }
+
+    private static int leadingWhitespace(String value) {
+        int index = 0;
+        while (index < value.length() && Character.isWhitespace(value.charAt(index))) {
+            index++;
+        }
+        return index;
     }
 
     /**

java/sdk/src/test/java/com/github/copilot/E2ETestContextTest.java

@@ -0,0 +1,39 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) Microsoft Corporation. All rights reserved.
+ *--------------------------------------------------------------------------------------------*/
+
+package com.github.copilot;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+class E2ETestContextTest {
+
+    @Test
+    void expectedUserPromptsParseFoldedBlockScalars() {
+        String snapshot = """
+                conversations:
+                  - messages:
+                      - role: user
+                        content: First prompt
+                          continued here.
+                      - role: assistant
+                        content: response
+                      - role: user
+                        content: >-
+                          <system_notification>
+
+                          Agent completed successfully.
+
+                          </system_notification>
+                """;
+
+        assertEquals(
+                List.of("First prompt continued here.",
+                        "<system_notification>\nAgent completed successfully.\n</system_notification>"),
+                E2ETestContext.parseExpectedUserPrompts(snapshot));
+    }
+}

python/e2e/test_pending_work_resume_e2e.py

@@ -20,11 +20,12 @@
     HandlePendingToolCallRequest,
     PermissionDecisionRequest,
     PermissionDecisionUserNotAvailable,
+    SessionsCheckInUseRequest,
 )
 from copilot.session import PermissionHandler
 from copilot.tools import Tool, ToolInvocation, ToolResult
 
-from .testharness import DEFAULT_GITHUB_TOKEN, E2ETestContext
+from .testharness import DEFAULT_GITHUB_TOKEN, E2ETestContext, wait_for_condition
 
 pytestmark = pytest.mark.asyncio(loop_scope="module")
 
@@ -464,6 +465,7 @@ async def blocking_external_tool(args):
         await server.start()
         try:
             cli_url = f"localhost:{server.runtime_port}"
+            lock_observer: CopilotClient | None = None
 
             suspended_client = CopilotClient(
                 connection=RuntimeConnection.for_uri(
@@ -487,8 +489,41 @@ async def blocking_external_tool(args):
                 assert (await asyncio.wait_for(tool_started, PENDING_WORK_TIMEOUT)) == "beta"
 
                 if disconnect_original_client:
+                    # force_stop closes the local socket before the server necessarily
+                    # processes that disconnect. Observe the session lock from another
+                    # runtime so resume cannot race the server's active-session cleanup.
+                    lock_observer = _make_subprocess_client(ctx)
+                    await lock_observer.start()
+
+                    async def session_lock_is_held() -> bool:
+                        result = await lock_observer.rpc.sessions.check_in_use(
+                            SessionsCheckInUseRequest(session_ids=[session_id])
+                        )
+                        return session_id in result.in_use
+
+                    await wait_for_condition(
+                        session_lock_is_held,
+                        timeout=PENDING_WORK_TIMEOUT,
+                        timeout_message=(
+                            f"Timed out waiting for session '{session_id}' to acquire its lock."
+                        ),
+                    )
                     await suspended_client.force_stop()
 
+                    async def session_lock_is_released() -> bool:
+                        result = await lock_observer.rpc.sessions.check_in_use(
+                            SessionsCheckInUseRequest(session_ids=[session_id])
+                        )
+                        return session_id not in result.in_use
+
+                    await wait_for_condition(
+                        session_lock_is_released,
+                        timeout=PENDING_WORK_TIMEOUT,
+                        timeout_message=(
+                            f"Timed out waiting for session '{session_id}' to release its lock."
+                        ),
+                    )
+
                 resumed_client = CopilotClient(
                     connection=RuntimeConnection.for_uri(
                         cli_url, connection_token="py-tcp-shared-test-token"
@@ -550,6 +585,8 @@ async def resumed_external_tool(args):
                 if not release_original.done():
                     release_original.set_result("ORIGINAL_SHOULD_NOT_WIN")
                 await _safe_force_stop(suspended_client)
+                if lock_observer is not None:
+                    await _safe_force_stop(lock_observer)
         finally:
             await _safe_force_stop(server)
 

python/e2e/test_rpc_shell_and_fleet_e2e.py

@@ -37,11 +37,10 @@
 
 def _write_file_command(marker_path: Path, marker: str) -> str:
     if sys.platform == "win32":
-        return (
-            f"powershell -NoLogo -NoProfile -Command "
-            f"\"Set-Content -LiteralPath '{marker_path}' -Value '{marker}'\""
-        )
-    return f"sh -c \"printf '%s' '{marker}' > '{marker_path}'\""
+        # shell.exec already runs through cmd.exe on Windows. Use its built-in echo
+        # instead of spawning a nested PowerShell process just to write the marker.
+        return f'echo {marker}>"{marker_path.name}"'
+    return f"sh -c \"printf '%s' '{marker}' > '{marker_path.name}'\""
 
 
 async def _wait_for_file_text(path: Path, expected: str, *, timeout: float = 30.0) -> None:
@@ -57,19 +56,21 @@ async def _wait_for_file_text(path: Path, expected: str, *, timeout: float = 30.
 
 class TestRpcShellAndFleet:
     async def test_should_execute_shell_command(self, ctx: E2ETestContext):
-        session = await ctx.client.create_session(
+        async with await ctx.client.create_session(
             on_permission_request=PermissionHandler.approve_all,
-        )
-        marker_path = Path(ctx.work_dir) / f"shell-rpc-{uuid.uuid4().hex}.txt"
-        marker = "copilot-sdk-shell-rpc"
-
-        result = await session.rpc.shell.exec(
-            ShellExecRequest(command=_write_file_command(marker_path, marker), cwd=ctx.work_dir)
-        )
-        assert (result.process_id or "").strip()
-        await _wait_for_file_text(marker_path, marker)
-
-        await session.disconnect()
+        ) as session:
+            command_dir = Path(ctx.work_dir) / f"shell-rpc-{uuid.uuid4().hex}"
+            command_dir.mkdir()
+            marker_path = command_dir / "marker.txt"
+            marker = "copilot-sdk-shell-rpc"
+
+            result = await session.rpc.shell.exec(
+                ShellExecRequest(
+                    command=_write_file_command(marker_path, marker), cwd=str(command_dir)
+                )
+            )
+            assert (result.process_id or "").strip()
+            await _wait_for_file_text(marker_path, marker)
 
     async def test_should_kill_shell_process(self, ctx: E2ETestContext):
         session = await ctx.client.create_session(