fix(code-runtime-python): close the log-fragment OOM, CPU classification, and done-send transitive-dependency findings

Addresses the bot's v16 review on the settlement-path code:
- critical: _LogStream._pending now seals the fragment list past a chunk cap
  (like the host captureStray seal), so a newline-free single-character drip no
  longer accumulates one list slot per write and OOMs on its own accounting.
- _clamped lowers a soft==hard result by one unit (when hard >= 2) so a
  dual-limit ulimit -t leaves SIGXCPU a window to fire and a definite CPU
  overrun is reported as a timeout, not a worker-exit.
- send_done wraps its encode+write in a try and, on any throw from a rebound
  transitive name (_dump_scalar/os), writes a fixed pre-encoded done frame via
  the import-time captured os.write, so a settled exception verdict is never
  downgraded to worker-exit.
- drainReplies clears the consumed replyQueue slot so a wide written payload is
  released immediately, bounding host memory to the current backlog under
  sustained fd-3 backpressure.
Tests added for each (fragment cap drip, dual-limit CPU overrun, transitive-name
rebind done frame).

packages/code-runtime/code-runtime-python/py/bootstrap.py

@@ -41,6 +41,22 @@ from protocol import PROTOCOL_FD, log_truncation_marker  # noqa: E402
 # simply takes more reads. 64 KiB matches the usual pipe capacity.
 _READ_CHUNK_BYTES = 65536
 
+# Captured primitive for the done-frame LAST-resort fallback. This bootstrap IS
+# ``__main__``, so ``import __main__; __main__.os = ...`` would rebind ``os.write``
+# at call time inside ``ProtocolChannel.write_encoded``. ``os_write`` is a closure
+# cell captured at import, before model code runs, so a one-line rebind cannot
+# change which write the fallback uses. See ``send_done``'s try/except below.
+_os_write = os.write
+
+# A fixed, pre-encoded done frame for the fallback. It carries no live model
+# value, so it can always be written even when a transitive name (a ``_dump_*``
+# helper or ``os``) has been rebound and the normal encode/write threw. The
+# message is the same fixed literal the failure reporter uses for an
+# unrenderable diagnostic; the host renders the run as an exception rather than
+# a worker-exit, which is the honest verdict for a settled run whose reporting
+# was sabotaged. The bytes are JSON-valid and newline-terminated.
+_FALLBACK_DONE_FRAME = b'{"type":"done","error":{"kind":"exception","message":"<unrenderable>"}}\n'
+
 # Code-unit ceiling on the exception class name interpolated into the LAST-resort
 # failure diagnostic. A metaclass `__name__` property can return any length, and
 # that construction runs outside the guard that would otherwise absorb a
@@ -179,6 +195,16 @@ class _LogStream(io.TextIOBase):
         # ``print("x", end="")`` must not concatenate quadratically.
         self._pending: list[str] = []
         self._pending_chars = 0
+        # A newline-free drip must not accumulate one list slot per ``write``:
+        # under a large ``maxLogBytes`` the list-of-fragments pointer array and
+        # the per-fragment str objects cost host memory well before the byte
+        # budget is reached, and a 25 M single-character drip would OOM on its
+        # own accounting (plus the same-size list ``_push_bounded_prefix`` then
+        # builds). Past this many fragments the chunks are sealed into one
+        # joined block (the character count is unchanged), bounding the live
+        # fragment count exactly as the host-side ``captureStray`` does with its
+        # ``MAX_PENDING_CHUNKS``.
+        self._PENDING_MAX_CHUNKS = 1024
 
     def writable(self) -> bool:  # noqa: D401 -- inherited contract
         return True
@@ -292,6 +318,18 @@ class _LogStream(io.TextIOBase):
         else:
             self._pending.append(text)
             self._pending_chars += len(text)
+            # Seal the fragment list past the chunk cap: a newline-free drip
+            # appends one fragment per write, so a 25 M single-character flood
+            # would accumulate that many list slots (and str objects) long before
+            # the byte budget is met — the pointer array alone being ~25 M slots.
+            # Joining the fragments into one block keeps the SAME character count
+            # (`_pending_chars` is unchanged) while bounding the live fragment
+            # count, mirroring the host-side `captureStray` seal. The join is
+            # only as large as the buffered characters, which the budget already
+            # bounds; the fragments are otherwise un-sealable mid-newline because
+            # a newline never starts a multi-byte sequence.
+            if len(self._pending) >= self._PENDING_MAX_CHUNKS:
+                self._pending = ["".join(self._pending)]
         # A newline-free flood must hit the budget while running, not at
         # settlement: once the buffered tail alone can no longer fit the
         # ledger (chars lower-bound the serialized cost), push it through — LogBuffer
@@ -640,7 +678,21 @@ def _clamped(which: int, soft: int, hard: int) -> tuple[int, int]:
     # invert them (a finite inherited soft below the clamped hard is fine, but a
     # requested hard below the inherited soft would leave soft > hard), so pin
     # soft under hard as the final step; the stricter hard ceiling wins.
