import { afterEach, describe, expect, it } from "bun:test"; import * as fs from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; import type { AgentSideConnection, PromptRequest, SessionNotification } from "@agentclientprotocol/sdk"; import type { Model } from "@oh-my-pi/pi-ai"; import { getConfigRootDir, setAgentDir } from "@oh-my-pi/pi-utils"; import { AcpAgent } from "../src/modes/acp/acp-agent"; import type { AgentSession, AgentSessionEvent } from "../src/session/agent-session"; import { SessionManager } from "../src/session/session-manager"; const TEST_MODELS: Model[] = [ { id: "claude-sonnet-4-20250514", name: "Claude Sonnet", api: "anthropic-messages", provider: "anthropic", baseUrl: "https://example.invalid", reasoning: true, input: ["text", "image"], cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: 200_000, maxTokens: 8_192, }, { id: "gpt-5.4", name: "GPT-5.4", api: "openai-responses", provider: "openai", baseUrl: "https://example.invalid", reasoning: true, input: ["text", "image"], cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: 200_000, maxTokens: 8_192, }, ]; function makeAssistantMessage(text: string, thinking?: string) { const content: Array<{ type: "text"; text: string } | { type: "thinking"; thinking: string }> = [ { type: "text", text }, ]; if (thinking) { content.push({ type: "thinking" as const, thinking }); } return { role: "assistant" as const, content, api: "anthropic-messages" as const, provider: "anthropic" as const, model: TEST_MODELS[0].id, usage: { input: 10, output: 5, cacheRead: 2, cacheWrite: 1, totalTokens: 18, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, stopReason: "stop" as const, timestamp: Date.now(), }; } class FakeAgentSession { sessionManager: SessionManager; sessionId: string; agent: { sessionId: string; waitForIdle: () => Promise }; model: Model | undefined; thinkingLevel: string | undefined; customCommands: [] = []; extensionRunner = undefined; isStreaming = false; queuedMessageCount = 0; systemPrompt = "system"; disposed = false; #listeners = new Set<(event: AgentSessionEvent) => void>(); constructor( cwd: string, private readonly models: Model[] = TEST_MODELS, ) { this.sessionManager = SessionManager.create(cwd); this.sessionId = this.sessionManager.getSessionId(); this.agent = { sessionId: this.sessionId, waitForIdle: async () => {}, }; this.model = models[0]; } get sessionName(): string { return this.sessionManager.getHeader()?.title ?? `Session ${this.sessionId}`; } get modelRegistry(): { getApiKey: (model: Model) => Promise } { return { getApiKey: async (_model: Model) => "test-key", }; } getAvailableModels(): Model[] { return this.models; } getAvailableThinkingLevels(): ReadonlyArray { return ["low", "medium", "high"]; } setThinkingLevel(level: string | undefined): void { this.thinkingLevel = level; } async setModel(model: Model): Promise { this.model = model; } subscribe(listener: (event: AgentSessionEvent) => void): () => void { this.#listeners.add(listener); return () => { this.#listeners.delete(listener); }; } async prompt(text: string): Promise { this.isStreaming = true; this.sessionManager.appendMessage({ role: "user", content: text, timestamp: Date.now() }); const assistantMessage = makeAssistantMessage("pong"); for (const listener of this.#listeners) { listener({ type: "message_update", message: assistantMessage, assistantMessageEvent: { type: "text_delta", delta: "pong" }, } as AgentSessionEvent); } this.sessionManager.appendMessage(assistantMessage); for (const listener of this.#listeners) { listener({ type: "agent_end", messages: [assistantMessage], } as AgentSessionEvent); } this.isStreaming = false; } async abort(): Promise { this.isStreaming = false; } async refreshMCPTools(_tools: unknown[]): Promise {} getContextUsage(): undefined { return undefined; } async switchSession(sessionPath: string): Promise { await this.sessionManager.setSessionFile(sessionPath); this.sessionId = this.sessionManager.getSessionId(); this.agent.sessionId = this.sessionId; return true; } async dispose(): Promise { this.disposed = true; await this.sessionManager.close(); } async reload(): Promise {} async newSession(): Promise { await this.sessionManager.newSession(); this.sessionId = this.sessionManager.getSessionId(); this.agent.sessionId = this.sessionId; return true; } async branch(_entryId: string): Promise<{ cancelled: boolean }> { return { cancelled: false }; } async navigateTree(_targetId: string): Promise<{ cancelled: boolean }> { return { cancelled: false }; } getActiveToolNames(): string[] { return []; } getAllToolNames(): string[] { return []; } setActiveToolsByName(_toolNames: string[]): void {} async sendCustomMessage(_message: string, _options?: unknown): Promise {} async sendUserMessage(_content: string, _options?: unknown): Promise {} async compact(_instructions?: string, _options?: unknown): Promise {} async fork(): Promise { await this.sessionManager.flush(); const forked = await this.sessionManager.fork(); if (!forked) { return false; } this.sessionId = this.sessionManager.getSessionId(); this.agent.sessionId = this.sessionId; return true; } } interface AgentHarness { agent: AcpAgent; updates: SessionNotification[]; abortController: AbortController; sessions: FakeAgentSession[]; cwdA: string; cwdB: string; findSession(sessionId: string): FakeAgentSession | undefined; } function getChunkMessageId(notification: SessionNotification): string | undefined { const update = notification.update as { messageId?: string | null }; return typeof update.messageId === "string" ? update.messageId : undefined; } const cleanupRoots: string[] = []; const originalAgentDir = process.env.PI_CODING_AGENT_DIR; const fallbackAgentDir = path.join(getConfigRootDir(), "agent"); afterEach(async () => { if (originalAgentDir) { setAgentDir(originalAgentDir); } else { setAgentDir(fallbackAgentDir); delete process.env.PI_CODING_AGENT_DIR; } for (const root of cleanupRoots.splice(0)) { await fs.promises.rm(root, { recursive: true, force: true }); } }); async function createHarness(): Promise { const root = await fs.promises.mkdtemp(path.join(os.tmpdir(), "omp-acp-test-")); cleanupRoots.push(root); const agentDir = path.join(root, "agent"); const cwdA = path.join(root, "cwd-a"); const cwdB = path.join(root, "cwd-b"); await fs.promises.mkdir(agentDir, { recursive: true }); await fs.promises.mkdir(cwdA, { recursive: true }); await fs.promises.mkdir(cwdB, { recursive: true }); setAgentDir(agentDir); const updates: SessionNotification[] = []; const abortController = new AbortController(); const sessions: FakeAgentSession[] = []; const connection = { sessionUpdate: async (notification: SessionNotification) => { updates.push(notification); }, signal: abortController.signal, closed: Promise.withResolvers().promise, } as unknown as AgentSideConnection; const initialSession = new FakeAgentSession(cwdA); sessions.push(initialSession); const factory = async (cwd: string): Promise => { const session = new FakeAgentSession(cwd); sessions.push(session); return session as unknown as AgentSession; }; return { agent: new AcpAgent(connection, initialSession as unknown as AgentSession, factory), updates, abortController, sessions, cwdA, cwdB, findSession: (sessionId: string) => sessions.find(session => session.sessionId === sessionId), }; } describe("ACP agent", () => { it("supports multiple live ACP sessions with model and lifecycle handlers", async () => { const harness = await createHarness(); const first = await harness.agent.newSession({ cwd: harness.cwdA, mcpServers: [] }); const second = await harness.agent.newSession({ cwd: harness.cwdB, mcpServers: [] }); expect(first.models?.availableModels.map(model => model.modelId)).toEqual( TEST_MODELS.map(model => `${model.provider}/${model.id}`), ); await harness.agent.unstable_setSessionModel({ sessionId: first.sessionId, modelId: `${TEST_MODELS[1]!.provider}/${TEST_MODELS[1]!.id}`, }); await harness.agent.setSessionConfigOption({ sessionId: first.sessionId, configId: "thinking", value: "high", }); const firstSession = harness.findSession(first.sessionId); const secondSession = harness.findSession(second.sessionId); expect(firstSession?.model?.id).toBe(TEST_MODELS[1]!.id); expect(firstSession?.thinkingLevel).toBe("high"); expect(secondSession?.model?.id).toBe(TEST_MODELS[0]!.id); expect(secondSession?.thinkingLevel).toBeUndefined(); firstSession?.sessionManager.appendMessage({ role: "user", content: "fork me", timestamp: Date.now() }); await firstSession?.sessionManager.flush(); const forked = await harness.agent.unstable_forkSession({ sessionId: first.sessionId, cwd: harness.cwdA, mcpServers: [], }); const forkedSession = harness.findSession(forked.sessionId); const forkedMessages = forkedSession?.sessionManager.buildSessionContext().messages ?? []; expect(forked.sessionId).not.toBe(first.sessionId); expect(forkedMessages.some(message => message.role === "user" && message.content === "fork me")).toBe(true); await harness.agent.unstable_closeSession({ sessionId: forked.sessionId }); await expect(harness.agent.setSessionMode({ sessionId: forked.sessionId, modeId: "default" })).rejects.toThrow( "Unsupported ACP session", ); harness.abortController.abort(); await Bun.sleep(0); }); it("replays messageIds and returns turn usage for prompts", async () => { const harness = await createHarness(); const stored = new FakeAgentSession(harness.cwdA); harness.sessions.push(stored); stored.sessionManager.appendMessage({ role: "user", content: "hello", timestamp: Date.now() }); stored.sessionManager.appendMessage(makeAssistantMessage("reply", "reasoning")); await stored.sessionManager.ensureOnDisk(); await stored.sessionManager.flush(); await harness.agent.loadSession({ sessionId: stored.sessionId, cwd: harness.cwdA, mcpServers: [] }); const replayChunks = harness.updates.filter( update => update.sessionId === stored.sessionId && (update.update.sessionUpdate === "user_message_chunk" || update.update.sessionUpdate === "agent_message_chunk" || update.update.sessionUpdate === "agent_thought_chunk"), ); const replayAssistantChunks = replayChunks.filter( update => update.update.sessionUpdate === "agent_message_chunk" || update.update.sessionUpdate === "agent_thought_chunk", ); expect( replayChunks.every( update => typeof getChunkMessageId(update) === "string" && getChunkMessageId(update)!.length > 0, ), ).toBe(true); expect(new Set(replayAssistantChunks.map(update => getChunkMessageId(update))).size).toBe(1); const live = await harness.agent.newSession({ cwd: harness.cwdB, mcpServers: [] }); const response = await harness.agent.prompt({ sessionId: live.sessionId, messageId: "05b17a6f-b310-4be7-b767-6b4f3a84eb63", prompt: [{ type: "text", text: "ping" }], } as PromptRequest); const liveChunks = harness.updates.filter( update => update.sessionId === live.sessionId && update.update.sessionUpdate === "agent_message_chunk", ); expect(response.userMessageId).toBe("05b17a6f-b310-4be7-b767-6b4f3a84eb63"); expect(response.usage).toEqual({ inputTokens: 10, outputTokens: 5, cachedReadTokens: 2, cachedWriteTokens: 1, totalTokens: 18, }); expect( liveChunks.some( update => typeof getChunkMessageId(update) === "string" && getChunkMessageId(update)!.length > 0, ), ).toBe(true); harness.abortController.abort(); await Bun.sleep(0); }); });