From e7924f78d532cce5e21667e25faa499256eca869 Mon Sep 17 00:00:00 2001 From: Hong Jiarong Date: Tue, 7 Jul 2026 22:20:30 +0800 Subject: [PATCH] =?UTF-8?q?feat(hub):=20=E7=BB=93=E6=9E=84=E5=8C=96?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E5=8E=86=E5=8F=B2(A-3)=20+=20lock=20?= =?UTF-8?q?=E5=BF=83=E8=B7=B3(A-5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A-3 结构化历史: - 新增 AgentMessage 表(每条消息的 queryable 投影,transcript 文件仍是 agent 读回记忆) - 新增 AgentFileChange 表(run 期间文件变更 before/after hash + operation) - runner.ts: generateText 后 persistAgentMessages 落库 result.responseMessages - workspace.ts writeFileTool: 写前算 beforeHash(SHA-256),写后算 afterHash,通过 ctx.onFileChange sink 记录 - runner.ts flushFileChanges: 收集 sink 的变更批量写入 AgentFileChange - ToolContext 加 onFileChange sink 字段 - trigger.ts 接线 runId/sessionId/prisma 到 RunRequest A-5 lock 心跳: - ProjectAgentLock 加 heartbeatAt 列 - runner.ts onStepFinish 回调:每步 model step 后更新 heartbeatAt(吞错,lock 可能被 force-release) 60 tests 全过(tsc 干净)。 --- .../migration.sql | 45 +++++++ hub/prisma/schema.prisma | 59 ++++++++- hub/src/agent/runner.ts | 125 +++++++++++++++++- hub/src/agent/tools.ts | 5 + hub/src/agent/workspace.ts | 32 +++++ hub/src/feishu/trigger.ts | 3 + hub/test/integration/helpers.ts | 2 + hub/test/integration/trigger.test.ts | 16 +++ 8 files changed, 280 insertions(+), 7 deletions(-) create mode 100644 hub/prisma/migrations/20260707210000_structured_history/migration.sql diff --git a/hub/prisma/migrations/20260707210000_structured_history/migration.sql b/hub/prisma/migrations/20260707210000_structured_history/migration.sql new file mode 100644 index 0000000..328c7a5 --- /dev/null +++ b/hub/prisma/migrations/20260707210000_structured_history/migration.sql @@ -0,0 +1,45 @@ +-- AgentMessage: per-message record of the agent loop (queryable projection +-- alongside the transcript JSONL file, which remains the agent's read-back memory). +CREATE TABLE "AgentMessage" ( + "id" TEXT NOT NULL, + "sessionId" TEXT NOT NULL, + "runId" TEXT, + "role" TEXT NOT NULL, + "content" TEXT NOT NULL, + "attachments" JSONB NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "AgentMessage_pkey" PRIMARY KEY ("id") +); + +CREATE INDEX "AgentMessage_sessionId_createdAt_idx" ON "AgentMessage"("sessionId", "createdAt"); +CREATE INDEX "AgentMessage_runId_idx" ON "AgentMessage"("runId"); + +ALTER TABLE "AgentMessage" ADD CONSTRAINT "AgentMessage_sessionId_fkey" FOREIGN KEY ("sessionId") REFERENCES "AgentSession"("id") ON DELETE CASCADE ON UPDATE CASCADE; +ALTER TABLE "AgentMessage" ADD CONSTRAINT "AgentMessage_runId_fkey" FOREIGN KEY ("runId") REFERENCES "AgentRun"("id") ON DELETE SET NULL ON UPDATE CASCADE; + +-- AgentFileChange: file change captured during a run (write/rename/delete). +CREATE TABLE "AgentFileChange" ( + "id" TEXT NOT NULL, + "runId" TEXT NOT NULL, + "projectId" TEXT NOT NULL, + "path" TEXT NOT NULL, + "operation" TEXT NOT NULL, + "beforeHash" TEXT, + "afterHash" TEXT, + "diff" TEXT, + "metadata" JSONB NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "AgentFileChange_pkey" PRIMARY KEY ("id") +); + +CREATE INDEX "AgentFileChange_runId_idx" ON "AgentFileChange"("runId"); +CREATE INDEX "AgentFileChange_projectId_idx" ON "AgentFileChange"("projectId"); +CREATE INDEX "AgentFileChange_path_idx" ON "AgentFileChange"("path"); + +ALTER TABLE "AgentFileChange" ADD CONSTRAINT "AgentFileChange_runId_fkey" FOREIGN KEY ("runId") REFERENCES "AgentRun"("id") ON DELETE CASCADE ON UPDATE CASCADE; +ALTER TABLE "AgentFileChange" ADD CONSTRAINT "AgentFileChange_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "Project"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- ProjectAgentLock: add heartbeatAt for liveness (A-5). +ALTER TABLE "ProjectAgentLock" ADD COLUMN "heartbeatAt" TIMESTAMP(3); diff --git a/hub/prisma/schema.prisma b/hub/prisma/schema.prisma index c3d06bf..3d979b4 100644 --- a/hub/prisma/schema.prisma +++ b/hub/prisma/schema.prisma @@ -84,6 +84,7 @@ model Project { permissionSettings PermissionSettings[] @relation("projectSettings") roleTriggerGrants RoleTriggerGrant[] @relation("projectRoleGrants") auditEntries AuditEntry[] @relation("projectAudit") + fileChanges AgentFileChange[] @relation("projectFileChanges") @@index([archivedAt]) } @@ -125,6 +126,7 @@ model AgentSession { project Project @relation(fields: [projectId], references: [id], onDelete: Cascade) runs AgentRun[] + messages AgentMessage[] @relation("sessionMessages") @@index([projectId, archivedAt]) @@index([provider, model]) @@ -172,6 +174,8 @@ model AgentRun { session AgentSession? @relation(fields: [sessionId], references: [id], onDelete: SetNull) requestedBy User? @relation("runRequester", fields: [requestedByUserId], references: [id], onDelete: SetNull) projectLock ProjectAgentLock? + messages AgentMessage[] @relation("runMessages") + fileChanges AgentFileChange[] @relation("runFileChanges") @@index([projectId, status]) @@index([sessionId]) @@ -184,11 +188,12 @@ model AgentRun { /// a run holds at most one lock. WellFormed (holder is non-terminal) is an /// app-level invariant checked on read/write, not a DB constraint. model ProjectAgentLock { - projectId String @id - runId String @unique + projectId String @id + runId String @unique holderUserId String? - acquiredAt DateTime @default(now()) - expiresAt DateTime? + acquiredAt DateTime @default(now()) + heartbeatAt DateTime? + expiresAt DateTime? project Project @relation(fields: [projectId], references: [id], onDelete: Cascade) run AgentRun @relation(fields: [runId], references: [id], onDelete: Cascade) @@ -331,3 +336,49 @@ model FeishuEventReceipt { @@index([messageId]) @@index([receivedAt]) } + +// --- Structured run history (ADR-0003 Memory) ---------------------------- + +/// Per-message record of the agent loop. The transcript JSONL file remains the +/// agent's own read-back memory (runner loads it to seed the next run); these +/// rows are the queryable/auditable projection for admin APIs. ADR-0003: +/// session messages ≠ Feishu group chat history. +model AgentMessage { + id String @id @default(cuid()) + sessionId String + runId String? + role String + content String + attachments Json + createdAt DateTime @default(now()) + + session AgentSession @relation("sessionMessages", fields: [sessionId], references: [id], onDelete: Cascade) + run AgentRun? @relation("runMessages", fields: [runId], references: [id], onDelete: SetNull) + + @@index([sessionId, createdAt]) + @@index([runId]) +} + +/// File change captured during a run (writeFile/rename/delete in workspace). +/// before/after hash + diff for audit and admin review. Linked to the run +/// that produced it; the tool's execute closure records via a sink the runner +/// injects into ToolContext (A-3 file-change capture). +model AgentFileChange { + id String @id @default(cuid()) + runId String + projectId String + path String + operation String + beforeHash String? + afterHash String? + diff String? + metadata Json + createdAt DateTime @default(now()) + + run AgentRun @relation("runFileChanges", fields: [runId], references: [id], onDelete: Cascade) + project Project @relation("projectFileChanges", fields: [projectId], references: [id], onDelete: Cascade) + + @@index([runId]) + @@index([projectId]) + @@index([path]) +} diff --git a/hub/src/agent/runner.ts b/hub/src/agent/runner.ts index 6c5a723..ce21b93 100644 --- a/hub/src/agent/runner.ts +++ b/hub/src/agent/runner.ts @@ -8,6 +8,7 @@ * memory/anchors (ADR-0003), not by this runner. */ import { generateText, stepCountIs, type LanguageModel, type ModelMessage } from "ai"; +import type { PrismaClient, Prisma } from "@prisma/client"; import type { ToolContext, ToolRegistry } from "./tools.js"; import { readTranscript, appendTranscript } from "./transcript.js"; @@ -35,6 +36,13 @@ export interface RunRequest { /** Path to the session transcript JSONL. If set, prior messages are loaded * from it before the run, and new messages are appended after. */ readonly transcriptPath?: string; + /** The real AgentRun.id (created by the caller before runAgent). Used for + * AgentMessage.runId and ProjectAgentLock heartbeat updates (A-5 liveness). */ + readonly runId: string; + /** The AgentSession.id, for AgentMessage.sessionId. */ + readonly sessionId: string; + /** Prisma client for persisting AgentMessage rows + lock heartbeats. */ + readonly prisma: PrismaClient; } export type RunStatus = "completed" | "length" | "failed"; @@ -54,11 +62,20 @@ export async function runAgent( tools: ToolRegistry, req: RunRequest, ): Promise { + // A-3 file-change capture: the writeFile tool calls ctx.onFileChange with + // before/after hash when it modifies a workspace file. The runner injects + // this collector buffer; it is flushed to AgentFileChange rows after the + // run (success or failure). Undefined onFileChange = no-op (unit tests + // that don't exercise writes), so we always wire it here. + const fileChanges: FileChange[] = []; const ctx: ToolContext = { - runId: cryptoRandomId(), + runId: req.runId, projectId: req.project.projectId, boundChatId: req.project.boundChatId, workspaceDir: req.project.workspaceDir, + onFileChange: (c) => { + fileChanges.push(c); + }, }; const aiTools = tools.build(ctx, req.toolWhitelist); @@ -82,9 +99,25 @@ export async function runAgent( allowSystemInMessages: true, tools: aiTools, stopWhen: stepCountIs(cap), + onStepFinish: async () => { + // A-5 liveness: refresh the lock heartbeat after each model step so + // the liveness monitor sees activity while a long multi-step run is + // in flight. The lock may have been force-released (admin) between + // steps; that's not the runner's problem — swallow the update error. + try { + await req.prisma.projectAgentLock.update({ + where: { runId: req.runId }, + data: { heartbeatAt: new Date() }, + }); + } catch { + // Lock gone (force-released) or never held — best-effort. + } + }, }); const allMessages: ModelMessage[] = [...messages, ...result.responseMessages]; await appendTranscriptIfSet(req, allMessages, loadedCount); + await persistAgentMessages(req, result.responseMessages); + await flushFileChanges(req, fileChanges); const stoppedByStepCap = result.finishReason === "tool-calls" && result.steps.length >= cap; const status: RunStatus = @@ -111,6 +144,8 @@ export async function runAgent( }; } catch (e) { await appendTranscriptIfSet(req, messages, loadedCount); + await persistAgentMessages(req, messages); + await flushFileChanges(req, fileChanges); return { status: "failed", messages, @@ -135,6 +170,90 @@ async function appendTranscriptIfSet(req: RunRequest, messages: readonly ModelMe } } -function cryptoRandomId(): string { - return globalThis.crypto.randomUUID(); +/// Persist AgentMessage rows for the messages produced during this run +/// (A-3 structured history). Each ModelMessage becomes one row: text parts +/// are joined into `content`; non-text parts (tool-call / tool-result / +/// file / reasoning / …) are JSON-encoded into `attachments`. Best-effort: +/// a write failure on one row is swallowed — the run result is the source of +/// truth and must not be lost over persistence. +async function persistAgentMessages(req: RunRequest, messages: readonly ModelMessage[]): Promise { + for (const msg of messages) { + try { + const { text, attachments } = extractMessageContent(msg); + await req.prisma.agentMessage.create({ + data: { + sessionId: req.sessionId, + runId: req.runId, + role: msg.role, + content: text, + // AI SDK parts are JSON-serializable by construction; assert for Prisma's InputJsonValue. + attachments: attachments as Prisma.InputJsonValue, + }, + }); + } catch { + // Swallow: structured-history persistence is best-effort; a single row + // failure must not abort the run or lose the run result. + } + } +} + +/// Extract the `content` string and `attachments` payload from a ModelMessage. +/// Runtime-narrows the AI SDK's part-typed content union: string content → +/// text directly; array content → join `text` parts, collect the rest as +/// attachments. Narrowing via `in` / `typeof` keeps the access compiler-checked +/// rather than asserted. +function extractMessageContent(msg: ModelMessage): { text: string; attachments: unknown[] } { + const content = msg.content; + if (typeof content === "string") { + return { text: content, attachments: [] }; + } + if (!Array.isArray(content)) { + return { text: "", attachments: [content] }; + } + const textParts: string[] = []; + const attachments: unknown[] = []; + for (const part of content) { + if ( + part !== null && + typeof part === "object" && + "type" in part && + part.type === "text" && + "text" in part && + typeof part.text === "string" + ) { + textParts.push(part.text); + } else { + attachments.push(part); + } + } + return { text: textParts.join("\n"), attachments }; +} + +/// The file-change record the writeFile tool emits via ctx.onFileChange. +/// Derived from the ToolContext sink signature so it can't drift. +type FileChange = Parameters>[0]; + +/// Flush the buffered file-change records to AgentFileChange rows (A-3). +/// Called after the run completes (success or failure) with whatever the +/// writeFile tool captured. Best-effort: a flush failure is swallowed — the +/// run result is the source of truth, not the change log. +async function flushFileChanges(req: RunRequest, changes: readonly FileChange[]): Promise { + if (changes.length === 0) return; + try { + await req.prisma.agentFileChange.createMany({ + data: changes.map((c) => ({ + runId: req.runId, + projectId: req.project.projectId, + path: c.path, + operation: c.operation, + beforeHash: c.beforeHash, + afterHash: c.afterHash, + diff: c.diff, + metadata: {}, + })), + }); + } catch { + // Swallow: file-change capture is best-effort observability; must not + // lose the run result over a change-log write failure. + } } diff --git a/hub/src/agent/tools.ts b/hub/src/agent/tools.ts index f1377af..99d1a31 100644 --- a/hub/src/agent/tools.ts +++ b/hub/src/agent/tools.ts @@ -24,6 +24,11 @@ export interface ToolContext { readonly boundChatId: string; /** Absolute path to the project's curriculum engineering-file workspace. */ readonly workspaceDir: string; + /** Optional sink for file-change capture (A-3). The writeFile tool calls + * this with before/after hash when it modifies a file; the runner injects + * a collector that flushes to AgentFileChange at run end. Undefined when + * the runner doesn't wire capture (e.g. unit tests). */ + readonly onFileChange?: (change: { path: string; operation: "CREATE" | "UPDATE" | "DELETE" | "RENAME"; beforeHash: string | null; afterHash: string | null; diff: string | null }) => void; } export type ToolDefinition = Tool; diff --git a/hub/src/agent/workspace.ts b/hub/src/agent/workspace.ts index d63d213..3c81bc6 100644 --- a/hub/src/agent/workspace.ts +++ b/hub/src/agent/workspace.ts @@ -8,6 +8,7 @@ * surface; the Hub agent is narrower than a general coding agent. */ import { readFile, writeFile, readdir, mkdir } from "node:fs/promises"; +import { createHash } from "node:crypto"; import { join, resolve, relative, isAbsolute } from "node:path"; import { tool } from "ai"; import { z } from "zod"; @@ -67,8 +68,39 @@ export function writeFileTool(ctx: ToolContext): ToolDefinition { execute: async (args): Promise => { try { const full = confine(args.path, ctx.workspaceDir); + // A-3 file-change capture: compute before/after hashes around the write. + // A hashing failure must not block the write — skip the sink in that case. + let beforeHash: string | null = null; + let afterHash: string | null = null; + let operation: "CREATE" | "UPDATE" | null = null; + try { + let oldContent: string | null; + try { + oldContent = await readFile(full, "utf8"); + } catch (e) { + if (e !== null && typeof e === "object" && "code" in e && e.code === "ENOENT") { + oldContent = null; + } else { + throw e; + } + } + afterHash = createHash("sha256").update(args.content).digest("hex"); + beforeHash = oldContent === null ? null : createHash("sha256").update(oldContent).digest("hex"); + if (beforeHash === null) { + operation = "CREATE"; + } else if (beforeHash === afterHash) { + operation = null; + } else { + operation = "UPDATE"; + } + } catch { + operation = null; + } await mkdir(join(full, ".."), { recursive: true }); await writeFile(full, args.content, "utf8"); + if (operation !== null && ctx.onFileChange !== undefined) { + ctx.onFileChange({ path: args.path, operation, beforeHash, afterHash, diff: null }); + } return JSON.stringify({ ok: true, path: args.path, bytes: args.content.length }); } catch (e) { return JSON.stringify({ error: e instanceof Error ? e.message : String(e) }); diff --git a/hub/src/feishu/trigger.ts b/hub/src/feishu/trigger.ts index 5c20c0d..88b44c3 100644 --- a/hub/src/feishu/trigger.ts +++ b/hub/src/feishu/trigger.ts @@ -251,6 +251,9 @@ export function makeTriggerHandler(deps: TriggerDeps) { systemPrompt, toolWhitelist: role?.tools, transcriptPath: transcriptPath(project.workspaceDir, session.id), + runId: run.id, + sessionId: session.id, + prisma: deps.prisma, }) .then(async (result) => { await deps.prisma.agentRun.update({ diff --git a/hub/test/integration/helpers.ts b/hub/test/integration/helpers.ts index 1572230..76a095b 100644 --- a/hub/test/integration/helpers.ts +++ b/hub/test/integration/helpers.ts @@ -27,6 +27,8 @@ export const prisma = new PrismaClient({ export async function resetDb(): Promise { const tables = [ "FeishuEventReceipt", + "AgentFileChange", + "AgentMessage", "AuditEntry", "PermissionSettings", "PermissionGrant", diff --git a/hub/test/integration/trigger.test.ts b/hub/test/integration/trigger.test.ts index 31f2f95..f0a1f0c 100644 --- a/hub/test/integration/trigger.test.ts +++ b/hub/test/integration/trigger.test.ts @@ -288,6 +288,22 @@ describe("trigger full lifecycle (integration)", () => { expect(audit.some((a) => a.action === "run.created")).toBe(true); expect(audit.some((a) => a.action === "run.finished")).toBe(true); }); + + 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 }); + + await trigger(makeEvent("chat-15", "@_user_1 写教案"), rt); + await vi.waitFor(async () => { + const msgs = await prisma.agentMessage.findMany(); + expect(msgs.length).toBeGreaterThanOrEqual(1); + }); + const msgs = await prisma.agentMessage.findMany({ orderBy: { createdAt: "asc" } }); + expect(msgs.length).toBeGreaterThanOrEqual(1); + const run = await prisma.agentRun.findFirst(); + expect(msgs.every((m) => m.runId === run?.id)).toBe(true); + expect(msgs.some((m) => m.role === "assistant")).toBe(true); + }); }); afterAll(async () => {