fix(code-runtime-python): send SIGKILL at the reap-poll deadline, not cancel it

The group-reap poll folded its deadline arm into the empty-group arm, so a host
event loop blocked past graceMs + CLOSE_REAP_MARGIN_MS would run the overdue
poll before the grace SIGKILL timer: the group is still non-empty, the deadline
has passed, and the shared arm cancelled the never-fired SIGKILL and finalized —
releasing a SIGTERM-ignoring same-group survivor for good. Split the arms: empty
group cancels the moot timer and finalizes; deadline-with-non-empty-group sends
SIGKILL itself (idempotent if the timer already ran) before finalizing. Adds a
regression test that busy-blocks the loop past both timers and asserts the
survivor's heartbeat freezes.

packages/code-runtime/code-runtime-python/src/index.ts

@@ -1082,14 +1082,22 @@ export class PythonCodeRuntime extends CodeRuntime {
         }
         const deadline = Date.now() + this.config.graceMs + CLOSE_REAP_MARGIN_MS
         const pollGroup = (): void => {
-          // The deadline is a backstop: the group is reachable by `kill(-pid)`
-          // and SIGKILL is uncatchable, so it always empties within graceMs — the
-          // `Date.now() >= deadline` arm exists only so a probe that never sees
-          // ESRCH (a kernel quirk) cannot hang disposal forever.
-          /* v8 ignore next -- SIGKILL always empties the reachable group before the deadline. */
-          if (groupEmpty() || Date.now() >= deadline) {
-            // graceTimer is always defined here: pollGroup runs only when
-            // `killing` is set, and kill() arms graceTimer before any settle.
+          if (groupEmpty()) {
+            // The group is gone; the grace SIGKILL is moot. Cancel it (it may not
+            // have fired yet) and finalize. graceTimer is always defined here:
+            // pollGroup runs only when `killing` is set, and kill() armed it.
+            clearTimeout(graceTimer)
+            finalize()
+            return
+          }
+          if (Date.now() >= deadline) {
+            // Deadline reached with the group still non-empty. This is reachable
+            // when the host event loop was blocked past both timers: Node runs
+            // this poll before the grace SIGKILL timer, so that SIGKILL may never
+            // have fired. Send it HERE before finalizing — idempotent if the timer
+            // already ran — so a SIGTERM-ignoring same-group survivor is actually
+            // reaped rather than released by cancelling an unfired escalation.
+            killGroup('SIGKILL')
             clearTimeout(graceTimer)
             finalize()
             return

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

@@ -2102,6 +2102,64 @@ describe('PythonCodeRuntime — budgets, termination, disposal', () => {
     await new Promise(resolve => setTimeout(resolve, 500))
     expect(mtime()).toBe(afterDispose)
   }, 20_000)
+
+  it('sends SIGKILL at the poll deadline when the event loop was blocked past both timers', async () => {
+    // If the host event loop is blocked (a big synchronous computation) from
+    // before the group-reap poll was scheduled until after the deadline, both the
+    // poll timer and the grace-window SIGKILL timer are overdue when the loop
+    // resumes. Node runs the earlier-scheduled poll first, so the SIGKILL timer
+    // may not have fired yet. The deadline arm must then send SIGKILL ITSELF
+    // 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 readyMarker = join(handoff, 'ready')
+    const heartbeat = join(handoff, 'heartbeat')
+    const graceMs = 300
+    const { runtime } = await setup({ maxWallMs: 10_000, graceMs })
+    const result = await runtime.run({
+      program: [
+        'import subprocess, sys, os, time',
+        `marker = ${JSON.stringify(readyMarker)}`,
+        `heartbeat = ${JSON.stringify(heartbeat)}`,
+        '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)',
+        'return "spawned"',
+      ].join('\n'),
+      bindings: [],
+    })
+    expect(result.error).toBeUndefined()
+    expect(existsSync(readyMarker)).toBe(true)
+    // Block the event loop synchronously past graceMs + CLOSE_REAP_MARGIN_MS
+    // (2000) with margin, so both timers are overdue when the loop resumes.
+    const blockUntil = Date.now() + graceMs + 2_000 + 800
+    while (Date.now() < blockUntil) { /* busy-wait, no yield */ }
+    // Yield: the overdue poll runs (group still non-empty, deadline passed) and
+    // must send SIGKILL itself. The survivor then stops bumping the heartbeat.
+    const mtime = (): number => { try { return statSync(heartbeat).mtimeMs } catch { return 0 } }
+    const stopDeadline = Date.now() + 5_000
+    let last = mtime()
+    let stopped = false
+    while (Date.now() < stopDeadline) {
+      await new Promise(resolve => setTimeout(resolve, 400))
+      const now = mtime()
+      if (now === last && now !== 0) { stopped = true; break }
+      last = now
+    }
+    expect(stopped).toBe(true)
+  }, 20_000)
 })
 
 describe('PythonCodeRuntime — hostile peer', () => {