Merge pull request #560 from cloudyli/codex/pr3-session-compaction

fix: wire SQLite session history compaction

Author: phodal

docs/operational/sqlite-session-history-maintenance.zh-CN.md

@@ -0,0 +1,32 @@
+# SQLite Session History Maintenance
+
+Routa 本地 SQLite 长时间运行后，ACP 会话历史可能同时占用：
+
+- `session_messages`：按事件追加的会话消息表。
+- `acp_sessions.message_history`：兼容旧读取路径的 JSON 历史列。
+- `routa.db-wal`：SQLite WAL 文件。
+
+维护入口：
+
+```bash
+npm run db:sqlite:compact-sessions -- --db ./routa.db --dry-run --json
+npm run db:sqlite:compact-sessions -- --db ./routa.db --apply --checkpoint
+```
+
+如需在确认服务已停止后回收更多磁盘空间：
+
+```bash
+npm run db:sqlite:compact-sessions -- --db ./routa.db --apply --checkpoint --vacuum
+```
+
+行为边界：
+
+- 默认 `dry-run`，只报告候选 session、可合并 chunk 和可删除行数。
+- `--apply` 才会写入数据库。
+- 默认保护 60 分钟内更新过的 session，也保护 `lease_expires_at` 尚未过期的 session。
+- 可以用 `--active-session <id>` 显式保护正在运行的 session。
+- 只合并连续的 `agent_message_chunk` 为 `agent_message`。
+- 不删除 artifact、task completion summary、verification report 或业务对象。
+- compaction 不更新 `acp_sessions.updated_at`，避免维护操作污染会话活跃时间线。
+- `--checkpoint` 会执行 `PRAGMA wal_checkpoint(TRUNCATE)`。
+- `--vacuum` 会执行 `VACUUM`，建议只在 Routa 服务停止后使用，避免 Windows SQLite 文件锁问题。

package.json

@@ -93,6 +93,7 @@
     "db:sqlite:generate": "drizzle-kit generate --config=drizzle-sqlite.config.ts",
     "db:sqlite:migrate": "drizzle-kit migrate --config=drizzle-sqlite.config.ts",
     "db:sqlite:push": "drizzle-kit push --config=drizzle-sqlite.config.ts",
+    "db:sqlite:compact-sessions": "node --import tsx scripts/maintenance/compact-session-history.ts",
     "api:check": "node --import tsx scripts/fitness/check-api-parity.ts",
     "api:check:json": "node --import tsx scripts/fitness/check-api-parity.ts --json",
     "api:check:fix": "node --import tsx scripts/fitness/check-api-parity.ts --fix-hint",

scripts/__tests__/compact-session-history.test.ts

