From ca31c9eec4a5bd1efb70cec4889d8c9a04f974d6 Mon Sep 17 00:00:00 2001 From: Tosd0 <65720409+Tosd0@users.noreply.github.com> Date: Tue, 21 Jul 2026 10:26:49 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(amsg-server):=20onBeforeFire=20?= =?UTF-8?q?=E6=94=AF=E6=8C=81=20{=20skip:=20true=20}=20=E7=94=9F=E6=88=90?= =?UTF-8?q?=E5=89=8D=E8=B7=B3=E8=BF=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 宿主在 fire 时刻就能判断消息已多余时，onBeforeFire 返回 { skip: true } 作废本次触发：算一次零推送成功（status: 'skipped'），一次性任务照删、 循环任务照推进到下次，不调 LLM、不耗 token。返回 null 的老语义不变。 Claude-Session: https://claude.ai/code/session_017mfb5T9eZU4iHpFbCEmeqR --- .changeset/amsg-server-before-fire-skip.md | 9 ++++++++ .../examples/cloudflare-single-user/README.md | 3 +++ .../server/src/server/lib/agentic-fire.js | 18 +++++++++++++--- .../src/server/lib/message-processor.js | 4 +++- .../server/test/agentic-fire.test.mjs | 21 +++++++++++++++++++ 5 files changed, 51 insertions(+), 4 deletions(-) create mode 100644 .changeset/amsg-server-before-fire-skip.md diff --git a/.changeset/amsg-server-before-fire-skip.md b/.changeset/amsg-server-before-fire-skip.md new file mode 100644 index 0000000..46bdf5b --- /dev/null +++ b/.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 老链路）。 diff --git a/packages/rei-standard-amsg/server/examples/cloudflare-single-user/README.md b/packages/rei-standard-amsg/server/examples/cloudflare-single-user/README.md index c1c8dea..32fa895 100644 --- a/packages/rei-standard-amsg/server/examples/cloudflare-single-user/README.md +++ b/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 的 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 4098651..84af7ef 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 @@ -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], 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 dc909a2..18e04ff 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 @@ -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; 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 5cd3db1..0097751 100644 --- a/packages/rei-standard-amsg/server/test/agentic-fire.test.mjs +++ b/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: '重来' }]; From 81133e4d09ff9d17234c1199535b8fd26316f944 Mon Sep 17 00:00:00 2001 From: Tosd0 <65720409+Tosd0@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:21:15 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat(amsg-server):=20GET=20/messages=20?= =?UTF-8?q?=E6=8A=95=E5=BD=B1=20charId/clientTaskId?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 每条任务额外返回 charId / clientTaskId（取自任务 metadata 的 charId / amsgClientTaskId，缺省 null），供宿主按角色归属筛选任务。metadata 其余 字段与凭据类字段照旧不回传。 Claude-Session: https://claude.ai/code/session_017mfb5T9eZU4iHpFbCEmeqR --- .changeset/amsg-server-messages-identity.md | 5 ++ .../server/src/server/handlers/messages.js | 9 ++- .../server/test/single-user-server.test.mjs | 63 ++++++++++++++++++- 3 files changed, 75 insertions(+), 2 deletions(-) create mode 100644 .changeset/amsg-server-messages-identity.md diff --git a/.changeset/amsg-server-messages-identity.md b/.changeset/amsg-server-messages-identity.md new file mode 100644 index 0000000..ecbee0a --- /dev/null +++ b/.changeset/amsg-server-messages-identity.md @@ -0,0 +1,5 @@ +--- +"@rei-standard/amsg-server": minor +--- + +`GET /messages` 每条任务额外返回 `charId` / `clientTaskId`（取自任务 metadata 的 `charId` / `amsgClientTaskId`，缺省为 null），供宿主按角色归属筛选任务。metadata 的其余字段不回传，凭据类字段照旧不回传。 diff --git a/packages/rei-standard-amsg/server/src/server/handlers/messages.js b/packages/rei-standard-amsg/server/src/server/handlers/messages.js index 4b77954..6eb3b86 100644 --- a/packages/rei-standard-amsg/server/src/server/handlers/messages.js +++ b/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 }; })); diff --git a/packages/rei-standard-amsg/server/test/single-user-server.test.mjs b/packages/rei-standard-amsg/server/test/single-user-server.test.mjs index 15fdc52..aa3172f 100644 --- a/packages/rei-standard-amsg/server/test/single-user-server.test.mjs +++ b/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));