fix(code-runtime-python): seal stray fragments, flush by serialized cost, charge lone surrogates fully

Three follow-ups the review caught in the stray-capture rewrite, plus a cost
undercount shared with the log ledger.

Seal the stray fragment list into blocks past MAX_PENDING_CHUNKS, mirroring the
fd-3 reader: a program pacing single-byte os.write(1, ...) calls otherwise
accumulates one live Buffer per write, and the per-object overhead no byte
count sees exhausts the host heap far below the budget.

Flush the residual by its running SERIALIZED cost (serializedBufferCost, a
per-byte lower bound) rather than raw byte count: a control-char-dense
newline-free flood serializes several-fold, so a raw-byte threshold let it grow
to a full budget's worth of raw bytes — up to ~6x what the ledger admits —
before flushStray concat/decoded the whole ~256 MiB residual at once.

Charge a lone surrogate its full six escaped bytes (\uXXXX under ES2019
well-formed JSON.stringify) in both jsonStringCostUpTo and serializedBufferCost,
not the three bytes Buffer.byteLength reports for U+FFFD: a forged log frame
flooding \ud800 escapes was undercharged by half and admitted ~2x maxLogBytes.

Key the sync-spawn leak assertion off the exact bootstrap path from the mocked
spawn's argv, immune to a sibling worker's concurrent staging. Refresh the
stale load-check comment that named the replaced JSON.stringify mechanism.

Add lone-surrogate, stray-sealing, and companion regression tests (per-file
100% coverage); update the Agent Note and zh pair.

packages/code-runtime/code-runtime-python/tests/boot-write-failure.spec.ts

@@ -1,7 +1,7 @@
 import { EventEmitter } from 'node:events'
-import { readdirSync } from 'node:fs'
+import { existsSync } from 'node:fs'
+import { dirname } from 'node:path'
 import { PassThrough } from 'node:stream'
