From 0fe6cabc6868ff8df5cdfa0c5269434c71d4b8fa Mon Sep 17 00:00:00 2001 From: Hong Jiarong Date: Tue, 7 Jul 2026 17:50:05 +0800 Subject: [PATCH] =?UTF-8?q?feat(hub):=20session=20transcript=20=E8=BF=9E?= =?UTF-8?q?=E7=BB=AD=E6=80=A7=20+=20slash=20commands=20(/new=20/resume=20/?= =?UTF-8?q?reset)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复每次 @bot 都是失忆状态的问题。 - transcript.ts: JSONL 文件存 session 对话消息(workspace/.cph/sessions/ .jsonl),append-only;和 Claude Code 一致;ADR-0003 说的'不存全 历史'是群聊历史,session 对话是 Hub 自己产生的,不冲突 - runner.ts: RunRequest 加 transcriptPath;run 开始时读历史拼进初始 messages(system prompt 只在首条);run 结束时 append 本轮新消息 (loadedCount 区分历史 vs 新增,不重复写);best-effort(写失败不丢 run 结果) - trigger.ts: slash commands 在 lock 检查之前(/new /resume /reset 不创建 run 不需要锁);transcriptPath 传给 runner /new = 归档当前 session,下次 @bot 自动建新的 /resume = 恢复最近归档的 session /reset = 同 /new 未知 /cmd 透传给 agent 当普通 prompt - unit: transcript.test.ts 读写回环/append 累积/损坏行恢复/路径(6) - integration: trigger.test.ts 加 /new /resume /reset /unknown(4) 修了 slash command 被 lock 挡住的问题(提到 lock 之前)。 vitest run 45/45 通过;tsc rc=0。 --- hub/src/agent/runner.ts | 39 +++++++++++++-- hub/src/agent/transcript.ts | 64 ++++++++++++++++++++++++ hub/src/feishu/trigger.ts | 52 +++++++++++++++++-- hub/test/integration/trigger.test.ts | 75 ++++++++++++++++++++++++++++ hub/test/unit/transcript.test.ts | 60 ++++++++++++++++++++++ 5 files changed, 283 insertions(+), 7 deletions(-) create mode 100644 hub/src/agent/transcript.ts create mode 100644 hub/test/unit/transcript.test.ts diff --git a/hub/src/agent/runner.ts b/hub/src/agent/runner.ts index 7d3965d..5409873 100644 --- a/hub/src/agent/runner.ts +++ b/hub/src/agent/runner.ts @@ -11,6 +11,7 @@ */ import type { AgentProvider, ChatResponse, Message, ToolCallPart } from "./provider.js"; import type { ToolContext, ToolRegistry } from "./tools.js"; +import { readTranscript, appendTranscript } from "./transcript.js"; /** What the runner needs about the project to seed and bound a run. */ export interface ProjectContext { @@ -29,6 +30,9 @@ export interface RunRequest { readonly systemPrompt?: string; /** Iteration cap before the run is returned as `length`. Default 25. */ readonly maxIterations?: number; + /** 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; } export type RunStatus = "completed" | "length" | "failed"; @@ -59,9 +63,18 @@ export async function runAgent( boundChatId: req.project.boundChatId, workspaceDir: req.project.workspaceDir, }; - const messages: Message[] = []; - if (req.systemPrompt !== undefined) { + // Load prior session messages from transcript (continuity across @bot calls + // in the same session). ADR-0003: session messages ≠ group chat history. + let loadedCount = 0; + if (req.transcriptPath !== undefined) { + const prior = await readTranscript(req.transcriptPath); + loadedCount = prior.length; + messages.push(...prior); + } + 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.push({ role: "system", parts: [{ type: "text", text: req.systemPrompt }] }); } messages.push({ role: "user", parts: [{ type: "text", text: req.prompt }] }); @@ -79,12 +92,14 @@ export async function runAgent( tools: tools.specs(), }); } catch (e) { - return { + const result: RunResult = { status: "failed", messages, usage: { inputTokens, outputTokens }, error: e instanceof Error ? e.message : String(e), }; + await appendTranscriptIfSet(req, messages, loadedCount); + return result; } messages.push(res.message); @@ -94,9 +109,11 @@ export async function runAgent( } if (res.finishReason === "stop") { + await appendTranscriptIfSet(req, messages, loadedCount); return { status: "completed", messages, usage: { inputTokens, outputTokens } }; } if (res.finishReason !== "tool_calls") { + await appendTranscriptIfSet(req, messages, loadedCount); return { status: "length", messages, usage: { inputTokens, outputTokens } }; } @@ -111,6 +128,7 @@ export async function runAgent( } } + await appendTranscriptIfSet(req, messages, loadedCount); return { status: "length", messages, @@ -119,6 +137,21 @@ export async function runAgent( }; } +/// 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 Message[], loadedCount: number): Promise { + 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. + } +} + function toolCallsOfAssistant(msg: Message): readonly ToolCallPart[] { if (msg.role !== "assistant") return []; return msg.parts.filter((p): p is ToolCallPart => p.type === "tool_call"); diff --git a/hub/src/agent/transcript.ts b/hub/src/agent/transcript.ts new file mode 100644 index 0000000..ef7b646 --- /dev/null +++ b/hub/src/agent/transcript.ts @@ -0,0 +1,64 @@ +/** + * Session transcript — append-only JSONL file in the workspace. + * + * Each line is one {@link Message}, JSON-encoded. The transcript lives at + * `workspaceDir/.cph/sessions/.jsonl`, following the same "workspace + * is a directory tree" principle as the engineering file itself (ADR-0007). + * + * This mirrors how Claude Code stores its transcript (a `.jsonl` file per + * session). The agent can read its own history via `read_file` — no special + * "load history" code path. ADR-0002's lock guarantees one run per project at a + * time, so the transcript is never concurrently written. + * + * ADR-0003: this stores **agent session messages** (prompt + response + tool + * calls produced by the Hub), NOT Feishu group chat history. The two are + * distinct; storing session messages does not conflict with "the Hub avoids + * becoming a full chat history store." + */ +import { readFile, writeFile, mkdir, appendFile } from "node:fs/promises"; +import { join, dirname } from "node:path"; +import type { Message } from "./provider.js"; + +/// Subdirectory within a workspace where session transcripts live. +export const TRANSCRIPT_DIR = ".cph/sessions"; + +/** Path for a session's transcript file within a workspace. */ +export function transcriptPath(workspaceDir: string, sessionId: string): string { + return join(workspaceDir, TRANSCRIPT_DIR, `${sessionId}.jsonl`); +} + +/** + * Read prior messages from a transcript file. Returns an empty array if the + * file doesn't exist yet (first run in this session). Corrupt lines are + * skipped — a partially-written last line (crash mid-append) won't break the + * next run. + */ +export async function readTranscript(path: string): Promise { + let content: string; + try { + content = await readFile(path, "utf8"); + } catch { + return []; + } + const messages: Message[] = []; + for (const line of content.split("\n")) { + const trimmed = line.trim(); + if (trimmed === "") continue; + try { + messages.push(JSON.parse(trimmed) as Message); + } catch { + // Skip corrupt line — likely a partial write from a crash. + } + } + return messages; +} + +/** + * Append messages to a transcript file. Creates the parent directory if needed. + * Each message is written as one JSON line. + */ +export async function appendTranscript(path: string, messages: readonly Message[]): Promise { + await mkdir(dirname(path), { recursive: true }); + const lines = messages.map((m) => JSON.stringify(m)).join("\n") + "\n"; + await appendFile(path, lines, "utf8"); +} diff --git a/hub/src/feishu/trigger.ts b/hub/src/feishu/trigger.ts index b2c8843..cd4cd2c 100644 --- a/hub/src/feishu/trigger.ts +++ b/hub/src/feishu/trigger.ts @@ -22,6 +22,7 @@ import type { ModelRegistry } from "../agent/models.js"; import { runAgent, type ProjectContext } from "../agent/runner.js"; import { acquireLock, currentLockRunId, releaseLock } from "../lock.js"; import { canTriggerAgent } from "../permission.js"; +import { transcriptPath } from "../agent/transcript.js"; interface TriggerDeps { readonly prisma: PrismaClient; @@ -69,6 +70,52 @@ export function makeTriggerHandler(deps: TriggerDeps) { } } + const prompt = extractPrompt(msg); + if (prompt === null) return; + + // Slash commands: session management, not agent runs. These bypass the + // lock — they don't create a run, so they can't conflict with one. + if (prompt.startsWith("/")) { + const cmd = prompt.split(/\s+/)[0]; + switch (cmd) { + case "/new": { + await deps.prisma.agentSession.updateMany({ + where: { projectId, archivedAt: null }, + data: { archivedAt: new Date() }, + }); + await sendText(rt, chatId, "已开新会话,下次 @bot 将从头开始。"); + return; + } + case "/resume": { + const latest = await deps.prisma.agentSession.findFirst({ + where: { projectId, archivedAt: { not: null } }, + orderBy: { archivedAt: "desc" }, + select: { id: true }, + }); + if (latest === null) { + await sendText(rt, chatId, "没有可恢复的会话。"); + return; + } + await deps.prisma.agentSession.update({ + where: { id: latest.id }, + data: { archivedAt: null }, + }); + await sendText(rt, chatId, "已恢复上一个会话。"); + return; + } + case "/reset": { + await deps.prisma.agentSession.updateMany({ + where: { projectId, archivedAt: null }, + data: { archivedAt: new Date() }, + }); + await sendText(rt, chatId, "已重置,下次 @bot 将从头开始。"); + return; + } + default: + break; + } + } + // ADR-0002: if locked by another run, reply busy. const existing = await currentLockRunId(deps.prisma, projectId); if (existing !== null) { @@ -76,9 +123,6 @@ export function makeTriggerHandler(deps: TriggerDeps) { return; } - const prompt = extractPrompt(msg); - if (prompt === null) return; - const project = await deps.prisma.project.findUnique({ where: { id: projectId }, select: { workspaceDir: true }, @@ -155,7 +199,7 @@ export function makeTriggerHandler(deps: TriggerDeps) { }; // Run the agent — fire-and-forget from the ws handler's perspective. - runAgent(deps.provider, deps.tools, { prompt, model, project: projectCtx }) + runAgent(deps.provider, deps.tools, { prompt, model, project: projectCtx, transcriptPath: transcriptPath(project.workspaceDir, session.id) }) .then(async (result) => { await deps.prisma.agentRun.update({ where: { id: run.id }, diff --git a/hub/test/integration/trigger.test.ts b/hub/test/integration/trigger.test.ts index 6cacd36..bf8aefe 100644 --- a/hub/test/integration/trigger.test.ts +++ b/hub/test/integration/trigger.test.ts @@ -123,6 +123,81 @@ describe("trigger full lifecycle (integration)", () => { const runs = await prisma.agentRun.findMany(); expect(runs).toHaveLength(0); }); + + it("/new archives current session (no run created)", async () => { + await seedProject("proj-6", "chat-6"); + const trigger = makeTriggerHandler({ prisma, provider, tools, models, logger: silentLogger }); + + // First @bot creates a session + run. + await trigger(makeEvent("chat-6", "@_user_1 写教案"), rt); + await vi.waitFor(async () => { + const runs = await prisma.agentRun.findMany(); + expect(runs).toHaveLength(1); + expect(runs[0]?.status).toBe("COMPLETED"); + }); + + // /new archives the session. + await trigger(makeEvent("chat-6", "@_user_1 /new"), rt); + expect(rt.sentTexts).toContain("已开新会话,下次 @bot 将从头开始。"); + + const sessions = await prisma.agentSession.findMany(); + expect(sessions).toHaveLength(1); + expect(sessions[0]?.archivedAt).not.toBeNull(); + // No new run created for /new. + expect(await prisma.agentRun.findMany()).toHaveLength(1); + }); + + it("/resume un-archives the most recent session", async () => { + await seedProject("proj-7", "chat-7"); + const trigger = makeTriggerHandler({ prisma, provider, tools, models, logger: silentLogger }); + + // Create + archive a session via /new. + await trigger(makeEvent("chat-7", "@_user_1 写教案"), rt); + await vi.waitFor(async () => { + expect(await prisma.agentRun.findMany()).toHaveLength(1); + }); + await trigger(makeEvent("chat-7", "@_user_1 /new"), rt); + + // /resume un-archives. + await trigger(makeEvent("chat-7", "@_user_1 /resume"), rt); + expect(rt.sentTexts).toContain("已恢复上一个会话。"); + + const sessions = await prisma.agentSession.findMany(); + expect(sessions).toHaveLength(1); + expect(sessions[0]?.archivedAt).toBeNull(); + }); + + it("/reset archives current session", async () => { + await seedProject("proj-8", "chat-8"); + const trigger = makeTriggerHandler({ prisma, provider, tools, models, logger: silentLogger }); + + await trigger(makeEvent("chat-8", "@_user_1 写教案"), rt); + await vi.waitFor(async () => { + const runs = await prisma.agentRun.findMany(); + expect(runs).toHaveLength(1); + expect(runs[0]?.status).toBe("COMPLETED"); + }); + + await trigger(makeEvent("chat-8", "@_user_1 /reset"), rt); + expect(rt.sentTexts).toContain("已重置,下次 @bot 将从头开始。"); + + const sessions = await prisma.agentSession.findMany(); + expect(sessions).toHaveLength(1); + expect(sessions[0]?.archivedAt).not.toBeNull(); + }); + + it("unknown slash command falls through to agent", async () => { + await seedProject("proj-9", "chat-9"); + const trigger = makeTriggerHandler({ prisma, provider, tools, models, logger: silentLogger }); + + await trigger(makeEvent("chat-9", "@_user_1 /unknown"), rt); + // Should create a run (falls through as a normal prompt). + await vi.waitFor(async () => { + const runs = await prisma.agentRun.findMany(); + expect(runs).toHaveLength(1); + expect(runs[0]?.prompt).toBe("/unknown"); + }); + }); }); afterAll(async () => { diff --git a/hub/test/unit/transcript.test.ts b/hub/test/unit/transcript.test.ts new file mode 100644 index 0000000..1490d68 --- /dev/null +++ b/hub/test/unit/transcript.test.ts @@ -0,0 +1,60 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { mkdtemp, rm, readFile, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { readTranscript, appendTranscript, transcriptPath, TRANSCRIPT_DIR } from "../../src/agent/transcript.js"; +import type { Message } from "../../src/agent/provider.js"; + +const sample: Message = { role: "user", parts: [{ type: "text", text: "hello" }] }; +const reply: Message = { role: "assistant", parts: [{ type: "text", text: "hi" }] }; + +describe("transcript JSONL", () => { + let dir: string; + let path: string; + + beforeEach(async () => { + dir = await mkdtemp(join(tmpdir(), "hub-transcript-")); + path = join(dir, "session.jsonl"); + }); + afterEach(async () => { + await rm(dir, { recursive: true, force: true }); + }); + + it("readTranscript returns empty array for nonexistent file", async () => { + expect(await readTranscript(path)).toEqual([]); + }); + + it("appendTranscript creates parent dirs and writes JSONL", async () => { + await appendTranscript(path, [sample]); + const content = await readFile(path, "utf8"); + expect(content.trim()).toBe(JSON.stringify(sample)); + }); + + it("readTranscript reads back appended messages", async () => { + await appendTranscript(path, [sample, reply]); + const messages = await readTranscript(path); + expect(messages).toHaveLength(2); + expect(messages[0]).toEqual(sample); + expect(messages[1]).toEqual(reply); + }); + + it("multiple appends accumulate", async () => { + await appendTranscript(path, [sample]); + await appendTranscript(path, [reply]); + const messages = await readTranscript(path); + expect(messages).toHaveLength(2); + }); + + it("skips corrupt lines (partial write recovery)", async () => { + await appendTranscript(path, [sample]); + // Append a corrupt line (simulating a crash mid-write). + await writeFile(path, '{"broken json\n', { flag: "a" }); + const messages = await readTranscript(path); + expect(messages).toHaveLength(1); // only the valid line + }); + + it("transcriptPath builds the expected path", () => { + const p = transcriptPath("/ws", "sess-1"); + expect(p).toBe(join("/ws", TRANSCRIPT_DIR, "sess-1.jsonl")); + }); +});