feat(amsg-server): onBeforeFire skip 出口 + GET /messages 投影 charId/clientTaskId

.changeset/amsg-server-before-fire-skip.md

@@ -0,0 +1,9 @@
+---
+"@rei-standard/amsg-server": minor
+---
+
+onBeforeFire 新增 `{ skip: true }` 出口：在第一次 LLM 调用之前结束本次 fire
+
+宿主在 fire 时刻就能判断这条消息已经多余（比如排程之后对话又有了新进展）时，`onBeforeFire` 返回 `{ skip: true }` 即可作废本次触发。这次 fire 算作一次零推送的成功投递（`status: 'skipped'`）：一次性任务照删、循环任务照推进到下次，且不调用 LLM、不消耗 token。
+
+返回 `null` 的既有语义不变（回退到排程时冻结的 completePrompt 老链路）。

.changeset/amsg-server-messages-identity.md

@@ -0,0 +1,5 @@
+---
+"@rei-standard/amsg-server": minor
+---
+
+`GET /messages` 每条任务额外返回 `charId` / `clientTaskId`（取自任务 metadata 的 `charId` / `amsgClientTaskId`，缺省为 null），供宿主按角色归属筛选任务。metadata 的其余字段不回传，凭据类字段照旧不回传。

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

@@ -74,6 +74,9 @@ export default createSingleUserCloudflareWorker((env) => ({
         { role: 'user', content: `根据这些记录写一条提醒：${notes.map(n => n.value).join('\n')}` },
       ];
       // 也可以返回 { messages, maxToolIterations, totalTimeoutMs } 按次放宽预算
+      // 或返回 { skip: true }：这次不生成，零推送直接算成功结束（不调 LLM）。
+      //   一次性任务照删、循环任务照推进到下次。适合排程后对话已有新进展、
+      //   这条到点已多余的情况。
     },
 
     // 每轮 LLM 输出后分类。ctx 形状与 @rei-standard/amsg-instant 的

packages/rei-standard-amsg/server/src/server/handlers/messages.js

@@ -65,7 +65,14 @@ export function createMessagesHandler(ctx) {
         status: task.status,
         retryCount: task.retry_count,
         createdAt: task.created_at,
-        updatedAt: task.updated_at
+        updatedAt: task.updated_at,
+        // Character ownership / client-side task identity, pulled from the
+        // scheduling host's metadata so it can filter tasks by character
+        // (contactName can collide across characters). Only these two
+        // metadata fields are surfaced — the rest of metadata may hold
+        // host-private data and stays server-side. Absent → null.
+        charId: decrypted.metadata?.charId ?? null,
+        clientTaskId: decrypted.metadata?.amsgClientTaskId ?? null
       };
     }));
 

packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js

@@ -4,7 +4,7 @@
  * When the host configures fire-time hooks, an LLM task stops replaying
  * the completePrompt frozen at schedule time. At fire time instead:
  *
- *   onBeforeFire(fireCtx) → fresh messages (may read client_state)
+ *   onBeforeFire(fireCtx) → fresh messages (may read client_state) | { skip: true }
  *     → callLlm → onLLMOutput(sessionCtx) → decision
  *         ├─ 'finish'       → push decision.pushPayloads, done
  *         ├─ 'skip-push'    → record, done (task counts as delivered)
@@ -105,7 +105,7 @@ function normalizeBeforeFireResult(result) {
     };
   }
   throw new TypeError(
-    'AGENTIC_BAD_BEFORE_FIRE: onBeforeFire must return ChatMessage[] | { messages, maxToolIterations?, totalTimeoutMs? } | null'
+    'AGENTIC_BAD_BEFORE_FIRE: onBeforeFire must return ChatMessage[] | { messages, maxToolIterations?, totalTimeoutMs? } | { skip: true } | null'
   );
 }
 
