fix(python-sdk): log the local cause when a command stream is cancelled

.changeset/command-stream-cancel-logging.md

@@ -0,0 +1,6 @@
+---
+"e2b": patch
+"@e2b/python-sdk": patch
+---
+
+Log the local cause when a command output stream is cancelled. `CommandHandle` (sync and async, in both the Python and JS SDKs) now records, via the logger the sandbox was constructed with, why the stream ended — explicit `disconnect()`, an early stop of iteration/cancellation, or a stream error before the end event — tagged with the command pid, at most once per handle. This is a no-op when no logger is configured, so behavior is unchanged for callers that did not opt into logging. It gives the client-side reason to correlate with the server-side "stream end" line (e2b-dev/E2B#1877, e2b-dev/runtime#3647).

packages/js-sdk/src/sandbox/commands/commandHandle.ts

@@ -4,6 +4,7 @@ import {
 } from '../../envd/rpc'
 import { SandboxError } from '../../errors'
 import { ConnectResponse, StartResponse } from '../../envd/process/process_pb'
+import { Logger } from '../../logs'
 import type { CommandRequestOpts } from '.'
 
 declare const __brand: unique symbol
@@ -121,11 +122,31 @@ export class CommandHandle
     private readonly handleCloseStdin?: (
       opts?: CommandRequestOpts
     ) => Promise<void>,
-    private readonly checkHealth?: SandboxHealthCheck
+    private readonly checkHealth?: SandboxHealthCheck,
+    private readonly logger?: Logger
   ) {
     this._wait = this.handleEvents()
   }
 
+  private streamEndLogged = false
+
+  /**
+   * Record why the command's output stream ended on the client side.
+   *
+   * The server (envd) only sees that the stream was cancelled, not why; this
+   * logs the local cause so the two can be correlated. No-op when no logger was
+   * configured. Records at most once per handle so the disconnect() and
+   * stream-error paths cannot emit two contradictory causes for the same
+   * command. See e2b-dev/E2B#1877.
+   */
+  private logStreamEnded(reason: string) {
+    if (!this.logger?.info || this.streamEndLogged) {
+      return
+    }
+    this.streamEndLogged = true
+    this.logger.info(`command stream ended (pid=${this.pid}): ${reason}`)
+  }
+
   /**
    * Command execution exit code.
    * `0` if the command finished successfully.
@@ -193,6 +214,7 @@ export class CommandHandle
    * whose stream produces no further output.
    */
   async disconnect() {
+    this.logStreamEnded('explicit disconnect() (command left running)')
     this.disconnected = true
     this.handleDisconnect()
   }
@@ -328,6 +350,9 @@ export class CommandHandle
       // failure). Flush any bytes still buffered in the decoders so incomplete
       // trailing sequences surface as replacement characters instead of being
       // silently dropped, then re-raise so the error is still surfaced.
+      this.logStreamEnded(
+        `stream error before end event: ${e instanceof Error ? `${e.name}: ${e.message}` : String(e)}`
+      )
       yield* this.flushDecoders()
       throw e
     }

packages/js-sdk/src/sandbox/commands/index.ts

@@ -37,9 +37,8 @@ export { Pty } from './pty'
 /**
  * Options for sending a command request.
  */
-export interface CommandRequestOpts extends Partial<
-  Pick<ConnectionOpts, 'requestTimeoutMs' | 'signal'>
-> {}
+export interface CommandRequestOpts
+  extends Partial<Pick<ConnectionOpts, 'requestTimeoutMs' | 'signal'>> {}
 
 /**
  * Options for starting a new command.
@@ -364,7 +363,8 @@ export class Commands {
         undefined,
         (data, stdinOpts) => this.sendStdin(pid, data, stdinOpts),
         (stdinOpts) => this.closeStdin(pid, stdinOpts),
-        this.checkHealth
+        this.checkHealth,
+        this.connectionConfig.logger
       )
     } catch (err) {
       cleanup()
@@ -479,7 +479,8 @@ export class Commands {
         undefined,
         (data, stdinOpts) => this.sendStdin(pid, data, stdinOpts),
         (stdinOpts) => this.closeStdin(pid, stdinOpts),
-        this.checkHealth
+        this.checkHealth,
+        this.connectionConfig.logger
       )
     } catch (err) {
       cleanup()

packages/js-sdk/src/sandbox/commands/pty.ts

@@ -30,10 +30,8 @@ import {
   handleProcessStartEvent,
 } from '../../envd/api'
 
-export interface PtyCreateOpts extends Pick<
-  ConnectionOpts,
-  'requestTimeoutMs' | 'signal'
-> {
+export interface PtyCreateOpts
+  extends Pick<ConnectionOpts, 'requestTimeoutMs' | 'signal'> {
   /**
    * Number of columns for the PTY.
    */