-    return (min(clamped_soft, clamped_hard), clamped_hard)
+    result_soft = min(clamped_soft, clamped_hard)
+    result_hard = clamped_hard
+    # A soft limit EQUAL to the hard limit leaves the kernel no window to send
+    # SIGXCPU: it checks the hard limit in the same tick and SIGKILLs directly
+    # (a `ulimit -t N` sets both, and a busy loop then dies by SIGKILL, not
+    # SIGXCPU). The host classifies a CPU overrun ONLY on ``signal ===
+    # 'SIGXCPU'``, so a definite budget exhaustion would be misreported as a
+    # `worker-exit`. Lowering the soft limit one unit below the hard (when the
+    # hard is at least 2, so soft stays positive) keeps the stricter-of-the-two
+    # containment semantics while giving SIGXCPU a window to fire — the CPU
+    # overrun is then reported as a timeout, not a worker-exit. For RLIMIT_AS
+    # this is one byte stricter, harmless.
+    if result_soft == result_hard and result_hard >= 2:
+        result_soft = result_hard - 1
+    return (result_soft, result_hard)
 
 
 # ---------------------------------------------------------------------------
@@ -913,10 +965,28 @@ async def _run(channel: ProtocolChannel) -> None:
     write_encoded_bound = channel.write_encoded
 
     def send_done(payload: dict[str, Any] | str) -> None:
-        if isinstance(payload, str):
-            write_encoded_bound(payload)
-        else:
-            write_encoded_bound(encode_plain_bound(payload))
+        try:
+            if isinstance(payload, str):
+                write_encoded_bound(payload)
+            else:
+                write_encoded_bound(encode_plain_bound(payload))
+        except BaseException:  # noqa: BLE001 -- a rebind must not cost the done frame
+            # `encode_plain_bound`/`write_encoded_bound` are bound callables, but
+            # their BODIES still resolve transitive module globals at call time —
+            # `_encode_json_plain` reaches `_dump_scalar`/`_dump_string`/`json.dumps`,
+            # `write_encoded` reaches `os.write` (via the `os` module). This
+            # bootstrap is `__main__`, so `__main__._dump_scalar = boom` (or
+            # `__main__.os = ...`) makes the error-frame encode/write throw AFTER
+            # the `except` block, which would drop the `done` frame and downgrade a
+            # settled `exception` verdict to a host-side `worker-exit`. Write a fixed
+            # literal done frame with the captured `_os_write` (itself immune to a
+            # rebind) so the host still gets a verdict. The literal is JSON-valid
+            # and newline-terminated; the lock is the channel's, so the write is
+            # serialized against any concurrent writer.
+            with channel._write_lock:
+                view = memoryview(_FALLBACK_DONE_FRAME)
+                while view:
+                    view = view[_os_write(channel._fd, view):]
 
     max_value_bytes = int(boot["maxValueBytes"])
     done: dict[str, Any] | str

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

