Merge main into model-registry

2026-06-22 14:00:18 +02:00
parent 019b022068 717a8f958a
commit abbd911693
220 changed files with 10488 additions and 4354 deletions
@@ -1,6 +1,7 @@
 import { type AssistantMessage, type AssistantMessageEvent, EventStream, getModel } from "@earendil-works/pi-ai/compat";
+import { Type } from "typebox";
 import { describe, expect, it } from "vitest";
-import { Agent } from "../src/index.ts";
+import { Agent, type AgentEvent, type AgentTool, type AgentToolUpdateCallback } from "../src/index.ts";

 // Mock stream that mimics AssistantMessageEventStream
 class MockAssistantStream extends EventStream<AssistantMessageEvent, AssistantMessage> {
@@ -36,6 +37,28 @@ function createAssistantMessage(text: string): AssistantMessage {
 	};
 }

+type ToolCallContent = Extract<AssistantMessage["content"][number], { type: "toolCall" }>;
+
+function createAssistantToolUseMessage(content: ToolCallContent[]): AssistantMessage {
+	return {
+		role: "assistant",
+		content,
+		api: "openai-responses",
+		provider: "openai",
+		model: "mock",
+		usage: {
+			input: 0,
+			output: 0,
+			cacheRead: 0,
+			cacheWrite: 0,
+			totalTokens: 0,
+			cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+		},
+		stopReason: "toolUse",
+		timestamp: Date.now(),
+	};
+}
+
 function createDeferred(): {
 	promise: Promise<void>;
 	resolve: () => void;
@@ -242,6 +265,147 @@ describe("Agent", () => {
 		expect(receivedSignal?.aborted).toBe(true);
 	});

+	it("should ignore tool updates after the tool execution settles", async () => {
+		const toolSchema = Type.Object({});
+		let delayedUpdate: AgentToolUpdateCallback<{ status: string }> | undefined;
+		const events: AgentEvent[] = [];
+		const unhandledRejections: unknown[] = [];
+		const onUnhandledRejection = (error: unknown) => {
+			unhandledRejections.push(error);
+		};
+		const tool: AgentTool<typeof toolSchema, { status: string }> = {
+			name: "delayed_tool",
+			label: "Delayed Tool",
+			description: "Captures progress callbacks",
+			parameters: toolSchema,
+			async execute(_toolCallId, _params, _signal, onUpdate) {
+				delayedUpdate = onUpdate;
+				onUpdate?.({
+					content: [{ type: "text", text: "running" }],
+					details: { status: "running" },
+				});
+				return {
+					content: [{ type: "text", text: "ok" }],
+					details: { status: "done" },
+					terminate: true,
+				};
+			},
+		};
+		const agent = new Agent({
+			initialState: { tools: [tool] },
+			streamFn: () => {
+				const stream = new MockAssistantStream();
+				queueMicrotask(() => {
+					stream.push({
+						type: "done",
+						reason: "toolUse",
+						message: createAssistantToolUseMessage([
+							{ type: "toolCall", id: "call-1", name: "delayed_tool", arguments: {} },
+						]),
+					});
+				});
+				return stream;
+			},
+		});
+		agent.subscribe((event) => {
+			events.push(event);
+		});
+
+		process.on("unhandledRejection", onUnhandledRejection);
+		try {
+			await agent.prompt("run tool");
+			const eventCountAfterPrompt = events.length;
+
+			delayedUpdate?.({
+				content: [{ type: "text", text: "late" }],
+				details: { status: "late" },
+			});
+			await new Promise((resolve) => setTimeout(resolve, 0));
+
+			expect(events.filter((event) => event.type === "tool_execution_update")).toHaveLength(1);
+			expect(events).toHaveLength(eventCountAfterPrompt);
+			expect(unhandledRejections).toEqual([]);
+		} finally {
+			process.off("unhandledRejection", onUnhandledRejection);
+		}
+	});
+
+	it("should ignore a settled parallel tool update while another tool is still running", async () => {
+		const toolSchema = Type.Object({});
+		const slowStarted = createDeferred();
+		const settledToolEnded = createDeferred();
+		const releaseSlow = createDeferred();
+		let settledToolUpdate: AgentToolUpdateCallback<{ status: string }> | undefined;
+		const events: AgentEvent[] = [];
+		const settledTool: AgentTool<typeof toolSchema, { status: string }> = {
+			name: "settled_tool",
+			label: "Settled Tool",
+			description: "Captures progress callbacks",
+			parameters: toolSchema,
+			async execute(_toolCallId, _params, _signal, onUpdate) {
+				settledToolUpdate = onUpdate;
+				return {
+					content: [{ type: "text", text: "done" }],
+					details: { status: "done" },
+					terminate: true,
+				};
+			},
+		};
+		const slowTool: AgentTool<typeof toolSchema, { status: string }> = {
+			name: "slow_tool",
+			label: "Slow Tool",
+			description: "Keeps the agent run active",
+			parameters: toolSchema,
+			async execute() {
+				slowStarted.resolve();
+				await releaseSlow.promise;
+				return {
+					content: [{ type: "text", text: "done" }],
+					details: { status: "done" },
+					terminate: true,
+				};
+			},
+		};
+		const agent = new Agent({
+			initialState: { tools: [settledTool, slowTool] },
+			streamFn: () => {
+				const stream = new MockAssistantStream();
+				queueMicrotask(() => {
+					stream.push({
+						type: "done",
+						reason: "toolUse",
+						message: createAssistantToolUseMessage([
+							{ type: "toolCall", id: "call-1", name: "settled_tool", arguments: {} },
+							{ type: "toolCall", id: "call-2", name: "slow_tool", arguments: {} },
+						]),
+					});
+				});
+				return stream;
+			},
+		});
+		agent.subscribe((event) => {
+			events.push(event);
+			if (event.type === "tool_execution_end" && event.toolCallId === "call-1") {
+				settledToolEnded.resolve();
+			}
+		});
+
+		const promptPromise = agent.prompt("run tools");
+		await Promise.all([slowStarted.promise, settledToolEnded.promise]);
+		const eventCountBeforeLateUpdate = events.length;
+
+		settledToolUpdate?.({
+			content: [{ type: "text", text: "late" }],
+			details: { status: "late" },
+		});
+		await new Promise((resolve) => setTimeout(resolve, 0));
+		expect(events).toHaveLength(eventCountBeforeLateUpdate);
+
+		releaseSlow.resolve();
+		await promptPromise;
+		expect(events.filter((event) => event.type === "tool_execution_update")).toHaveLength(0);
+	});
+
 	it("should update state with mutators", () => {
 		const agent = new Agent();

@@ -1,5 +1,5 @@
 import { access, chmod, realpath, symlink } from "node:fs/promises";
-import { join } from "node:path";
+import { delimiter, join } from "node:path";
 import { afterEach, describe, expect, it } from "vitest";
 import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
 import { FileError, getOrThrow } from "../../src/harness/types.ts";
@@ -201,6 +201,39 @@ describe("NodeExecutionEnv", () => {
 		expect(result).toEqual({ stdout: `${await realpath(root)}:ok`, stderr: "", exitCode: 0 });
 	});

+	it("uses stdin command transport for legacy WSL bash paths", async () => {
+		if (process.platform === "win32") return;
+		const root = createTempDir();
+		const shellPath = "C:\\Windows\\System32\\bash.exe";
+		const env = new NodeExecutionEnv({ cwd: root });
+		getOrThrow(await env.writeFile(shellPath, '#!/bin/sh\nprintf \'args:%s\\n\' "$*" >&2\nexec /bin/bash "$@"\n'));
+		await chmod(join(root, shellPath), 0o755);
+
+		const originalCwd = process.cwd();
+		const originalPath = process.env.PATH;
+		const platformDescriptor = Object.getOwnPropertyDescriptor(process, "platform");
+		try {
+			process.chdir(root);
+			process.env.PATH = `${root}${delimiter}${originalPath ?? ""}`;
+			Object.defineProperty(process, "platform", {
+				configurable: true,
+				value: "win32",
+			});
+
+			const wslEnv = new NodeExecutionEnv({ cwd: root, shellPath });
+			const nameExpansion = "$" + "{name}";
+			const result = getOrThrow(await wslEnv.exec(`name='World'; echo "Hello, ${nameExpansion}!"`));
+
+			expect(result).toEqual({ stdout: "Hello, World!\n", stderr: "args:-s\n", exitCode: 0 });
+		} finally {
+			process.chdir(originalCwd);
+			process.env.PATH = originalPath;
+			if (platformDescriptor) {
+				Object.defineProperty(process, "platform", platformDescriptor);
+			}
+		}
+	});
+
 	it("streams stdout and stderr chunks", async () => {
 		const root = createTempDir();
 		const env = new NodeExecutionEnv({ cwd: root });