fix(sdk): correct command/PTY stream handling in Python and JS SDKs (#1441)

## Summary

Fixes three command/PTY streaming issues in the Python and JS SDKs:

- **Multibyte UTF-8 corruption (JS + Python sync/async):** stdout/stderr
were decoded per-chunk, so a UTF-8 character split across two stream
chunks turned into replacement characters. Each handle now keeps a
persistent incremental decoder per stream
(`codecs.getincrementaldecoder` in Python, a shared `TextDecoder` with
`{ stream: true }` in JS) and flushes any incomplete trailing bytes to
`�` on the end event, preserving the existing broken-UTF-8 behavior.
- **`commands.list()` optionals (Python):** now returns `None` instead
of `""` for unset proto3-optional `tag` and `cwd` fields, matching the
declared `Optional[str]` types and the JS SDK.
- **Leaked connections (Python):** command/PTY/watch streams are now
closed when stream setup fails, instead of abandoning the generator (and
its pooled HTTP connection) until GC.

## Usage example

```python
# Split multibyte output is now decoded correctly instead of returning "ð\x9f\x98\x80"-style garbage
result = sandbox.commands.run("printf '😀'")
assert result.stdout == "😀"

# Unset fields are None rather than ""
proc = sandbox.commands.list()[0]
assert proc.tag is None  # previously ""
```

## Testing

- New unit tests for incremental/trailing UTF-8 decoding (Python sync +
async, JS).
- Live command/PTY/watch integration suites pass.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: mishushakov

.changeset/great-pots-deny.md

@@ -0,0 +1,8 @@
+---
+"e2b": patch
+---
+
+Fix command and PTY streaming issues:
+
+- Decode stdout/stderr incrementally so multibyte UTF-8 characters split across command stream chunks are no longer corrupted
+- Avoid mutating the caller's `envs` object when applying default `TERM`/`LANG`/`LC_ALL` values in `pty.create()`

.changeset/heavy-pugs-smile.md

@@ -0,0 +1,10 @@
+---
+"@e2b/python-sdk": patch
+---
+
+Fix several command and PTY streaming issues:
+
+- Decode stdout/stderr incrementally so multibyte UTF-8 characters split across stream chunks are no longer corrupted
+- Return `None` instead of empty strings for unset `tag` and `cwd` fields in `commands.list()`
+- Close command/PTY/watch stream connections when establishing the stream fails, instead of leaking pooled connections
+- Avoid mutating the caller's `envs` dict when applying default `TERM`/`LANG`/`LC_ALL` values in `pty.create()`

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

@@ -91,6 +91,9 @@ export class CommandHandle
   private _stdout = ''
   private _stderr = ''
 
+  private readonly stdoutDecoder = new TextDecoder()
+  private readonly stderrDecoder = new TextDecoder()
+
   private result?: CommandResult
   private iterationError?: Error
 
@@ -235,41 +238,89 @@ export class CommandHandle
     await this.handleCloseStdin(opts)
   }
 