@@ -1486,8 +1486,15 @@ export class PythonCodeRuntime extends CodeRuntime {
             // bindings awaiting fd 3's `drain` can queue many frames, and each
             // `shift()` re-slices the remaining array (O(n) per pop, O(n²) over
             // the whole drain). A head cursor keeps the cost linear; the `finally`
-            // below discards everything consumed once the drain ends.
+            // below discards everything consumed once the drain ends. The consumed
+            // slot is CLEARED here (not just advanced past) so a wide payload the
+            // pipe has already taken is released immediately: under sustained
+            // backpressure the drain loop can live across many `await drain`
+            // ticks, and leaving the slot set would pin the written value's bytes
+            // in `replyQueue` for the whole busy period, making host memory grow
+            // with cumulative processing rather than the current backlog.
             const payload = replyQueue[head] as ReplyMessage
+            replyQueue[head] = undefined as unknown as ReplyMessage
             head += 1
             // Encode inside the loop, not up front: a queued reply the run no
             // longer needs is dropped by the `settled` check above without ever

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

@@ -511,6 +511,37 @@ describe('PythonCodeRuntime — inherited resource limits', () => {
     expect(result.value).toBe(5)
   }, 15_000)
 
+  it('reports a CPU overrun under a dual-limit ulimit as a timeout, not a worker-exit', async () => {
+    // `ulimit -t N` sets BOTH the soft and hard CPU limit to N. The kernel
+    // checks the hard limit in the same tick and SIGKILLs a busy loop directly,
+    // so SIGXCPU is never delivered — and the host classifies a CPU overrun
+    // ONLY on `signal === 'SIGXCPU'`, so the overrun would be misreported as a
+    // `worker-exit` instead of a timeout. `_clamped` now lowers a clamped
+    // soft==hard result by one unit (when hard >= 2), so SIGXCPU fires at the
+    // softer limit and the run reports a timeout. This drives a busy loop past
+    // the inherited cap and asserts the run classifies as a timeout.
+    const dir = await mkdtemp(join(tmpdir(), 'dsh-rlimit-dual-'))
+    const wrapper = join(dir, 'python3-dual-capped')
+    // Both soft and hard CPU 1 s; configured cpuSeconds 30 s.
+    await writeFile(wrapper, '#!/bin/sh\nulimit -t 1\nexec python3 "$@"\n', { mode: 0o755 })
+    const { runtime } = await setup({ pythonBin: wrapper, cpuSeconds: 30, maxWallMs: 12_000 })
+    const result = await runtime.run({
+      program: [
+        'import signal',
+        // Trap SIGXCPU; with the soft limit one unit below the hard it fires at
+        // 1 s and the run is classified as a CPU timeout, not a worker-exit.
+        'signal.signal(signal.SIGXCPU, lambda *a: None)',
+        'end = 2.5',
+        'while True:',
+        '    pass',
+        'return "unreachable"',
+      ].join('\n'),
+      bindings: [],
+    })
+    expect(result.error?.kind).toBe('timeout')
+    expect(result.error?.message).toContain('SIGXCPU')
+  }, 15_000)
+
   it('rechecks CPU at settlement against the effective inherited soft limit', async () => {
     // The settlement-time CPU recheck must compare against the EFFECTIVE soft
     // limit (`_clamped` may have lowered it to a stricter inherited value), not
@@ -793,6 +824,31 @@ describe('PythonCodeRuntime — programs and bindings', () => {
     expect(result.logs.join('').length).toBeLessThan(4096)
   })
 
+  it('bounds a newline-free single-character Python write drip by the fragment cap, not OOM', async () => {
+    // The child-side `_LogStream` buffers one fragment per `write` (so
+    // `print("x", end="")` does not concatenate quadratically). A newline-free
+    // drip of one character per call past a large `maxLogBytes` would otherwise
+    // accumulate one list slot (and one str object) per call — 25 M calls =
+    // ~25 M slots, which OOMs the host on its own accounting before the byte
+    // budget is reached. The stream seals the fragment list past
+    // `_PENDING_MAX_CHUNKS` into one joined block (character count unchanged),
+    // bounding the live fragment count exactly as the host-side `captureStray`
+    // seal does. This drives well past the cap and asserts the run still
+    // completes with a truncation marker rather than a MemoryError.
+    const { runtime } = await setup({ maxLogBytes: 4096 })
+    const result = await runtime.run({
+      program: [
+        'import sys',
+        'for _ in range(200_000):',
+        '    sys.stdout.write("x")',
+        'return None',
+      ].join('\n'),
+      bindings: [],
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.logs.at(-1)).toBe(logTruncationMarker(4096))
+  })
+
   it('bounds a control-char-dense native residual by serialized cost, not raw length', async () => {
     // A newline-free NUL flood passes the cheap `length + 3` lower bound at a
     // raw length well under the budget, but each NUL serializes to `` (6
@@ -1427,6 +1483,34 @@ describe('PythonCodeRuntime — programs and bindings', () => {
     expect(result.error?.message).not.toContain('hijacked')
   }, 15_000)
 
+  it('still delivers a done frame when a transitive encode name is rebound', async () => {
+    // `send_done` binds `_encode_json_plain` and `ProtocolChannel.write_encoded`
+    // into locals, but those callables' BODIES still resolve transitive module
+    // globals at call time: `_encode_json_plain` reaches `_dump_scalar`/`_dump_string`/
+    // `json.dumps`, and `write_encoded` reaches `os.write`. This bootstrap is
+    // `__main__`, so rebinding `__main__._dump_scalar` to a raising function makes
+    // the error-frame encode throw AFTER the `except` block. `send_done` catches
+    // that and writes a fixed literal done frame (kind `exception`) with the
+    // captured `os.write`, so the host still gets a verdict — the run must be an
+    // `exception`, never a `worker-exit`. The real message is lost (the literal
+    // carries a fixed `<unrenderable>` text), which is acceptable: the verdict
+    // outranks the diagnostic detail.
+    const { runtime } = await setup({ maxWallMs: 10_000 })
+    const result = await runtime.run({
+      program: [
+        'import __main__',
+        'def boom(*a, **k):',
+        '    raise RuntimeError("hijacked")',
+        '__main__._dump_scalar = boom',
+        '__main__.os = boom',
+        'raise ValueError("real failure")',
+      ].join('\n'),
+      bindings: [],
+    })
+    expect(result.error?.kind).toBe('exception')
+    expect(result.error?.kind).not.toBe('worker-exit')
+  }, 15_000)
+
   it('bounds an over-cap exception-group nesting on the copy', async () => {
     // Exception groups link through `exceptions`, not the cause/context
     // dunders, so the cap has to count that edge too — otherwise a deeply