feat(hub): 事件去重(A-4) + 审计写入路径(A-8)

A-4 FeishuEventReceipt: - 新增 FeishuEventReceipt 表(eventId @unique),lark ws 至少一次投递的幂等兜底 - MessageReceiveEvent 加 header.event_id 字段 - trigger 入口查重:已存在的 event_id → 跳过(不建第二次 run) - resetDb 加入 FeishuEventReceipt A-8 审计写入: - AuditEntry 加 projectId(权限拒绝/slash 命令等无 run 的审计点可定位) - 新建 src/audit.ts:writeAudit(prisma, entry) best-effort 写入(吞错,不阻断 trigger) - trigger 五个审计点:trigger.denied / trigger.role_denied / run.created / run.lock_race / run.finished / run.failed - 新增 audit unit 测试(3) + trigger 集成测试(去重 + 审计,2) 59 tests 全过。
2026-07-07 21:34:15 +08:00
parent a2c8fa8eaf
commit f8c3e16a52
9 changed files with 261 additions and 7 deletions
@@ -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
@@ -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;
@@ -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)
+  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])
+}
@@ -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<string, unknown>;
+}
+
+/**
+ * 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<void> {
+  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.
+  }
+}
@@ -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;
@@ -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,
@@ -26,6 +26,7 @@ export const prisma = new PrismaClient({
 /** Truncate all tables before each test for isolation. */
 export async function resetDb(): Promise<void> {
  const tables = [
+    "FeishuEventReceipt",
    "AuditEntry",
    "PermissionSettings",
    "PermissionGrant",
@@ -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 () => {
@@ -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<typeof vi.fn>;
+}
+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);
+  });
+});