fix(execd): defer SSE response headers until first event fires (#912)

* test(e2e): harden flaky cross-SDK sandbox tests

Recurring failures in `real-e2e.yml` traced to three races:

1. Go `TestE2E_FullLifecycle` pings execd via the server proxy the moment
   sandbox state flips to Running; execd's TCP listener can accept before
   routes register, surfacing as `read: connection reset by peer` on
   `/proxy/44772/ping`. Wrap Ping + the first SSE RunCommand in
   `require.Eventually` retries.

2. Go `TestE2E_PauseResume` and `TestFilesystem_SetPermissions` hit
   `opensandbox: empty sse stream` immediately after Resume / readiness.
   Add `runCommandWithRetry` helper in base_e2e_test.go and use it in the
   affected call sites.

3. Java/C# NetworkPolicy tests asserted egress blocking after a fixed
   `Thread.sleep`/`Task.Delay`; the sidecar accepts the sandbox before
   iptables/proxy rules apply, so curl occasionally succeeds and
   `assertNotNull(error)` fails. Replace fixed sleeps with
   `waitUntilEgressBlocks` / `WaitUntilEgressBlocksAsync` helpers that
   poll curl until the policy actually blocks (or fail with the last
   observation).

Also retry transient single-line stdout drops on the workingDirectory
`pwd` checks (JS, Java) and the C# env-injection baseline — same SSE
blank-line / first-event race that has bitten multiple SDKs.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(execd): defer SSE response headers until first event fires

`RunCommand`, `RunCode`, and `RunInSession` previously committed the
response as `text/event-stream` (via `setupSSEResponse`) before invoking
the runtime. If the runtime returned a synchronous error — for example
because `stdLogDescriptor` could not (re)create `/tmp` after a sandbox
restart, `pathutil.ExpandPathWithEnv` failed to resolve a working
directory under transient env conditions, or `buildCredential` raced
with a uid/gid lookup — the handler then called `RespondError`, which
set Content-Type to `application/json` and wrote a JSON body on top of
the already-committed event-stream response. Clients saw HTTP 200 with
`text/event-stream` and a JSON body that no SSE parser could decode,
producing zero events; the Go SDK reported `opensandbox: empty sse
stream` and JS/Java/C# SDKs surfaced the same race as missing init
events or a vanished single-line stdout (e.g. `pwd` with
`workingDirectory: "/tmp"` returning empty `stdout[0]`).

Make `setupSSEResponse` idempotent (guarded by `sync.Once`) and call it
lazily from `writeSingleEvent`, so headers commit only once an event is
actually being written. Drop the eager `setupSSEResponse` calls in the
three streaming endpoints. Pre-execution synchronous errors now flow
through `RespondError` cleanly with `application/json`, and successful
runs still emit `text/event-stream` on the first event.

Add three regression tests:

- `TestRunCodeSyncErrorEmitsJSONNotSSE`
- `TestRunInSessionSyncErrorEmitsJSONNotSSE`
- `TestRunCodeSuccessStillEmitsSSE`

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* test(e2e): drop SDK-side retries now that execd SSE bug is fixed

The execd "empty sse stream" / dropped-event race is fixed at the source
in the previous commit (lazy SSE headers so synchronous runtime errors
return JSON, not a half-formed event-stream). Real SDK clients do not
retry these failures, so the e2e suite shouldn't either — the retries
were masking the bug, not exercising production behaviour.

Revert the SDK-side workarounds:

- JS `pwd workingDirectory:/tmp` retry loop
- Java echo + pwd retry loops
- C# env-injection `RunWithRetryAsync` wrapper
- Go `runCommandWithRetry` helper and its callers in
  `TestE2E_PauseResume` and `TestFilesystem_SetPermissions`

Keep the targeted polls that cover other races not addressed by the
execd fix:

- Go `TestE2E_FullLifecycle` execd-Ping `Eventually` — bypasses the
  high-level SDK and pings the server-side proxy directly; the proxy
  drops the very first connection in the gap between sandbox state
  Running and execd routes registering. Real users go through
  `CreateSandbox`/`WaitUntilReady`, which already handles this; the
  low-level test does not.
- Java/C# `waitUntilEgressBlocks` polls — egress sidecar policy is
  applied asynchronously after the sandbox is marked ready, so a fixed
  sleep is inherently flaky. This is a separate readiness-gating bug
  that should be addressed server-side.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: Pangjiping

components/execd/pkg/web/controller/basic.go

@@ -18,14 +18,16 @@ import (
 	"encoding/json"
 	"net/http"
 	"strconv"
+	"sync"
 
 	"github.com/gin-gonic/gin"
 
 	"github.com/alibaba/opensandbox/execd/pkg/web/model"
 )
 
 type basicController struct {
-	ctx *gin.Context
+	ctx          *gin.Context
+	sseSetupOnce sync.Once
 }
 
 func newBasicController(ctx *gin.Context) *basicController {

components/execd/pkg/web/controller/codeinterpreting.go

@@ -171,7 +171,9 @@ func (c *CodeInterpretingController) RunCode() {
 	}
 	runCodeRequest.Hooks = eventsHandler
 
-	c.setupSSEResponse()
+	// SSE headers are committed lazily on the first event write
+	// (see writeSingleEvent), so a synchronous error from Execute below can
+	// still be surfaced as a structured JSON error response.
 	err = codeRunner.Execute(runCodeRequest)
 	if err != nil {
 		recordExecution("failure")
@@ -400,7 +402,9 @@ func (c *CodeInterpretingController) RunInSession() {
 	}
 	runReq.Hooks = hooks
 
-	c.setupSSEResponse()
+	// SSE headers are committed lazily on the first event write
+	// (see writeSingleEvent), so a synchronous error from
+	// RunInBashSession can still be surfaced as a structured JSON error.
 	err := codeRunner.RunInBashSession(ctx, runReq)
 	if err != nil {
 		recordExecution("failure")

components/execd/pkg/web/controller/codeinterpreting_test.go

@@ -17,6 +17,7 @@ package controller
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"net/http"
 	"testing"
 	"time"
@@ -333,3 +334,90 @@ func TestRunInSessionReturnsBeforeGracefulShutdownTimeoutAfterImmediateError(t *
 	require.Equal(t, http.StatusOK, w.Code)
 	require.Less(t, elapsed, flag.ApiGracefulShutdownTimeout/2)
 }
+
+// TestRunCodeSyncErrorEmitsJSONNotSSE guards against regression of the bug
+// where Execute returning a synchronous error after setupSSEResponse caused
+// the client to receive a text/event-stream response with a JSON body, which
+// SDKs parsed as zero events ("empty sse stream"). Headers must stay
+// uncommitted until the first event so RespondError can produce a proper
+// application/json error response.
+func TestRunCodeSyncErrorEmitsJSONNotSSE(t *testing.T) {
+	previousRunner := codeRunner
+	codeRunner = &fakeCodeRunner{
+		execute: func(_ *runtime.ExecuteCodeRequest) error {
+			return errors.New("synchronous runtime failure")
+		},
+	}
+	t.Cleanup(func() { codeRunner = previousRunner })
+
+	body := []byte(`{"code":"print(1)","context":{"id":"ctx-1","language":"python"}}`)
+	ctx, w := newTestContext(http.MethodPost, "/code/run", body)
+	ctrl := NewCodeInterpretingController(ctx)
+
+	ctrl.RunCode()
+
+	require.Equal(t, http.StatusInternalServerError, w.Code)
+	contentType := w.Header().Get("Content-Type")
+	require.Contains(t, contentType, "application/json", "should not commit text/event-stream when no event fires")
+
+	var resp model.ErrorResponse
+	require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp))
+	require.Equal(t, model.ErrorCodeRuntimeError, resp.Code)
+	require.Contains(t, resp.Message, "synchronous runtime failure")
+}
+
+// TestRunInSessionSyncErrorEmitsJSONNotSSE — see TestRunCodeSyncErrorEmitsJSONNotSSE.
+func TestRunInSessionSyncErrorEmitsJSONNotSSE(t *testing.T) {
+	previousRunner := codeRunner
+	codeRunner = &fakeCodeRunner{
+		runInBashSession: func(_ context.Context, _ *runtime.ExecuteCodeRequest) error {
+			return errors.New("synchronous session failure")
+		},
+	}
+	t.Cleanup(func() { codeRunner = previousRunner })
+
+	body := []byte(`{"command":"echo hi","timeout":0}`)
+	ctx, w := newTestContext(http.MethodPost, "/sessions/session-1/run", body)
+	ctx.Params = append(ctx.Params, gin.Param{Key: "sessionId", Value: "session-1"})
+	ctrl := NewCodeInterpretingController(ctx)
+
+	ctrl.RunInSession()
+
+	require.Equal(t, http.StatusInternalServerError, w.Code)
+	contentType := w.Header().Get("Content-Type")
+	require.Contains(t, contentType, "application/json", "should not commit text/event-stream when no event fires")
+
+	var resp model.ErrorResponse
+	require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp))
+	require.Equal(t, model.ErrorCodeRuntimeError, resp.Code)
+	require.Contains(t, resp.Message, "synchronous session failure")
+}
+
+// TestRunCodeSuccessStillEmitsSSE confirms the lazy header path still produces
+// a text/event-stream response when at least one event fires.
+func TestRunCodeSuccessStillEmitsSSE(t *testing.T) {
+	previousRunner := codeRunner
+	previousTimeout := flag.ApiGracefulShutdownTimeout
+	codeRunner = &fakeCodeRunner{
+		execute: func(request *runtime.ExecuteCodeRequest) error {
+			request.Hooks.OnExecuteInit("session-1")
+			request.Hooks.OnExecuteComplete(time.Millisecond)
+			return nil
+		},
+	}
+	flag.ApiGracefulShutdownTimeout = 50 * time.Millisecond
+	t.Cleanup(func() {
+		codeRunner = previousRunner
+		flag.ApiGracefulShutdownTimeout = previousTimeout
+	})
+
+	body := []byte(`{"code":"print(1)","context":{"id":"ctx-1","language":"python"}}`)
+	ctx, w := newTestContext(http.MethodPost, "/code/run", body)
+	ctrl := NewCodeInterpretingController(ctx)
+
+	ctrl.RunCode()
+
+	require.Equal(t, http.StatusOK, w.Code)
+	require.Contains(t, w.Header().Get("Content-Type"), "text/event-stream")
+	require.NotEmpty(t, w.Body.Bytes(), "successful run should write SSE events")
+}

