/** * Feishu @bot trigger → AgentRun lifecycle (streaming). * * Sends a "processing" card immediately, patches it with text deltas + tool * status as the agent streams, then replaces it with a final status card on * completion. Throttled to ~1 patch/sec to avoid spamming the Feishu API. * * ADR-0001..0003, 0017: chat→project binding, permission gate, lock, * provider-bound session, transcript continuity, ADR-0003-authorized reads. */ import type { PrismaClient } from "@prisma/client"; import type { FastifyBaseLogger } from "fastify"; import { sendText, sendTextMessage, patchTextMessage, reactToMessage, downloadMessageFile, sendFile, type FeishuRuntime, type MessageReceiveEvent } from "./client.js"; import type { ModelRegistry } from "../agent/models.js"; import { runAgent, type ProjectContext } from "../agent/runner.js"; import { acquireLock, currentLockRunId, releaseLock } from "../lock.js"; import { canTriggerAgent, canTriggerRole } from "../permission.js"; import { transcriptPath } from "../agent/transcript.js"; import { writeAudit } from "../audit.js"; interface TriggerDeps { readonly prisma: PrismaClient; readonly models: ModelRegistry; readonly logger: FastifyBaseLogger; } /** * Build the message handler. Returns a function suitable for the lark * `im.message.receive_v1` event. The handler is async and long-running (the * agent run is the long leg); the lock guarantees one run per project at a time. */ export function makeTriggerHandler(deps: TriggerDeps) { return async (event: MessageReceiveEvent, rt: FeishuRuntime): Promise => { const msg = event.message; const sender = event.sender; void sender; // sender→user mapping is OPEN (principal sub-typology) // Dedup: the lark ws client redelivers at-least-once. A duplicate event // would spawn a second AgentRun — the lock would refuse it, but a FAILED // run row + a busy reply would leak. Record the event_id up front; if it // already exists, this is a redelivery → stop. const eventId = event.header?.event_id; if (eventId !== undefined && eventId !== "") { const existing = await deps.prisma.feishuEventReceipt.findUnique({ where: { eventId }, select: { id: true }, }); if (existing !== null) { deps.logger.debug({ eventId }, "feishu trigger: duplicate event, skipping"); return; } await deps.prisma.feishuEventReceipt.create({ data: { eventId, eventType: event.header?.event_type ?? "im.message.receive_v1", messageId: msg.message_id, }, }); } // Only react to text, file, or image messages in group chats. if (msg.message_type !== "text" && msg.message_type !== "file" && msg.message_type !== "image") return; const chatId = msg.chat_id; if (chatId === "") return; // ADR-0001: resolve chat → project. Unknown chat ⇒ not a project group. const binding = await deps.prisma.projectGroupBinding.findUnique({ where: { chatId }, select: { projectId: true }, }); if (binding === null) { deps.logger.debug({ chatId }, "feishu trigger: chat not bound to any project"); return; } const projectId = binding.projectId; // ADR-0004: triggerAgent requires edit on the project. principal=sender // open_id (sub-typology OPEN — pragmatic choice, see permission.ts). const senderOpenId = event.sender.sender_id.open_id ?? ""; if (senderOpenId !== "") { const perm = await canTriggerAgent(deps.prisma, projectId, senderOpenId); if (!perm.allowed) { deps.logger.info({ projectId, senderOpenId, reason: perm.reason }, "feishu trigger: permission denied"); await writeAudit(deps.prisma, { projectId, actorUserId: senderOpenId || undefined, action: "trigger.denied", metadata: { reason: perm.reason } }); await sendText(rt, chatId, "无权限触发。"); return; } } let cleanPrompt: string | null = null; let downloadedFiles: string[] = []; if (msg.message_type === "text") { cleanPrompt = extractPrompt(msg); } else if (msg.message_type === "file" || msg.message_type === "image") { // Download the file/image to the workspace, build a prompt around it. const project = await deps.prisma.project.findUnique({ where: { id: projectId }, select: { workspaceDir: true }, }); if (project === null) return; try { const content = JSON.parse(msg.content) as { file_key?: string; image_key?: string }; const key = content.file_key ?? content.image_key ?? ""; if (key !== "") { const fileName = `${msg.message_type}-${Date.now()}.${msg.message_type === "image" ? "png" : "bin"}`; const savePath = `${project.workspaceDir}/.cph/inbox/${fileName}`; await downloadMessageFile(rt, msg.message_id, key, savePath); downloadedFiles.push(savePath); cleanPrompt = `用户发来一个文件: ${savePath}`; } } catch { return; } } if (cleanPrompt === null) return; // Slash commands: session management, not agent runs. These bypass the // lock — they don't create a run, so they can't conflict with one. if (cleanPrompt.startsWith("/")) { const cmd = cleanPrompt.split(/\s+/)[0]; switch (cmd) { case "/new": { await deps.prisma.agentSession.updateMany({ where: { projectId, archivedAt: null }, data: { archivedAt: new Date() }, }); await sendText(rt, chatId, "已开新会话,下次 @bot 将从头开始。"); return; } case "/resume": { const latest = await deps.prisma.agentSession.findFirst({ where: { projectId, archivedAt: { not: null } }, orderBy: { archivedAt: "desc" }, select: { id: true }, }); if (latest === null) { await sendText(rt, chatId, "没有可恢复的会话。"); return; } await deps.prisma.agentSession.update({ where: { id: latest.id }, data: { archivedAt: null }, }); await sendText(rt, chatId, "已恢复上一个会话。"); return; } case "/reset": { await deps.prisma.agentSession.updateMany({ where: { projectId, archivedAt: null }, data: { archivedAt: new Date() }, }); await sendText(rt, chatId, "已重置,下次 @bot 将从头开始。"); return; } default: break; } } // ADR-0002: if locked by another run, reply busy. const existing = await currentLockRunId(deps.prisma, projectId); if (existing !== null) { await sendText(rt, chatId, "项目正在处理中,请稍候。"); return; } const project = await deps.prisma.project.findUnique({ where: { id: projectId }, select: { workspaceDir: true }, }); if (project === null) { deps.logger.error({ projectId }, "feishu trigger: project missing"); return; } // ADR-0017: role-as-data. Parse a leading `/` command; unknown // slashes are left as literal text (extractRole returns null). Falls back // to "draft" when no role is named — the registry degrades gracefully. const { roleId: parsedRole } = extractRole(cleanPrompt, deps.models); const roleId = parsedRole ?? "draft"; // Per-role trigger gate (ADR-0017). Orthogonal to ADR-0004's // canTriggerAgent above: that checked "can trigger an agent at all"; // this checks "can trigger *this* role". Unconfigured role ⇒ open. if (senderOpenId !== "") { const rolePerm = await canTriggerRole(deps.prisma, projectId, roleId, senderOpenId); if (!rolePerm.allowed) { deps.logger.info({ projectId, roleId, senderOpenId, reason: rolePerm.reason }, "feishu trigger: role permission denied"); await writeAudit(deps.prisma, { projectId, actorUserId: senderOpenId || undefined, action: "trigger.role_denied", metadata: { roleId, reason: rolePerm.reason } }); await sendText(rt, chatId, `无权限使用角色 ${roleId}。`); return; } } const role = deps.models.role(roleId); const model = deps.models.resolve(undefined, roleId); const providerId = "openrouter"; // Per-role bundle (ADR-0017): systemPrompt seeds persona; tools whitelist // is enforced inside runAgent when it builds the SDK tools record. `tools: // undefined` means the full registered set (unrestricted role). const systemPrompt = role?.systemPrompt; // ADR-0017: provider+model-bound session. Reuse if one exists for this // (project, provider, model) triple; otherwise create. const existingSession = await deps.prisma.agentSession.findFirst({ where: { projectId, provider: providerId, model, archivedAt: null }, select: { id: true }, }); const session = existingSession !== null ? await deps.prisma.agentSession.update({ where: { id: existingSession.id }, data: { provider: providerId, model, updatedAt: new Date() }, select: { id: true }, }) : await deps.prisma.agentSession.create({ data: { projectId, provider: providerId, model, title: cleanPrompt.slice(0, 40) || null, metadata: {}, }, select: { id: true }, }); const run = await deps.prisma.agentRun.create({ data: { projectId, sessionId: session.id, requestedByUserId: null, // sender→user mapping is OPEN (principal sub-typology) entrypoint: "FEISHU", status: "ACTIVE", prompt: cleanPrompt, model, provider: providerId, metadata: { roleId }, }, select: { id: true }, }); await writeAudit(deps.prisma, { runId: run.id, projectId, action: "run.created", metadata: { roleId, model, prompt: cleanPrompt.slice(0, 200) } }); // ADR-0002: acquire the project lock for this run. try { await acquireLock(deps.prisma, projectId, run.id, null); } catch { await deps.prisma.agentRun.update({ where: { id: run.id }, data: { status: "FAILED", error: "lock race", finishedAt: new Date() }, }); await writeAudit(deps.prisma, { runId: run.id, projectId, action: "run.lock_race", metadata: {} }); await sendText(rt, chatId, "项目正在处理中,请稍候。"); return; } const projectCtx: ProjectContext = { projectId, boundChatId: chatId, // ADR-0003: Feishu reads stay in this chat workspaceDir: project.workspaceDir, }; // Emoji reaction on the user's message (acknowledge receipt). await reactToMessage(rt, msg.message_id, "THUMBSUP"); // Stream agent response as text-as-card messages — patchable, markdown, // no truncation. Lower throttle (400ms) for smoother streaming. let currentMsgId = await sendTextMessage(rt, chatId, ""); let accText = ""; let lastPatch = 0; const PATCH_INTERVAL = 400; const MAX_MSG_LEN = 4000; const flushText = async () => { const now = Date.now(); if (now - lastPatch < PATCH_INTERVAL || currentMsgId === null) return; lastPatch = now; if (accText.length > MAX_MSG_LEN) { await patchTextMessage(rt, currentMsgId, accText.slice(0, MAX_MSG_LEN)); accText = accText.slice(MAX_MSG_LEN); currentMsgId = await sendTextMessage(rt, chatId, accText); } else { await patchTextMessage(rt, currentMsgId, accText); } }; runAgent({ prompt: cleanPrompt, model, project: projectCtx, systemPrompt, runId: run.id, sessionId: session.id, prisma: deps.prisma, onStream: (event) => { if (event.type === "text-delta") { accText += event.text; void flushText(); } }, }) .then(async (result) => { // Final flush: make sure the last message has complete text. if (currentMsgId !== null && accText.length > 0) { await patchTextMessage(rt, currentMsgId, accText); } await deps.prisma.agentRun.update({ where: { id: run.id }, data: { status: result.status === "completed" ? "COMPLETED" : result.status === "length" ? "TIMED_OUT" : "FAILED", inputTokens: result.usage.inputTokens, outputTokens: result.usage.outputTokens, error: result.error ?? null, finishedAt: new Date(), }, }); await writeAudit(deps.prisma, { runId: run.id, projectId, action: "run.finished", metadata: { status: result.status } }); // Send any new files in build/ as file messages. const { readdir, stat } = await import("node:fs/promises"); const { join } = await import("node:path"); const buildDir = join(project.workspaceDir, "build"); try { const entries = await readdir(buildDir); for (const entry of entries) { const fullPath = join(buildDir, entry); const s = await stat(fullPath); if (s.isFile() && s.mtimeMs > Date.now() - 5 * 60 * 1000) { // File created in the last 5 minutes — likely from this run. await sendFile(rt, chatId, fullPath, entry); } } } catch { /* no build/ dir — fine */ } }) .catch(async (e) => { try { await deps.prisma.agentRun.update({ where: { id: run.id }, data: { status: "FAILED", error: String(e), finishedAt: new Date() }, }); } catch (updateErr) { deps.logger.warn({ runId: run.id, err: String(updateErr) }, "trigger: could not mark run FAILED"); } await writeAudit(deps.prisma, { runId: run.id, projectId, action: "run.failed", metadata: { error: String(e) } }); if (currentMsgId !== null && accText.length === 0) { await patchTextMessage(rt, currentMsgId, `⚠️ 处理出错: ${String(e)}`); } }) .finally(async () => { await releaseLock(deps.prisma, run.id); }); }; } /** * Extract the prompt text from a text message, stripping @mentions. * Returns null if the message is empty after stripping or doesn't mention the bot. * Lark text content is JSON: `{"text":"@_user_1 do something"}`. */ export function extractPrompt(msg: MessageReceiveEvent["message"]): string | null { // Check for bot mention — mentions[].id.open_id should include the bot. // The dispatcher already routes only message events; we further require // that the bot is among the mentions (if no mentions, it's not @bot). const mentions = msg.mentions; if (mentions === undefined || mentions.length === 0) return null; try { const parsed = JSON.parse(msg.content) as { text?: string }; const text = parsed.text; if (typeof text !== "string") return null; // Strip @mentions (@_user_1 etc.) and trim; remainder is the prompt. const stripped = text.replace(/@_\w+\s*/g, "").trim(); return stripped === "" ? null : stripped; } catch { return null; } } /** * Parse a leading `/` command from a prompt (e.g. `/draft 帮我写教案`). * Returns the role id and the remaining prompt with the command stripped. * Control commands already handled upstream (`/new`, `/resume`, `/reset`) are * ignored here — they never reach this function. * * Unknown `/foo` that isn't a registered role: returns `null` role and the * original prompt unchanged (the slash is treated as literal text). Role * resolution still happens via the registry's fallback. */ export function extractRole( prompt: string, registry: { role(id: string): unknown }, ): { roleId: string | null; prompt: string } { const m = /^\/(\S+)\s*/.exec(prompt); if (m === null || m[1] === undefined) return { roleId: null, prompt }; const roleId = m[1]; if (registry.role(roleId) === undefined) { // Unknown slash command — leave the prompt as-is (no silent role switch). return { roleId: null, prompt }; } return { roleId, prompt: prompt.slice(m[0].length) }; }