feat(amsg): 单用户/Cloudflare 满血后台消息（client_state + fire 时刻 hooks + 服务端工具循环）

.changeset/amsg-server-agentic-fire.md

@@ -0,0 +1,9 @@
+---
+"@rei-standard/amsg-server": minor
+---
+
+单用户 / Cloudflare 模式新增「fire 时刻现场生成」能力：
+
+- 新表 `client_state`（init-tenant 幂等建表）+ 三个端点：`PUT /client-state` 批量上传状态（按 `updatedAt` last-write-wins，单条 value ≤ 200KB）、`GET /client-state?namespace=` 读取、`DELETE /client-state` 清空。value 用 per-user key 加密落库，鉴权与加密头沿用现有端点。
+- `createSingleUserCloudflareWorker` 的 config 接受可选 `hooks: { onBeforeFire, onLLMOutput, executeToolCalls }` 与 `maxToolIterations`（默认 5）、`totalTimeoutMs`（默认 240000，两者都可在 onBeforeFire 返回值里按次覆盖）。配置后，AI 类任务在触发时由 onBeforeFire 现场组装 messages（可经 `ctx.readState(namespace)` 读 client_state），工具在 worker 内就地执行、多轮循环闭环后推送成品；onLLMOutput 的 ctx 与 decision 契约与 `@rei-standard/amsg-instant` 的同名 hook 一致，instant 的 classifier 可直接复用（`tool-request` 同时接受 `toolCalls` 直传与 tool_request pushPayloads 两种形状）。
+- 不配 hooks、或 onBeforeFire 返回 null 时，任务照走排程时冻结的 completePrompt 老链路；固定文本任务永远走老链路。hook ctx 不含 apiKey / pushSubscription / VAPID。多租户入口（Netlify/Neon）行为不变。

.changeset/amsg-shared-agentic-contract.md

@@ -0,0 +1,6 @@
+---
+"@rei-standard/amsg-shared": minor
+"@rei-standard/amsg-instant": patch
+---
+
+amsg-shared 新增 agentic 循环契约工具：`buildSessionContext`、`extractAssistantMessage`、`assertValidDecision`（新增 `inlineToolCalls` 选项，允许 `tool-request` 直接携带 `toolCalls`，供服务端就地执行工具的场景用）、`extractToolCallsFromDecision`。amsg-instant 的 SessionContext 构建与 decision 校验改为从 amsg-shared 复用同一实现，对外行为与错误信息不变。

packages/rei-standard-amsg/instant/src/message-processor.js

@@ -23,6 +23,7 @@ import {
   buildErrorPush,
   readReasoningContent,
   stripReasoningTags,
+  assertValidDecision,
 } from '@rei-standard/amsg-shared';
 
 import { sendWebPush } from './webpush.js';
@@ -42,7 +43,6 @@ const SLEEP_BETWEEN_MESSAGES_MS = 1500;
 const DEFAULT_MAX_LOOP_ITERATIONS = 10;
 const DEFAULT_MAX_INLINE_BYTES = 2600;
 const DEFAULT_BLOB_TTL_SECONDS = 60;
-const VALID_DECISIONS = new Set(['finish', 'tool-request', 'continue', 'skip-push']);
 const PUSH_PAYLOAD_BYTE_ENCODER = new TextEncoder();
 
 /**
@@ -684,78 +684,11 @@ async function runAgenticLoop(payload, ctx) {
   return { status: 'loop_exceeded', sessionId, iteration };
 }
 
-/**
- * Assert that the hook returned a structurally valid decision.
- * TypeScript discriminated unions don't survive into runtime, and a
- * misbehaving hook can easily return `null` / `{ decision: 'idk' }`
- * / `undefined`. Treat any of those as a hook contract violation so
- * we route through the same HOOK_THREW pipeline.
- *
- * @param {unknown} decision
+/*
+ * assertValidDecision moved to @rei-standard/amsg-shared (imported above)
+ * so amsg-server's fire-time loop validates the exact same contract.
  */
