fix(code-runtime-python): keep a completed run in live until its group is reaped

settle() dropped the run from `live` eagerly, before the grace-window SIGKILL
reaped a same-group survivor. A dispose() racing a just-resolved run() then
snapshotted an empty `live` and returned while the descendant was still alive,
so teardown's "no subprocess outlives the fiber" (and its JSDoc) was false for
that window. The run now stays in `live` until the process-group poll confirms
the group empty, at which point it is both dropped from `live` and its finished
promise resolved. Adds a regression test asserting dispose() of a completed run
with a same-group survivor returns only after the survivor stops executing.

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

@@ -1032,20 +1032,7 @@ export class PythonCodeRuntime extends CodeRuntime {
       const settle = (result: Omit<CodeRunResult, 'logs'>): void => {
         if (resolved) return
         resolved = true
-        // The grace-window SIGKILL timer is intentionally NOT cleared here: a
-        // 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; 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
-        // may sit in the SIGTERM grace window, and a concurrent teardown()
-        // snapshot of `this.live` must still see it so disposal awaits its exit
-        // ("no subprocess outlives the fiber"). teardown's own settle() on an
-        // already-finished run hits the resolved guard as a no-op.
-        this.live.delete(live)
         // The child has exited by now (settle runs on `close`, or on a spawn
         // that produced no pid), so its staging directory is no longer read and
         // this run's copy goes away with it. Removed SYNCHRONOUSLY, before
@@ -1063,29 +1050,41 @@ export class PythonCodeRuntime extends CodeRuntime {
           // checked-in scripts.
         }
         resolve({ ...result, logs })
+        // Mark the fiber quiescent for THIS run: drop it from `live` and resolve
+        // `finished` (what teardown awaits). Deferred until the process group is
+        // actually empty — dropping from `live` before then would let a
+        // `dispose()` that races a just-resolved run() snapshot an empty `live`
+        // and return while a same-group survivor is still alive, making teardown's
+        // "no subprocess outlives the fiber" false for that window. Keeping the
+        // run in `live` until the group is reaped is exactly what makes a
+        // concurrent teardown await it.
+        const finalize = (): void => {
+          this.live.delete(live)
+          finishResolve()
+        }
         // `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, cancel the pending SIGKILL and resolve now.
+        // group is already empty, cancel the pending SIGKILL and finalize now.
         // Clearing it is what bounds the PID-reuse hazard: an armed `kill(-pid)`
         // left to fire up to graceMs later could hit a RECYCLED pgid once the
         // kernel reused the leader's pid, SIGKILLing an unrelated group. So the
         // timer stays armed only while a real survivor exists — a same-group
         // descendant that ignored SIGTERM but released the pipes, still alive
         // here because its `close` is what got us to settle. In that case
-        // withhold `finished` and poll the group on REF'd timers (a short-lived
+        // withhold finalize and poll the group on REF'd timers (a short-lived
         // host would otherwise exit before the unref'd SIGKILL fired, reparenting
         // the survivor to init), clearing the timer the moment the group empties;
         // the wait is bounded by the same graceMs + margin the escalation uses.
         if (!killing || groupEmpty()) {
           if (graceTimer !== undefined) clearTimeout(graceTimer)
-          finishResolve()
+          finalize()
           return
         }
         const deadline = Date.now() + this.config.graceMs + CLOSE_REAP_MARGIN_MS
         const pollGroup = (): void => {
           if (groupEmpty() || Date.now() >= deadline) {
             if (graceTimer !== undefined) clearTimeout(graceTimer)
-            finishResolve()
+            finalize()
             return
           }
           setTimeout(pollGroup, GROUP_REAP_POLL_MS)

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

@@ -2053,6 +2053,55 @@ describe('PythonCodeRuntime — budgets, termination, disposal', () => {
     }
     expect(still).toBe(true)
   }, 20_000)
+
+  it('dispose awaits reaping of a same-group survivor from a completed run', async () => {
+    // The quiescence contract also holds for a run that ALREADY resolved: the run
+    // stays tracked in `live` until its process group is reaped, so a `dispose()`
+    // that races a just-returned run() still awaits the survivor rather than
+    // snapshotting an empty `live` and returning while it lives. Here the run
+    // completes (leaving a SIGTERM-ignoring same-group descendant), then dispose()
+    // 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 readyMarker = join(handoff, 'ready')
+    const heartbeat = join(handoff, 'heartbeat')
+    const { runtime, fiber } = await setup({ maxWallMs: 10_000, graceMs: 300 })
+    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)
+    // dispose() must not return until the group is reaped. After it resolves, the
+    // heartbeat must already be stale: read its mtime, wait past the heartbeat
+    // interval, and confirm it did not advance — the descendant is no longer
+    // executing (reaped or zombie), so teardown was genuinely quiescent.
+    await fiber.dispose()
+    const mtime = (): number => { try { return statSync(heartbeat).mtimeMs } catch { return 0 } }
+    const afterDispose = mtime()
+    await new Promise(resolve => setTimeout(resolve, 500))
+    expect(mtime()).toBe(afterDispose)
+  }, 20_000)
 })
 
 describe('PythonCodeRuntime — hostile peer', () => {