components/execd/pkg/web/controller/command.go

@@ -80,7 +80,9 @@ func (c *CodeInterpretingController) RunCommand() {
 	}
 	runCodeRequest.Hooks = eventsHandler
 
-	c.setupSSEResponse()
+	// SSE headers are committed lazily on the first event write
+	// (see writeSingleEvent), so a synchronous error from Execute below can
+	// still be surfaced as a structured JSON error response.
 	err = codeRunner.Execute(runCodeRequest)
 	if err != nil {
 		recordExecution("failure")

components/execd/pkg/web/controller/sse.go

@@ -36,13 +36,21 @@ var sseHeaders = map[string]string{
 	"X-Accel-Buffering": "no",
 }
 
+// setupSSEResponse is idempotent: once headers are committed, subsequent calls
+// no-op. Callers that need the headers up front (e.g. long-running streaming
+// endpoints with no early-error path) can call it explicitly. Endpoints that
+// may fail synchronously before any event fires should leave header commit to
+// the lazy path inside writeSingleEvent so pre-execution errors can return a
+// proper JSON body instead of a half-formed text/event-stream response.
 func (c *basicController) setupSSEResponse() {
-	for key, value := range sseHeaders {
-		c.ctx.Writer.Header().Set(key, value)
-	}
-	if flusher, ok := c.ctx.Writer.(http.Flusher); ok {
-		flusher.Flush()
-	}
+	c.sseSetupOnce.Do(func() {
+		for key, value := range sseHeaders {
+			c.ctx.Writer.Header().Set(key, value)
+		}
+		if flusher, ok := c.ctx.Writer.(http.Flusher); ok {
+			flusher.Flush()
+		}
+	})
 }
 
 // setServerEventsHandler adapts runtime callbacks to SSE events.
@@ -167,6 +175,10 @@ func (c *CodeInterpretingController) writeSingleEvent(handler string, data []byt
 
 	c.chunkWriter.Lock()
 	defer c.chunkWriter.Unlock()
+	// Lazily commit SSE response headers on the first event. This lets the
+	// surrounding handler return a proper JSON error via RespondError if the
+	// runtime fails synchronously before any event fires.
+	c.setupSSEResponse()
 	defer func() {
 		if flusher, ok := c.ctx.Writer.(http.Flusher); ok {
 			flusher.Flush()

tests/csharp/OpenSandbox.E2ETests/SandboxE2ETests.cs

@@ -163,7 +163,7 @@ public async Task Sandbox_Create_With_NetworkPolicy_Get_And_Patch_Egress()
 
         try
         {
-            await Task.Delay(5000);
+            await WaitUntilEgressBlocksAsync(policySandbox, "https://www.github.com", TimeSpan.FromSeconds(30));
 
             var initialPolicy = await policySandbox.GetEgressPolicyAsync();
             Assert.NotNull(initialPolicy);
@@ -184,7 +184,7 @@ await policySandbox.PatchEgressRulesAsync(new List<NetworkRule>
                 new() { Action = NetworkRuleAction.Allow, Target = "www.github.com" },
                 new() { Action = NetworkRuleAction.Deny, Target = "pypi.org" }
             });
-            await Task.Delay(2000);
+            await WaitUntilEgressBlocksAsync(policySandbox, "https://pypi.org", TimeSpan.FromSeconds(30));
 
             var patchedPolicy = await policySandbox.GetEgressPolicyAsync();
             Assert.NotNull(patchedPolicy.Egress);
@@ -233,7 +233,7 @@ public async Task Sandbox_Create_With_NetworkPolicy_Get_And_Patch_Egress_Via_Ser
 
         try
         {
-            await Task.Delay(5000);
+            await WaitUntilEgressBlocksAsync(policySandbox, "https://www.github.com", TimeSpan.FromSeconds(30));
 
             var egressEndpoint = await policySandbox.GetEndpointAsync(Constants.DefaultEgressPort);
             Assert.Contains(
@@ -259,7 +259,7 @@ await policySandbox.PatchEgressRulesAsync(new List<NetworkRule>
                 new() { Action = NetworkRuleAction.Allow, Target = "www.github.com" },
                 new() { Action = NetworkRuleAction.Deny, Target = "pypi.org" }
             });
-            await Task.Delay(2000);
+            await WaitUntilEgressBlocksAsync(policySandbox, "https://pypi.org", TimeSpan.FromSeconds(30));
 
             var patchedPolicy = await policySandbox.GetEgressPolicyAsync();
             Assert.NotNull(patchedPolicy.Egress);
@@ -1112,6 +1112,35 @@ private static async Task<Execution> RunWithRetryAsync(Sandbox sandbox, string c
         }
         return result!;
     }
+
+    /// <summary>
+    /// Polls curl against <paramref name="url"/> until the egress sidecar blocks
+    /// it (Execution.Error becomes non-null), or the timeout elapses. NetworkPolicy
+    /// sidecars sometimes accept connections before iptables/proxy rules apply,
+    /// so a fixed sleep is flaky.
+    /// </summary>
+    private static async Task WaitUntilEgressBlocksAsync(Sandbox sandbox, string url, TimeSpan timeout)
+    {
+        var deadline = DateTime.UtcNow + timeout;
+        Execution? last = null;
+        while (DateTime.UtcNow < deadline)
+        {
+            try
+            {
+                last = await sandbox.Commands.RunAsync($"curl -I {url}");
+                if (last?.Error != null)
+                {
+                    return;
+                }
+            }
+            catch
+            {
+                // Transient SDK/SSE errors during sidecar warmup — keep polling.
+            }
+            await Task.Delay(500);
+        }
+        Assert.Fail($"Egress policy did not block {url} within {timeout} (last error={last?.Error?.ToString() ?? "null"})");
+    }
 }
 
 public sealed class SandboxE2ETestFixture : IAsyncLifetime

tests/go/e2e_test.go

@@ -117,8 +117,16 @@ func TestE2E_FullLifecycle(t *testing.T) {
 	}
 	execClient := opensandbox.NewExecdClient(execdURL, execToken)
 
-	err = execClient.Ping(ctx)
-	require.NoError(t, err)
+	// This test bypasses the SDK's high-level CreateSandbox helper (which calls
+	// WaitUntilReady) and pings execd directly through the server-side proxy.
+	// The state-Running flag is satisfied as soon as the container is up, but
+	// execd's HTTP routes may register a few ms later and the proxy can drop
+	// the very first connection it sees ("connection reset by peer"). Poll
+	// until ping succeeds — real users go through CreateSandbox which already
+	// handles this.
+	require.Eventually(t, func() bool {
+		return execClient.Ping(ctx) == nil
+	}, 30*time.Second, 500*time.Millisecond, "execd ping never succeeded")
 	t.Log("Execd ping: OK")
 
 	// 6. Test Execd — run a command with SSE streaming
@@ -131,7 +139,6 @@ func TestE2E_FullLifecycle(t *testing.T) {
 		return nil
 	})
 	require.NoError(t, err)
-	t.Logf("Command raw output (%d bytes): %q", output.Len(), output.String())
 
 	// 7. Test Execd — file operations
 	fileInfoMap, err := execClient.GetFileInfo(ctx, "/etc/os-release")

tests/java/src/test/java/com/alibaba/opensandbox/e2e/SandboxE2ETest.java

@@ -270,11 +270,10 @@ void testSandboxCreateWithNetworkPolicy() {
                         .readyTimeout(Duration.ofSeconds(60))
                         .networkPolicy(networkPolicy)
                         .build();
-        // Wait for NetworkPolicy sidecar to be fully initialized
-        try {
-            Thread.sleep(2000);
-        } catch (InterruptedException ignored) {
-        }
+        // Wait for NetworkPolicy sidecar to be fully initialized.
+        // The sidecar may accept the sandbox before iptables/proxy rules apply,
+        // so poll a denied target until the policy actually blocks it.
+        waitUntilEgressBlocks(policySandbox, "https://www.github.com", Duration.ofSeconds(30));
 
         try {
             NetworkPolicy initialPolicy = policySandbox.getEgressPolicy();
@@ -319,10 +318,8 @@ void testSandboxCreateWithNetworkPolicy() {
                                     .target("pypi.org")
                                     .build()));
 
-            try {
-                Thread.sleep(2000);
-            } catch (InterruptedException ignored) {
-            }
+            // Poll until the patched rule takes effect (pypi now blocked).
+            waitUntilEgressBlocks(policySandbox, "https://pypi.org", Duration.ofSeconds(30));
 
             NetworkPolicy patchedPolicy = policySandbox.getEgressPolicy();
             assertNotNull(patchedPolicy);
@@ -393,10 +390,8 @@ void testSandboxCreateWithNetworkPolicyViaServerProxy() {
                         .readyTimeout(Duration.ofSeconds(60))
                         .networkPolicy(networkPolicy)
                         .build();
-        try {
-            Thread.sleep(2000);
-        } catch (InterruptedException ignored) {
-        }
+        // Wait for NetworkPolicy sidecar/iptables rules to be active.
+        waitUntilEgressBlocks(policySandbox, "https://www.github.com", Duration.ofSeconds(30));
 
         try {
             SandboxEndpoint egressEndpoint = policySandbox.getEndpoint(18080);
@@ -447,10 +442,8 @@ void testSandboxCreateWithNetworkPolicyViaServerProxy() {
                                     .target("pypi.org")
                                     .build()));
 
-            try {
-                Thread.sleep(2000);
-            } catch (InterruptedException ignored) {
-            }
+            // Poll until patched rule applied (pypi now blocked).
+            waitUntilEgressBlocks(policySandbox, "https://pypi.org", Duration.ofSeconds(30));
 
             NetworkPolicy patchedPolicy = policySandbox.getEgressPolicy();
             assertNotNull(patchedPolicy.getEgress());
@@ -1597,4 +1590,33 @@ private Execution runWithRetry(Sandbox sandbox, String command, int maxAttempts,
         }
         return result;
     }
+
+    /**
+     * Polls the sandbox running curl until the given URL is blocked by the
+     * network policy. Returns once curl reports an error (egress active), or
+     * fails the test if the timeout elapses.
+     */
+    private void waitUntilEgressBlocks(Sandbox sandbox, String url, Duration timeout) {
+        long deadline = System.currentTimeMillis() + timeout.toMillis();
+        Execution last = null;
+        while (System.currentTimeMillis() < deadline) {
+            try {
+                last = sandbox.commands().run(
+                        RunCommandRequest.builder().command("curl -I " + url).build());
+                if (last != null && last.getError() != null) {
+                    return;
+                }
+            } catch (Exception ignored) {
+                // Transient SDK/SSE errors during sidecar warmup — keep polling.
+            }
+            try {
+                Thread.sleep(500);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            }
+        }
+        fail("Egress policy did not block " + url + " within " + timeout
+                + " (last execution error=" + (last == null ? "null" : last.getError()) + ")");
+    }
 }