+  /**
+   * Flush any bytes still buffered in the stream decoders.
+   *
+   * Incomplete trailing UTF-8 sequences are emitted as replacement
+   * characters, matching the per-chunk decoding behavior.
+   */
+  private *flushDecoders(): Generator<
+    [Stdout, null, null] | [null, Stderr, null]
+  > {
+    const stdoutRest = this.stdoutDecoder.decode()
+    if (stdoutRest) {
+      this._stdout += stdoutRest
+      yield [stdoutRest as Stdout, null, null]
+    }
+    const stderrRest = this.stderrDecoder.decode()
+    if (stderrRest) {
+      this._stderr += stderrRest
+      yield [null, stderrRest as Stderr, null]
+    }
+  }
+
   private async *iterateEvents(): AsyncGenerator<
     [Stdout, null, null] | [null, Stderr, null] | [null, null, PtyOutput]
   > {
-    for await (const event of this.events) {
-      const e = event?.event?.event
-      let out: string | undefined
-
-      switch (e?.case) {
-        case 'data':
-          switch (e.value.output.case) {
-            case 'stdout':
-              out = new TextDecoder().decode(e.value.output.value)
-              this._stdout += out
-              yield [out as Stdout, null, null]
-              break
-            case 'stderr':
-              out = new TextDecoder().decode(e.value.output.value)
-              this._stderr += out
-              yield [null, out as Stderr, null]
-              break
-            case 'pty':
-              yield [null, null, e.value.output.value as PtyOutput]
-              break
-          }
-          break
-        case 'end':
-          this.result = {
-            exitCode: e.value.exitCode,
-            error: e.value.error,
-            stdout: this.stdout,
-            stderr: this.stderr,
+    try {
+      for await (const event of this.events) {
+        const e = event?.event?.event
+        let out: string | undefined
+
+        switch (e?.case) {
+          case 'data':
+            switch (e.value.output.case) {
+              case 'stdout':
+                out = this.stdoutDecoder.decode(e.value.output.value, {
+                  stream: true,
+                })
+                if (out) {
+                  this._stdout += out
+                  yield [out as Stdout, null, null]
+                }
+                break
+              case 'stderr':
+                out = this.stderrDecoder.decode(e.value.output.value, {
+                  stream: true,
+                })
+                if (out) {
+                  this._stderr += out
+                  yield [null, out as Stderr, null]
+                }
+                break
+              case 'pty':
+                yield [null, null, e.value.output.value as PtyOutput]
+                break
+            }
+            break
+          case 'end': {
+            yield* this.flushDecoders()
+            this.result = {
+              exitCode: e.value.exitCode,
+              error: e.value.error,
+              stdout: this.stdout,
+              stderr: this.stderr,
+            }
+            break
           }
-          break
+        }
+        // TODO: Handle empty events like in python SDK
       }
-      // TODO: Handle empty events like in python SDK
+    } catch (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 re-raise so the error is still surfaced.
+      yield* this.flushDecoders()
+      throw e
+    }
+
+    // If the stream closed without an `end` event (e.g. disconnect or a
+    // dropped connection), flush any bytes still buffered in the decoders so
+    // incomplete trailing sequences surface as replacement characters instead
+    // of being silently dropped.
+    if (this.result === undefined) {
+      yield* this.flushDecoders()
     }
   }
 

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

@@ -106,7 +106,7 @@ export class Pty {
   async create(opts: PtyCreateOpts) {
     const requestTimeoutMs =
       opts?.requestTimeoutMs ?? this.connectionConfig.requestTimeoutMs
-    const envs = opts?.envs ?? {}
+    const envs = { ...(opts?.envs ?? {}) }
     envs.TERM = envs.TERM ?? 'xterm-256color'
     envs.LANG = envs.LANG ?? 'C.UTF-8'
     envs.LC_ALL = envs.LC_ALL ?? 'C.UTF-8'

packages/js-sdk/tests/sandbox/commands/commandHandle.test.ts

@@ -52,6 +52,36 @@ function createEvents(kind: EventKind): AsyncIterable<any> {
   return events()
 }
 
+function dataEvent(kind: 'stdout' | 'stderr', value: Uint8Array) {
+  return {
+    event: {
+      event: {
+        case: 'data',
+        value: {
+          output: {
+            case: kind,
+            value,
+          },
+        },
+      },
+    },
+  }
+}
+
+function endEvent(exitCode = 0) {
+  return {
+    event: {
+      event: {
+        case: 'end',
+        value: {
+          exitCode,
+          error: undefined,
+        },
+      },
+    },
+  }
+}
+
 describe('CommandHandle', () => {
   it.each<EventKind>(['stdout', 'stderr', 'pty'])(
     'wait awaits async %s callbacks',
@@ -94,4 +124,137 @@ describe('CommandHandle', () => {
       expect(waitResolved).toBe(true)
     }
   )
+
+  it('decodes multibyte characters split across chunks', async () => {
+    const emojiBytes = new TextEncoder().encode('😀')
+
+    async function* events() {
+      yield dataEvent(
+        'stdout',
+        new Uint8Array([
+          ...new TextEncoder().encode('a'),
+          ...emojiBytes.slice(0, 2),
+        ])
+      )
+      yield dataEvent(
+        'stdout',
+        new Uint8Array([
+          ...emojiBytes.slice(2),
+          ...new TextEncoder().encode('b'),
+        ])
+      )
+      yield dataEvent('stderr', emojiBytes.slice(0, 3))
+      yield dataEvent('stderr', emojiBytes.slice(3))
+      yield endEvent()
+    }
+
+    const stdoutChunks: string[] = []
+    const handle = new CommandHandle(
+      1,
+      () => {},
+      async () => true,
+      events(),
+      (out) => {
+        stdoutChunks.push(out)
+      }
+    )
+
+    const result = await handle.wait()
+
+    expect(result.stdout).toBe('a😀b')
+    expect(result.stderr).toBe('😀')
+    expect(result.stdout).not.toContain('�')
+    expect(result.stderr).not.toContain('�')
+    expect(stdoutChunks.join('')).toBe('a😀b')
+  })
+
+  it('replaces incomplete trailing utf-8 sequences at the end of the stream', async () => {
+    const emojiBytes = new TextEncoder().encode('😀')
+
+    async function* events() {
+      yield dataEvent(
+        'stdout',
+        new Uint8Array([
+          ...new TextEncoder().encode('a'),
+          ...emojiBytes.slice(0, 2),
+        ])
+      )
+      yield endEvent()
+    }
+
+    const handle = new CommandHandle(
+      1,
+      () => {},
+      async () => true,
+      events()
+    )
+
+    const result = await handle.wait()
+
+    expect(result.stdout).toBe('a�')
+  })
+
+  it('flushes incomplete trailing utf-8 sequences when the stream closes without an end event', async () => {
+    const emojiBytes = new TextEncoder().encode('😀')
+
+    async function* events() {
+      yield dataEvent(
+        'stdout',
+        new Uint8Array([
+          ...new TextEncoder().encode('a'),
+          ...emojiBytes.slice(0, 2),
+        ])
+      )
+    }
+
+    const stdoutChunks: string[] = []
+    const handle = new CommandHandle(
+      1,
+      () => {},
+      async () => true,
+      events(),
+      (out) => {
+        stdoutChunks.push(out)
+      }
+    )
+
+    // No end event arrives, so wait() rejects, but the buffered bytes must
+    // still be flushed to the stdout callback as a replacement character.
+    await expect(handle.wait()).rejects.toThrow()
+
+    expect(stdoutChunks.join('')).toBe('a�')
+  })
+
+  it('flushes incomplete trailing utf-8 sequences when the stream errors', async () => {
+    const emojiBytes = new TextEncoder().encode('😀')
+
+    async function* events() {
+      yield dataEvent(
+        'stdout',
+        new Uint8Array([
+          ...new TextEncoder().encode('a'),
+          ...emojiBytes.slice(0, 2),
+        ])
+      )
+      throw new Error('stream died')
+    }
+
+    const stdoutChunks: string[] = []
+    const handle = new CommandHandle(
+      1,
+      () => {},
+      async () => true,
+      events(),
+      (out) => {
+        stdoutChunks.push(out)
+      }
+    )
+
+    // The stream errors before an end event arrives, so wait() rejects, but the
+    // buffered bytes must still be flushed to the stdout callback as a
+    // replacement character.
+    await expect(handle.wait()).rejects.toThrow()
+
+    expect(stdoutChunks.join('')).toBe('a�')
+  })
 })

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

@@ -67,11 +67,11 @@ async def list(
             return [
                 ProcessInfo(
                     pid=p.pid,
-                    tag=p.tag,
+                    tag=p.tag if p.HasField("tag") else None,
                     cmd=p.config.cmd,
                     args=list(p.config.args),
                     envs=dict(p.config.envs),
-                    cwd=p.config.cwd,
+                    cwd=p.config.cwd if p.config.HasField("cwd") else None,
                 )
                 for p in res.processes
             ]
@@ -325,6 +325,10 @@ async def _start(
                 check_health=self._check_health,
             )
         except Exception as e:
+            try:
+                await events.aclose()
+            except Exception:
+                pass
             raise await ahandle_rpc_exception_with_health(e, self._check_health)
 
     async def connect(
@@ -384,4 +388,8 @@ async def connect(
                 check_health=self._check_health,
             )
         except Exception as e:
+            try:
+                await events.aclose()
+            except Exception:
+                pass
             raise await ahandle_rpc_exception_with_health(e, self._check_health)