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..ce683086 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,18 @@ 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 +627,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 +681,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 +753,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..c7c3e1be 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"; @@ -35,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 = { @@ -43,6 +47,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 +88,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; @@ -124,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, @@ -174,6 +192,91 @@ 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); + }); + let timeoutHandle: ReturnType | undefined; + try { + const abortOperation = this.runtime.session.abort(); + const abortFailure = new Promise((_, reject) => { + void abortOperation.catch(reject); + }); + 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: + 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 { + if (timeoutHandle) clearTimeout(timeoutHandle); + 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 +398,7 @@ export class PiWebRuntime implements WebRuntimeController { startedAt, started: false, queued, + userMessageObserved: false, } : undefined; this.retainRuntimeReference(agentRuntime); @@ -432,6 +536,7 @@ export class PiWebRuntime implements WebRuntimeController { }); } if (promptTrace) { + promptTrace.outcome = "failed"; traceWeb("prompt_operation_failed", { commandId: promptTrace.commandId, sessionId, @@ -725,13 +830,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 +885,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 +920,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 +955,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; }