From 160ca1b9ba357acd1a9a257e4ed740ba98358ec6 Mon Sep 17 00:00:00 2001
From: Tosd0 <65720409+Tosd0@users.noreply.github.com>
Date: Thu, 1 Oct 2026 01:21:01 +0900
Subject: [PATCH 1/2] feat(amsg-server): abort cancelled fire generation and
 tools

---
 .changeset/amsg-abort-active-fire.md          |   6 +
 packages/rei-standard-amsg/server/README.md   |  25 +++++
 .../server/src/server/lib/agentic-fire.js     |  52 ++++++++-
 .../src/server/lib/message-processor.js       |   4 +-
 .../server/src/server/lib/run-tick.js         |   7 +-
 .../server/test/agentic-fire.test.mjs         |   1 +
 .../server/test/run-tick.test.mjs             | 105 +++++++++++++++++-
 .../rei-standard-amsg/shared/src/llm-call.js  |   5 +-
 8 files changed, 196 insertions(+), 9 deletions(-)
 create mode 100644 .changeset/amsg-abort-active-fire.md

diff --git a/.changeset/amsg-abort-active-fire.md b/.changeset/amsg-abort-active-fire.md
new file mode 100644
index 0000000..061bce4
--- /dev/null
+++ b/.changeset/amsg-abort-active-fire.md
@@ -0,0 +1,6 @@
+---
+"@rei-standard/amsg-shared": patch
+"@rei-standard/amsg-server": patch
+---
+
+Abort active LLM requests when task cancellation or supersession invalidates the lease. Expose signal, isCancelled and throwIfCancelled to fire hooks; stop agentic continuation and outbox delivery after cancellation, and settle cancelled fires without retries. Shared callLlm now accepts an external AbortSignal while preserving its request timeout.
diff --git a/packages/rei-standard-amsg/server/README.md b/packages/rei-standard-amsg/server/README.md
index 58feb80..f852316 100644
--- a/packages/rei-standard-amsg/server/README.md
+++ b/packages/rei-standard-amsg/server/README.md
@@ -940,3 +940,28 @@ VERCEL_PROTECTION_BYPASS=YOUR_BYPASS_KEY
 - [SW 包 README](https://github.com/Tosd0/ReiStandard/blob/main/packages/rei-standard-amsg/sw/README.md)
 - [API 技术规范](https://github.com/Tosd0/ReiStandard/blob/main/standards/active-messaging-api.md)
 - [Service Worker 规范](https://github.com/Tosd0/ReiStandard/blob/main/standards/service-worker-specification.md)
+
+
+## 取消正在执行的生成与工具
+
+任务被 `DELETE /cancel-message` 取消或被新任务顶替后，租约心跳发现任务已经失效，会中断正在等待的模型请求，阻止新的模型轮次、工具阶段和推送。`leaseHeartbeatMs` 默认仍为 30000；需要及时停止的交互应用可配置为 1000（每个正在执行的任务每秒续租一次）。这是跨请求的数据库检测，延迟取决于心跳和数据库响应，不承诺瞬时中断。自定义适配器需要支持 `claimTask` / `renewTaskLease`；关掉心跳时也关掉了这条主动检测路径。
+
+`onBeforeFire`、`onLLMOutput`、`executeToolCalls` 的 context 以及 `onFireSettled` 回执新增：
+
+- `signal: AbortSignal`：传给工具内部的 `fetch`，与工具自己的超时信号合并。
+- `isCancelled(): boolean`：当前是否已收到取消信号。
+- `throwIfCancelled(): void`：已取消时抛出 `code: 'TASK_CANCELLED'` 的错误。每个副作用开始前检查，工具 catch 中也要先调用它，以免取消被转成普通工具失败后继续执行。
+
+```js
+async function executeToolCalls(calls, ctx) {
+  const results = [];
+  for (const call of calls) {
+    ctx.throwIfCancelled();
+    const response = await fetch(toolUrl(call), { signal: ctx.signal });
+    results.push({ tool_call_id: call.id, role: 'tool', content: await response.text() });
+  }
+  return results;
+}
+```
+
+取消收尾使用 `onFireSettled({ status: 'cancelled', willRetry: false, error, ... })`，不是生成失败。应用应在这里释放资源、结算已发生的用量，避免发出失败提示或把未送达内容记为已回复。取消时已经开始写入的 outbox 批次会在写完后检查信号并撤掉未投递部分；已经发到客户端的消息仍由客户端处理。应用需要记录停止的任务身份，拦住迟到推送、补收以及尚未显示的本地内容。已完成的外部副作用无法撤销；不支持 AbortSignal 的工具也需要自己在执行步骤之间检查取消。
diff --git a/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js b/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js
index 28d3e99..eed7848 100644
--- a/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js
+++ b/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js
@@ -309,6 +309,19 @@ function firstPositiveNumber(values, fallback) {
  */
 export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
   const hooks = ctx.hooks;
+  // One cancellation identity flows through fetch, hooks and settlement.
+  const signal = ctx.signal || new AbortController().signal;
+  const isCancelled = () => signal.aborted || ctx.isTaskCancelled?.() === true;
+  const throwIfCancelled = () => {
+    if (!isCancelled()) return;
+    if (isTaskCancelledError(signal.reason)) throw signal.reason;
+    const error = new Error('Task was cancelled or superseded');
+    error.code = 'TASK_CANCELLED';
+    throw error;
+  };
+  const cancellation = { signal, isCancelled, throwIfCancelled };
+  ctx = { ...ctx, ...cancellation };
+
   if (typeof hooks.onLLMOutput !== 'function') {
     throw new DeploymentConfigError(
       'AGENTIC_CONFIG_ERROR: hooks.onBeforeFire requires hooks.onLLMOutput to classify LLM rounds',
@@ -355,7 +368,7 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
     now: nowFn,
     // run-tick 在投递 ctx 上挂的取消信号（与 guardWebpushWithLease 读的是同一
     // 个租约状态），emitResult 不发推送的那条路要靠它拦下已取消任务的落行。
-    isCancelled: typeof ctx.isTaskCancelled === 'function' ? ctx.isTaskCancelled : null,
+    isCancelled,
   });
 
   const maxScheduledTasksPerFire =
@@ -415,6 +428,7 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
    *   认领。行读不回来（已经不是 pending，或适配器没有 getTaskByUuid）→ `null`。
    */
   const scheduleTask = async (options) => {
+    throwIfCancelled();
     if (!options || typeof options !== 'object' || Array.isArray(options)) {
       throw new TypeError('scheduleTask(options) 需要一个对象，至少包含 { firstSendTime }');
     }
@@ -570,6 +584,7 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
     scheduledTaskCount++;
 
     const encryptedPayload = await encryptForStorage(serializedTaskData, userKey);
+    throwIfCancelled();
 
     let created;
     try {
@@ -626,6 +641,7 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
    *   （已发出 / 已删除）——对「用户要它别响」来说结果已达成，不算错误。
    */
   const cancelTask = async (uuid) => {
+    throwIfCancelled();
     if (typeof uuid !== 'string' || !uuid.trim()) {
       throw new TypeError('cancelTask(uuid) 需要非空字符串 uuid');
     }
@@ -641,6 +657,7 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
         { code: 'AGENTIC_CANCEL_UNSUPPORTED' }
       );
     }
+    throwIfCancelled();
     const cancelled = await ctx.db.deleteTaskByUuid(uuid, task.user_id);
     if (cancelled) {
       // 与 DELETE /cancel-message 同一收尾：那条任务此前投递到一半失败过的话，
@@ -668,6 +685,7 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
    *   在路上了，通常没什么好补救的；真要顺延下一次，等这次发完再调一遍。
    */
   const renewTask = async (uuid, nextSendAt) => {
+    throwIfCancelled();
     if (typeof uuid !== 'string' || !uuid.trim()) {
       throw new TypeError('renewTask(uuid, nextSendAt) 需要非空字符串 uuid');
     }
@@ -701,6 +719,7 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
     const payload = JSON.parse(await decryptFromStorage(row.encrypted_payload, userKey));
     const nextSendAtIso = sendAt.toISOString();
     const encrypted = await encryptForStorage(JSON.stringify({ ...payload, firstSendTime: nextSendAtIso }), userKey);
+    throwIfCancelled();
     const updated = await ctx.db.updateTaskByUuid(uuid, task.user_id, encrypted, {
       next_send_at: nextSendAtIso,
       retry_count: 0,
@@ -738,6 +757,7 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
   const scratch = {};
 
   const fireCtx = Object.freeze({
+    ...cancellation,
     task: buildHookTask(task, decryptedPayload),
     userId: task.user_id,
     readState,
@@ -781,18 +801,26 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
       resolveLlmCredential, fireCtx, progress,
       sessionId, messageIdBase, occurrenceMs,
     });
+    throwIfCancelled();
     settledStatus = !outcome.handled
       ? 'not-handled'
       : (outcome.result.status === 'skipped' ? 'skipped' : 'sent');
     return outcome;
   } catch (error) {
+    if (isCancelled() && !isTaskCancelledError(error)) {
+      try { throwIfCancelled(); } catch (cancelled) { error = cancelled; }
+    }
+    if (isTaskCancelledError(error)) settledStatus = 'cancelled';
     settledError = error;
     throw error;
   } finally {
     await notifyFireSettled(ctx, {
       task,
+      ...cancellation,
       status: settledStatus,
-      ...(settledStatus === 'failed' && ctx._deliveryState
+      ...(settledStatus === 'cancelled'
+        ? { willRetry: false, failureStage: null }
+        : settledStatus === 'failed' && ctx._deliveryState
         ? failureRetryDecision(ctx._deliveryState, settledError)
         : { willRetry: null, failureStage: null }),
       skipReason: settledStatus === 'skipped' ? progress.skipReason : null,
@@ -837,7 +865,9 @@ async function runFireChain({
   resolveLlmCredential, fireCtx, progress,
   sessionId, messageIdBase, occurrenceMs,
 }) {
+  ctx.throwIfCancelled();
   const before = await hooks.onBeforeFire(fireCtx);
+  ctx.throwIfCancelled();
   if (before == null) return { handled: false };
 
   // Pre-LLM skip: the host judged this fire moot before generation (e.g. the
@@ -871,6 +901,7 @@ async function runFireChain({
   });
 
   for (let iteration = 0; iteration < maxToolIterations; iteration++) {
+    ctx.throwIfCancelled();
     if (nowFn() >= deadline) {
       throw new Error(`AGENTIC_TOTAL_TIMEOUT: fire chain exceeded ${totalTimeoutMs}ms after ${iteration} LLM round(s)`);
     }
@@ -890,7 +921,7 @@ async function runFireChain({
         messages,
         ...(normalized.tools ? { tools: normalized.tools, toolChoice: normalized.toolChoice } : {}),
       },
-      { requireContent: false, timeoutMs: roundTimeoutMs }
+      { requireContent: false, timeoutMs: roundTimeoutMs, signal: ctx.signal }
     );
 
     const assistantMessage = extractAssistantMessage(llmResponse);
@@ -899,6 +930,7 @@ async function runFireChain({
     // 逐轮累加，是整次 fire 的花费。
     progress.usage = (llmResponse && typeof llmResponse === 'object' && llmResponse.usage) || null;
     progress.usageTotal = accumulateUsage(progress.usageTotal, progress.usage);
+    ctx.throwIfCancelled();
 
     // 共享的 SessionContext（与 amsg-instant 同形状）之上，再挂任务身份、
     // 两个状态访问器和 scheduleTask：
@@ -909,6 +941,9 @@ async function runFireChain({
     //     时才知道，而那正是 onLLMOutput / executeToolCalls 的位置，
     //     onBeforeFire 早就返回了。
     const sessionCtx = Object.freeze({
+      signal: ctx.signal,
+      isCancelled: ctx.isCancelled,
+      throwIfCancelled: ctx.throwIfCancelled,
       ...buildSessionContext({
         sessionId,
         messages,
@@ -933,6 +968,7 @@ async function runFireChain({
     });
 
     const decision = await hooks.onLLMOutput(sessionCtx);
+    ctx.throwIfCancelled();
     try {
       assertValidDecision(decision, { inlineToolCalls: true });
     } catch (error) {
@@ -996,11 +1032,15 @@ async function runFireChain({
 
     let toolResults;
     try {
+      ctx.throwIfCancelled();
       toolResults = await hooks.executeToolCalls(toolCalls, sessionCtx);
+      ctx.throwIfCancelled();
       if (!Array.isArray(toolResults)) {
         throw new TypeError('executeToolCalls must resolve to an array of { tool_call_id, role: "tool", content }');
       }
     } catch (error) {
+      ctx.throwIfCancelled();
+      if (isTaskCancelledError(error)) throw error;
       // Feed the failure back as tool results and let the LLM talk its way
       // out, instead of failing the whole fire.
       toolResults = toolCalls.map((toolCall) => ({
@@ -1139,7 +1179,8 @@ async function notifyAfterSend(ctx, info) {
  * 了，但记账的代码挂在发送后，这次没发成就没人记，那条任务从此只活在数据库
  * 里；以及 fire 开头拿的锁没有可靠的释放点，一次 skip 就把资源占满整个 TTL。
  *
- * status 四种：
+ * status 五种：
+ *   - `cancelled`   —— 任务被取消/顶替；error.code 为 TASK_CANCELLED，不重试
  *   - `sent`        —— pushPayloads 全部发完（sentCount === total）
  *   - `skipped`     —— 这次不发。skipReason 区分是 onBeforeFire 直接
  *                      `{ skip: true }`（`'before-fire'`），还是模型跑完之后
@@ -1261,9 +1302,11 @@ async function sendHookPushPayloads({
       stampTaskIdentity(push, task, decryptedPayload, occurrenceMs);
       finalized.push(push);
     }
+    ctx.throwIfCancelled();
     outboxed = await appendPushesToOutbox({ db: ctx.db, userId: task.user_id, userKey, pushes: finalized });
     progress.outboxed = outboxed;
     if (ctx._deliveryState) ctx._deliveryState.outboxed = outboxed;
+    ctx.throwIfCancelled();
 
     if (!ctx.vapid || !ctx.vapid.email || !ctx.vapid.publicKey || !ctx.vapid.privateKey) {
       throw new Error('VAPID configuration missing - push notifications cannot be sent');
@@ -1287,6 +1330,7 @@ async function sendHookPushPayloads({
     const lastPushIndex = willPush.lastIndexOf(true);
 
     for (let i = 0; i < total; i++) {
+      ctx.throwIfCancelled();
       if (willPush[i]) {
         await sendTaggedPush(ctx.webpush, pushSubscription, JSON.stringify(finalized[i]));
         pushedCount++;
diff --git a/packages/rei-standard-amsg/server/src/server/lib/message-processor.js b/packages/rei-standard-amsg/server/src/server/lib/message-processor.js
index 8bb80b5..065a288 100644
--- a/packages/rei-standard-amsg/server/src/server/lib/message-processor.js
+++ b/packages/rei-standard-amsg/server/src/server/lib/message-processor.js
@@ -505,7 +505,7 @@ export async function processSingleMessage(task, ctx, providedMasterKey, predecr
         // credRefs.chat 任务按引用现读凭据；解析结果只合进发给 callLlm 的这
         // 一个对象，不写回 decryptedPayload（那份会流向 hook / push）。
         const chatCred = await resolveFireCredentials({ db: ctx.db, userId: task.user_id, userKey, decryptedPayload });
-        const aiResult = await callLlm(chatCred ? { ...decryptedPayload, ...chatCred } : decryptedPayload);
+        const aiResult = await callLlm(chatCred ? { ...decryptedPayload, ...chatCred } : decryptedPayload, { signal: ctx.signal });
         messageContent = aiResult.content;
         llmResponse = aiResult.response;
       } else if (decryptedPayload.userMessage) {
@@ -516,7 +516,7 @@ export async function processSingleMessage(task, ctx, providedMasterKey, predecr
 
     } else if (decryptedPayload.messageType === 'prompted' || decryptedPayload.messageType === 'auto') {
       const chatCred = await resolveFireCredentials({ db: ctx.db, userId: task.user_id, userKey, decryptedPayload });
-      const aiResult = await callLlm(chatCred ? { ...decryptedPayload, ...chatCred } : decryptedPayload);
+      const aiResult = await callLlm(chatCred ? { ...decryptedPayload, ...chatCred } : decryptedPayload, { signal: ctx.signal });
       messageContent = aiResult.content;
       llmResponse = aiResult.response;
     } else {
diff --git a/packages/rei-standard-amsg/server/src/server/lib/run-tick.js b/packages/rei-standard-amsg/server/src/server/lib/run-tick.js
index de040e7..793cd6a 100644
--- a/packages/rei-standard-amsg/server/src/server/lib/run-tick.js
+++ b/packages/rei-standard-amsg/server/src/server/lib/run-tick.js
@@ -477,7 +477,8 @@ async function deliverTasks(ctx, tasks) {
    * lost；同理，只有明确的 false 才算行没了，什么都不返回的自定义适配器照旧。
    */
   function startLeaseHeartbeat(task) {
-    const lease = { lost: false, released: false, stop: () => {} };
+    const controller = new AbortController();
+    const lease = { lost: false, released: false, signal: controller.signal, stop: () => {} };
     if (!heartbeatEnabled) return lease;
     let stopped = false;
     let timer = null;
@@ -499,6 +500,9 @@ async function deliverTasks(ctx, tasks) {
         // 就会在 tick 日志里看到一条正常送达的消息带着取消告警。
         if (stopped || lease.released) return;
         lease.lost = true;
+        const error = new Error('Task was cancelled or superseded');
+        error.code = TASK_CANCELLED_CODE;
+        controller.abort(error);
         console.warn(`[amsg-server] 任务 ${task.id} 的租约已失效（行被取消或顶替），剩余推送将中止`);
         return;
       }
@@ -1073,6 +1077,7 @@ async function deliverTasks(ctx, tasks) {
           ...ctx, db, masterKey,
           webpush: guardWebpushWithLease(ctx.webpush, lease),
           isTaskCancelled: () => lease.lost,
+          signal: lease.signal,
         },
         masterKey,
         { userKey, payload: decryptedPayload }
diff --git a/packages/rei-standard-amsg/server/test/agentic-fire.test.mjs b/packages/rei-standard-amsg/server/test/agentic-fire.test.mjs
index 2cb7d46..3f46551 100644
--- a/packages/rei-standard-amsg/server/test/agentic-fire.test.mjs
+++ b/packages/rei-standard-amsg/server/test/agentic-fire.test.mjs
@@ -406,6 +406,7 @@ describe('agentic fire loop', () => {
         'usage', 'cancelTask', 'renewTask',
         // 按 cred_id 现读一份凭据（返回新对象，不挂在 ctx 上——这里只是方法本身）
         'resolveLlmCredential',
+        'signal', 'isCancelled', 'throwIfCancelled',
       ]);
       for (const k of Object.keys(capturedSessionCtx)) {
         assert.ok(allowedSessionKeys.has(k), `unexpected sessionCtx key: ${k}`);
diff --git a/packages/rei-standard-amsg/server/test/run-tick.test.mjs b/packages/rei-standard-amsg/server/test/run-tick.test.mjs
index aaf8f61..aa6b69f 100644
--- a/packages/rei-standard-amsg/server/test/run-tick.test.mjs
+++ b/packages/rei-standard-amsg/server/test/run-tick.test.mjs
@@ -1519,7 +1519,7 @@ test('取消撞上 emitResult(show:false)：不落行、抛 TASK_CANCELLED', asy
   });
 
   assert.equal(thrown && thrown.code, 'TASK_CANCELLED', 'show:false 不走推送，取消信号也得拦得住');
-  assert.deepEqual(res.details.cancelledTasks.map((t) => t.status), ['cancelled_after_delivery']);
+  assert.deepEqual(res.details.cancelledTasks.map((t) => t.status), ['cancelled_mid_delivery']);
   assert.deepEqual(await adapter.listUnackedOutbox(USER, 0, 50), [], '取消后的结果不能留在收件箱等补收');
 });
 
@@ -1749,3 +1749,106 @@ test('正常收尾不会被心跳误报成「任务被取消」', async () => {
     `正常收尾不该报租约失效：${JSON.stringify(warnings)}`,
   );
 });
+
+
+for (const phase of ['llm', 'before-hook', 'output-hook', 'tool', 'tool-swallow']) {
+  test(`cancellation aborts active ${phase} and never starts another round`, { timeout: 4000 }, async () => {
+    const adapter = createD1Adapter(createTestD1());
+    await adapter.initSchema();
+    await seed(adapter, { uuid: `abort-${phase}`, recurrenceType: 'none', nextSendAt: recentDue(), payload: {
+      messageType: 'auto', apiUrl: 'https://example.com/v1/chat/completions', apiKey: 'key', primaryModel: 'test',
+    } });
+    let entered;
+    const started = new Promise(resolve => { entered = resolve; });
+    let settle;
+    let calls = 0;
+    let observedSignal;
+    const originalFetch = globalThis.fetch;
+    const blockUntilAbort = signal => new Promise((resolve, reject) => {
+      observedSignal = signal;
+      entered();
+      // A bounded fallback makes missing signal support fail as an assertion,
+      // rather than leaving a request hanging until the test runner times out.
+      const timer = setTimeout(() => reject(new Error('request was not aborted')), 150);
+      signal?.addEventListener('abort', () => { clearTimeout(timer); reject(signal.reason); }, { once: true });
+    });
+    globalThis.fetch = async (_url, options) => {
+      calls++;
+      if (phase === 'llm') return blockUntilAbort(options.signal);
+      return { ok: true, json: async () => ({ choices: [{ message: { role: 'assistant', content: 'tool' } }] }) };
+    };
+    const webpush = fakeWebpush();
+    try {
+      const tick = runScheduledTick({
+        db: adapter, masterKey: MASTER_KEY, vapid: VAPID, webpush, leaseHeartbeatMs: 5,
+        hooks: {
+          onBeforeFire: async ctx => {
+            if (phase === 'before-hook') await blockUntilAbort(ctx.signal);
+            return [{ role: 'user', content: 'hello' }];
+          },
+          onLLMOutput: async ctx => {
+            if (phase === 'output-hook') await blockUntilAbort(ctx.signal);
+            return { decision: 'tool-request', toolCalls: [{ id: 'x', type: 'function', function: { name: 'wait', arguments: '{}' } }] };
+          },
+          executeToolCalls: async (_calls, ctx) => {
+            try { await blockUntilAbort(ctx.signal); }
+            catch (error) { if (phase !== 'tool-swallow') throw error; }
+            return [];
+          },
+        },
+        onFireSettled: info => { settle = info; },
+      });
+      await started;
+      await adapter.deleteTaskByUuid(`abort-${phase}`, USER);
+      const result = await tick;
+      assert.equal(observedSignal?.aborted, true, 'active operation receives task abort');
+      assert.equal(settle.status, 'cancelled');
+      assert.equal(settle.error.code, 'TASK_CANCELLED');
+      assert.equal(settle.signal.aborted, true);
+      assert.equal(settle.isCancelled(), true);
+      assert.throws(() => settle.throwIfCancelled(), { code: 'TASK_CANCELLED' });
+      assert.equal(settle.willRetry, false);
+      assert.equal(calls, phase === 'before-hook' ? 0 : 1, 'no model continuation after cancellation');
+      assert.equal(result.failedCount, 0);
+      assert.equal(result.details.cancelledTasks.length, 1);
+      assert.equal(webpush.sent.length, 0);
+    } finally { globalThis.fetch = originalFetch; }
+  });
+}
+
+
+test('cancellation during outbox commit removes silent results after the commit completes', async () => {
+  const adapter = createD1Adapter(createTestD1());
+  await adapter.initSchema();
+  await seed(adapter, { uuid: 'abort-outbox', recurrenceType: 'none', nextSendAt: recentDue(), payload: {
+    messageType: 'auto', apiUrl: 'https://example.com/v1/chat/completions', apiKey: 'key', primaryModel: 'test',
+  } });
+  let signal;
+  let settle;
+  const db = new Proxy(adapter, {
+    get(target, prop) {
+      if (prop === 'appendOutboxMessages') return async (...args) => {
+        await adapter.deleteTaskByUuid('abort-outbox', USER);
+        await waitUntil(() => signal.aborted, 'lease cancellation was not observed');
+        // The cancel endpoint can clean an empty outbox before this write lands.
+        return target.appendOutboxMessages(...args);
+      };
+      const value = target[prop];
+      return typeof value === 'function' ? value.bind(target) : value;
+    }
+  });
+  const originalFetch = globalThis.fetch;
+  globalThis.fetch = async () => ({ ok: true, json: async () => ({ choices: [{ message: { content: 'hello' } }] }) });
+  try {
+    const result = await runScheduledTick({ db, masterKey: MASTER_KEY, vapid: VAPID, webpush: fakeWebpush(), leaseHeartbeatMs: 5,
+      hooks: {
+        onBeforeFire: ctx => { signal = ctx.signal; return [{ role: 'user', content: 'hi' }]; },
+        onLLMOutput: () => ({ decision: 'finish', pushPayloads: [{ messageKind: 'content', message: 'hidden', notification: { show: false } }] }),
+      },
+      onFireSettled: info => { settle = info; },
+    });
+    assert.equal(settle.status, 'cancelled');
+    assert.equal(result.successCount, 0);
+    assert.equal((await adapter.listUnackedOutbox(USER, 0, 100)).length, 0);
+  } finally { globalThis.fetch = originalFetch; }
+});
diff --git a/packages/rei-standard-amsg/shared/src/llm-call.js b/packages/rei-standard-amsg/shared/src/llm-call.js
index 81c6f93..6031222 100644
--- a/packages/rei-standard-amsg/shared/src/llm-call.js
+++ b/packages/rei-standard-amsg/shared/src/llm-call.js
@@ -53,6 +53,7 @@ const UPSTREAM_ERROR_BODY_MAX_BYTES = 16 * 1024;
  * @param {{
  *   requireContent?: boolean,
  *   timeoutMs?: number,
+ *   signal?: AbortSignal,
  *   fetch?: typeof globalThis.fetch,
  *   stream?: boolean,
  *   forwardTools?: boolean,
@@ -98,7 +99,9 @@ export async function callLlm(payload, options = {}) {
       'Authorization': `Bearer ${payload.apiKey}`
     },
     body: JSON.stringify(requestBody),
-    signal: AbortSignal.timeout(timeoutMs)
+    signal: options.signal
+      ? AbortSignal.any([options.signal, AbortSignal.timeout(timeoutMs)])
+      : AbortSignal.timeout(timeoutMs)
   });
 
   if (!aiResponse.ok) {

From 4aa13eac6f53c54ac467233aa5c0267b0bb4a7a0 Mon Sep 17 00:00:00 2001
From: Tosd0 <65720409+Tosd0@users.noreply.github.com>
Date: Thu, 1 Oct 2026 01:25:24 +0900
Subject: [PATCH 2/2] fix(amsg-server): guard cancelled hook state writes

---
 packages/rei-standard-amsg/server/README.md   |  2 ++
 .../server/src/server/lib/agentic-fire.js     |  8 ++++--
 .../server/test/run-tick.test.mjs             | 26 +++++++++++++++++++
 3 files changed, 34 insertions(+), 2 deletions(-)

diff --git a/packages/rei-standard-amsg/server/README.md b/packages/rei-standard-amsg/server/README.md
index f852316..9ced586 100644
--- a/packages/rei-standard-amsg/server/README.md
+++ b/packages/rei-standard-amsg/server/README.md
@@ -965,3 +965,5 @@ async function executeToolCalls(calls, ctx) {
 ```
 
 取消收尾使用 `onFireSettled({ status: 'cancelled', willRetry: false, error, ... })`，不是生成失败。应用应在这里释放资源、结算已发生的用量，避免发出失败提示或把未送达内容记为已回复。取消时已经开始写入的 outbox 批次会在写完后检查信号并撤掉未投递部分；已经发到客户端的消息仍由客户端处理。应用需要记录停止的任务身份，拦住迟到推送、补收以及尚未显示的本地内容。已完成的外部副作用无法撤销；不支持 AbortSignal 的工具也需要自己在执行步骤之间检查取消。
+
+执行阶段的 `ctx.writeState`、`scheduleTask`、`cancelTask`、`renewTask` 和 `emitResult` 会拒绝取消后新发起的操作。`onFireSettled` 的 `info.writeState` 特意仍可用，供宿主释放锁、保存账单等收尾；它不能用于继续生成阶段的业务动作。已经开始的数据库写入不支持回滚，宿主仍需处理并发状态版本。
diff --git a/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js b/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js
index eed7848..17eadf6 100644
--- a/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js
+++ b/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js
@@ -335,13 +335,17 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
 
   // client_state 的读写口。实现与 `GET/PUT /client-state` 共用一份（见
   // lib/state-accessors.js），fire 级和 config 级 hook 拿到的是同一套语义。
-  const { readState, writeState } = createStateAccessors({
+  const { readState, writeState: writeSettledState } = createStateAccessors({
     db: ctx.db,
     userId: task.user_id,
     userKey,
     maxStateValueBytes: ctx.maxStateValueBytes,
     now: nowFn,
   });
+  const writeState = async (namespace, entries) => {
+    throwIfCancelled();
+    return writeSettledState(namespace, entries);
+  };
 
   // 这一次 fire 的三个身份值，整条链共用一份：sessionId 钉在（任务 id + 名义
   // 触发时刻）上，同一 occurrence 的重试复用同一个 session、不同 occurrence 各
@@ -846,7 +850,7 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
       outboxed: progress.outboxed,
       scratch,
       readState,
-      writeState,
+      writeState: writeSettledState,
       emitResult,
     });
   }
diff --git a/packages/rei-standard-amsg/server/test/run-tick.test.mjs b/packages/rei-standard-amsg/server/test/run-tick.test.mjs
index aa6b69f..e8470ca 100644
--- a/packages/rei-standard-amsg/server/test/run-tick.test.mjs
+++ b/packages/rei-standard-amsg/server/test/run-tick.test.mjs
@@ -1852,3 +1852,29 @@ test('cancellation during outbox commit removes silent results after the commit
     assert.equal((await adapter.listUnackedOutbox(USER, 0, 100)).length, 0);
   } finally { globalThis.fetch = originalFetch; }
 });
+
+
+test('cancelled hook cannot write more state but settlement can release resources', async () => {
+  const adapter = createD1Adapter(createTestD1());
+  await adapter.initSchema();
+  await seed(adapter, { uuid: 'abort-write', recurrenceType: 'none', nextSendAt: recentDue(), payload: { messageType: 'auto' } });
+  let writeError;
+  const result = await runScheduledTick({ db: adapter, masterKey: MASTER_KEY, vapid: VAPID, webpush: fakeWebpush(), leaseHeartbeatMs: 5,
+    hooks: {
+      onBeforeFire: async ctx => {
+        await adapter.deleteTaskByUuid('abort-write', USER);
+        await waitUntil(() => ctx.signal.aborted, 'lease cancellation was not observed');
+        try { await ctx.writeState('cancel-test', [{ key: 'late', value: 'must not persist' }]); }
+        catch (error) { writeError = error; }
+        return { skip: true };
+      },
+      onLLMOutput: () => ({ decision: 'skip-push' }),
+    },
+    onFireSettled: async info => {
+      await info.writeState('cancel-test', [{ key: 'cleanup', value: 'released' }]);
+    },
+  });
+  assert.equal(writeError?.code, 'TASK_CANCELLED');
+  assert.deepEqual((await adapter.getClientState(USER, 'cancel-test')).map(row => row.key), ['cleanup']);
+  assert.equal(result.details.cancelledTasks.length, 1);
+});