fix(code-runtime-python): bind the reply-pump exception names as def-time default arguments

A body-local X = X binding in _pump_replies is too late: _run reaches the
model's top-level statements (which run first, since there is no suspension
point between create_task and await __dsh_main__) before the pump's first step,
so a __main__.RuntimeError rebind there would be captured by the body local and
a closed-loop failure would escape the except, killing the pump. Bind
_RuntimeError, _BindingRejection, str, and bool as DEF-TIME default arguments of
_pump_replies (evaluated at import, before any model code runs). Add a regression
test that rebinds __main__.RuntimeError as the first program statement and drives
the closed-loop worker pattern, asserting the pump survives and delivers the
later binding. Update the settlement note (en + zh) to describe the default-arg
capture; pairing re-recorded and consistent.

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

@@ -1115,6 +1115,19 @@ async def _pump_replies(
     channel: ProtocolChannel,
     pending: dict[int, tuple[asyncio.AbstractEventLoop, asyncio.Future[Any]]],
     pending_lock: "threading.Lock",
+    # Bound as DEFAULT ARGUMENTS so they are captured at def/import time, before
+    # ANY model code runs. This bootstrap IS `__main__`, so `__main__.RuntimeError
+    # = ...` (or `__main__._BindingRejection`, `__main__.str`, `__main__.bool`)
+    # as a program top-level statement would otherwise rebind the module globals
+    # these clauses resolve at runtime. A body-local `X = X` binding is too late:
+    # `_run` reaches `await __dsh_main__` (whose top-level statements run first)
+    # with no suspension point after `create_task`, so the model's rebind executes
+    # before the pump body. Defaults are evaluated in the enclosing scope at def
+    # time, truly before the program.
+    _RuntimeError: Any = RuntimeError,
+    _BindingRejection: Any = _BindingRejection,
+    _str: Any = str,
+    _bool: Any = bool,
 ) -> None:
     """Background task: read reply frames and settle pending futures.
 
@@ -1131,14 +1144,6 @@ async def _pump_replies(
     so a reply cannot race the claim that registers its id.
     """
 
-    # The exception class the closed-loop catch below resolves is bound into a
-    # LOCAL here, after the docstring, before any model code runs. This bootstrap
-    # IS `__main__`, so `__main__.RuntimeError = ...` would otherwise rebind the
-    # module global the `except RuntimeError` clause resolves at runtime, and a
-    # closed-loop scheduling failure would then escape the catch, killing the
-    # pump and stranding every later reply.
-    _RuntimeError = RuntimeError
-
     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
@@ -1148,7 +1153,7 @@ async def _pump_replies(
         if ok:
             fut.set_result(value)
         else:
-            fut.set_exception(_BindingRejection(str(message)))
+            fut.set_exception(_BindingRejection(_str(message)))
 
     while True:
         frame = await channel.read_frame_async()
@@ -1161,7 +1166,7 @@ async def _pump_replies(
         if entry is None:
             continue
         loop, fut = entry
-        ok = bool(frame.get("ok"))
+        ok = _bool(frame.get("ok"))
         value = frame.get("value")
         message = frame.get("message")
         try:

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

@@ -3100,6 +3100,51 @@ describe('PythonCodeRuntime — hostile peer', () => {
     expect(result.value).toBe('released')
   }, 15_000)
 
+  it('keeps the reply pump alive when RuntimeError is rebound before the program runs', async () => {
+    // `_pump_replies` catches a closed-loop scheduling failure with `except
+    // _RuntimeError`. If that name were bound as a pump BODY local, it would be
+    // captured at pump-start — but `_run` reaches the model's top-level
+    // statements (which run before the pump's first step, since there is no
+    // suspension point between `create_task` and `await __dsh_main__`) with the
+    // rebind already applied, so `_RuntimeError` would capture the REBOUND class
+    // and the closed-loop `RuntimeError` would escape, killing the pump. Binding
+    // it as a DEF-TIME default argument captures the original before any model
+    // code runs. This rebinds `__main__.RuntimeError` as the very first program
+    // statement and drives the closed-loop worker pattern: the pump must survive
+    // the dead-loop reply and deliver the later binding.
+    let releaseSlow!: () => void
+    const slowGate = new Promise<void>((resolve) => { releaseSlow = resolve })
+    const { runtime } = await setup({ maxWallMs: 6_000 })
+    const result = await runtime.run({
+      program: [
+        'import __main__',
+        '__main__.RuntimeError = ValueError',
+        'import asyncio, threading',
+        'closed = threading.Event()',
+        'def worker():',
+        '    async def body():',
+        '        try:',
+        '            await asyncio.wait_for(tools.slow({}), timeout=0.1)',
+        '        except asyncio.TimeoutError:',
+        '            pass',
+        '    asyncio.run(body())',
+        '    closed.set()',
+        't = threading.Thread(target=worker)',
+        't.start()',
+        'while not closed.is_set():',
+        '    await asyncio.sleep(0.02)',
+        'after = await tools.release({})',
+        'return after',
+      ].join('\n'),
+      bindings: tools({
+        slow: async () => { await slowGate; return 'late' },
+        release: async () => { releaseSlow(); await new Promise(resolve => setImmediate(resolve)); return 'released' },
+      }),
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.value).toBe('released')
+  }, 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