diff --git a/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.i18n.yaml b/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.i18n.yaml index 1c84c7dc95..1dca984554 100644 --- a/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.i18n.yaml +++ b/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.i18n.yaml @@ -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 .agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.md -2026-07-31-code-runtime-python-settlement-fixes.md: 27e6a78e5d164e31ae4d5bd24170e5c254d37e44 -2026-07-31-code-runtime-python-settlement-fixes.zh.md: 668f17cf2b504eb339b11fe311cc3593c99c569b +2026-07-31-code-runtime-python-settlement-fixes.md: fbf562ac7eeb407d137905649160916c6ce63c4f +2026-07-31-code-runtime-python-settlement-fixes.zh.md: 594d5074a5b96885416c4019d228eaafd10b08ae diff --git a/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.md b/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.md index 27e6a78e5d..fbf562ac7e 100644 --- a/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.md +++ b/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.md @@ -1,4 +1,4 @@ -# Agent Note: Three settlement and framing fixes in the CPython backend +# Agent Note: Settlement, framing, and lifecycle fixes in the CPython backend Status: implemented @@ -6,11 +6,11 @@ English | [中文](2026-07-31-code-runtime-python-settlement-fixes.zh.md) ## Problem -The CPython subprocess backend for Code Mode, built on the [fd-3 frame protocol](../architecture/2026-07-31-code-runtime-python-fd3-protocol.md), resolves every program outcome as a `CodeRunResult` and rejects `run()` only for seam misuse. Three defects broke that contract in ways unit coverage did not surface, because each hid behind a `/* v8 ignore */`, a captured-callable comment that read as a fix but was not, or a memory effect invisible through the seam. They were found by review of the backend as it stood, not by a failing test, so each fix ships with a test that fails without it. +The CPython subprocess backend for Code Mode, built on the [fd-3 frame protocol](../architecture/2026-07-31-code-runtime-python-fd3-protocol.md), resolves every program outcome as a `CodeRunResult`, rejects `run()` only for seam misuse, and disposes to quiescence so no subprocess outlives the fiber. A sequence of review passes surfaced defects that broke those contracts in ways unit coverage did not catch — each hid behind a `/* v8 ignore */`, a captured-callable that read as a fix but was not, a memory effect invisible through the seam, a load-time bound that double-counted, a process-group escalation that a survivor could outlast, or a cross-event-loop completion that silently deadlocked. Each fix ships with a test that fails without it. ## Decision -Three independent corrections, each in the package that owns the defect. +Six independent corrections, each in the package that owns the defect. ### Boot-write failure no longer rejects run() @@ -24,11 +24,23 @@ In [`py/bootstrap.py`](../../../../packages/code-runtime/code-runtime-python/py/ Also in `src/index.ts`, after the newline loop over a `Buffer.concat` of the pending fd-3 chunks, the leftover partial line was carried forward as the `subarray` VIEW it was sliced to. A view keeps the entire concat backing allocation alive, so a large frame followed by a tiny trailing fragment pinned a whole frame's worth of memory while `pendingBytes` — set to the fragment's length — reported far less than was retained. The residual is now detached into a fresh right-sized `Buffer` via the exported `detachResidual` helper, letting the concat allocation be collected and keeping `pendingBytes` an honest measure. +### Output-cap load bound is ceiling minus envelope, not divided by six + +The load-time check that rejects a `maxLogBytes`/`maxValueBytes` larger than one fd-3 frame can carry divided the frame ceiling by six for worst-case escape expansion. But both budgets are metered in ALREADY-ESCAPED serialized bytes — the host log ledger charges `Buffer.byteLength(JSON.stringify(text))` and `checkDoneValue` measures the escaped form — so a payload admitted under the cap occupies at most `cap + envelope` on the wire; escaping is inside the charge and must not be multiplied in again. The bound is now `FRAME_CEILING_BYTES - FRAME_ENVELOPE_BYTES`, and the unused `MAX_JSON_ESCAPE_EXPANSION` constant is gone. The old bound was not unsafe — it under-admitted — but it silently forbade legitimate large caps. + +### Same-group survivors are reaped before the fiber goes quiescent + +A model program can leave a descendant in the child's OWN process group (no `setsid`, so `kill(-pid)` reaches it) that ignores SIGTERM but releases the inherited stdout/stderr/fd-3 pipes. The leader then exits, its `close` fires because the pipes drained, and settlement runs while that descendant is still alive. `kill()` arms an `unref`'d SIGKILL timer after SIGTERM; the fix is that `settle()` no longer resolves the run's `finished` promise immediately when an escalation is in flight. Instead, when `killing` is set and the process group is not yet empty (`process.kill(-pid, 0)` does not throw ESRCH), it polls the group on a REF'd timer, bounded by `graceMs + CLOSE_REAP_MARGIN_MS`, and resolves `finished` only once the group has emptied. The ref'd poll is the load-bearing part: it keeps the host event loop alive until the SIGKILL has actually reaped the group, so even a short-lived host — a one-shot headless run, a config subprocess — cannot exit and reparent the survivor to init. In the normal case (the leader was the only member) the first probe returns ESRCH and settlement resolves with zero added latency. `teardown()` awaits each run's `finished`, so disposal is genuinely quiescent, matching its JSDoc. + +### Binding replies complete on the calling loop's thread + +Also in `py/bootstrap.py`, a binding reply Future is created on the loop that ran `dispatch`. When the model calls a binding from a worker THREAD via `asyncio.run(tools.x(...))`, that Future belongs to the thread's loop, not the main loop where `_pump_replies` reads the reply. `asyncio.Future` is not thread-safe: completing it from another thread does not wake its own loop, so the direct `set_result`/`set_exception` left the awaiting thread stranded and the run degraded to a wall-clock timeout. Each pending entry now records its Future's loop alongside the Future, and `_pump_replies` completes it via that loop's `call_soon_threadsafe`. The shared `pending`/`next_id` state is guarded by a `threading.Lock` held across the id claim, the fd-3 write, and the counter advance, so concurrent callers cannot interleave frames out of the id order the host requires. + ## Testing - `tests/boot-write-failure.spec.ts` mocks `spawn` so the fd-3 pipe throws on the boot write — the one path a real subprocess cannot be coerced into — and asserts `run()` resolves a `worker-exit` rather than rejecting. Isolated in its own spec so the real-subprocess suite is untouched. -- `tests/residual-detach.spec.ts` unit-tests `detachResidual`: the carried copy equals the residual, owns a backing store sized to its own length, and does not share the source frame's `ArrayBuffer`. -- `tests/runtime.spec.ts` adds a real-subprocess case where four daemon threads emit unterminated writes up to the moment the body returns and settlement flushes, repeated so the interleave lands; the run must complete cleanly. A pure data race has no single bad input to reject, so this maximizes overlap rather than asserting a deterministic rejection. +- `tests/residual-detach.spec.ts` unit-tests `detachResidual`: the carried copy equals the residual, owns a backing store sized to its own length (fixture kept above Node's Buffer pool threshold), and does not share the source frame's `ArrayBuffer`. +- `tests/runtime.spec.ts` — the output-cap case asserts the `ceiling - envelope` bound (268435392) and its message. A daemon-thread case drives four threads emitting unterminated writes through settlement's flush. The same-group reap case spawns a SIGTERM-ignoring same-group descendant that releases the pipes and bumps a heartbeat file; the test asserts the heartbeat STOPS after the grace-window SIGKILL — an assertion robust whether the killed descendant is reaped or lingers as a zombie, so it holds where PID 1 does not wait() orphans. The cross-loop case runs a binding from a worker thread's own `asyncio.run` loop while the main coroutine yields with `await asyncio.sleep`, asserting the reply round-trips instead of timing out. ## Alternatives considered @@ -40,6 +52,12 @@ Also in `src/index.ts`, after the newline loop over a `Buffer.concat` of the pen **Assert the residual memory effect through the seam.** Rejected: the retained allocation is not observable through `CodeRunResult`, so a black-box test could not distinguish fixed from unfixed. Extracting `detachResidual` makes the backing-store invariant a deterministic unit test instead. +**Reap the same-group survivor with a fire-and-forget `unref`'d SIGKILL timer alone.** Rejected: an `unref`'d timer does not keep the host alive, so a host that exits within the grace window (a one-shot run, a config subprocess) never fires the SIGKILL and the survivor is reparented to init — the same "no subprocess outlives the fiber" violation in a different shape, and `teardown`'s "await each child's exit" JSDoc would be false. Awaiting the group's death on a ref'd poll keeps the host alive exactly long enough to reap, at zero cost in the common empty-group case. + +**Assert the reap with `process.kill(pid, 0)` throwing ESRCH.** Rejected: a SIGKILL'd process lingers as a zombie until its parent `wait()`s it, and in a container whose PID 1 does not reap orphans the signal-0 probe keeps succeeding, so the assertion would false-fail cross-environment. A heartbeat file that stops advancing detects "no longer executing," which a reaped process and a zombie both satisfy. + +**Complete the cross-loop Future with a plain `set_result` and rely on the GIL.** Rejected: the GIL serializes bytecode but does not make `asyncio.Future` cross-loop-safe — completing a Future from a thread other than its loop's does not schedule its callbacks or wake the loop. `call_soon_threadsafe` on the owning loop is the documented mechanism. + ## Consequences -The seam's resolve-don't-reject contract now holds on the boot-write path, and its coverage is measured rather than ignored. Log capture is thread-safe at the cost of one re-entrant lock acquisition per write and flush — negligible against the os.write already on that path. Fd-3 residual memory is bounded by the actual retained bytes, and `pendingBytes` measures what it claims. Each fix carries a test that fails without it, so a future regression on any of the three goes red. +The seam's resolve-don't-reject contract holds on the boot-write path with measured coverage. Log capture is thread-safe at the cost of one re-entrant lock acquisition per write and flush. Fd-3 residual memory is bounded by the actual retained bytes. The output caps admit every value a frame can carry. Disposal is genuinely quiescent against a same-group survivor — bounded by the existing grace budget, zero-cost when the group is already empty — and bindings called from model-created threads complete instead of timing out. Each fix carries a test that fails without it, so a future regression on any of the six goes red. diff --git a/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.zh.md b/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.zh.md index 668f17cf2b..594d5074a5 100644 --- a/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.zh.md +++ b/.agents/notes/implemented/bug-fix/2026-07-31-code-runtime-python-settlement-fixes.zh.md @@ -1,4 +1,4 @@ -# Agent Note: CPython 后端的三处结算与分帧修复 +# Agent Note: CPython 后端中的结算、分帧与生命周期修复 Status: implemented @@ -6,11 +6,11 @@ Status: implemented ## Problem -用于 Code Mode 的 CPython 子进程后端建立在 [fd-3 帧协议](../architecture/2026-07-31-code-runtime-python-fd3-protocol.md)之上，把每个程序结果都 resolve 成一个 `CodeRunResult`，仅在 seam 被误用时才 reject `run()`。三个缺陷以单元测试覆盖率无法暴露的方式破坏了这一契约，因为它们各自藏在一处 `/* v8 ignore */` 之后、藏在一条读起来像修复但实际并非修复的"捕获可调用对象"注释之后，或藏在一处透过 seam 不可见的内存效应之后。这些缺陷是通过审查当时的后端代码发现的，而非由某个失败的测试发现，因此每处修复都附带一个在缺少该修复时会失败的测试。 +用于 Code Mode 的 CPython 子进程后端建立在 [fd-3 帧协议](../architecture/2026-07-31-code-runtime-python-fd3-protocol.md)之上，把每个程序结果都 resolve 成一个 `CodeRunResult`，仅在 seam 被误用时才 reject `run()`，并且会 dispose 到完全停稳，从而没有任何子进程存活得比 fiber 更久。一连串审查暴露出一些缺陷，它们以单元测试覆盖率无法捕获的方式破坏了这些契约：每一个都藏在一处 `/* v8 ignore */` 之后、一个读起来像修复但实际并非修复的捕获可调用对象之后、一处透过 seam 不可见的内存效应之后、一处重复计数的加载期上界之后、一处存活者能够熬过的进程组升级之后，或者一处静默死锁的跨事件循环完成之后。每处修复都附带一个在缺少它时会失败的测试。 ## Decision -三处相互独立的修正，各自位于拥有对应缺陷的包中。 +六处相互独立的修正，各自位于拥有对应缺陷的包中。 ### Boot-write failure no longer rejects run() @@ -24,11 +24,23 @@ Status: implemented 同样在 `src/index.ts` 中，在对待处理 fd-3 分片的 `Buffer.concat` 结果按换行符做循环之后，剩余的不完整行被以它被切出的 `subarray` 视图形式向前传递。视图会使整个 concat 的底层分配保持存活，因此一个大帧后面跟着一个极小的尾部片段，会钉住整整一帧大小的内存，而 `pendingBytes`（被设为该片段的长度）报告的值远小于实际保留的内存。现在，残余数据通过导出的 `detachResidual` 辅助函数被分离到一个大小恰当的新 `Buffer` 中，从而让 concat 分配得以被回收，并使 `pendingBytes` 成为一个诚实的度量值。 +### Output-cap load bound is ceiling minus envelope, not divided by six + +那处在加载期拒绝比单个 fd-3 帧所能承载更大的 `maxLogBytes`／`maxValueBytes` 的检查，会把帧上限除以六以应对最坏情况下的转义膨胀。但这两项预算都是以已转义的序列化字节来计量的：宿主日志账本按 `Buffer.byteLength(JSON.stringify(text))` 计费，而 `checkDoneValue` 度量的是转义后的形式，因此一个在上限之内被放行的载荷在传输时最多占用 `cap + envelope`；转义已经包含在计费之内，不能再被乘一次。现在该上界为 `FRAME_CEILING_BYTES - FRAME_ENVELOPE_BYTES`，未使用的 `MAX_JSON_ESCAPE_EXPANSION` 常量已被删除。旧的上界并非不安全（它是放行不足），但它静默地禁止了合法的大上限。 + +### Same-group survivors are reaped before the fiber goes quiescent + +模型程序可能在子进程自己的进程组里（没有 `setsid`，因此 `kill(-pid)` 能到达它）留下一个后代，它忽略 SIGTERM，但释放了继承而来的 stdout／stderr／fd-3 管道。随后 leader 退出，由于管道已被抽空，它的 `close` 触发，于是结算在那个后代仍存活时运行。`kill()` 在 SIGTERM 之后装设一个 `unref` 的 SIGKILL 定时器；本次修复是，当有一次升级正在进行时，`settle()` 不再立即 resolve 该次运行的 `finished` promise。取而代之的是，当 `killing` 被置位且进程组尚未为空时（`process.kill(-pid, 0)` 不抛出 ESRCH），它在一个 ref 的定时器上轮询该进程组，以 `graceMs + CLOSE_REAP_MARGIN_MS` 为界，仅当进程组已清空后才 resolve `finished`。这个 ref 的轮询是承重部分：它让宿主事件循环保持存活，直到 SIGKILL 真正回收了该进程组，因此即使是一个短命的宿主（一次性的 headless 运行、一个配置子进程）也无法退出并把存活者 reparent 给 init。在正常情况下（leader 是唯一成员），第一次探测返回 ESRCH，结算以零附加延迟完成 resolve。`teardown()` 会 await 每次运行的 `finished`，因此 dispose 是真正完全停稳的，与其 JSDoc 相符。 + +### Binding replies complete on the calling loop's thread + +同样在 `py/bootstrap.py` 中，一个绑定回复 Future 是在运行 `dispatch` 的那个事件循环上创建的。当模型通过 `asyncio.run(tools.x(...))` 从一个工作线程调用某个绑定时，该 Future 属于该线程的事件循环，而不是 `_pump_replies` 读取回复的主事件循环。`asyncio.Future` 不是线程安全的：从另一个线程完成它并不会唤醒它自己的事件循环，因此直接的 `set_result`／`set_exception` 会让那个正在等待的线程被搁置，该次运行退化为墙钟超时。现在每个待处理条目都会在记录 Future 的同时记录其 Future 所属的事件循环，`_pump_replies` 通过该事件循环的 `call_soon_threadsafe` 来完成它。共享的 `pending`／`next_id` 状态由一把 `threading.Lock` 保护，该锁跨越 id 认领、fd-3 写入和计数器推进这三步持有，因此并发调用方无法以违反宿主所要求的 id 顺序来交错帧。 + ## Testing - `tests/boot-write-failure.spec.ts` 对 `spawn` 做 mock，使 fd-3 管道在引导写入时抛出异常（这是真实子进程无法被迫进入的唯一路径），并断言 `run()` resolve 出一个 `worker-exit` 而非 reject。它被隔离在自己的 spec 中，因此真实子进程测试套件不受影响。 -- `tests/residual-detach.spec.ts` 对 `detachResidual` 做单元测试：向前传递的副本与残余数据相等、拥有一个大小与其自身长度一致的底层存储，并且不与源帧的 `ArrayBuffer` 共享。 -- `tests/runtime.spec.ts` 新增一个真实子进程用例：四个 daemon 线程持续发出未结束的写入，直到函数体返回、结算执行 flush 的那一刻，并反复运行以让交错真正出现；该次运行必须干净地完成。纯数据竞态没有单一的坏输入可供 reject，因此该测试最大化重叠而非断言一个确定性的 reject。 +- `tests/residual-detach.spec.ts` 对 `detachResidual` 做单元测试：向前传递的副本与残余数据相等、拥有一个大小与其自身长度一致的底层存储（fixture 保持在 Node 的 Buffer 池阈值之上），并且不与源帧的 `ArrayBuffer` 共享。 +- `tests/runtime.spec.ts`：output-cap 用例断言 `ceiling - envelope` 上界（268435392）及其消息。一个 daemon 线程用例驱动四个线程穿过结算的 flush 发出未结束的写入。same-group 回收用例 spawn 一个忽略 SIGTERM 的同进程组后代，它释放管道并递增一个心跳文件；该测试断言在宽限窗口的 SIGKILL 之后心跳停止：无论被杀死的后代是被回收还是作为僵尸进程滞留，这个断言都成立，因此它在 PID 1 不 wait() 孤儿进程的环境下同样成立。cross-loop 用例在主协程通过 `await asyncio.sleep` 让出时，从一个工作线程自己的 `asyncio.run` 事件循环运行一个绑定，断言该回复完成往返而不是超时。 ## Alternatives considered @@ -40,6 +52,12 @@ Status: implemented **通过 seam 断言残余数据的内存效应。** 已否决：被保留的分配透过 `CodeRunResult` 不可观测，因此黑盒测试无法区分已修复与未修复。转而抽取出 `detachResidual`，把底层存储的不变量变成一个确定性的单元测试。 +**仅用一个发后不理的 `unref` SIGKILL 定时器来回收同进程组存活者。** 已否决：`unref` 的定时器不会让宿主保持存活，因此一个在宽限窗口内退出的宿主（一次性运行、一个配置子进程）永远不会触发 SIGKILL，存活者被 reparent 给 init，这是同一个"没有子进程存活得比 fiber 更久"的违规换了个形态，而且 `teardown` 的"await 每个子进程退出"的 JSDoc 会变为不实。在一个 ref 的轮询上 await 进程组的消亡，让宿主恰好保持存活足够长以完成回收，在常见的空进程组情形下代价为零。 + +**用 `process.kill(pid, 0)` 抛出 ESRCH 来断言回收。** 已否决：一个被 SIGKILL 的进程会作为僵尸进程滞留，直到它的父进程 `wait()` 它，而在一个 PID 1 不回收孤儿进程的容器里，signal-0 探测会持续成功，因此该断言会在跨环境时误报失败。一个停止推进的心跳文件检测的是"不再执行"，而被回收的进程和僵尸进程都满足这一点。 + +**用一个普通的 `set_result` 完成跨事件循环的 Future 并依赖 GIL。** 已否决：GIL 序列化字节码，但并不使 `asyncio.Future` 跨事件循环安全：从一个并非其事件循环所属的线程完成一个 Future，不会调度它的回调，也不会唤醒该事件循环。在拥有该 Future 的事件循环上调用 `call_soon_threadsafe` 才是有文档记载的机制。 + ## Consequences -现在 seam 的"只 resolve、不 reject"契约在引导写入路径上得以成立，且其覆盖率是被度量而非被忽略的。日志捕获现在是线程安全的，代价是每次写入和 flush 都要获取一次可重入锁，这相对于该路径上已有的 os.write 可以忽略不计。fd-3 残余数据的内存现在受实际保留的字节数约束，且 `pendingBytes` 度量的正是它所声称的值。每处修复都附带一个在缺少它时会失败的测试，因此这三处中任何一处未来若发生回归都会变红。 +seam 的"只 resolve、不 reject"契约在引导写入路径上得以成立，且覆盖率是被度量的。日志捕获是线程安全的，代价是每次写入和 flush 都要获取一次可重入锁。fd-3 残余数据的内存受实际保留的字节数约束。输出上限放行一个帧所能承载的每一个值。dispose 面对同进程组存活者是真正完全停稳的（以既有的宽限预算为界，在进程组已为空时代价为零），并且从模型创建的线程调用的绑定会完成而不是超时。每处修复都附带一个在缺少它时会失败的测试，因此这六处中任何一处未来若发生回归都会变红。 diff --git a/packages/code-runtime/code-runtime-python/py/bootstrap.py b/packages/code-runtime/code-runtime-python/py/bootstrap.py index 707a50858e..e3866da968 100644 --- a/packages/code-runtime/code-runtime-python/py/bootstrap.py +++ b/packages/code-runtime/code-runtime-python/py/bootstrap.py @@ -616,8 +616,25 @@ async def _run(channel: ProtocolChannel) -> None: ) # 2. Wire the tools proxies and the ack. - pending: dict[int, asyncio.Future[Any]] = {} + # + # Each entry records the reply Future AND the loop it was created on. Model + # code may call a binding from a THREAD it started, spelled + # ``asyncio.run(tools.x(...))`` or its own new loop in that thread, so a + # Future here can belong to a loop other than the one ``_pump_replies`` runs + # on. ``asyncio.Future`` is not thread-safe: completing it from another + # thread does not wake its own loop, so the pump schedules the completion on + # the owning loop via ``call_soon_threadsafe`` (see ``_pump_replies``) rather + # than calling ``set_result`` directly. + pending: dict[int, tuple[asyncio.AbstractEventLoop, asyncio.Future[Any]]] = {} next_id = 0 + # Serializes the id claim + write + counter advance in ``dispatch`` against + # both other binding-calling threads and the pump's ``pop``. ``dispatch`` may + # run concurrently on several loops/threads, and the host answers a ``call`` + # only when its id is the exact successor of the last one — so ids must reach + # the wire in the order they are claimed. Holding this lock across the write + # (not just the counter arithmetic) is what keeps two threads' frames from + # interleaving on fd 3 out of id order, which the host would reject. + pending_lock = threading.Lock() error_classes: dict[str, type] = {} @@ -646,25 +663,35 @@ async def _run(channel: ProtocolChannel) -> None: # state it retains to a single number. A frame that never reaches the # host must therefore not consume an id, so the counter advances only # once the write has succeeded. - call_id = next_id - fut: asyncio.Future[Any] = asyncio.get_event_loop().create_future() - pending[call_id] = fut - try: - channel.send_sync( - { - "type": "call", - "id": call_id, - "global": global_name, - "name": name, - "args": args, - } - ) - except (TypeError, ValueError) as exc: - pending.pop(call_id, None) - raise call_failure( - f"binding arguments must be lossless JSON: {exc}" - ) from exc - next_id += 1 + # + # The whole claim-write-advance runs under ``pending_lock`` because a + # binding may be called from more than one thread/loop at once (the model + # can start a thread that runs ``asyncio.run(tools.x(...))``). Without + # the lock two callers could claim the same id, or write their frames to + # fd 3 in an order that does not match their ids — either of which the + # host rejects as an out-of-sequence call. The Future's own loop is + # captured here so ``_pump_replies`` can complete it thread-safely. + loop = asyncio.get_event_loop() + with pending_lock: + call_id = next_id + fut: asyncio.Future[Any] = loop.create_future() + pending[call_id] = (loop, fut) + try: + channel.send_sync( + { + "type": "call", + "id": call_id, + "global": global_name, + "name": name, + "args": args, + } + ) + except (TypeError, ValueError) as exc: + pending.pop(call_id, None) + raise call_failure( + f"binding arguments must be lossless JSON: {exc}" + ) from exc + next_id += 1 try: return await fut except _BindingRejection as exc: @@ -689,7 +716,9 @@ async def _run(channel: ProtocolChannel) -> None: # 3. Start a reply-pump task before the run message: replies can arrive # interleaved with the run's own binding traffic. - reply_task = asyncio.get_event_loop().create_task(_pump_replies(channel, pending)) + reply_task = asyncio.get_event_loop().create_task( + _pump_replies(channel, pending, pending_lock) + ) # 4. Read the run message. run = channel.read_frame() @@ -793,28 +822,51 @@ async def _run(channel: ProtocolChannel) -> None: async def _pump_replies( - channel: ProtocolChannel, pending: dict[int, asyncio.Future[Any]] + channel: ProtocolChannel, + pending: dict[int, tuple[asyncio.AbstractEventLoop, asyncio.Future[Any]]], + pending_lock: "threading.Lock", ) -> None: """Background task: read reply frames and settle pending futures. Cancelled after ``done`` is posted. Unknown ids and post-settlement replies are ignored (mirrors the worker backend's hostile-peer stance, though here the host is the trusted side; the guards defend against races). + + A pending Future may belong to a loop other than this pump's — the model can + call a binding from a thread running its own loop (``asyncio.run(tools.x())``). + ``asyncio.Future`` is not thread-safe, so the completion is scheduled on the + Future's OWN loop via ``call_soon_threadsafe`` rather than mutated here; a + direct ``set_result`` would never wake the waiting loop and the call would + hang to the wall clock. The ``pop`` shares ``pending_lock`` with ``dispatch`` + so a reply cannot race the claim that registers its id. """ + def complete(fut: asyncio.Future[Any], ok: bool, value: Any, message: Any) -> None: + # Runs on the Future's own loop. `done()` re-checked here because + # cancellation or a duplicate reply may have settled it between the pop + # and this callback. + if fut.done(): + return + if ok: + fut.set_result(value) + else: + fut.set_exception(_BindingRejection(str(message))) + while True: frame = await channel.read_frame_async() if frame is None: return if frame.get("type") != "reply": continue - fut = pending.pop(frame.get("id"), None) - if fut is None or fut.done(): + with pending_lock: + entry = pending.pop(frame.get("id"), None) + if entry is None: continue - if frame.get("ok"): - fut.set_result(frame.get("value")) - else: - fut.set_exception(_BindingRejection(str(frame.get("message")))) + loop, fut = entry + ok = bool(frame.get("ok")) + value = frame.get("value") + message = frame.get("message") + loop.call_soon_threadsafe(complete, fut, ok, value, message) _SCALAR_RE = re.compile( diff --git a/packages/code-runtime/code-runtime-python/src/index.ts b/packages/code-runtime/code-runtime-python/src/index.ts index 38935ab45f..a811185567 100644 --- a/packages/code-runtime/code-runtime-python/src/index.ts +++ b/packages/code-runtime/code-runtime-python/src/index.ts @@ -217,6 +217,17 @@ const FRAME_ENVELOPE_BYTES = 64 */ const CLOSE_REAP_MARGIN_MS = 2_000 +/** + * Interval between process-group liveness probes while settlement waits for an + * escalated SIGKILL to empty the group (see the `killing` branch in + * {@link PythonCodeRuntime.execute}'s settle). A poll rather than an event + * because the group members are the model's own descendants, which the host does + * not `wait()` for and gets no exit signal from; the probe is a signal-0 + * `process.kill(-pid, 0)`, so the interval only bounds how promptly a now-empty + * group is noticed, capped by `graceMs + CLOSE_REAP_MARGIN_MS`. + */ +const GROUP_REAP_POLL_MS = 50 + /** * Extract a human message from an unknown thrown value. * @@ -961,6 +972,7 @@ export class PythonCodeRuntime extends CodeRuntime { // Escalate SIGTERM → grace → SIGKILL on the entire process group. Idempotent // via `killing`. let killing = false + let graceTimer: NodeJS.Timeout | undefined // A backstop for the one case `close` cannot cover: model code that starts // a descendant with `os.setsid()`/`start_new_session=True` moves it into a // fresh process group, so the SIGTERM/SIGKILL aimed at the child's group @@ -983,22 +995,29 @@ export class PythonCodeRuntime extends CodeRuntime { if (killing) return killing = true killGroup('SIGTERM') - // The SIGKILL is left to fire on its own timer and is deliberately NOT - // cancelled at settlement. A model program can leave a descendant in the - // SAME process group `kill(-pid)` targets — no setsid, so it stays in the - // group — that ignores SIGTERM but releases the inherited stdout/stderr/ - // fd-3 pipes: the leader then exits, its `close` fires (the pipes drained), - // and settle() runs while that descendant is still alive. Cancelling the - // timer there would strand it, breaking "no subprocess outlives the fiber". - // Letting the timer elapse SIGKILLs the whole group, reaching the survivor; - // `killGroup` swallows ESRCH, so firing against an already-dead group (the - // normal case, where the leader was the only member) is harmless. `unref` - // so a pending SIGKILL never keeps the host process alive after run() - // resolves. (A setsid-escaped orphan in a FRESH group is the different case - // `closeDeadline` in finish() covers, since `close` never fires there.) - const graceTimer = setTimeout(() => { killGroup('SIGKILL') }, this.config.graceMs) + // Escalate to SIGKILL after the grace window. The timer is `unref`'d so a + // pending SIGKILL never keeps the host process alive on its own; the + // guarantee that a same-group survivor is actually reaped before the fiber + // goes quiescent is enforced by settle() awaiting the group's death (see + // there), NOT by this timer firing during host lifetime. A setsid-escaped + // orphan in a FRESH group is the different case `closeDeadline` in finish() + // covers, since `close` never fires there. + graceTimer = setTimeout(() => { killGroup('SIGKILL') }, this.config.graceMs) graceTimer.unref() } + // True once the group has no members left: a signal-0 probe to the whole + // group (`kill(-pid, 0)`) throws ESRCH when empty (EPERM would still mean a + // member exists). Only meaningful once a spawn produced a pid. + const groupEmpty = (): boolean => { + /* v8 ignore next -- pid is always defined once escalation runs; the guard narrows the type. */ + if (child.pid === undefined) return true + try { + process.kill(-child.pid, 0) + return false + } catch (error: unknown) { + return (error as NodeJS.ErrnoException).code === 'ESRCH' + } + } let finishResolve!: () => void const finished = new Promise((done) => { finishResolve = done }) @@ -1017,7 +1036,8 @@ export class PythonCodeRuntime extends CodeRuntime { // same-group descendant that ignored SIGTERM but released the pipes lets // `close` fire (and settle() run) while it is still alive, so the pending // SIGKILL must remain armed to reap it (see kill()). The timer is - // `unref`'d, so leaving it pending cannot keep the host process alive. + // `unref`'d; quiescence does not depend on it firing during host lifetime + // — `finished` (below) is withheld until the group is confirmed empty. if (closeDeadline !== undefined) clearTimeout(closeDeadline) // Drop from `live` only at settlement (close / pid-less spawn failure), // NOT at finish(): between finish() and the child's `close` the child @@ -1042,8 +1062,32 @@ export class PythonCodeRuntime extends CodeRuntime { // tracked; the directory holds no secret, only a copy of two // checked-in scripts. } - finishResolve() resolve({ ...result, logs }) + // `finished` is what teardown awaits to honor "no subprocess outlives the + // fiber". When no escalation ran (normal completion, no kill) or the group + // is already empty, resolve it now. Otherwise a same-group descendant that + // ignored SIGTERM but released the pipes is still alive here (its `close` + // is what got us to settle); withhold `finished` until the grace-window + // SIGKILL has emptied the group. The poll timers are REF'd on purpose: a + // short-lived host (a one-shot headless run, a config subprocess) would + // otherwise exit before the unref'd SIGKILL timer fired, reparenting the + // survivor to init — the leak this await exists to prevent. The wait is + // bounded by the same graceMs + margin the SIGKILL escalation uses, so a + // truly unreapable process (it cannot be, since it is in the group + // `kill(-pid)` reaches) could not hang disposal. + if (!killing || groupEmpty()) { + finishResolve() + return + } + const deadline = Date.now() + this.config.graceMs + CLOSE_REAP_MARGIN_MS + const pollGroup = (): void => { + if (groupEmpty() || Date.now() >= deadline) { + finishResolve() + return + } + setTimeout(pollGroup, GROUP_REAP_POLL_MS) + } + pollGroup() } const finish = (result: Omit): void => { diff --git a/packages/code-runtime/code-runtime-python/tests/runtime.spec.ts b/packages/code-runtime/code-runtime-python/tests/runtime.spec.ts index 2f78dcddc8..30d4134aa3 100644 --- a/packages/code-runtime/code-runtime-python/tests/runtime.spec.ts +++ b/packages/code-runtime/code-runtime-python/tests/runtime.spec.ts @@ -1,4 +1,4 @@ -import { existsSync, readdirSync, realpathSync } from 'node:fs' +import { existsSync, readdirSync, realpathSync, statSync } from 'node:fs' import { mkdtemp, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { basename, dirname, join } from 'node:path' @@ -1932,72 +1932,73 @@ describe('PythonCodeRuntime — budgets, termination, disposal', () => { // it does not hold — here by giving the Popen child DEVNULL streams and // letting close_fds drop fd 3. The leader then writes `done` and exits, its // `close` fires because the pipes drained, and settle() runs while that - // descendant is still alive. If settle() cancelled the grace-window SIGKILL - // the descendant would outlive the fiber; leaving the unref'd timer to fire - // SIGKILLs the whole group and reaps it. + // descendant is still alive. settle() then keeps a REF'd poll alive until the + // grace-window SIGKILL has emptied the whole process group, so the host cannot + // exit and reparent the survivor to init: no subprocess outlives the fiber. // // The descendant must have SIG_IGN installed BEFORE the host sends SIGTERM, // or it dies from the default SIGTERM whether the fix is present or not — so // it writes a readiness marker after trapping and the leader waits for that - // marker before returning. The descendant sleeps 30 s as a safety net so a - // broken fix cannot leak it forever; the assertion window is far shorter, so - // it genuinely tests the SIGKILL reaping rather than the self-timeout. + // marker before returning. While alive it bumps a heartbeat file every 50 ms; + // the test asserts the heartbeat STOPS, which is what "no longer executing" + // 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 readyMarker = join(handoff, 'ready') + const heartbeat = join(handoff, 'heartbeat') const { runtime } = await setup({ maxWallMs: 10_000, graceMs: 300 }) - let reportedPid!: (pid: number) => void - const childPid = new Promise((resolve) => { reportedPid = resolve }) const result = await runtime.run({ program: [ 'import subprocess, sys, os, time', `marker = ${JSON.stringify(readyMarker)}`, + `heartbeat = ${JSON.stringify(heartbeat)}`, // Same group (no start_new_session); ignores SIGTERM; holds none of the // leader's pipes (DEVNULL std streams, close_fds drops fd 3). It writes - // the marker (its argv[1]) only AFTER the trap is installed, so the - // leader cannot return — and the host cannot send SIGTERM — before the - // descendant ignores it. - 'code = "import signal, sys, time; signal.signal(signal.SIGTERM, signal.SIG_IGN); open(sys.argv[1], \'w\').close(); time.sleep(30)"', - 'child = subprocess.Popen([sys.executable, "-c", code, marker],', + // the marker (argv[1]) only AFTER the trap is installed — so the leader + // cannot return, and the host cannot send SIGTERM, before it is ignored — + // then rewrites the heartbeat (argv[2]) every 50 ms for up to 30 s. + 'code = ("import signal, sys, time\\n"', + ' "signal.signal(signal.SIGTERM, signal.SIG_IGN)\\n"', + ' "open(sys.argv[1], \'w\').close()\\n"', + ' "end = time.time() + 30\\n"', + ' "while time.time() < end:\\n"', + ' " open(sys.argv[2], \'w\').close()\\n"', + ' " time.sleep(0.05)\\n")', + 'child = subprocess.Popen([sys.executable, "-c", code, marker, heartbeat],', ' stdin=subprocess.DEVNULL,', ' stdout=subprocess.DEVNULL,', ' stderr=subprocess.DEVNULL)', 'deadline = time.time() + 5', 'while not os.path.exists(marker) and time.time() < deadline:', ' time.sleep(0.02)', - 'await tools.report({"pid": child.pid})', 'return "spawned"', ].join('\n'), - bindings: tools({ - report: async (args) => { - reportedPid((args as { pid: number }).pid) - return 'ok' - }, - }), + bindings: [], }) expect(result.error).toBeUndefined() expect(result.value).toBe('spawned') - const pid = await childPid - expect(Number.isInteger(pid) && pid > 0).toBe(true) // The trap really installed before the leader returned, so this is the // SIGTERM-ignoring descendant, not one that would have died to the default. expect(existsSync(readyMarker)).toBe(true) - // run() resolved inside the grace window, so the descendant is still alive - // here; the pending SIGKILL reaps it shortly after graceMs. Poll until it is - // gone, well within the descendant's own 30 s self-timeout. - const deadline = Date.now() + 5_000 - const alive = (): boolean => { - try { - process.kill(pid, 0) - return true - } catch { - return false - } + // The grace-window SIGKILL (graceMs 300 + reap margin) empties the group. Once + // it has, the descendant stops bumping the heartbeat. Poll the heartbeat's + // mtime: two consecutive reads far enough apart with no change means it is no + // longer executing — true whether it was reaped or lingers as a zombie, so + // the assertion holds in a container whose init does not wait() orphans. The + // window (well under the 30 s self-timeout) proves the SIGKILL did the work. + const mtime = (): number => { try { return statSync(heartbeat).mtimeMs } catch { return 0 } } + const stopDeadline = Date.now() + 8_000 + let last = mtime() + let still = false + while (Date.now() < stopDeadline) { + await new Promise(resolve => setTimeout(resolve, 400)) + const now = mtime() + if (now === last && now !== 0) { still = true; break } + last = now } - while (alive() && Date.now() < deadline) { - await new Promise(resolve => setTimeout(resolve, 50)) - } - expect(() => process.kill(pid, 0)).toThrow(/ESRCH/) - }, 15_000) + expect(still).toBe(true) + }, 20_000) }) describe('PythonCodeRuntime — hostile peer', () => { @@ -2305,6 +2306,47 @@ describe('PythonCodeRuntime — hostile peer', () => { } }, 30_000) + it('completes a binding called from a worker thread on its own event loop', async () => { + // A binding reply Future is created on the loop that ran `dispatch`. When the + // model calls a binding from a worker THREAD via `asyncio.run(tools.x(...))`, + // that Future belongs to the thread's loop, not the main loop where + // `_pump_replies` reads the reply. `asyncio.Future` is not thread-safe: + // completing it from another thread does not wake its own loop, so a direct + // `set_result` would strand the awaiting thread and the run would degrade to a + // wall-clock timeout. The pump must schedule completion on the Future's own + // loop via `call_soon_threadsafe`. The tight maxWallMs makes the pre-fix + // failure a fast timeout rather than a hang. + // + // The main coroutine yields with `await asyncio.sleep` while the worker runs, + // rather than a synchronous `t.join()`: joining would block the main thread, + // so the main loop could not run `_pump_replies` and the call would deadlock + // regardless of the fix — that blocks the pump, not the cross-loop delivery + // this test pins. + const { runtime } = await setup({ maxWallMs: 8_000 }) + const seen: unknown[] = [] + const result = await runtime.run({ + program: [ + 'import asyncio, threading', + 'result = {}', + 'def worker():', + // A fresh loop in this thread; the binding Future is created here. + ' result["value"] = asyncio.run(tools.echo({"from": "thread"}))', + 't = threading.Thread(target=worker)', + 't.start()', + 'while t.is_alive():', + ' await asyncio.sleep(0.02)', + 'return result["value"]', + ].join('\n'), + bindings: tools({ + echo: async (args) => { seen.push(args); return args as CodeJsonValue }, + }), + }) + expect(result.error).toBeUndefined() + expect(result.value).toEqual({ from: 'thread' }) + // The host binding actually ran (the reply round-tripped), not a timeout. + expect(seen).toEqual([{ from: 'thread' }]) + }, 15_000) + it('round-trips an exactly representable large integer through a binding echo', async () => { // The reply serializer must print BigInt digits for a beyond-safe // integral double: String(2**60) emits a rounded form, and the child