-function assertValidDecision(decision) {
-  if (!decision || typeof decision !== 'object') {
-    throw new TypeError(`onLLMOutput returned invalid decision: ${stringifyForError(decision)}`);
-  }
-  const tag = /** @type {{ decision?: unknown }} */ (decision).decision;
-  if (typeof tag !== 'string' || !VALID_DECISIONS.has(tag)) {
-    throw new TypeError(`onLLMOutput returned invalid decision tag: ${stringifyForError(tag)}`);
-  }
-
-  const hasSingular = Object.prototype.hasOwnProperty.call(decision, 'pushPayload');
-  const hasPlural = Object.prototype.hasOwnProperty.call(decision, 'pushPayloads');
-
-  if (hasSingular) {
-    throw new TypeError(
-      hasPlural
-        ? 'pushPayload (singular) is removed in 0.8.0, use pushPayloads'
-        : 'pushPayload (singular) is removed in 0.8.0, use pushPayloads: [yourPayload]'
-    );
-  }
-
-  if (tag === 'continue') {
-    if (!Array.isArray(/** @type {{ nextHistory?: unknown }} */ (decision).nextHistory)) {
-      throw new TypeError('decision:"continue" requires a nextHistory array');
-    }
-    return;
-  }
-
-  if (tag === 'skip-push') {
-    return;
-  }
-
-  // 'finish' / 'tool-request' — both need pushPayloads array
-  if (!hasPlural || !Array.isArray(/** @type {{ pushPayloads?: unknown }} */ (decision).pushPayloads)) {
-    throw new TypeError(`decision:"${tag}" requires a pushPayloads array`);
-  }
-  const pushes = /** @type {Array<unknown>} */ (decision.pushPayloads);
-  if (pushes.length === 0) {
-    throw new TypeError('pushPayloads: [] — use decision: skip-push to skip notification entirely');
-  }
-  for (let i = 0; i < pushes.length; i++) {
-    const p = pushes[i];
-    if (!p || typeof p !== 'object' || Array.isArray(p)) {
-      throw new TypeError(`pushPayloads[${i}] must be a plain object, got ${stringifyForError(p)}`);
-    }
-    if (Object.prototype.hasOwnProperty.call(p, 'splitPattern')) {
-      throw new TypeError(`pushPayloads[${i}].splitPattern is removed in 0.8.0; caller is responsible for splitting`);
-    }
-    if (Object.prototype.hasOwnProperty.call(p, 'messageId')) {
-      const id = p.messageId;
-      if (typeof id !== 'string' || id === '') {
-        throw new TypeError(`pushPayloads[${i}].messageId must be a non-empty string when set, got ${stringifyForError(id)}`);
-      }
-    }
-  }
-}
 
-function stringifyForError(value) {
-  try {
-    return JSON.stringify(value);
-  } catch {
-    return String(value);
-  }
-}
 
 /**
  * Push a payload (any of the four `messageKind` types or a free-form

packages/rei-standard-amsg/instant/src/session-context.js

@@ -17,6 +17,11 @@
  * The worker still owns those fields in its closure; the hook only
  * gets to *decide* (finish / tool-request / continue / skip-push), not
  * to re-execute LLM or push calls itself.
+ *
+ * Implementation lives in `@rei-standard/amsg-shared` so
+ * `@rei-standard/amsg-server`'s fire-time loop builds the exact same
+ * context shape. The typedefs stay here for this package's own d.ts
+ * generation and internal `import('./session-context.js')` references.
  */
 
 /**
@@ -41,93 +46,4 @@
  * @property {string}                   [avatarUrl]
  */
 
