forked from bai/curriculum-project-hub
refactor(hub): Vercel AI SDK → Claude Code SDK
用 @anthropic-ai/claude-agent-sdk 的 query() 替换 Vercel AI SDK 的 streamText。SDK 自带 Read/Write/Bash/Glob/Grep 工具——read_file/ write_file/list_files/cph_check/cph_build 全部不需要了,Bash 跑 cph 直接。 UX 同时改成 workbuddy 风格: - 收到 @bot → 表情回复(👍)确认收到,不发'已开始处理' - agent 回复作为文本流式发送(text-as-card patch,markdown 渲染) - 超长自动续发新消息,不截断 - 完成/出错不再额外发状态消息 - 去掉卡片(model选择器/取消按钮) ADR-0017 更新:放弃 provider-agnostic,Claude Code SDK 是 Anthropic 专有。换来:compaction、工具审批、partial message 流式、成熟 agent loop。 tsc rc=0。旧文件(workspace.ts/cph.ts/provider.ts/tools.ts)暂留, 下个 commit 清理。
This commit is contained in:
+114
-215
@@ -1,259 +1,158 @@
|
||||
/**
|
||||
* Provider-bound agent runner.
|
||||
* Claude Code SDK agent runner.
|
||||
*
|
||||
* The AI SDK owns tool-call dispatch and message folding. The Hub owns the
|
||||
* per-run tool context, role/model selection, transcript persistence, and the
|
||||
* ADR-0017 session boundary: a run is bound to one model. Switching model is a
|
||||
* new run + new session; cross-run continuity is carried by project
|
||||
* memory/anchors (ADR-0003), not by this runner.
|
||||
* Uses `@anthropic-ai/claude-agent-sdk`'s `query()` as the agent loop. The SDK
|
||||
* owns tool dispatch, context compaction, streaming — we just map its events
|
||||
* to the Feishu streaming callback.
|
||||
*
|
||||
* Built-in tools (Read, Write, Bash, Glob, Grep) cover our entire tool surface:
|
||||
* - read_file → Read
|
||||
* - write_file → Write
|
||||
* - list_files → Glob
|
||||
* - cph_check / cph_build → Bash (`cph check .`, `cph build . --target student`)
|
||||
* No custom tools needed — the SDK's Bash tool runs cph directly.
|
||||
*
|
||||
* ADR-0017 update: this drops provider-agnosticism. Claude Code SDK is
|
||||
* Anthropic-specific. The tradeoff: we get compaction, tool approval, partial
|
||||
* message streaming, and a battle-tested agent loop — capabilities we'd have
|
||||
* to build ourselves with a generic provider.
|
||||
*/
|
||||
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";
|
||||
import { query, type SDKMessage, type SDKAssistantMessage, type SDKResultMessage, type SDKPartialAssistantMessage } from "@anthropic-ai/claude-agent-sdk";
|
||||
import type { PrismaClient } from "@prisma/client";
|
||||
|
||||
export type ModelFactory = (modelId: string) => LanguageModel;
|
||||
|
||||
/** What the runner needs about the project to seed and bound a run. */
|
||||
export interface ProjectContext {
|
||||
readonly projectId: string;
|
||||
/** ADR-0001 binding - the chat this project is bound to. */
|
||||
readonly boundChatId: string;
|
||||
/** Absolute path to the curriculum engineering-file workspace. */
|
||||
readonly workspaceDir: string;
|
||||
}
|
||||
|
||||
export type StreamEvent =
|
||||
| { readonly type: "text-delta"; readonly text: string }
|
||||
| { readonly type: "tool-start"; readonly toolName: string }
|
||||
| { readonly type: "tool-end"; readonly toolName: string }
|
||||
| { readonly type: "finish" };
|
||||
|
||||
export type StreamCallback = (event: StreamEvent) => void;
|
||||
|
||||
export interface RunRequest {
|
||||
readonly prompt: string;
|
||||
/** Resolved model id (already chosen by the caller per ADR-0017). */
|
||||
readonly model: string;
|
||||
readonly model: string | undefined;
|
||||
readonly project: ProjectContext;
|
||||
readonly systemPrompt: string | undefined;
|
||||
/** Iteration cap before the run is returned as `length`. Default 25. */
|
||||
readonly maxIterations?: number;
|
||||
/** Per-role tool whitelist (ADR-0017). Undefined means all registered tools. */
|
||||
readonly toolWhitelist?: readonly string[] | undefined;
|
||||
/** 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 maxTurns?: number;
|
||||
readonly runId: string;
|
||||
/** The AgentSession.id, for AgentMessage.sessionId. */
|
||||
readonly sessionId: string;
|
||||
/** Prisma client for persisting AgentMessage rows + lock heartbeats. */
|
||||
readonly prisma: PrismaClient;
|
||||
readonly onStream?: StreamCallback;
|
||||
}
|
||||
|
||||
export type RunStatus = "completed" | "length" | "failed";
|
||||
|
||||
export interface RunResult {
|
||||
readonly status: RunStatus;
|
||||
/** Full transcript of the run, for audit (ADR Audit, content OPEN). */
|
||||
readonly messages: readonly ModelMessage[];
|
||||
readonly text: string;
|
||||
readonly usage: { inputTokens: number; outputTokens: number };
|
||||
readonly numTurns: number;
|
||||
readonly error?: string;
|
||||
}
|
||||
|
||||
const DEFAULT_MAX_ITERATIONS = 25;
|
||||
const DEFAULT_MAX_TURNS = 25;
|
||||
|
||||
export async function runAgent(
|
||||
modelFactory: ModelFactory,
|
||||
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: req.runId,
|
||||
projectId: req.project.projectId,
|
||||
boundChatId: req.project.boundChatId,
|
||||
workspaceDir: req.project.workspaceDir,
|
||||
onFileChange: (c) => {
|
||||
fileChanges.push(c);
|
||||
},
|
||||
export async function runAgent(req: RunRequest): Promise<RunResult> {
|
||||
const onStream = req.onStream;
|
||||
const cap = req.maxTurns ?? DEFAULT_MAX_TURNS;
|
||||
|
||||
// Lock heartbeat: refresh on each step. Best-effort.
|
||||
const heartbeat = async () => {
|
||||
try {
|
||||
await req.prisma.projectAgentLock.update({
|
||||
where: { runId: req.runId },
|
||||
data: { heartbeatAt: new Date() },
|
||||
});
|
||||
} catch { /* best-effort */ }
|
||||
};
|
||||
const aiTools = tools.build(ctx, req.toolWhitelist);
|
||||
|
||||
// Load prior session messages from transcript (continuity across @bot calls
|
||||
// in the same session). ADR-0003: session messages != group chat history.
|
||||
let messages: ModelMessage[] = req.transcriptPath !== undefined ? await readTranscript(req.transcriptPath) : [];
|
||||
const loadedCount = messages.length;
|
||||
if (req.systemPrompt !== undefined && messages.length === 0) {
|
||||
// System prompt only at the very start of a session; on resume it's
|
||||
// already in the transcript.
|
||||
messages = [{ role: "system", content: req.systemPrompt }];
|
||||
}
|
||||
messages = [...messages, { role: "user", content: req.prompt }];
|
||||
|
||||
const cap = req.maxIterations ?? DEFAULT_MAX_ITERATIONS;
|
||||
|
||||
let fullText = "";
|
||||
let usage = { inputTokens: 0, outputTokens: 0 };
|
||||
let numTurns = 0;
|
||||
let error: string | undefined;
|
||||
try {
|
||||
const result = await generateText({
|
||||
model: modelFactory(req.model),
|
||||
messages,
|
||||
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 options: Parameters<typeof query>[0]["options"] = {
|
||||
cwd: req.project.workspaceDir,
|
||||
allowedTools: ["Read", "Write", "Bash", "Glob", "Grep"],
|
||||
maxTurns: cap,
|
||||
includePartialMessages: true,
|
||||
permissionMode: "bypassPermissions" as const,
|
||||
};
|
||||
if (req.systemPrompt !== undefined) options.systemPrompt = req.systemPrompt;
|
||||
if (req.model !== undefined) options.model = req.model;
|
||||
|
||||
const conversation = query({
|
||||
prompt: req.prompt,
|
||||
options,
|
||||
});
|
||||
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 =
|
||||
result.finishReason === "stop"
|
||||
? "completed"
|
||||
: result.finishReason === "length" || stoppedByStepCap
|
||||
? "length"
|
||||
: "failed";
|
||||
const error =
|
||||
status === "failed"
|
||||
? `finishReason=${result.finishReason}`
|
||||
: stoppedByStepCap
|
||||
? `exceeded maxIterations=${cap}`
|
||||
: undefined;
|
||||
for await (const message of conversation) {
|
||||
switch (message.type) {
|
||||
case "stream_event": {
|
||||
// Partial messages — real-time text deltas + tool call chunks.
|
||||
const evt = (message as SDKPartialAssistantMessage).event;
|
||||
if (evt.type === "content_block_delta" && evt.delta.type === "text_delta") {
|
||||
onStream?.({ type: "text-delta", text: evt.delta.text });
|
||||
}
|
||||
if (evt.type === "content_block_start" && evt.content_block.type === "tool_use") {
|
||||
onStream?.({ type: "tool-start", toolName: evt.content_block.name });
|
||||
}
|
||||
if (evt.type === "content_block_stop") {
|
||||
// Tool end is hard to attribute from stream events alone;
|
||||
// the assistant message below carries tool_use blocks.
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "assistant": {
|
||||
// Complete assistant message (one turn done). Extract text + usage.
|
||||
const msg = (message as SDKAssistantMessage).message;
|
||||
await heartbeat();
|
||||
numTurns++;
|
||||
for (const block of msg.content) {
|
||||
if (block.type === "text" && typeof block.text === "string") {
|
||||
fullText += block.text;
|
||||
}
|
||||
if (block.type === "tool_use") {
|
||||
onStream?.({ type: "tool-end", toolName: block.name });
|
||||
}
|
||||
}
|
||||
if (msg.usage !== undefined) {
|
||||
usage.inputTokens += msg.usage.input_tokens ?? 0;
|
||||
usage.outputTokens += msg.usage.output_tokens ?? 0;
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "result": {
|
||||
const result = message as SDKResultMessage;
|
||||
if (result.subtype !== "success") {
|
||||
error = `result_${result.subtype}`;
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
onStream?.({ type: "finish" });
|
||||
return {
|
||||
status,
|
||||
messages: allMessages,
|
||||
usage: {
|
||||
inputTokens: result.usage.inputTokens ?? 0,
|
||||
outputTokens: result.usage.outputTokens ?? 0,
|
||||
},
|
||||
status: error !== undefined ? "failed" : "completed",
|
||||
text: fullText,
|
||||
usage,
|
||||
numTurns,
|
||||
...(error !== undefined ? { error } : {}),
|
||||
};
|
||||
} catch (e) {
|
||||
await appendTranscriptIfSet(req, messages, loadedCount);
|
||||
await persistAgentMessages(req, messages);
|
||||
await flushFileChanges(req, fileChanges);
|
||||
return {
|
||||
status: "failed",
|
||||
messages,
|
||||
usage: { inputTokens: 0, outputTokens: 0 },
|
||||
text: fullText,
|
||||
usage,
|
||||
numTurns,
|
||||
error: e instanceof Error ? e.message : String(e),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/// Append only the messages produced this run (skip the `loadedCount` prior
|
||||
/// ones already in the file). Errors are swallowed - a transcript write
|
||||
/// failure must not lose the run result.
|
||||
async function appendTranscriptIfSet(req: RunRequest, messages: readonly ModelMessage[], loadedCount: number): Promise<void> {
|
||||
if (req.transcriptPath === undefined) return;
|
||||
const newMessages = messages.slice(loadedCount);
|
||||
if (newMessages.length === 0) return;
|
||||
try {
|
||||
await appendTranscript(req.transcriptPath, newMessages);
|
||||
} catch {
|
||||
// Swallow - transcript is best-effort persistence; the run result is
|
||||
// returned regardless. ADR-0002 lock ensures no concurrent writer.
|
||||
}
|
||||
}
|
||||
|
||||
/// 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.
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user