From 1dbed5be7c8d5940bee6dbbe1543ab9c0e663aed Mon Sep 17 00:00:00 2001 From: zh Date: Mon, 28 Sep 2026 16:03:12 +0800 Subject: [PATCH] =?UTF-8?q?feat(storage):=20=E4=BC=9A=E8=AF=9D=E4=B8=8E?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E8=90=BD=E5=BA=93=20SQLite=EF=BC=88A3.7?= =?UTF-8?q?=EF=BC=8C=E5=86=B3=E7=AD=96=20D16=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 server/core/storage：openDatabase + 迁移（projects / sessions / messages / idempotency_keys 四张平台表，按 ADR-0007 用整数毫秒与应用侧主键），SessionRepository 负责读写。SessionStore 改为内存态 + 写透：会话、消息、幂等键落库，进程重启后按库补建运行时状态（事件总线与进行中的运行仍在内存，符合架构设计 5.1）。startServer 支持 databasePath（测试传 :memory: 隔离，默认取配置 ~/.aps/aps.sqlite），关停时关闭数据库。新增 storage 用例：同一库文件起两次服务，会话、消息、幂等键都可查且重试命中同一条受理记录。测试 54 → 56 全绿。过程中修掉一个自身缺陷：迁移表要先建，否则第一次查版本号就报 no such table。 --- scripts/core-invoke.test.mjs | 2 +- scripts/internal-tool-call.test.mjs | 2 +- scripts/m1-loop.test.mjs | 2 +- scripts/serve-client.test.mjs | 2 +- scripts/server-smoke.test.mjs | 2 +- scripts/storage.test.mjs | 75 ++++++++++++++++++ server/app/service.ts | 14 ++-- server/app/sessions.ts | 58 ++++++++++++-- server/core/storage/db.ts | 79 +++++++++++++++++++ server/core/storage/sessions.ts | 116 ++++++++++++++++++++++++++++ server/index.ts | 7 ++ 11 files changed, 344 insertions(+), 15 deletions(-) create mode 100644 scripts/storage.test.mjs create mode 100644 server/core/storage/db.ts create mode 100644 server/core/storage/sessions.ts diff --git a/scripts/core-invoke.test.mjs b/scripts/core-invoke.test.mjs index daa2a91..b474776 100644 --- a/scripts/core-invoke.test.mjs +++ b/scripts/core-invoke.test.mjs @@ -17,7 +17,7 @@ async function startPlatform(devProvider = true) { if (devProvider) process.env.APS_DEV_PROVIDER = '1'; else delete process.env.APS_DEV_PROVIDER; const { startServer } = await import('../server/dist/index.js'); - const server = await startServer({ host: '127.0.0.1', port: 0, version: 'test' }); + const server = await startServer({ host: '127.0.0.1', port: 0, version: 'test', databasePath: ':memory:' }); return server; } diff --git a/scripts/internal-tool-call.test.mjs b/scripts/internal-tool-call.test.mjs index 5c09e83..cb77942 100644 --- a/scripts/internal-tool-call.test.mjs +++ b/scripts/internal-tool-call.test.mjs @@ -10,7 +10,7 @@ import { test } from 'node:test'; async function startWithProvider() { process.env.APS_DEV_PROVIDER = '1'; const { startServer } = await import('../server/dist/index.js'); - const server = await startServer({ host: '127.0.0.1', port: 0, version: 'test' }); + const server = await startServer({ host: '127.0.0.1', port: 0, version: 'test', databasePath: ':memory:' }); const created = await fetch(`${server.url}/v1/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, diff --git a/scripts/m1-loop.test.mjs b/scripts/m1-loop.test.mjs index 6f15c9b..bf718ec 100644 --- a/scripts/m1-loop.test.mjs +++ b/scripts/m1-loop.test.mjs @@ -23,7 +23,7 @@ test('假引擎发起工具调用，经 Core.invoke 到假 Provider 并把回执 ], }); - const server = await startServer({ host: '127.0.0.1', port: 0, version: 'test', runtime }); + const server = await startServer({ host: '127.0.0.1', port: 0, version: 'test', databasePath: ':memory:', runtime }); try { const created = await fetch(`${server.url}/v1/sessions`, { method: 'POST', diff --git a/scripts/serve-client.test.mjs b/scripts/serve-client.test.mjs index a487058..2237dbb 100644 --- a/scripts/serve-client.test.mjs +++ b/scripts/serve-client.test.mjs @@ -16,7 +16,7 @@ test('平台服务托管客户端产物，且 API 不受影响', async () => { execFileSync('pnpm', ['--filter', '@aps/client-app', 'run', 'build'], { cwd: ROOT, stdio: 'pipe', shell: isWindows }); const { startServer } = await import('../server/dist/index.js'); - const server = await startServer({ host: '127.0.0.1', port: 0, version: 'test' }); + const server = await startServer({ host: '127.0.0.1', port: 0, version: 'test', databasePath: ':memory:' }); try { const page = await fetch(`${server.url}/`); assert.equal(page.status, 200, 'GET / 应返回客户端页面'); diff --git a/scripts/server-smoke.test.mjs b/scripts/server-smoke.test.mjs index 0ca7377..3b3eed9 100644 --- a/scripts/server-smoke.test.mjs +++ b/scripts/server-smoke.test.mjs @@ -31,7 +31,7 @@ async function validator() { async function startServer(options = {}) { const { startServer: start } = await import('../server/dist/index.js'); - return start({ host: '127.0.0.1', port: 0, version: 'test', ...options }); + return start({ host: '127.0.0.1', port: 0, version: 'test', databasePath: ':memory:', ...options }); } const json = (method, body) => ({ method, headers: { 'content-type': 'application/json' }, body: JSON.stringify(body) }); diff --git a/scripts/storage.test.mjs b/scripts/storage.test.mjs new file mode 100644 index 0000000..5b763d2 --- /dev/null +++ b/scripts/storage.test.mjs @@ -0,0 +1,75 @@ +/** + * 落库用例（任务 A3.7）：会话、消息与幂等键要在进程重启后仍然可查。 + * + * 用同一个 SQLite 文件起两次服务，模拟「平台重启」。 + */ +import assert from 'node:assert/strict'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; + +async function startWithDatabase(dbPath) { + const { startServer } = await import('../server/dist/index.js'); + return startServer({ host: '127.0.0.1', port: 0, version: 'test', databasePath: dbPath }); +} + +const postJson = (base, path, body) => + fetch(`${base}${path}`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(body) }); + +test('会话、消息与幂等键在重启后可查', async () => { + const dir = await mkdtemp(join(tmpdir(), 'aps-storage-')); + const dbPath = join(dir, 'aps.sqlite'); + try { + let sessionId = ''; + let firstRunId = ''; + { + const server = await startWithDatabase(dbPath); + try { + const created = await (await postJson(server.url, '/v1/sessions', { scope: 'project', projectId: 'p-default', title: '持久化用例' })).json(); + sessionId = created.session.sessionId; + + const accepted = await ( + await postJson(server.url, `/v1/sessions/${sessionId}/messages`, { text: '试排一版看看', idempotencyKey: 'idem-storage-0001' }) + ).json(); + firstRunId = accepted.runId; + + await new Promise((resolve) => setTimeout(resolve, 300)); + } finally { + await server.close(); + } + } + + // 重启：同一个库文件，新的服务实例 + { + const server = await startWithDatabase(dbPath); + try { + const sessions = await (await fetch(`${server.url}/v1/sessions`)).json(); + assert.equal(sessions.sessions.length, 1, '重启后应还能列出会话'); + assert.equal(sessions.sessions[0].sessionId, sessionId); + assert.equal(sessions.sessions[0].title, '持久化用例'); + + const detail = await (await fetch(`${server.url}/v1/sessions/${sessionId}`)).json(); + assert.ok(detail.messages.length >= 2, `重启后应还能读到消息，实际 ${detail.messages.length} 条`); + assert.equal(detail.messages[0].role, 'user'); + assert.equal(detail.messages.at(-1).role, 'assistant'); + + // 幂等键也持久化了：同一 key 重试仍返回同一个 run + const retry = await ( + await postJson(server.url, `/v1/sessions/${sessionId}/messages`, { text: '试排一版看看', idempotencyKey: 'idem-storage-0001' }) + ).json(); + assert.equal(retry.runId, firstRunId, '重启后重试应命中同一条受理记录'); + } finally { + await server.close(); + } + } + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test('默认数据库路径来自配置（~/.aps/aps.sqlite）', async () => { + const { loadConfig } = await import('@aps/server/config'); + const config = loadConfig({ env: {}, homeDir: 'C:/Users/example' }); + assert.equal(config.database.path.replace(/\\/g, '/'), 'C:/Users/example/.aps/aps.sqlite'); +}); \ No newline at end of file diff --git a/server/app/service.ts b/server/app/service.ts index e56110a..2a2333b 100644 --- a/server/app/service.ts +++ b/server/app/service.ts @@ -12,6 +12,7 @@ import { createFakeRuntime } from '../agent/fake-runtime.js'; import type { AgentRuntime, RuntimeSession } from '../agent/runtime.js'; import { createRulePlanner, plannerToFakeTurn } from '../agent/rule-planner.js'; import { toToolDescriptors } from '../agent/tools.js'; +import type { SessionRepository } from '../core/storage/sessions.js'; import { runTurn } from './turn-runner.js'; import type { Provider } from '@aps/provider-sdk'; import { badRequest, conflict } from './errors.js'; @@ -23,11 +24,13 @@ export interface ApplicationServiceOptions { version?: string; /** 引擎实现：默认是假引擎（A3.3 起可换成 Pi 子进程） */ runtime?: AgentRuntime; + /** 会话持久化（A3.7）：不传则纯内存，测试与临时运行用 */ + sessionRepository?: SessionRepository; } export class ApplicationService { private readonly projects = new ProjectStore(); - private readonly sessions = new SessionStore(); + private readonly sessions: SessionStore; private readonly version: string; /** 能力目录与执行入口：能力清单来自这里，执行只有一个出口 */ readonly registry = new CapabilityRegistry(); @@ -38,6 +41,7 @@ export class ApplicationService { constructor(options: ApplicationServiceOptions = {}) { this.version = options.version ?? '0.1.0'; + this.sessions = new SessionStore(options.sessionRepository); this.core = createCore({ registry: this.registry }); this.runtime = options.runtime ?? createFakeRuntime({ planner: plannerToFakeTurn(createRulePlanner()) }); } @@ -124,15 +128,15 @@ export class ApplicationService { throw badRequest('idempotencyKey 至少 8 个字符'); } // 幂等重试优先于「会话忙」判定：同一 key 的重试必须返回原结果，不能报冲突 - const cached = record.accepted.get(request.idempotencyKey); + const cached = this.sessions.findIdempotencyKey(sessionId, request.idempotencyKey); if (cached) return { accepted: true, messageId: cached.messageId, runId: cached.runId }; if (record.state !== 'idle') throw conflict('上一个运行还没结束，先取消或等它完成'); const messageId = newId('m'); const runId = newId('r'); - record.accepted.set(request.idempotencyKey, { messageId, runId }); - record.messages.push({ messageId, role: 'user', ts: Date.now(), text: request.text }); + this.sessions.rememberIdempotencyKey(sessionId, request.idempotencyKey, messageId, runId); + this.sessions.appendMessage(sessionId, { messageId, role: 'user', ts: Date.now(), text: request.text }); record.bus.publish({ sessionId, runId, type: 'message.user', data: { messageId, text: request.text } }); const abort = new AbortController(); @@ -207,7 +211,7 @@ export class ApplicationService { this.invokeFromSession(sessionId, { capabilityId, params: args, idempotencyKey }), }); if (result.text !== '') { - record.messages.push({ messageId: assistantMessageId, role: 'assistant', ts: Date.now(), text: result.text }); + this.sessions.appendMessage(sessionId, { messageId: assistantMessageId, role: 'assistant', ts: Date.now(), text: result.text }); } } catch (error) { record.bus.publish({ diff --git a/server/app/sessions.ts b/server/app/sessions.ts index 23b8606..f45be04 100644 --- a/server/app/sessions.ts +++ b/server/app/sessions.ts @@ -1,14 +1,15 @@ /** - * 会话存储（A1 阶段内存实现）。 + * 会话存储（A3.7：内存态 + SQLite 持久化）。 * - * ADR-0003：个人会话只读，写操作必须在项目会话内——这里在创建时就把住： - * scope=project 必须带 projectId，scope=personal 不允许带。 - * A3.7 落库后换成持久层；主键按 ADR-0007 用应用侧生成。 + * - 持久部分（会话、消息、幂等键）写进 SQLite：重启后仍可查（ADR-0007） + * - 运行态部分（事件总线、进行中的运行）留在内存：进程重启后由平台历史恢复（架构设计 5.1） + * ADR-0003：个人会话只读，写操作必须在项目会话内——创建时就把住。 */ import { randomUUID } from 'node:crypto'; import type { ChatMessage, SessionSummary } from '@aps/contracts'; +import type { SessionRepository } from '../core/storage/sessions.js'; import { badRequest, notFound } from './errors.js'; import type { Subject } from './subject.js'; import { EventBus } from './events.js'; @@ -45,6 +46,8 @@ export function newId(prefix: string): string { export class SessionStore { private readonly sessions = new Map(); + constructor(private readonly repository?: SessionRepository) {} + create(input: CreateSessionInput, owner: Subject): SessionRecord { if (input.scope === 'project' && !input.projectId) throw badRequest('项目会话必须带 projectId（ADR-0003）'); if (input.scope === 'personal' && input.projectId) throw badRequest('个人会话不能带 projectId'); @@ -67,11 +70,53 @@ export class SessionStore { accepted: new Map(), }; this.sessions.set(summary.sessionId, record); + this.repository?.insertSession({ ...summary, owner }); return record; } get(sessionId: string): SessionRecord | undefined { - return this.sessions.get(sessionId); + const cached = this.sessions.get(sessionId); + if (cached) return cached; + // 进程重启后按库里的记录“补建”运行态：历史与归属都从存储恢复 + const stored = this.repository?.getSession(sessionId); + if (!stored) return undefined; + const { owner, ...summary } = stored; + const record: SessionRecord = { + summary, + owner, + messages: this.repository?.listMessages(sessionId) ?? [], + bus: new EventBus(), + state: 'idle', + runs: new Map(), + accepted: new Map(), + }; + this.sessions.set(sessionId, record); + return record; + } + + /** 追加消息：内存与库一起写 */ + appendMessage(sessionId: string, message: ChatMessage): void { + const record = this.require(sessionId); + record.messages.push(message); + this.repository?.appendMessage(sessionId, message); + } + + /** 幂等键：受理时记录，重试与重启后都能命中 */ + rememberIdempotencyKey(sessionId: string, key: string, messageId: string, runId: string): void { + this.require(sessionId).accepted.set(key, { messageId, runId }); + this.repository?.rememberIdempotencyKey(key, sessionId, messageId, runId); + } + + findIdempotencyKey(sessionId: string, key: string): { messageId: string; runId: string } | undefined { + const cached = this.require(sessionId).accepted.get(key); + if (cached) return cached; + const stored = this.repository?.findByIdempotencyKey(key); + if (stored && stored.sessionId === sessionId) { + const value = { messageId: stored.messageId, runId: stored.runId }; + this.require(sessionId).accepted.set(key, value); + return value; + } + return undefined; } require(sessionId: string): SessionRecord { @@ -81,6 +126,9 @@ export class SessionStore { } list(projectId?: string): SessionSummary[] { + if (this.repository) { + return this.repository.listSessions(projectId).map(({ owner: _owner, ...summary }) => summary); + } return [...this.sessions.values()] .filter((record) => (projectId ? record.summary.projectId === projectId : true)) .map((record) => record.summary) diff --git a/server/core/storage/db.ts b/server/core/storage/db.ts new file mode 100644 index 0000000..0d633af --- /dev/null +++ b/server/core/storage/db.ts @@ -0,0 +1,79 @@ +/** + * SQLite 存储（任务 A3.7，依据 ADR-0007 与决策 D16）。 + * + * - 驱动用 Node 内置 node:sqlite：零原生依赖，不需要为 Electron ABI 重编译 + * - 时间统一存整数毫秒，主键由应用侧生成（ADR-0007 两条约定） + * - 表按 ADR-0008 的「平台库」建；审计、待确认动作、桌面授权三组等各自功能落地时再加 + */ +import { mkdirSync } from 'node:fs'; +import { dirname } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; + +export type Database = DatabaseSync; + +const MIGRATIONS: Array<{ version: number; sql: string }> = [ + { + version: 1, + sql: ` + create table if not exists projects ( + project_id text primary key, + name text not null, + status text not null, + writable integer not null default 1, + created_at integer not null + ); + create table if not exists sessions ( + session_id text primary key, + project_id text, + scope text not null, + status text not null, + title text not null, + owner_user_id text not null, + owner_tenant_id text not null, + owner_display_name text not null, + owner_client_kind text not null, + created_at integer not null + ); + create index if not exists idx_sessions_project on sessions(project_id, created_at); + create table if not exists messages ( + message_id text primary key, + session_id text not null, + role text not null, + text text not null, + ts integer not null + ); + create index if not exists idx_messages_session on messages(session_id, ts); + create table if not exists idempotency_keys ( + idempotency_key text primary key, + session_id text not null, + message_id text not null, + run_id text not null, + created_at integer not null + ); + `, + }, +]; + +export function openDatabase(path: string): Database { + if (path !== ':memory:') mkdirSync(dirname(path), { recursive: true }); + const db = new DatabaseSync(path); + try { + db.exec('pragma journal_mode = WAL'); + } catch { + /* 内存库不支持 WAL */ + } + db.exec('pragma foreign_keys = ON'); + migrate(db); + return db; +} + +export function migrate(db: Database): void { + // 先建迁移表本身，否则第一次检查版本号就会报 no such table + db.exec('create table if not exists schema_migrations (version integer primary key, applied_at integer not null)'); + for (const migration of MIGRATIONS) { + const applied = db.prepare('select version from schema_migrations where version = ?').get(migration.version); + if (applied) continue; + db.exec(migration.sql); + db.prepare('insert into schema_migrations (version, applied_at) values (?, ?)').run(migration.version, Date.now()); + } +} \ No newline at end of file diff --git a/server/core/storage/sessions.ts b/server/core/storage/sessions.ts new file mode 100644 index 0000000..665cba2 --- /dev/null +++ b/server/core/storage/sessions.ts @@ -0,0 +1,116 @@ +/** 会话与消息的读写（A3.7）。只碰平台库，不涉及业务数据（ADR-0008）。 */ +import type { ChatMessage, SessionSummary } from '@aps/contracts'; + +import type { Database } from './db.js'; + +export interface StoredSession extends SessionSummary { + owner: { userId: string; tenantId: string; displayName: string; clientKind: 'desktop' | 'browser' }; +} + +interface SessionRow { + session_id: string; + project_id: string | null; + scope: string; + status: string; + title: string; + owner_user_id: string; + owner_tenant_id: string; + owner_display_name: string; + owner_client_kind: string; + created_at: number; +} + +interface MessageRow { + message_id: string; + session_id: string; + role: string; + text: string; + ts: number; +} + +const toSession = (row: SessionRow): StoredSession => ({ + sessionId: row.session_id, + projectId: row.project_id, + scope: row.scope as SessionSummary['scope'], + status: row.status as SessionSummary['status'], + title: row.title, + createdAt: row.created_at, + owner: { + userId: row.owner_user_id, + tenantId: row.owner_tenant_id, + displayName: row.owner_display_name, + clientKind: row.owner_client_kind as 'desktop' | 'browser', + }, +}); + +const toMessage = (row: MessageRow): ChatMessage => ({ + messageId: row.message_id, + role: row.role as ChatMessage['role'], + ts: row.ts, + text: row.text, +}); + +export class SessionRepository { + constructor(private readonly db: Database) {} + + insertSession(session: StoredSession): void { + this.db + .prepare( + 'insert into sessions (session_id, project_id, scope, status, title, owner_user_id, owner_tenant_id, owner_display_name, owner_client_kind, created_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', + ) + .run( + session.sessionId, + session.projectId, + session.scope, + session.status, + session.title, + session.owner.userId, + session.owner.tenantId, + session.owner.displayName, + session.owner.clientKind, + session.createdAt, + ); + } + + getSession(sessionId: string): StoredSession | undefined { + const row = this.db.prepare('select * from sessions where session_id = ?').get(sessionId) as unknown as SessionRow | undefined; + return row ? toSession(row) : undefined; + } + + listSessions(projectId?: string): StoredSession[] { + const rows = ( + projectId === undefined + ? this.db.prepare('select * from sessions order by created_at desc').all() + : this.db.prepare('select * from sessions where project_id = ? order by created_at desc').all(projectId) + ) as unknown as SessionRow[]; + return rows.map(toSession); + } + + appendMessage(sessionId: string, message: ChatMessage): void { + this.db + .prepare('insert into messages (message_id, session_id, role, text, ts) values (?, ?, ?, ?, ?)') + .run(message.messageId, sessionId, message.role, message.text, message.ts); + } + + listMessages(sessionId: string): ChatMessage[] { + const rows = this.db + .prepare('select * from messages where session_id = ? order by ts asc, rowid asc') + .all(sessionId) as unknown as MessageRow[]; + return rows.map(toMessage); + } + + rememberIdempotencyKey(key: string, sessionId: string, messageId: string, runId: string): void { + this.db + .prepare( + 'insert or ignore into idempotency_keys (idempotency_key, session_id, message_id, run_id, created_at) values (?, ?, ?, ?, ?)', + ) + .run(key, sessionId, messageId, runId, Date.now()); + } + + findByIdempotencyKey(key: string): { sessionId: string; messageId: string; runId: string } | undefined { + const row = this.db.prepare('select session_id, message_id, run_id from idempotency_keys where idempotency_key = ?').get(key) as + | { session_id: string; message_id: string; run_id: string } + | undefined; + return row ? { sessionId: row.session_id, messageId: row.message_id, runId: row.run_id } : undefined; + } +} \ No newline at end of file diff --git a/server/index.ts b/server/index.ts index 0c9de0f..0c9bbde 100644 --- a/server/index.ts +++ b/server/index.ts @@ -9,6 +9,8 @@ import type { AddressInfo } from 'node:net'; import type { AgentRuntime } from './agent/runtime.js'; import { loadConfig } from './config.js'; +import { openDatabase, type Database } from './core/storage/db.js'; +import { SessionRepository } from './core/storage/sessions.js'; import { ApplicationService } from './app/service.js'; import { createHttpServer } from './entry/http.js'; @@ -22,6 +24,8 @@ export interface StartOptions { version?: string; /** 引擎实现：默认假引擎；A3.3 起可换成 Pi 子进程，测试也可注入脚本化假引擎 */ runtime?: AgentRuntime; + /** 数据库路径：默认取配置；测试传 ':memory:' 以隔离 */ + databasePath?: string; } export interface RunningServer { @@ -37,9 +41,11 @@ export async function startServer(options: StartOptions = {}): Promise