From 38b425569afe9ee208ea4547c7ead08a1181eb97 Mon Sep 17 00:00:00 2001 From: Ansel-2428 Date: Sat, 17 Oct 2026 18:10:15 -0400 Subject: [PATCH 01/28] fix(code-execution): the protocol gets fds of its own, and the host reads a script's output The code_execution runtime ran its protocol on fds 0 and 1, which every script and every command it runs shares, and kept a script's output in memory, read back at the script's end. So a command's output (subprocess without capture_output, os.system, C code writing to fds 1 and 2) reached only the host's log as "non-protocol" lines; a line without a newline ran into exec_result and left the exec waiting out its time limit; a command reading its input (cat, ssh) took the host's messages; os.close(0) or (1) broke the interpreter; and whatever a script printed died with the interpreter on a kill or a crash, along with a fatal error's text. Now the host spawns the runtime with stdio ignore/pipe/pipe/pipe/pipe: - fd 0 is /dev/null (Node's 'ignore'), so a command reading its input gets its end at once. - fds 3 (host -> py) and 4 (py -> host) carry the protocol. The runtime first checks both are open, and if not writes one sentence to fd 2 (the host's channel didn't reach it; a pythonPath wrapper must pass fds 3 and 4 on, which a container or sudo's defaults don't) and exits 2, so the failed start says what to fix. Then it makes them non-inheritable, so no command is given them. No fallback to fds 0 and 1, which would keep two protocols. - fds 1 and 2 are pipes the host reads itself (OutputStream). Each exec op carries a random mark; when the script returns, the runtime sends exec_result (now only the return code), flushes the script's writers and writes the mark into both streams through private copies of fds 1 and 2. The host settles once it has exec_result and both marks, so everything the script and the commands it waited for wrote is in the result; each stream is capped at 2,000,000 characters with the same note as before. - Output arriving while no exec runs (a command left running) goes to the host's log, labelled by stream; after a reclaim the old pipes stay read until whatever holds them closes them. - The runtime restores fds 0, 1 and 2 from its private copies at each exec's start, and gives the script fresh line-buffered writers over 1 and 2 (UTF-8, backslashreplace): a line reaches the host as it's printed, in order with commands' output. 100,000 prints cost about three times what CappedIO did (measured while designing). - faulthandler is enabled on the runtime's private copy of fd 2, so a segfault or an abort leaves a traceback; Py_FatalError's own text reaches the host because fd 2 is the host's pipe. - A kill, abort, dispose or crash now settles with the output so far ahead of its message (stopPending). An unexpected exit waits for both pipes to end, or EXIT_OUTPUT_GRACE_MS when a command still holds them, and logs the last of what the interpreter wrote. A start that fails says what the interpreter wrote to stderr. - Background: the runtime joins fd 2 onto fd 1's pipe, so the journal keeps the order written; the host writes the journal (Journal: same 50 MB cap, same 4,000-character tail) in place of the runtime's LogTee, so a crashed background script's tail is real. - When the runtime can't go on because a script closed one of its own fds (fd 4, or its private copies), it says why on stderr and exits (code 70) rather than leaving the exec to its time limit. #458's kill of an interpreter that stops reading moves to fd 3; since fd 3 is a duplex socket that ends when the far side closes, the first message that can't reach it is caught by the write's callback as well as by 'error'. Removed from the runtime: CappedIO, LogTee, the result assembly and the log-file handling. The tool description now says what the script and the commands it runs print comes back; CodeExecutionConfig.pythonPath's doc says a wrapper must pass fds 3 and 4 on. Changelog: the script-facing changes in a fixed fragment; the hosts' part (a pythonPath wrapper must pass fds 3 and 4 on; PYTHON_RUNTIME_SOURCE's new protocol) in a breaking fragment under **Hosts:**, so this releases in a new minor version. Tests: the stand-ins that stop reading now close fd 3 and answer on fd 4; the stop-table driver reads the script's output from fds 1 and 2 through a file; new real-runtime tests for each piece above. Co-Authored-By: Claude Opus 5.5 --- ...-code-execution-output-to-host.breaking.md | 1 + ...fix-code-execution-output-to-host.fixed.md | 7 + src/code-execution/py-runner.ts | 493 ++++++++++++++---- src/code-execution/runtime-py.ts | 302 ++++++----- src/code-execution/tool-definition.ts | 8 +- src/framework.ts | 6 +- src/types/framework.ts | 8 +- test/code-execution.test.ts | 333 +++++++++++- test/helpers/pytc-stop-rows.ts | 41 +- 9 files changed, 927 insertions(+), 272 deletions(-) create mode 100644 changelog.d/fix-code-execution-output-to-host.breaking.md create mode 100644 changelog.d/fix-code-execution-output-to-host.fixed.md diff --git a/changelog.d/fix-code-execution-output-to-host.breaking.md b/changelog.d/fix-code-execution-output-to-host.breaking.md new file mode 100644 index 000000000..731474ee2 --- /dev/null +++ b/changelog.d/fix-code-execution-output-to-host.breaking.md @@ -0,0 +1 @@ +- **Hosts:** the `code_execution` interpreter now takes its channel to the host on its file descriptors 3 and 4, and `PYTHON_RUNTIME_SOURCE`, which the package exports, speaks a new protocol there. **Who needs to act:** a host whose `pythonPath` starts Python through a wrapper that doesn't pass fds 3 and 4 on, such as one that runs it in a container or under sudo with its default settings, and a host that runs `PYTHON_RUNTIME_SOURCE` itself rather than through `PyRunner`. Until it acts, every script fails to start, and the result says the host's channel didn't reach the interpreter. **Migration:** have the wrapper `exec` Python, or pass fds 3 and 4 on. A host that runs the runtime itself opens fd 3 for its messages to the runtime and fd 4 for the runtime's, gives it `/dev/null` as fd 0, reads its fds 1 and 2 itself, puts a `mark` in each `exec`, and takes the script's output on each stream up to that mark, which the runtime writes once the script returns: `exec_result` now carries only the return code. **Unchanged:** `PyRunner`'s interface, `ExecResult` included, and a `pythonPath` that names an interpreter or a wrapper that `exec`s one. No sibling package needs a new version. diff --git a/changelog.d/fix-code-execution-output-to-host.fixed.md b/changelog.d/fix-code-execution-output-to-host.fixed.md new file mode 100644 index 000000000..94b150504 --- /dev/null +++ b/changelog.d/fix-code-execution-output-to-host.fixed.md @@ -0,0 +1,7 @@ +- What the commands a `code_execution` script runs write to stdout and stderr now comes back in the script's result, in the order written on each stream, beside the script's own prints: `subprocess` without `capture_output`, `os.system`, and C code writing to file descriptors 1 and 2. It used to reach only the host's log, and a line a command wrote without a newline ran into the runtime's own reply, so the script waited out its time limit. A command with a lot of output now fills the result as the script's own prints do: each stream keeps up to 2,000,000 characters, and a result past 24,000 characters spills to a workspace file, as any result does. The tool's description now says that what the script and the commands it runs print comes back. +- A command a `code_execution` script runs now reads the end of its input at once: its standard input is `/dev/null`. It used to read the runtime's channel from the host, so a command that reads its input, such as `cat`, or `ssh` without `-n`, took the host's messages, and the script failed or waited out its time limit. +- A script that closes or redirects its file descriptors 0, 1 or 2 no longer breaks the interpreter: the runtime's channel to the host has file descriptors of its own, 3 and 4, which no command a script runs is given, and each script starts with 0, 1 and 2 as the host set them. A script's `os.close(1)` used to leave it waiting out its time limit, and its `os.close(0)` sometimes left the next script doing so. +- What a script and its commands wrote before it was stopped now comes back ahead of the message saying why, when it's killed at its time limit, aborted, or its interpreter crashes; it used to be lost. A crash's own text comes back too, and goes to the host's log: a fatal error's, and now a traceback for a segfault or an abort, which faulthandler writes where CPython alone wrote nothing. +- A background script's journal now holds its commands' output too, and a background script that crashes leaves the last of what it wrote, its crash's text included, in the message that wakes its agent, where that read only "python runtime exited unexpectedly". +- A script's prints now reach the host as each line is complete, so a partial line waits for its newline, a flush or the end of the script. Printing costs more than it did: 100,000 short `print` calls took about half a second, against about a sixth of a second before, on a busy machine. +- An interpreter that exits as it starts now says in the result what it wrote to stderr. Output that arrives while no script runs, such as from a command a script left running, reaches the host's log as `stdout while no script runs: …` or `stderr while no script runs: …`, where stdout's used to read `non-protocol stdout line: …`. diff --git a/src/code-execution/py-runner.ts b/src/code-execution/py-runner.ts index 3b37f553a..8d76f003b 100644 --- a/src/code-execution/py-runner.ts +++ b/src/code-execution/py-runner.ts @@ -12,13 +12,21 @@ * What this class guarantees is liveness: a wedged or runaway script cannot * hang the agent turn (cancel + SIGINT -> grace -> SIGKILL -> respawn) and a crashed * interpreter surfaces as a tool result, never as an unhandled rejection. + * + * The protocol runs on the interpreter's fds 3 (host -> py) and 4 (py -> host); + * its fd 0 is /dev/null, and the host reads its fds 1 and 2 itself (see + * OutputStream), so what a script and the commands it runs write to them + * reaches the host whatever then happens to the interpreter. */ -import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'; -import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { spawn, type ChildProcess } from 'node:child_process'; +import { randomBytes } from 'node:crypto'; +import { closeSync, mkdirSync, mkdtempSync, openSync, rmSync, writeFileSync, writeSync } from 'node:fs'; import { tmpdir } from 'node:os'; -import { join } from 'node:path'; +import { dirname, join } from 'node:path'; import { createInterface, type Interface } from 'node:readline'; +import type { Readable, Writable } from 'node:stream'; +import { StringDecoder } from 'node:string_decoder'; import { PYTHON_RUNTIME_SOURCE } from './runtime-py.js'; export interface InjectedTool { @@ -41,7 +49,7 @@ export interface ExecResult { /** Per-exec options for background (daemon) scripts. */ export interface BackgroundExecOptions { - /** Absolute path for the line-buffered stdout+stderr journal (null: tail only). */ + /** Absolute path for the stdout+stderr journal, which the runner appends to as the output arrives (null: tail only). */ logPath: string | null; /** * Called on each wake_agent() from the script. Resolve with null to ack @@ -60,6 +68,7 @@ export type ScriptToolCallHandler = ( ) => Promise; export interface PyRunnerOptions { + /** Python to spawn (default 'python3'). A wrapper must pass fds 3 and 4 on, where the protocol runs (see CodeExecutionConfig). */ pythonPath?: string; /** Per-tool-call timeout surfaced as TimeoutError inside the script. */ toolCallTimeoutMs?: number; @@ -80,6 +89,20 @@ const DEFAULT_IDLE_RECLAIM_MS = 300_000; const CANCEL_GRACE_MS = 10_000; /** After the deadline, SIGINT is repeated at this interval until the script reports or is killed. */ const INTERRUPT_REPEAT_MS = 200; +/** Each of a script's stdout and stderr is kept up to this many characters in its result. */ +const OUTPUT_CAP_CHARS = 2_000_000; +/** A background script's journal stops growing past this many bytes. */ +const LOG_FILE_CAP_BYTES = 50_000_000; +/** How much of a background script's output its result keeps, for the envelope that tells the agent how it ended. */ +const TAIL_CHARS = 4000; +/** + * How long an interpreter that exited unexpectedly gets for its last output to + * arrive when something still holds its output pipes open, such as a command its + * script left running; otherwise their end says it has all arrived. + */ +const EXIT_OUTPUT_GRACE_MS = 200; +/** How much of what an interpreter wrote before it failed to start, or crashed, goes into the message and the host's log. */ +const LAST_WORDS_CHARS = 2000; /** The longest delay Node's timers honour (~24.8 days); a longer one fires after ~1 ms. */ export const MAX_TIMER_MS = 2_147_483_647; @@ -91,6 +114,197 @@ export function formatLimit(ms: number): string { return `${Math.round(ms / 360_000) / 10} h`; } +/** `line` after `text`, on a line of its own: a text that doesn't end with a newline gets one first. */ +function onItsOwnLine(text: string, line: string): string { + return (text === '' || text.endsWith('\n') ? text : `${text}\n`) + line; +} + +/** + * A background script's journal: the log file the agent can read with its own + * tools while the script runs, and the tail its result carries. + */ +class Journal { + private fd: number | null = null; + private written = 0; + private capped = false; + tail = ''; + + constructor(path: string | null, log: (text: string) => void) { + if (!path) return; + try { + mkdirSync(dirname(path), { recursive: true }); + this.fd = openSync(path, 'a'); + } catch (err) { + log(`cannot open log file: ${err instanceof Error ? err.message : String(err)}`); + } + } + + write(text: string): void { + if (text === '') return; + this.tail = (this.tail + text).slice(-TAIL_CHARS); + if (this.fd === null || this.capped) return; + const bytes = Buffer.byteLength(text); + try { + if (this.written + bytes > LOG_FILE_CAP_BYTES) { + this.capped = true; + writeSync(this.fd, `\n[log capped at ${LOG_FILE_CAP_BYTES} bytes]\n`); + } else { + writeSync(this.fd, text); + this.written += bytes; + } + } catch { + // A journal that can't be written to still keeps its tail. + } + } + + close(): void { + if (this.fd === null) return; + try { + closeSync(this.fd); + } catch { + // already gone + } + this.fd = null; + } +} + +const NOTHING = Buffer.alloc(0); + +/** + * One of the interpreter's output pipes, its fd 1 or fd 2, as the host reads it. + * While an exec runs, what arrives before that exec's end mark is the exec's: + * kept for its result, up to OUTPUT_CAP_CHARS, or written to a background + * script's journal. The mark itself is cut out. What arrives while no exec runs, + * such as a command a script left running writes, goes to the host's log. + */ +class OutputStream { + /** Every process holding the pipe has closed it, so nothing more will come. */ + ended = false; + /** The running exec's mark has arrived, so all of its output on this pipe has. */ + marked = false; + /** The end of what arrived while no exec ran: an interpreter that failed to start says why here. */ + lastIdle = ''; + private mark: Buffer | null = null; + /** The end of a chunk that could be the start of a mark the next chunk completes. */ + private carry: Buffer = NOTHING; + private readonly decoder = new StringDecoder('utf8'); + private text = ''; + private truncated = false; + private journal: Journal | null = null; + private onEnd: (() => void)[] = []; + + constructor( + private readonly name: 'stdout' | 'stderr', + pipe: Readable, + private readonly log: (text: string) => void, + private readonly onMarked: () => void, + ) { + pipe.on('data', (chunk: Buffer) => this.push(chunk)); + const ended = () => { + if (this.ended) return; + this.ended = true; + for (const fn of this.onEnd.splice(0)) fn(); + }; + pipe.on('end', ended); + pipe.on('close', ended); + pipe.on('error', (err) => { + log(`reading the python runtime's ${name} failed: ${err.message}`); + ended(); + }); + } + + /** Begin an exec's share of the pipe, which its mark ends. */ + begin(mark: Buffer, journal: Journal | null): void { + this.mark = mark; + this.marked = false; + this.journal = journal; + this.text = ''; + this.truncated = false; + } + + /** The exec's share as it stands, with a note if it was cut at the cap; what arrives next goes to the log. */ + take(): string { + if (this.mark !== null) { + // Settled before its mark came (a kill, an abort or a crash): what arrived counts. + this.keep(this.carry); + this.carry = NOTHING; + this.keep(this.decoder.end()); + } + const text = this.text + (this.truncated ? `\n[${this.name} truncated at ${OUTPUT_CAP_CHARS} chars]` : ''); + this.mark = null; + this.journal = null; + this.text = ''; + this.truncated = false; + return text; + } + + /** Run fn once the pipe has ended (at once if it has). */ + whenEnded(fn: () => void): void { + if (this.ended) fn(); + else this.onEnd.push(fn); + } + + private push(chunk: Buffer): void { + let bytes = this.carry.length > 0 ? Buffer.concat([this.carry, chunk]) : chunk; + this.carry = NOTHING; + if (this.mark !== null) { + const at = bytes.indexOf(this.mark); + if (at < 0) { + const hold = Math.min(bytes.length, this.mark.length - 1); + this.keep(bytes.subarray(0, bytes.length - hold)); + this.carry = bytes.subarray(bytes.length - hold); + return; + } + this.keep(bytes.subarray(0, at)); + this.keep(this.decoder.end()); + bytes = bytes.subarray(at + this.mark.length); + this.mark = null; + this.marked = true; + this.onMarked(); + } + if (bytes.length > 0) { + const text = this.decoder.write(bytes); + this.lastIdle = (this.lastIdle + text).slice(-LAST_WORDS_CHARS); + if (text.trim() !== '') this.log(`${this.name} while no script runs: ${text.trimEnd()}`); + } + } + + /** Keep what arrived for the running exec. */ + private keep(part: Buffer | string): void { + const text = typeof part === 'string' ? part : this.decoder.write(part); + if (text === '') return; + if (this.journal) { + this.journal.write(text); + return; + } + if (this.truncated) return; + const room = OUTPUT_CAP_CHARS - this.text.length; + if (text.length <= room) { + this.text += text; + return; + } + this.text += text.slice(0, room); + this.truncated = true; + } +} + +/** + * Run fn once both of an exited interpreter's output pipes have ended, which + * says the last of what it wrote has arrived, or after EXIT_OUTPUT_GRACE_MS if + * something, such as a command its script left running, still holds them. + */ +function afterOutputEnds(stdout: OutputStream, stderr: OutputStream, fn: () => void): void { + let done = false; + const finish = () => { + if (done) return; + done = true; + clearTimeout(timer); + fn(); + }; + const timer = setTimeout(finish, EXIT_OUTPUT_GRACE_MS); + stdout.whenEnded(() => stderr.whenEnded(finish)); +} + interface PendingExec { id: string; resolve: (result: ExecResult) => void; @@ -105,6 +319,11 @@ interface PendingExec { * waiting for the interpreter to start, and an interpreter that exits then * failed to start (see exec). */ sent: boolean; + /** A background exec's journal, which its output goes to instead of its result. */ + journal: Journal | null; + /** The script's return code, once the interpreter has reported it; the exec + * settles when its output's end marks have arrived too. */ + returnCode: number | null; } export class PyRunner { @@ -117,9 +336,17 @@ export class PyRunner { private readonly label: string; private onWake: ((line: number, payload: unknown) => Promise) | null = null; - private child: ChildProcessWithoutNullStreams | null = null; + private child: ChildProcess | null = null; private childReady: Promise | null = null; + /** The interpreter's fd 3, where the host's messages go. */ + private protoIn: Writable | null = null; + /** Ends the interpreter whose fd 3 a message couldn't reach (see ensureChild). */ + private unreachable: ((why: string) => void) | null = null; + /** Reads the interpreter's messages from its fd 4. */ private reader: Interface | null = null; + /** The interpreter's stdout and stderr, as the host reads them. */ + private out: OutputStream | null = null; + private err: OutputStream | null = null; private runtimeDir: string | null = null; private pending: PendingExec | null = null; private idleTimer: ReturnType | null = null; @@ -187,6 +414,8 @@ export class PyRunner { settled: false, deadlineMs: null, sent: false, + journal: null, + returnCode: null, }; this.pending = pending; try { @@ -227,30 +456,30 @@ export class PyRunner { pending.interruptTimer = setInterval(() => this.interruptChild(), INTERRUPT_REPEAT_MS); pending.killTimer = setTimeout(() => { pending.deadlineMs = null; // this message already says why - this.settlePending({ - stdout: '', - stderr: - `script stopped: it reached its ${formatLimit(deadlineMs)} time limit and did not respond, ` + + this.stopPending( + `script stopped: it reached its ${formatLimit(deadlineMs)} time limit and did not respond, ` + 'so it was killed and the interpreter restarted (variables from earlier scripts are gone)', - returnCode: 1, - aborted: true, - }); + ); this.reclaim('deadline-kill'); }, this.cancelGraceMs); }, deadlineMs); // A day-scale background deadline must not hold the process open. if (background) pending.deadlineTimer.unref?.(); + // What the script writes, up to its mark, is this exec's: a background + // script's goes to its journal. + const mark = Buffer.from(`\x1epytc-end-${randomBytes(16).toString('hex')}\x1e`); + pending.journal = background ? new Journal(background.logPath, (text) => this.log(text)) : null; + this.out?.begin(mark, pending.journal); + this.err?.begin(mark, pending.journal); pending.sent = true; this.send({ op: 'init', tools: tools.map((t) => ({ py_name: t.pyName, tool_name: t.toolName })), call_timeout_s: Math.round(this.toolCallTimeoutMs / 1000), - ...(background - ? { background: true, log_path: background.logPath ?? null } - : {}), + ...(background ? { background: true } : {}), }); - this.send({ op: 'exec', id: execId, code }); + this.send({ op: 'exec', id: execId, code, mark: mark.toString('utf8') }); const result = await settled; @@ -270,26 +499,14 @@ export class PyRunner { abort(reason: string): void { if (!this.pending) return; this.send({ op: 'cancel', id: this.pending.id, reason }); - this.settlePending({ - stdout: '', - stderr: `script aborted by host: ${reason}`, - returnCode: 1, - aborted: true, - }); + this.stopPending(`script aborted by host: ${reason}`); this.reclaim(`abort: ${reason}`); } dispose(): void { if (this.disposed) return; this.disposed = true; - if (this.pending) { - this.settlePending({ - stdout: '', - stderr: 'code_execution runner disposed', - returnCode: 1, - aborted: true, - }); - } + this.stopPending('code_execution runner disposed'); this.reclaim('dispose'); } @@ -306,32 +523,38 @@ export class PyRunner { writeFileSync(runtimePath, PYTHON_RUNTIME_SOURCE, 'utf8'); const child = spawn(this.pythonPath, [runtimePath], { - stdio: ['pipe', 'pipe', 'pipe'], + // fd 0 is /dev/null, the host reads fds 1 and 2, and the protocol has 3 and 4. + stdio: ['ignore', 'pipe', 'pipe', 'pipe', 'pipe'], env: { ...process.env, PYTHONUNBUFFERED: '1' }, }); this.child = child; - - child.stderr.setEncoding('utf8'); - child.stderr.on('data', (chunk: string) => { - // Outside an exec, python stderr is runtime diagnostics (crash - // tracebacks etc.) — surface them in host logs. - console.error(`[pytc:${this.label}] ${chunk.trimEnd()}`); - }); - - // A write to an interpreter that no longer reads its input fails on the - // stream after write() returns; with no listener, that error is thrown as - // an uncaught exception. Such an interpreter can go on running, as after a - // script closes file descriptor 0, and every later message to it is lost: - // its exec would wait out its time limit. So end it, and its exit settles - // the exec now. One already dead or killed is left to its own exit. + const protoIn = child.stdio[3] as Writable; + const protoOut = child.stdio[4] as Readable; + this.protoIn = protoIn; + const stdout = new OutputStream('stdout', child.stdout as Readable, (text) => this.log(text), () => this.settleIfComplete(child)); + const stderr = new OutputStream('stderr', child.stderr as Readable, (text) => this.log(text), () => this.settleIfComplete(child)); + this.out = stdout; + this.err = stderr; + protoOut.on('error', (e) => this.log(`reading the python runtime's messages failed: ${e.message}`)); + + // An interpreter that no longer reads the host's messages, as after a + // script closes file descriptor 3, could go on running, and every later + // message to it is lost: its exec would wait out its time limit. So the + // first message that can't reach it ends it, and its exit settles the exec + // now. Such a write fails after write() returns: the pipe refuses it, or + // has already ended, its far end closed. Its error, which with no listener + // would be thrown as an uncaught exception, is heard here too. One already + // dead or killed is left to its own exit. let inputFailed = false; - child.stdin.on('error', (err) => { - console.error(`[pytc:${this.label}] write to the python runtime failed: ${err.message}`); + const unreachable = (why: string) => { + this.log(`write to the python runtime failed: ${why}`); if (child.exitCode === null && !child.killed && child.pid !== undefined) { inputFailed = true; child.kill('SIGKILL'); } - }); + }; + this.unreachable = unreachable; + protoIn.on('error', (e) => unreachable(e.message)); child.on('error', (err) => { // Spawn failure lands here (e.g. python3 missing); readiness rejects. @@ -341,51 +564,58 @@ export class PyRunner { child.on('exit', (exitCode, signal) => { // An exec still waiting for the interpreter is settled by exec, as a // runtime that failed to start. - if (this.pending?.sent) { - this.settlePending({ - stdout: '', - stderr: - inputFailed && signal === 'SIGKILL' - ? 'python runtime stopped reading its input, so it was killed and the interpreter restarted ' + - '(variables from earlier scripts are gone). A script that closes its standard input ' + - '(file descriptor 0) does this.' - : `python runtime exited unexpectedly (code=${exitCode}, signal=${signal ?? 'none'})`, - returnCode: 1, - aborted: true, - }); - } - if (this.child === child) { - this.teardownChild(); + const pending = this.pending; + if (!pending?.sent) { + if (this.child === child) this.teardownChild(); + return; } + const why = + inputFailed && signal === 'SIGKILL' + ? "python runtime stopped reading the host's messages, so it was killed and the interpreter restarted " + + '(variables from earlier scripts are gone). A script that closes file descriptor 3, where they ' + + 'arrive, does this.' + : `python runtime exited unexpectedly (code=${exitCode}, signal=${signal ?? 'none'})`; + // What it wrote last, such as a fatal error's own text, is already on its + // way through its output pipes: settle once they've ended, or after a + // short grace while a command it started still holds them. + afterOutputEnds(stdout, stderr, () => { + if (this.pending === pending) { + const last = this.stopPending(why).trim(); + this.log(last ? `${why}; the last it wrote: ${last.slice(-LAST_WORDS_CHARS)}` : why); + } + if (this.child === child) this.teardownChild(); + }); }); - this.reader = createInterface({ input: child.stdout }); + this.reader = createInterface({ input: protoOut }); this.reader.on('line', (line) => this.handleLine(line, child)); this.childReady = new Promise((resolve, reject) => { - const onReady = () => { - cleanup(); - resolve(); - }; - const onExit = () => { - cleanup(); - reject(new Error('python runtime exited before becoming ready')); - }; - const onError = (err: Error) => { - cleanup(); - reject(err); - }; - const timeout = setTimeout(() => { - cleanup(); - reject(new Error('python runtime did not become ready within 15s')); - }, 15_000); - const cleanup = () => { + let over = false; + const finish = (outcome: () => void) => { + if (over) return; + over = true; clearTimeout(timeout); - this.readyResolver = null; - this.readyFailer = null; + // Another interpreter may have started since this one exited. + if (this.readyResolver === onReady) this.readyResolver = null; + if (this.readyFailer === onError) this.readyFailer = null; child.off('exit', onExit); child.off('error', onError); + outcome(); }; + const onReady = () => finish(resolve); + const onExit = () => { + // Why it ended, such as a traceback, is on its way through its stderr. + afterOutputEnds(stdout, stderr, () => { + const said = stderr.lastIdle.trim(); + finish(() => reject(new Error(`python runtime exited before becoming ready${said ? `, saying: ${said}` : ''}`))); + }); + }; + const onError = (e: Error) => finish(() => reject(e)); + const timeout = setTimeout( + () => finish(() => reject(new Error('python runtime did not become ready within 15s'))), + 15_000, + ); this.readyResolver = onReady; this.readyFailer = onError; child.on('exit', onExit); @@ -394,16 +624,20 @@ export class PyRunner { return this.childReady; } + private log(text: string): void { + console.error(`[pytc:${this.label}] ${text}`); + } + private readyResolver: (() => void) | null = null; /** Fails the readiness of an interpreter that is still starting (see reclaim). */ private readyFailer: ((err: Error) => void) | null = null; - private handleLine(line: string, child: ChildProcessWithoutNullStreams): void { - let msg: { op?: string; id?: string; name?: string; args?: unknown; stdout?: string; stderr?: string; return_code?: number }; + private handleLine(line: string, child: ChildProcess): void { + let msg: { op?: string; id?: string; name?: string; args?: unknown; return_code?: number }; try { msg = JSON.parse(line); } catch { - console.error(`[pytc:${this.label}] non-protocol stdout line: ${line.slice(0, 200)}`); + this.log(`non-protocol line from the python runtime: ${line.slice(0, 200)}`); return; } @@ -447,24 +681,58 @@ export class PyRunner { } case 'exec_result': { - if (this.pending && msg.id === this.pending.id) { - this.settlePending({ - stdout: msg.stdout ?? '', - stderr: msg.stderr ?? '', - returnCode: typeof msg.return_code === 'number' ? msg.return_code : 1, - ...(typeof (msg as { tail?: unknown }).tail === 'string' - ? { tail: (msg as { tail: string }).tail } - : {}), - }); + const pending = this.pending; + if (pending && msg.id === pending.id && pending.returnCode === null) { + pending.returnCode = typeof msg.return_code === 'number' ? msg.return_code : 1; + this.settleIfComplete(child); } return; } default: - console.error(`[pytc:${this.label}] unknown protocol op: ${String(msg.op)}`); + this.log(`unknown protocol op: ${String(msg.op)}`); + } + } + + /** + * Settle the running exec once its script has reported and the end marks + * of both its output pipes have arrived, which says all of its output has. + */ + private settleIfComplete(child: ChildProcess): void { + const pending = this.pending; + if (child !== this.child || !pending || pending.returnCode === null) return; + if (!this.out?.marked || !this.err?.marked) return; + const stdout = this.out.take(); + const stderr = this.err.take(); + const journal = pending.journal; + if (journal) { + journal.close(); + this.settlePending({ stdout: '', stderr: '', returnCode: pending.returnCode, tail: journal.tail }); + } else { + this.settlePending({ stdout, stderr, returnCode: pending.returnCode }); } } + /** + * Settle the running exec now: what its script wrote so far, then `why` it + * stopped, on a line of its own. Returns what the script wrote to stderr (a + * background script's tail), for the host's log. + */ + private stopPending(why: string): string { + const pending = this.pending; + if (!pending || pending.settled) return ''; + const stdout = this.out?.take() ?? ''; + const stderr = this.err?.take() ?? ''; + const journal = pending.journal; + if (journal) { + journal.close(); + this.settlePending({ stdout: '', stderr: why, returnCode: 1, aborted: true, tail: onItsOwnLine(journal.tail, why) }); + return journal.tail; + } + this.settlePending({ stdout, stderr: onItsOwnLine(stderr, why), returnCode: 1, aborted: true }); + return stderr; + } + private settlePending(result: ExecResult): void { const pending = this.pending; if (!pending || pending.settled) return; @@ -474,8 +742,9 @@ export class PyRunner { // and finished its work returns 0, and must not read as stopped. if (pending.deadlineMs !== null && result.returnCode !== 0) { const note = `script stopped: it reached its ${formatLimit(pending.deadlineMs)} time limit\n`; - // On a line of its own: the runner's own texts, such as an interpreter's exit, end without a newline. - const withNote = (text: string) => (text === '' || text.endsWith('\n') ? text : `${text}\n`) + note; + // On a line of its own: a script's output, and the runner's own texts, such as an + // interpreter's exit, can end without a newline. + const withNote = (text: string) => onItsOwnLine(text, note); // A background script reports its output tail, so the note goes there too. result = { ...result, stderr: withNote(result.stderr), ...(result.tail !== undefined ? { tail: withNote(result.tail) } : {}) }; } @@ -492,9 +761,9 @@ export class PyRunner { * aborted, killed or crashed, or the interpreter reclaimed) would resolve * the same id in the one that replaced it, inside another script: drop it. */ - private reply(child: ChildProcessWithoutNullStreams, msg: unknown, what: string): void { + private reply(child: ChildProcess, msg: unknown, what: string): void { if (child !== this.child) { - console.error(`[pytc:${this.label}] dropped late ${what}: the interpreter that asked for it is gone`); + this.log(`dropped late ${what}: the interpreter that asked for it is gone`); return; } this.send(msg); @@ -502,13 +771,15 @@ export class PyRunner { private send(obj: unknown): void { const child = this.child; - if (!child || child.exitCode !== null || child.killed) return; + const input = this.protoIn; + const unreachable = this.unreachable; + if (!child || !input || child.exitCode !== null || child.killed) return; try { - child.stdin.write(JSON.stringify(obj) + '\n'); + input.write(JSON.stringify(obj) + '\n', (err) => { + if (err) unreachable?.(err.message); + }); } catch (err) { - console.error( - `[pytc:${this.label}] protocol write failed: ${err instanceof Error ? err.message : String(err)}`, - ); + this.log(`protocol write failed: ${err instanceof Error ? err.message : String(err)}`); } } @@ -534,7 +805,7 @@ export class PyRunner { /** Kill interpreter + clean temp dir. State is lost (by design). */ private reclaim(reason: string): void { if (this.child) { - console.error(`[pytc:${this.label}] reclaiming python interpreter (${reason})`); + this.log(`reclaiming python interpreter (${reason})`); } // An interpreter killed as it starts never says it's ready: fail its // readiness now, so an exec waiting on it returns at once rather than at @@ -553,13 +824,21 @@ export class PyRunner { } if (this.child) { const child = this.child; + const input = this.protoIn; this.child = null; this.childReady = null; this.readyResolver = null; this.readyFailer = null; + this.protoIn = null; + this.unreachable = null; + // Its output pipes stay read, into the host's log, until whatever still + // holds them, such as a command its script left running, closes them: + // breaking them under such a command could end it. + this.out = null; + this.err = null; child.removeAllListeners('exit'); try { - child.stdin.end(); + input?.end(); } catch { // stream may already be destroyed } diff --git a/src/code-execution/runtime-py.ts b/src/code-execution/runtime-py.ts index e1304f557..a655a0aa1 100644 --- a/src/code-execution/runtime-py.ts +++ b/src/code-execution/runtime-py.ts @@ -19,23 +19,38 @@ * the payload to the host, which injects it into the agent's context and * triggers inference. Awaits a host ack so rate limiting backpressures * inside the script instead of dropping wakes. - * - stdout/stderr tee'd line-buffered to `log_path` (agent-readable via - * their file tools) with an in-memory tail kept for the crash envelope. + * - its stderr joined to its stdout (fd 2 onto fd 1's pipe), so the host's + * journal of its output (a log file the agent can read, and a tail for the + * crash envelope) holds the two in the order they were written. + * + * OUTPUT. The protocol has file descriptors of its own: the host opens the + * runtime's fds 3 (host -> py) and 4 (py -> host), which the runtime makes + * non-inheritable as it starts, so no command a script runs is given them. fd 0 + * is /dev/null, and fds 1 and 2 are pipes the host reads itself: a script's + * prints and the output of every command it runs go there in the order + * written, and what reaches them reaches the host whatever then happens to the + * interpreter. The host gives each exec what arrived on each stream before + * that exec's end mark (the exec op's `mark`), which the runtime writes into + * both streams once the script has returned; each stream is first in, first + * out, so what the script and the commands it waited for wrote is ahead of + * it. Nothing a script or a command does with fds 0, 1 or 2 reaches the + * protocol, and the runtime puts them back as the host set them at each + * exec's start. * * Embedded as a TS string because tsc does not copy non-TS assets to dist/; * PyRunner materializes it to a temp file at spawn. * - * Protocol (JSON lines): + * Protocol (JSON lines, on fds 3 and 4): * host -> py : {op:"init", tools:[{py_name,tool_name}], call_timeout_s, - * background?, log_path?, tail_chars?} - * {op:"exec", id, code} + * background?} + * {op:"exec", id, code, mark} * {op:"tool_result", id, result} * {op:"wake_ack", id, error?} * {op:"cancel", id, reason?} * py -> host : {op:"ready"} * {op:"tool_call", id, exec_id, name, args} * {op:"wake", id, exec_id, line, payload} - * {op:"exec_result", id, stdout, stderr, return_code, tail?} + * {op:"exec_result", id, return_code} * * At a deadline the host sends cancel and then SIGINT, repeated until the * script reports (not on Windows): the cancel stops a script waiting on an @@ -49,6 +64,7 @@ export const PYTHON_RUNTIME_SOURCE: string = String.raw` import ast import asyncio +import faulthandler import inspect import io import json @@ -58,15 +74,21 @@ import sys import threading import traceback -PROTO_OUT = sys.stdout # protocol channel (real stdout) -REAL_STDERR = sys.stderr # runtime diagnostics outside execs CALL_TIMEOUT_S = 270.0 -OUTPUT_CAP_CHARS = 2_000_000 -LOG_FILE_CAP_BYTES = 50_000_000 - BACKGROUND = False -LOG_PATH = None -TAIL_CHARS = 4000 + +# The protocol's own channel, fds 3 and 4 as the host opened them (_start). +PROTO_IN = None +PROTO_OUT = None + +# Private copies of fds 0, 1 and 2 as the host set them up: /dev/null, and the +# two output pipes the host reads (_take_std_fds). Each exec puts fds 0, 1 and 2 +# back from them, whatever an earlier script did with its own, and writes its +# end marks through them. os.dup's copies are non-inheritable, so no command a +# script runs holds them. +_STD_IN = None +_STD_OUT = None +_STD_ERR = None _send_lock = threading.Lock() @@ -74,92 +96,112 @@ _send_lock = threading.Lock() def send(obj): line = json.dumps(obj, ensure_ascii=False, default=str) with _send_lock: - PROTO_OUT.write(line + "\n") - PROTO_OUT.flush() + try: + PROTO_OUT.write(line + "\n") + PROTO_OUT.flush() + except OSError as exc: + _fatal("cannot send the host a message any more (" + str(exc) + ")") + + +def _start(): + global PROTO_IN, PROTO_OUT + # A wrapper between the host and this interpreter can close fds 3 and 4 + # (a container, or sudo with its default settings, does): say so where the + # host reads, for a failed start that names what to fix, and stop. + for fd in (3, 4): + try: + os.fstat(fd) + except OSError: + try: + os.write(2, ( + "The host's channel didn't reach this Python interpreter (its file descriptors 3 and 4 " + "aren't open): a pythonPath that starts Python through a wrapper must pass fds 3 and 4 on, " + "which a container, or sudo with its default settings, doesn't.\n" + ).encode()) + except OSError: + pass + sys.exit(2) + # fds 3 and 4 came from the host inheritable; no command a script runs may + # hold them, or read the host's messages. + os.set_inheritable(3, False) + os.set_inheritable(4, False) + PROTO_IN = open(3, "r", encoding="utf-8", newline="\n") + PROTO_OUT = open(4, "w", encoding="utf-8", newline="\n") + _take_std_fds() + + +def _take_std_fds(): + global _STD_IN, _STD_OUT, _STD_ERR + _STD_IN, _STD_OUT, _STD_ERR = os.dup(0), os.dup(1), os.dup(2) + # A crash's own text goes where the host reads it, a pipe that outlives the + # process: faulthandler adds a traceback for a segfault or an abort, where + # CPython alone writes nothing. + faulthandler.enable(file=_STD_ERR, all_threads=False) + + +def _diag(text): + # A note of the runtime's own for the host's log, through fd 2's copy. + try: + os.write(_STD_ERR, ("[pytc-runtime] " + text + "\n").encode("utf-8", "backslashreplace")) + except OSError: + pass -class CappedIO(io.StringIO): - """StringIO that stops recording past a cap (keeps a truncation flag).""" +def _write_all(fd, data): + while data: + data = data[os.write(fd, data):] - def __init__(self, cap): - super().__init__() - self._cap = cap - self.truncated = False - def write(self, s): - if not isinstance(s, str): - s = str(s) - if self.tell() >= self._cap: - self.truncated = True - return len(s) - remaining = self._cap - self.tell() - if len(s) > remaining: - self.truncated = True - return super().write(s[:remaining]) - return super().write(s) +def _script_writer(fd): + # Line-buffered, as Python's own stdout is on a terminal: each line reaches + # the host as it's printed, in order with what the commands a script runs + # write, for one write per line. A partial line waits for its newline, a + # flush or the end of the exec. It writes to the fd by number, so one an + # earlier script kept, such as a logging handler's, writes into the exec + # that's running. + return io.TextIOWrapper(io.BufferedWriter(io.FileIO(fd, "w", closefd=False)), encoding="utf-8", + errors="backslashreplace", newline="\n", line_buffering=True) -class LogTee: - """Background-mode writer: line-buffered file append + rolling tail. +def _open_streams(): + # fds 0, 1 and 2 as the host set them, whatever the last script did. + os.dup2(_STD_IN, 0) + os.dup2(_STD_OUT, 1) + # A background script's journal is one stream: its stderr joins its stdout's + # pipe, so the two keep the order they were written in. + os.dup2(_STD_OUT if BACKGROUND else _STD_ERR, 2) + return _script_writer(1), _script_writer(2) - The file is the agent-readable journal (their read/grep/shell tools work - on it); the tail rides the exec_result so the host can build a crash - envelope without re-reading the file. - """ - def __init__(self, path, tail_chars): - self._fh = None - self._written = 0 - self._tail = "" - self._tail_chars = tail_chars - self.truncated = False - if path: - try: - os.makedirs(os.path.dirname(path), exist_ok=True) - self._fh = open(path, "a", encoding="utf-8", errors="replace") - except Exception as exc: - REAL_STDERR.write("[pytc-runtime] cannot open log file: " + str(exc) + "\n") - REAL_STDERR.flush() - - def write(self, s): - if not isinstance(s, str): - s = str(s) - self._tail = (self._tail + s)[-self._tail_chars:] - if self._fh is not None and not self.truncated: - if self._written + len(s) > LOG_FILE_CAP_BYTES: - self.truncated = True - try: - self._fh.write("\n[log capped at " + str(LOG_FILE_CAP_BYTES) + " bytes]\n") - self._fh.flush() - except Exception: - pass - else: - try: - self._fh.write(s) - if "\n" in s: - self._fh.flush() - self._written += len(s) - except Exception: - pass - return len(s) - - def flush(self): - if self._fh is not None: - try: - self._fh.flush() - except Exception: - pass +def _note(stream, text): + # The runtime's own line in a script's stream, which may have broken under it. + try: + stream.write(text) + except Exception: + pass - def close(self): - if self._fh is not None: - try: - self._fh.close() - except Exception: - pass - self._fh = None - def tail(self): - return self._tail +def _end_streams(streams, mark): + # Send what the script left in its writers, then end this exec's share of + # each stream with its mark: everything the script wrote, and all of every + # command it waited for, is ahead of it. + for writer in streams: + try: + writer.flush() + except Exception: + pass # its fd closed or broke under it, so that much can't be sent + data = mark.encode("utf-8") + _write_all(_STD_OUT, data) + _write_all(_STD_ERR, data) + + +def _fatal(why): + # The runtime can't go on, as when a script closed one of the runtime's own + # fds under it: say why on fd 2, which the host reads, and end, so the host + # settles the exec on the exit with the output that arrived, rather than + # waiting out its time limit for a result that can't come. + _diag(why + "; exiting") + os._exit(70) # Persistent interpreter state (mirrors container reuse: variables survive @@ -330,15 +372,11 @@ async def wake_agent(payload=None): def handle_init(msg): - global CALL_TIMEOUT_S, BACKGROUND, LOG_PATH, TAIL_CHARS + global CALL_TIMEOUT_S, BACKGROUND timeout = msg.get("call_timeout_s") if isinstance(timeout, (int, float)) and timeout > 0: CALL_TIMEOUT_S = float(timeout) BACKGROUND = bool(msg.get("background")) - LOG_PATH = msg.get("log_path") or None - tail = msg.get("tail_chars") - if isinstance(tail, int) and tail > 0: - TAIL_CHARS = tail # Remove functions injected by a previous init that are no longer present # (tool list changed) without touching user-defined globals. @@ -368,17 +406,16 @@ def handle_init(msg): SCRIPT_GLOBALS.pop("wake_agent", None) -async def _run_script(exec_id, code, st): +async def _run_script(exec_id, code, mark, st): global _current_exec_id, _current_exec_task, _stop - if BACKGROUND: - out = LogTee(LOG_PATH, TAIL_CHARS) - err = out # interleave, terminal-style; tail is shared - else: - out = CappedIO(OUTPUT_CAP_CHARS) - err = CappedIO(OUTPUT_CAP_CHARS) + try: + streams = _open_streams() + except OSError as exc: + _fatal("cannot reach the output pipes any more (" + str(exc) + ")") + out, err = streams old_out, old_err, old_in = sys.stdout, sys.stderr, sys.stdin sys.stdout, sys.stderr = out, err - sys.stdin = io.StringIO("") # input() must not steal the protocol channel + sys.stdin = io.StringIO("") # input() ends at once, as fd 0's /dev/null does for a command return_code = 0 try: compiled = compile(code, "