diff --git a/docs/adr/0017-agent-session-is-provider-bound.md b/docs/adr/0017-agent-session-is-provider-bound.md index 88ed160..37af679 100644 --- a/docs/adr/0017-agent-session-is-provider-bound.md +++ b/docs/adr/0017-agent-session-is-provider-bound.md @@ -36,9 +36,12 @@ Per-run model selection (the run carries its model) is the switching point. - Switching model mid-project = new session; the new run seeds from project memory/anchors (ADR-0003), not the prior session's live context. -- The agent layer is provider-agnostic: the agent loop is ours; model calls go - through an OpenAI-compatible client. No hard dependency on a single-vendor - agent SDK. +- The agent layer is provider-agnostic: model calls go through an + OpenAI-compatible client, and the tool-calling loop is delegated to the + Vercel AI SDK (`generateText` + `tool`). The provider seam is the SDK's + `LanguageModel` interface, not a hand-rolled loop; swapping provider is + swapping the `createOpenAICompatible` factory, not rewriting the loop. No + hard dependency on a single-vendor agent SDK. - Role-based model routing (different run kinds default to different models) is a product/admin configuration concern, not a spec invariant. - "AgentSession reused by many runs" (ADR-0002) holds *within* one diff --git a/hub/prisma/migrations/20260707200000_dedup_and_audit_project/migration.sql b/hub/prisma/migrations/20260707200000_dedup_and_audit_project/migration.sql new file mode 100644 index 0000000..8f35eae --- /dev/null +++ b/hub/prisma/migrations/20260707200000_dedup_and_audit_project/migration.sql @@ -0,0 +1,23 @@ +-- FeishuEventReceipt: idempotency for inbound ws events (at-least-once redelivery). +CREATE TABLE "FeishuEventReceipt" ( + "id" TEXT NOT NULL, + "eventId" TEXT NOT NULL, + "eventType" TEXT NOT NULL, + "messageId" TEXT, + "receivedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "FeishuEventReceipt_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "FeishuEventReceipt_eventId_key" ON "FeishuEventReceipt"("eventId"); +CREATE INDEX "FeishuEventReceipt_eventType_idx" ON "FeishuEventReceipt"("eventType"); +CREATE INDEX "FeishuEventReceipt_messageId_idx" ON "FeishuEventReceipt"("messageId"); +CREATE INDEX "FeishuEventReceipt_receivedAt_idx" ON "FeishuEventReceipt"("receivedAt"); + +-- AuditEntry: add projectId so audit points before/without a run (permission +-- denials, slash commands) are locatable to a project. +ALTER TABLE "AuditEntry" ADD COLUMN "projectId" TEXT; + +CREATE INDEX "AuditEntry_projectId_createdAt_idx" ON "AuditEntry"("projectId", "createdAt"); + +ALTER TABLE "AuditEntry" ADD CONSTRAINT "AuditEntry_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "Project"("id") ON DELETE SET NULL ON UPDATE CASCADE; diff --git a/hub/prisma/schema.prisma b/hub/prisma/schema.prisma index c0bc8b4..c3d06bf 100644 --- a/hub/prisma/schema.prisma +++ b/hub/prisma/schema.prisma @@ -83,6 +83,7 @@ model Project { permissionGrants PermissionGrant[] @relation("projectGrants") permissionSettings PermissionSettings[] @relation("projectSettings") roleTriggerGrants RoleTriggerGrant[] @relation("projectRoleGrants") + auditEntries AuditEntry[] @relation("projectAudit") @@index([archivedAt]) } @@ -294,18 +295,39 @@ model RoleTriggerGrant { /// spec AuditEntry: minimal skeleton — one entry relates to a run. Event type, /// actor, timestamp, details are OPEN. This table mirrors that: `runId` is the -/// PINNED relation; the rest is open-shaped columns. model AuditEntry { id String @id @default(cuid()) runId String? + projectId String? actorUserId String? action String metadata Json createdAt DateTime @default(now()) - actor User? @relation("auditActor", fields: [actorUserId], references: [id], onDelete: SetNull) + actor User? @relation("auditActor", fields: [actorUserId], references: [id], onDelete: SetNull) + project Project? @relation("projectAudit", fields: [projectId], references: [id], onDelete: SetNull) @@index([runId]) + @@index([projectId, createdAt]) @@index([actorUserId]) @@index([createdAt]) } + +// --- Feishu event dedup -------------------------------------------------- + +/// Idempotency receipt for inbound Feishu ws events. The lark ws client may +/// redeliver an event (at-least-once); without dedup a duplicate would spawn a +/// second AgentRun — the lock would refuse it, but a FAILED run row + a busy +/// reply would still leak. `eventId @unique` makes the second insert a no-op +/// signal: the handler checks existence before processing. +model FeishuEventReceipt { + id String @id @default(cuid()) + eventId String @unique + eventType String + messageId String? + receivedAt DateTime @default(now()) + + @@index([eventType]) + @@index([messageId]) + @@index([receivedAt]) +} diff --git a/hub/src/audit.ts b/hub/src/audit.ts new file mode 100644 index 0000000..f92941a --- /dev/null +++ b/hub/src/audit.ts @@ -0,0 +1,59 @@ +/** + * Audit write path — A-8. The `AuditEntry` table existed but had no writer; + * callers (trigger path, permission denials, slash commands) had no way to + * record what happened. This module is that single write seam. + * + * ADR reference: spec Audit. The skeleton pins only the `runId` relation; + * event type / actor / timestamp / details are content-OPEN. This helper + * exposes `action` (event type) + `actorUserId` (actor) + `metadata` + * (details) + `projectId` (locates audit points that don't have a run, e.g. + * permission denials) + `runId` (loose string ref, not a Prisma relation — + * see schema). `createdAt` is `@default(now())` so we don't pass it. + * + * Best-effort by design: an audit failure must never break the caller's path + * (a trigger failing because the audit log is down would be a spec breach — + * audit is observability, not control). Errors are swallowed here. There's no + * logger in this layer yet; a future logger can be threaded in without + * changing the signature. + */ +import type { PrismaClient, Prisma } from "@prisma/client"; + +export interface AuditInput { + readonly runId?: string | undefined; + readonly projectId?: string | undefined; + readonly actorUserId?: string | undefined; + readonly action: string; + readonly metadata: Record; +} + +/** + * Write one audit entry. Best-effort: any Prisma error is swallowed so the + * caller's path (trigger, permission gate, slash command) is unaffected. + * + * Optional fields are only included when defined — under + * `exactOptionalPropertyTypes`, assigning `undefined` to an optional key in a + * Prisma `create` payload is a type error (and would also write an explicit + * NULL rather than omit the column). Conditional assignment keeps the payload + * honest. The data object is typed as `AuditEntryUncheckedCreateInput` (the + * direct-FK variant) so it matches one branch of Prisma's `create` input + * union exactly — the relation-connect branch has no `projectId`/`runId` + * columns and would otherwise reject these fields. + */ +export async function writeAudit( + prisma: PrismaClient, + entry: AuditInput, +): Promise { + const data: Prisma.AuditEntryUncheckedCreateInput = { + action: entry.action, + metadata: entry.metadata as Prisma.InputJsonValue, + }; + if (entry.runId !== undefined) data.runId = entry.runId; + if (entry.projectId !== undefined) data.projectId = entry.projectId; + if (entry.actorUserId !== undefined) data.actorUserId = entry.actorUserId; + + try { + await prisma.auditEntry.create({ data }); + } catch { + // Swallow: audit is best-effort and must not break the caller's path. + } +} diff --git a/hub/src/feishu/client.ts b/hub/src/feishu/client.ts index e7c3570..4f01a73 100644 --- a/hub/src/feishu/client.ts +++ b/hub/src/feishu/client.ts @@ -26,6 +26,10 @@ export interface FeishuRuntime { /** Raw `im.message.receive_v1` event payload (subset we use). */ export interface MessageReceiveEvent { + readonly header?: { + readonly event_id?: string; + readonly event_type?: string; + }; readonly message: { readonly message_id: string; readonly chat_id: string; diff --git a/hub/src/feishu/trigger.ts b/hub/src/feishu/trigger.ts index 019ec29..5c20c0d 100644 --- a/hub/src/feishu/trigger.ts +++ b/hub/src/feishu/trigger.ts @@ -22,6 +22,7 @@ import { runAgent, type ModelFactory, type ProjectContext } from "../agent/runne 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; @@ -42,6 +43,29 @@ export function makeTriggerHandler(deps: TriggerDeps) { 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 messages in group chats. if (msg.message_type !== "text") return; const chatId = msg.chat_id; @@ -64,6 +88,7 @@ export function makeTriggerHandler(deps: TriggerDeps) { 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; } @@ -143,6 +168,7 @@ export function makeTriggerHandler(deps: TriggerDeps) { 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; } @@ -193,16 +219,17 @@ export function makeTriggerHandler(deps: TriggerDeps) { }, 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 { - // Race: another run grabbed the lock between our check and create. 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; } @@ -241,6 +268,7 @@ export function makeTriggerHandler(deps: TriggerDeps) { finishedAt: new Date(), }, }); + await writeAudit(deps.prisma, { runId: run.id, projectId, action: "run.finished", metadata: { status: result.status, inputTokens: result.usage.inputTokens, outputTokens: result.usage.outputTokens } }); await sendCard(rt, chatId, buildRunStatusCard({ status: statusReply(result.status), model, @@ -260,6 +288,7 @@ export function makeTriggerHandler(deps: TriggerDeps) { } catch (updateErr) { deps.logger.warn({ runId: run.id, err: String(updateErr) }, "trigger: could not mark run FAILED (already gone?)"); } + await writeAudit(deps.prisma, { runId: run.id, projectId, action: "run.failed", metadata: { error: String(e) } }); await sendCard(rt, chatId, buildRunStatusCard({ status: "处理出错", model, diff --git a/hub/test/integration/helpers.ts b/hub/test/integration/helpers.ts index b2dbbc7..1572230 100644 --- a/hub/test/integration/helpers.ts +++ b/hub/test/integration/helpers.ts @@ -26,6 +26,7 @@ export const prisma = new PrismaClient({ /** Truncate all tables before each test for isolation. */ export async function resetDb(): Promise { const tables = [ + "FeishuEventReceipt", "AuditEntry", "PermissionSettings", "PermissionGrant", diff --git a/hub/test/integration/trigger.test.ts b/hub/test/integration/trigger.test.ts index c6f30b4..31f2f95 100644 --- a/hub/test/integration/trigger.test.ts +++ b/hub/test/integration/trigger.test.ts @@ -8,8 +8,10 @@ import type { ModelFactory } from "../../src/agent/runner.js"; const bot = { key: "@_user_1", id: { open_id: "ou_bot" }, name: "Bot" }; -function makeEvent(chatId: string, text: string, senderOpenId = "ou_test_user"): MessageReceiveEvent { +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; return { + header, message: { message_id: "m_" + Math.random().toString(36).slice(2), chat_id: chatId, @@ -248,6 +250,44 @@ describe("trigger full lifecycle (integration)", () => { expect(runs[0]?.metadata).toMatchObject({ roleId: "draft" }); }); }); + + 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 }); + + // First delivery: processes normally. + await trigger(makeEvent("chat-13", "@_user_1 写教案", "ou_test_user", "evt-dedup-1"), rt); + await vi.waitFor(async () => { + const runs = await prisma.agentRun.findMany(); + expect(runs).toHaveLength(1); + expect(runs[0]?.status).toBe("COMPLETED"); + }); + + // Redelivery of the same event_id: must not create a second run. + await trigger(makeEvent("chat-13", "@_user_1 写教案", "ou_test_user", "evt-dedup-1"), rt); + const runs2 = await prisma.agentRun.findMany(); + expect(runs2).toHaveLength(1); + + // Only one event receipt row. + const receipts = await prisma.feishuEventReceipt.findMany(); + expect(receipts).toHaveLength(1); + expect(receipts[0]?.eventId).toBe("evt-dedup-1"); + }); + + it("writes audit entries across the run lifecycle", async () => { + await seedProject("proj-14", "chat-14"); + const trigger = makeTriggerHandler({ prisma, modelFactory, tools, models, logger: silentLogger }); + + await trigger(makeEvent("chat-14", "@_user_1 写教案"), rt); + await vi.waitFor(async () => { + const audit = await prisma.auditEntry.findMany(); + expect(audit.length).toBeGreaterThanOrEqual(2); + }); + + const audit = await prisma.auditEntry.findMany({ orderBy: { createdAt: "asc" } }); + expect(audit.some((a) => a.action === "run.created")).toBe(true); + expect(audit.some((a) => a.action === "run.finished")).toBe(true); + }); }); afterAll(async () => { diff --git a/hub/test/unit/audit.test.ts b/hub/test/unit/audit.test.ts new file mode 100644 index 0000000..24311fa --- /dev/null +++ b/hub/test/unit/audit.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it, vi } from "vitest"; +import type { PrismaClient } from "@prisma/client"; +import { writeAudit } from "../../src/audit.js"; + +/// Minimal mock: only the surface `writeAudit` touches. Cast through unknown +/// so we don't have to implement the rest of the PrismaClient surface — this +/// is a unit test, not a DB test. +interface AuditEntryMock { + create: ReturnType; +} +function mockPrisma(): PrismaClient & { auditEntry: AuditEntryMock } { + return { auditEntry: { create: vi.fn() } } as unknown as PrismaClient & { + auditEntry: AuditEntryMock; + }; +} + +describe("writeAudit (A-8 audit write path)", () => { + it("creates a row with all fields set", async () => { + const prisma = mockPrisma(); + prisma.auditEntry.create.mockResolvedValue({ id: "aud-1" }); + + await writeAudit(prisma, { + runId: "run-1", + projectId: "proj-1", + actorUserId: "user-1", + action: "run.triggered", + metadata: { source: "feishu" }, + }); + + expect(prisma.auditEntry.create).toHaveBeenCalledTimes(1); + expect(prisma.auditEntry.create).toHaveBeenCalledWith({ + data: { + runId: "run-1", + projectId: "proj-1", + actorUserId: "user-1", + action: "run.triggered", + metadata: { source: "feishu" }, + }, + }); + }); + + it("creates a row with only required fields (action + metadata)", async () => { + const prisma = mockPrisma(); + prisma.auditEntry.create.mockResolvedValue({ id: "aud-2" }); + + await writeAudit(prisma, { + action: "permission.denied", + metadata: { reason: "no-edit-role" }, + }); + + expect(prisma.auditEntry.create).toHaveBeenCalledTimes(1); + expect(prisma.auditEntry.create).toHaveBeenCalledWith({ + data: { + action: "permission.denied", + metadata: { reason: "no-edit-role" }, + }, + }); + }); + + it("swallows a prisma error (best-effort, must not throw)", async () => { + const prisma = mockPrisma(); + prisma.auditEntry.create.mockRejectedValue(new Error("db down")); + + // Must not throw — audit is observability, not control. + await expect( + writeAudit(prisma, { + action: "run.triggered", + metadata: {}, + }), + ).resolves.toBeUndefined(); + expect(prisma.auditEntry.create).toHaveBeenCalledTimes(1); + }); +});