@@ -132,7 +132,11 @@ function firstPositiveNumber(values, fallback) {
  * @param {string} args.userKey - per-user storage key (for readState decryption)
  * @param {Object} args.ctx - processor ctx ({ db, webpush, vapid, hooks, maxToolIterations, totalTimeoutMs })
  * @returns {Promise<{ handled: false } | { handled: true, result: { success: true, messagesSent: number, status: 'finished'|'skipped', iterations: number } }>}
- *   `handled: false` → caller falls back to the legacy frozen-prompt path.
+ *   `handled: false` → caller falls back to the legacy frozen-prompt path
+ *   (onBeforeFire returned null).
+ *   `onBeforeFire` may also return `{ skip: true }` to complete the fire
+ *   before the first LLM call → `status: 'skipped', iterations: 0`, same
+ *   success handling as the post-LLM skip-push path.
  *   Failures (timeout / loop exceeded / config errors) throw — the caller's
  *   existing error handling turns them into task retry/failure.
  */
@@ -178,6 +182,14 @@ export async function runAgenticFire({ task, decryptedPayload, userKey, ctx }) {
   const before = await hooks.onBeforeFire(fireCtx);
   if (before == null) return { handled: false };
 
+  // Pre-LLM skip: the host judged this fire moot before generation (e.g. the
+  // conversation moved on after the task was scheduled). Shaped exactly like
+  // the post-LLM 'skipped' result, so run-tick's success handling (delete
+  // once-off / advance recurrence) applies unchanged — and zero tokens spent.
+  if (typeof before === 'object' && before.skip === true) {
+    return { handled: true, result: { success: true, messagesSent: 0, status: 'skipped', iterations: 0 } };
+  }
+
   const normalized = normalizeBeforeFireResult(before);
   const maxToolIterations = firstPositiveInt(
     [normalized.maxToolIterations, ctx.maxToolIterations],

packages/rei-standard-amsg/server/src/server/lib/message-processor.js

@@ -113,7 +113,9 @@ export async function processSingleMessage(task, ctx, providedMasterKey) {
     // needs the LLM, offer the agentic path first. onBeforeFire → null
     // falls straight through to the frozen-prompt chain below, and
     // deployments without hooks never enter this branch — legacy behavior
-    // is byte-identical.
+    // is byte-identical. onBeforeFire → { skip: true } completes the fire
+    // here as a zero-push success (no LLM call, no frozen-prompt fallback):
+    // use it when the host can tell at fire time the message is moot.
     if (ctx.hooks && typeof ctx.hooks.onBeforeFire === 'function' && taskNeedsLlm(decryptedPayload)) {
       const agentic = await runAgenticFire({ task, decryptedPayload, userKey, ctx });
       if (agentic.handled) return agentic.result;

packages/rei-standard-amsg/server/test/agentic-fire.test.mjs

@@ -288,6 +288,27 @@ describe('agentic fire loop', () => {
     }
   });
 
+  test('onBeforeFire { skip: true }: fire ends before the first LLM call; zero-push success', async () => {
+    const { task } = await makeTask();
+    const hooks = {
+      onBeforeFire: async () => ({ skip: true }),
+      // If the skip ever leaked into the LLM loop, onLLMOutput would run and
+      // this throw would flip result.success to false — the deepEqual guards it.
+      onLLMOutput: async () => { throw new Error('onLLMOutput must not run when onBeforeFire skips'); },
+    };
+    const pushes = [];
+    const llm = stubLlm([finishRound]);
+    try {
+      const result = await processSingleMessage(task, makeCtx({ hooks, pushSpy: (_s, p) => pushes.push(p) }));
+      // Same shape as the post-LLM skip-push path, but iterations: 0 — no round ran.
+      assert.deepEqual(result, { success: true, messagesSent: 0, status: 'skipped', iterations: 0 });
+      assert.equal(llm.calls.length, 0);   // never reached the LLM
+      assert.equal(pushes.length, 0);
+    } finally {
+      llm.restore();
+    }
+  });
+
   test('continue: nextHistory replaces the running messages', async () => {
     const { task } = await makeTask();
     const nextHistory = [{ role: 'user', content: '重来' }];

packages/rei-standard-amsg/server/test/single-user-server.test.mjs

@@ -3,7 +3,7 @@ import assert from 'node:assert/strict';
 import { createSingleUserServer } from '../src/server/single-user.js';
 import { createD1Adapter } from '../src/server/adapters/d1.js';
 import { createTestD1 } from './helpers/sqlite-d1.mjs';
-import { deriveUserEncryptionKey, encryptPayload, encryptForStorage, decryptFromStorage } from '../src/server/lib/encryption.js';
+import { deriveUserEncryptionKey, encryptPayload, encryptForStorage, decryptFromStorage, decryptPayload } from '../src/server/lib/encryption.js';
 
 const USER = '550e8400-e29b-41d4-a716-446655440000';
 const MASTER_KEY = 'a'.repeat(64);
@@ -55,6 +55,67 @@ test('schedule → list → cancel round-trips through single-user server over D
   assert.equal(cancelled.status, 200);
 });
 
+// GET /messages projects the task's charId / clientTaskId (taken from the
+// stored payload's metadata) so the host can filter tasks by character
+// ownership — contactName alone is ambiguous (characters can share a name).
+const SCHEDULE_HEADERS = {
+  'X-User-Id': USER,
+  'X-Payload-Encrypted': 'true',
+  'X-Encryption-Version': '1',
+};
+
+function basePayload(overrides = {}) {
+  return {
+    contactName: 'Rei',
+    messageType: 'fixed',
+    userMessage: 'hi',
+    firstSendTime: '2999-01-01T00:00:00.000Z',
+    recurrenceType: 'none',
+    pushSubscription: { endpoint: 'https://example.com/x', keys: { p256dh: 'k', auth: 'a' } },
+    ...overrides,
+  };
+}
+
+// Decrypt the encrypted GET /messages response back into { tasks, pagination }.
+async function listTasksDecrypted(server) {
+  const listed = await server.handlers.messages.GET('/messages?status=all', { 'X-User-Id': USER });
+  assert.equal(listed.status, 200);
+  const userKey = await deriveUserEncryptionKey(USER, MASTER_KEY);
+  return decryptPayload(listed.body.data, userKey);
+}
+
+test('GET /messages projects charId / clientTaskId from task metadata', async () => {
+  const server = await makeServer();
+  const clientTaskId = 'a1b2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d'; // uuid v4
+  const created = await server.handlers.scheduleMessage.POST(
+    SCHEDULE_HEADERS,
+    await encBody(basePayload({ metadata: { charId: 'char-1', amsgClientTaskId: clientTaskId } }))
+  );
+  assert.equal(created.status, 201);
+  const uuid = created.body.data.uuid;
+
+  const { tasks } = await listTasksDecrypted(server);
+  const row = tasks.find(t => t.uuid === uuid);
+  assert.ok(row, 'scheduled task should appear in the list');
+  assert.equal(row.charId, 'char-1');
+  assert.equal(row.clientTaskId, clientTaskId);
+  // the whole metadata object must NOT leak onto the response
+  assert.equal('metadata' in row, false);
+});
+
+test('GET /messages: charId / clientTaskId are null when metadata is absent', async () => {
+  const server = await makeServer();
+  const created = await server.handlers.scheduleMessage.POST(SCHEDULE_HEADERS, await encBody(basePayload()));
+  assert.equal(created.status, 201);
+  const uuid = created.body.data.uuid;
+
+  const { tasks } = await listTasksDecrypted(server);
+  const row = tasks.find(t => t.uuid === uuid);
+  assert.ok(row, 'scheduled task should appear in the list');
+  assert.equal(row.charId, null);
+  assert.equal(row.clientTaskId, null);
+});
+
 test('masterKey wiring: storage encrypt/decrypt round-trips', async () => {
   const userKey = await deriveUserEncryptionKey(USER, MASTER_KEY);
   const round = JSON.parse(await decryptFromStorage(await encryptForStorage(JSON.stringify({ a: 1 }), userKey), userKey));