Merge corrected #1148 checkpoint into #3289

# Conflicts: # docs/config-catalog.i18n.yaml # docs/config-catalog.zh.md # packages/experimental/code-runtime-python/README.i18n.yaml # packages/experimental/code-runtime-python/README.md # packages/experimental/code-runtime-python/README.zh.md # packages/experimental/code-runtime-python/src/index.ts # packages/experimental/code-runtime-python/tests/runtime.spec.ts
2026-09-11 04:00:38 +00:00 · 2026-08-31 16:10:13 +08:00
parent d6bd5eb973 61b3e06e97
commit 4e2c568efe
20 changed files with 691 additions and 155 deletions
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/experimental/code-runtime-python/README.md
-README.md: 117daeac38c329521e5334b41106b71c629a0efc
-README.zh.md: 0727c705a8eb024f068804c0a1af6e81c2e02d6b
+README.md: 0857660d983834cabc57340a9a61a5f6e402acaa
+README.zh.md: e87b0623d593da324e6ce149077a9cd9f732ce2b
@@ -25,11 +25,11 @@ English | [中文](README.zh.md)
 <a id="use-this-package"></a>
 ## Use this package

-Choose this private experimental package only in an explicit source-checkout composition. Register `PythonCodeRuntime` beside `dsh-tools` and `run()` executes each program in a fresh CPython 3.10+ subprocess, resolving with `result.value` on success and `result.error` on failure (the orthogonal `CodeRunFailure.kind` taxonomy classifies parse failures, thrown exceptions, invalid completions, output overflows, budget expiry, aborts, and substrate death). It rejects only for seam misuse — a malformed binding namespace, or a call after disposal. Configuration is rejected at load: a non-Unix platform; an explicit `pythonBin` that is not an executable regular file or a bare name that does not resolve on `PATH`; a non-CPython, pre-3.10, or probe-failing interpreter; a non-positive or non-integer budget; a `maxLogBytes` below the truncation-marker floor (64); a timer value `setTimeout` would clamp; a budget larger than one fd-3 frame can carry; or an `addressSpaceMb`/output-budget pair whose worst-case peak would breach `RLIMIT_AS`.
+Choose this private experimental package only in an explicit source-checkout composition. Register `PythonCodeRuntime` beside `dsh-tools` and `run()` executes each program in a fresh CPython 3.10+ subprocess, resolving with `result.value` on success and `result.error` on failure (the orthogonal `CodeRunFailure.kind` taxonomy classifies parse failures, thrown exceptions, invalid completions, output overflows, budget expiry, aborts, and substrate death). It rejects only for seam misuse — a malformed binding namespace, or a call after disposal. Configuration is rejected at load: a non-Unix platform; an explicit `pythonBin` that is not an executable regular file or a bare name that does not resolve on `PATH`; a non-CPython, pre-3.10, or probe-failing interpreter; a non-positive or non-integer budget; a `maxLogBytes` below the truncation-marker floor (64); a timer value `setTimeout` would clamp; a budget larger than the effective fd-3 frame cap (lowered when the host heap cannot safely parse a near-cap frame); or an `addressSpaceMb`/output-budget pair whose worst-case peak would breach `RLIMIT_AS`.

 ### What you get

-The package's default export is the `PythonCodeRuntime` plugin. Its public surface also re-exports the host-side protocol vocabulary: `validateChildFrame` (rebuilds every inbound frame), the lossless-JSON codec and meters (`encodeJsonPlain`, `checkDoneValue`, `hasUnsafeIntegerToken`, `hasNonLosslessNumber`), `logTruncationMarker` (the shared truncation-marker text), plus `resolvePythonBin` (interpreter lookup against the current `PATH`), `readProcessStart` (process-start statistics for tests), and `detachResidual` (a test seam for the settled run's resource cleanup). Every cap is a validated `Config` field with a default: `cpuSeconds` (60), `maxWallMs` (600000), `addressSpaceMb` (512, not applied on Darwin), `maxLogBytes` (65536), `maxValueBytes` (32768), `graceMs` (3000), and `pythonBin` (`python3`, resolved, executable-checked, version-probed under a five-second force-kill deadline, and frozen at load). Each child receives only `TMPDIR`; ambient credentials, `PATH`, `HOME`, and other host state stay unavailable.
+The package's default export is the `PythonCodeRuntime` plugin. Its public surface also re-exports the host-side protocol vocabulary: `validateChildFrame` (rebuilds every inbound frame), the lossless-JSON codec and meters (`encodeJsonPlain`, `checkDoneValue`, `hasUnsafeIntegerToken`, `hasNonLosslessNumber`), `logTruncationMarker` (the shared truncation-marker text), plus `resolvePythonBin` (interpreter lookup against the current `PATH`), `readProcessStart` (process-start statistics for tests), `detachResidual` (a test seam for the settled run's resource cleanup), and `hostFrameParseCeiling` (the heap-derived frame parse cap a given heap limit admits). Every cap is a validated `Config` field with a default: `cpuSeconds` (60), `maxWallMs` (600000), `addressSpaceMb` (512, not applied on Darwin), `maxLogBytes` (65536), `maxValueBytes` (32768), `graceMs` (3000), and `pythonBin` (`python3`, resolved, executable-checked, version-probed under a five-second force-kill deadline, and frozen at load). Each child receives only `TMPDIR`; ambient credentials, `PATH`, `HOME`, and other host state stay unavailable.

 ### The wire

@@ -37,7 +37,7 @@ Frames travel on the child's fd 3 as JSON-lines — one object per line — so s

 ### What can go wrong

