feat(coding-agent/eval): added local python-runner subprocess execution
- Replaced Python execution with a local `python -u runner.py` subprocess and NDJSON stdin/stdout framing. - Removed shared-gateway architecture, including coordinator lifecycle APIs, `useSharedGateway` wiring, and `jupyter` CLI/actions. - Simplified setup checks to a plain Python 3 availability probe and removed automatic dependency-install fallbacks. - Updated kernel cancellation and display processing to use status frames, SIGINT/SIGTERM escalation, and normalized output coercion. - Added `python-runner` integration and display tests while deleting legacy websocket and kernel lifecycle test suites.
This commit is contained in:
can1357
@@ -0,0 +1,39 @@
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import { renderKernelDisplay } from "@oh-my-pi/pi-coding-agent/eval/py/display";
|
||||
|
||||
describe("renderKernelDisplay (raw bundle shape)", () => {
|
||||
it("renders status events without text output", async () => {
|
||||
const { text, outputs } = await renderKernelDisplay({
|
||||
"application/x-omp-status": { op: "find", count: 12, pattern: "foo" },
|
||||
});
|
||||
expect(text).toBe("");
|
||||
expect(outputs).toEqual([{ type: "status", event: { op: "find", count: 12, pattern: "foo" } }]);
|
||||
});
|
||||
|
||||
it("prefers text/markdown over text/plain", async () => {
|
||||
const { text, outputs } = await renderKernelDisplay({
|
||||
"text/markdown": "**bold**",
|
||||
"text/plain": "bold",
|
||||
});
|
||||
expect(text).toBe("**bold**\n");
|
||||
expect(outputs).toContainEqual({ type: "markdown" });
|
||||
});
|
||||
|
||||
it("collects image/png alongside text/plain", async () => {
|
||||
const { text, outputs } = await renderKernelDisplay({
|
||||
"image/png": "base64data",
|
||||
"text/plain": "<Figure>",
|
||||
});
|
||||
expect(text).toBe("<Figure>\n");
|
||||
expect(outputs).toContainEqual({ type: "image", data: "base64data", mimeType: "image/png" });
|
||||
});
|
||||
|
||||
it("emits json bundle and includes text/plain when present", async () => {
|
||||
const { text, outputs } = await renderKernelDisplay({
|
||||
"application/json": { ok: true },
|
||||
"text/plain": "{ ok: true }",
|
||||
});
|
||||
expect(text).toBe("{ ok: true }\n");
|
||||
expect(outputs).toEqual([{ type: "json", data: { ok: true } }]);
|
||||
});
|
||||
});
|
||||
@@ -4,7 +4,6 @@ import {
|
||||
disposeKernelSessionsByOwner,
|
||||
executePython,
|
||||
} from "@oh-my-pi/pi-coding-agent/eval/py/executor";
|
||||
import * as gatewayCoordinator from "@oh-my-pi/pi-coding-agent/eval/py/gateway-coordinator";
|
||||
import type {
|
||||
KernelExecuteResult,
|
||||
KernelShutdownResult,
|
||||
@@ -320,80 +319,6 @@ describe("python executor owner cleanup", () => {
|
||||
expect(kernel.shutdown).toHaveBeenCalledWith(expect.objectContaining({ timeoutMs: expect.any(Number) }));
|
||||
expect(startSpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
it("keeps local owner-cleanup disposals counted during resource-exhaustion recovery", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const staleKernels = [new FakeKernel(), new FakeKernel(), new FakeKernel()];
|
||||
const recoveredKernel = new FakeKernel();
|
||||
const laterKernel = new FakeKernel();
|
||||
const staleShutdownDeferreds = staleKernels.map(() => Promise.withResolvers<KernelShutdownResult>());
|
||||
for (const [index, kernel] of staleKernels.entries()) {
|
||||
kernel.shutdown = vi.fn(() => staleShutdownDeferreds[index]!.promise);
|
||||
}
|
||||
const shutdownSharedGatewaySpy = vi.spyOn(gatewayCoordinator, "shutdownSharedGateway").mockResolvedValue();
|
||||
vi.spyOn(pythonKernel, "checkPythonKernelAvailability").mockResolvedValue({ ok: true });
|
||||
const startSpy = vi.spyOn(PythonKernel, "start");
|
||||
for (const kernel of staleKernels) {
|
||||
startSpy.mockResolvedValueOnce(kernel as unknown as PythonKernelInstance);
|
||||
}
|
||||
startSpy
|
||||
.mockRejectedValueOnce(new Error("EMFILE: too many open files"))
|
||||
.mockResolvedValueOnce(recoveredKernel as unknown as PythonKernelInstance)
|
||||
.mockResolvedValueOnce(laterKernel as unknown as PythonKernelInstance);
|
||||
|
||||
for (const [index] of staleKernels.entries()) {
|
||||
await executePython(`print(${index})`, {
|
||||
cwd: `/tmp/recovery-stale-${index}`,
|
||||
sessionId: `recovery-stale-session-${index}`,
|
||||
kernelMode: "session",
|
||||
kernelOwnerId: "owner-a",
|
||||
});
|
||||
}
|
||||
|
||||
const ownerCleanup = disposeKernelSessionsByOwner("owner-a");
|
||||
await Promise.resolve();
|
||||
for (const kernel of staleKernels) {
|
||||
expect(kernel.shutdown).toHaveBeenCalledWith({ timeoutMs: 2_000 });
|
||||
}
|
||||
vi.advanceTimersByTime(2_000);
|
||||
await ownerCleanup;
|
||||
|
||||
await executePython("print('recovered')", {
|
||||
cwd: "/tmp/recovery-after-emfile",
|
||||
sessionId: "recovery-session",
|
||||
kernelMode: "session",
|
||||
});
|
||||
expect(shutdownSharedGatewaySpy).toHaveBeenCalledTimes(1);
|
||||
expect(startSpy).toHaveBeenCalledTimes(5);
|
||||
expect(recoveredKernel.execute).toHaveBeenCalledTimes(1);
|
||||
|
||||
const blockedExecution = executePython("print('later')", {
|
||||
cwd: "/tmp/recovery-after-emfile-later",
|
||||
sessionId: "recovery-session-later",
|
||||
kernelMode: "session",
|
||||
});
|
||||
await flushMicrotasks();
|
||||
expect(startSpy).toHaveBeenCalledTimes(5);
|
||||
expect(recoveredKernel.shutdown).not.toHaveBeenCalled();
|
||||
expect(laterKernel.execute).not.toHaveBeenCalled();
|
||||
|
||||
staleShutdownDeferreds[0]!.resolve({ confirmed: true });
|
||||
await blockedExecution;
|
||||
expect(startSpy).toHaveBeenCalledTimes(6);
|
||||
expect(laterKernel.execute).toHaveBeenCalledTimes(1);
|
||||
|
||||
for (const deferred of staleShutdownDeferreds.slice(1)) {
|
||||
deferred.resolve({ confirmed: true });
|
||||
}
|
||||
await flushMicrotasks();
|
||||
await disposeAllKernelSessions();
|
||||
expect(recoveredKernel.shutdown).toHaveBeenCalledTimes(1);
|
||||
expect(laterKernel.shutdown).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("returns owner cleanup promptly but keeps retained capacity reserved until shutdown is confirmed", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
|
||||
@@ -1,108 +0,0 @@
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import {
|
||||
deserializeWebSocketMessage,
|
||||
type JupyterMessage,
|
||||
serializeWebSocketMessage,
|
||||
} from "@oh-my-pi/pi-coding-agent/eval/py/kernel";
|
||||
|
||||
const encoder = new TextEncoder();
|
||||
|
||||
function buildFrame(message: Omit<JupyterMessage, "buffers">, buffers: Uint8Array[] = []): ArrayBuffer {
|
||||
const msgBytes = encoder.encode(JSON.stringify(message));
|
||||
const offsetCount = 1 + buffers.length;
|
||||
const headerSize = 4 + offsetCount * 4;
|
||||
|
||||
let totalSize = headerSize + msgBytes.length;
|
||||
for (const buffer of buffers) {
|
||||
totalSize += buffer.length;
|
||||
}
|
||||
|
||||
const frame = new ArrayBuffer(totalSize);
|
||||
const view = new DataView(frame);
|
||||
const bytes = new Uint8Array(frame);
|
||||
|
||||
view.setUint32(0, offsetCount, true);
|
||||
view.setUint32(4, headerSize, true);
|
||||
bytes.set(msgBytes, headerSize);
|
||||
|
||||
let offset = headerSize + msgBytes.length;
|
||||
for (let i = 0; i < buffers.length; i++) {
|
||||
view.setUint32(4 + (i + 1) * 4, offset, true);
|
||||
bytes.set(buffers[i], offset);
|
||||
offset += buffers[i].length;
|
||||
}
|
||||
|
||||
return frame;
|
||||
}
|
||||
|
||||
describe("deserializeWebSocketMessage", () => {
|
||||
it("parses offset tables and buffers", () => {
|
||||
const message = {
|
||||
channel: "iopub",
|
||||
header: {
|
||||
msg_id: "msg-1",
|
||||
session: "session-1",
|
||||
username: "omp",
|
||||
date: "2024-01-01T00:00:00Z",
|
||||
msg_type: "stream",
|
||||
version: "5.5",
|
||||
},
|
||||
parent_header: {},
|
||||
metadata: {},
|
||||
content: { text: "hello" },
|
||||
};
|
||||
const buffer = new Uint8Array([1, 2, 3]);
|
||||
const frame = buildFrame(message, [buffer]);
|
||||
|
||||
const parsed = deserializeWebSocketMessage(frame);
|
||||
|
||||
expect(parsed).not.toBeNull();
|
||||
expect(parsed?.header.msg_id).toBe("msg-1");
|
||||
expect(parsed?.content).toEqual({ text: "hello" });
|
||||
expect(parsed?.buffers?.[0]).toEqual(buffer);
|
||||
});
|
||||
|
||||
it("returns null for invalid frames", () => {
|
||||
const headerSize = 8;
|
||||
const bytes = encoder.encode("not-json");
|
||||
const frame = new ArrayBuffer(headerSize + bytes.length);
|
||||
const view = new DataView(frame);
|
||||
const data = new Uint8Array(frame);
|
||||
view.setUint32(0, 1, true);
|
||||
view.setUint32(4, headerSize, true);
|
||||
data.set(bytes, headerSize);
|
||||
|
||||
expect(deserializeWebSocketMessage(frame)).toBeNull();
|
||||
const emptyFrame = new ArrayBuffer(4);
|
||||
new DataView(emptyFrame).setUint32(0, 0, true);
|
||||
expect(deserializeWebSocketMessage(emptyFrame)).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe("serializeWebSocketMessage", () => {
|
||||
it("round trips message payloads", () => {
|
||||
const message: JupyterMessage = {
|
||||
channel: "shell",
|
||||
header: {
|
||||
msg_id: "msg-2",
|
||||
session: "session-2",
|
||||
username: "omp",
|
||||
date: "2024-02-01T00:00:00Z",
|
||||
msg_type: "execute_request",
|
||||
version: "5.5",
|
||||
},
|
||||
parent_header: { parent: "root" },
|
||||
metadata: { tag: "meta" },
|
||||
content: { code: "print('hi')" },
|
||||
buffers: [new Uint8Array([9, 8, 7])],
|
||||
};
|
||||
|
||||
const frame = serializeWebSocketMessage(message);
|
||||
const parsed = deserializeWebSocketMessage(frame);
|
||||
|
||||
expect(parsed).not.toBeNull();
|
||||
expect(parsed?.header.msg_type).toBe("execute_request");
|
||||
expect(parsed?.content).toEqual({ code: "print('hi')" });
|
||||
expect(parsed?.buffers?.[0]).toEqual(new Uint8Array([9, 8, 7]));
|
||||
});
|
||||
});
|
||||
@@ -1,469 +0,0 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "bun:test";
|
||||
import * as gatewayCoordinator from "@oh-my-pi/pi-coding-agent/eval/py/gateway-coordinator";
|
||||
import { PythonKernel } from "@oh-my-pi/pi-coding-agent/eval/py/kernel";
|
||||
import { hookFetch, TempDir } from "@oh-my-pi/pi-utils";
|
||||
import type { Subprocess } from "bun";
|
||||
|
||||
type SpawnOptions = Bun.SpawnOptions.SpawnOptions<
|
||||
Bun.SpawnOptions.Writable,
|
||||
Bun.SpawnOptions.Readable,
|
||||
Bun.SpawnOptions.Readable
|
||||
>;
|
||||
|
||||
type FetchCall = { url: string; init?: RequestInit };
|
||||
|
||||
type FetchResponse = {
|
||||
ok: boolean;
|
||||
status: number;
|
||||
json: () => Promise<unknown>;
|
||||
text: () => Promise<string>;
|
||||
};
|
||||
|
||||
type MockEnvironment = {
|
||||
fetchCalls: FetchCall[];
|
||||
spawnCalls: { cmd: string[]; options: SpawnOptions }[];
|
||||
};
|
||||
|
||||
type MessageEventPayload = { data: ArrayBuffer };
|
||||
|
||||
type WebSocketHandler = (event: unknown) => void;
|
||||
|
||||
type WebSocketMessageHandler = (event: MessageEventPayload) => void;
|
||||
|
||||
class FakeWebSocket {
|
||||
static OPEN = 1;
|
||||
static CLOSED = 3;
|
||||
static instances: FakeWebSocket[] = [];
|
||||
|
||||
readyState = FakeWebSocket.OPEN;
|
||||
binaryType = "arraybuffer";
|
||||
url: string;
|
||||
sent: ArrayBuffer[] = [];
|
||||
|
||||
onopen: WebSocketHandler | null = null;
|
||||
onerror: WebSocketHandler | null = null;
|
||||
onclose: WebSocketHandler | null = null;
|
||||
onmessage: WebSocketMessageHandler | null = null;
|
||||
|
||||
constructor(url: string) {
|
||||
this.url = url;
|
||||
FakeWebSocket.instances.push(this);
|
||||
queueMicrotask(() => {
|
||||
this.onopen?.(undefined);
|
||||
});
|
||||
}
|
||||
|
||||
send(data: ArrayBuffer): void {
|
||||
this.sent.push(data);
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.readyState = FakeWebSocket.CLOSED;
|
||||
this.onclose?.(undefined);
|
||||
}
|
||||
}
|
||||
|
||||
const createResponse = (options: { ok: boolean; status?: number; json?: unknown; text?: string }): FetchResponse => {
|
||||
return {
|
||||
ok: options.ok,
|
||||
status: options.status ?? (options.ok ? 200 : 500),
|
||||
json: async () => options.json ?? {},
|
||||
text: async () => options.text ?? "",
|
||||
};
|
||||
};
|
||||
|
||||
const createFakeProcess = (): Subprocess => {
|
||||
const exited = new Promise<number>(() => undefined);
|
||||
return { pid: 999999, exited } as Subprocess;
|
||||
};
|
||||
|
||||
const expectResolvesWithin = async <T>(promise: Promise<T>, timeoutMs: number, message: string): Promise<T> => {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
return await Promise.race([
|
||||
promise,
|
||||
new Promise<T>((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error(message)), timeoutMs);
|
||||
timer.unref?.();
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timer !== undefined) clearTimeout(timer);
|
||||
}
|
||||
};
|
||||
|
||||
describe("PythonKernel gateway lifecycle", () => {
|
||||
const originalWebSocket = globalThis.WebSocket;
|
||||
const originalGatewayUrl = Bun.env.PI_PYTHON_GATEWAY_URL;
|
||||
const originalGatewayToken = Bun.env.PI_PYTHON_GATEWAY_TOKEN;
|
||||
const originalBunEnv = Bun.env.BUN_ENV;
|
||||
|
||||
let tempDir: TempDir;
|
||||
let env: MockEnvironment;
|
||||
|
||||
const stubKernelRuntime = () => {
|
||||
function mockSpawn(options: SpawnOptions & { cmd: string[] }): Subprocess;
|
||||
function mockSpawn(cmd: string[], options?: SpawnOptions): Subprocess;
|
||||
function mockSpawn(first: string[] | (SpawnOptions & { cmd: string[] }), second?: SpawnOptions): Subprocess {
|
||||
if (Array.isArray(first)) {
|
||||
env.spawnCalls.push({ cmd: first, options: second ?? {} });
|
||||
} else {
|
||||
const { cmd, ...options } = first;
|
||||
env.spawnCalls.push({ cmd, options });
|
||||
}
|
||||
return createFakeProcess();
|
||||
}
|
||||
|
||||
const spawnSpy = vi.spyOn(Bun, "spawn").mockImplementation(mockSpawn);
|
||||
const sleepSpy = vi.spyOn(Bun, "sleep").mockImplementation(async () => undefined);
|
||||
const whichSpy = vi.spyOn(Bun, "which").mockImplementation(() => "/usr/bin/python");
|
||||
const executeSpy = vi.spyOn(PythonKernel.prototype, "execute").mockResolvedValue({
|
||||
status: "ok",
|
||||
cancelled: false,
|
||||
timedOut: false,
|
||||
stdinRequested: false,
|
||||
});
|
||||
|
||||
return {
|
||||
[Symbol.dispose]() {
|
||||
spawnSpy.mockRestore();
|
||||
sleepSpy.mockRestore();
|
||||
whichSpy.mockRestore();
|
||||
executeSpy.mockRestore();
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
tempDir = TempDir.createSync("@omp-python-kernel-");
|
||||
env = { fetchCalls: [], spawnCalls: [] };
|
||||
|
||||
Bun.env.BUN_ENV = "test";
|
||||
delete Bun.env.PI_PYTHON_GATEWAY_URL;
|
||||
delete Bun.env.PI_PYTHON_GATEWAY_TOKEN;
|
||||
|
||||
FakeWebSocket.instances = [];
|
||||
globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
if (tempDir) {
|
||||
tempDir.removeSync();
|
||||
}
|
||||
|
||||
if (originalBunEnv === undefined) {
|
||||
delete Bun.env.BUN_ENV;
|
||||
} else {
|
||||
Bun.env.BUN_ENV = originalBunEnv;
|
||||
}
|
||||
if (originalGatewayUrl === undefined) {
|
||||
delete Bun.env.PI_PYTHON_GATEWAY_URL;
|
||||
} else {
|
||||
Bun.env.PI_PYTHON_GATEWAY_URL = originalGatewayUrl;
|
||||
}
|
||||
if (originalGatewayToken === undefined) {
|
||||
delete Bun.env.PI_PYTHON_GATEWAY_TOKEN;
|
||||
} else {
|
||||
Bun.env.PI_PYTHON_GATEWAY_TOKEN = originalGatewayToken;
|
||||
}
|
||||
|
||||
globalThis.WebSocket = originalWebSocket;
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("starts shared gateway, interrupts, and shuts down", async () => {
|
||||
using _runtime = stubKernelRuntime();
|
||||
vi.spyOn(gatewayCoordinator, "acquireSharedGateway").mockResolvedValue({
|
||||
url: "http://127.0.0.1:9999",
|
||||
isShared: true,
|
||||
});
|
||||
|
||||
using _hook = hookFetch((input, init) => {
|
||||
const url = String(input);
|
||||
env.fetchCalls.push({ url, init });
|
||||
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return createResponse({ ok: true, json: { id: "kernel-123" } }) as unknown as Response;
|
||||
}
|
||||
|
||||
return createResponse({ ok: true }) as unknown as Response;
|
||||
});
|
||||
|
||||
const kernel = await PythonKernel.start({ cwd: tempDir.path() });
|
||||
|
||||
expect(env.fetchCalls.some(call => call.url.endsWith("/api/kernels") && call.init?.method === "POST")).toBe(true);
|
||||
|
||||
await kernel.interrupt();
|
||||
expect(env.fetchCalls.some(call => call.url.includes("/interrupt") && call.init?.method === "POST")).toBe(true);
|
||||
|
||||
await kernel.shutdown();
|
||||
expect(env.fetchCalls.some(call => call.init?.method === "DELETE")).toBe(true);
|
||||
expect(kernel.isAlive()).toBe(false);
|
||||
});
|
||||
|
||||
it("aborts stalled startup after websocket connect and cleans up the kernel", async () => {
|
||||
vi.spyOn(gatewayCoordinator, "acquireSharedGateway").mockResolvedValue({
|
||||
url: "http://127.0.0.1:9999",
|
||||
isShared: true,
|
||||
});
|
||||
|
||||
let executeCallCount = 0;
|
||||
const preludeStarted = Promise.withResolvers<void>();
|
||||
vi.spyOn(PythonKernel.prototype, "execute").mockImplementation(async (_code, options) => {
|
||||
executeCallCount += 1;
|
||||
if (executeCallCount === 1) {
|
||||
return { status: "ok", cancelled: false, timedOut: false, stdinRequested: false };
|
||||
}
|
||||
preludeStarted.resolve();
|
||||
return await new Promise((_, reject) => {
|
||||
const onAbort = () => {
|
||||
const reason = options?.signal?.reason;
|
||||
reject(reason instanceof Error ? reason : new Error("Python kernel startup aborted"));
|
||||
};
|
||||
if (options?.signal?.aborted) {
|
||||
onAbort();
|
||||
return;
|
||||
}
|
||||
options?.signal?.addEventListener("abort", onAbort, { once: true });
|
||||
});
|
||||
});
|
||||
|
||||
using _hook = hookFetch((input, init) => {
|
||||
const url = String(input);
|
||||
env.fetchCalls.push({ url, init });
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return createResponse({ ok: true, json: { id: "kernel-stalled" } }) as unknown as Response;
|
||||
}
|
||||
return createResponse({ ok: true }) as unknown as Response;
|
||||
});
|
||||
|
||||
const abortController = new AbortController();
|
||||
const startPromise = PythonKernel.start({ cwd: tempDir.path(), signal: abortController.signal });
|
||||
await preludeStarted.promise;
|
||||
abortController.abort(new Error("cancel startup"));
|
||||
|
||||
const pending = Symbol("pending");
|
||||
const settled = await Promise.race([
|
||||
startPromise.then(
|
||||
() => "resolved",
|
||||
error => error,
|
||||
),
|
||||
Bun.sleep(50).then(() => pending),
|
||||
]);
|
||||
|
||||
expect(settled).toBeInstanceOf(Error);
|
||||
expect(settled).not.toBe(pending);
|
||||
expect((settled as Error).message).toContain("cancel startup");
|
||||
await expect(startPromise).rejects.toThrow("cancel startup");
|
||||
expect(
|
||||
env.fetchCalls.some(
|
||||
call => call.url.endsWith("/api/kernels/kernel-stalled") && call.init?.method === "DELETE",
|
||||
),
|
||||
).toBe(true);
|
||||
expect(FakeWebSocket.instances.at(-1)?.readyState).toBe(FakeWebSocket.CLOSED);
|
||||
});
|
||||
|
||||
it("preserves timeout classification when startup environment initialization is cancelled", async () => {
|
||||
vi.spyOn(gatewayCoordinator, "acquireSharedGateway").mockResolvedValue({
|
||||
url: "http://127.0.0.1:9999",
|
||||
isShared: true,
|
||||
});
|
||||
|
||||
vi.spyOn(PythonKernel.prototype, "execute").mockResolvedValue({
|
||||
status: "ok",
|
||||
cancelled: true,
|
||||
timedOut: true,
|
||||
stdinRequested: false,
|
||||
});
|
||||
|
||||
using _hook = hookFetch((input, init) => {
|
||||
const url = String(input);
|
||||
env.fetchCalls.push({ url, init });
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return createResponse({ ok: true, json: { id: "kernel-init-timeout" } }) as unknown as Response;
|
||||
}
|
||||
return createResponse({ ok: true }) as unknown as Response;
|
||||
});
|
||||
|
||||
await expect(PythonKernel.start({ cwd: tempDir.path() })).rejects.toMatchObject({
|
||||
name: "TimeoutError",
|
||||
message: "Failed to initialize Python kernel environment",
|
||||
});
|
||||
expect(
|
||||
env.fetchCalls.some(
|
||||
call => call.url.endsWith("/api/kernels/kernel-init-timeout") && call.init?.method === "DELETE",
|
||||
),
|
||||
).toBe(true);
|
||||
expect(FakeWebSocket.instances.at(-1)?.readyState).toBe(FakeWebSocket.CLOSED);
|
||||
});
|
||||
|
||||
it("preserves timeout classification when startup prelude execution is cancelled", async () => {
|
||||
vi.spyOn(gatewayCoordinator, "acquireSharedGateway").mockResolvedValue({
|
||||
url: "http://127.0.0.1:9999",
|
||||
isShared: true,
|
||||
});
|
||||
|
||||
let executeCallCount = 0;
|
||||
vi.spyOn(PythonKernel.prototype, "execute").mockImplementation(async () => {
|
||||
executeCallCount += 1;
|
||||
if (executeCallCount === 1) {
|
||||
return { status: "ok", cancelled: false, timedOut: false, stdinRequested: false };
|
||||
}
|
||||
return { status: "ok", cancelled: true, timedOut: true, stdinRequested: false };
|
||||
});
|
||||
|
||||
using _hook = hookFetch((input, init) => {
|
||||
const url = String(input);
|
||||
env.fetchCalls.push({ url, init });
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return createResponse({ ok: true, json: { id: "kernel-prelude-timeout" } }) as unknown as Response;
|
||||
}
|
||||
return createResponse({ ok: true }) as unknown as Response;
|
||||
});
|
||||
|
||||
await expect(PythonKernel.start({ cwd: tempDir.path() })).rejects.toMatchObject({
|
||||
name: "TimeoutError",
|
||||
message: "Failed to initialize Python kernel prelude",
|
||||
});
|
||||
expect(
|
||||
env.fetchCalls.some(
|
||||
call => call.url.endsWith("/api/kernels/kernel-prelude-timeout") && call.init?.method === "DELETE",
|
||||
),
|
||||
).toBe(true);
|
||||
expect(FakeWebSocket.instances.at(-1)?.readyState).toBe(FakeWebSocket.CLOSED);
|
||||
});
|
||||
|
||||
it("throws when shared gateway kernel creation never succeeds", async () => {
|
||||
using _runtime = stubKernelRuntime();
|
||||
vi.spyOn(gatewayCoordinator, "acquireSharedGateway").mockResolvedValue({
|
||||
url: "http://127.0.0.1:9999",
|
||||
isShared: true,
|
||||
});
|
||||
|
||||
using _hook = hookFetch((input, init) => {
|
||||
const url = String(input);
|
||||
env.fetchCalls.push({ url, init });
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return createResponse({ ok: false, status: 503, text: "oops" }) as unknown as Response;
|
||||
}
|
||||
return createResponse({ ok: true }) as unknown as Response;
|
||||
});
|
||||
|
||||
await expect(PythonKernel.start({ cwd: tempDir.path() })).rejects.toThrow(
|
||||
"Failed to create kernel on shared gateway",
|
||||
);
|
||||
});
|
||||
|
||||
it("treats initial 404 and 410 shutdown responses as confirmed", async () => {
|
||||
using _runtime = stubKernelRuntime();
|
||||
vi.spyOn(gatewayCoordinator, "acquireSharedGateway").mockResolvedValue({
|
||||
url: "http://127.0.0.1:9999",
|
||||
isShared: true,
|
||||
});
|
||||
|
||||
for (const status of [404, 410]) {
|
||||
let deleteCalls = 0;
|
||||
using _hook = hookFetch((input, init) => {
|
||||
const url = String(input);
|
||||
env.fetchCalls.push({ url, init });
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return createResponse({ ok: true, json: { id: `kernel-missing-${status}` } }) as unknown as Response;
|
||||
}
|
||||
if (url.endsWith(`/api/kernels/kernel-missing-${status}`) && init?.method === "DELETE") {
|
||||
deleteCalls += 1;
|
||||
return createResponse({ ok: false, status, text: "gone" }) as unknown as Response;
|
||||
}
|
||||
return createResponse({ ok: true }) as unknown as Response;
|
||||
});
|
||||
|
||||
const kernel = await PythonKernel.start({ cwd: tempDir.path() });
|
||||
|
||||
await expect(kernel.shutdown()).resolves.toEqual({ confirmed: true });
|
||||
expect(deleteCalls).toBe(1);
|
||||
expect(kernel.isAlive()).toBe(false);
|
||||
expect(FakeWebSocket.instances.at(-1)?.readyState).toBe(FakeWebSocket.CLOSED);
|
||||
await expect(kernel.shutdown()).resolves.toEqual({ confirmed: true });
|
||||
expect(deleteCalls).toBe(1);
|
||||
}
|
||||
});
|
||||
|
||||
it("returns unconfirmed when shutdown times out and can confirm on retry", async () => {
|
||||
using _runtime = stubKernelRuntime();
|
||||
vi.spyOn(gatewayCoordinator, "acquireSharedGateway").mockResolvedValue({
|
||||
url: "http://127.0.0.1:9999",
|
||||
isShared: true,
|
||||
});
|
||||
let deleteCalls = 0;
|
||||
const firstDeleteStarted = Promise.withResolvers<void>();
|
||||
const firstDeleteAborted = Promise.withResolvers<void>();
|
||||
using _hook = hookFetch((input, init) => {
|
||||
const url = String(input);
|
||||
env.fetchCalls.push({ url, init });
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return createResponse({ ok: true, json: { id: "kernel-shutdown-timeout" } }) as unknown as Response;
|
||||
}
|
||||
if (url.endsWith("/api/kernels/kernel-shutdown-timeout") && init?.method === "DELETE") {
|
||||
deleteCalls += 1;
|
||||
if (deleteCalls === 1) {
|
||||
firstDeleteStarted.resolve();
|
||||
return new Promise<Response>((_, reject) => {
|
||||
const abortSignal = init.signal;
|
||||
if (!abortSignal) return;
|
||||
const rejectOnAbort = () => {
|
||||
firstDeleteAborted.resolve();
|
||||
const reason = abortSignal.reason;
|
||||
reject(reason instanceof Error ? reason : new Error("Python kernel shutdown timed out"));
|
||||
};
|
||||
if (abortSignal.aborted) {
|
||||
rejectOnAbort();
|
||||
return;
|
||||
}
|
||||
abortSignal.addEventListener("abort", rejectOnAbort, { once: true });
|
||||
});
|
||||
}
|
||||
return createResponse({ ok: false, status: 404, text: "gone" }) as unknown as Response;
|
||||
}
|
||||
return createResponse({ ok: true }) as unknown as Response;
|
||||
});
|
||||
const kernel = await PythonKernel.start({ cwd: tempDir.path() });
|
||||
const shutdownPromise = kernel.shutdown({ timeoutMs: 25 });
|
||||
await expectResolvesWithin(firstDeleteStarted.promise, 250, "kernel shutdown never issued a delete request");
|
||||
await expectResolvesWithin(
|
||||
firstDeleteAborted.promise,
|
||||
500,
|
||||
"timed out waiting for the first delete request to abort",
|
||||
);
|
||||
await expect(
|
||||
expectResolvesWithin(shutdownPromise, 500, "kernel shutdown did not settle after timing out"),
|
||||
).resolves.toEqual({
|
||||
confirmed: false,
|
||||
});
|
||||
expect(kernel.isAlive()).toBe(false);
|
||||
expect(FakeWebSocket.instances.at(-1)?.readyState).toBe(FakeWebSocket.CLOSED);
|
||||
await expect(kernel.shutdown()).resolves.toEqual({ confirmed: true });
|
||||
expect(deleteCalls).toBe(2);
|
||||
});
|
||||
it("does not throw when shutdown API fails", async () => {
|
||||
using _runtime = stubKernelRuntime();
|
||||
vi.spyOn(gatewayCoordinator, "acquireSharedGateway").mockResolvedValue({
|
||||
url: "http://127.0.0.1:9999",
|
||||
isShared: true,
|
||||
});
|
||||
|
||||
using _hook = hookFetch((input, init) => {
|
||||
const url = String(input);
|
||||
env.fetchCalls.push({ url, init });
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return createResponse({ ok: true, json: { id: "kernel-456" } }) as unknown as Response;
|
||||
}
|
||||
if (init?.method === "DELETE") {
|
||||
throw new Error("delete failed");
|
||||
}
|
||||
return createResponse({ ok: true }) as unknown as Response;
|
||||
});
|
||||
|
||||
const kernel = await PythonKernel.start({ cwd: tempDir.path() });
|
||||
|
||||
await expect(kernel.shutdown()).resolves.toEqual({ confirmed: false });
|
||||
});
|
||||
});
|
||||
@@ -1,377 +0,0 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "bun:test";
|
||||
import { type KernelDisplayOutput, PythonKernel } from "@oh-my-pi/pi-coding-agent/eval/py/kernel";
|
||||
import { PYTHON_PRELUDE } from "@oh-my-pi/pi-coding-agent/eval/py/prelude";
|
||||
import { hookFetch } from "@oh-my-pi/pi-utils";
|
||||
|
||||
type JupyterMessage = {
|
||||
channel: string;
|
||||
header: {
|
||||
msg_id: string;
|
||||
session: string;
|
||||
username: string;
|
||||
date: string;
|
||||
msg_type: string;
|
||||
version: string;
|
||||
};
|
||||
parent_header: Record<string, unknown>;
|
||||
metadata: Record<string, unknown>;
|
||||
content: Record<string, unknown>;
|
||||
buffers?: Uint8Array[];
|
||||
};
|
||||
|
||||
const textEncoder = new TextEncoder();
|
||||
const textDecoder = new TextDecoder();
|
||||
|
||||
function encodeMessage(msg: JupyterMessage): ArrayBuffer {
|
||||
const msgText = JSON.stringify({
|
||||
channel: msg.channel,
|
||||
header: msg.header,
|
||||
parent_header: msg.parent_header,
|
||||
metadata: msg.metadata,
|
||||
content: msg.content,
|
||||
});
|
||||
const msgBytes = textEncoder.encode(msgText);
|
||||
const buffers = msg.buffers ?? [];
|
||||
const offsetCount = 1 + buffers.length;
|
||||
const headerSize = 4 + offsetCount * 4;
|
||||
let totalSize = headerSize + msgBytes.length;
|
||||
for (const buffer of buffers) {
|
||||
totalSize += buffer.length;
|
||||
}
|
||||
const result = new ArrayBuffer(totalSize);
|
||||
const view = new DataView(result);
|
||||
const bytes = new Uint8Array(result);
|
||||
view.setUint32(0, offsetCount, true);
|
||||
let offset = headerSize;
|
||||
view.setUint32(4, offset, true);
|
||||
bytes.set(msgBytes, offset);
|
||||
offset += msgBytes.length;
|
||||
buffers.forEach((buffer, index) => {
|
||||
view.setUint32(4 + (index + 1) * 4, offset, true);
|
||||
bytes.set(buffer, offset);
|
||||
offset += buffer.length;
|
||||
});
|
||||
return result;
|
||||
}
|
||||
|
||||
function decodeMessage(data: ArrayBuffer): JupyterMessage {
|
||||
const view = new DataView(data);
|
||||
const offsetCount = view.getUint32(0, true);
|
||||
const offsets: number[] = [];
|
||||
for (let i = 0; i < offsetCount; i++) {
|
||||
offsets.push(view.getUint32(4 + i * 4, true));
|
||||
}
|
||||
const msgStart = offsets[0];
|
||||
const msgEnd = offsets.length > 1 ? offsets[1] : data.byteLength;
|
||||
const msgBytes = new Uint8Array(data, msgStart, msgEnd - msgStart);
|
||||
const msgText = textDecoder.decode(msgBytes);
|
||||
return JSON.parse(msgText) as JupyterMessage;
|
||||
}
|
||||
|
||||
function sendOkExecution(ws: FakeWebSocket, msgId: string, executionCount = 1) {
|
||||
const reply: JupyterMessage = {
|
||||
channel: "shell",
|
||||
header: {
|
||||
msg_id: `reply-${msgId}`,
|
||||
session: "session",
|
||||
username: "omp",
|
||||
date: new Date().toISOString(),
|
||||
msg_type: "execute_reply",
|
||||
version: "5.5",
|
||||
},
|
||||
parent_header: { msg_id: msgId },
|
||||
metadata: {},
|
||||
content: { status: "ok", execution_count: executionCount },
|
||||
};
|
||||
const status: JupyterMessage = {
|
||||
channel: "iopub",
|
||||
header: {
|
||||
msg_id: `status-${msgId}`,
|
||||
session: "session",
|
||||
username: "omp",
|
||||
date: new Date().toISOString(),
|
||||
msg_type: "status",
|
||||
version: "5.5",
|
||||
},
|
||||
parent_header: { msg_id: msgId },
|
||||
metadata: {},
|
||||
content: { execution_state: "idle" },
|
||||
};
|
||||
ws.onmessage?.({ data: encodeMessage(reply) });
|
||||
ws.onmessage?.({ data: encodeMessage(status) });
|
||||
}
|
||||
|
||||
class FakeWebSocket {
|
||||
static OPEN = 1;
|
||||
static CLOSED = 3;
|
||||
static lastInstance: FakeWebSocket | null = null;
|
||||
readyState = FakeWebSocket.OPEN;
|
||||
binaryType = "arraybuffer";
|
||||
onopen?: () => void;
|
||||
onmessage?: (event: { data: ArrayBuffer }) => void;
|
||||
onerror?: (event: unknown) => void;
|
||||
onclose?: () => void;
|
||||
readonly url: string;
|
||||
readonly sent: (ArrayBuffer | string)[] = [];
|
||||
private handleSend: ((data: ArrayBuffer | string) => void) | null = null;
|
||||
private pendingMessages: (ArrayBuffer | string)[] = [];
|
||||
|
||||
constructor(url: string) {
|
||||
this.url = url;
|
||||
FakeWebSocket.lastInstance = this;
|
||||
queueMicrotask(() => this.onopen?.());
|
||||
}
|
||||
|
||||
setSendHandler(handler: (data: ArrayBuffer | string) => void) {
|
||||
this.handleSend = handler;
|
||||
for (const msg of this.pendingMessages) {
|
||||
handler(msg);
|
||||
}
|
||||
this.pendingMessages = [];
|
||||
}
|
||||
|
||||
send(data: ArrayBuffer | string) {
|
||||
this.sent.push(data);
|
||||
if (this.handleSend) {
|
||||
this.handleSend(data);
|
||||
} else {
|
||||
this.pendingMessages.push(data);
|
||||
}
|
||||
}
|
||||
|
||||
close() {
|
||||
this.readyState = FakeWebSocket.CLOSED;
|
||||
this.onclose?.();
|
||||
}
|
||||
}
|
||||
|
||||
describe("PythonKernel (external gateway)", () => {
|
||||
const originalEnv = { ...Bun.env };
|
||||
const originalWebSocket = globalThis.WebSocket;
|
||||
|
||||
beforeEach(() => {
|
||||
Bun.env.PI_PYTHON_GATEWAY_URL = "http://gateway.test";
|
||||
Bun.env.PI_PYTHON_SKIP_CHECK = "1";
|
||||
globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
for (const key of Object.keys(Bun.env)) {
|
||||
if (!(key in originalEnv)) {
|
||||
delete Bun.env[key];
|
||||
}
|
||||
}
|
||||
for (const [key, value] of Object.entries(originalEnv)) {
|
||||
Bun.env[key] = value;
|
||||
}
|
||||
globalThis.WebSocket = originalWebSocket;
|
||||
FakeWebSocket.lastInstance = null;
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("executes code via websocket stream and display data", async () => {
|
||||
const fetchMock = vi.fn(async (url: string, init?: RequestInit) => {
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return new Response(JSON.stringify({ id: "kernel-1" }), { status: 201 });
|
||||
}
|
||||
if (url.includes("/api/kernels/") && init?.method === "DELETE") {
|
||||
return new Response("", { status: 204 });
|
||||
}
|
||||
return new Response("", { status: 200 });
|
||||
});
|
||||
using _hook = hookFetch((input, init) => fetchMock(String(input), init));
|
||||
|
||||
let initSeen = false;
|
||||
let preludeSeen = false;
|
||||
|
||||
const kernelPromise = PythonKernel.start({ cwd: "/" });
|
||||
await Bun.sleep(10);
|
||||
const ws = FakeWebSocket.lastInstance;
|
||||
if (!ws) throw new Error("WebSocket not initialized");
|
||||
ws.setSendHandler(data => {
|
||||
const msg = typeof data === "string" ? (JSON.parse(data) as JupyterMessage) : decodeMessage(data);
|
||||
const code = String(msg.content.code ?? "");
|
||||
if (!initSeen) {
|
||||
// First execution is kernel environment init
|
||||
initSeen = true;
|
||||
sendOkExecution(ws, msg.header.msg_id);
|
||||
return;
|
||||
}
|
||||
if (!preludeSeen) {
|
||||
expect(code).toBe(PYTHON_PRELUDE);
|
||||
preludeSeen = true;
|
||||
sendOkExecution(ws, msg.header.msg_id);
|
||||
return;
|
||||
}
|
||||
|
||||
if (code === "print('hello')") {
|
||||
const stream: JupyterMessage = {
|
||||
channel: "iopub",
|
||||
header: {
|
||||
msg_id: "stream-1",
|
||||
session: "session",
|
||||
username: "omp",
|
||||
date: new Date().toISOString(),
|
||||
msg_type: "stream",
|
||||
version: "5.5",
|
||||
},
|
||||
parent_header: { msg_id: msg.header.msg_id },
|
||||
metadata: {},
|
||||
content: { text: "hello\n" },
|
||||
};
|
||||
const display: JupyterMessage = {
|
||||
channel: "iopub",
|
||||
header: {
|
||||
msg_id: "display-1",
|
||||
session: "session",
|
||||
username: "omp",
|
||||
date: new Date().toISOString(),
|
||||
msg_type: "execute_result",
|
||||
version: "5.5",
|
||||
},
|
||||
parent_header: { msg_id: msg.header.msg_id },
|
||||
metadata: {},
|
||||
content: {
|
||||
data: {
|
||||
"text/plain": "result",
|
||||
"application/json": { answer: 42 },
|
||||
},
|
||||
},
|
||||
};
|
||||
const reply: JupyterMessage = {
|
||||
channel: "shell",
|
||||
header: {
|
||||
msg_id: "reply-2",
|
||||
session: "session",
|
||||
username: "omp",
|
||||
date: new Date().toISOString(),
|
||||
msg_type: "execute_reply",
|
||||
version: "5.5",
|
||||
},
|
||||
parent_header: { msg_id: msg.header.msg_id },
|
||||
metadata: {},
|
||||
content: { status: "ok", execution_count: 2 },
|
||||
};
|
||||
const status: JupyterMessage = {
|
||||
channel: "iopub",
|
||||
header: {
|
||||
msg_id: "status-2",
|
||||
session: "session",
|
||||
username: "omp",
|
||||
date: new Date().toISOString(),
|
||||
msg_type: "status",
|
||||
version: "5.5",
|
||||
},
|
||||
parent_header: { msg_id: msg.header.msg_id },
|
||||
metadata: {},
|
||||
content: { execution_state: "idle" },
|
||||
};
|
||||
ws.onmessage?.({ data: encodeMessage(stream) });
|
||||
ws.onmessage?.({ data: encodeMessage(display) });
|
||||
ws.onmessage?.({ data: encodeMessage(reply) });
|
||||
ws.onmessage?.({ data: encodeMessage(status) });
|
||||
return;
|
||||
}
|
||||
|
||||
sendOkExecution(ws, msg.header.msg_id);
|
||||
});
|
||||
|
||||
const kernel = await kernelPromise;
|
||||
const chunks: string[] = [];
|
||||
const displays: KernelDisplayOutput[] = [];
|
||||
|
||||
const result = await kernel.execute("print('hello')", {
|
||||
onChunk: text => {
|
||||
chunks.push(text);
|
||||
},
|
||||
onDisplay: output => {
|
||||
displays.push(output);
|
||||
},
|
||||
});
|
||||
|
||||
expect(result.status).toBe("ok");
|
||||
expect(chunks.join("")).toContain("hello");
|
||||
expect(chunks.join("")).toContain("result");
|
||||
expect(displays).toEqual([{ type: "json", data: { answer: 42 } }]);
|
||||
|
||||
const shutdown = await kernel.shutdown();
|
||||
expect(shutdown).toEqual({ confirmed: true });
|
||||
expect(fetchMock).toHaveBeenCalledWith("http://gateway.test/api/kernels/kernel-1", {
|
||||
method: "DELETE",
|
||||
headers: {},
|
||||
});
|
||||
});
|
||||
|
||||
it("returns an unconfirmed shutdown result when kernel deletion is not acknowledged", async () => {
|
||||
let deleteAttempts = 0;
|
||||
const fetchMock = vi.fn(async (url: string, init?: RequestInit) => {
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return new Response(JSON.stringify({ id: "kernel-delete-failure" }), { status: 201 });
|
||||
}
|
||||
if (url.includes("/api/kernels/") && init?.method === "DELETE") {
|
||||
deleteAttempts += 1;
|
||||
if (deleteAttempts === 1) {
|
||||
return new Response("delete failed", { status: 500, statusText: "Server Error" });
|
||||
}
|
||||
return new Response("already gone", { status: 404, statusText: "Not Found" });
|
||||
}
|
||||
return new Response("", { status: 200 });
|
||||
});
|
||||
using _hook = hookFetch((input, init) => fetchMock(String(input), init));
|
||||
|
||||
const kernelPromise = PythonKernel.start({ cwd: "/" });
|
||||
await Bun.sleep(10);
|
||||
const ws = FakeWebSocket.lastInstance;
|
||||
if (!ws) throw new Error("WebSocket not initialized");
|
||||
ws.setSendHandler(data => {
|
||||
const msg = typeof data === "string" ? (JSON.parse(data) as JupyterMessage) : decodeMessage(data);
|
||||
sendOkExecution(ws, msg.header.msg_id);
|
||||
});
|
||||
|
||||
const kernel = await kernelPromise;
|
||||
await expect(kernel.shutdown()).resolves.toEqual({ confirmed: false });
|
||||
expect(deleteAttempts).toBe(1);
|
||||
await expect(kernel.shutdown()).resolves.toEqual({ confirmed: true });
|
||||
expect(deleteAttempts).toBe(2);
|
||||
await expect(kernel.shutdown()).resolves.toEqual({ confirmed: true });
|
||||
expect(deleteAttempts).toBe(2);
|
||||
});
|
||||
|
||||
it("initializes the IPython prelude", async () => {
|
||||
const fetchMock = vi.fn(async (url: string, init?: RequestInit) => {
|
||||
if (url.endsWith("/api/kernels") && init?.method === "POST") {
|
||||
return new Response(JSON.stringify({ id: "kernel-3" }), { status: 201 });
|
||||
}
|
||||
return new Response("", { status: 200 });
|
||||
});
|
||||
using _hook = hookFetch((input, init) => fetchMock(String(input), init));
|
||||
|
||||
let initSeen = false;
|
||||
let preludeSeen = false;
|
||||
|
||||
const kernelPromise = PythonKernel.start({ cwd: "/" });
|
||||
await Bun.sleep(10);
|
||||
const ws = FakeWebSocket.lastInstance;
|
||||
if (!ws) throw new Error("WebSocket not initialized");
|
||||
ws.setSendHandler(data => {
|
||||
const msg = typeof data === "string" ? (JSON.parse(data) as JupyterMessage) : decodeMessage(data);
|
||||
const code = String(msg.content.code ?? "");
|
||||
if (!initSeen) {
|
||||
initSeen = true;
|
||||
sendOkExecution(ws, msg.header.msg_id);
|
||||
return;
|
||||
}
|
||||
if (!preludeSeen) {
|
||||
expect(code).toBe(PYTHON_PRELUDE);
|
||||
preludeSeen = true;
|
||||
}
|
||||
sendOkExecution(ws, msg.header.msg_id);
|
||||
});
|
||||
|
||||
const kernel = await kernelPromise;
|
||||
expect(kernel.isAlive()).toBe(true);
|
||||
await kernel.shutdown();
|
||||
});
|
||||
});
|
||||
|
||||
// TODO: add coverage for gateway process exit handling once PythonKernel exposes a test hook.
|
||||
@@ -0,0 +1,106 @@
|
||||
/**
|
||||
* End-to-end exercise of the new subprocess-backed Python runner.
|
||||
*
|
||||
* Gated by `PI_PYTHON_INTEGRATION=1` so CI without a real Python interpreter
|
||||
* (or sandboxes where subprocess spawning is restricted) does not fail.
|
||||
*/
|
||||
import { afterEach, describe, expect, it } from "bun:test";
|
||||
import * as path from "node:path";
|
||||
import { disposeAllKernelSessions, executePythonWithKernel } from "@oh-my-pi/pi-coding-agent/eval/py/executor";
|
||||
import { PythonKernel } from "@oh-my-pi/pi-coding-agent/eval/py/kernel";
|
||||
import { TempDir } from "@oh-my-pi/pi-utils";
|
||||
|
||||
const SHOULD_RUN = Bun.env.PI_PYTHON_INTEGRATION === "1";
|
||||
|
||||
describe.skipIf(!SHOULD_RUN)("python runner subprocess", () => {
|
||||
afterEach(async () => {
|
||||
await disposeAllKernelSessions();
|
||||
});
|
||||
|
||||
it("streams stdout chunks as they are produced", async () => {
|
||||
using tempDir = TempDir.createSync("@python-runner-stream-");
|
||||
const kernel = await PythonKernel.start({ cwd: tempDir.path() });
|
||||
try {
|
||||
const chunks: string[] = [];
|
||||
const result = await executePythonWithKernel(
|
||||
kernel,
|
||||
["import sys", "for i in range(5):", " print(i, flush=True)"].join("\n"),
|
||||
{
|
||||
onChunk: chunk => {
|
||||
chunks.push(chunk);
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(result.exitCode).toBe(0);
|
||||
// 5 lines * (digit + newline) → at least 5 distinct chunks once printed.
|
||||
const text = chunks.join("");
|
||||
expect(text).toContain("0\n");
|
||||
expect(text).toContain("4\n");
|
||||
} finally {
|
||||
await kernel.shutdown();
|
||||
}
|
||||
});
|
||||
|
||||
it("cancels a long sleep via SIGINT within 500ms", async () => {
|
||||
using tempDir = TempDir.createSync("@python-runner-cancel-");
|
||||
const kernel = await PythonKernel.start({ cwd: tempDir.path() });
|
||||
try {
|
||||
const start = Date.now();
|
||||
const ac = new AbortController();
|
||||
const pending = executePythonWithKernel(kernel, "import time\ntime.sleep(30)", {
|
||||
signal: ac.signal,
|
||||
});
|
||||
setTimeout(() => ac.abort(new DOMException("user cancelled", "AbortError")), 50);
|
||||
const result = await pending;
|
||||
const elapsed = Date.now() - start;
|
||||
expect(result.cancelled).toBe(true);
|
||||
expect(elapsed).toBeLessThan(2_000);
|
||||
// Kernel must survive cancellation and remain usable.
|
||||
const next = await executePythonWithKernel(kernel, "print('alive')");
|
||||
expect(next.exitCode).toBe(0);
|
||||
expect(next.output).toContain("alive");
|
||||
} finally {
|
||||
await kernel.shutdown();
|
||||
}
|
||||
});
|
||||
|
||||
it("preserves user namespace across calls", async () => {
|
||||
using tempDir = TempDir.createSync("@python-runner-session-");
|
||||
const kernel = await PythonKernel.start({ cwd: tempDir.path() });
|
||||
try {
|
||||
await executePythonWithKernel(kernel, "x = 41");
|
||||
const result = await executePythonWithKernel(kernel, "x + 1");
|
||||
expect(result.exitCode).toBe(0);
|
||||
expect(result.output).toContain("42");
|
||||
} finally {
|
||||
await kernel.shutdown();
|
||||
}
|
||||
});
|
||||
|
||||
it("emits an error frame when user code raises", async () => {
|
||||
using tempDir = TempDir.createSync("@python-runner-error-");
|
||||
const kernel = await PythonKernel.start({ cwd: tempDir.path() });
|
||||
try {
|
||||
const result = await executePythonWithKernel(kernel, "raise ValueError('boom')");
|
||||
expect(result.exitCode).toBe(1);
|
||||
expect(result.output).toContain("ValueError");
|
||||
expect(result.output).toContain("boom");
|
||||
} finally {
|
||||
await kernel.shutdown();
|
||||
}
|
||||
});
|
||||
|
||||
it("translates %pwd magic to the user namespace", async () => {
|
||||
using tempDir = TempDir.createSync("@python-runner-magic-");
|
||||
const kernel = await PythonKernel.start({ cwd: tempDir.path() });
|
||||
try {
|
||||
const result = await executePythonWithKernel(kernel, "%pwd");
|
||||
expect(result.exitCode).toBe(0);
|
||||
// %pwd returns the cwd string, which becomes the last-expression result.
|
||||
// On macOS, the OS may resolve /var to /private/var, so check by basename.
|
||||
expect(result.output).toContain(path.basename(tempDir.path()));
|
||||
} finally {
|
||||
await kernel.shutdown();
|
||||
}
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user