test(python): control stray-output fragments for sealing coverage

2026-09-14 04:01:35 +00:00 · 2026-09-10 18:29:04 +08:00
parent 3b5daf10be
commit 0963bdd7d3
5 changed files with 62 additions and 54 deletions
@@ -4759,58 +4759,6 @@ 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 appends a finished block rather than re-merging everything held, so
-    // each byte is copied a bounded number of times. Re-merging the whole
-    // residual at every seal threshold instead makes the cumulative copy volume
-    // quadratic. Measured like the fd-3 sibling above rather than reasoned about:
-    // this sealed shape copies about 120 KB for 60000 trickled bytes, the
-    // re-merging shape about 538 KB (the stray path adds one whole-residual
-    // concat at the terminating newline over the fd-3 sibling's 119/540, landing
-    // at the same order). 256 KiB sits between them with margin on both sides, so
-    // reverting the seal to a re-merge turns this assertion red.
-    expect(copied).toBeLessThan(256 * 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
@@ -0,0 +1,56 @@
+import { Context } from '@deepseek-ai/cordis'
+import { expect, it, vi } from 'vitest'
+
+// Keep the interpreter and pipe lifecycle real; only OS-dependent read sizes
+// change. Each byte reaches the runtime as its own data event.
+vi.mock('node:child_process', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('node:child_process')>()
+  return {
+    ...actual,
+    spawn: vi.fn(actual.spawn).mockImplementation((...args) => {
+      const child = actual.spawn(...args)
+      const stdout = child.stdout!
+      const emit = stdout.emit.bind(stdout)
+      stdout.emit = (event: string | symbol, ...values: unknown[]) => {
+        if (event !== 'data') return emit(event, ...values)
+        const chunk = values[0] as Buffer
+        for (let offset = 0; offset < chunk.length; offset++) {
+          emit('data', chunk.subarray(offset, offset + 1))
+        }
+        return true
+      }
+      return child
+    }),
+  }
+})
+
+const { PythonCodeRuntime } = await import('../src/index.ts')
+
+it('seals stray fragments without recopying the sealed prefix', async () => {
+  const ctx = new Context()
+  const fiber = await ctx.plugin(PythonCodeRuntime, { maxLogBytes: 200_000, maxWallMs: 30_000 })
+  const realConcat = Buffer.concat.bind(Buffer)
+  let copied = 0
+  let maxFragments = 0
+  const concat = vi.spyOn(Buffer, 'concat').mockImplementation((list, total) => {
+    maxFragments = Math.max(maxFragments, list.length)
+    for (const part of list) copied += part.length
+    return realConcat(list, total)
+  })
+  try {
+    const result = await ctx.codeRuntime.run({
+      program: 'import os\nos.write(1, b"x" * 60000 + b"\\n")\nreturn "done"',
+      bindings: [],
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.value).toBe('done')
+    expect(result.logs).toEqual(['x'.repeat(60_000)])
+    // Sealing copies each byte at most twice; merging every accumulated prefix
+    // instead copies over a megabyte for these 60,001 controlled fragments.
+    expect(maxFragments).toBeLessThanOrEqual(1024)
+    expect(copied).toBeLessThan(256 * 1024)
+  } finally {
+    concat.mockRestore()
+    await fiber.dispose()
+  }
+}, 40_000)