-Host-side validation drops junk without throwing, so a malformed or forged frame never crashes the host process: `validateChildFrame` returns `undefined` for anything that does not rebuild cleanly, a non-number call id can never be echoed into a reply, and forged extra fields never ride along. A completion value that is not lossless JSON, or that exceeds the configured byte budget, is rejected explicitly (`non-lossless` / `over-budget`) rather than silently rounded or truncated. An fd-3 frame whose raw length exceeds 64 MiB settles the run as a `worker-exit` (the receive path caps raw frames before `toString`/`JSON.parse` so a compact wide frame cannot decode to far more host memory than its wire bytes admitted).
+Host-side validation drops junk without throwing, so a malformed or forged frame never crashes the host process: `validateChildFrame` returns `undefined` for anything that does not rebuild cleanly, a non-number call id can never be echoed into a reply, and forged extra fields never ride along. A completion value that is not lossless JSON, or that exceeds the configured byte budget, is rejected explicitly (`non-lossless` / `over-budget`) rather than silently rounded or truncated. An fd-3 frame whose raw length exceeds the effective frame parse cap (64 MiB, or lower when the host's configured heap cannot safely parse a near-cap frame — see `hostFrameParseCeiling`) settles the run as a `worker-exit` (the receive path caps raw frames before `toString`/`JSON.parse` so a compact wide frame cannot decode to far more host memory than its wire bytes admitted).

 -----

@@ -120,7 +120,7 @@ These limits define what the package does and does not cover; they are current p
 - **The truncation-marker text and the tempdir prefix keep the pre-rename short names** — the marker `[dsh-code-runtime-python] log capture truncated at <N> bytes` and the `dsh-code-runtime-python-` tempdir prefix are byte-anchored by tests and are independent of the npm package name; promotion (dropping the `experimental-` prefix) does not rename them.
 - **`run()` is one-shot** — `logs` become available only after `CodeRunResult` resolves; there is no streaming-log or progress interface for output produced by a running program.
 - **No state persists across runs** — every request executes in a fresh subprocess; a persistent REPL-style kernel stays deferred until a backend brings its own logging scheme.
- **An fd-3 frame whose raw length exceeds 64 MiB settles the run as a worker-exit** — `maxLogBytes`/`maxValueBytes` are load-bounded to the same parser cap so an honest child's frames always fit; a model-constructed binding ARGUMENT above 64 MiB (a value with no seam-level budget) trips the same cap — an accepted residual of the OOM guard.
+- **An fd-3 frame whose raw length exceeds the effective frame parse cap settles the run as a worker-exit** — the cap is 64 MiB, or lower when the host's configured heap cannot safely parse a near-cap frame (`hostFrameParseCeiling`); `maxLogBytes`/`maxValueBytes` are load-bounded to the same cap so an honest child's frames always fit; a model-constructed binding ARGUMENT above the cap (a value with no seam-level budget) trips it too — an accepted residual of the OOM guard.
 - **A child that stops reading its replies settles the run as a worker-exit once the reply backlog passes 1024 frames** — the host writes replies one at a time, waiting for `drain` when the pipe is full; a child that keeps sending calls without consuming replies would otherwise grow the retained backlog (and the binding results it pins) until the wall clock, so the backlog cap fails the run early. Binding results carry no seam-level byte cap, so this is a count bound, not a byte bound.
 - **A child that floods calls against a binding that never settles settles the run as a worker-exit once 1024 calls are in flight** — binding calls are counted before dispatch and released when the async body settles, so a binding whose promise never resolves would otherwise accumulate one async closure per call frame until the wall clock. Like the reply backlog, this is a count bound, not a byte bound.
 - **A combined log-and-value peak is not modelled by the load gate** — a model daemon thread that keeps writing while the completion value is metered and framed can add the two peaks in a way no gate admits or rejects; the run dies as `worker-exit`, containment holds, and only the failure classification is degraded.
@@ -25,11 +25,11 @@ kind: "package-reference"
 <a id="use-this-package"></a>
 ## 使用本包

-仅在显式源码检出组合中选择这个私有实验包。将 `PythonCodeRuntime` 与 `dsh-tools` 一起注册后，`run()` 会在全新的 CPython 3.10+ 子进程中执行每个程序；成功时以 `result.value` resolve，失败时以 `result.error` resolve（正交的 `CodeRunFailure.kind` 分类涵盖解析失败、抛出异常、无效完成值、输出溢出、预算到期、中止与执行基底终止）。仅有 seam 误用会 reject——binding 命名空间不合法，或在 dispose 后调用。配置在加载期拒绝：非 Unix 平台；不是可执行普通文件的显式 `pythonBin`，或无法在 `PATH` 上解析的裸名；非 CPython、低于 3.10 或探测失败的解释器；非正或非整数预算；低于截断标记下限（64）的 `maxLogBytes`；会被 `setTimeout` 截断的定时器值；超过单个 fd-3 帧承载能力的预算；或最坏峰值会突破 `RLIMIT_AS` 的 `addressSpaceMb`／输出预算组合。
+仅在显式源码检出组合中选择这个私有实验包。将 `PythonCodeRuntime` 与 `dsh-tools` 一起注册后，`run()` 会在全新的 CPython 3.10+ 子进程中执行每个程序；成功时以 `result.value` resolve，失败时以 `result.error` resolve（正交的 `CodeRunFailure.kind` 分类涵盖解析失败、抛出异常、无效完成值、输出溢出、预算到期、中止与执行基底终止）。仅有 seam 误用会 reject——binding 命名空间不合法，或在 dispose 后调用。配置在加载期拒绝：非 Unix 平台；不是可执行普通文件的显式 `pythonBin`，或无法在 `PATH` 上解析的裸名；非 CPython、低于 3.10 或探测失败的解释器；非正或非整数预算；低于截断标记下限（64）的 `maxLogBytes`；会被 `setTimeout` 截断的定时器值；超过有效 fd-3 帧上限的预算（宿主堆无法安全解析接近上限的帧时，该上限会降低）；或最坏峰值会突破 `RLIMIT_AS` 的 `addressSpaceMb`／输出预算组合。

 ### 你得到什么

-包的默认导出是 `PythonCodeRuntime` 插件。其公开面还重新导出宿主侧协议词汇：`validateChildFrame`（重建每条入站帧）、无损 JSON codec 与计量器（`encodeJsonPlain`、`checkDoneValue`、`hasUnsafeIntegerToken`、`hasNonLosslessNumber`）、`logTruncationMarker`（共享截断标记文本），以及 `resolvePythonBin`（对照当前 `PATH` 的解释器查找）、`readProcessStart`（供测试用的进程启动统计）和 `detachResidual`（已结算运行的资源清理测试 seam）。每个上限都是带默认值并经校验的 `Config` 字段：`cpuSeconds`（60）、`maxWallMs`（600000）、`addressSpaceMb`（512，Darwin 上不生效）、`maxLogBytes`（65536）、`maxValueBytes`（32768）、`graceMs`（3000）与 `pythonBin`（`python3`，在加载期解析、检查可执行性，在五秒强制终止期限内探测版本并固定）。每个子进程只接收 `TMPDIR`；环境中的凭证、`PATH`、`HOME` 与其他宿主状态均不可见。
+包的默认导出是 `PythonCodeRuntime` 插件。其公开面还重新导出宿主侧协议词汇：`validateChildFrame`（重建每条入站帧）、无损 JSON codec 与计量器（`encodeJsonPlain`、`checkDoneValue`、`hasUnsafeIntegerToken`、`hasNonLosslessNumber`）、`logTruncationMarker`（共享截断标记文本），以及 `resolvePythonBin`（对照当前 `PATH` 的解释器查找）、`readProcessStart`（供测试用的进程启动统计）、`detachResidual`（已结算运行的资源清理测试 seam）与 `hostFrameParseCeiling`（给定堆上限可容纳的堆推导帧解析上限）。每个上限都是带默认值并经校验的 `Config` 字段：`cpuSeconds`（60）、`maxWallMs`（600000）、`addressSpaceMb`（512，Darwin 上不生效）、`maxLogBytes`（65536）、`maxValueBytes`（32768）、`graceMs`（3000）与 `pythonBin`（`python3`，在加载期解析、检查可执行性，在五秒强制终止期限内探测版本并固定）。每个子进程只接收 `TMPDIR`；环境中的凭证、`PATH`、`HOME` 与其他宿主状态均不可见。

 ### wire

@@ -37,7 +37,7 @@ kind: "package-reference"

 ### 可能出错的地方

-宿主侧校验在不抛异常的情况下丢弃垃圾，因此畸形或伪造帧永远不会让宿主进程崩溃：`validateChildFrame` 对任何不能干净重建的内容返回 `undefined`，非数字的 call id 永远不会被回显进 reply，伪造的额外字段永远不会被带走。非无损 JSON 或超过配置字节预算的完成值会被显式拒绝（`non-lossless`／`over-budget`），而不是被静默取整或截断。原始长度超过 64 MiB 的 fd-3 帧会让本次运行以 `worker-exit` 结算（接收路径在 `toString`/`JSON.parse` 之前限制原始帧，紧凑宽帧不能解码出远超其线上字节的宿主内存）。
+宿主侧校验在不抛异常的情况下丢弃垃圾，因此畸形或伪造帧永远不会让宿主进程崩溃：`validateChildFrame` 对任何不能干净重建的内容返回 `undefined`，非数字的 call id 永远不会被回显进 reply，伪造的额外字段永远不会被带走。非无损 JSON 或超过配置字节预算的完成值会被显式拒绝（`non-lossless`／`over-budget`），而不是被静默取整或截断。原始长度超过有效帧解析上限（64 MiB，或当宿主的配置堆无法安全解析接近上限的帧时更低——见 `hostFrameParseCeiling`）的 fd-3 帧会让本次运行以 `worker-exit` 结算（接收路径在 `toString`/`JSON.parse` 之前限制原始帧，紧凑宽帧不能解码出远超其线上字节的宿主内存）。

 -----

@@ -119,7 +119,7 @@ kind: "package-reference"
 - **需要 CPython 3.10 或更高版本**——配置的可执行文件会在加载期完成解析与版本探测；不受支持的解释器会在 `ctx.codeRuntime` 注册前失败。
 - **`run()` 是一次性的**——`logs` 只有在 `CodeRunResult` resolve 后才能获得；没有为运行中程序产生的输出提供流式日志或进度接口。
 - **运行之间不保留状态**——每次请求都在全新子进程中执行；持久 REPL 风格内核在某个后端带来自己的日志方案之前保持延期。
- **原始长度超过 64 MiB 的 fd-3 帧会让本次运行以 worker-exit 结算**——`maxLogBytes`/`maxValueBytes` 在加载期被限制到同一解析器上限，因此诚实子进程的帧总能放得下；模型构造的超过 64 MiB 的 binding 实参（一个在 seam 层没有预算的值）会触发同一上限——这是该 OOM 防护的已接受残余。
+- **原始长度超过有效帧解析上限的 fd-3 帧会让本次运行以 worker-exit 结算**——上限为 64 MiB，或当宿主的配置堆无法安全解析接近上限的帧时更低（`hostFrameParseCeiling`）；`maxLogBytes`/`maxValueBytes` 在加载期被限制到同一上限，因此诚实子进程的帧总能放得下；模型构造的超过该上限的 binding 实参（一个在 seam 层没有预算的值）会触发同一上限——这是该 OOM 防护的已接受残余。
 - **停止读取回复的子进程会在回复积压超过 1024 帧时以 worker-exit 结算运行**——宿主每次写一条回复，管道满时等待 `drain`；只持续发送调用而不消费回复的子进程会让保留的积压（及其钉住的 binding 结果）一直增长到墙钟，因此积压上限让运行提前失败。binding 结果在 seam 层没有字节上限，所以这是计数上限而非字节上限。
 - **向永不结算的 binding 洪泛调用的子进程会在 1024 个调用在途时以 worker-exit 结算运行**——binding 调用在分发前计数、异步体结算时释放，否则 promise 永不 resolve 的 binding 会让每个调用帧累积一个异步闭包直到墙钟。与回复积压一样，这是计数上限而非字节上限。
 - **组合日志与值的峰值不被加载门建模**——持续写入的模型 daemon 线程与完成值计量、分帧相加的峰值没有任何门会放行或拒绝；运行以 `worker-exit` 告终，隔离成立，只有失败分类降级。
@@ -31,16 +31,16 @@
  "peerDependencies": {
    "@deepseek-ai/dsh-code-runtime": "workspace:^",
    "@deepseek-ai/dsh-invariants": "workspace:^",
-    "@deepseek-ai/dsh-session": "workspace:^",
    "@deepseek-ai/dsh-timeout": "workspace:^",
-    "@deepseek-ai/cordis": "workspace:^"
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-util-values": "workspace:^"
  },
  "devDependencies": {
    "@deepseek-ai/dsh-code-runtime": "workspace:^",
    "@deepseek-ai/dsh-invariants": "workspace:^",
-    "@deepseek-ai/dsh-session": "workspace:^",
    "@deepseek-ai/dsh-timeout": "workspace:^",
-    "@deepseek-ai/cordis": "workspace:^"
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-util-values": "workspace:^"
  },
  "dependencies": {
    "@deepseek-ai/schemastery": "workspace:^"
@@ -1883,7 +1883,7 @@ def _check_done_value(value: Any, max_bytes: int):
    # forgery before any iteration begins.
    exhausted = object()
    visit, list_cursor, dict_cursor = 0, 1, 2
-    stack: list[tuple[int, Any, Any]] = [(visit, value, None)]
+    stack: list[tuple[int, Any, Any, Any]] = [(visit, value, None, None)]
    while stack:
        frame = stack.pop()
        kind = frame[0]
@@ -1896,10 +1896,10 @@ def _check_done_value(value: Any, max_bytes: int):
            # Resume this cursor after the child is fully walked; the child goes
            # on top so it is visited next (order does not affect the byte total).
            stack.append(frame)
-            stack.append((visit, child, None))
+            stack.append((visit, child, None, None))
            continue
        if kind == dict_cursor:
-            container, iterator = frame[1], frame[2]
+            container, iterator, seen = frame[1], frame[2], frame[3]
            entry = next(iterator, exhausted)
            if entry is exhausted:
                on_path.discard(id(container))
@@ -1910,6 +1910,16 @@ def _check_done_value(value: Any, max_bytes: int):
            # the encoder emits its real characters.
            if type(key) is not str:
                return invalid(f"non-string dict key ({type(key).__name__})")
+            # The key's JSON form folds a spelled-out surrogate pair into its
+            # astral code point (`_dump_string`), so two DIFFERENT Python keys —
+            # the two code units and the single character — encode to the same
+            # JSON member, and the host's JSON.parse silently drops one of them.
+            # That is a lossless-JSON violation, so the collision is rejected
+            # here, before any encoding.
+            combined_key = _SURROGATE_PAIR.sub(_combine_surrogate_pair, key)
+            if combined_key in seen:
+                return invalid("duplicate dict key after surrogate-pair combining")
+            seen.add(combined_key)
            # The same string lower bound, before escaping the key.
            if total + len(key) + 3 > max_bytes:
                return over_budget
@@ -1919,7 +1929,7 @@ def _check_done_value(value: Any, max_bytes: int):
            if total > max_bytes:
                return over_budget
            stack.append(frame)
-            stack.append((visit, item, None))
+            stack.append((visit, item, None, None))
            continue
        current = frame[1]
        if current is None or type(current) is bool:
@@ -1969,7 +1979,7 @@ def _check_done_value(value: Any, max_bytes: int):
            if total + count > max_bytes:
                return over_budget
            on_path.add(id(current))
-            stack.append((list_cursor, current, iter(current)))
+            stack.append((list_cursor, current, iter(current), None))
        elif type(current) is dict:
            if id(current) in on_path:
                return invalid("circular reference")
@@ -1984,7 +1994,10 @@ def _check_done_value(value: Any, max_bytes: int):
            if total + count * 4 > max_bytes:
                return over_budget
            on_path.add(id(current))
-            stack.append((dict_cursor, current, iter(current.items())))
+            # The seen-set holds one combined key per member — O(keys), the same
+            # order as the dict itself — so the surrogate-collision check below
+            # can detect two keys that fold to one JSON member.
+            stack.append((dict_cursor, current, iter(current.items()), set()))
        else:
            # tuple, set, or any other type: not round-trippable JSON.
            return invalid(f"unsupported type ({type(current).__name__})")
@@ -2034,12 +2047,13 @@ def _lossless_json_violation(value: Any) -> str | None:
    # iterator; children are pulled one at a time.
    exhausted = object()
    visit, container_cursor = 0, 1
-    # A visit frame is (visit, value); a cursor frame is (cursor, container, iterator).
-    stack: list[tuple[int, Any, Any]] = [(visit, value, None)]
+    # A visit frame is (visit, value, None, None); a cursor frame is
+    # (cursor, container, iterator, seen-keys-for-dicts).
+    stack: list[tuple[int, Any, Any, Any]] = [(visit, value, None, None)]
    while stack:
        kind = stack[-1][0]
        if kind == container_cursor:
-            _, container, iterator = stack[-1]
+            _, container, iterator, seen = stack[-1]
            child = next(iterator, exhausted)
            if child is exhausted:
                # Leaving the container: it is no longer on the current path, so
@@ -2055,9 +2069,19 @@ def _lossless_json_violation(value: Any) -> str | None:
                key, child = child
                if type(key) is not str:
                    return f"non-string dict key ({type(key).__name__})"
-            stack.append((visit, child, None))
+                # The key's JSON form folds a spelled-out surrogate pair into its
+                # astral code point (`_dump_string`), so two DIFFERENT Python
+                # keys -- the two code units and the single character -- encode
+                # to the same JSON member, and the host's JSON.parse silently
+                # drops one of them. A lossless-JSON violation, rejected here
+                # before any encoding.
+                combined_key = _SURROGATE_PAIR.sub(_combine_surrogate_pair, key)
+                if combined_key in seen:
+                    return "duplicate dict key after surrogate-pair combining"
+                seen.add(combined_key)
+            stack.append((visit, child, None, None))
            continue
-        _, current, _unused = stack.pop()
+        _, current, _unused, _unused2 = stack.pop()
        if current is None or type(current) is bool:
            continue
        if type(current) is str:
@@ -2095,10 +2119,12 @@ def _lossless_json_violation(value: Any) -> str | None:
                # Keys are checked as the cursor pulls each entry, not in a
                # separate pass: ``current.values()`` would need a second walk,
                # and materializing ``items()`` up front allocates one tuple per
-                # member -- the very spike the cursor removes.
-                stack.append((container_cursor, current, iter(current.items())))
+                # member -- the very spike the cursor removes. The seen-set holds
+                # one combined key per member -- O(keys), the same order as the
+                # dict itself -- for the surrogate-collision check.
+                stack.append((container_cursor, current, iter(current.items()), set()))
            else:
-                stack.append((container_cursor, current, iter(current)))
+                stack.append((container_cursor, current, iter(current), None))
            continue
        return f"unsupported type ({type(current).__name__})"
    return None
@@ -5,10 +5,8 @@
 * boundary: model code has bash-equivalent trust, contained by a tempdir-only environment,
 * RLIMIT_CPU + RLIMIT_AS, wall-clock timeout, and SIGTERM→grace→SIGKILL on the process group.
 *
- * The package owns the versionless fd-3 wire protocol between the Node host and
- * the CPython subprocess. The protocol's host-side codec and hostile-frame
- * validators are re-exported so every consumer of the wire shares one
- * vocabulary.
+ * The package also owns the versionless fd-3 wire protocol itself; its host-side codec and
+ * hostile-frame validators are re-exported so every consumer of the wire shares one vocabulary.
 * @module @deepseek-ai/dsh-experimental-code-runtime-python
 */

@@ -17,12 +15,13 @@ import { accessSync, copyFileSync, constants as fsConstants, mkdtempSync, readFi
 import { tmpdir } from 'node:os'
 import { delimiter, dirname, isAbsolute, join, resolve } from 'node:path'
 import { fileURLToPath } from 'node:url'
+import { getHeapStatistics } from 'node:v8'
 import type { Duplex } from 'node:stream'
-import { Context } from 'cordis'
-import z from 'schemastery'
+import { Context } from '@deepseek-ai/cordis'
+import z from '@deepseek-ai/schemastery'
 import { CodeRuntime, DUNDER_MEMBER, PORTABLE_RESERVED_WORDS, RESERVED_BINDING_GLOBALS, RESERVED_ERROR_MEMBERS } from '@deepseek-ai/dsh-code-runtime'
 import type { CodeBindingErrorClass, CodeBindingFunction, CodeJsonValue, CodeRunFailure, CodeRunRequest, CodeRunResult } from '@deepseek-ai/dsh-code-runtime'
-import { snapshotJsonValue } from '@deepseek-ai/dsh-session'
+import { snapshotJsonValue } from '@deepseek-ai/dsh-util-values'
 import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
 import type { BootMessage, ChildToHost, ReplyMessage } from './protocol.ts'
 import { checkDoneValue, encodeJsonPlain, hasUnsafeIntegerToken, logTruncationMarker, validateChildFrame } from './protocol.ts'
@@ -70,7 +69,9 @@ export interface Config {
   * under RLIMIT_AS with several copies live at once, so this cap times the
   * worst-case Unicode expansion must fit the address space left after the
   * interpreter baseline (see `addressSpaceMb`) — a load-time rejection, not a
-   * runtime clamp.
+   * runtime clamp. Also bounded at load by the host's configured heap like
+   * `maxValueBytes` (see its JSDoc): the effective frame cap minus the frame
+   * envelope.
   */
  maxLogBytes?: number
  /**
@@ -78,7 +79,12 @@ export interface Config {
   * the same way `maxLogBytes` is: the child builds and encodes a near-budget
   * value under RLIMIT_AS with several copies live at once, so this cap times the
   * worst-case Unicode expansion must fit the address space left after the
-   * interpreter baseline.
+   * interpreter baseline. Both budgets are ALSO bounded at load by the host's
+   * configured heap: the effective frame cap (the protocol cap, or a lower
+   * heap-derived ceiling when the host heap cannot safely parse a near-cap
+   * frame — see `hostFrameParseCeiling`) minus the frame envelope, so a budget
+   * whose honest frame could OOM the host's own JSON.parse is rejected up
+   * front.
   */
  maxValueBytes?: number
  /** SIGTERM→SIGKILL grace period on kill, matching bash-local's default. */
@@ -310,6 +316,56 @@ const OUTPUT_BUDGET_WORST_CASE_ADDRESS_SPACE_MULTIPLE = 12
 */
 const INTERPRETER_BASELINE_BYTES = 64 * 1024 * 1024

+/**
+ * Worst-case peak host-heap bytes the PARSE of one inbound fd-3 frame can
+ * transiently occupy, expressed as a multiple of the frame's raw bytes.
+ * `JSON.parse` of a wide container materializes the object's property storage
+ * and key strings on top of the raw text; the WORST shape is a dict of many
+ * SHORT UNIQUE keys, which forces V8's dictionary-mode property storage
+ * (~32-64 bytes per entry) plus one interned string per key (header + data)
+ * plus string-table growth: measured 6.4x for a 3,000,000-key frame (~31 MB
+ * raw) on a 1 GiB heap, trending up with key count (a flat unique-key array
+ * is ~4x, a repeated-key dict ~3x). On a constrained heap the parse also
+ * retains the raw frame string while the object builds, so the safety factor
+ * is 16x — ~2.5x over the measured worst shape, ~1.6x over the claimed
+ * GC-headroom bound. Used with the host's configured heap limit to derive the
+ * largest frame whose parse cannot OOM the host process. This bounds the
+ * HOST's parse; {@link OUTPUT_BUDGET_WORST_CASE_ADDRESS_SPACE_MULTIPLE} bounds
+ * the CHILD's build and encode under RLIMIT_AS, a different resource. A fixed
+ * safety invariant, not a knob.
+ */
+const HOST_PARSE_WORST_CASE_MULTIPLE = 16
+
+/**
+ * Fixed host-heap headroom reserved for the application itself (the dsh
+ * fiber, plugins, and this runtime's own state) before the frame-parse
+ * multiple claims the rest: the effective frame cap is derived from
+ * `heap_size_limit - HOST_PARSE_BASELINE_BYTES`, so a constrained host's
+ * parse ceiling never spends the application's working set. A fixed safety
+ * margin, not a knob.
+ */
+const HOST_PARSE_BASELINE_BYTES = 64 * 1024 * 1024
+
+/**
+ * The largest inbound fd-3 frame the HOST can parse without risking a
+ * process-level OOM on its current heap: the configured heap limit (honoring
+ * `--max-old-space-size`) minus the application baseline, divided by the
+ * worst-case parse multiple, floored to the protocol frame cap. The
+ * raw-byte cap alone does not protect the heap — `JSON.parse` of a
+ * ≤64 MiB wide-object frame materializes several times that in property
+ * storage — so the effective cap is the smaller of the two. A default Node
+ * heap (~4 GiB) never binds; a constrained host (e.g.
+ * `--max-old-space-size=256` reports a ~300 MiB limit) lowers it to ~29 MiB,
+ * and the load gate rejects budgets that cannot cross it.
+ * @param heapLimit - the host's configured heap limit; the live
+ * `heap_size_limit` when omitted. A parameter so the derivation is unit
+ * testable against simulated heap sizes.
+ * @returns the effective frame parse cap in bytes.
+ */
+export function hostFrameParseCeiling(heapLimit: number = getHeapStatistics().heap_size_limit): number {
+  return Math.min(FRAME_PARSE_CAP_BYTES, Math.floor((heapLimit - HOST_PARSE_BASELINE_BYTES) / HOST_PARSE_WORST_CASE_MULTIPLE))
+}
+
 /**
 * Interval between process-group liveness probes while settlement waits for an
 * escalated SIGKILL to empty the group (see the `killing` branch in
@@ -762,6 +818,11 @@ export class PythonCodeRuntime extends CodeRuntime {

  private readonly config: ResolvedConfig
  private readonly pythonBin: string
+  // The frame cap this instance enforces: the protocol cap, or the host's
+  // heap-derived parse ceiling when a constrained heap makes the protocol cap
+  // unsafe to parse (see {@link hostFrameParseCeiling}). Computed per
+  // instance so the config gate and the inbound checks agree.
+  private readonly frameParseCapBytes = hostFrameParseCeiling()
  private readonly live = new Set<LiveRun>()
  private disposed = false

@@ -852,10 +913,12 @@ export class PythonCodeRuntime extends CodeRuntime {
    // into `CodeRunResult.error.message` and never re-crosses a frame-bounded
    // channel, so it is not part of the wire-width bound (see its JSDoc). The
    // admissible cap is therefore `parse-cap - envelope`: the receive path
-    // rejects raw frames past FRAME_PARSE_CAP_BYTES before decoding (the run
-    // settles as a worker-exit; a hostile compact-wide-frame OOM guard), so a
-    // budget must not exceed what an honest child's frame can actually carry
-    // through that parser.
+    // rejects raw frames past the effective parse cap (`frameParseCapBytes` —
+    // the protocol cap, or the host's heap-derived ceiling when a constrained
+    // heap makes the protocol cap unsafe to parse; see hostFrameParseCeiling)
+    // before decoding (the run settles as a worker-exit; a hostile
+    // compact-wide-frame OOM guard), so a budget must not exceed what an
+    // honest child's frame can actually carry through that parser.
    for (const key of ['maxLogBytes', 'maxValueBytes'] as const) {
      // Require an integer: the child reads these budgets through `int(...)`,
      // which silently floors a float, so `maxLogBytes: 3.5` would truncate at 3
@@ -865,9 +928,16 @@ export class PythonCodeRuntime extends CodeRuntime {
      if (!Number.isInteger(this.config[key])) {
        throw new Error(`dsh-code-runtime-python: config.${key} must be a positive integer (the child reads it as an int, so a float diverges from the host), got ${String(this.config[key])}`)
      }
-      const limit = FRAME_PARSE_CAP_BYTES - FRAME_ENVELOPE_BYTES
+      const limit = this.frameParseCapBytes - FRAME_ENVELOPE_BYTES
      if (this.config[key] > limit) {
-        throw new Error(`dsh-code-runtime-python: config.${key} must not exceed ${limit} (a payload that large cannot cross the fd-3 frame PARSER, which rejects raw frames past ${FRAME_PARSE_CAP_BYTES} bytes before decoding to bound host memory — a larger budget would admit a config whose honest child frames the host then rejects as a worker-exit), got ${String(this.config[key])}`)
+        // Only a host whose heap is below the protocol cap reaches the
+        // heap-constrained note; the constrained-heap rejection is exercised
+        // by the subprocess load test, but subprocess runs are not
+        // coverage-instrumented, so the note's arm is not schedulable from the
+        // instrumented suite (whose heap never binds).
+        /* v8 ignore next -- the heap-constrained message arm needs a host heap below the protocol cap. */
+        const heapNote = this.frameParseCapBytes < FRAME_PARSE_CAP_BYTES ? ` — this host's heap limits the parse to ${this.frameParseCapBytes} bytes, so the protocol cap of ${FRAME_PARSE_CAP_BYTES} would be unsafe` : ''
+        throw new Error(`dsh-code-runtime-python: config.${key} must not exceed ${limit} (a payload that large cannot cross the fd-3 frame PARSER, which rejects raw frames past ${this.frameParseCapBytes} bytes before decoding to bound host memory${heapNote} — a larger budget would admit a config whose honest child frames the host then rejects as a worker-exit), got ${String(this.config[key])}`)
      }
      // Reject a log budget too small to honor: the truncation marker alone
      // must serialize within the budget, or a marker-only truncated run
@@ -1446,6 +1516,29 @@ export class PythonCodeRuntime extends CodeRuntime {
        // fd 3 between finish() and close must not regrow the host buffer.
        /* v8 ignore next -- post-settlement data needs the child to outrace close after we decided. */
        if (settled) return
+        // Schedule ONE post-batch outstanding-call check per macrotask. The
+        // check must see the TRUE count — the live count is inflated by this
+        // batch's own frames (the finallys run on the microtask queue, which
+        // drains only when the macrotask ends), and a per-event snapshot is
+        // stale when flowing mode fires several 'data' events within one
+        // macrotask before any microtask drains. setImmediate runs after the
+        // current macrotask's microtasks, so the count is exact; the flag
+        // dedupes the check across the events of one macrotask. The threshold
+        // is STRICT: exactly MAX_PENDING_REPLIES outstanding calls are allowed,
+        // so a program that returns with calls it never awaited still
+        // completes (the done frame settles the run; the check no-ops on
+        // `settled`).
+        if (!postBatchCheckPending) {
+          postBatchCheckPending = true
+          setImmediate(() => {
+            postBatchCheckPending = false
+            /* v8 ignore next -- the done frame can settle the run between the schedule and this callback. */
+            if (settled) return
+            if (pendingCalls > MAX_PENDING_REPLIES) {
+              finish({ error: { kind: 'worker-exit', message: `call backlog exceeded ${MAX_PENDING_REPLIES} in-flight binding calls (a binding never settled)` } })
+            }
+          })
+        }
        pendingChunks.push(chunk)
        pendingBytes += chunk.length
        // Check the counter BEFORE the join, not the joined line afterwards:
@@ -1476,11 +1569,11 @@ export class PythonCodeRuntime extends CodeRuntime {
        // of the wire bytes. When this chunk DOES carry a newline the buffer
        // holds several frames, so the FIRST-FRAME check below (not this
        // counter, which charges them all) decides.
-        if (pendingBytes > FRAME_PARSE_CAP_BYTES && !chunk.includes(0x0a)) {
+        if (pendingBytes > this.frameParseCapBytes && !chunk.includes(0x0a)) {
          pendingChunks = []
          sealedBlocks = []
          pendingBytes = 0
-          finish({ error: { kind: 'worker-exit', message: `protocol frame exceeded ${FRAME_PARSE_CAP_BYTES} bytes on fd 3` } })
+          finish({ error: { kind: 'worker-exit', message: `protocol frame exceeded ${this.frameParseCapBytes} bytes on fd 3` } })
          return
        }
        // Bound the FRAGMENT COUNT as well as the byte total, but only AFTER the
@@ -1535,11 +1628,11 @@ export class PythonCodeRuntime extends CodeRuntime {
            }
            firstFrameLen += c.length
          }
-          if (sawNewline && firstFrameLen > FRAME_PARSE_CAP_BYTES) {
+          if (sawNewline && firstFrameLen > this.frameParseCapBytes) {
            pendingChunks = []
            sealedBlocks = []
            pendingBytes = 0
-            finish({ error: { kind: 'worker-exit', message: `protocol frame exceeded ${FRAME_PARSE_CAP_BYTES} bytes on fd 3` } })
+            finish({ error: { kind: 'worker-exit', message: `protocol frame exceeded ${this.frameParseCapBytes} bytes on fd 3` } })
            return
          }
          let buffered = Buffer.concat(sealedBlocks.length > 0 ? [...sealedBlocks, ...pendingChunks] : pendingChunks)
@@ -1724,6 +1817,19 @@ export class PythonCodeRuntime extends CodeRuntime {
            admit(message.text)
            return
          case 'done': {
+            // The call-backlog cap must also hold when the child finishes in
+            // the SAME batch as its flood: the post-macrotask check no-ops once
+            // this done frame settles the run, so a done arriving right after
+            // more than MAX_PENDING_REPLIES call frames in one data event would
+            // otherwise complete successfully with the outstanding closures
+            // left behind (a single sub-64 KiB write can carry 1025 compact
+            // calls plus a done). The strict threshold lets exactly
+            // MAX_PENDING_REPLIES outstanding calls — a program that returned
+            // without awaiting its calls — complete normally.
+            if (pendingCalls > MAX_PENDING_REPLIES) {
+              finish({ error: { kind: 'worker-exit', message: `call backlog exceeded ${MAX_PENDING_REPLIES} in-flight binding calls (a binding never settled)` } })
+              return
+            }
            if (message.error) {
              finish({ error: { kind: message.error.kind, message: capMessage(message.error.message, this.config.maxValueBytes) } })
              return
@@ -1788,17 +1894,12 @@ export class PythonCodeRuntime extends CodeRuntime {
              sendReply({ type: 'reply', id: message.id, ok: false, message: capMessage(`unknown binding ${preview}`, cap) })
              return
            }
-            // A binding that never settles (or resolves too slowly to keep up
-            // with the child's call rate) must not let the flood accumulate one
-            // async closure per frame until the wall clock: the reply cap only
-            // counts resolved calls, so it never trips for in-flight ones.
-            // Count the outstanding binding calls here, before dispatch, and
-            // release the slot in the body's finally — bounding in-flight
-            // closures to MAX_PENDING_REPLIES exactly like the reply backlog.
-            if (pendingCalls >= MAX_PENDING_REPLIES) {
-              finish({ error: { kind: 'worker-exit', message: `call backlog exceeded ${MAX_PENDING_REPLIES} in-flight binding calls (a binding never settled)` } })
-              return
-            }
+            // Count the outstanding binding call before dispatch and release the
+            // slot in the async body's finally. The CAP CHECK runs in the data
+            // handler's post-macrotask pass (where the finallys have drained),
+            // not here: a per-frame check would see every frame of one event as
+            // in-flight and false-positive on a legitimate gather of more than
+            // MAX_PENDING_REPLIES instant calls.
            pendingCalls += 1
            void (async () => {
              try {
@@ -1875,9 +1976,14 @@ export class PythonCodeRuntime extends CodeRuntime {
      // RESOLVED calls — `pendingReplies` grows after the await — so a child
      // flooding calls against a binding that never settles would accumulate
      // one async closure per frame until the wall clock without tripping it.
-      // Counted here before dispatch and released in the body's finally, the
-      // in-flight closures are bounded to the same MAX_PENDING_REPLIES.
+      // Counted here before dispatch and released in the body's finally; the
+      // data handler schedules a post-macrotask check (see there) that settles
+      // the run as worker-exit when the true outstanding count passes
+      // MAX_PENDING_REPLIES.
      let pendingCalls = 0
+      // Dedupes the post-batch outstanding-call check across the 'data' events
+      // of one macrotask (see the data handler).
+      let postBatchCheckPending = false
      let draining = false
      // Resolve when fd 3 can take another frame, OR when it is gone: a pipe
      // destroyed under the drain (child exited, close-deadline teardown) never
@@ -291,6 +291,12 @@ export function logTruncationMarker(maxBytes: number): string {
 * @returns the compact JSON encoding.
 */
 export function encodeJsonPlain(value: unknown): string {
+  // The task stack holds every member of the currently open containers — O(width)
+  // — but the encoded OUTPUT is itself O(total bytes) and the stack holds only
+  // references, so the walk's auxiliary state is same-order as its result; the
+  // metering walks (checkDoneValue/hasNonLosslessNumber) are the ones that must
+  // stay O(depth), since they can reject a wide payload without producing any
+  // output. Exempted by that same-order argument.
  type Task = { text: string } | { value: unknown }
  const chunks: string[] = []
  const tasks: Task[] = [{ value }]
@@ -430,9 +436,36 @@ export function checkDoneValue(value: unknown, maxBytes: number): { ok: true; by
  // classify differently (non-lossless vs over-budget), and the JSDoc promises
  // an over-budget value is rejected as over-budget regardless.
  let nonLossless = false
-  const stack: unknown[] = [value]
-  while (stack.length > 0) {
-    const current = stack.pop()
+  // One cursor per OPEN container (a values iterator for the root and arrays,
+  // an entries iterator for objects), mirroring hasNonLosslessNumber and the
+  // child's _check_done_value: a wide completion near the frame cap would
+  // otherwise copy every member's reference onto an explicit work stack —
+  // O(width) — OOMing the host after the parse already succeeded. The byte
+  // budget still bounds the walk: each member is metered as its cursor yields
+  // it, and the width lower-bound checks below bail an over-budget container
+  // before the cursor descends.
+  const cursors: Cursor[] = [{ kind: 'values', iter: [value].values() }]
+  while (cursors.length > 0) {
+    // The loop condition guarantees a top cursor.
+    const cursor = cursors.at(-1) as Cursor
+    const step = cursor.iter.next()
+    if (step.done === true) {
+      cursors.pop()
+      continue
+    }
+    let current: unknown
+    if (cursor.kind === 'entries') {
+      // Meter the key's escaped form without allocating it (same reason as the
+      // string branch), then add the colon separator, before the value's own
+      // bytes are counted.
+      const [key, member] = step.value as readonly [string, unknown]
+      const keyBytes = jsonStringBytesUpTo(key, maxBytes - bytes)
+      if (keyBytes === undefined) return { ok: false, reason: 'over-budget' }
+      bytes += keyBytes + 1
+      current = member
+    } else {
+      current = step.value
+    }
    if (typeof current === 'number') {
      // Flag a non-lossless number but keep counting its encoded bytes: a value
      // that is BOTH non-lossless and over-budget must classify as over-budget
@@ -451,13 +484,13 @@ export function checkDoneValue(value: unknown, maxBytes: number): { ok: true; by
      bytes += stringBytes
    } else if (Array.isArray(current)) {
      // Brackets plus one comma per gap; elements add themselves. Reject
-      // BEFORE enqueuing children: every element serializes to at least one
+      // BEFORE the cursor descends: every element serializes to at least one
      // byte, so a forged flat array far above the budget fails here without
-      // pushing its elements onto the host stack. (The array itself is already
-      // materialized by the upstream parse; this only bounds the extra stack.)
+      // the cursor yielding any of them. (The array itself is already
+      // materialized by the upstream parse; this only bounds the extra walk.)
      bytes += 2 + (current.length > 1 ? current.length - 1 : 0)
      if (bytes + current.length > maxBytes) return { ok: false, reason: 'over-budget' }
-      for (const item of current) stack.push(item)
+      cursors.push({ kind: 'values', iter: (current as unknown[]).values() })
    } else if (typeof current === 'object' && current !== null) {
      const record = current as Record<string, unknown>
      // Count own keys with for...in + hasOwn. This IS O(keys) — JS has no lazy
@@ -469,15 +502,7 @@ export function checkDoneValue(value: unknown, maxBytes: number): { ok: true; by
      for (const key in record) if (Object.hasOwn(record, key)) count += 1
      bytes += 2 + (count > 1 ? count - 1 : 0)
      if (bytes + count * 4 > maxBytes) return { ok: false, reason: 'over-budget' }
-      for (const key in record) {
-        if (!Object.hasOwn(record, key)) continue
-        // Meter the key's escaped form without allocating it (same reason as the
-        // string branch), then add the colon separator. `+ 1` for the `:`.
-        const keyBytes = jsonStringBytesUpTo(key, maxBytes - bytes)
-        if (keyBytes === undefined) return { ok: false, reason: 'over-budget' }
-        bytes += keyBytes + 1
-        stack.push(record[key])
-      }
+      cursors.push({ kind: 'entries', iter: ownEntries(record) })
    } else {
      bytes += Buffer.byteLength(scalarJson(current), 'utf8')
    }
@@ -538,6 +563,31 @@ export function hasUnsafeIntegerToken(line: string): boolean {
  return false
 }

+/**
+ * One open container in checkDoneValue's cursor walk: a values iterator (the
+ * root and arrays) or an entries iterator (objects, so each key's escaped
+ * bytes can be metered when the entry is reached). A cursor bounds the walk's
+ * auxiliary state to O(depth), not O(width).
+ */
+type Cursor =
+  | { kind: 'values'; iter: Iterator<unknown> }
+  | { kind: 'entries'; iter: Iterator<readonly [string, unknown]> }
+
+/**
+ * Lazily yield one plain object's own enumerable [key, value] entries. The
+ * key escapes are metered when {@link checkDoneValue}'s cursor walk reaches
+ * each entry, so a wide object never materializes a member list: each entry
+ * is produced straight off the already-parsed record, and the escaped key
+ * bytes are counted without building the escaped string.
+ * @param record - a JSON-parse-produced object.
+ * @yields each own enumerable [key, value] pair, in key order.
+ */
+function* ownEntries(record: Record<string, unknown>): Generator<readonly [string, unknown]> {
+  for (const key in record) {
+    if (Object.hasOwn(record, key)) yield [key, record[key]]
+  }
+}
+
 /**
 * Lazily yield one plain object's own enumerable property values. A generator
 * (not `Object.values`/`Object.entries`) because {@link hasNonLosslessNumber}
@@ -3,7 +3,7 @@ import { existsSync } from 'node:fs'
 import { dirname } from 'node:path'
 import { PassThrough } from 'node:stream'
 import { afterEach, describe, expect, it, vi } from 'vitest'
-import { Context } from 'cordis'
+import { Context } from '@deepseek-ai/cordis'

 /**
 * A synchronous `proto.write` throw on the fd-3 pipe is the one boot path a real
@@ -1,5 +1,5 @@
 import { describe, expect, it } from 'vitest'
-import { checkDoneValue, encodeJsonPlain, hasNonLosslessNumber, hasUnsafeIntegerToken, logTruncationMarker, validateChildFrame } from '../src/index.ts'
+import { checkDoneValue, encodeJsonPlain, hasNonLosslessNumber, hasUnsafeIntegerToken, hostFrameParseCeiling, logTruncationMarker, validateChildFrame } from '../src/index.ts'

 describe('logTruncationMarker', () => {
  it('names the configured byte budget', () => {
@@ -301,4 +301,46 @@ describe('checkDoneValue', () => {
    expect(encodeJsonPlain(v)).toBe('[1152921504606846976]')
    expect(checkDoneValue(v, 100)).toEqual({ ok: true, bytes: Buffer.byteLength('[1152921504606846976]', 'utf8') })
  })
+
+  it('walks wide arrays and objects one member at a time', () => {
+    // A completion value has a seam byte budget, but the budget alone does not
+    // bound the traversal's AUXILIARY state: a wide value near the frame cap
+    // (millions of members) must not have every member's reference copied onto
+    // a work stack — that O(width) allocation would OOM the host after the
+    // parse already succeeded. The walk holds one cursor per nesting level, so
+    // a wide value meters exactly and a violation anywhere in it is found
+    // wherever it sits.
+    const wideArray = new Array(2_000_000).fill(0) as unknown[]
+    const arrayJson = `[${wideArray.join(',')}]`
+    const arrayExact = Buffer.byteLength(arrayJson, 'utf8')
+    expect(checkDoneValue(wideArray, arrayExact)).toEqual({ ok: true, bytes: arrayExact })
+    expect(checkDoneValue(wideArray, arrayExact - 1)).toEqual({ ok: false, reason: 'over-budget' })
+    // Last element, so the cursor must run the whole breadth lazily to find it.
+    wideArray[wideArray.length - 1] = -0
+    expect(checkDoneValue(wideArray, arrayExact)).toEqual({ ok: false, reason: 'non-lossless' })
+    wideArray[wideArray.length - 1] = 0
+    const wideObject: Record<string, unknown> = {}
+    for (let i = 0; i < 100_000; i++) wideObject[`k${i}`] = i
+    const objectExact = Buffer.byteLength(JSON.stringify(wideObject), 'utf8')
+    expect(checkDoneValue(wideObject, objectExact)).toEqual({ ok: true, bytes: objectExact })
+    expect(checkDoneValue(wideObject, objectExact - 1)).toEqual({ ok: false, reason: 'over-budget' })
+    wideObject.last = -0
+    expect(checkDoneValue(wideObject, Buffer.byteLength(JSON.stringify(wideObject), 'utf8'))).toEqual({ ok: false, reason: 'non-lossless' })
+  })
+})
+
+describe('hostFrameParseCeiling', () => {
+  it('caps the parse at the protocol limit on a default heap and lower on a constrained one', () => {
+    // The raw-byte frame cap does not protect the host heap: JSON.parse of a
+    // wide-object frame materializes several times the raw bytes in property
+    // storage, so the effective cap is min(protocol cap, heap-derived
+    // ceiling). A default Node heap (~4 GiB) never binds.
+    expect(hostFrameParseCeiling(4 * 1024 * 1024 * 1024)).toBe(64 * 1024 * 1024)
+    // A constrained host (--max-old-space-size=256 reports a ~304 MiB limit)
+    // derives floor((304 - 64) / 16) = 15 MiB: a 50 MiB budget would be
+    // rejected at load, where the address-space gate alone would admit it.
+    expect(hostFrameParseCeiling(304 * 1024 * 1024)).toBe(15 * 1024 * 1024)
+    // A tiny heap leaves almost no parse room — the load gate fails loud.
+    expect(hostFrameParseCeiling(128 * 1024 * 1024)).toBe(4 * 1024 * 1024)
+  })
 })
@@ -1,10 +1,11 @@
-import { existsSync, realpathSync, rmSync, statSync, writeFileSync } from 'node:fs'
+import { execFileSync } from 'node:child_process'
+import { existsSync, mkdtempSync, realpathSync, rmSync, statSync, writeFileSync } from 'node:fs'
 import { mkdtemp, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
-import { basename, dirname, join, relative } from 'node:path'
-import { describe, expect, it, vi } from 'vitest'
+import { basename, dirname, join, relative, resolve } from 'node:path'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
-import { PythonCodeRuntime, readProcessStart, resolvePythonBin } from '../src/index.ts'
+import { PythonCodeRuntime, hostFrameParseCeiling, readProcessStart, resolvePythonBin } from '../src/index.ts'
 import { logTruncationMarker } from '../src/protocol.ts'
 import type { Config } from '../src/index.ts'

@@ -27,9 +28,16 @@ import type { CodeBindingFunction, CodeJsonValue, CodeRunResult } from '@deepsee
 * records the same race and solves it with argv-based identity; recording the
 * mkdtempSync results is the fs-mock equivalent.
 */
-const { failNextCopyOf, stagedDirs } = vi.hoisted(() => ({
+const { failNextCopyOf, stagedDirs, tempDirs, tempFiles } = vi.hoisted(() => ({
  failNextCopyOf: { value: undefined as string | undefined },
  stagedDirs: [] as string[],
+  // Test-created temp dirs/files, registered by the helpers below and removed
+  // after each test: a suite run over real python3 subprocesses must not
+  // permanently accumulate `dsh-*` fixtures in the shared tmpdir (the runtime
+  // cleans its own per-run staging dir; these are the stubs and wrappers the
+  // tests themselves build).
+  tempDirs: [] as string[],
+  tempFiles: [] as string[],
 }))
 vi.mock('node:fs', async (importOriginal) => {
  const actual = await importOriginal<typeof import('node:fs')>()
@@ -68,6 +76,27 @@ function tools(functions: Record<string, CodeBindingFunction>) {
  return [{ global: 'tools', functions }]
 }

+/** Create a test temp dir registered for afterEach removal. */
+async function makeTempDir(prefix: string): Promise<string> {
+  const dir = await mkdtemp(join(tmpdir(), prefix))
+  tempDirs.push(dir)
+  return dir
+}
+
+/** Synchronous variant of {@link makeTempDir} for the PATH-stub fixtures. */
+function makeTempDirSync(prefix: string): string {
+  const dir = mkdtempSync(join(tmpdir(), prefix))
+  tempDirs.push(dir)
+  return dir
+}
+
+// Remove every fixture this file created, so repeated runs do not accumulate
+// `dsh-*` directories and wrappers in the shared tmpdir.
+afterEach(() => {
+  for (const dir of tempDirs.splice(0)) rmSync(dir, { recursive: true, force: true })
+  for (const file of tempFiles.splice(0)) rmSync(file, { force: true })
+})
+
 describe('PythonCodeRuntime — seam descriptors and misuse', () => {
  it('registers the seam descriptors', async () => {
    const { runtime } = await setup()
@@ -149,6 +178,66 @@ describe('PythonCodeRuntime — seam descriptors and misuse', () => {
    await boundary.dispose()
  })

+  it('rejects a completion budget whose frame a constrained host heap cannot safely parse', async () => {
+    // The load gate bounds the CHILD's build-and-encode under RLIMIT_AS; it
+    // does not bound the HOST's JSON.parse, which materializes several times a
+    // wide frame's raw bytes in property storage. In a child node with a
+    // 128 MiB old space the heap-derived frame cap is ~14 MiB, so a 50 MiB
+    // budget is rejected at load even though the address-space gate alone
+    // would admit it (50 MiB * 12 = 600 MiB < 1 GiB - 64 MiB).
+    const script = [
+      "import { Context } from '@deepseek-ai/cordis'",
+      "import { PythonCodeRuntime } from './packages/experimental/code-runtime-python/src/index.ts'",
+      'const ctx = new Context()',
+      'try {',
+      '  await ctx.plugin(PythonCodeRuntime, { maxValueBytes: 50 * 1024 * 1024, addressSpaceMb: 1024 })',
+      "  console.log('LOADED')",
+      '  process.exit(1)',
+      '} catch (error) {',
+      "  console.log('REJECTED:' + (error instanceof Error ? error.message : String(error)))",
+      '  process.exit(0)',
+      '}',
+    ].join('\n')
+    const out = execFileSync(process.execPath, ['--max-old-space-size=128', '--import', 'tsx', '-e', script], {
+      cwd: resolve(import.meta.dirname, '../../../..'),
+      encoding: 'utf8',
+      timeout: 60_000,
+      env: { ...process.env, TSX_TSCONFIG_PATH: resolve(import.meta.dirname, '../../../../tsconfig.json') },
+    })
+    expect(out).toContain('REJECTED:')
+    expect(out).toContain('must not exceed')
+  }, 60_000)
+
+  it('parses a worst-shape frame at the derived cap on a constrained heap', async () => {
+    // The host-heap frame cap must be measured against the WORST parse shape —
+    // a dict of many short unique keys, which forces dictionary-mode property
+    // storage plus interned keys (~6.4x at 3M keys, trending up), not the ~3x
+    // of a repeated-key dict. A child node with a 128 MiB old space (~176 MiB
+    // heap limit) derives a cap of floor((176 - 64) / 16) = 7 MiB; the
+    // subprocess builds a unique-key dict whose frame is AT that cap and
+    // parses it, which must survive. Verified fail-before: with the multiple
+    // at 8 the derived cap doubles to 14 MiB and the same subprocess OOMs
+    // during the parse (plain JS, no tsx — the frame and parse are builtins).
+    const cap = hostFrameParseCeiling(176 * 1024 * 1024)
+    const script = [
+      `const cap = ${cap}`,
+      // Each entry "k<base36>:1," is ~9-12 raw bytes; a few hundred thousand
+      // unique keys put the frame just at the cap.
+      'const count = Math.floor(cap / 12)',
+      'const obj = {}',
+      'for (let i = 0; i < count; i++) obj[`k${i.toString(36)}`] = 1',
+      'const frame = JSON.stringify(obj)',
+      "if (Buffer.byteLength(frame, 'utf8') > cap) throw new Error('frame over cap: ' + frame.length)",
+      'JSON.parse(frame)',
+      "console.log('SURVIVED:' + Buffer.byteLength(frame, 'utf8'))",
+    ].join('\n')
+    const out = execFileSync(process.execPath, ['--max-old-space-size=128', '-e', script], {
+      encoding: 'utf8',
+      timeout: 60_000,
+    })
+    expect(out).toContain('SURVIVED:')
+  }, 60_000)
+
  it('rejects a pythonBin that spawn() would throw on, at load', async () => {
    // Both values pass the string schema and both make `spawn` throw
    // SYNCHRONOUSLY from inside run() — ERR_INVALID_ARG_VALUE for the empty
@@ -170,8 +259,8 @@ describe('PythonCodeRuntime — seam descriptors and misuse', () => {
    // The message distinguishes the explicit-path failure from a basename that
    // simply does not resolve on PATH.
    const nodePath = await import('node:path')
-    const { mkdtempSync, writeFileSync, mkdirSync } = await import('node:fs')
-    const dir = mkdtempSync(nodePath.join(tmpdir(), 'dsh-bad-bin-'))
+    const { writeFileSync, mkdirSync } = await import('node:fs')
+    const dir = makeTempDirSync('dsh-bad-bin-')
    const notExecutable = nodePath.join(dir, 'not-executable')
    writeFileSync(notExecutable, '#!/bin/sh\nexit 0\n') // Regular file, but no X bit.
    const directory = nodePath.join(dir, 'is-a-directory')
@@ -386,10 +475,9 @@ describe('PythonCodeRuntime — seam descriptors and misuse', () => {
    // real interpreter is used.
    const cp = await import('node:child_process')
    const nodePath = await import('node:path')
-    const { mkdtempSync, mkdirSync } = await import('node:fs')
-    const { tmpdir } = await import('node:os')
+    const { mkdirSync } = await import('node:fs')
    const realPythonDir = nodePath.dirname(cp.execFileSync('which', ['python3'], { encoding: 'utf8' }).trim())
-    const fakeDir = mkdtempSync(nodePath.join(tmpdir(), 'dsh-fake-bin-'))
+    const fakeDir = makeTempDirSync('dsh-fake-bin-')
    mkdirSync(nodePath.join(fakeDir, 'python3')) // A directory named python3, executable by default.
    vi.stubEnv('PATH', `${fakeDir}:${realPythonDir}`)
    try {
@@ -608,7 +696,7 @@ describe('PythonCodeRuntime — seam descriptors and misuse', () => {
    // `os.tmpdir()`, so pointing it at a path that is not a directory makes the
    // real call fail without stubbing the module under test.
    const previous = process.env.TMPDIR
-    const notADirectory = join(await mkdtemp(join(tmpdir(), 'dsh-staging-')), 'file')
+    const notADirectory = join(await makeTempDir('dsh-staging-'), 'file')
    await writeFile(notADirectory, '')
    process.env.TMPDIR = notADirectory
    try {
@@ -694,7 +782,7 @@ describe('PythonCodeRuntime — inherited resource limits', () => {
    // `pythonBin` is the honest lever: a wrapper that lowers RLIMIT_AS and then
    // execs the real interpreter reproduces the inherited-limit condition
    // without touching this test process's own limits.
-    const dir = await mkdtemp(join(tmpdir(), 'dsh-rlimit-'))
+    const dir = await makeTempDir('dsh-rlimit-')
    const wrapper = join(dir, 'python3-capped')
    // 256 MiB, half the 512 MiB addressSpaceMb default, so the requested cap is
    // unambiguously above the inherited ceiling.
@@ -722,7 +810,7 @@ describe('PythonCodeRuntime — inherited resource limits', () => {
    // rejected. The rejection surfaces as an 'exception' (bootstrap's
    // setrlimit-phase failure class), not a mid-run OOM. The repro is Linux-only
    // (macOS ignores `ulimit -v`); there the run proceeds.
-    const dir = await mkdtemp(join(tmpdir(), 'dsh-rlimit-'))
+    const dir = await makeTempDir('dsh-rlimit-')
    const wrapper = join(dir, 'python3-tight')
    await writeFile(wrapper, `#!/bin/sh\nulimit -v 131072\nexec "${PYABS}" "$@"\n`, { mode: 0o755 })
    const { runtime } = await setup({ pythonBin: wrapper, maxLogBytes: 32 * 1024 * 1024, addressSpaceMb: 512 })
@@ -772,7 +860,7 @@ describe('PythonCodeRuntime — inherited resource limits', () => {
    // requested soft (`cpuSeconds`) sits above the inherited soft — the case that
    // exposed the bug. RLIMIT_CPU is used because macOS ignores `ulimit -v`
    // (RLIMIT_AS), which is exactly why the backend skips address space there.
-    const dir = await mkdtemp(join(tmpdir(), 'dsh-rlimit-soft-'))
+    const dir = await makeTempDir('dsh-rlimit-soft-')
    const wrapper = join(dir, 'python3-soft-capped')
    // Soft CPU 5 s, well below the configured 30 s, hard left unlimited.
    await writeFile(wrapper, `#!/bin/sh\nulimit -S -t 5\nexec "${PYABS}" "$@"\n`, { mode: 0o755 })
@@ -797,7 +885,7 @@ describe('PythonCodeRuntime — inherited resource limits', () => {
    // timeout. This uses `ulimit -t 2` (hard == 2, so the soft is lowered to 1)
    // and leaves SIGXCPU unhandled, so the kernel terminates the busy loop at
    // 1 s with SIGXCPU and the host classifies it as a timeout.
-    const dir = await mkdtemp(join(tmpdir(), 'dsh-rlimit-dual-'))
+    const dir = await makeTempDir('dsh-rlimit-dual-')
    const wrapper = join(dir, 'python3-dual-capped')
    // Both soft and hard CPU 2 s; configured cpuSeconds 30 s.
    await writeFile(wrapper, `#!/bin/sh\nulimit -t 2\nexec "${PYABS}" "$@"\n`, { mode: 0o755 })
@@ -848,6 +936,7 @@ describe('PythonCodeRuntime — inherited resource limits', () => {
    // before model code runs, so a busy loop still ends as a timeout rather
    // than running to the hard limit and being misclassified as worker-exit.
    const wrapper = join(tmpdir(), `dsh-xcpu-ignore-${process.pid}.sh`)
+    tempFiles.push(wrapper)
    writeFileSync(wrapper, `#!/bin/sh\ntrap "" XCPU\nexec "${PYABS}" "$@"\n`, { mode: 0o755 })
    try {
      const { runtime } = await setup({ maxWallMs: 30_000, cpuSeconds: 1, pythonBin: wrapper })
@@ -900,7 +989,7 @@ describe('PythonCodeRuntime — inherited resource limits', () => {
    // the inherited limit. The wrapper sets a 1 s soft CPU limit; the program
    // traps SIGXCPU and busy-loops past it, then returns — the recheck must
    // re-deliver SIGXCPU so the host classifies the run as a timeout.
-    const dir = await mkdtemp(join(tmpdir(), 'dsh-cpu-recheck-'))
+    const dir = await makeTempDir('dsh-cpu-recheck-')
    const wrapper = join(dir, 'python3-cpu-capped')
    await writeFile(wrapper, `#!/bin/sh\nulimit -S -t 1\nexec "${PYABS}" "$@"\n`, { mode: 0o755 })
    const { runtime } = await setup({ pythonBin: wrapper, cpuSeconds: 30, maxWallMs: 12_000 })
@@ -3587,7 +3676,7 @@ describe('PythonCodeRuntime — budgets, termination, disposal', () => {
    // means whether the killed descendant is reaped or lingers as a zombie (a
    // SIGKILL'd process runs no more code either way). It sleeps 30 s as a safety
    // net so a broken fix cannot leak it forever.
-    const handoff = await mkdtemp(join(tmpdir(), 'dsh-samegroup-'))
+    const handoff = await makeTempDir('dsh-samegroup-')
    const readyMarker = join(handoff, 'ready')
    const heartbeat = join(handoff, 'heartbeat')
    const { runtime } = await setup({ maxWallMs: 10_000, graceMs: 300 })
@@ -3652,7 +3741,7 @@ describe('PythonCodeRuntime — budgets, termination, disposal', () => {
    // is called; the heartbeat must be stale BY THE TIME dispose() resolves —
    // proving teardown waited for the reap, not merely that the reap eventually
    // happened.
-    const handoff = await mkdtemp(join(tmpdir(), 'dsh-dispose-quiesce-'))
+    const handoff = await makeTempDir('dsh-dispose-quiesce-')
    const readyMarker = join(handoff, 'ready')
    const heartbeat = join(handoff, 'heartbeat')
    const { runtime, fiber } = await setup({ maxWallMs: 10_000, graceMs: 300 })
@@ -3705,7 +3794,7 @@ describe('PythonCodeRuntime — budgets, termination, disposal', () => {
    // rather than cancel the unfired escalation — otherwise a SIGTERM-ignoring
    // same-group survivor is released for good. A synchronous busy-loop after
    // run() resolves reproduces the block deterministically.
-    const handoff = await mkdtemp(join(tmpdir(), 'dsh-deadline-'))
+    const handoff = await makeTempDir('dsh-deadline-')
    const readyMarker = join(handoff, 'ready')
    const heartbeat = join(handoff, 'heartbeat')
    const graceMs = 300
@@ -5309,52 +5398,203 @@ describe('PythonCodeRuntime — hostile peer', () => {
    expect(result.error?.message).toContain('call backlog exceeded')
  }, 30_000)

-  it('compacts the reply queue mid-drain without dropping pending frames', async () => {
-    // A reply larger than the writable high-water mark makes the FIRST write
-    // return false, suspending the drain loop while the child's synchronous
-    // flood starves the reply pump; the frames queued behind it push the
-    // drain's consumed head past MAX_PENDING_REPLIES, so the resumed drain
-    // compacts the queue mid-run. The child reads fd 3 itself (blocking the
-    // asyncio pump, so its reads cannot race the host's pushes) and sends a
-    // second wave of calls AFTER reading part of the first wave's replies —
-    // those replies are still pending when the drain's head crosses the
-    // compaction bound, so a compaction that dropped pending frames would
-    // leave the child's reply count short and the read loop spinning to the
-    // wall clock. The second wave is sent mid-delivery (not with the first
-    // flood): pushing it earlier would trip the 1024-pending reply cap
-    // before the drain resumed.
+  it('runs a legitimate gather of more than 1024 concurrent binding calls', async () => {
+    // The in-flight call cap must not count a synchronous batch of instant
+    // calls: the async bodies' finallys run on the microtask queue, which
+    // drains only between 'data' events, so a per-frame check would trip on
+    // the 1025th frame of a single event even though every binding settled
+    // immediately — killing a valid large concurrent gather as worker-exit.
+    // The cap is checked at event boundaries (after the microtask queue
+    // drained), so this gather of 1025 instant calls completes.
+    const { runtime } = await setup({ maxWallMs: 30_000 })
+    const result = await runtime.run({
+      program: [
+        'import asyncio',
+        'return len(await asyncio.gather(*[tools.echo(i) for i in range(1025)]))',
+      ].join('\n'),
+      bindings: [{ global: 'tools', functions: { echo: async (args: unknown) => args as CodeJsonValue } }],
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.value).toBe(1025)
+  }, 30_000)
+
+  it('completes normally when a program returns with binding calls still outstanding', async () => {
+    // The in-flight call cap refuses to admit NEW calls past the bound; it must
+    // not reclassify a `done` frame as worker-exit just because the program
+    // returned with calls it started but never awaited. The child schedules
+    // exactly 1024 slow bindings (still pending when the program returns), so
+    // the done frame arrives with the outstanding count AT the cap — the event
+    // must complete with its value, not settle as `call backlog exceeded`.
+    const { runtime } = await setup({ maxWallMs: 30_000 })
+    const result = await runtime.run({
+      program: [
+        'import asyncio',
+        'for i in range(1024):',
+        '    asyncio.create_task(tools.slow(i))',
+        'await asyncio.sleep(0.2)',
+        'return "done"',
+      ].join('\n'),
+      bindings: [{
+        global: 'tools',
+        functions: { slow: async () => { await new Promise((resolve) => { setTimeout(resolve, 5_000) }); return 1 } },
+      }],
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.value).toBe('done')
+  }, 30_000)
+
+  it('settles a single-batch never-settling flood as worker-exit without further frames', async () => {
+    // The outstanding-call cap must take effect even when the whole flood fits
+    // in ONE data event: a per-event admission snapshot never re-checks once no
+    // further frames arrive, so a single 62 KiB write of 1025 compact calls
+    // against a never-settling binding would otherwise wait out the full wall
+    // clock instead of tripping the cap. The post-macrotask check runs after
+    // the batch's finallys (which never run for this binding) and settles the
+    // run as worker-exit long before maxWallMs.
    const { runtime } = await setup({ maxWallMs: 30_000 })
    const result = await runtime.run({
      program: [
        'import os, time',
+        'frame = b\'{"type":"call","id":%d,"global":"tools","name":"hang","args":{}}\\n\'',
+        'payload = b"".join(frame % i for i in range(1025))',
+        'view = memoryview(payload)',
+        'while view:',
+        '    view = view[os.write(3, view):]',
+        'time.sleep(30)',
+        'return "unreachable"',
+      ].join('\n'),
+      bindings: [{ global: 'tools', functions: { hang: async () => await new Promise<never>(() => {}) } }],
+    })
+    expect(result.error?.kind).toBe('worker-exit')
+    expect(result.error?.message).toContain('call backlog exceeded')
+  }, 30_000)
+
+  it('runs a burst of 1300 instant calls whose frames split across pipe reads', async () => {
+    // Flowing mode can fire several 'data' events within one macrotask, before
+    // any microtask drains, so a per-event snapshot of the outstanding count
+    // could see the first chunk's in-flight calls in the second chunk's check
+    // and false-positive on a legitimate burst. The post-macrotask check always
+    // sees the true count (all finallys have run), so this burst of compact
+    // frames — sized so the pipe read splits it — completes with all results.
+    const { runtime } = await setup({ maxWallMs: 30_000 })
+    const result = await runtime.run({
+      program: [
+        'import asyncio',
+        'return len(await asyncio.gather(*[t.e(i) for i in range(1300)]))',
+      ].join('\n'),
+      bindings: [{ global: 't', functions: { e: async (args: unknown) => args as CodeJsonValue } }],
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.value).toBe(1300)
+  }, 30_000)
+
+  it('settles as worker-exit when a done frame lands in the same batch as a call flood', async () => {
+    // A done frame processed in the SAME data event as more than 1024 call
+    // frames settles the run before the post-macrotask check runs (which no-ops
+    // once settled), so a child could finish "successfully" while leaving the
+    // outstanding closures behind — one sub-64 KiB write carries 1025 compact
+    // calls plus a done. The done handler re-checks the count before accepting
+    // the frame, so the run settles as worker-exit with the call-backlog
+    // message instead.
+    const { runtime } = await setup({ maxWallMs: 30_000 })
+    const result = await runtime.run({
+      program: [
+        'import os, time',
+        'frame = b\'{"type":"call","id":%d,"global":"tools","name":"hang","args":{}}\\n\'',
+        'payload = b"".join(frame % i for i in range(1025)) + b\'{"type":"done","value":1}\\n\'',
+        'view = memoryview(payload)',
+        'while view:',
+        '    view = view[os.write(3, view):]',
+        'time.sleep(30)',
+        'return "unreachable"',
+      ].join('\n'),
+      bindings: [{ global: 'tools', functions: { hang: async () => await new Promise<never>(() => {}) } }],
+    })
+    expect(result.error?.kind).toBe('worker-exit')
+    expect(result.error?.message).toContain('call backlog exceeded')
+  }, 30_000)
+
+  it('rejects a completion whose dict keys fold to one JSON member', async () => {
+    // `_dump_string` folds a spelled-out surrogate pair into its astral code
+    // point, so `"\ud83d\ude00"` and `"\U0001f600"` are DIFFERENT Python keys
+    // that encode to the SAME JSON member — the host's JSON.parse would
+    // silently drop one of them, violating the lossless-JSON completion
+    // contract. The child's meter rejects the collision before encoding.
+    const { runtime } = await setup()
+    const result = await runtime.run({
+      program: 'return {"\\ud83d\\ude00": 1, "\\U0001f600": 2}',
+      bindings: [],
+    })
+    expect(result.error?.kind).toBe('invalid-output')
+    expect(result.error?.message).toContain('duplicate dict key')
+  }, 30_000)
+
+  it('rejects binding arguments whose dict keys fold to one JSON member', async () => {
+    // The same collision on the binding-argument path: the call is rejected as
+    // not lossless JSON, so the program's `await` raises and the program
+    // surfaces the rejection message.
+    const { runtime } = await setup()
+    const result = await runtime.run({
+      program: [
+        'try:',
+        '    await tools.echo({"\\ud83d\\ude00": 1, "\\U0001f600": 2})',
+        '    return "no-error"',
+        'except Exception as e:',
+        '    return str(e)',
+      ].join('\n'),
+      bindings: [{ global: 'tools', functions: { echo: async (args: unknown) => args as CodeJsonValue } }],
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.value).toContain('duplicate dict key')
+  }, 30_000)
+
+  it('compacts the reply queue mid-drain without dropping pending frames', async () => {
+    // A reply larger than the writable high-water mark makes the FIRST write
+    // return false, suspending the drain loop; the frames queued behind it
+    // push the drain's consumed head past MAX_PENDING_REPLIES, so the resumed
+    // drain compacts the queue mid-run. The child reads fd 3 itself (blocking
+    // the asyncio pump, so its reads cannot race the host's pushes) and sends
+    // a second wave of calls AFTER reading part of the first wave's replies —
+    // those replies are still pending when the drain's head crosses the
+    // compaction bound, so a compaction that dropped pending frames would
+    // leave the child's reply count short and the read loop spinning to the
+    // wall clock. No fixed sleep: the child's reads pace at the drain's
+    // delivery rate (each write blocks until the child reads), and the host
+    // finishes pushing all of a wave within milliseconds — orders of magnitude
+    // before the head crosses the bound — so the queue is always full at the
+    // splice. Newlines are counted per chunk (each reply carries exactly one),
+    // never by re-scanning the accumulated total, which would be O(n²).
+    const { runtime } = await setup({ maxWallMs: 60_000 })
+    const result = await runtime.run({
+      program: [
+        'import os',
        'frame = b\'{"type":"call","id":%d,"global":"tools","name":"big","args":{}}\\n\'',
        'for i in range(1024):',
        '    view = memoryview(frame % i)',
        '    while view:',
        '        view = view[os.write(3, view):]',
-        'time.sleep(0.5)',
-        'total = b""',
-        'while total.count(b"\\n") < 500:',
+        'seen = 0',
+        'while seen < 500:',
        '    chunk = os.read(3, 65536)',
        '    if not chunk:',
        '        break',
-        '    total += chunk',
+        '    seen += chunk.count(b"\\n")',
        'for i in range(500):',
        '    view = memoryview(frame % (1024 + i))',
        '    while view:',
        '        view = view[os.write(3, view):]',
-        'while total.count(b"\\n") < 1524:',
+        'while seen < 1524:',
        '    chunk = os.read(3, 65536)',
        '    if not chunk:',
        '        break',
-        '    total += chunk',
+        '    seen += chunk.count(b"\\n")',
        'return "done"',
      ].join('\n'),
      bindings: [{ global: 'tools', functions: { big: async () => 'x'.repeat(65 * 1024) } }],
    })
    expect(result.error).toBeUndefined()
    expect(result.value).toBe('done')
-  }, 30_000)
+  }, 60_000)

  it('bounds a flood of zero-byte log lines through the per-entry separator charge', async () => {
    // Blank print() lines carry zero content bytes; without the +1 separator
@@ -20,14 +20,14 @@
    {
      "path": "../../code-runtime/code-runtime"
    },
-    {
-      "path": "../../core/session"
-    },
    {
      "path": "../../runtime-diagnostics/invariants"
    },
    {
      "path": "../../util/timeout"
+    },
+    {
+      "path": "../../util/values"
    }
  ]
 }