feat(web): add turn-scoped cancellation

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`。
 
 ## 对话无响应排查

tests/web/app-render.test.ts

@@ -401,6 +401,10 @@ async function renderApp(
       "sendPrompt",
       context as vm.Context,
     ) as () => Promise<void>,
+    cancelActiveTurn: vm.runInContext(
+      "cancelActiveTurn",
+      context as vm.Context,
+    ) as () => Promise<void>,
     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<string>).size, 32);
 });
 
+test("app.js stops only the canonical active turn without optimistic settlement", async () => {
+  const app = await renderApp();
+  const cancellation = deferred<ReturnType<typeof response>>();
+  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<ReturnType<typeof response>>();

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 }),

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<string>;
+  turnSettlementWaiters: Map<string, Set<(settlement: unknown) => void>>;
 };
 
 function deferred() {
@@ -84,11 +90,17 @@ type PromptRuntimeHarness = {
   runtimeDisposalPromises: WeakMap<FakeAgentRuntime, Promise<void>>;
   promptAdmission: Promise<void>;
   pendingPromptTraces: Trace[];
+  activePromptTrace?: Trace;
+  nextTurnEpoch: number;
+  terminalTurnKeys: Set<string>;
+  turnSettlementWaiters: Map<string, Set<(settlement: unknown) => void>>;
+  controllerMutation: Promise<void>;
   disposed: boolean;
   hasSelectedWorkspace: boolean;
   dispatcherLease: { release: () => Promise<void> };
   webHostLease: { release: () => Promise<void> };
   sendPrompt: PiWebRuntime["sendPrompt"];
+  cancelTurn: PiWebRuntime["cancelTurn"];
   subscribe: PiWebRuntime["subscribe"];
   dispose: PiWebRuntime["dispose"];
 };
@@ -237,6 +249,10 @@ function promptHarness(session: ReturnType<typeof promptSession>) {
   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 },
+      },
+    ],
+  );
 });