feat(hub): 结构化运行历史(A-3) + lock 心跳(A-5)

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 干净)。
This commit is contained in:
2026-07-07 22:20:30 +08:00
parent f8c3e16a52
commit e7924f78d5
8 changed files with 280 additions and 7 deletions
+122 -3
View File
@@ -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<RunResult> {
// 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<void> {
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<NonNullable<ToolContext["onFileChange"]>>[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<void> {
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.
}
}
+5
View File
@@ -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;
+32
View File
@@ -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<string> => {
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) });
+3
View File
@@ -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({