diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 2685559d7..3fa2c2d52 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -12,6 +12,9 @@ - Fixed high memory usage in multi-target ast_grep searches by only retaining the final page window while preserving exact match and file counts. - Fixed async job manager disposal and task cancellation handling to properly release session semaphores when aborted. - Updated and expanded omp:// documentation coverage for managed memory, skill tools, image/speech generation, and package CLIs. +### Fixed + +- Fixed Python eval shell helpers buffering child-process output until exit or newline by streaming fixed-size chunks from `!cmd`, `%%bash`, and `%pip`. ([#3950](https://github.com/can1357/oh-my-pi/issues/3950)) ## [16.2.10] - 2026-06-30 diff --git a/packages/coding-agent/src/eval/py/__tests__/runner-shell-output.test.ts b/packages/coding-agent/src/eval/py/__tests__/runner-shell-output.test.ts new file mode 100644 index 000000000..1c63e63b1 --- /dev/null +++ b/packages/coding-agent/src/eval/py/__tests__/runner-shell-output.test.ts @@ -0,0 +1,121 @@ +import { describe, expect, it } from "bun:test"; +import * as path from "node:path"; + +interface RunnerFrame { + type?: string; + id?: string; + data?: string; + status?: string; +} + +const pythonPath = Bun.env.PYTHON ?? "python3"; +const runnerPath = path.resolve(import.meta.dir, "..", "runner.py"); +const repoRoot = path.resolve(import.meta.dir, "../../../../../.."); +const encoder = new TextEncoder(); + +function shellQuote(value: string): string { + return `'${value.replaceAll("'", `'"'"'`)}'`; +} + +async function runCell(code: string): Promise { + const proc = Bun.spawn([pythonPath, "-u", runnerPath], { + cwd: repoRoot, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + env: { + ...process.env, + PYTHONUNBUFFERED: "1", + PYTHONIOENCODING: "utf-8", + }, + }); + const stderr = new Response(proc.stderr).text(); + const reader = proc.stdout.getReader(); + const decoder = new TextDecoder(); + let pending = ""; + const frames: RunnerFrame[] = []; + + async function readFrame(): Promise { + while (true) { + const newline = pending.indexOf("\n"); + if (newline >= 0) { + const line = pending.slice(0, newline); + pending = pending.slice(newline + 1); + return JSON.parse(line) as RunnerFrame; + } + const { value, done } = await reader.read(); + if (done) { + throw new Error(`Python runner exited before done frame: ${await stderr}`); + } + pending += decoder.decode(value, { stream: true }); + } + } + + try { + proc.stdin.write(encoder.encode(`${JSON.stringify({ id: "r1", code })}\n`)); + proc.stdin.flush(); + while (true) { + const frame = await readFrame(); + frames.push(frame); + if (frame.type === "done") break; + } + proc.stdin.write(encoder.encode(`${JSON.stringify({ type: "exit" })}\n`)); + proc.stdin.end(); + const exitCode = await proc.exited; + if (exitCode !== 0) { + throw new Error(`Python runner exited ${exitCode}: ${await stderr}`); + } + return frames; + } finally { + try { + reader.releaseLock(); + } catch { + // Reader may already be released by stream closure. + } + try { + proc.kill("SIGKILL"); + } catch { + // Process already exited. + } + } +} + +describe("Python runner shell output streaming", () => { + it("streams !cmd output chunks before the child process exits", async () => { + const child = [ + "import sys,time", + "sys.stdout.write('first\\n')", + "sys.stdout.flush()", + "time.sleep(0.2)", + "sys.stdout.write('second\\n')", + "sys.stdout.flush()", + ].join(";"); + const frames = await runCell( + [ + `result = !${pythonPath} -c ${shellQuote(child)}`, + "print('return=' + str(result.returncode) + ' lines=' + repr(list(result)))", + ].join("\n"), + ); + const stdout = frames.filter(frame => frame.type === "stdout").map(frame => frame.data); + + expect(stdout[0]).toBe("first\n"); + expect(stdout.join("")).toContain("second\n"); + expect(stdout.join("")).toContain("return=0 lines=['first', 'second']"); + }); + + it("streams newline-free %%bash output without waiting for EOF", async () => { + const child = [ + "import sys,time", + "sys.stdout.write('first')", + "sys.stdout.flush()", + "time.sleep(0.2)", + "sys.stdout.write('second')", + "sys.stdout.flush()", + ].join(";"); + const frames = await runCell(`%%bash\n${pythonPath} -c ${shellQuote(child)}`); + const stdout = frames.filter(frame => frame.type === "stdout").map(frame => frame.data); + + expect(stdout[0]).toBe("first"); + expect(stdout.join("")).toBe("firstsecond"); + }); +}); diff --git a/packages/coding-agent/src/eval/py/runner.py b/packages/coding-agent/src/eval/py/runner.py index dbb40e820..48a347a31 100644 --- a/packages/coding-agent/src/eval/py/runner.py +++ b/packages/coding-agent/src/eval/py/runner.py @@ -26,14 +26,16 @@ when installed. from __future__ import annotations -import asyncio import ast -import contextvars +import asyncio import base64 import builtins +import codecs +import contextvars import inspect import io import json +import locale import os import re import runpy @@ -45,7 +47,7 @@ import threading import time import traceback from pathlib import Path -from typing import Any +from typing import Any, Callable # --------------------------------------------------------------------------- # Frame writer @@ -396,6 +398,78 @@ def _emit_status(op: str, **data: Any) -> None: return _emit({"type": "display", "id": rid, "bundle": bundle}) +_SHELL_READ_CHUNK_BYTES = 8192 +_SHELL_RESULT_CAPTURE_CHARS = 1024 * 1024 +_PIP_LINE_SCAN_CHARS = 64 * 1024 + + +def _process_output_decoder() -> codecs.IncrementalDecoder: + encoding = locale.getpreferredencoding(False) or "utf-8" + return codecs.getincrementaldecoder(encoding)(errors="strict") + + +def _stream_process_output(proc: subprocess.Popen, on_text: Callable[[str], None] | None = None) -> None: + assert proc.stdout is not None + decoder = _process_output_decoder() + while True: + chunk = os.read(proc.stdout.fileno(), _SHELL_READ_CHUNK_BYTES) + if not chunk: + break + text = decoder.decode(chunk) + if text: + sys.stdout.write(text) + sys.stdout.flush() + if on_text is not None: + on_text(text) + tail = decoder.decode(b"", final=True) + if tail: + sys.stdout.write(tail) + sys.stdout.flush() + if on_text is not None: + on_text(tail) + + +class _BoundedTextCapture: + def __init__(self, max_chars: int) -> None: + self._remaining = max_chars + self._parts: list[str] = [] + + def add(self, text: str) -> None: + if self._remaining <= 0: + return + part = text[: self._remaining] + self._parts.append(part) + self._remaining -= len(part) + + def text(self) -> str: + return "".join(self._parts) + + +class _BoundedLineScanner: + def __init__(self, max_chars: int, on_line: Callable[[str], None]) -> None: + self._max_chars = max_chars + self._on_line = on_line + self._partial = "" + + def add(self, text: str) -> None: + data = self._partial + text + lines = data.splitlines(keepends=True) + if not lines: + return + if lines[-1].endswith(("\n", "\r")): + self._partial = "" + else: + self._partial = lines.pop() + for line in lines: + self._on_line(line) + if len(self._partial) > self._max_chars: + self._partial = self._partial[-self._max_chars :] + + def finish(self) -> None: + if self._partial: + self._on_line(self._partial) + self._partial = "" + @line_magic("pip") def _magic_pip(args: str) -> None: @@ -405,19 +479,20 @@ def _magic_pip(args: str) -> None: cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, - bufsize=1, ) installed_packages: list[str] = [] - assert proc.stdout is not None - for raw_line in proc.stdout: - sys.stdout.write(raw_line) + + def scan_pip_line(raw_line: str) -> None: m = re.search(r"Successfully installed\s+(.+)$", raw_line) if m: for token in m.group(1).split(): # Token is name-version; drop the version suffix. pkg = token.rsplit("-", 1)[0] installed_packages.append(pkg.replace("_", "-")) + + scanner = _BoundedLineScanner(_PIP_LINE_SCAN_CHARS, scan_pip_line) + _stream_process_output(proc, scanner.add) + scanner.finish() proc.wait() if installed_packages: import importlib @@ -599,12 +674,8 @@ def _run_shell_body(body: str, *, shell_arg: str) -> int: [shell_arg, "-c", body], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, - bufsize=1, ) - assert proc.stdout is not None - for raw_line in proc.stdout: - sys.stdout.write(raw_line) + _stream_process_output(proc) proc.wait() return proc.returncode @@ -640,16 +711,16 @@ class _ShellResult(list): def __omp_shell(cmd: str) -> _ShellResult: - proc = subprocess.run( + proc = subprocess.Popen( cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, ) - if proc.stdout: - sys.stdout.write(proc.stdout) - lines = [line for line in (proc.stdout or "").splitlines()] + capture = _BoundedTextCapture(_SHELL_RESULT_CAPTURE_CHARS) + _stream_process_output(proc, capture.add) + proc.wait() + lines = [line for line in capture.text().splitlines()] return _ShellResult(lines, proc.returncode)