@@ -0,0 +1,144 @@
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import BetterSqlite3 from "better-sqlite3";
+import { afterEach, describe, expect, it } from "vitest";
+import { runSessionHistoryMaintenance } from "../maintenance/compact-session-history";
+
+const tempDirs: string[] = [];
+type CountRow = { count: number };
+type PayloadRow = { payload: string };
+type SessionHistoryRow = { message_history: string };
+
+function createFixtureDb(): string {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "routa-session-compact-"));
+  tempDirs.push(dir);
+  const dbPath = path.join(dir, "routa.db");
+  const sqlite = new BetterSqlite3(dbPath);
+  const old = Date.now() - 10 * 24 * 60 * 60 * 1000;
+  const active = Date.now();
+  sqlite.exec(`
+    CREATE TABLE acp_sessions (
+      id TEXT PRIMARY KEY,
+      message_history TEXT DEFAULT '[]',
+      updated_at INTEGER,
+      lease_expires_at INTEGER
+    );
+    CREATE TABLE session_messages (
+      id TEXT PRIMARY KEY,
+      session_id TEXT NOT NULL,
+      message_index INTEGER NOT NULL,
+      event_type TEXT NOT NULL,
+      payload TEXT NOT NULL,
+      created_at INTEGER
+    );
+  `);
+  sqlite.prepare("INSERT INTO acp_sessions (id, message_history, updated_at, lease_expires_at) VALUES (?, ?, ?, ?)").run(
+    "old-session",
+    JSON.stringify([
+      { sessionId: "old-session", update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "Hel" } } },
+      { sessionId: "old-session", update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "lo" } } },
+      { sessionId: "old-session", update: { sessionUpdate: "turn_complete" } },
+    ]),
+    old,
+    null,
+  );
+  sqlite.prepare("INSERT INTO acp_sessions (id, message_history, updated_at, lease_expires_at) VALUES (?, ?, ?, ?)").run(
+    "active-session",
+    JSON.stringify([
+      { sessionId: "active-session", update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "Act" } } },
+      { sessionId: "active-session", update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "ive" } } },
+    ]),
+    active,
+    Date.now() + 60_000,
+  );
+  sqlite.prepare("INSERT INTO acp_sessions (id, message_history, updated_at, lease_expires_at) VALUES (?, ?, ?, ?)").run(
+    "invalid-session",
+    JSON.stringify([]),
+    old,
+    null,
+  );
+  const insertMessage = sqlite.prepare(
+    "INSERT INTO session_messages (id, session_id, message_index, event_type, payload, created_at) VALUES (?, ?, ?, ?, ?, ?)",
+  );
+  insertMessage.run("m1", "old-session", 0, "agent_message_chunk", JSON.stringify({ type: "agent_message_chunk", text: "A" }), old);
+  insertMessage.run("m2", "old-session", 1, "agent_message_chunk", JSON.stringify({ type: "agent_message_chunk", text: "B" }), old);
+  insertMessage.run("m3", "active-session", 0, "agent_message_chunk", JSON.stringify({ sessionId: "active-session", update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "C" } } }), old);
+  insertMessage.run("m4", "active-session", 1, "agent_message_chunk", JSON.stringify({ sessionId: "active-session", update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "D" } } }), old);
+  insertMessage.run("m5", "invalid-session", 0, "agent_message_chunk", JSON.stringify({ type: "agent_message", text: "Keep" }), old);
+  insertMessage.run("m6", "invalid-session", 1, "agent_message_chunk", JSON.stringify({ type: "agent_message", text: "Me" }), old);
+  sqlite.close();
+  return dbPath;
+}
+
+afterEach(() => {
+  for (const dir of tempDirs.splice(0)) {
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+describe("compact-session-history maintenance script", () => {
+  it("reports dry-run changes without writing", () => {
+    const dbPath = createFixtureDb();
+    const summary = runSessionHistoryMaintenance({
+      dbPath,
+      mode: "dry-run",
+      sessionCutoffDays: 7,
+      activeWindowMinutes: 60,
+      activeSessionIds: new Set(),
+      checkpoint: true,
+      vacuum: true,
+      json: true,
+    });
+
+    expect(summary.sessionMessages.mergedGroups).toBe(1);
+    expect(summary.sessionMessages.deletedRows).toBe(1);
+    expect(summary.acpSessions.compactedSessions).toBe(1);
+    expect(summary.protectedActiveSessions).toEqual(["active-session"]);
+
+    const sqlite = new BetterSqlite3(dbPath);
+    expect(sqlite.prepare("SELECT COUNT(*) AS count FROM session_messages").get()).toEqual({ count: 6 });
+    sqlite.close();
+  });
+
+  it("applies compaction and skips active sessions", () => {
+    const dbPath = createFixtureDb();
+    const summary = runSessionHistoryMaintenance({
+      dbPath,
+      mode: "apply",
+      sessionCutoffDays: 7,
+      activeWindowMinutes: 60,
+      activeSessionIds: new Set(),
+      checkpoint: false,
+      vacuum: false,
+      json: true,
+    });
+
+    expect(summary.sessionMessages.deletedRows).toBe(1);
+    const sqlite = new BetterSqlite3(dbPath);
+    expect(sqlite.prepare("SELECT COUNT(*) AS count FROM session_messages").get() as CountRow).toEqual({ count: 5 });
+    const oldPayloadRow = sqlite.prepare("SELECT payload FROM session_messages WHERE id = ?").get("m1") as PayloadRow;
+    const oldPayload = JSON.parse(oldPayloadRow.payload);
+    expect(oldPayload.update.sessionUpdate).toBe("agent_message");
+    expect(oldPayload.update.content.text).toBe("AB");
+    expect(oldPayload.type).toBe("agent_message");
+    expect(oldPayload.text).toBe("AB");
+
+    const activePayloadRow = sqlite.prepare("SELECT payload FROM session_messages WHERE id = ?").get("m3") as PayloadRow;
+    const activePayload = JSON.parse(activePayloadRow.payload);
+    expect(activePayload.update.sessionUpdate).toBe("agent_message_chunk");
+
+    const invalidPayloadRow = sqlite.prepare("SELECT event_type, payload FROM session_messages WHERE id = ?").get("m6") as {
+      event_type: string;
+      payload: string;
+    };
+    expect(invalidPayloadRow.event_type).toBe("agent_message_chunk");
+    expect(JSON.parse(invalidPayloadRow.payload).text).toBe("Me");
+
+    const oldSession = sqlite.prepare("SELECT message_history FROM acp_sessions WHERE id = ?").get("old-session") as SessionHistoryRow;
+    expect(JSON.parse(oldSession.message_history)).toHaveLength(2);
+    const activeSession = sqlite.prepare("SELECT message_history FROM acp_sessions WHERE id = ?").get("active-session") as SessionHistoryRow;
+    expect(JSON.parse(activeSession.message_history)).toHaveLength(2);
+    sqlite.close();
+  });
+});