fix(code-runtime-python): meter stdout and stderr stray residual against one shared budget

stdout and stderr each checked their pending serialized cost against the full
logBudget independently, so both could retain nearly a budget's worth of
newline-free residual at once — double the intended peak, up to ~512 MiB near
the ceiling. The flush threshold now reads the COMBINED cost of both pipes and
flushes both when it crosses, since they share one ledger.

Remove the post-truncation admit() v8-ignore: captureStray's per-line loop
makes that branch deterministically reachable within one data callback (a chunk
whose first newline-terminated line exhausts the budget hits it on the second),
so it is measured by a new regression test rather than ignored.

Refresh two stray-output test comments that still named the removed
StringDecoder; the raw-chunk buffer reassembles a split multibyte sequence by
concatenating before it decodes, and the end flush renders a stranded partial
as U+FFFD via toString('utf8').

packages/code-runtime/code-runtime-python/src/index.ts

@@ -767,7 +767,10 @@ export class PythonCodeRuntime extends CodeRuntime {
       let logBudget = this.config.maxLogBytes
       let logsTruncated = false
       const admit = (text: string): void => {
-        /* v8 ignore next -- post-truncation admits no-op; needs child to keep streaming after ledger drops. */
+        // Post-truncation admits are no-ops: once the ledger has truncated, the
+        // marker is the last entry. Reachable within one `data` callback — a
+        // chunk carrying two newline-terminated lines where the first exhausts
+        // the budget hits this on the second — so it is a measured branch.
         if (logsTruncated) return
         // Each entry is charged its SERIALIZED cost — JSON.stringify's quotes
         // and escapes plus one separator byte — because the seam bounds the
@@ -872,22 +875,29 @@ export class PythonCodeRuntime extends CodeRuntime {
         }
         // Newline-free residual is bounded by the ledger, not left to grow with
         // the stream: an `os.write(1, b"A"*N)` flood carrying no newline would
-        // otherwise accumulate N bytes in host memory before `end`. When the
-        // pending residual's serialized cost would cross the budget, flush it now
-        // — admit() charges the exact serialized cost, truncates, and marks the
-        // ledger, and the truncation short-circuit above stops buffering on the
-        // next chunk. `+ 3` covers the two quotes and one separator admit adds.
-        if (stray.cost + 3 > logBudget) {
-          flushStray(stray)
+        // otherwise accumulate N bytes in host memory before `end`. The bound is
+        // on the COMBINED pending cost of both pipes, not each alone: stdout and
+        // stderr share one `logBudget`, so checking each against the full budget
+        // independently would let both retain nearly a budget's worth at once —
+        // ~2x peak, up to ~512 MiB near the ceiling — before either flushed.
+        // When the sum would cross the budget, flush both now. admit() charges
+        // the exact serialized cost, truncates, and marks the ledger, and the
+        // truncation short-circuit above stops buffering on the next chunk.
+        // `+ 3` covers the two quotes and one separator admit adds.
+        if (strayOut.cost + strayErr.cost + 3 > logBudget) {
+          flushStray(strayOut)
+          flushStray(strayErr)
         }
       }
-      // Flush a pipe's residual into `logs`. Called on the budget threshold
-      // above, on the pipe's `end` (normal drain), and — for the setsid-escapee
-      // path where destroy() forces settlement without an `end` — explicitly in
-      // the closeDeadline handler. Idempotent: it clears what it admits, so a
-      // later flush is a no-op. The `chunks`/`blocks` guard is the only emptiness
-      // check needed — `data` never emits a zero-length Buffer, so a non-empty
-      // fragment list always decodes to a non-empty tail.
+      // Flush a pipe's residual into `logs`. Called on the combined-budget
+      // threshold above, on the pipe's `end` (normal drain), and — for the
+      // setsid-escapee path where destroy() forces settlement without an `end` —
+      // explicitly in the closeDeadline handler. Idempotent: it clears what it
+      // admits, so a later flush is a no-op, and it returns early on an empty
+      // buffer so flushing the sibling that had nothing pending is a no-op. The
+      // `chunks`/`blocks` guard is the only emptiness check needed — `data` never
+      // emits a zero-length Buffer, so a non-empty fragment list always decodes
+      // to a non-empty tail.
       function flushStray(stray: StrayBuffer): void {
         if (stray.chunks.length === 0 && stray.blocks.length === 0) return
         const tail = Buffer.concat([...stray.blocks, ...stray.chunks]).toString('utf8')

packages/code-runtime/code-runtime-python/tests/runtime.spec.ts

@@ -774,6 +774,22 @@ describe('PythonCodeRuntime — programs and bindings', () => {
     expect(result.logs.at(-1)).toBe(logTruncationMarker(4096))
   })
 
+  it('drops a second stray line in the same chunk once the first truncated the ledger', async () => {
+    // One `os.write` carrying two newline-terminated lines where the first
+    // exhausts maxLogBytes: the first line's admit truncates and marks the
+    // ledger, and the second line's admit — reached in the same `data` callback
+    // — must be the post-truncation no-op. Proves that branch is exercised, so
+    // it carries no v8-ignore.
+    const { runtime } = await setup({ maxLogBytes: 64 })
+    const result = await runtime.run({
+      program: ['import os', 'os.write(1, b"A" * 5000 + b"\\nSECOND\\n")', 'return None'].join('\n'),
+      bindings: [],
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.logs.at(-1)).toBe(logTruncationMarker(64))
+    expect(result.logs.join('\n')).not.toContain('SECOND')
+  })
+
   it('charges the exact serialized cost of short-escape and quote/backslash characters', async () => {
     // Exercises every branch of jsonStringCostUpTo's per-character cost: a tab
     // and other C0 controls with short JSON forms (\t etc., 2 bytes), a quote
@@ -3440,7 +3456,9 @@ describe('PythonCodeRuntime — hostile peer', () => {
     // A single os.write far past the 64 KiB pipe buffer forces multiple
     // 'data' chunks; when the boundary lands inside the emoji's 4-byte
     // sequence, per-chunk decoding would corrupt it into replacement
-    // characters. The streaming decoder must reassemble it.
+    // characters. Raw bytes are buffered and only decoded once a complete line
+    // (or the whole tail at flush) is assembled, so the split sequence is whole
+    // by the time it is decoded.
     const { runtime } = await setup({ maxLogBytes: 1024 * 1024 })
     const result = await runtime.run({
       program: [
@@ -3464,9 +3482,9 @@ describe('PythonCodeRuntime — hostile peer', () => {
 
   it('flushes a stray-output byte sequence left incomplete when the pipe ends', async () => {
     // The child writes the first two bytes of a 3-byte UTF-8 character to fd 1
-    // and exits, so the pipe closes with the sequence unfinished inside the
-    // streaming decoder. The 'end' flush must render the stranded bytes as
-    // U+FFFD instead of dropping the evidence with the decoder.
+    // and exits, so the pipe closes with the sequence unfinished in the raw
+    // residual. The 'end' flush decodes the residual with `toString('utf8')`,
+    // which renders the stranded bytes as U+FFFD instead of dropping them.
     const { runtime } = await setup({ maxLogBytes: 1024 * 1024 })
     const result = await runtime.run({
       program: [