@@ -157,7 +155,8 @@ export class Pty {
         opts.onData,
         undefined,
         undefined,
-        this.checkHealth
+        this.checkHealth,
+        this.connectionConfig.logger
       )
     } catch (err) {
       cleanup()
@@ -214,7 +213,8 @@ export class Pty {
         opts?.onData,
         undefined,
         undefined,
-        this.checkHealth
+        this.checkHealth,
+        this.connectionConfig.logger
       )
     } catch (err) {
       cleanup()

packages/python-sdk/e2b/sandbox_async/commands/command.py

@@ -334,6 +334,7 @@ async def _start(
                     pid, request_timeout
                 ),
                 check_health=self._check_health,
+                logger=self._connection_config.logger,
             )
         except Exception as e:
             try:
@@ -393,6 +394,7 @@ async def connect(
                     pid, request_timeout
                 ),
                 check_health=self._check_health,
+                logger=self._connection_config.logger,
             )
         except Exception as e:
             try:

packages/python-sdk/e2b/sandbox_async/commands/command_handle.py

@@ -1,6 +1,7 @@
 import asyncio
 import codecs
 import inspect
+import logging
 from typing import (
     Optional,
     Callable,
@@ -95,13 +96,17 @@ def __init__(
             Callable[[Optional[float]], Coroutine[Any, Any, None]]
         ] = None,
         check_health: Optional[Callable[[], Awaitable[Optional[bool]]]] = None,
+        *,
+        logger: Optional[logging.Logger] = None,
     ):
         self._pid = pid
         self._handle_kill = handle_kill
         self._handle_send_stdin = handle_send_stdin
         self._handle_close_stdin = handle_close_stdin
         self._check_health = check_health
         self._events = events
+        self._logger = logger
+        self._stream_end_logged = False
 
         self._stdout_chunks: List[str] = []
         self._stderr_chunks: List[str] = []
