Skip to content

fix(agent): harden structured reply delivery #36

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zs-andy merged 1 commit into from
Jul 25, 2026
Merged

fix(agent): harden structured reply delivery #36

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 51 additions & 0 deletions apps/api/src/app.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down
122 changes: 99 additions & 23 deletions apps/api/src/app.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 };
}
Expand All @@ -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<string | null> {
if (!options.adventurexTestPoolAccessTokenMatches) {
reply.code(503).send({
Expand Down Expand Up @@ -396,31 +416,51 @@ 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 必须同时提供"
).parse(request.body)
: {};
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: {
userId: input.userId,
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);
Expand Down Expand Up @@ -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(
Expand All @@ -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}`;
Expand All @@ -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: {
Expand All @@ -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
Expand Down Expand Up @@ -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("开放局候选已经变化");
Expand Down
22 changes: 15 additions & 7 deletions apps/api/src/wechat-routes.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -79,6 +79,13 @@ async function setup(
return { app, store, wechatStore, fetchMock, verifyAccessToken, sentMessages };
}

function sentMessageTexts(messages: Array<Record<string, unknown>>): 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([{
Expand Down Expand Up @@ -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",
Expand All @@ -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({
Expand All @@ -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();
});
Expand Down Expand Up @@ -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([
Expand Down Expand Up @@ -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();
});
Expand Down Expand Up @@ -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",
Expand Down
19 changes: 17 additions & 2 deletions apps/intelligence-worker/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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";

Expand Down Expand Up @@ -124,7 +129,17 @@ async function runSlot(slot: number): Promise<void> {
} 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) }));
Expand Down
Loading
Loading