-/**
- * Build a SessionContext from a validated payload + the latest LLM
- * response. Caller must pass the already-appended messages array
- * (i.e. `[...originalHistory, choices[0].message]`).
- *
- * The returned object is frozen so a hook implementation cannot
- * accidentally mutate the live messages array — if the hook chooses
- * `decision:'continue'`, the worker still owns its copy.
- *
- * @param {Object} args
- * @param {string} args.sessionId
- * @param {ChatMessage[]} args.messages
- * @param {unknown} args.llmResponse
- * @param {number} args.iteration
- * @param {string} args.contactName
- * @param {string} [args.avatarUrl]
- * @param {string} [args.charId]
- * @param {Record<string, unknown>} [args.metadata]
- * @returns {SessionContext}
- */
-export function buildSessionContext({
-  sessionId,
-  messages,
-  llmResponse,
-  iteration,
-  contactName,
-  avatarUrl,
-  charId,
-  metadata,
-}) {
-  const llmOutputText = readLlmOutputText(llmResponse);
-  const ctx = {
-    sessionId,
-    charId,
-    messages,
-    llmResponse,
-    llmOutputText,
-    iteration,
-    metadata: metadata && typeof metadata === 'object' ? metadata : {},
-    contactName,
-    avatarUrl: avatarUrl || undefined,
-  };
-  return Object.freeze(ctx);
-}
-
-/**
- * Safely read `choices[0].message.content` as a string. Pure
- * tool-call responses legitimately have empty content, so we return
- * '' rather than throwing — matches the documented contract:
- *   "llmOutputText: ... 注意可能为空字符串 (纯 tool_calls 响应)".
- *
- * @param {unknown} llmResponse
- * @returns {string}
- */
-function readLlmOutputText(llmResponse) {
-  if (!llmResponse || typeof llmResponse !== 'object') return '';
-  const choices = /** @type {{ choices?: unknown }} */ (llmResponse).choices;
-  if (!Array.isArray(choices) || choices.length === 0) return '';
-  const message = /** @type {{ message?: { content?: unknown } }} */ (choices[0])?.message;
-  const content = message?.content;
-  return typeof content === 'string' ? content : '';
-}
-
-/**
- * Extract the `choices[0].message` whole object — preserving
- * `tool_calls` / `reasoning_content` / `refusal` etc. — for appending
- * to the running history. Falls back to a minimal placeholder when
- * the response is malformed so the hook still gets a chance to react
- * via `llmOutputText === ''`.
- *
- * Critically, we keep the entire message object (not just
- * `{role, content}`): the next round may need to forward a
- * `tool_calls` array to OpenAI alongside the matching tool-result
- * messages, and stripping the field would make the API reject the
- * request.
- *
- * @param {unknown} llmResponse
- * @returns {ChatMessage}
- */
-export function extractAssistantMessage(llmResponse) {
-  const message =
-    llmResponse &&
-    typeof llmResponse === 'object' &&
-    Array.isArray(/** @type {{ choices?: unknown }} */ (llmResponse).choices) &&
-    /** @type {{ choices: Array<{ message?: unknown }> }} */ (llmResponse).choices[0]?.message;
-  if (message && typeof message === 'object') {
-    return /** @type {ChatMessage} */ (message);
-  }
-  return { role: 'assistant', content: '' };
-}
+export { buildSessionContext, extractAssistantMessage } from '@rei-standard/amsg-shared';

packages/rei-standard-amsg/server/examples/cloudflare-single-user/README.md

@@ -27,13 +27,86 @@
 
 ## 端点
 
-`/get-user-key`、`/schedule-message`、`/messages`、`/update-message`、`/cancel-message`、`/init-tenant`、`/vapid-public-key`。
+`/get-user-key`、`/schedule-message`、`/messages`、`/update-message`、`/cancel-message`、`/init-tenant`、`/vapid-public-key`、`/client-state`。
 **没有 HTTP `/send-notifications`**——定时投递由 CF Cron Trigger 直接触发 `scheduled()`。
 
 `GET /vapid-public-key` 返回本 Worker 的 `VAPID_PUBLIC_KEY`，供前端创建 Web Push 订阅时作 `applicationServerKey`；未配置 VAPID 时返回 503。跟其它端点一样受 CORS 和 `serverToken` 约束。
 
 VAPID 和 webpush 都要配齐：定时投递（cron）和 `instant` 类型消息都靠它推送，缺了就发不出去。
 