@@ -138,6 +143,21 @@ def _flush_decoders(
             events.append((None, err, None))
         return events
 
+    def _log_stream_ended(self, reason: str) -> None:
+        """
+        Record why the command's output stream ended on the client side.
+
+        The server (envd) only sees that the stream was cancelled, not why; this
+        logs the local cause so the two can be correlated. No-op when no logger
+        was configured. Records at most once per handle so the disconnect() and
+        reader-task paths cannot emit two contradictory causes for the same
+        command.
+        """
+        if self._logger is None or self._stream_end_logged:
+            return
+        self._stream_end_logged = True
+        self._logger.info("command stream ended (pid=%s): %s", self._pid, reason)
+
     async def _iterate_events(
         self,
     ) -> AsyncGenerator[
@@ -209,6 +229,7 @@ async def disconnect(self) -> None:
         The command is not killed, but SDK stops receiving events from the command.
         You can reconnect to the command using `sandbox.commands.connect` method.
         """
+        self._log_stream_ended("explicit disconnect() (command left running)")
         self._wait.cancel()
         await asyncio.wait([self._wait])
         try:
@@ -234,6 +255,9 @@ async def _handle_events(self):
         except StopAsyncIteration:
             pass
         except Exception as e:
+            self._log_stream_ended(
+                f"stream error before end event: {type(e).__name__}: {e}"
+            )
             self._iteration_exception = await ahandle_rpc_exception_with_health(
                 e, self._check_health
             )

packages/python-sdk/e2b/sandbox_async/commands/pty.py

@@ -172,6 +172,7 @@ async def create(
                 events=events,
                 on_pty=on_data,
                 check_health=self._check_health,
+                logger=self._connection_config.logger,
             )
         except Exception as e:
             try:
@@ -221,6 +222,7 @@ async def connect(
                 events=events,
                 on_pty=on_data,
                 check_health=self._check_health,
+                logger=self._connection_config.logger,
             )
         except Exception as e:
             try:

packages/python-sdk/e2b/sandbox_sync/commands/command.py

@@ -330,6 +330,7 @@ def _start(
                     pid, request_timeout
                 ),
                 check_health=self._check_health,
+                logger=self._connection_config.logger,
             )
         except Exception as e:
             try:
@@ -381,6 +382,7 @@ def connect(
                     pid, request_timeout
                 ),
                 check_health=self._check_health,
+                logger=self._connection_config.logger,
             )
         except Exception as e:
             try:

packages/python-sdk/e2b/sandbox_sync/commands/command_handle.py

@@ -1,4 +1,5 @@
 import codecs
+import logging
 
 from typing import Optional, Callable, Any, Generator, List, Union, Tuple
 
@@ -42,13 +43,17 @@ def __init__(
         ] = None,
         handle_close_stdin: Optional[Callable[[Optional[float]], None]] = None,
         check_health: Optional[Callable[[], Optional[bool]]] = None,
+        *,
+        logger: Optional[logging.Logger] = None,
     ):
         self._pid = pid
         self._handle_kill = handle_kill
         self._handle_send_stdin = handle_send_stdin
         self._handle_close_stdin = handle_close_stdin
         self._check_health = check_health
         self._events = events
+        self._logger = logger
+        self._stream_end_logged = False
 
         self._stdout_chunks: List[str] = []
         self._stderr_chunks: List[str] = []
@@ -67,6 +72,21 @@ def __iter__(self):
         """
         return self._handle_events()
 
+    def _log_stream_ended(self, reason: str) -> None:
+        """
+        Record why the command's output stream ended on the client side.
+
+        The server (envd) only sees that the stream was cancelled, not why; this
+        logs the local cause so the two can be correlated. No-op when no logger
+        was configured. Records at most once per handle so the disconnect() and
+        generator-close paths cannot emit two contradictory causes for the same
+        command.
+        """
+        if self._logger is None or self._stream_end_logged:
+            return
+        self._stream_end_logged = True
+        self._logger.info("command stream ended (pid=%s): %s", self._pid, reason)
+
     def _flush_decoders(
         self,
     ) -> List[Union[Tuple[Stdout, None, None], Tuple[None, Stderr, None]]]:
@@ -140,11 +160,24 @@ def _handle_events(
             # characters instead of being silently dropped.
             if self._result is None:
                 yield from self._flush_decoders()
+        except GeneratorExit:
+            # The consumer stopped iterating before an end event (e.g. it broke
+            # out of the loop, was cancelled by a timeout/sibling task, or the
+            # caller exited). Closing this generator closes the underlying event
+            # stream, which the server (envd) records as a client cancellation —
+            # note the process itself is NOT killed and keeps running. Log the
+            # local cause so it can be correlated with the server-side line;
+            # see e2b-dev/E2B#1877.
+            self._log_stream_ended("consumer stopped iterating (break/cancel/exit)")
+            raise
         except Exception as e:
             # The stream raised before an end event (e.g. disconnect or RPC
             # failure). Flush any bytes still buffered in the decoders so
             # incomplete trailing sequences surface as replacement characters
             # instead of being silently dropped, then surface the error.
+            self._log_stream_ended(
+                f"stream error before end event: {type(e).__name__}: {e}"
+            )
             yield from self._flush_decoders()
             raise handle_rpc_exception_with_health(e, self._check_health)
 
@@ -155,6 +188,7 @@ def disconnect(self) -> None:
         The command is not killed, but SDK stops receiving events from the command.
         You can reconnect to the command using `sandbox.commands.connect` method.
         """
+        self._log_stream_ended("explicit disconnect() (command left running)")
         self._events.close()
 
     def wait(

packages/python-sdk/e2b/sandbox_sync/commands/pty.py

@@ -163,6 +163,7 @@ def create(
                 handle_kill=lambda: self.kill(pid),
                 events=events,
                 check_health=self._check_health,
+                logger=self._connection_config.logger,
             )
         except Exception as e:
             try:
@@ -207,6 +208,7 @@ def connect(
                 handle_kill=lambda: self.kill(pid),
                 events=events,
                 check_health=self._check_health,
+                logger=self._connection_config.logger,
             )
         except Exception as e:
             try: