diff --git a/tests/web/openpi-web-provider.e2e.ts b/tests/web/openpi-web-provider.e2e.ts index 0a0b535b..d505e680 100644 --- a/tests/web/openpi-web-provider.e2e.ts +++ b/tests/web/openpi-web-provider.e2e.ts @@ -139,3 +139,440 @@ test("thinking level reaches the provider request end to end", async ({ await rm(workspace, { recursive: true, force: true }); } }); + +test("preserves a held turn identity across A/B/A replacement and reload", async ({ + page, +}, testInfo) => { + const provider = await startFakeProvider(); + const workspace = await mkdtemp(join(tmpdir(), "openpi-stop-provider-")); + try { + const imported = await page.request.post("/api/workspaces", { + headers: authHeaders, + data: { path: workspace }, + }); + expect(imported.status()).toBe(201); + const { path: canonicalWorkspace } = await imported.json(); + + const createdA = await page.request.post("/api/sessions", { + headers: authHeaders, + data: { + workspacePath: canonicalWorkspace, + commandId: "stop-provider-session-a", + }, + }); + expect(createdA.status()).toBe(201); + const createdABody = (await createdA.json()) as { + sessionId?: unknown; + sessionPath?: unknown; + }; + expect(typeof createdABody.sessionId).toBe("string"); + expect(typeof createdABody.sessionPath).toBe("string"); + const sessionAPath = createdABody.sessionPath as string; + + const initialSnapshot = await page.request.get("/api/snapshot", { + headers: authHeaders, + }); + const sessionAId = ( + (await initialSnapshot.json()) as { + currentSessionId?: unknown; + } + ).currentSessionId; + expect(typeof sessionAId).toBe("string"); + + const selected = await page.request.post("/api/model", { + headers: authHeaders, + data: { + provider: PROVIDER_ID, + modelId: MODEL_ID, + sessionId: sessionAId, + }, + }); + expect(selected.status()).toBe(200); + + await page.goto("/", { waitUntil: "domcontentloaded" }); + const prompt = page.getByRole("textbox", { name: "描述任务" }); + await expect(prompt).toBeVisible(); + + // Persist A before replacing the active runtime; Pi writes a newly created + // session file once its first assistant message reaches a terminal event. + await prompt.fill("Persist Session A before switching"); + await page.getByRole("button", { name: "发送", exact: true }).click(); + await expect + .poll(() => provider.requests.length, { timeout: 15_000 }) + .toBe(1); + await expect(page.locator(".message-row.assistant").last()).toContainText( + "Thinking level acknowledged.", + ); + + provider.holdNextResponse(); + const firstPromptRequest = page.waitForRequest( + (request) => + new URL(request.url()).pathname === "/api/prompt" && + request.method() === "POST", + ); + await prompt.fill("Keep the first Session running"); + await page.getByRole("button", { name: "发送", exact: true }).click(); + const firstPromptBody = (await firstPromptRequest).postDataJSON() as { + commandId?: unknown; + sessionId?: unknown; + }; + expect(typeof firstPromptBody.commandId).toBe("string"); + expect(firstPromptBody.sessionId).toBe(sessionAId); + + await expect + .poll(() => provider.requests.length, { timeout: 15_000 }) + .toBe(2); + const runningAResponse = await page.request.get("/api/snapshot", { + headers: authHeaders, + }); + const runningA = (await runningAResponse.json()) as { + currentSessionId?: unknown; + runtime: { + status: string; + activeTurn?: { + sessionId: string; + commandId: string; + epoch: number; + }; + }; + }; + expect(runningA.currentSessionId).toBe(sessionAId); + expect(runningA.runtime.status).toBe("running"); + const firstTurn = runningA.runtime.activeTurn; + expect(firstTurn).toEqual({ + sessionId: sessionAId, + commandId: firstPromptBody.commandId, + epoch: expect.any(Number), + }); + + const createdB = await page.request.post("/api/sessions", { + headers: authHeaders, + data: { + workspacePath: canonicalWorkspace, + commandId: "stop-provider-session-b", + }, + }); + expect(createdB.status()).toBe(201); + const createdBBody = (await createdB.json()) as { + sessionId?: unknown; + }; + expect(typeof createdBBody.sessionId).toBe("string"); + expect(createdBBody.sessionId).not.toBe(sessionAId); + + const runningBResponse = await page.request.get("/api/snapshot", { + headers: authHeaders, + }); + const runningB = (await runningBResponse.json()) as { + currentSessionId?: unknown; + runtime: { status: string; activeTurn?: unknown }; + }; + expect(runningB.currentSessionId).toBe(createdBBody.sessionId); + expect(runningB.runtime.status).toBe("idle"); + expect(runningB.runtime.activeTurn).toBeUndefined(); + + // The original A handle is scoped to A and cannot stop the selected B. + const staleOnB = await page.request.post("/api/turns/cancel", { + headers: authHeaders, + data: firstTurn, + }); + expect(staleOnB.status()).toBe(409); + expect(await staleOnB.json()).toMatchObject({ + state: "stale-session", + accepted: false, + }); + expect(provider.requests).toHaveLength(2); + + const selectedA = await page.request.post("/api/sessions/select", { + headers: authHeaders, + data: { path: sessionAPath }, + }); + const selectedABody = await selectedA.json(); + expect(selectedA.status(), JSON.stringify(selectedABody)).toBe(200); + await expect + .poll( + async () => + ( + await page.request.get("/api/snapshot", { + headers: authHeaders, + }) + ).json() as Promise<{ + currentSessionId?: unknown; + runtime: { status: string; activeTurn?: unknown }; + }>, + { timeout: 15_000 }, + ) + .toMatchObject({ + currentSessionId: sessionAId, + runtime: { status: "running", activeTurn: firstTurn }, + }); + + await page.reload({ waitUntil: "domcontentloaded" }); + await expect(prompt).toBeVisible(); + const stop = page.getByRole("button", { + name: "停止当前轮次", + exact: true, + }); + await expect(stop).toBeVisible(); + await expect + .poll( + async () => + ( + await page.request.get("/api/snapshot", { + headers: authHeaders, + }) + ).json() as Promise<{ + runtime: { status: string; activeTurn?: unknown }; + }>, + { timeout: 15_000 }, + ) + .toMatchObject({ + runtime: { status: "running", activeTurn: firstTurn }, + }); + await page.screenshot({ + path: testInfo.outputPath("stop-session-switch-running.png"), + fullPage: true, + }); + + const firstCancelResponse = page.waitForResponse( + (response) => + new URL(response.url()).pathname === "/api/turns/cancel" && + response.request().method() === "POST", + ); + await stop.click(); + const firstCancel = await firstCancelResponse; + expect(firstCancel.status()).toBe(202); + expect(await firstCancel.json()).toMatchObject({ + state: "accepted", + accepted: true, + }); + await expect + .poll( + async () => + ( + await page.request.get("/api/snapshot", { + headers: authHeaders, + }) + ).json() as Promise<{ + runtime: { status: string; activeTurn?: unknown }; + }>, + { timeout: 15_000 }, + ) + .toMatchObject({ runtime: { status: "idle" } }); + await expect(stop).toHaveCount(0); + await expect(page.locator(".composer-hint")).toContainText( + "当前轮次已停止。", + ); + provider.release(); + + // A new turn gets a new identity. Reusing the old handle is idempotent and + // must not cancel the new one. + provider.holdNextResponse(); + const secondPromptRequest = page.waitForRequest( + (request) => + new URL(request.url()).pathname === "/api/prompt" && + request.method() === "POST", + ); + await prompt.fill("Start a fresh turn in Session A"); + await page.getByRole("button", { name: "发送", exact: true }).click(); + const secondPromptBody = (await secondPromptRequest).postDataJSON() as { + commandId?: unknown; + }; + await expect + .poll(() => provider.requests.length, { timeout: 15_000 }) + .toBe(3); + const runningNewAResponse = await page.request.get("/api/snapshot", { + headers: authHeaders, + }); + const runningNewA = (await runningNewAResponse.json()) as { + runtime: { + status: string; + activeTurn?: { + sessionId: string; + commandId: string; + epoch: number; + }; + }; + }; + expect(runningNewA.runtime.status).toBe("running"); + const secondTurn = runningNewA.runtime.activeTurn; + expect(secondTurn).toEqual({ + sessionId: sessionAId, + commandId: secondPromptBody.commandId, + epoch: expect.any(Number), + }); + expect(secondTurn).not.toEqual(firstTurn); + + const staleOnNewA = await page.request.post("/api/turns/cancel", { + headers: authHeaders, + data: firstTurn, + }); + expect(staleOnNewA.status()).toBe(200); + expect(await staleOnNewA.json()).toMatchObject({ + state: "already-settled", + accepted: false, + }); + expect(provider.requests).toHaveLength(3); + await expect(stop).toBeVisible(); + + const secondCancelResponse = page.waitForResponse( + (response) => + new URL(response.url()).pathname === "/api/turns/cancel" && + response.request().method() === "POST", + ); + await stop.click(); + const secondCancel = await secondCancelResponse; + expect(secondCancel.status()).toBe(202); + expect(await secondCancel.json()).toMatchObject({ + state: "accepted", + accepted: true, + }); + await expect + .poll( + async () => + ( + await page.request.get("/api/snapshot", { + headers: authHeaders, + }) + ).json() as Promise<{ + runtime: { status: string; activeTurn?: unknown }; + }>, + { timeout: 15_000 }, + ) + .toMatchObject({ runtime: { status: "idle" } }); + provider.release(); + } finally { + await provider.close(); + await rm(workspace, { recursive: true, force: true }); + } +}); + +test("keeps native cron wake cancellation tied to its runtime identity", async ({ + page, +}, testInfo) => { + test.setTimeout(100_000); + const provider = await startFakeProvider(); + const workspace = await mkdtemp(join(tmpdir(), "openpi-cron-provider-")); + try { + const imported = await page.request.post("/api/workspaces", { + headers: authHeaders, + data: { path: workspace }, + }); + expect(imported.status()).toBe(201); + const { path: canonicalWorkspace } = await imported.json(); + + const created = await page.request.post("/api/sessions", { + headers: authHeaders, + data: { + workspacePath: canonicalWorkspace, + commandId: "cron-provider-session", + }, + }); + expect(created.status()).toBe(201); + const snapshot = await page.request.get("/api/snapshot", { + headers: authHeaders, + }); + const sessionId = ( + (await snapshot.json()) as { + currentSessionId?: unknown; + } + ).currentSessionId; + expect(typeof sessionId).toBe("string"); + + const selected = await page.request.post("/api/model", { + headers: authHeaders, + data: { provider: PROVIDER_ID, modelId: MODEL_ID, sessionId }, + }); + expect(selected.status()).toBe(200); + + await page.goto("/", { waitUntil: "domcontentloaded" }); + const prompt = page.getByRole("textbox", { name: "描述任务" }); + await expect(prompt).toBeVisible(); + + // The isolated package source registers /cron. Hold its eventual provider + // response so the native agent_start identity is visible to the UI. + provider.holdNextResponse(); + const cronPromptResponse = page.waitForResponse( + (response) => + new URL(response.url()).pathname === "/api/prompt" && + response.request().method() === "POST", + ); + await prompt.fill("/cron in 30s Verify native wake cancellation"); + await page.getByRole("button", { name: "发送", exact: true }).click(); + const cronResponse = await cronPromptResponse; + expect(cronResponse.status()).toBe(202); + expect(await cronResponse.json()).toMatchObject({ + accepted: true, + state: "accepted", + }); + + await expect + .poll(() => provider.requests.length, { timeout: 85_000 }) + .toBe(1); + const nativeSnapshotResponse = await page.request.get("/api/snapshot", { + headers: authHeaders, + }); + const nativeSnapshot = (await nativeSnapshotResponse.json()) as { + currentSessionId?: unknown; + runtime: { + status: string; + activeTurn?: { + sessionId: string; + commandId: string; + epoch: number; + }; + }; + }; + expect(nativeSnapshot.currentSessionId).toBe(sessionId); + expect(nativeSnapshot.runtime.status).toBe("running"); + const nativeTurn = nativeSnapshot.runtime.activeTurn; + expect(nativeTurn).toEqual({ + sessionId, + commandId: expect.stringMatching(/^native-/u), + epoch: expect.any(Number), + }); + + const stop = page.getByRole("button", { + name: "停止当前轮次", + exact: true, + }); + await expect(stop).toBeVisible(); + await page.screenshot({ + path: testInfo.outputPath("native-cron-wake-running.png"), + fullPage: true, + }); + + const cancelResponse = page.waitForResponse( + (response) => + new URL(response.url()).pathname === "/api/turns/cancel" && + response.request().method() === "POST", + ); + await stop.click(); + const cancel = await cancelResponse; + expect(cancel.status()).toBe(202); + expect(await cancel.json()).toMatchObject({ + state: "accepted", + accepted: true, + }); + await expect + .poll( + async () => + ( + await page.request.get("/api/snapshot", { + headers: authHeaders, + }) + ).json() as Promise<{ + runtime: { status: string; activeTurn?: unknown }; + }>, + { timeout: 15_000 }, + ) + .toMatchObject({ runtime: { status: "idle" } }); + await expect(stop).toHaveCount(0); + await expect(page.locator(".composer-hint")).toContainText( + "当前轮次已停止。", + ); + provider.release(); + } finally { + await provider.close(); + await rm(workspace, { recursive: true, force: true }); + } +}); diff --git a/tests/web/pi-runtime.test.ts b/tests/web/pi-runtime.test.ts index 32f33898..582415bc 100644 --- a/tests/web/pi-runtime.test.ts +++ b/tests/web/pi-runtime.test.ts @@ -23,8 +23,18 @@ type Trace = { outcome?: "completed" | "cancelled" | "failed" | "uncertain"; }; +type ExecutionMap = Map< + object, + { + turn: NonNullable>; + startedAt: number; + outcome?: Trace["outcome"]; + } +>; + type RuntimeHarness = { runtime: { session: object; dispose?: () => Promise }; + runtimeExecutions: ExecutionMap; activePromptTrace?: Trace; pendingPromptTraces: Trace[]; liveMessageSequence: number; @@ -100,8 +110,92 @@ type FakeAgentRuntime = { session: PromptSession; dispose: () => Promise; }; + +function executionSession( + sessionId: string, + sessionPath: string, + isStreaming: boolean, +) { + type RuntimeEvent = { type: string; message?: Record }; + const listeners = new Set<(event: RuntimeEvent) => void>(); + const session = { + isStreaming, + sessionManager: { + getSessionId: () => sessionId, + getSessionFile: () => sessionPath, + getCwd: () => "/workspace", + }, + aborts: 0, + async abort() { + session.aborts += 1; + }, + subscribe(listener: (event: RuntimeEvent) => void) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + emit(event: RuntimeEvent) { + for (const listener of listeners) listener(event); + }, + }; + return session; +} + +function executionRuntime(session: ReturnType) { + return { + session, + cwd: "/workspace", + setRebindSession() {}, + async dispose() {}, + }; +} + +type ExecutionHarness = ReturnType; + +function executionHarness(runtime: ReturnType) { + return Object.assign( + Object.create(PiWebRuntime.prototype) as Pick< + PiWebRuntime, + "getActiveTurn" | "cancelTurn" | "switchSession" | "subscribe" + >, + { + runtime, + runtimeExecutions: new Map(), + listeners: new Set<(event: WebRuntimeEvent) => void>(), + retainedRuntimes: new Set(), + retainedSubscriptions: new Map(), + inFlightRuntimes: new Map(), + runtimeDisposalPromises: new WeakMap(), + runtimeDisposals: new Set(), + promptOperations: new Set(), + runtimeOperations: new Set(), + candidateRuntimes: new Set(), + pendingPromptTraces: [], + turnAbortOperations: new Map(), + terminalTurnKeys: new Set(), + turnSettlementWaiters: new Map(), + nextTurnEpoch: 0, + liveMessageSequence: 0, + controllerMutation: Promise.resolve(), + promptAdmission: Promise.resolve(), + disposed: false, + hasSelectedWorkspace: true, + }, + ); +} + +const projectExecutionEvent = ( + PiWebRuntime.prototype as unknown as { + projectEvent( + this: ExecutionHarness, + session: ReturnType, + event: { type: string; message?: Record }, + ): void; + } +).projectEvent; + type PromptRuntimeHarness = { runtime: FakeAgentRuntime; + runtimeExecutions: ExecutionMap; listeners: Set<(event: WebRuntimeEvent) => void>; retainedRuntimes: Set; retainedSubscriptions: Map void>; @@ -125,6 +219,7 @@ type PromptRuntimeHarness = { dispatcherLease: { release: () => Promise }; webHostLease: { release: () => Promise }; sendPrompt: PiWebRuntime["sendPrompt"]; + projectEvent(session: object, event: object): void; cancelTurn: PiWebRuntime["cancelTurn"]; subscribe: PiWebRuntime["subscribe"]; dispose: PiWebRuntime["dispose"]; @@ -265,6 +360,7 @@ function promptHarness(session: ReturnType) { PiWebRuntime.prototype, ) as unknown as PromptRuntimeHarness; harness.runtime = { session, dispose: async () => undefined }; + harness.runtimeExecutions = new Map(); harness.listeners = new Set(); harness.retainedRuntimes = new Set(); harness.retainedSubscriptions = new Map(); @@ -811,6 +907,153 @@ test("later prompt failures retain their command and Session correlation", async }); }); +test("an active execution keeps its Stop identity across a Session switch", async () => { + const sessionA = executionSession("session-a", "/sessions/a.jsonl", true); + const runtimeA = executionRuntime(sessionA); + const runtime = executionHarness(runtimeA); + projectExecutionEvent.call(runtime, sessionA, { type: "agent_start" }); + const activeTurn = runtime.getActiveTurn(); + assert.ok(activeTurn); + + const sessionB = executionSession("session-b", "/sessions/b.jsonl", false); + const runtimeB = executionRuntime(sessionB); + runtime.retainedRuntimes.add(runtimeB); + runtime.retainedSubscriptions.set(runtimeB, () => undefined); + + await runtime.switchSession(sessionB.sessionManager.getSessionFile()); + assert.equal(runtime.getActiveTurn(), undefined); + await runtime.switchSession(sessionA.sessionManager.getSessionFile()); + + assert.deepEqual(runtime.getActiveTurn(), activeTurn); + sessionA.abort = async () => { + sessionA.aborts++; + sessionA.emit({ + type: "message_end", + message: { role: "assistant", stopReason: "aborted" }, + }); + sessionA.isStreaming = false; + sessionA.emit({ type: "agent_settled" }); + }; + assert.equal((await runtime.cancelTurn(activeTurn)).state, "accepted"); + assert.equal(sessionA.aborts, 1); + assert.equal(runtime.turnAbortOperations.size, 0); + assert.equal(runtime.getActiveTurn(), undefined); + assert.equal((await runtime.cancelTurn(activeTurn)).state, "already-settled"); + + sessionA.isStreaming = true; + sessionA.emit({ type: "agent_start" }); + const nextTurn = runtime.getActiveTurn(); + assert.ok(nextTurn); + assert.ok(nextTurn.epoch > activeTurn.epoch); + assert.notEqual(nextTurn.commandId, activeTurn.commandId); + assert.equal((await runtime.cancelTurn(activeTurn)).state, "already-settled"); + assert.equal( + (await runtime.cancelTurn({ ...nextTurn, epoch: nextTurn.epoch + 1 })) + .state, + "stale-turn", + ); + assert.equal( + (await runtime.cancelTurn({ ...nextTurn, sessionId: "session-b" })).state, + "stale-session", + ); + assert.equal(sessionA.aborts, 1); +}); + +test("a native agent start receives a cancellable identity without a Web prompt", () => { + const session = executionSession( + "native-session", + "/sessions/native.jsonl", + true, + ); + const runtime = executionHarness(executionRuntime(session)); + const events: WebRuntimeEvent[] = []; + runtime.subscribe((event) => events.push(event)); + + projectExecutionEvent.call(runtime, session, { type: "agent_start" }); + + const activeTurn = runtime.getActiveTurn(); + assert.ok(activeTurn); + assert.equal(activeTurn.sessionId, "native-session"); + assert.match(activeTurn.commandId, /^native-/u); + assert.equal(activeTurn.epoch, 1); + assert.deepEqual( + events.filter((event) => event.type === "turn_started")[0]?.detail, + activeTurn, + ); + assert.deepEqual( + events.filter((event) => event.type === "agent_start")[0]?.detail, + { sessionId: "native-session", activeTurn }, + ); + projectExecutionEvent.call(runtime, session, { type: "agent_start" }); + assert.deepEqual(runtime.getActiveTurn(), activeTurn); + assert.equal( + events.filter((event) => event.type === "turn_started").length, + 1, + ); +}); + +test("a retained Session settles its execution identity while another Session is active", async () => { + const sessionA = executionSession("session-a", "/sessions/a.jsonl", true); + const runtimeA = executionRuntime(sessionA); + const runtime = executionHarness(runtimeA); + const events: WebRuntimeEvent[] = []; + runtime.subscribe((event) => events.push(event)); + projectExecutionEvent.call(runtime, sessionA, { type: "agent_start" }); + const activeTurn = runtime.getActiveTurn(); + assert.ok(activeTurn); + + const sessionB = executionSession("session-b", "/sessions/b.jsonl", false); + const runtimeB = executionRuntime(sessionB); + runtime.retainedRuntimes.add(runtimeB); + runtime.retainedSubscriptions.set(runtimeB, () => undefined); + await runtime.switchSession(sessionB.sessionManager.getSessionFile()); + + sessionA.emit({ + type: "message_end", + message: { role: "assistant", stopReason: "stop" }, + }); + sessionA.emit({ type: "agent_settled" }); + + assert.deepEqual( + events.filter((event) => event.type === "turn_settled").at(-1)?.detail, + { ...activeTurn, outcome: "completed" }, + ); + await runtime.switchSession(sessionA.sessionManager.getSessionFile()); + assert.equal(runtime.getActiveTurn(), undefined); +}); + +test("a retained native wake receives a Stop identity before it is promoted", async () => { + const currentSession = executionSession( + "current", + "/sessions/current.jsonl", + false, + ); + const runtime = executionHarness(executionRuntime(currentSession)); + const retainedSession = executionSession( + "native-session", + "/sessions/native.jsonl", + true, + ); + const retainedRuntime = executionRuntime(retainedSession); + const retainRuntime = ( + PiWebRuntime.prototype as unknown as { + retainRuntime( + this: typeof runtime, + candidate: typeof retainedRuntime, + ): void; + } + ).retainRuntime; + retainRuntime.call(runtime, retainedRuntime); + + retainedSession.emit({ type: "agent_start" }); + await runtime.switchSession(retainedSession.sessionManager.getSessionFile()); + + const activeTurn = runtime.getActiveTurn(); + assert.ok(activeTurn); + assert.equal(activeTurn.sessionId, "native-session"); + assert.match(activeTurn.commandId, /^native-/u); +}); + test("turn cancellation reports uncertainty without assistant terminal evidence", async () => { const session = promptSession("session-a"); let aborts = 0; @@ -822,11 +1065,13 @@ test("turn cancellation reports uncertainty without assistant terminal evidence" commandId: "command-a", sessionId: "session-a", startedAt: 1, - started: true, + started: false, queued: false, epoch: 7, }; runtime.activePromptTrace = trace; + runtime.nextTurnEpoch = 6; + runtime.projectEvent(session, { type: "agent_start" }); const projectEvent = ( PiWebRuntime.prototype as unknown as { projectEvent( @@ -898,11 +1143,16 @@ test("turn cancellation loses to a naturally completed terminal run", async () = commandId: "command-a", sessionId: "session-a", startedAt: 1, - started: true, + started: false, queued: false, epoch: 1, outcome: "completed", }; + runtime.projectEvent(session, { type: "agent_start" }); + runtime.projectEvent(session, { + type: "message_end", + message: { role: "assistant", stopReason: "stop" }, + }); const projectEvent = ( PiWebRuntime.prototype as unknown as { projectEvent( @@ -936,10 +1186,11 @@ test("a repeated cancellation does not issue another native abort while settling commandId: "command-a", sessionId: "session-a", startedAt: 1, - started: true, + started: false, queued: false, epoch: 1, }; + runtime.projectEvent(session, { type: "agent_start" }); runtime.turnAbortOperations.set( "session-a\u0000command-a\u00001", new Promise(() => undefined), @@ -972,10 +1223,11 @@ test("turn cancellation reports native abort failures", async () => { commandId: "command-a", sessionId: "session-a", startedAt: 1, - started: true, + started: false, queued: false, epoch: 1, }; + runtime.projectEvent(session, { type: "agent_start" }); assert.deepEqual( await runtime.cancelTurn({ @@ -1369,6 +1621,7 @@ test("message_end and queued prompts do not settle a running turn", () => { const session = { sessionManager: { getSessionId: () => "session" } }; const harness = Object.create(PiWebRuntime.prototype) as RuntimeHarness; harness.runtime = { session }; + harness.runtimeExecutions = new Map(); harness.pendingPromptTraces = []; harness.liveMessageSequence = 0; harness.listeners = new Set(); @@ -1518,6 +1771,7 @@ test("toolUse message_end without a terminal result settles as uncertain", () => const session = { sessionManager: { getSessionId: () => "session" } }; const harness = Object.create(PiWebRuntime.prototype) as RuntimeHarness; harness.runtime = { session }; + harness.runtimeExecutions = new Map(); harness.pendingPromptTraces = []; harness.liveMessageSequence = 0; harness.listeners = new Set(); @@ -1583,6 +1837,7 @@ test("toolUse message_end without a terminal result settles as uncertain", () => }); type ThinkingHarness = { + runtimeExecutions: ExecutionMap; runtime: { session: ReturnType["session"] }; listeners: Set<(event: WebRuntimeEvent) => void>; controllerMutation: Promise; @@ -1647,6 +1902,7 @@ function thinkingHarness( PiWebRuntime.prototype, ) as unknown as ThinkingHarness; harness.runtime = { session }; + harness.runtimeExecutions = new Map(); harness.listeners = new Set(); harness.controllerMutation = Promise.resolve(); harness.runtimeOperations = new Set(); diff --git a/tests/web/provider-e2e-support.ts b/tests/web/provider-e2e-support.ts index 998b1c97..2e2a2de2 100644 --- a/tests/web/provider-e2e-support.ts +++ b/tests/web/provider-e2e-support.ts @@ -1,6 +1,7 @@ import { mkdirSync, writeFileSync } from "node:fs"; import { createServer, type IncomingMessage, type Server } from "node:http"; -import { join } from "node:path"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; /** * Shared identity for the hermetic provider round-trip end-to-end test. @@ -17,6 +18,10 @@ export const PROVIDER_ID = "fake-provider"; export const MODEL_ID = "fake-reasoner"; export const MODEL_NAME = "Fake Reasoner"; export const PROVIDER_BASE_URL = `http://127.0.0.1:${PROVIDER_PORT}/v1`; +export const OPENPI_SOURCE = resolve( + dirname(fileURLToPath(import.meta.url)), + "../..", +); export type RecordedProviderRequest = { method: string; @@ -61,7 +66,11 @@ export function seedAgentDirectory(agentDirectory: string) { // reach the provider, rather than a no-op against the initial clamped level. writeFileSync( join(agentDirectory, "settings.json"), - `${JSON.stringify({ defaultThinkingLevel: "off" }, null, 2)}\n`, + `${JSON.stringify( + { defaultThinkingLevel: "off", packages: [OPENPI_SOURCE] }, + null, + 2, + )}\n`, ); } diff --git a/web/runtime/pi-runtime.ts b/web/runtime/pi-runtime.ts index 2bce0cc3..36a37713 100644 --- a/web/runtime/pi-runtime.ts +++ b/web/runtime/pi-runtime.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { mkdir, realpath, stat } from "node:fs/promises"; import { join, resolve } from "node:path"; import { createEvidenceWriteTool } from "./write-evidence.ts"; @@ -79,13 +80,18 @@ type PromptTrace = { queued: boolean; userMessageObserved: boolean; epoch?: number; - outcome?: "completed" | "cancelled" | "failed" | "uncertain"; }; type TurnSettlement = WebActiveTurn & { outcome: "completed" | "cancelled" | "failed" | "uncertain"; }; +type RuntimeExecution = { + turn: WebActiveTurn; + startedAt: number; + outcome?: TurnSettlement["outcome"]; +}; + function errorText(error: unknown) { return error instanceof Error ? error.message : String(error); } @@ -145,6 +151,8 @@ export class PiWebRuntime implements WebRuntimeController { }; private activePromptTrace?: PromptTrace; private readonly pendingPromptTraces: PromptTrace[] = []; + /** Pi owns execution identity across Web prompt and Session projections. */ + private readonly runtimeExecutions = new WeakMap(); private nextTurnEpoch = 0; private readonly terminalTurnKeys = new Set(); private readonly turnSettlementWaiters = new Map< @@ -269,7 +277,7 @@ export class PiWebRuntime implements WebRuntimeController { } getActiveTurn() { - return this.activeTurnFromTrace(this.activePromptTrace); + return this.runtimeExecutions.get(this.runtime)?.turn; } cancelTurn(options: WebTurnCancellationOptions) { @@ -317,7 +325,7 @@ export class PiWebRuntime implements WebRuntimeController { let timeoutHandle: ReturnType | undefined; try { const abortOperation = this.runtime.session.abort(); - this.turnAbortOperations.set(key, abortOperation); + if (!this.terminalTurnKeys.has(key)) this.turnAbortOperations.set(key, abortOperation); void abortOperation.catch(() => { if (this.turnAbortOperations.get(key) === abortOperation) { this.turnAbortOperations.delete(key); @@ -814,7 +822,6 @@ export class PiWebRuntime implements WebRuntimeController { }); } if (promptTrace) { - promptTrace.outcome = "failed"; traceWeb("prompt_operation_failed", { commandId: promptTrace.commandId, sessionId, @@ -1131,25 +1138,28 @@ export class PiWebRuntime implements WebRuntimeController { private projectEvent(session: AgentSession, event: AgentSessionEvent) { if (session !== this.runtime.session) return; - if (event.type === "agent_start" && this.activePromptTrace) { - this.startPromptTrace(this.activePromptTrace); - } + this.observeExecutionEvent(this.runtime, event); if (event.type === "message_start" && event.message.role === "user") { if (!this.activePromptTrace) { this.activePromptTrace = this.pendingPromptTraces.shift(); } if (this.activePromptTrace) { - this.startPromptTrace(this.activePromptTrace); + const execution = this.runtimeExecutions.get(this.runtime); + if (execution) { + this.activePromptTrace.started = true; + this.activePromptTrace.epoch = execution.turn.epoch; + } this.activePromptTrace.userMessageObserved = true; } } const promptTrace = this.activePromptTrace; - if (promptTrace) { + const execution = this.runtimeExecutions.get(this.runtime); + if (promptTrace || execution) { const eventDetail: Record = { - commandId: promptTrace.commandId, - sessionId: promptTrace.sessionId, + commandId: execution?.turn.commandId ?? promptTrace?.commandId, + sessionId: execution?.turn.sessionId ?? promptTrace?.sessionId, type: event.type, - elapsedMs: elapsed(promptTrace.startedAt), + elapsedMs: elapsed(execution?.startedAt ?? promptTrace?.startedAt ?? 0), }; if (event.type === "message_update") { eventDetail.contentChars = projectMessage(event.message).content.length; @@ -1191,9 +1201,6 @@ export class PiWebRuntime implements WebRuntimeController { // Pi emits this only after the whole agent run (including tool loops // and admitted follow-ups) has reached a terminal state. A // message_end is only one model response and must not settle a turn. - if (this.activePromptTrace?.started) { - this.settlePromptTrace(this.activePromptTrace); - } this.activePromptTrace = undefined; this.pendingPromptTraces.length = 0; this.emit(event.type, { @@ -1222,30 +1229,6 @@ export class PiWebRuntime implements WebRuntimeController { break; case "message_update": case "message_end": - if ( - event.type === "message_end" && - event.message.role === "assistant" && - this.activePromptTrace - ) { - // Preserve the terminal model result for classification, but defer - // publication until Pi confirms the entire run is settled. - const outcome = - event.message.stopReason === "aborted" - ? "cancelled" - : event.message.stopReason === "error" - ? "failed" - : event.message.stopReason === "stop" || - event.message.stopReason === "length" - ? "completed" - : undefined; - // A later queued continuation must not erase proof that the - // provider result targeted by Stop was aborted. The control remains - // owned until agent_settled; this outcome does not claim that every - // queued follow-up in the same Pi execution was cancelled. - if (outcome && this.activePromptTrace.outcome !== "cancelled") { - this.activePromptTrace.outcome = outcome; - } - } this.emit(event.type, { message: projectMessage(event.message, (path) => resolve(this.cwd, path)), ...(this.liveMessageKey ? { messageKey: this.liveMessageKey } : {}), @@ -1272,32 +1255,49 @@ export class PiWebRuntime implements WebRuntimeController { for (const listener of this.listeners) listener({ type, detail }); } - private activeTurnFromTrace(trace?: PromptTrace): WebActiveTurn | undefined { - if (!trace?.started || trace.epoch === undefined) return undefined; - return { - sessionId: trace.sessionId, - commandId: trace.commandId, - epoch: trace.epoch, - }; - } - - private startPromptTrace(trace: PromptTrace) { - if (trace.started) return; - trace.started = true; - trace.epoch = ++this.nextTurnEpoch; - const activeTurn = this.activeTurnFromTrace(trace); - if (activeTurn) this.emit("turn_started", { ...activeTurn }); - } - - private settlePromptTrace(trace: PromptTrace) { - const activeTurn = this.activeTurnFromTrace(trace); - if (!activeTurn) return; + private observeExecutionEvent(runtime: AgentSessionRuntime, event: AgentSessionEvent) { + let execution = this.runtimeExecutions.get(runtime); + if (event.type === "agent_start" && !execution) { + const sessionId = runtime.session.sessionManager.getSessionId(); + const trace = runtime === this.runtime && + this.activePromptTrace?.sessionId === sessionId && !this.activePromptTrace.started + ? this.activePromptTrace : undefined; + execution = { + turn: { + sessionId, + commandId: trace?.commandId ?? `native-${randomUUID()}`, + epoch: ++this.nextTurnEpoch, + }, + startedAt: performance.now(), + }; + this.runtimeExecutions.set(runtime, execution); + if (trace) { + trace.started = true; + trace.epoch = execution.turn.epoch; + } + this.emit("turn_started", { ...execution.turn }); + } + if (!execution) return; + if (event.type === "message_end" && event.message.role === "assistant") { + const outcome = + event.message.stopReason === "aborted" + ? "cancelled" + : event.message.stopReason === "error" + ? "failed" + : event.message.stopReason === "stop" || event.message.stopReason === "length" + ? "completed" + : undefined; + // Admitted follow-ups share the Pi run. Keep proof that the response + // targeted by Stop was aborted, even if a later continuation completes. + if (outcome && execution.outcome !== "cancelled") execution.outcome = outcome; + } + if (event.type !== "agent_settled") return; + this.runtimeExecutions.delete(runtime); const settlement: TurnSettlement = { - ...activeTurn, - outcome: - trace.outcome ?? "uncertain", + ...execution.turn, + outcome: execution.outcome ?? "uncertain", }; - const key = this.turnKey(activeTurn); + const key = this.turnKey(execution.turn); if (this.terminalTurnKeys.has(key)) return; this.terminalTurnKeys.add(key); this.turnAbortOperations.delete(key); @@ -1382,6 +1382,7 @@ export class PiWebRuntime implements WebRuntimeController { private retainRuntime(runtime: AgentSessionRuntime) { this.retainedRuntimes.add(runtime); const unsubscribe = runtime.session.subscribe((event) => { + this.observeExecutionEvent(runtime, event); if (event.type !== "agent_settled") return; this.emit("session_progress", { sessionId: runtime.session.sessionManager.getSessionId(), @@ -1486,6 +1487,5 @@ export class PiWebRuntime implements WebRuntimeController { private resetPromptTraces() { this.activePromptTrace = undefined; this.pendingPromptTraces.length = 0; - this.turnAbortOperations.clear(); } }