feat(hub): session transcript 连续性 + slash commands (/new /resume /reset)

修复每次 @bot 都是失忆状态的问题。
- transcript.ts: JSONL 文件存 session 对话消息(workspace/.cph/sessions/
  <id>.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。
This commit is contained in:
2026-07-07 17:50:05 +08:00
parent e60aa43731
commit 0fe6cabc68
5 changed files with 283 additions and 7 deletions
+36 -3
View File
@@ -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<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.
}
}
function toolCallsOfAssistant(msg: Message): readonly ToolCallPart[] {
if (msg.role !== "assistant") return [];
return msg.parts.filter((p): p is ToolCallPart => p.type === "tool_call");
+64
View File
@@ -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/<sessionId>.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<Message[]> {
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<void> {
await mkdir(dirname(path), { recursive: true });
const lines = messages.map((m) => JSON.stringify(m)).join("\n") + "\n";
await appendFile(path, lines, "utf8");
}
+48 -4
View File
@@ -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 },
+75
View File
@@ -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 () => {
+60
View File
@@ -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"));
});
});