fix for streaming

Author: whoiskatrin

cf-sandbox-bug

@@ -1227,6 +1227,8 @@ export class Sandbox<Env = unknown> extends Container<Env> implements ISandbox {
       // Command execution - delegate to internal session-aware methods
       exec: (command, options) => this.execWithSession(command, sessionId, options),
       execStream: (command, options) => this.execStreamWithSession(command, sessionId, options),
+      execStreamWithCallback: (command, onEvent, options) =>
+        this.execStreamWithCallback(command, onEvent, { ...options, sessionId }),
 
       // Process management
       startProcess: (command, options) => this.startProcess(command, options, sessionId),
@@ -1237,6 +1239,8 @@ export class Sandbox<Env = unknown> extends Container<Env> implements ISandbox {
       cleanupCompletedProcesses: () => this.cleanupCompletedProcesses(),
       getProcessLogs: (id) => this.getProcessLogs(id),
       streamProcessLogs: (processId, options) => this.streamProcessLogs(processId, options),
+      streamProcessLogsWithCallback: (processId, onEvent, options) =>
+        this.streamProcessLogsWithCallback(processId, onEvent, options),
 
       // File operations - pass sessionId via options or parameter
       writeFile: (path, content, options) => this.writeFile(path, content, { ...options, sessionId }),

packages/sandbox/src/sandbox.ts

@@ -583,7 +583,14 @@ export interface ExecutionSession {
   // Command execution
   exec(command: string, options?: ExecOptions): Promise<ExecResult>;
   execStream(command: string, options?: StreamOptions): Promise<ReadableStream<Uint8Array>>;
-  
+
+  // Callback-based streaming 
+  execStreamWithCallback(
+    command: string,
+    onEvent: (event: ExecEvent) => void | Promise<void>,
+    options?: { signal?: AbortSignal }
+  ): Promise<void>;
+
   // Background process management
   startProcess(command: string, options?: ProcessOptions): Promise<Process>;
   listProcesses(): Promise<Process[]>;
@@ -593,6 +600,11 @@ export interface ExecutionSession {
   cleanupCompletedProcesses(): Promise<number>;
   getProcessLogs(id: string): Promise<{ stdout: string; stderr: string; processId: string }>;
   streamProcessLogs(processId: string, options?: { signal?: AbortSignal }): Promise<ReadableStream<Uint8Array>>;
+  streamProcessLogsWithCallback(
+    processId: string,
+    onEvent: (event: ExecEvent) => void | Promise<void>,
+    options?: { signal?: AbortSignal }
+  ): Promise<void>;
 
   // File operations
   writeFile(path: string, content: string, options?: { encoding?: string }): Promise<WriteFileResult>;

packages/shared/src/types.ts

@@ -593,7 +593,7 @@ describe('Streaming Operations Workflow', () => {
 
       // With callback-based streaming, it should complete successfully
       const streamResponse = await vi.waitFor(
-        async () => fetchWithStartup(`${workerUrl}/api/execStream`, {
+        async () => fetchWithStartup(`${workerUrl}/api/execStreamCallback`, {
           method: 'POST',
           headers,
           body: JSON.stringify({
@@ -641,7 +641,7 @@ describe('Streaming Operations Workflow', () => {
 
       // This is the exact pattern that was failing before
       const streamResponse = await vi.waitFor(
-        async () => fetchWithStartup(`${workerUrl}/api/execStream`, {
+        async () => fetchWithStartup(`${workerUrl}/api/execStreamCallback`, {
           method: 'POST',
           headers,
           body: JSON.stringify({

tests/e2e/streaming-operations-workflow.test.ts

@@ -86,6 +86,51 @@ export default {
         });
       }
 
+      // Command execution with callback-based streaming (for long-running commands)
+      if (url.pathname === '/api/execStreamCallback' && request.method === 'POST') {
+        console.log('[TestWorker] execStreamCallback called for command:', body.command);
+        const startTime = Date.now();
+
+        // Create a TransformStream to convert callbacks to SSE
+        const { readable, writable } = new TransformStream();
+        const writer = writable.getWriter();
+        const encoder = new TextEncoder();
+
+        // Start streaming in the background using callback-based method
+        (async () => {
+          try {
+            await executor.execStreamWithCallback(
+              body.command,
+              async (event) => {
+                // Convert event to SSE format
+                const sseData = `data: ${JSON.stringify(event)}\n\n`;
+                await writer.write(encoder.encode(sseData));
+              }
+            );
+          } catch (error: any) {
+            // Send error event
+            const errorEvent = {
+              type: 'error',
+              timestamp: new Date().toISOString(),
+              error: error.message,
+            };
+            await writer.write(encoder.encode(`data: ${JSON.stringify(errorEvent)}\n\n`));
+          } finally {
+            await writer.close();
+            console.log('[TestWorker] Stream completed in', Date.now() - startTime, 'ms');
+          }
+        })();
+
+        // Return SSE stream
+        return new Response(readable, {
+          headers: {
+            'Content-Type': 'text/event-stream',
+            'Cache-Control': 'no-cache',
+            'Connection': 'keep-alive',
+          },
+        });
+      }
+
       // Git clone
       if (url.pathname === '/api/git/clone' && request.method === 'POST') {
         await executor.gitCheckout(body.repoUrl, {