+## 客户端状态同步（/client-state）
+
+这是啥：一张给客户端存状态的云端表。客户端平时把要给「fire 时刻 hooks」用的数据（最近聊天摘要、设置项……存啥由你定）批量同步上来，worker 触发定时任务时就能读到最新状态。一份活状态，按 namespace 组织，客户端是唯一写者。
+
+| 端点 | 语义 |
+|------|------|
+| `PUT /client-state` | 批量 upsert。body =（加密后的）`{ entries: [{ namespace, key, value, updatedAt }] }`。`updatedAt` 是 epoch 毫秒，旧于库内的条目跳过（last-write-wins）；单条 `value` ≤ 200KB，单次 ≤ 200 条 |
+| `GET /client-state?namespace=<ns>` | 取一个 namespace 的全部条目（解密后返回，响应加密） |
+| `DELETE /client-state` | 清空该用户的全部状态（设置页做「清除云端状态」按钮用） |
+
+`value` 是任意字符串（想存对象就自己 `JSON.stringify`），落库前用 per-user key 加密。鉴权和加密头跟其它端点完全一样。
+
+## Fire 时刻 hooks（服务端工具循环）
+
+这是啥：默认情况下，AI 类任务的 prompt 在排程那一刻就冻结进数据库，cron 到点后拿冻结文本调一次 LLM 就推送——上下文停留在排程时。配置 hooks 后，worker 会在**触发那一刻**现场组装 prompt，LLM 要查资料时直接在 worker 里执行工具、多轮循环，最后推送成品，全程不需要客户端在线。
+
+啥时候用：任务触发离排程隔得久（比如每周提醒）、希望消息基于最新状态生成，或生成过程需要查数据（配合 `/client-state`）。不配 hooks 一切照旧。
+
+```js
+export default createSingleUserCloudflareWorker((env) => ({
+  // ...其余 config
+  hooks: {
+    // 触发时组装 prompt。返回 null 则这个任务走冻结 prompt 老链路。
+    // ctx: { task, userId, readState(namespace), now }
+    //   task 是解密后的任务字段（不含 apiKey / pushSubscription）；
+    //   自定义字段排程时放 metadata 里，这里原样读回。
+    async onBeforeFire(ctx) {
+      const notes = await ctx.readState('notes'); // [{ namespace, key, value, updatedAt }]
+      return [
+        { role: 'system', content: '你是一个提醒助手。' },
+        { role: 'user', content: `根据这些记录写一条提醒：${notes.map(n => n.value).join('\n')}` },
+      ];
+      // 也可以返回 { messages, maxToolIterations, totalTimeoutMs } 按次放宽预算
+    },
+
+    // 每轮 LLM 输出后分类。ctx 形状与 @rei-standard/amsg-instant 的
+    // onLLMOutput 一致（sessionId / messages / llmResponse / llmOutputText /
+    // iteration / metadata / contactName / avatarUrl），instant 的
+    // classifier 可以直接拿来用。四种 decision：
+    //   { decision: 'finish', pushPayloads }        → 推送这些 payload，结束
+    //   { decision: 'tool-request', toolCalls }     → 交给 executeToolCalls 执行
+    //     （也接受 instant 形状：pushPayloads 里带 tool_request push）
+    //   { decision: 'continue', nextHistory }       → 换个 history 再来一轮
+    //   { decision: 'skip-push' }                   → 这次不发，结束
+    async onLLMOutput(ctx) {
+      const toolCalls = ctx.llmResponse?.choices?.[0]?.message?.tool_calls;
+      if (Array.isArray(toolCalls) && toolCalls.length > 0) {
+        return { decision: 'tool-request', toolCalls };
+      }
+      return {
+        decision: 'finish',
+        pushPayloads: [{ messageKind: 'content', message: ctx.llmOutputText }],
+      };
+    },
+
+    // 工具在 worker 里就地执行（触发时客户端多半不在线，没有人替你执行）。
+    // 返回 OpenAI tool-result 形状；抛错的话错误文本会作为 tool result
+    // 回填给 LLM，让它自己圆场，不会整条链失败。
+    async executeToolCalls(toolCalls, ctx) {
+      return Promise.all(toolCalls.map(async (call) => ({
+        tool_call_id: call.id,
+        role: 'tool',
+        content: await runMyTool(call.function.name, call.function.arguments),
+      })));
+    },
+  },
+  maxToolIterations: 5,    // LLM 轮数上限（默认 5）
+  totalTimeoutMs: 240_000, // 整链墙钟超时（默认 240s）
+}));
+```
+
+预算兜底：轮数到上限、或整链超过 `totalTimeoutMs`，按任务失败处理（沿用现有重试/标记逻辑）。hook 收到的 ctx 里没有 apiKey、pushSubscription、VAPID——`console.log(ctx)` 不会把密钥打进日志。
+
 ## 导入入口
 
 Worker 从 `@rei-standard/amsg-server/cloudflare` 导入（不是包根）。这个子路径只含单用户 + D1 + Web Crypto 推送那条路径，不牵扯 pg / neon / web-push，所以只装了 D1 的环境也能打包通过。

packages/rei-standard-amsg/server/examples/cloudflare-single-user/schema.sql

@@ -29,3 +29,14 @@ CREATE INDEX IF NOT EXISTS idx_user_id
 CREATE UNIQUE INDEX IF NOT EXISTS uidx_uuid
   ON scheduled_messages (uuid)
   WHERE uuid IS NOT NULL;
+
+-- 客户端状态的云端镜像（/client-state 端点用）。
+-- value 是密文；updated_at 是客户端给的 epoch 毫秒整数，用于 last-write-wins。
+CREATE TABLE IF NOT EXISTS client_state (
+  user_id TEXT NOT NULL,
+  namespace TEXT NOT NULL,
+  key TEXT NOT NULL,
+  value TEXT NOT NULL,
+  updated_at INTEGER NOT NULL,
+  PRIMARY KEY (user_id, namespace, key)
+);