-import { tmpdir } from 'node:os'
 import { afterEach, describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
 
@@ -74,14 +74,18 @@ describe('PythonCodeRuntime — boot-write failure', () => {
     // misuse) and stranded the staging directory materializePyScripts had just
     // written, which only settle() removes. The fix catches it, unlinks the
     // directory, and resolves the same `worker-exit` class as an async ENOENT.
-    // Snapshot as a SET, then assert no dir NEW relative to it survives. Strict
-    // array equality would flake: vitest's forks pool runs runtime.spec.ts in a
-    // sibling worker that concurrently creates and removes
-    // `dsh-code-runtime-python-*` dirs, so a concurrent create OR delete in the
-    // window would fail `toEqual`. The set difference is immune to both — it
-    // only asserts THIS run left nothing behind.
-    const before = new Set(readdirSync(tmpdir()).filter(name => name.startsWith('dsh-code-runtime-python-')))
-    spawnMock.mockImplementation(() => { throw Object.assign(new Error('EMFILE: too many open files'), { code: 'EMFILE' }) })
+    //
+    // Capture THIS run's exact staging dir from the argv the mocked spawn
+    // received (`['-I', <dir>/bootstrap.py]`) and assert only that path is gone.
+    // A tmpdir scan — even a set difference against a pre-run snapshot — would
+    // flake under vitest's forks pool: a sibling worker creating its own
+    // `dsh-code-runtime-python-*` dir in the window reads as a leak here. Keying
+    // off our own argv is fully isolated from concurrent staging.
+    let stagedBootstrap: string | undefined
+    spawnMock.mockImplementation((_bin: string, args: string[]) => {
+      stagedBootstrap = args[args.length - 1]
+      throw Object.assign(new Error('EMFILE: too many open files'), { code: 'EMFILE' })
+    })
     const ctx = new Context()
     const fiber = await ctx.plugin(PythonCodeRuntime)
     const runtime = ctx.codeRuntime as InstanceType<typeof PythonCodeRuntime>
@@ -90,8 +94,8 @@ describe('PythonCodeRuntime — boot-write failure', () => {
 
     expect(result.error?.kind).toBe('worker-exit')
     expect(result.error?.message).toContain('python spawn error')
-    const leaked = readdirSync(tmpdir()).filter(name => name.startsWith('dsh-code-runtime-python-') && !before.has(name))
-    expect(leaked).toEqual([])
+    expect(stagedBootstrap).toBeDefined()
+    expect(existsSync(dirname(stagedBootstrap as string))).toBe(false)
     await fiber.dispose()
   })
 })

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

@@ -747,6 +747,33 @@ describe('PythonCodeRuntime — programs and bindings', () => {
     expect(result.logs.at(-1)).toBe(logTruncationMarker(4096))
   })
 
+  it('charges a lone surrogate its full six escaped bytes, not three', async () => {
+    // A forged `log` frame carrying `\ud800` escapes materializes lone
+    // surrogates after JSON.parse. `Buffer.byteLength` of U+FFFD is 3, but
+    // ES2019 well-formed `JSON.stringify` emits `\ud800` at 6 bytes, so charging
+    // the raw width would admit ~2x the configured budget of serialized bytes
+    // (the same family as the NUL-flood undercount, at 2x rather than 6x). The
+    // cost walker charges surrogates the full 6, so a flood truncates at budget.
+    // Forged on fd 3 because Python stdout will not emit lone surrogates.
+    const { runtime } = await setup({ maxLogBytes: 4096 })
+    const result = await runtime.run({
+      program: [
+        'import os',
+        // 1000 \ud800 escapes: charged at the buggy raw width 1000 * 3 = 3000
+        // bytes fits under 4096 (wrongly admitted), but the correct serialized
+        // width 1000 * 6 = 6000 bytes is over budget — so the ledger must
+        // truncate. The count sits in the 683..1365 window where the two
+        // chargings disagree, making the test discriminate.
+        String.raw`frame = b'{"type":"log","text":"' + b'\\ud800' * 1000 + b'"}\n'`,
+        'os.write(3, frame)',
+        'return None',
+      ].join('\n'),
+      bindings: [],
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.logs.at(-1)).toBe(logTruncationMarker(4096))
+  })
+
   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
@@ -3074,6 +3101,51 @@ describe('PythonCodeRuntime — hostile peer', () => {
     expect(copied).toBeLessThan(256 * 1024)
   }, 40_000)
 
+  it('seals trickled stray fragments into blocks without recopying the sealed prefix', async () => {
+    // The stray-capture buffer has the same object-overhead exposure as the fd-3
+    // reader above: each newline-free `data` chunk is its own Buffer, so a
+    // program pacing single-byte `os.write(1, ...)` accumulates one object per
+    // write, which the serialized-cost counter cannot see. Past MAX_PENDING_CHUNKS
+    // the fragments seal into a finished block; re-merging the whole residual at
+    // each threshold instead would copy the sealed prefix again and again, making
+    // the cumulative copy volume quadratic. `Buffer.concat` is wrapped to measure
+    // that volume — both shapes admit the same final log entry, so the copy total
+    // is the discriminator. maxLogBytes is raised so the trickle is retained,
+    // not truncated, which is what forces the fragments to accumulate and seal.
+    const realConcat = Buffer.concat.bind(Buffer)
+    let copied = 0
+    Buffer.concat = (list: readonly Uint8Array[], total?: number): Buffer<ArrayBuffer> => {
+      for (const part of list) copied += part.length
+      return realConcat(list, total)
+    }
+    let result: CodeRunResult
+    try {
+      const { runtime } = await setup({ maxLogBytes: 200_000, maxWallMs: 30_000 })
+      result = await runtime.run({
+        program: [
+          'import os',
+          'for _ in range(60000):',
+          '    os.write(1, b"x")',
+          '    os.sched_yield()',
+          'os.write(1, b"\\n")',
+          'return "done"',
+        ].join('\n'),
+        bindings: [],
+      })
+    } finally {
+      Buffer.concat = realConcat
+    }
+    expect(result.error).toBeUndefined()
+    expect(result.value).toBe('done')
+    // The trickle coalesces into one log line (no interior newlines). Its exact
+    // length depends on pipe coalescing, but it is one entry and non-empty.
+    expect(result.logs.length).toBe(1)
+    expect((result.logs[0] as string).length).toBeGreaterThan(0)
+    // Sealing keeps each byte copied a bounded number of times; re-merging the
+    // whole residual per threshold would push the total far past this.
+    expect(copied).toBeLessThan(2 * 1024 * 1024)
+  }, 40_000)
+
   it('caps a huge exception diagnostic child-side before it crosses the wire', async () => {
     // A program can raise with a multi-megabyte message; the child must cap
     // it at maxValueBytes before formatting/sending, not ship the whole