diff --git a/apps/api/src/app.test.ts b/apps/api/src/app.test.ts index e9dfbe2..9c44da3 100644 --- a/apps/api/src/app.test.ts +++ b/apps/api/src/app.test.ts @@ -434,6 +434,57 @@ describe("TOMEET core flow", () => { .toHaveLength(1); }); + it("persists each superseded WeChat source message only once", async () => { + const store = new MemoryStore(); + const processor = new JobProcessor(store, new MockAgentIntelligence(), new MockMatchmakingIntelligence()); + const internalApiToken = "stable-wechat-turn-token-at-least-32-characters"; + const app = await buildApp({ store, inlineProcessor: processor, internalApiToken }); + apps.push(app); + const userId = randomUUID(); + const connectionId = randomUUID(); + const firstMessageKey = randomUUID(); + const secondMessageKey = randomUUID(); + + for (const [index, generation] of [randomUUID(), randomUUID()].entries()) { + const response = await app.inject({ + method: "POST", + url: "/internal/agent/response-generations", + headers: { "x-tomeet-internal-token": internalApiToken }, + payload: { connectionId, generationToken: generation } + }); + expect(response.statusCode).toBe(204); + + const messages = [ + { messageId: "wechat-1", content: "第一条原话", idempotencyKey: firstMessageKey } + ]; + if (index === 1) { + messages.push({ messageId: "wechat-2", content: "第二条原话", idempotencyKey: secondMessageKey }); + } + const turn = await app.inject({ + method: "POST", + url: "/internal/agent/messages", + headers: { "x-tomeet-internal-token": internalApiToken }, + payload: { + userId, + displayName: "微信用户", + content: messages.map((message, index) => `${index + 1}. ${message.content}`).join("\n"), + idempotencyKey: randomUUID(), + connectionId, + generationToken: generation, + messages + } + }); + expect(turn.statusCode).toBe(200); + } + + const userMessages = (await store.listRecentMessages(userId, 20)) + .filter((message) => message.role === "user"); + expect(userMessages.map((message) => message.content)).toEqual([ + "第一条原话", + "第二条原话" + ]); + }); + it("auto-provisions deterministic channel users only when explicitly enabled", async () => { const store = new MemoryStore(); const processor = new JobProcessor(store, new MockAgentIntelligence(), new MockMatchmakingIntelligence()); diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 52da91f..27605f2 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -278,6 +278,8 @@ export async function buildApp(options: BuildAppOptions) { const { language } = z.object({ language: adventurexLanguageSchema.default("zh") }).parse(request.body ?? {}); + const state = await options.store.ensureAdventurexOnboardingState(userId); + if (state.welcomeDeliveredAt) return { message: null }; const message = await options.store.startAdventurexOnboarding(userId, language); return { message }; } @@ -292,13 +294,31 @@ export async function buildApp(options: BuildAppOptions) { isNewWechatIdentity: async (externalUserId) => ( (await options.store.resolveChannelIdentity("wechat", externalUserId)) === null ), - onActivated: async ({ userId }) => { + onActivated: async ({ userId, deliverText }) => { const onboardingState = await options.store.ensureAdventurexOnboardingState(userId); - if (onboardingState.welcomeSentAt) return; - await options.store.startAdventurexOnboarding(userId, "zh"); + if (onboardingState.welcomeDeliveredAt) return; + const message = await options.store.startAdventurexOnboarding(userId, "zh"); + if (!message || !deliverText) return; + const bubbles = message.content.split(/\n\s*\n+/u).filter(Boolean); + for (const [index, text] of bubbles.entries()) { + await deliverText({ text, runId: `activation-welcome-${userId}-${index + 1}` }); + } + await options.store.markAdventurexWelcomeDelivered(userId); } }); + app.post( + "/internal/users/:userId/adventurex-onboarding/welcome-delivered", + { config: { rateLimit: false } }, + async (request, reply) => { + if (!internalTokenMatches(request.headers["x-tomeet-internal-token"])) { + return reply.code(401).send({ error: "unauthorized", message: "内部服务认证失败" }); + } + const { userId } = z.object({ userId: uuidSchema }).parse(request.params); + return { state: await options.store.markAdventurexWelcomeDelivered(userId) }; + } + ); + async function requireAdventurexTestPoolOwner(request: FastifyRequest, reply: FastifyReply): Promise { if (!options.adventurexTestPoolAccessTokenMatches) { reply.code(503).send({ @@ -396,7 +416,12 @@ export async function buildApp(options: BuildAppOptions) { const generation = sourceChannel === "wechat" ? z.object({ connectionId: uuidSchema.optional(), - generationToken: z.string().min(8).max(128).optional() + generationToken: z.string().min(8).max(128).optional(), + messages: z.array(z.object({ + messageId: z.string().min(1).max(255), + content: z.string().trim().min(1).max(20_000), + idempotencyKey: z.string().min(8).max(128) + })).min(1).max(20).optional() }).refine( (value) => Boolean(value.connectionId) === Boolean(value.generationToken), "connectionId 与 generationToken 必须同时提供" @@ -404,13 +429,25 @@ export async function buildApp(options: BuildAppOptions) { : {}; assertCurrentUser(request, input.userId); await options.store.ensureUser(input.userId, input.displayName); - const userMessage = await options.store.appendMessage({ - userId: input.userId, - role: "user", - content: input.content, - idempotencyKey: input.idempotencyKey, - sourceChannel - }); + const rawMessages = "messages" in generation ? generation.messages : undefined; + const savedMessages = rawMessages + ? await Promise.all(rawMessages.map((item) => options.store.appendMessage({ + userId: input.userId, + role: "user", + content: item.content, + idempotencyKey: item.idempotencyKey, + sourceChannel + }))) + : [await options.store.appendMessage({ + userId: input.userId, + role: "user", + content: input.content, + idempotencyKey: input.idempotencyKey, + sourceChannel + })]; + const userMessage = savedMessages.at(-1)!; + const connectionId = "connectionId" in generation ? generation.connectionId : undefined; + const generationToken = "generationToken" in generation ? generation.generationToken : undefined; const job = await options.store.enqueueJob({ type: "agent_reply", payload: { @@ -418,9 +455,12 @@ export async function buildApp(options: BuildAppOptions) { content: input.content, userMessageId: userMessage.id, sourceChannel, - ...generation + connectionId, + generationToken }, - idempotencyKey: `agent:${userMessage.id}`, + idempotencyKey: generationToken + ? `agent-generation:${connectionId}:${generationToken}` + : `agent:${userMessage.id}`, partitionKey: `user:${input.userId}` }); const currentJob = await runInline(job.id); @@ -540,9 +580,17 @@ export async function buildApp(options: BuildAppOptions) { connectionId: uuidSchema.optional(), generationToken: z.string().min(8).max(128).optional(), images: z.array(z.object({ + messageId: z.string().min(1).max(255).optional(), mimeType: z.enum(["image/jpeg", "image/png", "image/webp"]), dataBase64: z.string().min(4).max(14 * 1024 * 1024) })).min(1).max(9), + turns: z.array(z.object({ + messageId: z.string().min(1).max(255), + content: z.string().trim().min(1).max(2_000).optional(), + imageCount: z.number().int().min(0).max(9), + idempotencyKey: z.string().min(8).max(128) + })).min(1).max(20).optional(), + messageIds: z.array(z.string().min(1).max(255)).min(1).max(20).optional(), hint: z.string().trim().min(1).max(2_000).optional(), idempotencyKey: z.string().min(8).max(128) }).refine( @@ -558,6 +606,16 @@ export async function buildApp(options: BuildAppOptions) { throw new StoreConflictError("单批微信图片总大小必须在 30MB 以内"); } await options.store.ensureUser(input.userId, "微信用户"); + const imageMessageIds = input.images.map((image, index) => + image.messageId ?? input.messageIds?.[index] ?? `batch-${index + 1}` + ); + const turns = input.turns ?? [...new Set(input.messageIds ?? imageMessageIds)].map((messageId, index) => ({ + messageId, + ...(index === 0 && input.hint ? { content: input.hint } : {}), + imageCount: imageMessageIds.filter((id) => id === messageId).length, + idempotencyKey: `multimodal-turn:${input.idempotencyKey}:${messageId}` + })); + const turnByMessageId = new Map(turns.map((turn) => [turn.messageId, turn])); const records = await Promise.all(decoded.map(async (image, index) => { const extension = image.mimeType === "image/jpeg" ? "jpg" : image.mimeType.split("/")[1]!; const storagePath = `${input.userId}/${randomUUID()}.${extension}`; @@ -568,17 +626,24 @@ export async function buildApp(options: BuildAppOptions) { storagePath, mimeType: image.mimeType, sizeBytes: image.bytes.length, - hint: index === 0 ? input.hint : undefined + hint: turnByMessageId.get(imageMessageIds[index]!)?.content }); return { inputId, storagePath, mimeType: image.mimeType }; })); - await options.store.appendMessage({ - userId: input.userId, - role: "user", - content: input.images.length === 1 ? "[发送了一张图片]" : `[一次发送了 ${input.images.length} 张图片]`, - idempotencyKey: `multimodal-user:${input.idempotencyKey}`, - sourceChannel: "wechat" - }); + const savedMessages = await Promise.all(turns.map((turn) => { + const imagePrefix = turn.imageCount === 0 + ? "" + : turn.imageCount === 1 ? "[发送了一张图片]" : `[一次发送了 ${turn.imageCount} 张图片]`; + return options.store.appendMessage({ + userId: input.userId, + role: "user", + content: [imagePrefix, turn.content].filter(Boolean).join("\n"), + idempotencyKey: turn.idempotencyKey, + sourceChannel: "wechat" + }); + })); + const userMessage = savedMessages.at(-1)!; + const hint = turns.flatMap((turn) => turn.content ? [turn.content] : []).join("\n") || undefined; const job = await options.store.enqueueJob({ type: "multimodal_understanding", payload: { @@ -587,7 +652,8 @@ export async function buildApp(options: BuildAppOptions) { storagePaths: records.map((record) => record.storagePath), mimeTypes: records.map((record) => record.mimeType), kind: "image", - hint: input.hint, + hint, + userMessageId: userMessage.id, sourceChannel: "wechat", connectionId: input.connectionId, generationToken: input.generationToken @@ -733,7 +799,17 @@ export async function buildApp(options: BuildAppOptions) { } const choices = await options.store.saveMatchChoices(id, input); const preferredOpenRoom = choices.find((choice) => choice.preferenceRank === 1 && choice.sourceType === "open_room"); - if (!preferredOpenRoom) return { choices, status: "waiting_for_settlement" }; + if (!preferredOpenRoom) { + const notifyJob = choices.some((choice) => choice.draftId) + ? await options.store.enqueueJob({ + type: "room_change_notify", + payload: { roundId: matchRequest.activeRoundId }, + idempotencyKey: `draft-change-notify:${matchRequest.requestId}:${matchRequest.updatedAt}`, + partitionKey: `match-round:${matchRequest.activeRoundId}` + }) + : null; + return { choices, status: "waiting_for_settlement", jobId: notifyJob?.id ?? null }; + } const context = await options.store.listCurrentMatchOptions(matchRequest.userId); const offer = context?.options.find((option) => option.roomId === preferredOpenRoom.roomId); if (!offer) throw new StoreConflictError("开放局候选已经变化"); diff --git a/apps/api/src/wechat-routes.test.ts b/apps/api/src/wechat-routes.test.ts index 86431be..b7f606e 100644 --- a/apps/api/src/wechat-routes.test.ts +++ b/apps/api/src/wechat-routes.test.ts @@ -1,6 +1,6 @@ import { randomBytes, randomUUID } from "node:crypto"; import { MockAgentIntelligence } from "@tomeet/agent-core"; -import { adventurexWelcomeContent } from "@tomeet/contracts"; +import { adventurexWelcomeBubbles, adventurexWelcomeContent } from "@tomeet/contracts"; import { MemoryStore, MemoryWechatStore } from "@tomeet/data"; import { JobProcessor } from "@tomeet/intelligence"; import { MockMatchmakingIntelligence } from "@tomeet/matchmaking"; @@ -79,6 +79,13 @@ async function setup( return { app, store, wechatStore, fetchMock, verifyAccessToken, sentMessages }; } +function sentMessageTexts(messages: Array>): string[] { + return messages.map((message) => { + const msg = message.msg as { item_list?: Array<{ text_item?: { text?: string } }> }; + return msg.item_list?.[0]?.text_item?.text ?? ""; + }); +} + describe("WeChat one-time QR onboarding", () => { it("returns QR creation secrets once and never exposes them from status or SSE", async () => { const { app } = await setup([{ @@ -246,7 +253,7 @@ describe("WeChat one-time QR onboarding", () => { const firstUserId = firstIdentity!.userId; expect((await store.getUserModel(firstUserId)).userId).toBe(firstUserId); expect(enqueueWelcome).not.toHaveBeenCalled(); - expect(sentMessages).toEqual([]); + expect(sentMessageTexts(sentMessages)).toEqual(adventurexWelcomeBubbles.zh); expect(await store.listRecentMessages(firstUserId)).toEqual([ expect.objectContaining({ role: "assistant", @@ -255,7 +262,8 @@ describe("WeChat one-time QR onboarding", () => { ]); expect(await store.ensureAdventurexOnboardingState(firstUserId)).toMatchObject({ preferredLanguage: "zh", - welcomeSentAt: expect.any(String) + welcomeSentAt: expect.any(String), + welcomeDeliveredAt: expect.any(String) }); const secondCreate = await app.inject({ @@ -273,7 +281,7 @@ describe("WeChat one-time QR onboarding", () => { expect(secondConfirmed.json()).not.toHaveProperty("userId"); expect(await store.resolveChannelIdentity("wechat", "wechat-owner-1")) .toMatchObject({ userId: firstUserId }); - expect(sentMessages).toEqual([]); + expect(sentMessageTexts(sentMessages)).toEqual(adventurexWelcomeBubbles.zh); expect(await store.listRecentMessages(firstUserId)).toHaveLength(1); expect(verifyAccessToken).not.toHaveBeenCalled(); }); @@ -305,7 +313,7 @@ describe("WeChat one-time QR onboarding", () => { { code: 200, body: expect.objectContaining({ status: "active" }) }, { code: 200, body: expect.objectContaining({ status: "active" }) } ]); - expect(sentMessages).toEqual([]); + expect(sentMessageTexts(sentMessages)).toEqual(adventurexWelcomeBubbles.zh); const identity = await store.resolveChannelIdentity("wechat", "wechat-owner-concurrent"); expect(identity).not.toBeNull(); expect(await store.listRecentMessages(identity!.userId)).toEqual([ @@ -352,7 +360,7 @@ describe("WeChat one-time QR onboarding", () => { await vi.waitFor(async () => { expect(await wechatStore.getWechatSession(claimedSession.sessionId)) .toMatchObject({ status: "active" }); - expect(sentMessages).toEqual([]); + expect(sentMessageTexts(sentMessages)).toEqual(adventurexWelcomeBubbles.zh); }); expect(await store.resolveChannelIdentity("wechat", ownerIlinkUserId)).not.toBeNull(); }); @@ -380,7 +388,7 @@ describe("WeChat one-time QR onboarding", () => { await vi.waitFor(async () => { expect(await wechatStore.getWechatSession(session.sessionId)) .toMatchObject({ status: "active" }); - expect(sentMessages).toEqual([]); + expect(sentMessageTexts(sentMessages)).toEqual(adventurexWelcomeBubbles.zh); }); const identity = await store.resolveChannelIdentity( "wechat", diff --git a/apps/intelligence-worker/src/index.ts b/apps/intelligence-worker/src/index.ts index 56ce78d..71887f3 100644 --- a/apps/intelligence-worker/src/index.ts +++ b/apps/intelligence-worker/src/index.ts @@ -1,7 +1,12 @@ import { randomUUID } from "node:crypto"; import { resolve } from "node:path"; import { SupabaseStore } from "@tomeet/data"; -import { HostedLlmIntelligence, JobProcessor, TavilyWebSearchProvider } from "@tomeet/intelligence"; +import { + HostedLlmIntelligence, + JobProcessor, + StructuredOutputValidationError, + TavilyWebSearchProvider +} from "@tomeet/intelligence"; import { config } from "dotenv"; import { createWorkerHealthServer } from "./health-server.js"; @@ -124,7 +129,17 @@ async function runSlot(slot: number): Promise { } catch (error) { const message = error instanceof Error ? error.message : String(error); await store.failJob(job.id, message, slotId); - console.error(JSON.stringify({ level: "error", event: "job_failed", worker: slotId, jobId: job.id, type: job.type, error: message })); + console.error(JSON.stringify({ + level: "error", + event: "job_failed", + worker: slotId, + jobId: job.id, + type: job.type, + error: message, + ...(error instanceof StructuredOutputValidationError + ? { errorStage: error.stage, validationIssues: error.issues } + : {}) + })); } } catch (error) { console.error(JSON.stringify({ level: "error", event: "worker_loop_error", worker: slotId, error: error instanceof Error ? error.message : String(error) })); diff --git a/apps/wechat-ilink-worker/src/runtime.test.ts b/apps/wechat-ilink-worker/src/runtime.test.ts index 21b724a..73660fd 100644 --- a/apps/wechat-ilink-worker/src/runtime.test.ts +++ b/apps/wechat-ilink-worker/src/runtime.test.ts @@ -65,6 +65,7 @@ function setup() { } satisfies WechatTransport; const tomeet = { startOnboarding: vi.fn(async () => null), + markOnboardingWelcomeDelivered: vi.fn(async () => undefined), setResponseGeneration: vi.fn(async () => undefined), sendTextBatch: vi.fn(async () => ({ reply: "Agent reply", @@ -180,6 +181,7 @@ describe("WeChat worker runtime", () => { it("forwards one user message and completes its idempotency receipt", async () => { const runtime = setup(); const activeConnection = connection(runtime.cipher); + activeConnection.lastMessageAt = "2026-07-25T12:00:00.000Z"; await expect(handleWechatMessage( runtime.dependencies, @@ -197,9 +199,8 @@ describe("WeChat worker runtime", () => { expect(runtime.tomeet.sendTextBatch).toHaveBeenCalledWith({ connectionId: activeConnection.id, generationToken: expect.any(String), - messageIds: ["42"], userId: activeConnection.userId, - contents: ["你好"] + turns: [{ messageId: "42", content: "你好" }] }); expect(runtime.ilink.sendText).toHaveBeenCalledWith( expect.objectContaining({ @@ -216,7 +217,37 @@ describe("WeChat worker runtime", () => { expect(JSON.stringify(runtime.logger.info.mock.calls)).not.toContain("bot-secret"); }); - it("sends the onboarding welcome before the first Agent reply and only once", async () => { + it("consumes the first inbound trigger when activation already delivered the welcome", async () => { + const runtime = setup(); + const activeConnection = connection(runtime.cipher); + + await expect(handleWechatMessage( + runtime.dependencies, + activeConnection, + "bot-secret", + { + message_id: 41, + message_type: 1, + from_user_id: activeConnection.ownerIlinkUserId, + context_token: "context-opening-trigger", + item_list: [{ type: 1, text_item: { text: "123456" } }] + } + )).resolves.toBe(true); + + expect(runtime.tomeet.startOnboarding).toHaveBeenCalledTimes(1); + expect(runtime.tomeet.markOnboardingWelcomeDelivered).not.toHaveBeenCalled(); + expect(runtime.tomeet.setResponseGeneration).not.toHaveBeenCalled(); + expect(runtime.tomeet.sendTextBatch).not.toHaveBeenCalled(); + expect(runtime.tomeet.sendImages).not.toHaveBeenCalled(); + expect(runtime.ilink.sendText).not.toHaveBeenCalled(); + expect(runtime.store.completeWechatMessage).toHaveBeenCalledWith( + activeConnection.id, + "41" + ); + expect(activeConnection.lastMessageAt).not.toBeNull(); + }); + + it("consumes the opening trigger after sending the onboarding welcome", async () => { const runtime = setup(); const activeConnection = connection(runtime.cipher); runtime.tomeet.startOnboarding.mockResolvedValue(adventurexWelcomeContent("zh")); @@ -230,10 +261,19 @@ describe("WeChat worker runtime", () => { }); expect(runtime.tomeet.startOnboarding).toHaveBeenCalledTimes(1); + expect(runtime.tomeet.markOnboardingWelcomeDelivered).toHaveBeenCalledWith({ + userId: activeConnection.userId + }); expect(runtime.ilink.sendText.mock.calls.map(([input]) => input.text)).toEqual([ - ...adventurexWelcomeBubbles.zh, - "Agent reply" + ...adventurexWelcomeBubbles.zh ]); + expect(runtime.tomeet.setResponseGeneration).not.toHaveBeenCalled(); + expect(runtime.tomeet.sendTextBatch).not.toHaveBeenCalled(); + expect(runtime.tomeet.sendImages).not.toHaveBeenCalled(); + expect(runtime.store.completeWechatMessage).toHaveBeenCalledWith( + activeConnection.id, + "45" + ); expect(runtime.ilink.sendText.mock.calls.slice(0, 4).map(([input]) => input.runId)).toEqual([ `first-inbound-welcome-${activeConnection.id}-45-bubble-1`, `first-inbound-welcome-${activeConnection.id}-45-bubble-2`, @@ -243,6 +283,8 @@ describe("WeChat worker runtime", () => { expect(activeConnection.lastMessageAt).not.toBeNull(); runtime.tomeet.startOnboarding.mockClear(); + runtime.tomeet.startOnboarding.mockResolvedValue(null); + runtime.tomeet.sendTextBatch.mockClear(); runtime.ilink.sendText.mockClear(); await handleWechatMessage(runtime.dependencies, activeConnection, "bot-secret", { message_id: 46, @@ -252,7 +294,10 @@ describe("WeChat worker runtime", () => { item_list: [{ type: 1, text_item: { text: "第二条" } }] }); - expect(runtime.tomeet.startOnboarding).not.toHaveBeenCalled(); + expect(runtime.tomeet.startOnboarding).toHaveBeenCalledTimes(1); + expect(runtime.tomeet.sendTextBatch).toHaveBeenCalledWith(expect.objectContaining({ + turns: [{ messageId: "46", content: "第二条" }] + })); expect(runtime.ilink.sendText.mock.calls.map(([input]) => input.text)).toEqual([ "Agent reply" ]); @@ -261,6 +306,7 @@ describe("WeChat worker runtime", () => { it("groups multiple inbound image messages into one vision request and one reply", async () => { const runtime = setup(); const activeConnection = connection(runtime.cipher); + activeConnection.lastMessageAt = "2026-07-25T12:00:00.000Z"; runtime.ilink.getUpdates.mockResolvedValueOnce({ ret: 0, get_updates_buf: "cursor-1", @@ -305,11 +351,14 @@ describe("WeChat worker runtime", () => { expect(runtime.tomeet.sendImages).toHaveBeenCalledWith(expect.objectContaining({ connectionId: activeConnection.id, generationToken: expect.any(String), - messageIds: ["51", "52"], userId: activeConnection.userId, images: [ - { bytes: Uint8Array.from([1]), mimeType: "image/jpeg" }, - { bytes: Uint8Array.from([2]), mimeType: "image/jpeg" } + { messageId: "51", bytes: Uint8Array.from([1]), mimeType: "image/jpeg" }, + { messageId: "52", bytes: Uint8Array.from([2]), mimeType: "image/jpeg" } + ], + turns: [ + { messageId: "51", imageCount: 1 }, + { messageId: "52", imageCount: 1 } ] })); expect(runtime.ilink.sendText).toHaveBeenCalledTimes(1); @@ -406,9 +455,12 @@ describe("WeChat worker runtime", () => { expect(runtime.tomeet.sendImages).toHaveBeenCalledTimes(2); expect(runtime.tomeet.sendImages.mock.calls[1]?.[0]).toMatchObject({ - messageIds: ["71", "72", "73"], - images: [{ bytes: Uint8Array.from([7]), mimeType: "image/jpeg" }], - hint: "这张照片是我上周拍的\n是在苏州河边" + images: [{ messageId: "71", bytes: Uint8Array.from([7]), mimeType: "image/jpeg" }], + turns: [ + { messageId: "71", imageCount: 1 }, + { messageId: "72", content: "这张照片是我上周拍的", imageCount: 0 }, + { messageId: "73", content: "是在苏州河边", imageCount: 0 } + ] }); expect(runtime.ilink.sendText).toHaveBeenCalledTimes(1); expect(runtime.ilink.sendText).toHaveBeenCalledWith(expect.objectContaining({ @@ -451,8 +503,10 @@ describe("WeChat worker runtime", () => { expect(runtime.tomeet.sendTextBatch).toHaveBeenCalledTimes(1); expect(runtime.tomeet.sendTextBatch).toHaveBeenCalledWith(expect.objectContaining({ - messageIds: ["61", "62"], - contents: ["我平时喜欢拍城市夜景", "最近也在学胶片摄影"] + turns: [ + { messageId: "61", content: "我平时喜欢拍城市夜景" }, + { messageId: "62", content: "最近也在学胶片摄影" } + ] })); expect(runtime.ilink.sendText).toHaveBeenCalledTimes(1); }); @@ -464,6 +518,7 @@ describe("WeChat worker runtime", () => { stale: false }); const activeConnection = connection(runtime.cipher); + activeConnection.lastMessageAt = "2026-07-25T12:00:00.000Z"; await handleWechatMessage(runtime.dependencies, activeConnection, "bot-secret", { message_id: 44, @@ -510,6 +565,7 @@ describe("WeChat worker runtime", () => { it("asks the Agent to compose the reply for unsupported message content", async () => { const runtime = setup(); const activeConnection = connection(runtime.cipher); + activeConnection.lastMessageAt = "2026-07-25T12:00:00.000Z"; await expect(handleWechatMessage( runtime.dependencies, @@ -546,6 +602,7 @@ describe("WeChat worker runtime", () => { it("reports an unreadable picture as received instead of unsupported", async () => { const runtime = setup(); const activeConnection = connection(runtime.cipher); + activeConnection.lastMessageAt = "2026-07-25T12:00:00.000Z"; runtime.dependencies.downloadImage = vi.fn(async () => { throw new Error("微信图片缺少 CDN 下载参数"); }); @@ -596,7 +653,7 @@ describe("WeChat worker runtime", () => { expect(runtime.tomeet.sendEvent).not.toHaveBeenCalled(); expect(runtime.tomeet.sendTextBatch).toHaveBeenCalledWith(expect.objectContaining({ - contents: ["看看这张"] + turns: [{ messageId: "82", content: "看看这张" }] })); }); @@ -698,7 +755,7 @@ describe("WeChat worker runtime", () => { turnBatchWindowMs: 0 }); - expect(runtime.tomeet.sendTextBatch.mock.calls.map(([input]) => input.messageIds)).toEqual([ + expect(runtime.tomeet.sendTextBatch.mock.calls.map(([input]) => input.turns.map((turn) => turn.messageId))).toEqual([ ["101"], ["102"] ]); @@ -773,7 +830,7 @@ describe("WeChat worker runtime", () => { turnBatchWindowMs: 0 }); - expect(runtime.tomeet.sendTextBatch.mock.calls.map(([input]) => input.messageIds)).toEqual([ + expect(runtime.tomeet.sendTextBatch.mock.calls.map(([input]) => input.turns.map((turn) => turn.messageId))).toEqual([ ["111"], ["111", "112"] ]); diff --git a/apps/wechat-ilink-worker/src/runtime.ts b/apps/wechat-ilink-worker/src/runtime.ts index 8639c03..5421467 100644 --- a/apps/wechat-ilink-worker/src/runtime.ts +++ b/apps/wechat-ilink-worker/src/runtime.ts @@ -29,6 +29,7 @@ type RuntimeStore = Pick< export interface AgentTextClient { startOnboarding(input: { userId: string }): Promise; + markOnboardingWelcomeDelivered(input: { userId: string }): Promise; setResponseGeneration(input: { connectionId: string; generationToken: string; @@ -36,9 +37,8 @@ export interface AgentTextClient { sendTextBatch(input: { connectionId: string; generationToken: string; - messageIds: string[]; userId: string; - contents: string[]; + turns: Array<{ messageId: string; content: string }>; }): Promise<{ reply: string | null; stale: boolean }>; sendText(input: { connectionId: string; @@ -49,13 +49,13 @@ export interface AgentTextClient { sendImages(input: { connectionId: string; generationToken: string; - messageIds: string[]; userId: string; images: Array<{ + messageId: string; bytes: Uint8Array; mimeType: "image/jpeg" | "image/png" | "image/webp"; }>; - hint?: string; + turns: Array<{ messageId: string; content?: string; imageCount: number }>; }): Promise<{ reply: string | null; stale: boolean }>; sendEvent(input: { connectionId: string; @@ -330,26 +330,31 @@ class WechatTurnBatcher implements WechatTurnBatchSink { const latest = batch.messages.at(-1)!; try { const images = batch.messages - .flatMap((message) => message.images) + .flatMap((message) => message.images.map((image) => ({ + ...image, + messageId: message.messageId + }))) .slice(0, WECHAT_IMAGE_BATCH_MAX); - const contents = batch.messages - .map((message) => message.content) - .filter((value): value is string => Boolean(value)); + const turns = batch.messages.map((message) => ({ + messageId: message.messageId, + ...(message.content ? { content: message.content } : {}), + imageCount: message.images.length + })); const result = images.length > 0 ? await this.dependencies.tomeet.sendImages({ connectionId: this.connection.id, generationToken: latest.generationToken, - messageIds, userId: this.connection.userId, images, - hint: contents.join("\n") || undefined + turns }) : await this.dependencies.tomeet.sendTextBatch({ connectionId: this.connection.id, generationToken: latest.generationToken, - messageIds, userId: this.connection.userId, - contents + turns: turns.flatMap((turn) => turn.content + ? [{ messageId: turn.messageId, content: turn.content }] + : []) }); batch.settled = true; if ( @@ -489,7 +494,11 @@ export async function handleWechatMessage( if (!started) return false; try { - if (!connection.lastMessageAt && message.context_token) { + // iLink needs this first contextual message to open the bot conversation. The activation + // callback may already have delivered the welcome, but the handshake still must not reach + // the Agent as user-authored content. + const isOpeningTrigger = !connection.lastMessageAt && Boolean(message.context_token); + if (message.context_token) { const welcome = await dependencies.tomeet.startOnboarding({ userId: connection.userId }); @@ -502,6 +511,12 @@ export async function handleWechatMessage( contextToken: message.context_token, runIdBase: `first-inbound-welcome-${connection.id}-${id}` }); + await dependencies.tomeet.markOnboardingWelcomeDelivered({ userId: connection.userId }); + } + if (welcome || isOpeningTrigger) { + await dependencies.store.completeWechatMessage(connection.id, id); + connection.lastMessageAt = new Date().toISOString(); + return true; } } const content = WechatILinkClient.extractText(message); diff --git a/apps/wechat-ilink-worker/src/tomeet-client.test.ts b/apps/wechat-ilink-worker/src/tomeet-client.test.ts index d2e3708..88ba456 100644 --- a/apps/wechat-ilink-worker/src/tomeet-client.test.ts +++ b/apps/wechat-ilink-worker/src/tomeet-client.test.ts @@ -145,11 +145,14 @@ describe("TomeetClient", () => { await expect(client.sendImages({ connectionId: "connection-images", generationToken: "generation-images", - messageIds: ["image-1", "image-2"], userId: "25000000-0000-4000-8000-000000000001", images: [ - { bytes: Uint8Array.from([1, 2]), mimeType: "image/jpeg" }, - { bytes: Uint8Array.from([3, 4]), mimeType: "image/png" } + { messageId: "image-1", bytes: Uint8Array.from([1, 2]), mimeType: "image/jpeg" }, + { messageId: "image-2", bytes: Uint8Array.from([3, 4]), mimeType: "image/png" } + ], + turns: [ + { messageId: "image-1", imageCount: 1 }, + { messageId: "image-2", imageCount: 1 } ] })).resolves.toEqual({ reply: "两张图片放在一起看，最吸引你的是哪一处？", @@ -161,8 +164,12 @@ describe("TomeetClient", () => { connectionId: "connection-images", generationToken: "generation-images", images: [ - { dataBase64: "AQI=", mimeType: "image/jpeg" }, - { dataBase64: "AwQ=", mimeType: "image/png" } + { messageId: "image-1", dataBase64: "AQI=", mimeType: "image/jpeg" }, + { messageId: "image-2", dataBase64: "AwQ=", mimeType: "image/png" } + ], + turns: [ + { messageId: "image-1", imageCount: 1, idempotencyKey: expect.stringMatching(/^wechat:[a-f0-9]{64}$/) }, + { messageId: "image-2", imageCount: 1, idempotencyKey: expect.stringMatching(/^wechat:[a-f0-9]{64}$/) } ] }); expect(String(postedBody?.idempotencyKey)).toMatch(/^wechat:[a-f0-9]{64}$/); diff --git a/apps/wechat-ilink-worker/src/tomeet-client.ts b/apps/wechat-ilink-worker/src/tomeet-client.ts index a0062cf..9c63790 100644 --- a/apps/wechat-ilink-worker/src/tomeet-client.ts +++ b/apps/wechat-ilink-worker/src/tomeet-client.ts @@ -76,6 +76,13 @@ export class TomeetClient { return message.content; } + async markOnboardingWelcomeDelivered(input: { userId: string }): Promise { + await this.request( + `/internal/users/${input.userId}/adventurex-onboarding/welcome-delivered`, + { method: "POST" } + ); + } + async sendText(input: { connectionId: string; messageId: string; @@ -127,25 +134,29 @@ export class TomeetClient { async sendTextBatch(input: { connectionId: string; generationToken: string; - messageIds: string[]; userId: string; - contents: string[]; + turns: Array<{ messageId: string; content: string }>; }): Promise { - const content = input.contents.length === 1 - ? input.contents[0]! - : input.contents.map((item, index) => `${index + 1}. ${item}`).join("\n"); + const content = input.turns.length === 1 + ? input.turns[0]!.content + : input.turns.map((item, index) => `${index + 1}. ${item.content}`).join("\n"); const payload = agentMessageInputSchema.parse({ userId: input.userId, displayName: "微信用户", content, - idempotencyKey: idempotencyKey(input.connectionId, `turn:${input.messageIds.join(":")}`) + idempotencyKey: idempotencyKey(input.connectionId, `generation:${input.generationToken}`) }); const response = await this.request<{ job: unknown }>("/internal/agent/messages", { method: "POST", body: JSON.stringify({ ...payload, connectionId: input.connectionId, - generationToken: input.generationToken + generationToken: input.generationToken, + messages: input.turns.map((turn) => ({ + messageId: turn.messageId, + content: turn.content, + idempotencyKey: idempotencyKey(input.connectionId, turn.messageId) + })) }) }); return this.waitForTurnResult(llmJobSchema.parse(response.job)); @@ -154,24 +165,28 @@ export class TomeetClient { async sendImages(input: { connectionId: string; generationToken: string; - messageIds: string[]; userId: string; images: Array<{ + messageId: string; bytes: Uint8Array; mimeType: "image/jpeg" | "image/png" | "image/webp"; }>; - hint?: string; + turns: Array<{ messageId: string; content?: string; imageCount: number }>; }): Promise { - const messageKey = idempotencyKey(input.connectionId, `images:${input.messageIds.join(":")}`); + const messageKey = idempotencyKey(input.connectionId, `images:${input.generationToken}`); const response = await this.request<{ job: unknown }>("/internal/agent/multimodal-inputs", { method: "POST", body: JSON.stringify({ userId: input.userId, images: input.images.map((image) => ({ + messageId: image.messageId, mimeType: image.mimeType, dataBase64: Buffer.from(image.bytes).toString("base64") })), - hint: input.hint, + turns: input.turns.map((turn) => ({ + ...turn, + idempotencyKey: idempotencyKey(input.connectionId, turn.messageId) + })), idempotencyKey: messageKey, connectionId: input.connectionId, generationToken: input.generationToken diff --git a/docs/adventurex-live-matchmaking-implementation-spec.md b/docs/adventurex-live-matchmaking-implementation-spec.md index 3aafc9d..3e21565 100644 --- a/docs/adventurex-live-matchmaking-implementation-spec.md +++ b/docs/adventurex-live-matchmaking-implementation-spec.md @@ -167,6 +167,7 @@ stateDiagram-v2 - 这条首次欢迎语及其英文翻译是明确的硬编码产品例外，MemoryStore 与 Supabase RPC 必须保持完全一致； - 每句话是一个气泡，普通气泡结尾不使用中文句号或英文句点； - 消息按语言使用幂等键； +- 触发首次欢迎语的入站消息只用于开启对话，不再作为用户内容提交给 Agent 回复； - 用户明确要求英文时保存 `preferred_language=en` 并用四个英文气泡重新播放欢迎语，明确要求切回中文时同理； - 用户已开始正常对话时不得再次补发； - 初步了解结束时宽松地问一次有没有雷点、明确边界或不想遇到的情况，并记录 `boundary_prompted_at` 防止重复追问； diff --git a/packages/agent-core/src/index.test.ts b/packages/agent-core/src/index.test.ts index 11fd4d1..b920d83 100644 --- a/packages/agent-core/src/index.test.ts +++ b/packages/agent-core/src/index.test.ts @@ -148,6 +148,7 @@ describe("mock agent intelligence", () => { preferredLanguage: "zh", boundaryPromptedAt: null, welcomeSentAt: nowIso(), + welcomeDeliveredAt: nowIso(), updatedAt: nowIso() } }); @@ -166,6 +167,7 @@ describe("mock agent intelligence", () => { preferredLanguage: "zh", boundaryPromptedAt: null, welcomeSentAt: nowIso(), + welcomeDeliveredAt: nowIso(), updatedAt: nowIso() } }); @@ -193,6 +195,7 @@ describe("mock agent intelligence", () => { preferredLanguage: "zh" as const, boundaryPromptedAt: null, welcomeSentAt: nowIso(), + welcomeDeliveredAt: nowIso(), updatedAt: nowIso() }; const first = await intelligence.reply(buildAgentContext(messages, createDefaultUserModel("u1"), { diff --git a/packages/agent-core/src/index.ts b/packages/agent-core/src/index.ts index b4de586..2c45ee9 100644 --- a/packages/agent-core/src/index.ts +++ b/packages/agent-core/src/index.ts @@ -502,6 +502,7 @@ export class MockAgentIntelligence implements AgentIntelligence { match_confirmation_incomplete: "这次候选没有完成成局确认。你的选择已经收到；如果你愿意，有新的合适安排时我可以主动告诉你。", room_intro: "成局信息已经准备好。", match_expired: "这次匹配已经超时并结束了。如果还想再匹配，告诉我就行。", + match_progress: "我还在处理这次匹配，会继续主动同步进展。", room_change: "当前活动的信息发生了变化，我把最新情况同步给你。", draft_change: "你看过的候选发生了变化，我会按最新情况重新说明。", unsupported_channel_message: "这条消息目前无法读取，你可以换一种方式告诉我。", diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index c9f806b..a406f86 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -38,6 +38,7 @@ export const agentProductEventKindSchema = z.enum([ "match_confirmation_incomplete", "room_intro", "match_expired", + "match_progress", "room_change", "draft_change", "unsupported_channel_message", @@ -102,6 +103,22 @@ export const channelTurnFailureNotice: Record = { en: "I could not process that last batch of messages. Could you send it again?" }; +// Sent one at a time while an Agent turn is still running, so the user can tell the agent +// is working rather than stuck. Also delivered straight over the channel transport: the +// Agent that would otherwise word them is exactly the thing still being awaited. +export const channelTurnProgressNotices: Record = { + zh: [ + "收到，我想一想…", + "还在想，稍等我一下…", + "快好了，马上回你…" + ], + en: [ + "Got it, let me think…", + "Still thinking, give me a moment…", + "Almost there, one second…" + ] +}; + export const adventurexOnboardingStateSchema = z.object({ userId: idSchema, stage: adventurexOnboardingStageSchema, @@ -109,6 +126,7 @@ export const adventurexOnboardingStateSchema = z.object({ preferredLanguage: adventurexLanguageSchema.default("zh"), boundaryPromptedAt: z.string().datetime().nullable().default(null), welcomeSentAt: z.string().datetime().nullable(), + welcomeDeliveredAt: z.string().datetime().nullable().default(null), createdAt: z.string().datetime().optional(), updatedAt: z.string().datetime() }); @@ -497,6 +515,7 @@ export const llmJobTypeSchema = z.enum([ "matchmaking", "match_round_generate", "match_round_settle", + "match_status_notify", "room_change_notify", "feedback_update", "memory_extract", diff --git a/packages/data/src/adventurex.test.ts b/packages/data/src/adventurex.test.ts index 9798b3f..c8cfcff 100644 --- a/packages/data/src/adventurex.test.ts +++ b/packages/data/src/adventurex.test.ts @@ -21,6 +21,10 @@ describe("AdventureX MemoryStore", () => { preferredLanguage: "zh", boundaryPromptedAt: null }); + await store.markAdventurexWelcomeDelivered(userId); + expect(await store.ensureAdventurexOnboardingState(userId)).toMatchObject({ + welcomeDeliveredAt: expect.any(String) + }); }); it("does not inject a welcome into an existing conversation", async () => { @@ -124,6 +128,7 @@ describe("AdventureX MemoryStore", () => { rawText: "1" }); } + expect((await store.listPendingDraftChangeNotifications()).length).toBeGreaterThan(0); const state = await store.getRoundSettlementState(round.roundId); const roomIds = await store.settleMatchRound(round.roundId, [{ draftId: state.drafts[0]!.draftId, @@ -180,9 +185,9 @@ describe("AdventureX MemoryStore", () => { await expect(store.leaveRoom(joined.roomId, userIds[3]!)).rejects.toThrow("理由"); await store.leaveRoom(joined.roomId, userIds[3]!, "临时有事"); expect(await store.getLatestMatchRequestForUser(userIds[3]!)).toMatchObject({ - status: "matching", - phase: "watching", - proactivePushEnabled: true, + status: "cancelled", + phase: "waiting", + proactivePushEnabled: false, roomId: null }); expect((await store.listSuitableOpenRooms(userIds[3]!)).map((item) => item.roomId)).not.toContain(joined.roomId); diff --git a/packages/data/src/memory-store.ts b/packages/data/src/memory-store.ts index 691c2a0..3b6d949 100644 --- a/packages/data/src/memory-store.ts +++ b/packages/data/src/memory-store.ts @@ -134,6 +134,7 @@ export class MemoryStore implements DataStore { preferredLanguage: "zh", boundaryPromptedAt: null, welcomeSentAt: null, + welcomeDeliveredAt: null, createdAt: now, updatedAt: now }; @@ -238,6 +239,17 @@ export class MemoryStore implements DataStore { return message; } + async markAdventurexWelcomeDelivered(userId: string): Promise { + await this.ensureUser(userId); + const state = this.onboardingStates.get(userId)!; + const now = new Date().toISOString(); + state.welcomeDeliveredAt ??= now; + state.welcomeSentAt ??= now; + if (state.stage === "new") state.stage = "awaiting_image_or_text"; + state.updatedAt = now; + return structuredClone(state); + } + async updateAdventurexOnboardingState( userId: string, patch: { @@ -890,7 +902,7 @@ export class MemoryStore implements DataStore { candidateRequestIds: [...proposal.candidateRequestIds], rationale: proposal.rationale, createdAt: new Date().toISOString(), - expiresAt: input.offerExpiresAt + expiresAt: input.offerExpiresAt ?? new Date(Date.now() + 10 * 60_000).toISOString() }); this.draftTempKeys.set(`${input.roundId}:${proposal.tempDraftId}`, draftId); } @@ -940,6 +952,48 @@ export class MemoryStore implements DataStore { return saved; } + async activateMatchOfferWindow( + requestId: string, + roundId: string, + windowSeconds: number + ): Promise { + const request = this.matchRequests.get(requestId); + if ( + !request + || request.status !== "matching" + || request.activeRoundId !== roundId + || request.phase !== "offered" + ) { + throw new StoreConflictError("候选当前不能开始选择计时"); + } + if (!Number.isInteger(windowSeconds) || windowSeconds < 10 || windowSeconds > 600) { + throw new StoreConflictError("候选选择窗口必须在 10-600 秒之间"); + } + const expiresAt = new Date(Date.now() + windowSeconds * 1_000).toISOString(); + request.optionsExpiresAt = expiresAt; + request.updatedAt = new Date().toISOString(); + const round = this.rounds.get(roundId); + if (round) { + round.offerExpiresAt = !round.offerExpiresAt || round.offerExpiresAt < expiresAt + ? expiresAt + : round.offerExpiresAt; + round.updatedAt = request.updatedAt; + } + for (const draft of this.drafts.values()) { + if (draft.roundId === roundId && draft.status === "collecting" && draft.expiresAt < expiresAt) { + draft.expiresAt = expiresAt; + } + } + await this.enqueueJob({ + type: "match_round_settle", + payload: { roundId }, + idempotencyKey: `match-round-settle:${roundId}:${Math.floor(new Date(expiresAt).getTime() / 1_000)}`, + partitionKey: `match-round:${roundId}`, + runAt: expiresAt + }); + return structuredClone(request); + } + async listCurrentMatchOptions(userId: string): Promise { const request = [...this.matchRequests.values()] .filter((item) => item.userId === userId && item.status === "matching" && ["offered", "selected"].includes(item.phase)) @@ -968,8 +1022,8 @@ export class MemoryStore implements DataStore { async saveMatchChoices(requestId: string, input: SaveMatchChoicesInput): Promise { const request = this.matchRequests.get(requestId); if (!request || request.status !== "matching" || !request.activeRoundId) throw new StoreConflictError("匹配请求当前不能选择"); - if (request.optionsExpiresAt && new Date(request.optionsExpiresAt).getTime() <= Date.now()) { - throw new StoreConflictError("候选已过期"); + if (!request.optionsExpiresAt || new Date(request.optionsExpiresAt).getTime() <= Date.now()) { + throw new StoreConflictError("候选尚未送达或已过期"); } const acceptedNumbers = [...new Set(input.acceptedOptionNumbers)]; const currentOffers = [...this.offers.values()].filter( @@ -981,6 +1035,13 @@ export class MemoryStore implements DataStore { if (input.requiredHookIds.some((hookId) => !allowedRequiredHooks.has(hookId))) { throw new StoreConflictError("required hook 必须来自已接受候选"); } + const previousDraftIds = new Set([...this.choices.values()] + .filter((choice) => + choice.requestId === requestId + && choice.roundId === request.activeRoundId + && choice.draftId + ) + .map((choice) => choice.draftId!)); for (const [choiceId, choice] of this.choices) { if (choice.requestId === requestId && choice.roundId === request.activeRoundId) this.choices.delete(choiceId); } @@ -1010,6 +1071,28 @@ export class MemoryStore implements DataStore { } request.phase = "selected"; request.updatedAt = now; + for (const choice of saved.filter((item) => item.draftId && !previousDraftIds.has(item.draftId))) { + const draft = this.drafts.get(choice.draftId!); + if (!draft || draft.status !== "collecting") continue; + draft.version += 1; + const eventId = randomUUID(); + const payload = { joinedUserId: request.userId }; + const acceptedUserIds = [...this.choices.values()] + .filter((item) => item.draftId === choice.draftId && item.requestId !== requestId) + .map((item) => this.matchRequests.get(item.requestId)?.userId) + .filter((userId): userId is string => Boolean(userId)); + for (const userId of new Set(acceptedUserIds)) { + const notification: DraftChangeNotification = { + eventId, + draftId: choice.draftId!, + userId, + changeType: "confirmed_member_joined", + payload, + idempotencyKey: `draft-change:${eventId}:${userId}` + }; + this.draftNotifications.set(`${eventId}:${userId}`, notification); + } + } return saved; } @@ -1579,8 +1662,9 @@ export class MemoryStore implements DataStore { } const request = [...this.matchRequests.values()].find((item) => item.roomId === roomId && item.userId === userId); if (request) { - request.status = request.proactivePushEnabled ? "matching" : "cancelled"; - request.phase = request.proactivePushEnabled ? "watching" : "waiting"; + request.status = "cancelled"; + request.phase = "waiting"; + request.proactivePushEnabled = false; request.roomId = null; request.activeRoundId = null; request.optionsExpiresAt = null; @@ -1714,6 +1798,12 @@ export class MemoryStore implements DataStore { throw new StoreConflictError("Web 对话消息不能投递到微信"); } this.wechatOutboundMessages.set(message.id, structuredClone(message)); + if (message.id.startsWith("match-options:")) { + const [, roundId, requestId] = message.id.split(":"); + if (roundId && requestId) { + await this.activateMatchOfferWindow(requestId, roundId, 90); + } + } } async enqueueJob(input: EnqueueJobInput): Promise { diff --git a/packages/data/src/migration.test.ts b/packages/data/src/migration.test.ts index 300e6c0..091690a 100644 --- a/packages/data/src/migration.test.ts +++ b/packages/data/src/migration.test.ts @@ -789,9 +789,9 @@ describe("Supabase migration", () => { `, [roomId, userIds[0]]); expect(authorizedExit.rows[0]).toEqual({ withdrawal_reason: "临时有事", - status: "matching", - phase: "watching", - proactive_push_enabled: true, + status: "cancelled", + phase: "waiting", + proactive_push_enabled: false, room_id: null, active_round_id: null, options_expires_at: null diff --git a/packages/data/src/store.ts b/packages/data/src/store.ts index 449df61..5dde052 100644 --- a/packages/data/src/store.ts +++ b/packages/data/src/store.ts @@ -61,7 +61,7 @@ export interface SaveRoundPlanInput { roundId: string; proposal: MatchRoundProposal | null; offers: PreparedMatchOffer[]; - offerExpiresAt: string; + offerExpiresAt: string | null; } export interface RoundSettlementState { @@ -135,6 +135,7 @@ export interface DataStore { ensureUser(userId: string, displayName?: string): Promise; ensureAdventurexOnboardingState(userId: string): Promise; startAdventurexOnboarding(userId: string, language?: AdventurexLanguage): Promise; + markAdventurexWelcomeDelivered(userId: string): Promise; updateAdventurexOnboardingState( userId: string, patch: { @@ -223,6 +224,11 @@ export interface DataStore { addRequestToRound(roundId: string, requestId: string): Promise; listRoundCandidates(roundId: string): Promise; saveRoundProposals(input: SaveRoundPlanInput): Promise; + activateMatchOfferWindow( + requestId: string, + roundId: string, + windowSeconds: number + ): Promise; listCurrentMatchOptions(userId: string): Promise; saveMatchChoices(requestId: string, input: SaveMatchChoicesInput): Promise; expireMatchOptions(requestId: string): Promise; diff --git a/packages/data/src/supabase-store.ts b/packages/data/src/supabase-store.ts index 2f9ce6b..84c7971 100644 --- a/packages/data/src/supabase-store.ts +++ b/packages/data/src/supabase-store.ts @@ -191,6 +191,9 @@ function mapOnboardingState(row: JsonRow): AdventurexOnboardingState { welcomeSentAt: normalizeDateTime( row.adventurex_welcome_sent_at ?? row.welcome_sent_at ?? row.welcomeSentAt ?? null ), + welcomeDeliveredAt: normalizeDateTime( + row.adventurex_welcome_delivered_at ?? row.welcome_delivered_at ?? row.welcomeDeliveredAt ?? null + ), createdAt: normalizeDateTime( row.adventurex_state_created_at ?? row.created_at ?? row.createdAt ), @@ -358,6 +361,14 @@ export class SupabaseStore implements DataStore { return message ? mapMessage(message) : null; } + async markAdventurexWelcomeDelivered(userId: string): Promise { + const { data, error } = await this.client.rpc("mark_adventurex_welcome_delivered", { + p_user_id: userId + }); + if (error) this.throwError("标记 AdventureX 欢迎语已投递", error); + return mapOnboardingState(unwrapRpcData(data) as JsonRow); + } + async updateAdventurexOnboardingState( userId: string, patch: { @@ -881,6 +892,20 @@ export class SupabaseStore implements DataStore { return ((data ?? []) as JsonRow[]).map((row) => mapOffer(row)); } + async activateMatchOfferWindow( + requestId: string, + roundId: string, + windowSeconds: number + ): Promise { + const { data, error } = await this.client.rpc("activate_match_offer_window", { + p_request_id: requestId, + p_round_id: roundId, + p_window_seconds: windowSeconds + }); + if (error) this.throwError("启动候选选择窗口", error); + return mapMatchRequest(unwrapRpcData(data) as JsonRow); + } + async listCurrentMatchOptions(userId: string): Promise { const { data, error } = await this.client.rpc("list_current_match_options", { p_user_id: userId }); if (error) this.throwError("读取当前候选", error); diff --git a/packages/intelligence/src/adventurex-flow.test.ts b/packages/intelligence/src/adventurex-flow.test.ts index 4d8af4a..a283ce5 100644 --- a/packages/intelligence/src/adventurex-flow.test.ts +++ b/packages/intelligence/src/adventurex-flow.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { MockAgentIntelligence } from "@tomeet/agent-core"; import { MemoryStore } from "@tomeet/data"; import { @@ -8,9 +8,14 @@ import { selectNonOverlappingGroups, validateMatchRoundProposal } from "@tomeet/matchmaking"; -import { JobProcessor, scheduleAdventurexMatchRequest } from "./index.js"; +import { + JobProcessor, + scheduleAdventurexMatchRequest, + scheduleMatchStatusHeartbeat +} from "./index.js"; describe("AdventureX 12-user integration", () => { + afterEach(() => vi.useRealTimers()); it("offers real drafts and settles at least two non-overlapping confirmed rooms", async () => { const store = new MemoryStore(); const matcher = new MockMatchmakingIntelligence(); @@ -116,7 +121,7 @@ describe("AdventureX 12-user integration", () => { previewText: "**1｜故事交换桌**\n你可能遇见其他候选成员。", hooks: [] }], - offerExpiresAt: new Date(Date.now() + 90_000).toISOString() + offerExpiresAt: new Date(Date.now() - 1).toISOString() }); const job = await store.enqueueJob({ type: "match_round_settle", @@ -137,7 +142,35 @@ describe("AdventureX 12-user integration", () => { expect(await store.listRoundCandidates(round.roundId)).toHaveLength(0); }); + it("publishes a ten-second matching heartbeat and schedules the next one", async () => { + vi.useFakeTimers(); + const now = new Date("2026-07-25T12:00:00.000Z"); + vi.setSystemTime(now); + const store = new MemoryStore(); + const processor = new JobProcessor( + store, + new MockAgentIntelligence(), + new MockMatchmakingIntelligence(), + { adventurexMatchingV1: true } + ); + const userId = randomUUID(); + await store.ensureUser(userId, "心跳用户"); + const request = await store.createMatchRequest(userId, { rawText: "现在开始匹配" }); + const heartbeat = await scheduleMatchStatusHeartbeat(store, request.requestId, { now }); + expect(heartbeat.runAt).toBe("2026-07-25T12:00:10.000Z"); + + vi.setSystemTime(new Date(now.getTime() + 10_000)); + const result = await processor.process(heartbeat); + expect(result).toMatchObject({ sent: true, phase: "waiting", nextJobId: expect.any(String) }); + const progress = (await store.listRecentMessages(userId, 10)) + .find((message) => message.id.startsWith(`match-progress:${request.requestId}:`)); + expect(progress?.content).toContain("处理"); + }); + it("ends an unformed confirmation neutrally and prioritizes the accepting user among watchers", async () => { + vi.useFakeTimers(); + const now = new Date("2026-07-25T12:00:00.000Z"); + vi.setSystemTime(now); const store = new MemoryStore(); const processor = new JobProcessor( store, @@ -177,7 +210,7 @@ describe("AdventureX 12-user integration", () => { previewText: "**1｜故事交换桌**\n等待其他候选成员确认。", hooks: [] })), - offerExpiresAt: new Date(Date.now() + 90_000).toISOString() + offerExpiresAt: new Date(now.getTime() + 90_000).toISOString() }); await store.saveMatchChoices(users[0]!.requestId, { preferredOptionNumber: 1, @@ -192,6 +225,7 @@ describe("AdventureX 12-user integration", () => { partitionKey: `match-round:${round.roundId}` }); + vi.setSystemTime(new Date(now.getTime() + 91_000)); await processor.process(settleJob); expect(await store.getMatchRequest(users[0]!.requestId)).toMatchObject({ @@ -326,9 +360,15 @@ describe("AdventureX 12-user integration", () => { requiredHookIds: [], rawText: "1" }); - const settleJob = await store.getJob(String(result.settleJobId)); - expect(settleJob).not.toBeNull(); - const settled = await processor.process(settleJob!); + vi.useFakeTimers(); + vi.setSystemTime(new Date(Date.now() + 91_000)); + const settleJob = await store.enqueueJob({ + type: "match_round_settle", + payload: { roundId: scheduled.round.roundId }, + idempotencyKey: `virtual-settle-test:${scheduled.round.roundId}`, + partitionKey: `match-round:${scheduled.round.roundId}` + }); + const settled = await processor.process(settleJob); expect(settled.roomIds).toHaveLength(1); const room = await store.getRoom(String((settled.roomIds as string[])[0])); expect(room?.members).toHaveLength(5); @@ -365,7 +405,8 @@ describe("AdventureX 12-user integration", () => { const resumed = await processor.process(job); expect(first.offerCount).toBe(3); - expect(resumed).toMatchObject({ resumed: true, offerCount: 3, settleJobId: first.settleJobId }); + expect(resumed).toMatchObject({ resumed: true, offerCount: 3 }); + expect(resumed.settleJobId).toEqual(expect.any(String)); for (const [index, requestId] of requestIds.entries()) { expect(await store.getMatchRequest(requestId)).toMatchObject({ status: "matching", phase: "offered" }); const messages = await store.listRecentMessages(userIds[index]!, 10); diff --git a/packages/intelligence/src/agent-policy.ts b/packages/intelligence/src/agent-policy.ts index 57cbc5f..0f95af7 100644 --- a/packages/intelligence/src/agent-policy.ts +++ b/packages/intelligence/src/agent-policy.ts @@ -69,6 +69,7 @@ export function defaultDeliveryClassForEvent(kind: AgentProductEventKind): Outbo case "match_confirmation_incomplete": case "room_intro": case "match_expired": + case "match_progress": case "room_change": case "draft_change": case "legacy_match_ready": diff --git a/packages/intelligence/src/hosted-llm.test.ts b/packages/intelligence/src/hosted-llm.test.ts index dcbe8e8..f9b574c 100644 --- a/packages/intelligence/src/hosted-llm.test.ts +++ b/packages/intelligence/src/hosted-llm.test.ts @@ -355,6 +355,68 @@ describe("hosted Agent web search", () => { }); describe("hosted Agent exploration pressure", () => { + it("normalizes common structured text shapes without failing the Agent job", async () => { + const lookupMemories = vi.fn(async () => []); + const requestBodies = stubChatResponses( + plannedReply({ + replyDraft: ["好，我记下了", "那我们继续往前走"], + memoryPlan: { + queries: [{ query: "用户明确表达过的摄影经历" }], + reviewSuggested: false + }, + socialHooks: [{ + hookText: { text: "平时会拍街头摄影" }, + evidenceMessageIds: "message-1" + }], + searchPlan: { required: false, queries: [] } + }), + { + reply: ["好，我记下了", "那我们继续往前走"], + usedSourceIndexes: [], + usedMemoryIds: [] + }, + { + status: "verified", + reply: { paragraphs: ["好，我记下了", "那我们继续往前走"] }, + issues: [{ message: "无需修改" }], + usedSourceIndexes: [], + usedMemoryIds: [] + } + ); + + const insight = await hostedWithSearch().reply( + agentContext(), + "我平时会拍街头摄影", + lookupMemories, + "message-1" + ); + + expect(insight.reply).toBe("好，我记下了\n\n那我们继续往前走"); + expect(insight.socialHooks).toEqual([{ + hookText: "平时会拍街头摄影", + evidenceMessageIds: ["message-1"] + }]); + expect(lookupMemories).toHaveBeenCalledWith(["用户明确表达过的摄影经历"]); + expect(requestBodies).toHaveLength(3); + }); + + it("reports the exact stage and field after deterministic repair is exhausted", async () => { + const invalidPlan = plannedReply({ + replyDraft: { unexpected: true }, + searchPlan: { required: false, queries: [] } + }); + const requestBodies = stubChatResponses(invalidPlan, invalidPlan, invalidPlan); + + await expect(hostedWithSearch().reply(agentContext(), "没什么特别介意的")) + .rejects.toThrow( + "LLM 结构化输出校验失败 stage=agent_reply.plan: replyDraft invalid_type expected=string received=object" + ); + + expect(requestBodies).toHaveLength(3); + const repairRequest = JSON.parse(requestBodies[1]!) as { temperature: number }; + expect(repairRequest.temperature).toBe(0); + }); + it("forbids mirror replies in planning and rewrites one at the publish gate", async () => { const requestBodies = stubChatResponses( plannedReply({ @@ -504,6 +566,7 @@ describe("hosted Agent proactive matching actions", () => { preferredLanguage: "zh", boundaryPromptedAt: null, welcomeSentAt: now, + welcomeDeliveredAt: now, updatedAt: now } }); diff --git a/packages/intelligence/src/hosted-llm.ts b/packages/intelligence/src/hosted-llm.ts index aa3f271..c8f1daa 100644 --- a/packages/intelligence/src/hosted-llm.ts +++ b/packages/intelligence/src/hosted-llm.ts @@ -144,6 +144,180 @@ const multimodalInsightSchema = z.object({ recentImpression: z.string().min(1).max(4_000) }).passthrough(); +type CandidateNormalizer = (candidate: unknown) => unknown; + +interface ParseOrRepairOptions { + model?: string; + temperature?: number; + stage?: string; + normalize?: CandidateNormalizer; +} + +interface StructuredOutputIssue { + path: string; + code: string; + expected?: string; + received?: string; +} + +export class StructuredOutputValidationError extends Error { + constructor( + readonly stage: string, + readonly issues: StructuredOutputIssue[], + cause: z.ZodError + ) { + const summary = issues.map((issue) => { + const typeMismatch = issue.expected + ? ` expected=${issue.expected}${issue.received ? ` received=${issue.received}` : ""}` + : ""; + return `${issue.path || ""} ${issue.code}${typeMismatch}`; + }).join("; "); + super(`LLM 结构化输出校验失败 stage=${stage}: ${summary}`, { cause }); + this.name = "StructuredOutputValidationError"; + } +} + +function isRecord(value: unknown): value is Record { + return Boolean(value) && typeof value === "object" && !Array.isArray(value); +} + +function normalizeTextValue(value: unknown, joiner = "\n\n"): unknown { + if (typeof value === "string") return value; + if (Array.isArray(value)) { + const parts = value.map((item) => normalizeTextValue(item, " ")); + return parts.every((item): item is string => typeof item === "string") + ? parts.join(joiner) + : value; + } + if (!isRecord(value)) return value; + for (const key of [ + "text", + "content", + "value", + "query", + "message", + "reason", + "summary", + "hookText", + "reply", + "replyDraft", + "answer" + ]) { + if (typeof value[key] === "string") return value[key]; + } + for (const key of ["paragraphs", "bubbles", "lines"]) { + if (Array.isArray(value[key])) return normalizeTextValue(value[key], joiner); + } + return value; +} + +function normalizeStringArray(value: unknown): unknown { + if (typeof value === "string") return [value]; + if (Array.isArray(value)) { + const items = value.map((item) => normalizeTextValue(item, " ")); + return items.every((item): item is string => typeof item === "string") ? items : value; + } + const singleton = normalizeTextValue(value, " "); + return typeof singleton === "string" ? [singleton] : value; +} + +function normalizeIdentifierArray(value: unknown): unknown { + if (typeof value === "string") return [value]; + if (isRecord(value) && typeof value.id === "string") return [value.id]; + if (!Array.isArray(value)) return value; + const items = value.map((item) => { + if (typeof item === "string") return item; + return isRecord(item) && typeof item.id === "string" ? item.id : item; + }); + return items.every((item): item is string => typeof item === "string") ? items : value; +} + +function normalizeConversationPlanCandidate(candidate: unknown): unknown { + if (!isRecord(candidate)) return candidate; + const normalized: Record = { + ...candidate, + replyDraft: normalizeTextValue(candidate.replyDraft) + }; + + if (isRecord(candidate.memoryPlan)) { + normalized.memoryPlan = { + ...candidate.memoryPlan, + queries: normalizeStringArray(candidate.memoryPlan.queries) + }; + } + if (isRecord(candidate.searchPlan) && Array.isArray(candidate.searchPlan.queries)) { + normalized.searchPlan = { + ...candidate.searchPlan, + queries: candidate.searchPlan.queries.map((query) => isRecord(query) + ? { + ...query, + query: normalizeTextValue(query.query, " "), + ...(query.timeRange === undefined + ? {} + : { timeRange: normalizeTextValue(query.timeRange, " ") }) + } + : query) + }; + } + if (Array.isArray(candidate.socialHooks)) { + normalized.socialHooks = candidate.socialHooks.map((hook) => isRecord(hook) + ? { + ...hook, + hookText: normalizeTextValue(hook.hookText, " "), + evidenceMessageIds: normalizeIdentifierArray(hook.evidenceMessageIds) + } + : hook); + } + if (Array.isArray(candidate.actions)) { + normalized.actions = candidate.actions.map((action) => { + if (!isRecord(action)) return action; + const next: Record = { ...action }; + for (const key of ["rawText", "reason", "peopleFeedback", "gameFeedback", "nextIntent"]) { + if (key in next) next[key] = normalizeTextValue(next[key], " "); + } + if ("requiredHookIds" in next) next.requiredHookIds = normalizeIdentifierArray(next.requiredHookIds); + if ("connectionUserIds" in next) next.connectionUserIds = normalizeIdentifierArray(next.connectionUserIds); + if (typeof next.intent === "string") { + next.intent = { rawText: next.intent }; + } else if (isRecord(next.intent) && "rawText" in next.intent) { + next.intent = { ...next.intent, rawText: normalizeTextValue(next.intent.rawText, " ") }; + } + return next; + }); + } + return normalized; +} + +function normalizeReplyCandidate(candidate: unknown): unknown { + if (!isRecord(candidate)) return candidate; + return { + ...candidate, + ...(candidate.reply === undefined ? {} : { reply: normalizeTextValue(candidate.reply) }), + ...(candidate.content === undefined ? {} : { content: normalizeTextValue(candidate.content) }), + ...(candidate.summary === undefined ? {} : { summary: normalizeTextValue(candidate.summary) }), + ...(candidate.issues === undefined ? {} : { issues: normalizeStringArray(candidate.issues) }), + ...(candidate.usedMemoryIds === undefined + ? {} + : { usedMemoryIds: normalizeIdentifierArray(candidate.usedMemoryIds) }), + ...(Array.isArray(candidate.optionPreviews) + ? { + optionPreviews: candidate.optionPreviews.map((preview) => isRecord(preview) + ? { ...preview, text: normalizeTextValue(preview.text, " ") } + : preview) + } + : {}) + }; +} + +function structuredOutputIssues(error: z.ZodError): StructuredOutputIssue[] { + return error.issues.map((issue) => ({ + path: issue.path.map(String).join("."), + code: issue.code, + ...("expected" in issue && typeof issue.expected === "string" ? { expected: issue.expected } : {}), + ...("received" in issue && typeof issue.received === "string" ? { received: issue.received } : {}) + })); +} + const leftFrameTitles = { match_options: "TOMEET 组局邀请", room_intro: "TOMEET 成局确认函" @@ -221,14 +395,17 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte result: unknown, contract: string, source: unknown, - model = this.options.textModel, - temperature = 0.3 + options: ParseOrRepairOptions = {} ): Promise { - let candidate = result; + const normalize = options.normalize ?? ((candidate: unknown) => candidate); + const model = options.model ?? this.options.textModel; + const temperature = options.temperature ?? 0; + const stage = options.stage ?? "unknown"; + let candidate = normalize(result); for (let attempt = 0; attempt < 2; attempt += 1) { const parsed = schema.safeParse(candidate); if (parsed.success) return parsed.data; - candidate = await this.chatJson( + candidate = normalize(await this.chatJson( [ "你是 JSON 契约修复器。只修复结构和与状态冲突的 action，不新增没有证据的动作。", contract, @@ -241,9 +418,11 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte }), model, temperature - ); + )); } - return schema.parse(candidate); + const parsed = schema.safeParse(candidate); + if (parsed.success) return parsed.data; + throw new StructuredOutputValidationError(stage, structuredOutputIssues(parsed.error), parsed.error); } private emitWebSearchEvent(event: WebSearchEvent): void { @@ -348,7 +527,8 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte userQuestion: userContent, memoryIds: memories.map((memory) => memory.id), evidenceCount: search.results.length - } + }, + { stage: "agent_reply.grounding", normalize: normalizeReplyCandidate } ); candidateReply = grounded.reply; candidateUsedMemoryIds = grounded.usedMemoryIds; @@ -412,8 +592,20 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte this.options.textModel, 0 ); - verified = verifiedReplySchema.parse(verificationResult); + verified = await this.parseOrRepair( + verifiedReplySchema, + verificationResult, + "只输出 status、reply:string、issues:string[]、usedMemoryIds:string[] 和 usedSourceIndexes:number[]。reply 必须是单个字符串，多气泡用两个换行符分隔，不得输出字符串数组。", + { + userMessage: userContent, + candidateReply, + memoryIds: memories.map((memory) => memory.id), + evidenceCount: search.results.length + }, + { stage: "agent_reply.verification", normalize: normalizeReplyCandidate } + ); } catch (error) { + if (error instanceof StructuredOutputValidationError) throw error; throw new Error("发布前事实校验失败", { cause: error }); } const memoryIds = new Set(memories.map((memory) => memory.id)); @@ -489,7 +681,7 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte "如果上一轮刚询问退出理由，用户当前只回复一个简单理由，也视为继续完成退出。确认函之前的受邀成员仍可直接退出。", "runtime.matchRequest.phase=push_consent 表示本次具体尝试已经结束，可能是没有足够合适的候选，也可能是用户已经选择但候选最终未成局，当前正在征求未来主动推送授权。用户明确同意以后有合适的主动告诉他时输出 enable_match_push；用户明确要求现在立即重新匹配时输出 activate_match；明确拒绝继续留意时输出 disable_match_push。", "runtime.matchRequest.phase=watching 表示用户已经授权未来主动推送，但当前没有占用实时匹配优先级。用户明确说现在就想再匹配时输出 activate_match；用户要求停止留意或停止推送时输出 disable_match_push。不要把普通寒暄误判为重新激活。", - "cancel_match 的回复可以结合上下文询问用户是否希望重新匹配，但不要未经同意直接重启。leave_room 的回复绝不能询问或暗示重新匹配：proactivePushEnabled=true 时说明退出后回到被动留意状态，有真正合适的安排再主动通知；否则说明本次组局到此结束。", + "cancel_match 的回复可以结合上下文询问用户是否希望重新匹配，但不要未经同意直接重启。leave_room 后本次请求一律结束，不进入 watching 或任何自动匹配；可以说明只有用户之后明确提出重新匹配才会开始新请求。", this.options.adventurexMatchingV1 ? "AdventureX V1 接受候选即加入并确认，禁止输出 confirm_room。runtime.room.status=confirming 只表示人数尚未达到活动最低要求、仍在补位；只允许 leave_room 或 actions=[]，向用户说明无需再次确认。" : "只有用户明确接受当前 confirming 房间时才输出 confirm_room。", @@ -561,10 +753,12 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte "memoryPlan 必须包含 queries:string[] 和 reviewSuggested:boolean；queries 最多 2 条。", "socialHooks 最多 4 条，每条包含 hookText 和 evidenceMessageIds；证据 ID 只能来自输入。", "searchPlan.required=false 时 queries 必须为空；required=true 时 queries 必须有 1–2 个 {query, topic, timeRange?}。", + "replyDraft 必须是单个字符串，多气泡使用两个换行符分隔，绝不能输出字符串数组。memoryPlan.queries 的每一项必须直接是字符串，不能写成 {query:...}。", "如果 type=submit_feedback，peopleFeedback、gameFeedback、connectionUserIds、nextIntent 必须与 action 同级。", actionPolicy ].join("\n"), - { newMessage: userContent, userMessageId, runtime: context.promptRuntime, roomStatus: context.room?.status ?? null } + { newMessage: userContent, userMessageId, runtime: context.promptRuntime, roomStatus: context.room?.status ?? null }, + { stage: "agent_reply.plan", normalize: normalizeConversationPlanCandidate } ); insight = normalizeRoomExitReason(insight, userContent); if (insight.actions.some((action) => !isActionAllowed(action, context, userContent, this.options.adventurexMatchingV1 === true))) { @@ -577,7 +771,13 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte ].join("\n"), JSON.stringify({ output: insight, newMessage: userContent }) ); - insight = normalizeRoomExitReason(plannedConversationInsightSchema.parse(corrected), userContent); + insight = normalizeRoomExitReason(await this.parseOrRepair( + plannedConversationInsightSchema, + corrected, + "保持完整 Agent 规划结构，只修正当前状态不允许的 actions。replyDraft 必须是字符串，memoryPlan.queries 必须是字符串数组。", + { newMessage: userContent, roomStatus: context.room?.status ?? null, actionPolicy }, + { stage: "agent_reply.action_correction", normalize: normalizeConversationPlanCandidate } + ), userContent); } if (insight.actions.some((action) => !isActionAllowed(action, context, userContent, this.options.adventurexMatchingV1 === true))) { throw new Error("模型返回了当前状态不允许的产品动作"); @@ -623,9 +823,11 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte "match_confirmation_incomplete 表示用户已经做出选择，但这次候选最终没有成局。先确认用户的选择已收到，再中性说明本次安排没有完成成局确认；不得说或暗示某个具体用户拒绝了他，不得归因于用户不够合适，也不得虚构拒绝原因。currentAttemptEnded=true 时要明确这次具体尝试已经结束。canEnableProactivePush=true 时可以询问是否授权未来主动推送；proactivePushAlreadyEnabled=true 时说明会继续留意。followUpPriority=confirmation_follow_up 表示之后再次出现合格机会时，该用户在 watching 用户中优先，但仍不得承诺一定或立即成局。", "room_intro 只能描述最终已确认成员和当前房间事实，不能使用查看者自己的人物事实。room_intro 的 content 使用与 match_options 相同的无右边框字符卡片，但第二行只能是‘┃ TOMEET 成局确认函’；正文以结构化事实自然组织，不得为了排版补充不存在的信息。", "match_expired 用于候选窗口内没有完成选择等真正超时情形；不得把用户已经选择但未成局描述成用户超时，也不得声称系统会自动重新匹配。", + "match_progress 是系统主动处理期间的短状态反馈。只说明仍在处理、当前不需要用户操作，不虚构已找到候选、人数、完成比例或预计完成时间；每次措辞应自然变化，避免机械重复。", "room_change 和 draft_change 只说明输入中真实发生的变化，并自然给出可用的下一步，不替用户做决定。room_change 中 currentlyFormed=false 时必须明确当前人数暂未达到活动最低人数，不能继续说已经成局；可以说明系统正在留意合适补位。", "legacy_match_ready 只使用当前房间、活动和成员事实。unsupported_channel_message 只说明能力边界并邀请用户换一种可处理的表达，facts.supportedInputs 里已有的输入方式不得说成不支持。", "channel_media_unreadable 表示用户确实发来了 facts.receivedKinds 里的内容，只是这一次没有取到。必须承认收到了，说明这一次没读出来并邀请重发一次；不得说不支持这种内容，不得复述内部错误原文，不得凭空猜测内容。optionPreviews=[]。", + "match_progress 只能保留仍在处理且用户无需操作这一事实，不得增加候选、人数、完成比例或完成时间。", "只输出 JSON：{\"content\":\"...\",\"optionPreviews\":[{\"optionNumber\":1,\"text\":\"...\"}]}。非 match_options 时 optionPreviews=[]。" ].join("\n"), JSON.stringify({ @@ -638,7 +840,8 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte }) ), "只使用输入事实，返回可直接发送的个性化 content；match_options 的 optionPreviews 编号必须与输入完全一致。", - { eventKind: event.kind, expectedOptionNumbers } + { eventKind: event.kind, expectedOptionNumbers }, + { stage: "agent_event.draft", normalize: normalizeReplyCandidate } ); const result = await this.parseOrRepair( agentProductMessageSchema, @@ -666,7 +869,8 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte 0 ), "发布前移除无事实依据的内容；候选编号必须完整且唯一，非候选事件 optionPreviews=[]。", - { eventKind: event.kind, expectedOptionNumbers } + { eventKind: event.kind, expectedOptionNumbers }, + { stage: "agent_event.verification", normalize: normalizeReplyCandidate } ); if (event.kind === "match_options") { const actual = result.optionPreviews.map((option) => option.optionNumber).sort(); @@ -696,7 +900,8 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte z.object({ summary: z.string().min(1).max(4_000) }), result, "只输出 {summary:string}。", - { previousSummary, newMessages: messages } + { previousSummary, newMessages: messages }, + { stage: "conversation_summary", normalize: normalizeReplyCandidate } )).summary; } @@ -736,7 +941,7 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte result, "只输出 observableDetails、uncertainty、personCues、suggestedQuestion，不要输出面向用户的回复。", { kind: input.kind, hint: input.hint }, - this.options.visionModel ?? this.options.textModel + { stage: "multimodal.image", model: this.options.visionModel ?? this.options.textModel } ); return { ...parsed, @@ -1003,8 +1208,7 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte result, "只输出 drafts 和 userOptions；所有 ID 必须来自输入。", input, - this.options.textModel, - 0.1 + { stage: "match_round.proposal", model: this.options.textModel, temperature: 0.1 } ); } @@ -1042,8 +1246,7 @@ export class HostedLlmIntelligence implements AgentIntelligence, MatchmakingInte result, "只输出 verdict、isolationRiskUserIds、reasoning；ID 必须来自输入。", input, - this.options.textModel, - 0 + { stage: "match_round.judgement", model: this.options.textModel, temperature: 0 } ); } } diff --git a/packages/intelligence/src/index.ts b/packages/intelligence/src/index.ts index 686ab26..d521bda 100644 --- a/packages/intelligence/src/index.ts +++ b/packages/intelligence/src/index.ts @@ -122,9 +122,29 @@ export async function scheduleAdventurexMatchRequest( partitionKey: `match-round:${round.roundId}`, runAt: scheduledAt }); + if (request.intentSnapshot.virtualTestUser !== true) { + await scheduleMatchStatusHeartbeat(store, request.requestId, { now: options.now }); + } return { round, job }; } +export async function scheduleMatchStatusHeartbeat( + store: DataStore, + requestId: string, + options: { now?: Date; intervalSeconds?: number } = {} +): Promise { + const intervalSeconds = options.intervalSeconds ?? 10; + const runAt = new Date((options.now?.getTime() ?? Date.now()) + intervalSeconds * 1_000); + const bucket = Math.floor(runAt.getTime() / 1_000); + return store.enqueueJob({ + type: "match_status_notify", + payload: { requestId, intervalSeconds }, + idempotencyKey: `match-status:${requestId}:${bucket}`, + partitionKey: `user-match-status:${requestId}`, + runAt: runAt.toISOString() + }); +} + export async function scheduleAdventurexClearingTick( store: DataStore, options: { roundBucketSeconds?: number; now?: Date } = {} @@ -173,6 +193,8 @@ export class JobProcessor { return await this.processMatchRoundGenerate(job); case "match_round_settle": return await this.processMatchRoundSettle(job); + case "match_status_notify": + return await this.processMatchStatusNotify(job); case "room_change_notify": return await this.processRoomChangeNotify(); case "feedback_update": @@ -247,6 +269,44 @@ export class JobProcessor { return { roundId: scheduled.round.roundId, jobId: scheduled.job.id }; } + private async processMatchStatusNotify(job: LlmJob): Promise> { + const requestId = requireString(job.payload, "requestId"); + const request = await this.store.getMatchRequest(requestId); + if ( + !request + || request.status !== "matching" + || !["waiting", "selected", "settling"].includes(request.phase) + ) { + return { + sent: false, + stopped: true, + phase: request?.phase ?? null, + status: request?.status ?? null + }; + } + if (request.intentSnapshot.virtualTestUser === true) { + return { sent: false, stopped: true, virtualTestUser: true }; + } + await this.appendProactiveProductMessage({ + userId: request.userId, + event: { + kind: "match_progress", + facts: { + phase: request.phase, + matchingStartedAt: request.createdAt, + hasActiveRound: Boolean(request.activeRoundId), + userActionRequired: false + } + }, + idempotencyKey: `match-progress:${request.requestId}:${Math.floor(Date.now() / 1_000)}` + }); + const intervalSeconds = typeof job.payload.intervalSeconds === "number" + ? Math.min(Math.max(Math.floor(job.payload.intervalSeconds), 5), 60) + : 10; + const next = await scheduleMatchStatusHeartbeat(this.store, requestId, { intervalSeconds }); + return { sent: true, phase: request.phase, nextJobId: next.id }; + } + private async buildProductContext(userId: string) { const [model, matchRequest, room, memoryProfile, matchOptions, onboardingState, messages, socialHooks] = await Promise.all([ this.store.getUserModel(userId), @@ -609,6 +669,9 @@ export class JobProcessor { idempotencyKey: `match:${matchRequest.requestId}`, partitionKey: `user:${userId}` }).then((queued) => ({ roundId: null, jobId: queued.id })); + if (!this.adventurexMatchingV1 && matchRequest.intentSnapshot.virtualTestUser !== true) { + await scheduleMatchStatusHeartbeat(this.store, matchRequest.requestId); + } if (this.adventurexMatchingV1) { await this.store.updateAdventurexOnboardingState(userId, { stage: "matching" }); } @@ -722,7 +785,19 @@ export class JobProcessor { replyOverride: intro }; } - return { result: { choices }, matchRequest: await this.store.getMatchRequest(currentMatchRequest.requestId), room: currentRoom }; + const notifyJob = choices.some((choice) => choice.draftId) + ? await this.store.enqueueJob({ + type: "room_change_notify", + payload: { roundId: currentMatchRequest.activeRoundId }, + idempotencyKey: `draft-change-notify:${currentMatchRequest.requestId}:${currentMatchRequest.updatedAt}`, + partitionKey: `match-round:${currentMatchRequest.activeRoundId}` + }) + : null; + return { + result: { choices, jobId: notifyJob?.id ?? null }, + matchRequest: await this.store.getMatchRequest(currentMatchRequest.requestId), + room: currentRoom + }; } case "refresh_match_options": { if (!this.adventurexMatchingV1) { @@ -1003,9 +1078,13 @@ export class JobProcessor { ? understanding.summary : ""; if (kind === "image") { + const userMessageId = typeof job.payload.userMessageId === "string" + ? job.payload.userMessageId + : undefined; const turn = await this.runAgentTurn(job, { userId, userContent: describeImageObservation(understanding, storagePaths.length, hint), + userMessageId, replyIdempotencyKey: `multimodal-reply:${job.id}`, memorySource: memoryContent ? { sourceType: "multimodal", sourceId: inputIds[0]!, content: memoryContent } @@ -1532,12 +1611,11 @@ export class JobProcessor { } optionMessages.set(requestId, composed.content); })); - const offerExpiresAt = new Date(Date.now() + (this.options.offerWindowSeconds ?? 90) * 1_000).toISOString(); const offers = await this.store.saveRoundProposals({ roundId, proposal, offers: preparedOffers, - offerExpiresAt + offerExpiresAt: null }); if (offers.length === 0) { const stillActive = activeEntrants.filter((candidate) => !noOfferEntrants.some( @@ -1575,6 +1653,11 @@ export class JobProcessor { const content = optionMessages.get(requestId); if (!content) throw new Error("候选消息没有完成个性化生成"); if (candidate.request.intentSnapshot.virtualTestUser === true) { + await this.store.activateMatchOfferWindow( + requestId, + roundId, + this.options.offerWindowSeconds ?? 90 + ); const requestOffers = offersByRequest.get(requestId) ?? []; const optionNumbers = requestOffers .map((offer) => offer.optionNumber) @@ -1597,20 +1680,13 @@ export class JobProcessor { }); })); await notifyUnavailable(noOfferEntrants, unavailableCause, "partial"); - const settleJob = await this.store.enqueueJob({ - type: "match_round_settle", - payload: { roundId }, - idempotencyKey: `match-round-settle:${roundId}`, - partitionKey: `match-round:${roundId}`, - runAt: offerExpiresAt - }); return { roundId, candidateCount: candidates.length, draftCount: proposal?.drafts.length ?? 0, offerCount: offers.length, unavailableCount: noOfferEntrants.length, - settleJobId: settleJob.id + settleJobId: null }; } @@ -1620,6 +1696,22 @@ export class JobProcessor { this.store.getRoundSettlementState(roundId), this.store.listOfflineGames() ]); + const latestOfferExpiry = state.requests + .filter((request) => request.status === "matching" && request.activeRoundId === roundId) + .map((request) => request.optionsExpiresAt) + .filter((value): value is string => Boolean(value)) + .sort() + .at(-1); + if (latestOfferExpiry && new Date(latestOfferExpiry).getTime() > Date.now()) { + const deferred = await this.store.enqueueJob({ + type: "match_round_settle", + payload: { roundId }, + idempotencyKey: `match-round-settle:${roundId}:${Math.floor(new Date(latestOfferExpiry).getTime() / 1_000)}`, + partitionKey: `match-round:${roundId}`, + runAt: latestOfferExpiry + }); + return { roundId, deferred: true, runAt: latestOfferExpiry, jobId: deferred.id }; + } const candidateByRequest = new Map(); await Promise.all(state.requests.map(async (request) => { const [userModel, socialHooks, profile] = await Promise.all([ diff --git a/scripts/migration-safety.config.json b/scripts/migration-safety.config.json index 9b6f46c..73c504a 100644 --- a/scripts/migration-safety.config.json +++ b/scripts/migration-safety.config.json @@ -40,6 +40,14 @@ "20260725230000_dynamic_room_matchmaking.sql": { "sha256": "8b164fb510321928434c5f5438086e6ec35a9bcdc4e5bdb5733a2d48f6dde937", "reason": "Replace matching status constraints and legacy execution grants while adding transactional invitations, RLS, and synchronized room lifecycle state." + }, + "20260725240000_llm_job_lease_heartbeat.sql": { + "sha256": "397867a188acbc0ee8d76262e5ca6d065bc934de24aae84edda1f438931e15d6", + "reason": "Replace legacy completion and failure RPC signatures with lease-owner-aware overloads, add the heartbeat RPC, and restrict execution to privileged roles." + }, + "20260726010000_end_to_end_delivery_guards.sql": { + "sha256": "ef70495c73d7f64ea8d3ddc054db4ede9afcdb71edc52c516356cf31f1f694a5", + "reason": "Replace the LLM job type constraint, recreate transactional matchmaking RPC bodies, and restrict new delivery/window RPCs to service_role; DELETE only replaces a user's choices atomically." } } } diff --git a/supabase/migrations/20260726010000_end_to_end_delivery_guards.sql b/supabase/migrations/20260726010000_end_to_end_delivery_guards.sql new file mode 100644 index 0000000..35d3164 --- /dev/null +++ b/supabase/migrations/20260726010000_end_to_end_delivery_guards.sql @@ -0,0 +1,403 @@ +alter table public.users + add column if not exists adventurex_welcome_delivered_at timestamptz; + +alter table public.llm_jobs drop constraint if exists llm_jobs_job_type_check; +alter table public.llm_jobs add constraint llm_jobs_job_type_check check (job_type in ( + 'agent_reply','agent_event_reply','multimodal_understanding','matchmaking','match_round_generate', + 'match_round_settle','match_status_notify','room_change_notify','feedback_update', + 'memory_extract','memory_consolidate' +)); + +create or replace view public.adventurex_onboarding_states as +select + id as user_id, + adventurex_stage as stage, + adventurex_image_declined as image_declined, + adventurex_welcome_sent_at as welcome_sent_at, + adventurex_state_created_at as created_at, + adventurex_state_updated_at as updated_at, + adventurex_preferred_language as preferred_language, + adventurex_boundary_prompted_at as boundary_prompted_at, + adventurex_welcome_delivered_at as welcome_delivered_at +from public.users; + +create or replace function public.adventurex_onboarding_state(p_user_id uuid) +returns jsonb +language sql +stable +security definer +set search_path = public +as $$ + select jsonb_build_object( + 'user_id', id, + 'stage', adventurex_stage, + 'image_declined', adventurex_image_declined, + 'preferred_language', adventurex_preferred_language, + 'boundary_prompted_at', adventurex_boundary_prompted_at, + 'welcome_sent_at', adventurex_welcome_sent_at, + 'welcome_delivered_at', adventurex_welcome_delivered_at, + 'created_at', adventurex_state_created_at, + 'updated_at', adventurex_state_updated_at + ) + from users where id = p_user_id +$$; + +create or replace function public.mark_adventurex_welcome_delivered(p_user_id uuid) +returns jsonb +language plpgsql +security definer +set search_path = public +as $$ +begin + perform ensure_tomeet_user(p_user_id, '新朋友'); + update users set + adventurex_welcome_delivered_at = coalesce(adventurex_welcome_delivered_at, now()), + adventurex_welcome_sent_at = coalesce(adventurex_welcome_sent_at, now()), + adventurex_stage = case when adventurex_stage = 'new' then 'awaiting_image_or_text' else adventurex_stage end, + adventurex_state_updated_at = now() + where id = p_user_id; + return adventurex_onboarding_state(p_user_id); +end; +$$; + +create or replace function public.save_match_round_proposals( + p_round_id uuid, + p_proposal jsonb, + p_offers jsonb, + p_offer_expires_at timestamptz +) +returns setof jsonb +language plpgsql +security definer +set search_path = public +as $$ +declare + v_draft jsonb; v_offer jsonb; v_hook jsonb; + v_draft_id uuid; v_offer_id uuid; v_request_id uuid; v_source_type text; + v_ordinal integer; v_draft_expires_at timestamptz; +begin + perform 1 from match_rounds where id = p_round_id and status not in ('completed','expired') for update; + if not found then raise exception '匹配轮次不可保存' using errcode = 'P0001'; end if; + v_draft_expires_at := coalesce(p_offer_expires_at, now() + interval '10 minutes'); + for v_draft in select value from jsonb_array_elements(coalesce(p_proposal->'drafts','[]'::jsonb)) loop + insert into match_drafts ( + round_id, temp_draft_id, offline_game_id, target_players, rationale, expires_at + ) values ( + p_round_id, v_draft->>'tempDraftId', v_draft->>'offlineGameId', + (v_draft->>'targetPlayers')::smallint, v_draft->>'rationale', v_draft_expires_at + ) on conflict (round_id, temp_draft_id) do update set updated_at = now() + returning id into v_draft_id; + insert into match_draft_candidates (draft_id, request_id, user_id) + select v_draft_id, source.value::uuid, mr.user_id + from jsonb_array_elements_text(v_draft->'candidateRequestIds') source + join match_requests mr on mr.id = source.value::uuid and mr.status = 'matching' + on conflict do nothing; + end loop; + for v_offer in select value from jsonb_array_elements(coalesce(p_offers,'[]'::jsonb)) loop + v_request_id := (v_offer->>'requestId')::uuid; + v_source_type := v_offer->>'sourceType'; + v_draft_id := null; + if v_source_type = 'draft' then + select id into v_draft_id from match_drafts + where round_id = p_round_id and temp_draft_id = v_offer->>'tempDraftId'; + if not found then raise exception '候选局映射不存在' using errcode = 'P0001'; end if; + end if; + insert into match_option_offers ( + request_id, round_id, source_type, draft_id, room_id, source_version, + option_number, offline_game_id, preview_text + ) values ( + v_request_id, p_round_id, v_source_type, v_draft_id, + case when v_source_type = 'open_room' then (v_offer->>'roomId')::uuid else null end, + (v_offer->>'sourceVersion')::integer, (v_offer->>'optionNumber')::smallint, + v_offer->>'offlineGameId', v_offer->>'previewText' + ) on conflict (request_id, round_id, option_number) do update set preview_text = excluded.preview_text + returning id into v_offer_id; + v_ordinal := 0; + for v_hook in select value from jsonb_array_elements(coalesce(v_offer->'hooks','[]'::jsonb)) loop + v_ordinal := v_ordinal + 1; + perform 1 from user_social_hooks h + where h.id = (v_hook->>'hookId')::uuid and h.user_id = (v_hook->>'sourceUserId')::uuid and h.status = 'active'; + if not found then raise exception '候选引用了无效社交钩子' using errcode = 'P0001'; end if; + insert into match_option_offer_hooks (offer_id, hook_id, source_user_id, certainty, ordinal) + values (v_offer_id, (v_hook->>'hookId')::uuid, (v_hook->>'sourceUserId')::uuid, v_hook->>'certainty', v_ordinal) + on conflict do nothing; + end loop; + update match_requests set phase = 'offered', options_expires_at = p_offer_expires_at, updated_at = now() + where id = v_request_id and status = 'matching' and active_round_id = p_round_id; + return next offer_with_hooks(v_offer_id); + end loop; + update match_rounds set status = 'collecting', offer_expires_at = p_offer_expires_at, updated_at = now() + where id = p_round_id; +end; +$$; + +create or replace function public.activate_match_offer_window( + p_request_id uuid, + p_round_id uuid, + p_window_seconds integer default 90 +) +returns jsonb +language plpgsql +security definer +set search_path = public +as $$ +declare v_request match_requests%rowtype; v_expires_at timestamptz; +begin + if p_window_seconds not between 10 and 600 then + raise exception '候选选择窗口必须在 10-600 秒之间' using errcode = 'P0001'; + end if; + select * into v_request from match_requests where id=p_request_id for update; + if not found or v_request.status <> 'matching' or v_request.phase <> 'offered' + or v_request.active_round_id is distinct from p_round_id then + raise exception '候选当前不能开始选择计时' using errcode = 'P0001'; + end if; + v_expires_at := now() + make_interval(secs => p_window_seconds); + update match_requests set options_expires_at=v_expires_at,updated_at=now() where id=p_request_id + returning * into v_request; + update match_rounds set + offer_expires_at=greatest(coalesce(offer_expires_at,v_expires_at),v_expires_at),updated_at=now() + where id=p_round_id; + update match_drafts set expires_at=greatest(expires_at,v_expires_at),updated_at=now() + where round_id=p_round_id and status='collecting'; + perform enqueue_llm_job( + 'match_round_settle', + jsonb_build_object('roundId',p_round_id), + 'match-round-settle:' || p_round_id::text || ':' || floor(extract(epoch from v_expires_at))::bigint::text, + 3, + 'match-round:' || p_round_id::text, + v_expires_at + ); + return to_jsonb(v_request); +end; +$$; + +create or replace function public.list_current_match_options(p_user_id uuid) +returns jsonb +language sql +stable +security definer +set search_path = public +as $$ + select jsonb_build_object( + 'request_id', mr.id, + 'round_id', mr.active_round_id, + 'expires_at', mr.options_expires_at, + 'options', coalesce(( + select jsonb_agg(offer_with_hooks(o.id) || jsonb_build_object( + 'activity_name', g.name, + 'activity_description', g.description + ) order by o.option_number) + from match_option_offers o join offline_games g on g.id = o.offline_game_id + where o.request_id = mr.id and o.round_id = mr.active_round_id and o.status <> 'expired' + ), '[]'::jsonb) + ) + from match_requests mr + where mr.user_id = p_user_id and mr.status = 'matching' and mr.phase in ('offered','selected') + and mr.options_expires_at is not null and mr.options_expires_at > now() + order by mr.created_at desc limit 1 +$$; + +create or replace function public.save_match_choices( + p_request_id uuid, + p_preferred_option_number smallint, + p_accepted_option_numbers smallint[], + p_required_hook_ids uuid[] default '{}', + p_raw_user_text text default '结构化选择' +) +returns setof jsonb +language plpgsql +security definer +set search_path = public +as $$ +declare + v_request match_requests%rowtype; v_offer match_option_offers%rowtype; v_choice_id uuid; + v_number smallint; v_index integer := 0; v_rank smallint; v_hook_id uuid; + v_previous_draft_ids uuid[]; v_draft_version integer; v_event_id uuid; +begin + select * into v_request from match_requests where id=p_request_id for update; + if not found or v_request.status<>'matching' or v_request.active_round_id is null then + raise exception '匹配请求当前不能选择' using errcode='P0001'; + end if; + if v_request.options_expires_at is null or v_request.options_expires_at<=now() then + raise exception '候选尚未送达或已过期' using errcode='P0001'; + end if; + if cardinality(coalesce(p_accepted_option_numbers,'{}')) not between 1 and 3 then + raise exception '至少选择一个候选' using errcode='P0001'; + end if; + if exists ( + select 1 from unnest(coalesce(p_required_hook_ids,'{}')) hook_id + where not exists ( + select 1 from match_option_offers o join match_option_offer_hooks oh on oh.offer_id=o.id + where o.request_id=p_request_id and o.round_id=v_request.active_round_id + and o.option_number=any(p_accepted_option_numbers) and oh.hook_id=hook_id + ) + ) then raise exception 'required hook 必须来自已接受候选' using errcode='P0001'; end if; + select coalesce(array_agg(draft_id) filter (where draft_id is not null),'{}'::uuid[]) + into v_previous_draft_ids from match_choices + where request_id=p_request_id and round_id=v_request.active_round_id; + delete from match_choices where request_id=p_request_id and round_id=v_request.active_round_id; + foreach v_number in array p_accepted_option_numbers loop + v_index := v_index+1; + select * into v_offer from match_option_offers + where request_id=p_request_id and round_id=v_request.active_round_id + and option_number=v_number and status<>'expired' for update; + if not found then raise exception '选择包含不存在的候选编号' using errcode='P0001'; end if; + v_rank := case when p_preferred_option_number is null or v_number=p_preferred_option_number then 1 else least(3,v_index+1) end; + insert into match_choices (request_id,round_id,source_type,draft_id,room_id,preference_rank,raw_user_text) + values (p_request_id,v_request.active_round_id,v_offer.source_type,v_offer.draft_id,v_offer.room_id,v_rank,p_raw_user_text) + returning id into v_choice_id; + for v_hook_id in select unnest(coalesce(p_required_hook_ids,'{}')) loop + insert into match_choice_required_hooks (choice_id,hook_id,source_user_id) + select v_choice_id,oh.hook_id,oh.source_user_id from match_option_offer_hooks oh + where oh.offer_id=v_offer.id and oh.hook_id=v_hook_id on conflict do nothing; + end loop; + if v_offer.source_type='draft' and not (v_offer.draft_id=any(v_previous_draft_ids)) then + update match_drafts set version=version+1,updated_at=now() + where id=v_offer.draft_id and status='collecting' returning version into v_draft_version; + if found then + insert into match_draft_change_events (draft_id,draft_version,change_type,payload) + values (v_offer.draft_id,v_draft_version,'confirmed_member_joined', + jsonb_build_object('joinedUserId',v_request.user_id)) returning id into v_event_id; + insert into match_draft_change_notifications (event_id,user_id,idempotency_key) + select distinct v_event_id,mr.user_id, + 'draft-change:'||v_event_id::text||':'||mr.user_id::text + from match_choices c join match_requests mr on mr.id=c.request_id + where c.draft_id=v_offer.draft_id and mr.user_id<>v_request.user_id + on conflict do nothing; + end if; + end if; + return next to_jsonb(v_offer)||jsonb_build_object( + 'id',v_choice_id,'preference_rank',v_rank,'raw_user_text',p_raw_user_text, + 'created_at',now(),'required_hook_ids',coalesce(( + select jsonb_agg(hook_id) from match_choice_required_hooks where choice_id=v_choice_id + ),'[]'::jsonb) + ); + end loop; + update match_option_offers set + status=case when option_number=any(p_accepted_option_numbers) then 'accepted' else 'rejected' end, + responded_at=now() + where request_id=p_request_id and round_id=v_request.active_round_id and status<>'expired'; + update match_requests set phase='selected',updated_at=now() where id=p_request_id; +end; +$$; + +create or replace function public.enqueue_wechat_outbound_message( + p_user_id uuid, + p_message_id uuid, + p_content text +) +returns void +language plpgsql +security definer +set search_path = public +as $$ +declare v_message messages%rowtype; +begin + select * into v_message from messages where id=p_message_id and user_id=p_user_id; + if not found or v_message.role <> 'assistant' then + raise exception '微信投递消息不存在或不是 Agent 回复' using errcode='P0001'; + end if; + if v_message.source_channel = 'web' then + raise exception 'Web 对话消息不能投递到微信' using errcode='P0001'; + end if; + if exists (select 1 from wechat_ilink_connections where user_id=p_user_id and status='active') then + insert into channel_message_deliveries ( + provider,direction,message_id,payload_override,status + ) values ( + 'wechat','outbound',p_message_id, + case when p_content is distinct from v_message.content then p_content else null end, + 'pending' + ) + on conflict (provider,message_id) where direction='outbound' do nothing; + end if; +end; +$$; + +create or replace function public.complete_wechat_outbound_message( + p_outbound_id uuid, + p_worker_id text, + p_error text default null +) +returns void +language plpgsql +security definer +set search_path = public +as $$ +declare v_delivery channel_message_deliveries%rowtype; v_idempotency_key text; +begin + select * into v_delivery from channel_message_deliveries d + where d.id=p_outbound_id and d.provider='wechat' and d.direction='outbound' and d.locked_by=p_worker_id + for update; + if not found then return; end if; + select idempotency_key into v_idempotency_key from messages where id=v_delivery.message_id; + if p_error is null then + update channel_message_deliveries set + status='sent',completed_at=now(),locked_by=null,locked_at=null,last_error=null,updated_at=now() + where id=p_outbound_id; + if v_idempotency_key like 'match-options:%' then + perform activate_match_offer_window( + split_part(v_idempotency_key,':',3)::uuid, + split_part(v_idempotency_key,':',2)::uuid, + 90 + ); + end if; + else + update channel_message_deliveries set + status=case when attempts>=5 then 'failed' else 'retry' end, + run_at=now()+make_interval(secs=>least(300,5*power(2,greatest(attempts-1,0))::integer)), + locked_by=null,locked_at=null,last_error=left(p_error,1000),updated_at=now() + where id=p_outbound_id; + end if; +end; +$$; + +create or replace function public.withdraw_room_member_with_reason( + p_room_id uuid, + p_user_id uuid, + p_reason text default null +) +returns jsonb +language plpgsql +security definer +set search_path = public +as $$ +declare + v_room match_rooms%rowtype; v_game offline_games%rowtype; v_member room_members%rowtype; + v_reason text; v_remaining integer; +begin + select * into v_room from match_rooms where id=p_room_id for update; + if not found then raise exception '房间不存在' using errcode='P0002'; end if; + if v_room.status='completed' then raise exception '活动已完成' using errcode='P0001'; end if; + select * into v_member from room_members + where room_id=p_room_id and user_id=p_user_id and participation_status<>'withdrawn' for update; + if not found then raise exception '用户不在当前房间中' using errcode='P0001'; end if; + v_reason := nullif(btrim(coalesce(p_reason,'')),''); + if (v_room.status='confirmed' or v_member.confirmed) and v_reason is null then + raise exception '正式成局后退出需要说明一个理由' using errcode='P0001'; + end if; + if v_reason is not null and char_length(v_reason)>500 then + raise exception '退出理由不能超过 500 字' using errcode='P0001'; + end if; + select * into v_game from offline_games where id=v_room.offline_game_id; + update room_members set confirmed=false,participation_status='withdrawn',withdrawn_at=now(),withdrawal_reason=v_reason + where room_id=p_room_id and user_id=p_user_id; + select count(*) into v_remaining from room_members + where room_id=p_room_id and participation_status='confirmed'; + update match_rooms set version=version+1, + status=case when v_remaining>=v_game.min_players then 'confirmed' else 'confirming' end, + recruitment_status='open' + where id=p_room_id returning * into v_room; + update match_requests set status='cancelled',phase='waiting',proactive_push_enabled=false,room_id=null, + active_round_id=null,options_expires_at=null,updated_at=now() + where room_id=p_room_id and user_id=p_user_id; + perform record_room_change_event( + p_room_id,'member_withdrawn',jsonb_build_object('withdrawnUserId',p_user_id,'memberCount',v_remaining) + ); + return get_match_room(p_room_id); +end; +$$; + +revoke all on function public.mark_adventurex_welcome_delivered(uuid) from public,anon,authenticated; +revoke all on function public.activate_match_offer_window(uuid,uuid,integer) from public,anon,authenticated; +grant execute on function public.mark_adventurex_welcome_delivered(uuid) to service_role; +grant execute on function public.activate_match_offer_window(uuid,uuid,integer) to service_role;