diff --git a/hub/src/agent/runner.ts b/hub/src/agent/runner.ts index 81a5c4d..b2c33d8 100644 --- a/hub/src/agent/runner.ts +++ b/hub/src/agent/runner.ts @@ -119,6 +119,8 @@ export async function runAgent(req: RunRequest): Promise { let sdkSessionId: string | undefined; let error: string | undefined; try { + await persistAgentMessage(req, "user", req.prompt); + const options: Parameters[0]["options"] = { cwd: req.project.workspaceDir, allowedTools: ["Read", "Write", "Bash", "Glob", "Grep"], @@ -174,14 +176,17 @@ export async function runAgent(req: RunRequest): Promise { const msg = (message as SDKAssistantMessage).message; await heartbeat(); numTurns++; + let assistantText = ""; for (const block of msg.content) { if (block.type === "text" && typeof block.text === "string") { fullText += block.text; + assistantText += block.text; } if (block.type === "tool_use") { onStream?.({ type: "tool-end", toolName: block.name }); } } + await persistAgentMessage(req, "assistant", assistantText); if (msg.usage !== undefined) { usage.inputTokens += msg.usage.input_tokens ?? 0; usage.outputTokens += msg.usage.output_tokens ?? 0; @@ -219,3 +224,20 @@ export async function runAgent(req: RunRequest): Promise { }; } } + +async function persistAgentMessage(req: RunRequest, role: string, content: string): Promise { + if (content === "") return; + try { + await req.prisma.agentMessage.create({ + data: { + sessionId: req.sessionId, + runId: req.runId, + role, + content, + attachments: [], + }, + }); + } catch { + // Best-effort projection: a history write failure must not break the run. + } +} diff --git a/hub/src/feishu/trigger.ts b/hub/src/feishu/trigger.ts index 270b864..4e6dd5c 100644 --- a/hub/src/feishu/trigger.ts +++ b/hub/src/feishu/trigger.ts @@ -13,7 +13,7 @@ import { z } from "zod"; import type { FastifyBaseLogger } from "fastify"; import { sendText, sendTextMessage, patchTextMessage, reactToMessage, downloadMessageFile, type FeishuRuntime, type MessageReceiveEvent } from "./client.js"; import type { ModelRegistry } from "../agent/models.js"; -import { runAgent, type ProjectContext } from "../agent/runner.js"; +import { runAgent as defaultRunAgent, type ProjectContext, type RunRequest, type RunResult } from "../agent/runner.js"; import { acquireLock, currentLockRunId, releaseLock } from "../lock.js"; import { canTriggerAgent, canTriggerRole } from "../permission.js"; import { writeAudit } from "../audit.js"; @@ -24,6 +24,7 @@ interface TriggerDeps { readonly prisma: PrismaClient; readonly models: ModelRegistry; readonly logger: FastifyBaseLogger; + readonly runAgent?: (req: RunRequest) => Promise; } /** @@ -33,6 +34,7 @@ interface TriggerDeps { */ export function makeTriggerHandler(deps: TriggerDeps) { return async (event: MessageReceiveEvent, rt: FeishuRuntime): Promise => { + const runAgent = deps.runAgent ?? defaultRunAgent; const msg = event.message; const sender = event.sender; void sender; // sender→user mapping is OPEN (principal sub-typology) @@ -165,6 +167,7 @@ export function makeTriggerHandler(deps: TriggerDeps) { const existing = await currentLockRunId(deps.prisma, projectId); if (existing !== null) { await reactToMessage(rt, msg.message_id, "OnIt"); + await sendText(rt, chatId, "项目正在处理中,请稍候。"); return; } @@ -251,6 +254,7 @@ export function makeTriggerHandler(deps: TriggerDeps) { }); await writeAudit(deps.prisma, { runId: run.id, projectId, action: "run.lock_race", metadata: {} }); await reactToMessage(rt, msg.message_id, "OnIt"); + await sendText(rt, chatId, "项目正在处理中,请稍候。"); return; } @@ -342,7 +346,11 @@ export function makeTriggerHandler(deps: TriggerDeps) { await writeAudit(deps.prisma, { runId: run.id, projectId, action: "run.failed", metadata: { error: String(e) } }); }) .finally(async () => { - await releaseLock(deps.prisma, run.id); + try { + await releaseLock(deps.prisma, run.id); + } catch (e) { + deps.logger.warn({ runId: run.id, err: e instanceof Error ? e.message : String(e) }, "trigger: could not release lock"); + } }); }; } diff --git a/hub/test/integration/helpers.ts b/hub/test/integration/helpers.ts index 05e53e5..4088e20 100644 --- a/hub/test/integration/helpers.ts +++ b/hub/test/integration/helpers.ts @@ -57,23 +57,41 @@ export const silentLogger: FastifyBaseLogger = { export interface MockFeishuRuntime extends FeishuRuntime { readonly sentTexts: string[]; readonly sentCards: unknown[]; + readonly sentPatches: unknown[]; + readonly reactions: Array<{ readonly messageId: string; readonly emoji: string }>; } export function mockFeishuRuntime(): MockFeishuRuntime { const sentTexts: string[] = []; const sentCards: unknown[] = []; + const sentPatches: unknown[] = []; + const reactions: Array<{ messageId: string; emoji: string }> = []; // Mock the client — sendText/sendCard are in client.ts, not on the runtime // object directly. We patch by providing a runtime whose client is a stub; // the actual send functions cast through shape, so a minimal stub works. const rt: MockFeishuRuntime = { client: { + request: async (p: unknown) => { + const payload = p as { url?: string; data?: { reaction_type?: { emoji_type?: string } } }; + const match = payload.url?.match(/\/messages\/([^/]+)\/reactions$/); + if (match?.[1] !== undefined) { + reactions.push({ + messageId: match[1], + emoji: payload.data?.reaction_type?.emoji_type ?? "", + }); + } + return {}; + }, im: { v1: { message: { create: async (p: unknown) => { const payload = p as { data?: { msg_type?: string; content?: string } }; if (payload.data?.msg_type === "interactive") { - sentCards.push(payload.data.content ? JSON.parse(payload.data.content) : null); + const card = payload.data.content ? JSON.parse(payload.data.content) : null; + sentCards.push(card); + const text = textFromCard(card); + if (text !== null) sentTexts.push(text); } else { try { const c = JSON.parse(payload.data?.content ?? "{}") as { text?: string }; @@ -84,6 +102,14 @@ export function mockFeishuRuntime(): MockFeishuRuntime { } return { data: { message_id: "mock-msg-id" } }; }, + patch: async (p: unknown) => { + const payload = p as { data?: { content?: string } }; + const card = payload.data?.content ? JSON.parse(payload.data.content) : null; + sentPatches.push(card); + const text = textFromCard(card); + if (text !== null) sentTexts.push(text); + return {}; + }, }, }, }, @@ -91,10 +117,27 @@ export function mockFeishuRuntime(): MockFeishuRuntime { logger: silentLogger, sentTexts, sentCards, + sentPatches, + reactions, }; return rt; } +function textFromCard(card: unknown): string | null { + if (typeof card !== "object" || card === null) return null; + const elements = (card as { elements?: unknown }).elements; + if (!Array.isArray(elements)) return null; + const parts: string[] = []; + for (const element of elements) { + if (typeof element !== "object" || element === null) continue; + const text = (element as { text?: unknown }).text; + if (typeof text !== "object" || text === null) continue; + const content = (text as { content?: unknown }).content; + if (typeof content === "string") parts.push(content); + } + return parts.length === 0 ? null : parts.join("\n"); +} + export interface MockToolCall { readonly toolCallId: string; readonly toolName: string; diff --git a/hub/test/integration/trigger.test.ts b/hub/test/integration/trigger.test.ts index f0a1f0c..3796435 100644 --- a/hub/test/integration/trigger.test.ts +++ b/hub/test/integration/trigger.test.ts @@ -1,12 +1,12 @@ import { describe, it, expect, beforeEach, afterAll, vi } from "vitest"; -import { prisma, resetDb, mockFeishuRuntime, createMockModelFactory, seedProject, silentLogger } from "./helpers.js"; +import { prisma, resetDb, mockFeishuRuntime, seedProject, silentLogger } from "./helpers.js"; import { InMemoryModelRegistry } from "../../src/agent/models.js"; -import { ToolRegistry } from "../../src/agent/tools.js"; import { makeTriggerHandler, extractPrompt } from "../../src/feishu/trigger.js"; import type { MessageReceiveEvent } from "../../src/feishu/client.js"; -import type { ModelFactory } from "../../src/agent/runner.js"; +import type { RunRequest, RunResult } from "../../src/agent/runner.js"; const bot = { key: "@_user_1", id: { open_id: "ou_bot" }, name: "Bot" }; +type TestRunner = (req: RunRequest) => Promise; function makeEvent(chatId: string, text: string, senderOpenId = "ou_test_user", eventId?: string): MessageReceiveEvent { const header = eventId !== undefined ? { event_id: eventId, event_type: "im.message.receive_v1" } : undefined; @@ -24,16 +24,39 @@ function makeEvent(chatId: string, text: string, senderOpenId = "ou_test_user", }; } +function createMockRunAgent(calls: RunRequest[] = []): TestRunner { + return async (req) => { + calls.push(req); + req.onStream?.({ type: "text-delta", text: "mock response" }); + await req.prisma.agentMessage.create({ + data: { + sessionId: req.sessionId, + runId: req.runId, + role: "assistant", + content: "mock response", + attachments: [], + }, + }); + return { + status: "completed", + text: "mock response", + usage: { inputTokens: 10, outputTokens: 5 }, + numTurns: 1, + sdkSessionId: "sdk-session-1", + }; + }; +} + describe("trigger full lifecycle (integration)", () => { - let modelFactory: ModelFactory; - let tools: ToolRegistry; let models: InMemoryModelRegistry; let rt: ReturnType; + let runAgentCalls: RunRequest[]; + let runAgent: TestRunner; beforeEach(async () => { await resetDb(); - modelFactory = createMockModelFactory().modelFactory; - tools = new ToolRegistry(); + runAgentCalls = []; + runAgent = createMockRunAgent(runAgentCalls); models = new InMemoryModelRegistry( [{ id: "mock-model", label: "Mock", toolCapable: true }], [ @@ -46,7 +69,7 @@ describe("trigger full lifecycle (integration)", () => { it("creates a run, acquires + releases the lock, sends status card", async () => { await seedProject("proj-1", "chat-1"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-1", "@_user_1 写教案"), rt); @@ -58,21 +81,25 @@ describe("trigger full lifecycle (integration)", () => { }); // Lock released (no lock row remains). - const locks = await prisma.projectAgentLock.findMany(); - expect(locks).toHaveLength(0); + await vi.waitFor(async () => { + const locks = await prisma.projectAgentLock.findMany(); + expect(locks).toHaveLength(0); + }); // A status card was sent. expect(rt.sentCards.length).toBeGreaterThanOrEqual(1); - expect(rt.sentTexts).toContain("已开始处理(role: draft, model: mock-model)。"); + expect(rt.sentTexts).toContain("mock response"); + expect(runAgentCalls).toHaveLength(1); }); it("rejects a sender without edit grant (ADR-0004)", async () => { await seedProject("proj-2", "chat-2", { role: "READ" }); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-2", "@_user_1 写教案"), rt); expect(rt.sentTexts).toContain("无权限触发。"); + expect(runAgentCalls).toHaveLength(0); const runs = await prisma.agentRun.findMany(); expect(runs).toHaveLength(0); }); @@ -87,10 +114,11 @@ describe("trigger full lifecycle (integration)", () => { data: { projectId: "proj-3", runId: existingRun.id }, }); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-3", "@_user_1 写教案"), rt); expect(rt.sentTexts).toContain("项目正在处理中,请稍候。"); + expect(runAgentCalls).toHaveLength(0); // No new run created. const runs = await prisma.agentRun.findMany(); expect(runs).toHaveLength(1); @@ -98,7 +126,7 @@ describe("trigger full lifecycle (integration)", () => { it("ignores messages from unbound chats (ADR-0001)", async () => { await seedProject("proj-4", "chat-4"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-UNKNOWN", "@_user_1 写教案"), rt); @@ -110,7 +138,7 @@ describe("trigger full lifecycle (integration)", () => { it("ignores messages without @bot mention", async () => { await seedProject("proj-5", "chat-5"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); const event: MessageReceiveEvent = { message: { @@ -132,7 +160,7 @@ describe("trigger full lifecycle (integration)", () => { it("/new archives current session (no run created)", async () => { await seedProject("proj-6", "chat-6"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); // First @bot creates a session + run. await trigger(makeEvent("chat-6", "@_user_1 写教案"), rt); @@ -155,7 +183,7 @@ describe("trigger full lifecycle (integration)", () => { it("/resume un-archives the most recent session", async () => { await seedProject("proj-7", "chat-7"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); // Create + archive a session via /new. await trigger(makeEvent("chat-7", "@_user_1 写教案"), rt); @@ -175,7 +203,7 @@ describe("trigger full lifecycle (integration)", () => { it("/reset archives current session", async () => { await seedProject("proj-8", "chat-8"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-8", "@_user_1 写教案"), rt); await vi.waitFor(async () => { @@ -194,7 +222,7 @@ describe("trigger full lifecycle (integration)", () => { it("unknown slash command falls through to agent", async () => { await seedProject("proj-9", "chat-9"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-9", "@_user_1 /unknown"), rt); // Should create a run (falls through as a normal prompt). @@ -211,11 +239,12 @@ describe("trigger full lifecycle (integration)", () => { await prisma.roleTriggerGrant.create({ data: { projectId: "proj-10", roleId: "review", principal: "ou_other" }, }); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-10", "@_user_1 /review 看看这节"), rt); expect(rt.sentTexts).toContain("无权限使用角色 review。"); + expect(runAgentCalls).toHaveLength(0); const runs = await prisma.agentRun.findMany(); expect(runs).toHaveLength(0); }); @@ -225,7 +254,7 @@ describe("trigger full lifecycle (integration)", () => { await prisma.roleTriggerGrant.create({ data: { projectId: "proj-11", roleId: "review", principal: "ou_test_user" }, }); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-11", "@_user_1 /review 看看这节"), rt); @@ -239,7 +268,7 @@ describe("trigger full lifecycle (integration)", () => { it("extractRole: /draft sets roleId=draft, strips command from prompt", async () => { await seedProject("proj-12", "chat-12"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-12", "@_user_1 /draft 写第三单元"), rt); @@ -253,7 +282,7 @@ describe("trigger full lifecycle (integration)", () => { it("dedups a redelivered event by event_id (no second run)", async () => { await seedProject("proj-13", "chat-13"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); // First delivery: processes normally. await trigger(makeEvent("chat-13", "@_user_1 写教案", "ou_test_user", "evt-dedup-1"), rt); @@ -276,7 +305,7 @@ describe("trigger full lifecycle (integration)", () => { it("writes audit entries across the run lifecycle", async () => { await seedProject("proj-14", "chat-14"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-14", "@_user_1 写教案"), rt); await vi.waitFor(async () => { @@ -291,7 +320,7 @@ describe("trigger full lifecycle (integration)", () => { it("persists AgentMessage rows for the run (A-3 structured history)", async () => { await seedProject("proj-15", "chat-15"); - const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + const trigger = makeTriggerHandler({ prisma, models, logger: silentLogger, runAgent }); await trigger(makeEvent("chat-15", "@_user_1 写教案"), rt); await vi.waitFor(async () => { diff --git a/hub/test/unit/runner.test.ts b/hub/test/unit/runner.test.ts index 2bd4528..f738524 100644 --- a/hub/test/unit/runner.test.ts +++ b/hub/test/unit/runner.test.ts @@ -90,4 +90,41 @@ describe("runAgent", () => { const call = queryMock.mock.calls[0]?.[0] as { options?: Record } | undefined; expect(call?.options).not.toHaveProperty("resume"); }); + + it("persists structured user and assistant messages best-effort", async () => { + queryMock.mockReturnValue(messages(assistantMessage("ok"), resultMessage("sdk-session-1"))); + const createMessage = vi.fn().mockResolvedValue({}); + const prisma = { + projectAgentLock: { update: async () => ({}) }, + agentMessage: { create: createMessage }, + } as unknown as import("@prisma/client").PrismaClient; + + await runAgent({ + prompt: "你好", + model: undefined, + project: { projectId: "p", boundChatId: "c", workspaceDir: "/tmp/ws" }, + systemPrompt: undefined, + runId: "run-1", + sessionId: "hub-session-1", + prisma, + }); + + expect(createMessage).toHaveBeenCalledTimes(2); + expect(createMessage).toHaveBeenNthCalledWith(1, { + data: expect.objectContaining({ + sessionId: "hub-session-1", + runId: "run-1", + role: "user", + content: "你好", + }), + }); + expect(createMessage).toHaveBeenNthCalledWith(2, { + data: expect.objectContaining({ + sessionId: "hub-session-1", + runId: "run-1", + role: "assistant", + content: "ok", + }), + }); + }); });