From bd854751c1afbd0de9be7cbd04bf51cb069ecfdb Mon Sep 17 00:00:00 2001 From: testikun <320479488+testikun@users.noreply.github.com> Date: Fri, 4 Sep 2026 17:44:32 +0800 Subject: [PATCH 1/6] feat(web): add turn-scoped cancellation --- docs/development/OPENPI_WEB_DEVELOPMENT.md | 8 + tests/web/app-render.test.ts | 70 +++++++++ tests/web/pi-adapter.test.ts | 2 + tests/web/pi-runtime.test.ts | 175 ++++++++++++++++++++- tests/web/web-host.test.ts | 84 ++++++++++ web/adapter/pi-adapter.ts | 3 + web/host/web-host.ts | 45 ++++++ web/protocol/types.ts | 2 + web/runtime/pi-runtime.ts | 168 +++++++++++++++++++- web/runtime/types.ts | 24 +++ web/ui/app.js | 109 +++++++++++-- web/ui/index.html | 3 + web/ui/styles.css | 5 + 13 files changed, 684 insertions(+), 14 deletions(-) diff --git a/docs/development/OPENPI_WEB_DEVELOPMENT.md b/docs/development/OPENPI_WEB_DEVELOPMENT.md index eef5bc97..8a2f71f6 100644 --- a/docs/development/OPENPI_WEB_DEVELOPMENT.md +++ b/docs/development/OPENPI_WEB_DEVELOPMENT.md @@ -51,6 +51,14 @@ bun run dev:web -- /absolute/path/to/workspace 异常进程恢复会在 Web Session 目录的 `.openpi-web-host.artifacts/` 中保留安全围栏。只有确认没有存活或暂停的 Web Host 仍依赖这些记录后，才可人工删除其中过期的 `candidate-*`、`released-*` 或 `stale-*` 目录。OpenPI 不会自动删除围栏；达到 128 个租约产物或 64 个 stale 围栏时会 fail closed，并在错误信息中给出该目录。普通 Session 文件不占用这个预算。 +## 活动回合取消协议 + +Web 的 Stop 只取消当前活动的 provider 回合，不等同于停止 Host，也不会清空已经排队的 follow-up。取消请求必须回传当前快照或 `turn_started` 事件给出的 `sessionId`、`commandId` 和 `epoch`；Runtime 在自己的串行 mutation 边界内重新核对三者，再调用 Pi 原生 `AgentSession.abort()`。 + +Host 返回 `accepted`、`already-settled`、`stale-session`、`stale-turn` 或 `failed` 的明确收据。浏览器不会因点击按钮而乐观结束运行态；只有 Pi 消息的 `stopReason: "aborted"` 投影成 `turn_settled(outcome: "cancelled")` 后才显示取消终态。活动回合身份也包含在快照中，因此刷新和 SSE 重连仍能恢复正确的 Stop 控件。多客户端的旧请求不能取消更新的回合，重复请求则按已终结回合幂等返回。 + +这个边界源自 [Issue #342](https://github.com/openpi-dev/openpi/issues/342)。Host disposal 仍由独立生命周期处理；全局暂停属于其他设计范围。 + `dev:web` 和 `dev:web:backend` 默认会在启动它们的终端输出 Web 诊断日志；设置 `OPENPI_WEB_DEBUG=0` 可关闭。正式运行 `openpi web` 默认关闭日志，排查时设置 `OPENPI_WEB_DEBUG=1`。 ## 对话无响应排查 diff --git a/tests/web/app-render.test.ts b/tests/web/app-render.test.ts index c35a3f88..a19c9cba 100644 --- a/tests/web/app-render.test.ts +++ b/tests/web/app-render.test.ts @@ -401,6 +401,10 @@ async function renderApp( "sendPrompt", context as vm.Context, ) as () => Promise, + cancelActiveTurn: vm.runInContext( + "cancelActiveTurn", + context as vm.Context, + ) as () => Promise, updateComposer: vm.runInContext( "updateComposer", context as vm.Context, @@ -867,6 +871,72 @@ test("app.js settles an admitted prompt that Pi handles without an agent turn", assert.equal((app.state.terminalPromptIds as Set).size, 32); }); +test("app.js stops only the canonical active turn without optimistic settlement", async () => { + const app = await renderApp(); + const cancellation = deferred>(); + app.context.fetch = async (url: unknown) => { + if (String(url) === "/api/turns/cancel") return cancellation.promise; + if (String(url).startsWith("/api/snapshot")) return response(SNAPSHOT); + throw new Error(`unexpected request: ${String(url)}`); + }; + vm.runInContext( + 'applyRuntimeEvent({sequence: 2, type: "turn_started", detail: {sessionId: "s1", commandId: "c1", epoch: 4}})', + app.context as vm.Context, + ); + + assert.equal(app.state.liveRunning, true); + assert.equal(app.elements.get("stop-turn")?.hidden, false); + assert.equal(app.elements.get("send-prompt")?.hidden, true); + const stopping = app.cancelActiveTurn(); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(app.state.liveRunning, true); + assert.equal(app.state.turnCancellationPending, true); + + vm.runInContext( + 'applyRuntimeEvent({sequence: 3, type: "turn_settled", detail: {sessionId: "s1", commandId: "c1", epoch: 4, outcome: "cancelled"}})', + app.context as vm.Context, + ); + cancellation.resolve( + response({ + sessionId: "s1", + commandId: "c1", + epoch: 4, + state: "accepted", + accepted: true, + }), + ); + await stopping; + + assert.equal(app.state.liveRunning, false); + assert.equal(app.state.activeTurn, null); + assert.equal(app.elements.get("stop-turn")?.hidden, true); + assert.equal(app.elements.get("send-prompt")?.hidden, false); + assert.equal( + app.elements.get("composer-hint")?.textContent, + "Current turn stopped.", + ); +}); + +test("app.js restores the active turn and Stop control from a snapshot", async () => { + const running = structuredClone(SNAPSHOT) as SnapshotFixture & { + runtime: typeof SNAPSHOT.runtime & { + activeTurn: { sessionId: string; commandId: string; epoch: number }; + }; + }; + running.runtime.status = "running"; + running.runtime.activeTurn = { + sessionId: "s1", + commandId: "c1", + epoch: 9, + }; + const app = await renderApp({ snapshot: running }); + + assert.deepEqual(app.state.activeTurn, running.runtime.activeTurn); + assert.equal(app.state.liveRunning, true); + assert.equal(app.elements.get("stop-turn")?.hidden, false); + assert.equal(app.elements.get("send-prompt")?.hidden, true); +}); + test("app.js scopes model selection to its session epoch", async () => { const app = await renderApp(); const model = deferred>(); diff --git a/tests/web/pi-adapter.test.ts b/tests/web/pi-adapter.test.ts index af2f7476..bdb4fcff 100644 --- a/tests/web/pi-adapter.test.ts +++ b/tests/web/pi-adapter.test.ts @@ -29,6 +29,8 @@ function runtimeFor( sessionDirectory, sessionManager, isIdle: () => true, + getActiveTurn: () => undefined, + cancelTurn: async (options) => ({ ...options, state: "stale-turn" }), sendPrompt: async () => {}, newSession: async () => ({ cancelled: false }), switchSession: async () => ({ cancelled: false }), diff --git a/tests/web/pi-runtime.test.ts b/tests/web/pi-runtime.test.ts index ae6f3f78..b96106c1 100644 --- a/tests/web/pi-runtime.test.ts +++ b/tests/web/pi-runtime.test.ts @@ -17,6 +17,9 @@ type Trace = { startedAt: number; started: boolean; queued: boolean; + userMessageObserved?: boolean; + epoch?: number; + outcome?: "completed" | "cancelled" | "failed"; }; type RuntimeHarness = { @@ -26,6 +29,9 @@ type RuntimeHarness = { liveMessageSequence: number; liveMessageKey?: string; listeners: Set<(event: WebRuntimeEvent) => void>; + nextTurnEpoch: number; + terminalTurnKeys: Set; + turnSettlementWaiters: Map void>>; }; function deferred() { @@ -84,11 +90,17 @@ type PromptRuntimeHarness = { runtimeDisposalPromises: WeakMap>; promptAdmission: Promise; pendingPromptTraces: Trace[]; + activePromptTrace?: Trace; + nextTurnEpoch: number; + terminalTurnKeys: Set; + turnSettlementWaiters: Map void>>; + controllerMutation: Promise; disposed: boolean; hasSelectedWorkspace: boolean; dispatcherLease: { release: () => Promise }; webHostLease: { release: () => Promise }; sendPrompt: PiWebRuntime["sendPrompt"]; + cancelTurn: PiWebRuntime["cancelTurn"]; subscribe: PiWebRuntime["subscribe"]; dispose: PiWebRuntime["dispose"]; }; @@ -237,6 +249,10 @@ function promptHarness(session: ReturnType) { harness.runtimeDisposalPromises = new WeakMap(); harness.promptAdmission = Promise.resolve(); harness.pendingPromptTraces = []; + harness.nextTurnEpoch = 0; + harness.terminalTurnKeys = new Set(); + harness.turnSettlementWaiters = new Map(); + harness.controllerMutation = Promise.resolve(); harness.disposed = false; harness.hasSelectedWorkspace = true; harness.dispatcherLease = { release: async () => undefined }; @@ -556,6 +572,108 @@ test("later prompt failures retain their command and Session correlation", async }); }); +test("turn cancellation is bound, canonical, and idempotent", async () => { + const session = promptSession("session-a"); + let aborts = 0; + session.abort = async () => { + aborts += 1; + }; + const runtime = promptHarness(session); + const trace: Trace = { + commandId: "command-a", + sessionId: "session-a", + startedAt: 1, + started: true, + queued: false, + epoch: 7, + outcome: "cancelled", + }; + runtime.activePromptTrace = trace; + const settlePromptTrace = ( + PiWebRuntime.prototype as unknown as { + settlePromptTrace(this: PromptRuntimeHarness, trace: Trace): void; + } + ).settlePromptTrace; + + const cancellation = runtime.cancelTurn({ + sessionId: "session-a", + commandId: "command-a", + epoch: 7, + }); + await Promise.resolve(); + settlePromptTrace.call(runtime, trace); + + assert.deepEqual(await cancellation, { + sessionId: "session-a", + commandId: "command-a", + epoch: 7, + state: "accepted", + }); + assert.equal(aborts, 1); + assert.equal( + ( + await runtime.cancelTurn({ + sessionId: "session-a", + commandId: "command-a", + epoch: 7, + }) + ).state, + "already-settled", + ); + assert.equal( + ( + await runtime.cancelTurn({ + sessionId: "session-a", + commandId: "command-a", + epoch: 8, + }) + ).state, + "stale-turn", + ); + assert.equal( + ( + await runtime.cancelTurn({ + sessionId: "session-b", + commandId: "command-a", + epoch: 7, + }) + ).state, + "stale-session", + ); + assert.equal(aborts, 1); +}); + +test("turn cancellation reports native abort failures", async () => { + const session = promptSession("session-a"); + session.abort = async () => { + throw new Error("abort failed"); + }; + const runtime = promptHarness(session); + runtime.activePromptTrace = { + commandId: "command-a", + sessionId: "session-a", + startedAt: 1, + started: true, + queued: false, + epoch: 1, + }; + + assert.deepEqual( + await runtime.cancelTurn({ + sessionId: "session-a", + commandId: "command-a", + epoch: 1, + }), + { + sessionId: "session-a", + commandId: "command-a", + epoch: 1, + state: "failed", + error: "abort failed", + }, + ); +}); + test("retained Session cleanup waits for all of its prompt operations", async () => { const sessionA = promptSession("session-a"); const sessionB = promptSession("session-b"); @@ -910,12 +1028,17 @@ test("runtime creation failure releases the Web Host lease", async () => { }); test("prompt traces advance with queued user messages", () => { - const session = {}; + const session = { sessionManager: { getSessionId: () => "session" } }; const harness = Object.create(PiWebRuntime.prototype) as RuntimeHarness; harness.runtime = { session }; harness.pendingPromptTraces = []; harness.liveMessageSequence = 0; harness.listeners = new Set(); + harness.nextTurnEpoch = 0; + harness.terminalTurnKeys = new Set(); + harness.turnSettlementWaiters = new Map(); + const events: WebRuntimeEvent[] = []; + harness.listeners.add((event) => events.push(event)); harness.activePromptTrace = { commandId: "first", sessionId: "session", @@ -941,12 +1064,62 @@ test("prompt traces advance with queued user messages", () => { message: { role: "user", content: [{ type: "text", text }] }, }); + projectEvent.call(harness, session, { type: "agent_start" }); projectEvent.call(harness, session, userMessage("first")); assert.equal(harness.activePromptTrace?.commandId, "first"); assert.equal(harness.activePromptTrace?.started, true); + projectEvent.call(harness, session, { + type: "message_end", + message: { + role: "assistant", + content: [], + stopReason: "aborted", + timestamp: 3, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + }, + }); + projectEvent.call(harness, session, userMessage("second")); assert.equal(harness.activePromptTrace?.commandId, "second"); assert.equal(harness.activePromptTrace?.started, true); + assert.equal(harness.activePromptTrace?.epoch, 2); assert.equal(harness.pendingPromptTraces.length, 0); + assert.deepEqual( + events + .filter((event) => event.type.startsWith("turn_")) + .map((event) => ({ type: event.type, detail: event.detail })), + [ + { + type: "turn_started", + detail: { sessionId: "session", commandId: "first", epoch: 1 }, + }, + { + type: "turn_settled", + detail: { + sessionId: "session", + commandId: "first", + epoch: 1, + outcome: "cancelled", + }, + }, + { + type: "turn_started", + detail: { sessionId: "session", commandId: "second", epoch: 2 }, + }, + ], + ); }); diff --git a/tests/web/web-host.test.ts b/tests/web/web-host.test.ts index d80df4fa..681f7c26 100644 --- a/tests/web/web-host.test.ts +++ b/tests/web/web-host.test.ts @@ -49,6 +49,8 @@ test("serves workspaces through a runtime isolated from terminal sessions", asyn return sessionManager; }, isIdle: () => false, + getActiveTurn: () => undefined, + cancelTurn: async (options) => ({ ...options, state: "stale-turn" }), sendPrompt: async (content) => { prompts.push(content); }, @@ -527,6 +529,8 @@ test("an unbound Host exposes no bootstrap Session and rejects prompt bypasses", sessionDirectory: root, sessionManager, isIdle: () => true, + getActiveTurn: () => undefined, + cancelTurn: async (options) => ({ ...options, state: "stale-turn" }), sendPrompt: async () => { prompts++; }, @@ -586,6 +590,21 @@ test("an unbound Host exposes no bootstrap Session and rejects prompt bypasses", }); assert.equal(prompts, 0); + const cancellation = await fetch( + `${launched.origin}/api/turns/cancel`, + { + method: "POST", + headers, + body: JSON.stringify({ + sessionId: sessionManager.getSessionId(), + commandId: "command-a", + epoch: 1, + }), + }, + ); + assert.equal(cancellation.status, 409); + assert.equal((await cancellation.json()).code, "WORKSPACE_REQUIRED"); + const started = events.find((event) => event.type === "web_host_started"); assert.ok(started); assert.equal("cwd" in (started.detail ?? {}), false); @@ -611,6 +630,8 @@ test("returns accepted only after Pi admits the prompt", async () => { cwd, sessionManager, isIdle: () => false, + getActiveTurn: () => undefined, + cancelTurn: async (options) => ({ ...options, state: "stale-turn" }), sendPrompt: async () => { promptStarted = true; await promptAdmitted; @@ -663,6 +684,67 @@ test("returns accepted only after Pi admits the prompt", async () => { } }); +test("returns an exact receipt for a turn-bound cancellation", async () => { + const cwd = await mkdtemp(join(tmpdir(), "openpi-web-cancel-")); + const runtime = testRuntime(cwd); + const activeTurn = { + sessionId: runtime.sessionManager.getSessionId(), + commandId: "command-a", + epoch: 3, + }; + runtime.getActiveTurn = () => activeTurn; + runtime.cancelTurn = async (options) => ({ + ...options, + state: "accepted", + }); + const { host, launched, headers } = await startTestHost(runtime); + try { + const snapshotResponse = await fetch(`${launched.origin}/api/snapshot`, { + headers, + }); + assert.equal(snapshotResponse.status, 200); + assert.deepEqual( + (await snapshotResponse.json()).runtime.activeTurn, + activeTurn, + ); + + const response = await fetch(`${launched.origin}/api/turns/cancel`, { + method: "POST", + headers: { ...headers, "Content-Type": "application/json" }, + body: JSON.stringify(activeTurn), + }); + assert.equal(response.status, 202); + assert.deepEqual(await response.json(), { + ...activeTurn, + state: "accepted", + accepted: true, + cursor: 1, + }); + + runtime.cancelTurn = async (options) => ({ + ...options, + state: "stale-turn", + }); + const stale = await fetch(`${launched.origin}/api/turns/cancel`, { + method: "POST", + headers: { ...headers, "Content-Type": "application/json" }, + body: JSON.stringify(activeTurn), + }); + assert.equal(stale.status, 409); + assert.equal((await stale.json()).state, "stale-turn"); + + const invalid = await fetch(`${launched.origin}/api/turns/cancel`, { + method: "POST", + headers: { ...headers, "Content-Type": "application/json" }, + body: JSON.stringify({ ...activeTurn, epoch: 0 }), + }); + assert.equal(invalid.status, 400); + } finally { + await host.stop(); + await rm(cwd, { recursive: true, force: true }); + } +}); + function testRuntime( cwd: string, sendPrompt: WebRuntimeController["sendPrompt"] = async () => {}, @@ -674,6 +756,8 @@ function testRuntime( cwd, sessionManager, isIdle: () => true, + getActiveTurn: () => undefined, + cancelTurn: async (options) => ({ ...options, state: "stale-turn" }), sendPrompt, newSession: async () => ({ cancelled: false }), switchSession: async () => ({ cancelled: false }), diff --git a/web/adapter/pi-adapter.ts b/web/adapter/pi-adapter.ts index fb4c0bc9..ba46852c 100644 --- a/web/adapter/pi-adapter.ts +++ b/web/adapter/pi-adapter.ts @@ -490,6 +490,9 @@ export class PiWebAdapter { status: this.runtime.isIdle() ? ("idle" as const) : ("running" as const), + ...(this.runtime.getActiveTurn() + ? { activeTurn: this.runtime.getActiveTurn() } + : {}), capabilities: webCapabilitySnapshot(this.runtime.sessionManager), }, truncation: { diff --git a/web/host/web-host.ts b/web/host/web-host.ts index 6203e647..f542d155 100644 --- a/web/host/web-host.ts +++ b/web/host/web-host.ts @@ -276,6 +276,7 @@ export class WebHost { if (request.method === "GET" || request.method === "HEAD") return false; const pathname = new URL(request.url ?? "/", `http://${HOST}`).pathname; if (pathname === "/api/prompt") return false; + if (pathname === "/api/turns/cancel") return true; return pathname.startsWith("/api/workspaces") || pathname.startsWith("/api/sessions") || pathname === "/api/model"; @@ -541,6 +542,50 @@ export class WebHost { cursor: this.sequence, }); } + if (url.pathname === "/api/turns/cancel" && request.method === "POST") { + const body = await this.readJson(request); + if ( + typeof body.sessionId !== "string" || + body.sessionId.length === 0 || + body.sessionId.length > 128 || + typeof body.commandId !== "string" || + body.commandId.length === 0 || + body.commandId.length > 128 || + typeof body.epoch !== "number" || + !Number.isSafeInteger(body.epoch) || + body.epoch <= 0 + ) { + return this.json(response, 400, { + code: "INVALID_TURN", + error: "bounded sessionId, commandId, and positive turn epoch are required", + }); + } + if (this.runtime.workspaceSelected !== true) { + return this.json(response, 409, { + code: "WORKSPACE_REQUIRED", + error: "Choose a workspace before using the Web runtime", + }); + } + const result = await this.runtime.cancelTurn({ + sessionId: body.sessionId, + commandId: body.commandId, + epoch: body.epoch, + }); + traceWeb("turn_cancel_receipt", { ...result }); + const status = + result.state === "accepted" + ? 202 + : result.state === "already-settled" + ? 200 + : result.state === "failed" + ? 500 + : 409; + return this.json(response, status, { + ...result, + accepted: result.state === "accepted", + cursor: this.sequence, + }); + } if (request.method !== "GET") { return this.json(response, 405, { error: "method not allowed" }); } diff --git a/web/protocol/types.ts b/web/protocol/types.ts index a7429b6e..ecaa6632 100644 --- a/web/protocol/types.ts +++ b/web/protocol/types.ts @@ -1,5 +1,6 @@ import type { SessionEntry } from "@earendil-works/pi-coding-agent"; import type { WebCapabilitySnapshot } from "../../extensions/shared/web-observer-registry.ts"; +import type { WebActiveTurn } from "../runtime/types.ts"; export const WEB_PROTOCOL_VERSION = 1; export const WEB_MAX_EVENTS = 200; @@ -113,6 +114,7 @@ export interface WebSnapshot { models: WebModelSummary[]; runtime: { status: "idle" | "running" | "unknown"; + activeTurn?: WebActiveTurn; capabilities: WebCapabilitySnapshot; }; truncation: WebSnapshotTruncation; diff --git a/web/runtime/pi-runtime.ts b/web/runtime/pi-runtime.ts index cf792d39..2df53879 100644 --- a/web/runtime/pi-runtime.ts +++ b/web/runtime/pi-runtime.ts @@ -15,11 +15,14 @@ import { hasTrustRequiringProjectResources, } from "@earendil-works/pi-coding-agent"; import { + type WebActiveTurn, type WebModelSelectionOptions, type WebPromptOptions, type WebRuntimeController, type WebRuntimeEvent, type WebSessionCreationOptions, + type WebTurnCancellationOptions, + type WebTurnCancellationResult, WebRuntimeRequestError, } from "./types.ts"; import { projectMessage } from "../protocol/types.ts"; @@ -43,6 +46,13 @@ type PromptTrace = { startedAt: number; started: boolean; queued: boolean; + userMessageObserved: boolean; + epoch?: number; + outcome?: "completed" | "cancelled" | "failed"; +}; + +type TurnSettlement = WebActiveTurn & { + outcome: "completed" | "cancelled" | "failed"; }; function errorText(error: unknown) { @@ -77,6 +87,12 @@ export class PiWebRuntime implements WebRuntimeController { private promptAdmission: Promise = Promise.resolve(); private activePromptTrace?: PromptTrace; private readonly pendingPromptTraces: PromptTrace[] = []; + private nextTurnEpoch = 0; + private readonly terminalTurnKeys = new Set(); + private readonly turnSettlementWaiters = new Map< + string, + Set<(settlement: TurnSettlement) => void> + >(); private liveMessageKey?: string; private liveMessageSequence = 0; private readonly webSessionDirectory: string; @@ -174,6 +190,74 @@ export class PiWebRuntime implements WebRuntimeController { return !this.runtime.session.isStreaming; } + getActiveTurn() { + return this.activeTurnFromTrace(this.activePromptTrace); + } + + cancelTurn(options: WebTurnCancellationOptions) { + return this.serializeControllerMutation(() => + this.cancelActiveTurn(options), + ); + } + + private async cancelActiveTurn( + options: WebTurnCancellationOptions, + ): Promise { + this.assertActive(); + this.assertWorkspaceSelected(); + const activeSessionId = this.runtime.session.sessionManager.getSessionId(); + if (options.sessionId !== activeSessionId) { + return { ...options, state: "stale-session" }; + } + const key = this.turnKey(options); + if (this.terminalTurnKeys.has(key)) { + return { ...options, state: "already-settled" }; + } + const activeTurn = this.getActiveTurn(); + if ( + !activeTurn || + activeTurn.commandId !== options.commandId || + activeTurn.epoch !== options.epoch + ) { + return { ...options, state: "stale-turn" }; + } + + let ownWaiter: ((settlement: TurnSettlement) => void) | undefined; + const settlement = new Promise((resolveSettlement) => { + ownWaiter = resolveSettlement; + const waiters = this.turnSettlementWaiters.get(key) ?? new Set(); + waiters.add(resolveSettlement); + this.turnSettlementWaiters.set(key, waiters); + }); + try { + const abortOperation = this.runtime.session.abort(); + const abortFailure = new Promise((_, reject) => { + void abortOperation.catch(reject); + }); + const terminal = await Promise.race([settlement, abortFailure]); + return { + ...options, + state: + terminal.outcome === "cancelled" + ? "accepted" + : terminal.outcome === "completed" + ? "already-settled" + : "failed", + ...(terminal.outcome === "failed" + ? { error: "The active turn failed while cancellation was requested" } + : {}), + }; + } catch (error) { + return { ...options, state: "failed", error: errorText(error) }; + } finally { + const waiters = this.turnSettlementWaiters.get(key); + if (waiters && ownWaiter) { + waiters.delete(ownWaiter); + if (waiters.size === 0) this.turnSettlementWaiters.delete(key); + } + } + } + listModels() { const current = this.runtime.session.model; const available = [...this.runtime.services.modelRuntime.getAvailableSnapshot()]; @@ -295,6 +379,7 @@ export class PiWebRuntime implements WebRuntimeController { startedAt, started: false, queued, + userMessageObserved: false, } : undefined; this.retainRuntimeReference(agentRuntime); @@ -432,6 +517,7 @@ export class PiWebRuntime implements WebRuntimeController { }); } if (promptTrace) { + promptTrace.outcome = "failed"; traceWeb("prompt_operation_failed", { commandId: promptTrace.commandId, sessionId, @@ -725,13 +811,23 @@ 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); + } if (event.type === "message_start" && event.message.role === "user") { if (!this.activePromptTrace) { this.activePromptTrace = this.pendingPromptTraces.shift(); - } else if (this.activePromptTrace.started && this.pendingPromptTraces.length > 0) { + } else if ( + this.activePromptTrace.userMessageObserved && + this.pendingPromptTraces.length > 0 + ) { + this.settlePromptTrace(this.activePromptTrace); this.activePromptTrace = this.pendingPromptTraces.shift(); } - if (this.activePromptTrace) this.activePromptTrace.started = true; + if (this.activePromptTrace) { + this.startPromptTrace(this.activePromptTrace); + this.activePromptTrace.userMessageObserved = true; + } } const promptTrace = this.activePromptTrace; if (promptTrace) { @@ -770,15 +866,24 @@ export class PiWebRuntime implements WebRuntimeController { } switch (event.type) { case "agent_start": + this.emit(event.type, { + sessionId: session.sessionManager.getSessionId(), + ...(this.getActiveTurn() + ? { activeTurn: this.getActiveTurn() } + : {}), + }); + break; case "agent_settled": - this.emit(event.type); if ( - event.type === "agent_settled" && this.activePromptTrace?.started && this.pendingPromptTraces.length === 0 ) { + this.settlePromptTrace(this.activePromptTrace); this.activePromptTrace = undefined; } + this.emit(event.type, { + sessionId: session.sessionManager.getSessionId(), + }); break; case "auto_retry_start": this.emit(event.type, { @@ -796,6 +901,18 @@ export class PiWebRuntime implements WebRuntimeController { break; case "message_update": case "message_end": + if ( + event.type === "message_end" && + event.message.role === "assistant" && + this.activePromptTrace + ) { + this.activePromptTrace.outcome = + event.message.stopReason === "aborted" + ? "cancelled" + : event.message.stopReason === "error" + ? "failed" + : "completed"; + } this.emit(event.type, { message: projectMessage(event.message), ...(this.liveMessageKey ? { messageKey: this.liveMessageKey } : {}), @@ -819,10 +936,53 @@ 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; + const settlement: TurnSettlement = { + ...activeTurn, + outcome: trace.outcome ?? "completed", + }; + const key = this.turnKey(activeTurn); + if (this.terminalTurnKeys.has(key)) return; + this.terminalTurnKeys.add(key); + while (this.terminalTurnKeys.size > 64) { + const oldest = this.terminalTurnKeys.values().next().value; + if (typeof oldest === "string") this.terminalTurnKeys.delete(oldest); + } + this.emit("turn_settled", { ...settlement }); + for (const resolveSettlement of this.turnSettlementWaiters.get(key) ?? []) { + resolveSettlement(settlement); + } + this.turnSettlementWaiters.delete(key); + } + + private turnKey(turn: WebActiveTurn) { + return `${turn.sessionId}\u0000${turn.commandId}\u0000${turn.epoch}`; + } + private removePromptTrace(trace: PromptTrace) { const pendingIndex = this.pendingPromptTraces.indexOf(trace); if (pendingIndex !== -1) this.pendingPromptTraces.splice(pendingIndex, 1); if (this.activePromptTrace !== trace) return; + if (trace.started) this.settlePromptTrace(trace); this.activePromptTrace = this.pendingPromptTraces.shift(); } diff --git a/web/runtime/types.ts b/web/runtime/types.ts index 2b66c3f0..6bc8b9ee 100644 --- a/web/runtime/types.ts +++ b/web/runtime/types.ts @@ -33,6 +33,26 @@ export interface WebPromptOptions { expectedSessionId?: string; } +export interface WebActiveTurn { + sessionId: string; + commandId: string; + epoch: number; +} + +export interface WebTurnCancellationOptions extends WebActiveTurn {} + +export type WebTurnCancellationState = + | "accepted" + | "already-settled" + | "stale-session" + | "stale-turn" + | "failed"; + +export interface WebTurnCancellationResult extends WebActiveTurn { + state: WebTurnCancellationState; + error?: string; +} + export interface WebModelSelectionOptions { expectedSessionId?: string; } @@ -54,7 +74,11 @@ export interface WebRuntimeController { readonly sessionDirectory: string; readonly sessionManager: SessionManager; isIdle(): boolean; + getActiveTurn(): WebActiveTurn | undefined; sendPrompt(content: string, options?: WebPromptOptions): Promise; + cancelTurn( + options: WebTurnCancellationOptions, + ): Promise; newSession( workspacePath: string, options?: WebSessionCreationOptions, diff --git a/web/ui/app.js b/web/ui/app.js index bcff449d..575ca5a2 100644 --- a/web/ui/app.js +++ b/web/ui/app.js @@ -19,6 +19,9 @@ const state = { collapsed: readCollapsedWorkspaces(), liveMessages: [], liveRunning: false, + activeTurn: null, + turnCancellationPending: false, + turnTerminalStatus: null, promptAdmissionPending: false, promptAdmissionToken: null, promptAdmissionSequence: 0, @@ -67,6 +70,9 @@ const translations = { enterHint: "Enter to send, Shift+Enter for a new line.", activeOnlyHint: "Only the active Web session accepts messages.", acceptedHint: "Message accepted by OpenPI Web.", + stopTurn: "Stop turn", + stoppingTurn: "Stopping current turn...", + stoppedTurn: "Current turn stopped.", modelRunning: "Working...", modelPreparing: "Preparing task...", modelRetrying: "Retrying model request...", @@ -103,6 +109,9 @@ const translations = { enterHint: "按 Enter 发送，Shift+Enter 换行。", activeOnlyHint: "只有当前 Web 会话可以接收消息。", acceptedHint: "OpenPI Web 已接收消息。", + stopTurn: "停止当前回合", + stoppingTurn: "正在停止当前回合...", + stoppedTurn: "当前回合已停止。", modelRunning: "正在运行...", modelPreparing: "正在准备任务...", modelRetrying: "模型请求重试中...", @@ -482,6 +491,13 @@ function updateComposer() { !selected && !state.snapshot?.currentSessionId; const canCompose = active || newSessionDraft; + const activeTurn = active + ? state.activeTurn || state.snapshot?.runtime.activeTurn || null + : null; + const canStop = Boolean( + activeTurn && + (state.snapshot?.runtime.status === "running" || state.liveRunning), + ); $("prompt-input").disabled = state.sessionSwitching || (!canCompose && Boolean(state.selectedWorkspace)); $("send-prompt").disabled = @@ -489,6 +505,9 @@ function updateComposer() { !canCompose || !state.selectedWorkspace || state.promptAdmissionPending; + $("send-prompt").hidden = canStop; + $("stop-turn").hidden = !canStop; + $("stop-turn").disabled = state.turnCancellationPending; const modelPicker = $("model-picker"); const modelPickerValue = $("model-picker-value"); const modelMenu = $("model-menu"); @@ -527,11 +546,14 @@ function updateComposer() { : active ? t("promptMessage") : t("promptReadonly"); - $("composer-hint").textContent = canCompose - ? state.snapshot.runtime.status === "running" || state.liveRunning - ? t("queuedHint") - : t("enterHint") - : t("activeOnlyHint"); + $("composer-hint").textContent = + state.turnTerminalStatus === "cancelled" + ? t("stoppedTurn") + : canCompose + ? state.snapshot.runtime.status === "running" || state.liveRunning + ? t("queuedHint") + : t("enterHint") + : t("activeOnlyHint"); } async function selectModel(value) { @@ -605,6 +627,9 @@ async function refreshSnapshot({ return false; } state.snapshot = snapshot; + if (resetCursor) resetLiveState(); + state.activeTurn = snapshot.runtime.activeTurn || null; + if (snapshot.runtime.status === "running") state.liveRunning = true; if ( state.snapshot.runtime.status !== "running" && !state.promptAdmissionPending @@ -613,7 +638,6 @@ async function refreshSnapshot({ state.livePhase = "idle"; state.liveRetry = null; } - if (resetCursor) resetLiveState(); state.cursor = resetCursor || state.cursor === null ? state.snapshot.cursor : Math.max(state.cursor, state.snapshot.cursor); @@ -738,6 +762,7 @@ async function sendPrompt() { ].slice(-8); state.promptAdmissionPending = true; state.promptAdmissionToken = admissionToken; + state.turnTerminalStatus = null; renderConversation(); $("composer-hint").classList.remove("error"); try { @@ -772,6 +797,36 @@ async function sendPrompt() { } } +async function cancelActiveTurn() { + const turn = state.activeTurn || state.snapshot?.runtime.activeTurn; + if (!turn || state.turnCancellationPending || state.sessionSwitching) return; + const epoch = state.sessionEpoch; + state.turnCancellationPending = true; + $("composer-hint").classList.remove("error"); + $("composer-hint").textContent = t("stoppingTurn"); + updateComposer(); + try { + const receipt = await api("/api/turns/cancel", { + method: "POST", + body: JSON.stringify(turn), + }); + if (epoch !== state.sessionEpoch) return; + if (receipt.state === "accepted" || receipt.state === "already-settled") { + $("composer-hint").textContent = t("stoppedTurn"); + } + } catch (error) { + if (epoch !== state.sessionEpoch) return; + $("composer-hint").textContent = error.message; + $("composer-hint").classList.add("error"); + await refreshSnapshot({ epoch }); + } finally { + if (epoch === state.sessionEpoch) { + state.turnCancellationPending = false; + renderConversation(); + } + } +} + function resizePrompt() { const input = $("prompt-input"); const maxHeight = 220; @@ -1070,16 +1125,44 @@ function applyRuntimeEvent(event) { state.livePhase = alreadySettled ? "idle" : "preparing"; state.liveRetry = null; renderConversation(); + } else if (event.type === "turn_started") { + state.activeTurn = { + sessionId: event.detail?.sessionId, + commandId: event.detail?.commandId, + epoch: event.detail?.epoch, + }; + state.liveRunning = true; + state.turnTerminalStatus = null; + state.livePhase = "running"; + state.liveRetry = null; + renderConversation(); } else if (event.type === "agent_start") { + if (event.detail?.activeTurn) state.activeTurn = event.detail.activeTurn; state.liveRunning = true; state.livePhase = "running"; state.liveRetry = null; renderConversation(); + } else if (event.type === "turn_settled") { + rememberTerminalPrompt(event.detail?.commandId); + const isActiveTurn = + state.activeTurn?.sessionId === event.detail?.sessionId && + state.activeTurn?.commandId === event.detail?.commandId && + state.activeTurn?.epoch === event.detail?.epoch; + if (isActiveTurn) { + state.activeTurn = null; + state.liveRunning = false; + state.livePhase = "idle"; + state.liveRetry = null; + state.turnTerminalStatus = event.detail?.outcome || null; + } + renderConversation(); } else if (event.type === "agent_settled" || event.type === "prompt_settled") { if (event.type === "prompt_settled") rememberTerminalPrompt(event.detail?.commandId); - state.liveRunning = false; - state.livePhase = "idle"; - state.liveRetry = null; + if (!state.activeTurn) { + state.liveRunning = false; + state.livePhase = "idle"; + state.liveRetry = null; + } renderConversation(); } else if (event.detail?.message) { if (event.detail.message.role === "user") { @@ -1100,6 +1183,8 @@ function applyRuntimeEvent(event) { [ "agent_start", "agent_settled", + "turn_started", + "turn_settled", "prompt_settled", "message_end", "tool_execution_end", @@ -1163,6 +1248,9 @@ let eventLoopStarted = false; function resetLiveState() { state.liveMessages = []; state.liveRunning = false; + state.activeTurn = null; + state.turnCancellationPending = false; + state.turnTerminalStatus = null; state.livePhase = "idle"; state.liveRetry = null; } @@ -1384,6 +1472,9 @@ $("composer")?.addEventListener("submit", (event) => { if (state.selectedWorkspace) void sendPrompt(); else void chooseWorkspace(); }); +$("stop-turn")?.addEventListener("click", () => { + void cancelActiveTurn(); +}); $("prompt-input")?.addEventListener("input", resizePrompt); $("prompt-input")?.addEventListener("keydown", (event) => { if (event.isComposing || event.keyCode === 229) return; diff --git a/web/ui/index.html b/web/ui/index.html index 7aa60047..3b5da19a 100644 --- a/web/ui/index.html +++ b/web/ui/index.html @@ -118,6 +118,9 @@ +
diff --git a/web/ui/styles.css b/web/ui/styles.css index 0270ed9a..47711c3e 100644 --- a/web/ui/styles.css +++ b/web/ui/styles.css @@ -506,6 +506,11 @@ body.sidebar-collapsed .icon-button { width: 36px; height: 36px; } .send-button:hover { background: var(--warm-accent-hover); } .send-button:disabled { background: var(--warm-accent-disabled); color: var(--subtle); } .send-button svg { width: 17px; height: 17px; stroke-width: 2; } +.stop-button { display: grid; width: 34px; height: 34px; flex: 0 0 auto; place-items: center; border: 1px solid #d6cbc1; border-radius: 50%; background: #f7eee8; color: #8a3f32; } +.stop-button:hover { border-color: #c8b3a5; background: #f1e2d9; } +.stop-button:disabled { border-color: var(--border); background: #f3f1ed; color: var(--subtle); } +.stop-button[hidden] { display: none; } +.stop-button svg { width: 16px; height: 16px; fill: currentColor; stroke: none; } .composer-hint { display: none; } .composer-hint.error { color: var(--error); } .sidebar-scrim { display: none; } From a0d18d25fc661d1230b267287d4a8a54d68ca6cf Mon Sep 17 00:00:00 2001 From: testikun <320479488+testikun@users.noreply.github.com> Date: Fri, 4 Sep 2026 18:19:56 +0800 Subject: [PATCH 2/6] style(web): format cancellation endpoint test --- tests/web/web-host.test.ts | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/tests/web/web-host.test.ts b/tests/web/web-host.test.ts index 681f7c26..ce683086 100644 --- a/tests/web/web-host.test.ts +++ b/tests/web/web-host.test.ts @@ -590,18 +590,15 @@ test("an unbound Host exposes no bootstrap Session and rejects prompt bypasses", }); assert.equal(prompts, 0); - const cancellation = await fetch( - `${launched.origin}/api/turns/cancel`, - { - method: "POST", - headers, - body: JSON.stringify({ - sessionId: sessionManager.getSessionId(), - commandId: "command-a", - epoch: 1, - }), - }, - ); + const cancellation = await fetch(`${launched.origin}/api/turns/cancel`, { + method: "POST", + headers, + body: JSON.stringify({ + sessionId: sessionManager.getSessionId(), + commandId: "command-a", + epoch: 1, + }), + }); assert.equal(cancellation.status, 409); assert.equal((await cancellation.json()).code, "WORKSPACE_REQUIRED"); From c4305ad039b3cff9b073c77eeed9a9ff5a4ae087 Mon Sep 17 00:00:00 2001 From: testikun <320479488+testikun@users.noreply.github.com> Date: Sat, 5 Sep 2026 10:37:44 +0800 Subject: [PATCH 3/6] fix(web): bound cancellation settlement wait --- web/runtime/pi-runtime.ts | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/web/runtime/pi-runtime.ts b/web/runtime/pi-runtime.ts index 2df53879..c7c3e1be 100644 --- a/web/runtime/pi-runtime.ts +++ b/web/runtime/pi-runtime.ts @@ -38,6 +38,7 @@ import { } from "./web-host-lease.ts"; const STARTUP_TIMEOUT_MS = 15_000; +const TURN_CANCELLATION_SETTLEMENT_TIMEOUT_MS = 10_000; const BOOTSTRAP_WORKSPACE_DIRECTORY = ".bootstrap-workspace"; type PromptTrace = { @@ -140,6 +141,7 @@ export class PiWebRuntime implements WebRuntimeController { const webSessionDirectory = join(getAgentDir(), "web-sessions"); const webHostLease = await acquireWebHostLease(webSessionDirectory); let runtime: PiWebRuntime | undefined; + let timeoutHandle: ReturnType | undefined; try { const created = await PiWebRuntime.createRuntime( canonicalCwd, @@ -229,12 +231,28 @@ export class PiWebRuntime implements WebRuntimeController { waiters.add(resolveSettlement); this.turnSettlementWaiters.set(key, waiters); }); + let timeoutHandle: ReturnType | undefined; try { const abortOperation = this.runtime.session.abort(); const abortFailure = new Promise((_, reject) => { void abortOperation.catch(reject); }); - const terminal = await Promise.race([settlement, abortFailure]); + const settlementTimeout = new Promise((_, reject) => { + timeoutHandle = setTimeout( + () => + reject( + new Error( + "Cancellation did not settle within the bounded wait window", + ), + ), + TURN_CANCELLATION_SETTLEMENT_TIMEOUT_MS, + ); + }); + const terminal = await Promise.race([ + settlement, + abortFailure, + settlementTimeout, + ]); return { ...options, state: @@ -250,6 +268,7 @@ export class PiWebRuntime implements WebRuntimeController { } catch (error) { return { ...options, state: "failed", error: errorText(error) }; } finally { + if (timeoutHandle) clearTimeout(timeoutHandle); const waiters = this.turnSettlementWaiters.get(key); if (waiters && ownWaiter) { waiters.delete(ownWaiter); From 2f4192e22f8c3c9b08af656023cab531960b0822 Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Sun, 6 Sep 2026 01:27:42 +0800 Subject: [PATCH 4/6] fix(web): settle cancellation on Pi execution completion --- docs/development/OPENPI_WEB_DEVELOPMENT.md | 4 +- tests/web/app-render.test.ts | 70 +++++++++ tests/web/pi-runtime.test.ts | 161 +++++++++++++++++++-- web/runtime/pi-runtime.ts | 62 +++++--- web/ui/app.js | 111 ++++++++++++-- 5 files changed, 368 insertions(+), 40 deletions(-) diff --git a/docs/development/OPENPI_WEB_DEVELOPMENT.md b/docs/development/OPENPI_WEB_DEVELOPMENT.md index 8a2f71f6..c8a48eeb 100644 --- a/docs/development/OPENPI_WEB_DEVELOPMENT.md +++ b/docs/development/OPENPI_WEB_DEVELOPMENT.md @@ -53,9 +53,9 @@ bun run dev:web -- /absolute/path/to/workspace ## 活动回合取消协议 -Web 的 Stop 只取消当前活动的 provider 回合，不等同于停止 Host，也不会清空已经排队的 follow-up。取消请求必须回传当前快照或 `turn_started` 事件给出的 `sessionId`、`commandId` 和 `epoch`；Runtime 在自己的串行 mutation 边界内重新核对三者，再调用 Pi 原生 `AgentSession.abort()`。 +Web 的 Stop 请求 Pi 停止当前 agent execution，不等同于停止 Host，也不会清空已经排队的 follow-up。取消请求必须回传当前快照或 `turn_started` 事件给出的 `sessionId`、`commandId` 和 `epoch`；Runtime 在自己的串行 mutation 边界内重新核对三者，再调用 Pi 原生 `AgentSession.abort()`。这不是“一条输入一个回合”的额外队列：Pi 可以在同一次 execution 中继续处理已排队的 follow-up；只有下一次真实 `agent_start` 才会取得新的 Stop identity。 -Host 返回 `accepted`、`already-settled`、`stale-session`、`stale-turn` 或 `failed` 的明确收据。浏览器不会因点击按钮而乐观结束运行态；只有 Pi 消息的 `stopReason: "aborted"` 投影成 `turn_settled(outcome: "cancelled")` 后才显示取消终态。活动回合身份也包含在快照中，因此刷新和 SSE 重连仍能恢复正确的 Stop 控件。多客户端的旧请求不能取消更新的回合，重复请求则按已终结回合幂等返回。 +Host 返回 `accepted`、`already-settled`、`stale-session`、`stale-turn` 或 `failed` 的明确收据。浏览器不会因点击按钮而乐观结束运行态；只有 Pi 的完整 execution 发出 `agent_settled`，并且其中有被 Stop 目标对应的 assistant 结果 `stopReason: "aborted"`，才投影为 `turn_settled(outcome: "cancelled")`。这个 outcome 只描述被请求停止的 provider 结果，不概括同一次 execution 中 Pi 随后处理的 follow-up 是否成功。单条 `message_end` 只提供结果证据，不能单独结束 execution；若 Pi settled 时没有终态 assistant 证据，Runtime 投影 `uncertain` 并返回 `failed`，不会猜测取消成功。活动回合身份也包含在快照中，因此刷新和 SSE 重连仍能恢复正确的 Stop 控件。多客户端的旧请求不能取消更新的回合，重复请求则按已终结回合幂等返回。 这个边界源自 [Issue #342](https://github.com/openpi-dev/openpi/issues/342)。Host disposal 仍由独立生命周期处理；全局暂停属于其他设计范围。 diff --git a/tests/web/app-render.test.ts b/tests/web/app-render.test.ts index 46848da0..e74a795c 100644 --- a/tests/web/app-render.test.ts +++ b/tests/web/app-render.test.ts @@ -401,6 +401,10 @@ async function renderApp( "sendPrompt", context as vm.Context, ) as () => Promise, + cancelActiveTurn: vm.runInContext( + "cancelActiveTurn", + context as vm.Context, + ) as () => Promise, updateComposer: vm.runInContext( "updateComposer", context as vm.Context, @@ -867,6 +871,72 @@ test("app.js settles an admitted prompt that Pi handles without an agent turn", assert.equal((app.state.terminalPromptIds as Set).size, 32); }); +test("app.js stops only the canonical active turn without optimistic settlement", async () => { + const app = await renderApp(); + const cancellation = deferred>(); + app.context.fetch = async (url: unknown) => { + if (String(url) === "/api/turns/cancel") return cancellation.promise; + if (String(url).startsWith("/api/snapshot")) return response(SNAPSHOT); + throw new Error(`unexpected request: ${String(url)}`); + }; + vm.runInContext( + 'applyRuntimeEvent({sequence: 2, type: "turn_started", detail: {sessionId: "s1", commandId: "c1", epoch: 4}})', + app.context as vm.Context, + ); + + assert.equal(app.state.liveRunning, true); + assert.equal(app.elements.get("stop-turn")?.hidden, false); + assert.equal(app.elements.get("send-prompt")?.hidden, true); + const stopping = app.cancelActiveTurn(); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(app.state.liveRunning, true); + assert.equal(app.state.turnCancellationPending, true); + + vm.runInContext( + 'applyRuntimeEvent({sequence: 3, type: "turn_settled", detail: {sessionId: "s1", commandId: "c1", epoch: 4, outcome: "cancelled"}})', + app.context as vm.Context, + ); + cancellation.resolve( + response({ + sessionId: "s1", + commandId: "c1", + epoch: 4, + state: "accepted", + accepted: true, + }), + ); + await stopping; + + assert.equal(app.state.liveRunning, false); + assert.equal(app.state.activeTurn, null); + assert.equal(app.elements.get("stop-turn")?.hidden, true); + assert.equal(app.elements.get("send-prompt")?.hidden, false); + assert.equal( + app.elements.get("composer-hint")?.textContent, + "Current turn stopped.", + ); +}); + +test("app.js restores the active turn and Stop control from a snapshot", async () => { + const running = structuredClone(SNAPSHOT) as SnapshotFixture & { + runtime: typeof SNAPSHOT.runtime & { + activeTurn: { sessionId: string; commandId: string; epoch: number }; + }; + }; + running.runtime.status = "running"; + running.runtime.activeTurn = { + sessionId: "s1", + commandId: "c1", + epoch: 9, + }; + const app = await renderApp({ snapshot: running }); + + assert.deepEqual(app.state.activeTurn, running.runtime.activeTurn); + assert.equal(app.state.liveRunning, true); + assert.equal(app.elements.get("stop-turn")?.hidden, false); + assert.equal(app.elements.get("send-prompt")?.hidden, true); +}); + test("app.js keeps an active agent running when a handled prompt settles", async () => { const app = await renderApp(); vm.runInContext( diff --git a/tests/web/pi-runtime.test.ts b/tests/web/pi-runtime.test.ts index 8dee27f1..b7ef155b 100644 --- a/tests/web/pi-runtime.test.ts +++ b/tests/web/pi-runtime.test.ts @@ -19,7 +19,7 @@ type Trace = { queued: boolean; userMessageObserved?: boolean; epoch?: number; - outcome?: "completed" | "cancelled" | "failed"; + outcome?: "completed" | "cancelled" | "failed" | "uncertain"; }; type RuntimeHarness = { @@ -32,6 +32,7 @@ type RuntimeHarness = { nextTurnEpoch: number; terminalTurnKeys: Set; turnSettlementWaiters: Map void>>; + turnAbortOperations: Map>; }; function deferred() { @@ -94,6 +95,7 @@ type PromptRuntimeHarness = { nextTurnEpoch: number; terminalTurnKeys: Set; turnSettlementWaiters: Map void>>; + turnAbortOperations: Map>; controllerMutation: Promise; disposed: boolean; hasSelectedWorkspace: boolean; @@ -142,6 +144,7 @@ type LifecycleHarness = { candidateRuntimes: Set; pendingPromptTraces: Trace[]; activePromptTrace?: Trace; + turnAbortOperations: Map>; liveMessageSequence: number; disposed: boolean; dispatcherLease: { release: () => Promise }; @@ -210,6 +213,7 @@ function lifecycleHarness(runtime: LifecycleRuntime) { harness.runtimeOperations = new Set(); harness.candidateRuntimes = new Set(); harness.pendingPromptTraces = []; + harness.turnAbortOperations = new Map(); harness.liveMessageSequence = 0; harness.disposed = false; harness.dispatcherLease = { release: async () => undefined }; @@ -252,6 +256,7 @@ function promptHarness(session: ReturnType) { harness.nextTurnEpoch = 0; harness.terminalTurnKeys = new Set(); harness.turnSettlementWaiters = new Map(); + harness.turnAbortOperations = new Map(); harness.controllerMutation = Promise.resolve(); harness.disposed = false; harness.hasSelectedWorkspace = true; @@ -572,7 +577,7 @@ test("later prompt failures retain their command and Session correlation", async }); }); -test("turn cancellation is bound, canonical, and idempotent", async () => { +test("turn cancellation reports uncertainty without assistant terminal evidence", async () => { const session = promptSession("session-a"); let aborts = 0; session.abort = async () => { @@ -586,14 +591,17 @@ test("turn cancellation is bound, canonical, and idempotent", async () => { started: true, queued: false, epoch: 7, - outcome: "cancelled", }; runtime.activePromptTrace = trace; - const settlePromptTrace = ( + const projectEvent = ( PiWebRuntime.prototype as unknown as { - settlePromptTrace(this: PromptRuntimeHarness, trace: Trace): void; + projectEvent( + this: PromptRuntimeHarness, + session: object, + event: { type: string }, + ): void; } - ).settlePromptTrace; + ).projectEvent; const cancellation = runtime.cancelTurn({ sessionId: "session-a", @@ -601,13 +609,15 @@ test("turn cancellation is bound, canonical, and idempotent", async () => { epoch: 7, }); await Promise.resolve(); - settlePromptTrace.call(runtime, trace); + projectEvent.call(runtime, session, { type: "agent_settled" }); assert.deepEqual(await cancellation, { sessionId: "session-a", commandId: "command-a", epoch: 7, - state: "accepted", + state: "failed", + error: + "Pi settled without a terminal assistant outcome for this cancellation", }); assert.equal(aborts, 1); assert.equal( @@ -643,6 +653,81 @@ test("turn cancellation is bound, canonical, and idempotent", async () => { assert.equal(aborts, 1); }); +test("turn cancellation loses to a naturally completed terminal run", async () => { + const session = promptSession("session-a"); + let aborts = 0; + session.abort = async () => { + aborts += 1; + }; + const runtime = promptHarness(session); + runtime.activePromptTrace = { + commandId: "command-a", + sessionId: "session-a", + startedAt: 1, + started: true, + queued: false, + epoch: 1, + outcome: "completed", + }; + const projectEvent = ( + PiWebRuntime.prototype as unknown as { + projectEvent( + this: PromptRuntimeHarness, + session: object, + event: { type: string }, + ): void; + } + ).projectEvent; + + const cancellation = runtime.cancelTurn({ + sessionId: "session-a", + commandId: "command-a", + epoch: 1, + }); + await Promise.resolve(); + projectEvent.call(runtime, session, { type: "agent_settled" }); + + assert.equal((await cancellation).state, "already-settled"); + assert.equal(aborts, 1); +}); + +test("a repeated cancellation does not issue another native abort while settling", async () => { + const session = promptSession("session-a"); + let aborts = 0; + session.abort = async () => { + aborts += 1; + }; + const runtime = promptHarness(session); + runtime.activePromptTrace = { + commandId: "command-a", + sessionId: "session-a", + startedAt: 1, + started: true, + queued: false, + epoch: 1, + }; + runtime.turnAbortOperations.set( + "session-a\u0000command-a\u00001", + new Promise(() => undefined), + ); + + assert.deepEqual( + await runtime.cancelTurn({ + sessionId: "session-a", + commandId: "command-a", + epoch: 1, + }), + { + sessionId: "session-a", + commandId: "command-a", + epoch: 1, + state: "failed", + error: "Cancellation is already waiting for Pi to settle this turn", + }, + ); + assert.equal(aborts, 0); +}); + test("turn cancellation reports native abort failures", async () => { const session = promptSession("session-a"); session.abort = async () => { @@ -1027,7 +1112,7 @@ test("runtime creation failure releases the Web Host lease", async () => { } }); -test("prompt traces advance with queued user messages", () => { +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 }; @@ -1037,6 +1122,7 @@ test("prompt traces advance with queued user messages", () => { harness.nextTurnEpoch = 0; harness.terminalTurnKeys = new Set(); harness.turnSettlementWaiters = new Map(); + harness.turnAbortOperations = new Map(); const events: WebRuntimeEvent[] = []; harness.listeners.add((event) => events.push(event)); harness.activePromptTrace = { @@ -1093,11 +1179,62 @@ test("prompt traces advance with queued user messages", () => { }, }); + projectEvent.call(harness, session, { + type: "message_end", + message: { + role: "assistant", + content: [], + stopReason: "stop", + timestamp: 4, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + }, + }); + projectEvent.call(harness, session, userMessage("second")); - assert.equal(harness.activePromptTrace?.commandId, "second"); + assert.equal(harness.activePromptTrace?.commandId, "first"); assert.equal(harness.activePromptTrace?.started, true); - assert.equal(harness.activePromptTrace?.epoch, 2); + assert.equal(harness.activePromptTrace?.epoch, 1); + assert.equal(harness.pendingPromptTraces.length, 1); + assert.deepEqual( + events.filter((event) => event.type === "turn_settled"), + [], + ); + + projectEvent.call(harness, session, { type: "agent_settled" }); + assert.equal(harness.activePromptTrace, undefined); assert.equal(harness.pendingPromptTraces.length, 0); + + harness.activePromptTrace = { + commandId: "third", + sessionId: "session", + startedAt: 5, + started: false, + queued: false, + }; + projectEvent.call(harness, session, { type: "agent_start" }); + projectEvent.call(harness, session, userMessage("third")); + assert.deepEqual(harness.activePromptTrace, { + commandId: "third", + sessionId: "session", + startedAt: 5, + started: true, + queued: false, + userMessageObserved: true, + epoch: 2, + }); assert.deepEqual( events .filter((event) => event.type.startsWith("turn_")) @@ -1118,7 +1255,7 @@ test("prompt traces advance with queued user messages", () => { }, { type: "turn_started", - detail: { sessionId: "session", commandId: "second", epoch: 2 }, + detail: { sessionId: "session", commandId: "third", epoch: 2 }, }, ], ); diff --git a/web/runtime/pi-runtime.ts b/web/runtime/pi-runtime.ts index c7c3e1be..457ab6a3 100644 --- a/web/runtime/pi-runtime.ts +++ b/web/runtime/pi-runtime.ts @@ -49,11 +49,11 @@ type PromptTrace = { queued: boolean; userMessageObserved: boolean; epoch?: number; - outcome?: "completed" | "cancelled" | "failed"; + outcome?: "completed" | "cancelled" | "failed" | "uncertain"; }; type TurnSettlement = WebActiveTurn & { - outcome: "completed" | "cancelled" | "failed"; + outcome: "completed" | "cancelled" | "failed" | "uncertain"; }; function errorText(error: unknown) { @@ -94,6 +94,8 @@ export class PiWebRuntime implements WebRuntimeController { string, Set<(settlement: TurnSettlement) => void> >(); + /** Native aborts remain owned by Pi until its agent_settled event arrives. */ + private readonly turnAbortOperations = new Map>(); private liveMessageKey?: string; private liveMessageSequence = 0; private readonly webSessionDirectory: string; @@ -223,6 +225,13 @@ export class PiWebRuntime implements WebRuntimeController { ) { return { ...options, state: "stale-turn" }; } + if (this.turnAbortOperations.has(key)) { + return { + ...options, + state: "failed", + error: "Cancellation is already waiting for Pi to settle this turn", + }; + } let ownWaiter: ((settlement: TurnSettlement) => void) | undefined; const settlement = new Promise((resolveSettlement) => { @@ -234,6 +243,12 @@ export class PiWebRuntime implements WebRuntimeController { let timeoutHandle: ReturnType | undefined; try { const abortOperation = this.runtime.session.abort(); + this.turnAbortOperations.set(key, abortOperation); + void abortOperation.catch(() => { + if (this.turnAbortOperations.get(key) === abortOperation) { + this.turnAbortOperations.delete(key); + } + }); const abortFailure = new Promise((_, reject) => { void abortOperation.catch(reject); }); @@ -263,7 +278,12 @@ export class PiWebRuntime implements WebRuntimeController { : "failed", ...(terminal.outcome === "failed" ? { error: "The active turn failed while cancellation was requested" } - : {}), + : terminal.outcome === "uncertain" + ? { + error: + "Pi settled without a terminal assistant outcome for this cancellation", + } + : {}), }; } catch (error) { return { ...options, state: "failed", error: errorText(error) }; @@ -836,12 +856,6 @@ export class PiWebRuntime implements WebRuntimeController { if (event.type === "message_start" && event.message.role === "user") { if (!this.activePromptTrace) { this.activePromptTrace = this.pendingPromptTraces.shift(); - } else if ( - this.activePromptTrace.userMessageObserved && - this.pendingPromptTraces.length > 0 - ) { - this.settlePromptTrace(this.activePromptTrace); - this.activePromptTrace = this.pendingPromptTraces.shift(); } if (this.activePromptTrace) { this.startPromptTrace(this.activePromptTrace); @@ -893,13 +907,14 @@ export class PiWebRuntime implements WebRuntimeController { }); break; case "agent_settled": - if ( - this.activePromptTrace?.started && - this.pendingPromptTraces.length === 0 - ) { + // 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.activePromptTrace = undefined; + this.pendingPromptTraces.length = 0; this.emit(event.type, { sessionId: session.sessionManager.getSessionId(), }); @@ -925,12 +940,21 @@ export class PiWebRuntime implements WebRuntimeController { event.message.role === "assistant" && this.activePromptTrace ) { - this.activePromptTrace.outcome = + // 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" : "completed"; + // 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 (this.activePromptTrace.outcome !== "cancelled") { + this.activePromptTrace.outcome = outcome; + } } this.emit(event.type, { message: projectMessage(event.message), @@ -977,11 +1001,13 @@ export class PiWebRuntime implements WebRuntimeController { if (!activeTurn) return; const settlement: TurnSettlement = { ...activeTurn, - outcome: trace.outcome ?? "completed", + outcome: + trace.outcome ?? "uncertain", }; const key = this.turnKey(activeTurn); if (this.terminalTurnKeys.has(key)) return; this.terminalTurnKeys.add(key); + this.turnAbortOperations.delete(key); while (this.terminalTurnKeys.size > 64) { const oldest = this.terminalTurnKeys.values().next().value; if (typeof oldest === "string") this.terminalTurnKeys.delete(oldest); @@ -1001,7 +1027,8 @@ export class PiWebRuntime implements WebRuntimeController { const pendingIndex = this.pendingPromptTraces.indexOf(trace); if (pendingIndex !== -1) this.pendingPromptTraces.splice(pendingIndex, 1); if (this.activePromptTrace !== trace) return; - if (trace.started) this.settlePromptTrace(trace); + // A started trace can only be terminally projected by agent_settled. + if (trace.started) return; this.activePromptTrace = this.pendingPromptTraces.shift(); } @@ -1166,5 +1193,6 @@ export class PiWebRuntime implements WebRuntimeController { private resetPromptTraces() { this.activePromptTrace = undefined; this.pendingPromptTraces.length = 0; + this.turnAbortOperations.clear(); } } diff --git a/web/ui/app.js b/web/ui/app.js index 53f020ea..b5f76615 100644 --- a/web/ui/app.js +++ b/web/ui/app.js @@ -19,6 +19,9 @@ const state = { collapsed: readCollapsedWorkspaces(), liveMessages: [], liveRunning: false, + activeTurn: null, + turnCancellationPending: false, + turnTerminalStatus: null, promptAdmissionPending: false, promptAdmissionToken: null, promptAdmissionSequence: 0, @@ -67,6 +70,9 @@ const translations = { enterHint: "Enter to send, Shift+Enter for a new line.", activeOnlyHint: "Only the active Web session accepts messages.", acceptedHint: "Message accepted by OpenPI Web.", + stopTurn: "Stop turn", + stoppingTurn: "Stopping current turn...", + stoppedTurn: "Current turn stopped.", modelRunning: "Working...", modelPreparing: "Preparing task...", modelRetrying: "Retrying model request...", @@ -103,6 +109,9 @@ const translations = { enterHint: "按 Enter 发送，Shift+Enter 换行。", activeOnlyHint: "只有当前 Web 会话可以接收消息。", acceptedHint: "OpenPI Web 已接收消息。", + stopTurn: "停止当前回合", + stoppingTurn: "正在停止当前回合...", + stoppedTurn: "当前回合已停止。", modelRunning: "正在运行...", modelPreparing: "正在准备任务...", modelRetrying: "模型请求重试中...", @@ -494,6 +503,13 @@ function updateComposer() { !selected && !state.snapshot?.currentSessionId; const canCompose = active || newSessionDraft; + const activeTurn = active + ? state.activeTurn || state.snapshot?.runtime.activeTurn || null + : null; + const canStop = Boolean( + activeTurn && + (state.snapshot?.runtime.status === "running" || state.liveRunning), + ); $("prompt-input").disabled = state.sessionSwitching || (!canCompose && Boolean(state.selectedWorkspace)); $("send-prompt").disabled = @@ -501,6 +517,9 @@ function updateComposer() { !canCompose || !state.selectedWorkspace || state.promptAdmissionPending; + $("send-prompt").hidden = canStop; + $("stop-turn").hidden = !canStop; + $("stop-turn").disabled = state.turnCancellationPending; const modelPicker = $("model-picker"); const modelPickerValue = $("model-picker-value"); const modelMenu = $("model-menu"); @@ -539,11 +558,16 @@ function updateComposer() { : active ? t("promptMessage") : t("promptReadonly"); - $("composer-hint").textContent = canCompose - ? state.snapshot.runtime.status === "running" || state.liveRunning - ? t("queuedHint") - : t("enterHint") - : t("activeOnlyHint"); + $("composer-hint").textContent = + state.turnCancellationPending + ? t("stoppingTurn") + : state.turnTerminalStatus === "cancelled" + ? t("stoppedTurn") + : canCompose + ? state.snapshot.runtime.status === "running" || state.liveRunning + ? t("queuedHint") + : t("enterHint") + : t("activeOnlyHint"); } async function selectModel(value) { @@ -617,6 +641,9 @@ async function refreshSnapshot({ return false; } state.snapshot = snapshot; + if (resetCursor) resetLiveState(); + state.activeTurn = snapshot.runtime.activeTurn || null; + if (snapshot.runtime.status === "running") state.liveRunning = true; if ( state.snapshot.runtime.status !== "running" && !state.promptAdmissionPending @@ -625,7 +652,6 @@ async function refreshSnapshot({ state.livePhase = "idle"; state.liveRetry = null; } - if (resetCursor) resetLiveState(); state.cursor = resetCursor || state.cursor === null ? state.snapshot.cursor : Math.max(state.cursor, state.snapshot.cursor); @@ -750,6 +776,7 @@ async function sendPrompt() { ].slice(-8); state.promptAdmissionPending = true; state.promptAdmissionToken = admissionToken; + state.turnTerminalStatus = null; renderConversation(); $("composer-hint").classList.remove("error"); try { @@ -783,6 +810,36 @@ async function sendPrompt() { } } +async function cancelActiveTurn() { + const turn = state.activeTurn || state.snapshot?.runtime.activeTurn; + if (!turn || state.turnCancellationPending || state.sessionSwitching) return; + const epoch = state.sessionEpoch; + state.turnCancellationPending = true; + $("composer-hint").classList.remove("error"); + $("composer-hint").textContent = t("stoppingTurn"); + updateComposer(); + try { + const receipt = await api("/api/turns/cancel", { + method: "POST", + body: JSON.stringify(turn), + }); + if (epoch !== state.sessionEpoch) return; + if (receipt.state === "accepted" || receipt.state === "already-settled") { + $("composer-hint").textContent = t("stoppedTurn"); + } + } catch (error) { + if (epoch !== state.sessionEpoch) return; + const message = error.message; + await refreshSnapshot({ epoch }); + if (epoch === state.sessionEpoch) showNotice(message); + } finally { + if (epoch === state.sessionEpoch) { + state.turnCancellationPending = false; + renderConversation(); + } + } +} + function resizePrompt() { const input = $("prompt-input"); const maxHeight = 220; @@ -1080,15 +1137,43 @@ function applyRuntimeEvent(event) { applyPromptAcceptedState(alreadySettled); state.liveRetry = null; renderConversation(); + } else if (event.type === "turn_started") { + state.activeTurn = { + sessionId: event.detail?.sessionId, + commandId: event.detail?.commandId, + epoch: event.detail?.epoch, + }; + state.liveRunning = true; + state.turnTerminalStatus = null; + state.livePhase = "running"; + state.liveRetry = null; + renderConversation(); } else if (event.type === "agent_start") { + if (event.detail?.activeTurn) state.activeTurn = event.detail.activeTurn; state.liveRunning = true; state.livePhase = "running"; state.liveRetry = null; renderConversation(); + } else if (event.type === "turn_settled") { + rememberTerminalPrompt(event.detail?.commandId); + const isActiveTurn = + state.activeTurn?.sessionId === event.detail?.sessionId && + state.activeTurn?.commandId === event.detail?.commandId && + state.activeTurn?.epoch === event.detail?.epoch; + if (isActiveTurn) { + state.activeTurn = null; + state.liveRunning = false; + state.livePhase = "idle"; + state.liveRetry = null; + state.turnTerminalStatus = event.detail?.outcome || null; + } + renderConversation(); } else if (event.type === "agent_settled") { - state.liveRunning = false; - state.livePhase = "idle"; - state.liveRetry = null; + if (!state.activeTurn) { + state.liveRunning = false; + state.livePhase = "idle"; + state.liveRetry = null; + } renderConversation(); } else if (event.type === "prompt_settled") { rememberTerminalPrompt(event.detail?.commandId); @@ -1117,6 +1202,8 @@ function applyRuntimeEvent(event) { [ "agent_start", "agent_settled", + "turn_started", + "turn_settled", "prompt_settled", "message_end", "tool_execution_end", @@ -1180,6 +1267,9 @@ let eventLoopStarted = false; function resetLiveState() { state.liveMessages = []; state.liveRunning = false; + state.activeTurn = null; + state.turnCancellationPending = false; + state.turnTerminalStatus = null; state.livePhase = "idle"; state.liveRetry = null; } @@ -1401,6 +1491,9 @@ $("composer")?.addEventListener("submit", (event) => { if (state.selectedWorkspace) void sendPrompt(); else void chooseWorkspace(); }); +$("stop-turn")?.addEventListener("click", () => { + void cancelActiveTurn(); +}); $("prompt-input")?.addEventListener("input", resizePrompt); $("prompt-input")?.addEventListener("keydown", (event) => { if (event.isComposing || event.keyCode === 229) return; From d1360db430c1aee1bc7b127747c316437b4f3973 Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Sun, 6 Sep 2026 01:31:11 +0800 Subject: [PATCH 5/6] fix(web): classify tool-use responses as uncertain --- tests/web/pi-runtime.test.ts | 68 ++++++++++++++++++++++++++++++++++++ web/runtime/pi-runtime.ts | 7 ++-- 2 files changed, 73 insertions(+), 2 deletions(-) diff --git a/tests/web/pi-runtime.test.ts b/tests/web/pi-runtime.test.ts index b7ef155b..def9743b 100644 --- a/tests/web/pi-runtime.test.ts +++ b/tests/web/pi-runtime.test.ts @@ -1260,3 +1260,71 @@ test("message_end and queued prompts do not settle a running turn", () => { ], ); }); + +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.pendingPromptTraces = []; + harness.liveMessageSequence = 0; + harness.listeners = new Set(); + harness.nextTurnEpoch = 0; + harness.terminalTurnKeys = new Set(); + harness.turnSettlementWaiters = new Map(); + harness.turnAbortOperations = new Map(); + const events: WebRuntimeEvent[] = []; + harness.listeners.add((event) => events.push(event)); + harness.activePromptTrace = { + commandId: "tool-use", + sessionId: "session", + startedAt: 1, + started: false, + queued: false, + }; + + const projectEvent = ( + PiWebRuntime.prototype as unknown as { + projectEvent(this: RuntimeHarness, session: object, event: object): void; + } + ).projectEvent; + + projectEvent.call(harness, session, { type: "agent_start" }); + projectEvent.call(harness, session, { + type: "message_end", + message: { + role: "assistant", + content: [], + stopReason: "toolUse", + timestamp: 2, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + }, + }); + projectEvent.call(harness, session, { type: "agent_settled" }); + + assert.deepEqual( + events + .filter((event) => event.type === "turn_settled") + .map((event) => event.detail), + [ + { + sessionId: "session", + commandId: "tool-use", + epoch: 1, + outcome: "uncertain", + }, + ], + ); +}); diff --git a/web/runtime/pi-runtime.ts b/web/runtime/pi-runtime.ts index 457ab6a3..b1cb8293 100644 --- a/web/runtime/pi-runtime.ts +++ b/web/runtime/pi-runtime.ts @@ -947,12 +947,15 @@ export class PiWebRuntime implements WebRuntimeController { ? "cancelled" : event.message.stopReason === "error" ? "failed" - : "completed"; + : 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 (this.activePromptTrace.outcome !== "cancelled") { + if (outcome && this.activePromptTrace.outcome !== "cancelled") { this.activePromptTrace.outcome = outcome; } } From 86d9a6e666b12d004f2d93e106140b8569e1d4be Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Sun, 6 Sep 2026 01:33:37 +0800 Subject: [PATCH 6/6] docs(web): clarify stop identity lifetime --- docs/development/OPENPI_WEB_DEVELOPMENT.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/development/OPENPI_WEB_DEVELOPMENT.md b/docs/development/OPENPI_WEB_DEVELOPMENT.md index c8a48eeb..2ea11ab0 100644 --- a/docs/development/OPENPI_WEB_DEVELOPMENT.md +++ b/docs/development/OPENPI_WEB_DEVELOPMENT.md @@ -53,7 +53,7 @@ bun run dev:web -- /absolute/path/to/workspace ## 活动回合取消协议 -Web 的 Stop 请求 Pi 停止当前 agent execution，不等同于停止 Host，也不会清空已经排队的 follow-up。取消请求必须回传当前快照或 `turn_started` 事件给出的 `sessionId`、`commandId` 和 `epoch`；Runtime 在自己的串行 mutation 边界内重新核对三者，再调用 Pi 原生 `AgentSession.abort()`。这不是“一条输入一个回合”的额外队列：Pi 可以在同一次 execution 中继续处理已排队的 follow-up；只有下一次真实 `agent_start` 才会取得新的 Stop identity。 +Web 的 Stop 请求 Pi 停止当前 agent execution，不等同于停止 Host，也不会清空已经排队的 follow-up。取消请求必须回传当前快照或 `turn_started` 事件给出的 `sessionId`、`commandId` 和 `epoch`；Runtime 在自己的串行 mutation 边界内重新核对三者，再调用 Pi 原生 `AgentSession.abort()`。这不是“一条输入一个回合”的额外队列：Pi 可以在同一次 execution 中继续处理已排队的 follow-up；当前 execution 终结后，下一次 execution 的真实 `agent_start` 才会取得新的 Stop identity；同一 execution 内的 retry 或 continue 保留原 identity。 Host 返回 `accepted`、`already-settled`、`stale-session`、`stale-turn` 或 `failed` 的明确收据。浏览器不会因点击按钮而乐观结束运行态；只有 Pi 的完整 execution 发出 `agent_settled`，并且其中有被 Stop 目标对应的 assistant 结果 `stopReason: "aborted"`，才投影为 `turn_settled(outcome: "cancelled")`。这个 outcome 只描述被请求停止的 provider 结果，不概括同一次 execution 中 Pi 随后处理的 follow-up 是否成功。单条 `message_end` 只提供结果证据，不能单独结束 execution；若 Pi settled 时没有终态 assistant 证据，Runtime 投影 `uncertain` 并返回 `failed`，不会猜测取消成功。活动回合身份也包含在快照中，因此刷新和 SSE 重连仍能恢复正确的 Stop 控件。多客户端的旧请求不能取消更新的回合，重复请求则按已终结回合幂等返回。