From 73601ebe26f129fae24e9c530bbc00fbe2fa1119 Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 10 Aug 2026 10:50:18 +0800 Subject: [PATCH] fix(topics): coordinate session lifecycle across runtimes --- .../src/routes/shared-routes.test.ts | 22 ++ .../src/routes/web/ai-chat-topics.test.ts | 1 + .../src/routes/web/ai-chat-topics.ts | 9 +- .../http-routes/src/routes/web/register-ai.ts | 2 +- .../http-routes/src/routes/web/sessions.ts | 16 +- packages/node-runtime/src/data-dir-compat.ts | 6 + .../node-runtime/src/database-manager.test.ts | 106 +++++-- packages/node-runtime/src/database-manager.ts | 14 +- packages/node-runtime/src/index.ts | 1 + .../node-runtime/src/services/topics/index.ts | 2 +- .../src/services/topics/service.test.ts | 194 +++++++++++- .../src/services/topics/service.ts | 236 ++++++++++++--- .../src/services/topics/store.test.ts | 70 ++++- .../node-runtime/src/services/topics/store.ts | 278 ++++++++++++++---- 14 files changed, 808 insertions(+), 149 deletions(-) diff --git a/packages/http-routes/src/routes/shared-routes.test.ts b/packages/http-routes/src/routes/shared-routes.test.ts index 89a807e7..4bddd2fa 100644 --- a/packages/http-routes/src/routes/shared-routes.test.ts +++ b/packages/http-routes/src/routes/shared-routes.test.ts @@ -673,6 +673,28 @@ describe('registerSharedRoutes smoke tests', () => { assert.deepEqual(lifecycle, ['close:chat-1', 'delete:chat-1']) }) + it('DELETE /_web/sessions/:id preserves a topic generation conflict without deleting the session', async () => { + const db = createSessionDb() + const dbs = new Map([['chat-1', db]]) + const routeApp = Fastify() + const ctx = createTestContext(dbs) + ctx.sessionAdapter.deleteSessionFile = () => { + throw Object.assign(new Error('Chat topics are being generated for this session in another runtime'), { + statusCode: 409, + }) + } + registerSessionRoutes(routeApp, ctx) + await routeApp.ready() + + const resp = await routeApp.inject({ method: 'DELETE', url: '/_web/sessions/chat-1' }) + + await routeApp.close() + db.close() + + assert.equal(resp.statusCode, 409) + assert.equal(dbs.has('chat-1'), true) + }) + it('owner profile routes select, apply and dismiss across sessions', async () => { const prefDir = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-owner-routes-')) const current = createSessionDb() diff --git a/packages/http-routes/src/routes/web/ai-chat-topics.test.ts b/packages/http-routes/src/routes/web/ai-chat-topics.test.ts index 12195e0e..054e6030 100644 --- a/packages/http-routes/src/routes/web/ai-chat-topics.test.ts +++ b/packages/http-routes/src/routes/web/ai-chat-topics.test.ts @@ -29,6 +29,7 @@ test('chat topic routes expose persisted run state and reject generation without registerAiChatTopicRoutes(app, { sessionAdapter: {} as SessionRuntimeAdapter, pathProvider, + runtimeIdentity: { version: '0.35.1', kind: 'cli' }, nativeBinding, }) await app.ready() diff --git a/packages/http-routes/src/routes/web/ai-chat-topics.ts b/packages/http-routes/src/routes/web/ai-chat-topics.ts index 5e353eba..71b36f3b 100644 --- a/packages/http-routes/src/routes/web/ai-chat-topics.ts +++ b/packages/http-routes/src/routes/web/ai-chat-topics.ts @@ -9,15 +9,20 @@ import type { CreateChatTopicsRequest } from '@openchatlab/shared-types' import type { AiRouteContext } from '../../context/ai' import type { RuntimeRouteContext } from '../../context/runtime' -type ChatTopicRouteContext = Pick & +type ChatTopicRouteContext = Pick< + RuntimeRouteContext, + 'sessionAdapter' | 'pathProvider' | 'runtimeIdentity' | 'nativeBinding' +> & Pick export function registerAiChatTopicRoutes(server: FastifyInstance, ctx: ChatTopicRouteContext): void { let service: ChatTopicService | null = null const getService = () => { + if (!ctx.runtimeIdentity) throw new Error('Chat topic routes require a runtime identity') service ??= createChatTopicService({ runtime: ctx.sessionAdapter, - userDataDir: ctx.pathProvider.getUserDataDir(), + pathProvider: ctx.pathProvider, + runtimeIdentity: ctx.runtimeIdentity, nativeBinding: ctx.nativeBinding, getModelClient() { const config = ctx.llmConfigStore?.getFastModelConfig() diff --git a/packages/http-routes/src/routes/web/register-ai.ts b/packages/http-routes/src/routes/web/register-ai.ts index 22097d7b..ab1ef9b1 100644 --- a/packages/http-routes/src/routes/web/register-ai.ts +++ b/packages/http-routes/src/routes/web/register-ai.ts @@ -15,7 +15,7 @@ import { registerAiSummaryRoutes } from './ai-summaries' import { registerAiToolRoutes } from './ai-tools' export type AiRoutesContext = AiRouteContext & - Pick & + Pick & Pick export interface AiRouteOptions { diff --git a/packages/http-routes/src/routes/web/sessions.ts b/packages/http-routes/src/routes/web/sessions.ts index 74909d92..10a0695c 100644 --- a/packages/http-routes/src/routes/web/sessions.ts +++ b/packages/http-routes/src/routes/web/sessions.ts @@ -49,17 +49,13 @@ export function registerSessionRoutes(server: FastifyInstance, ctx: SessionRoute server.delete<{ Params: { id: string } }>('/_web/sessions/:id', async (request, reply) => { const { id } = request.params - try { - await chatTopicWorkCoordinator.prepareSessionDelete(id) - await ctx.beforeDeleteSession?.(id) - const deleted = sessionService.deleteSession(adapter, id) - if (!deleted) { - return reply.code(404).send({ success: false, error: 'File not found' }) - } - return { success: true } - } catch (err) { - return reply.code(500).send({ success: false, error: String(err) }) + await chatTopicWorkCoordinator.prepareSessionDelete(id) + await ctx.beforeDeleteSession?.(id) + const deleted = sessionService.deleteSession(adapter, id) + if (!deleted) { + return reply.code(404).send({ success: false, error: 'File not found' }) } + return { success: true } }) server.patch<{ Params: { id: string }; Body: { name: string } }>('/_web/sessions/:id/name', async (request) => { diff --git a/packages/node-runtime/src/data-dir-compat.ts b/packages/node-runtime/src/data-dir-compat.ts index 6c79395e..809b12a6 100644 --- a/packages/node-runtime/src/data-dir-compat.ts +++ b/packages/node-runtime/src/data-dir-compat.ts @@ -131,6 +131,12 @@ export function assertDataDirCompatible( } } +export function isRuntimeVersionAtLeast(runtimeVersion: string, minimumVersion: string): boolean { + const normalizedRuntimeVersion = normalizeRuntimeStableVersion(runtimeVersion) + if (!normalizedRuntimeVersion || !isStableSemver(minimumVersion)) return false + return compareStableSemver(normalizedRuntimeVersion, minimumVersion) >= 0 +} + export function raiseDataDirMinRuntimeVersion( pathProvider: PathProvider, input: RaiseDataDirCompatibilityInput diff --git a/packages/node-runtime/src/database-manager.test.ts b/packages/node-runtime/src/database-manager.test.ts index 59fe12bb..15ab6d83 100644 --- a/packages/node-runtime/src/database-manager.test.ts +++ b/packages/node-runtime/src/database-manager.test.ts @@ -46,6 +46,36 @@ function assertAnalysisToolIndexes(db: Database.Database): void { assert.ok(indexes.includes('idx_message_type_ts')) } +function seedTopicRun(userDataDir: string, sessionId: string, runId = 'topic-run'): void { + const topicStore = new ChatTopicStore(getChatTopicsDbPath(userDataDir), { nativeBinding }) + topicStore.createRun({ + id: runId, + sessionId, + rangeKind: 'today', + timezone: 'Asia/Shanghai', + locale: 'zh-CN', + startDay: '2026-08-09', + endDay: '2026-08-09', + status: 'paused', + totalDays: 1, + completedDays: 0, + totalBlocks: 1, + completedBlocks: 0, + currentDay: '2026-08-09', + currentBlockIndex: 0, + modelId: 'test-model', + promptVersion: 'topics-v1', + algorithmVersion: 'topics-v1', + inputTokens: 0, + outputTokens: 0, + modelCalls: 0, + lastError: null, + createdAt: 1, + updatedAt: 1, + }) + topicStore.close() +} + test('constructor rejects missing runtime unless the test-only bypass is explicit', () => { const root = makeTempDir() @@ -1188,33 +1218,7 @@ test('deleting a session also removes its derived topic data', () => { fs.mkdirSync(paths.getDatabaseDir(), { recursive: true }) fs.writeFileSync(path.join(paths.getDatabaseDir(), 'session-with-topics.db'), 'placeholder') - const topicStore = new ChatTopicStore(getChatTopicsDbPath(paths.getUserDataDir()), { nativeBinding }) - topicStore.createRun({ - id: 'topic-run', - sessionId: 'session-with-topics', - rangeKind: 'today', - timezone: 'Asia/Shanghai', - locale: 'zh-CN', - startDay: '2026-08-09', - endDay: '2026-08-09', - status: 'paused', - totalDays: 1, - completedDays: 0, - totalBlocks: 1, - completedBlocks: 0, - currentDay: '2026-08-09', - currentBlockIndex: 0, - modelId: 'test-model', - promptVersion: 'topics-v1', - algorithmVersion: 'topics-v1', - inputTokens: 0, - outputTokens: 0, - modelCalls: 0, - lastError: null, - createdAt: 1, - updatedAt: 1, - }) - topicStore.close() + seedTopicRun(paths.getUserDataDir(), 'session-with-topics') const manager = new DatabaseManager(paths, { nativeBinding, allowMissingRuntimeForTests: true }) assert.equal(manager.deleteSessionDatabaseFiles('session-with-topics'), true) @@ -1226,3 +1230,51 @@ test('deleting a session also removes its derived topic data', () => { reopened.close() } }) + +test('failed primary database deletion preserves derived topic data', () => { + const root = makeTempDir() + const paths = createPathProvider(root) + const dbPath = path.join(paths.getDatabaseDir(), 'blocked.db') + fs.mkdirSync(dbPath, { recursive: true }) + seedTopicRun(paths.getUserDataDir(), 'blocked', 'blocked-topic-run') + + const manager = new DatabaseManager(paths, { nativeBinding, allowMissingRuntimeForTests: true }) + assert.throws(() => manager.deleteSessionDatabaseFiles('blocked')) + assert.equal(fs.existsSync(dbPath), true) + + const reopened = new ChatTopicStore(getChatTopicsDbPath(paths.getUserDataDir()), { nativeBinding }) + try { + assert.equal(reopened.getRun('blocked-topic-run')?.sessionId, 'blocked') + } finally { + reopened.close() + } +}) + +test('session deletion rejects a live topic execution owned by another runtime', () => { + const root = makeTempDir() + const paths = createPathProvider(root) + const dbPath = path.join(paths.getDatabaseDir(), 'active-topics.db') + fs.mkdirSync(paths.getDatabaseDir(), { recursive: true }) + fs.writeFileSync(dbPath, 'placeholder') + seedTopicRun(paths.getUserDataDir(), 'active-topics', 'active-topic-run') + + const topicStore = new ChatTopicStore(getChatTopicsDbPath(paths.getUserDataDir()), { nativeBinding }) + const run = topicStore.getRun('active-topic-run')! + topicStore.updateRun({ ...run, status: 'running' }) + assert.equal(topicStore.tryAcquireExecutionLease(run.id, 'other-runtime', Date.now(), Date.now() + 30_000), true) + topicStore.close() + + const manager = new DatabaseManager(paths, { nativeBinding, allowMissingRuntimeForTests: true }) + assert.throws( + () => manager.deleteSessionDatabaseFiles('active-topics'), + (error: unknown) => (error as { code?: string }).code === 'TOPIC_EXECUTION_IN_PROGRESS' + ) + assert.equal(fs.existsSync(dbPath), true) + + const reopened = new ChatTopicStore(getChatTopicsDbPath(paths.getUserDataDir()), { nativeBinding }) + try { + assert.equal(reopened.getRun('active-topic-run')?.status, 'running') + } finally { + reopened.close() + } +}) diff --git a/packages/node-runtime/src/database-manager.ts b/packages/node-runtime/src/database-manager.ts index 940281e8..73716628 100644 --- a/packages/node-runtime/src/database-manager.ts +++ b/packages/node-runtime/src/database-manager.ts @@ -25,7 +25,7 @@ import { getContactsFactsCacheDir } from './services/contacts/paths' import { getGlobalInsightDir, getGlobalInsightFactsCacheDir } from './services/global-insight/paths' import { deleteAnnualSummarySnapshots } from './services/global-insight/snapshot' import { getPeopleRelationshipsFactsCacheDir } from './services/people/relationships/paths' -import { deleteSessionChatTopics } from './services/topics' +import { assertSessionChatTopicsIdle, deleteSessionChatTopics } from './services/topics' interface DatabaseManagerOptions { nativeBinding?: string @@ -248,14 +248,14 @@ export class DatabaseManager { * SQLite WAL 模式会产生 sidecar 文件，前端查询还会写 JSON 缓存；只删主库会让列表和后续同步读到脏状态。 */ deleteSessionDatabaseFiles(sessionId: string): boolean { + assertSessionChatTopicsIdle(this.pathProvider.getUserDataDir(), sessionId, { nativeBinding: this.nativeBinding }) this.close(sessionId) - // 话题是独立派生库中的会话数据，先删除它；失败时保留主库，避免出现“界面已删但派生数据残留”。 - deleteSessionChatTopics(this.pathProvider.getUserDataDir(), sessionId, { nativeBinding: this.nativeBinding }) - const dbPath = this.getDbPath(sessionId) const existed = ['', '-wal', '-shm'].some((suffix) => fs.existsSync(dbPath + suffix)) - for (const suffix of ['', '-wal', '-shm']) { + if (fs.existsSync(dbPath)) fs.unlinkSync(dbPath) + + for (const suffix of ['-wal', '-shm']) { try { const filePath = dbPath + suffix if (fs.existsSync(filePath)) fs.unlinkSync(filePath) @@ -263,6 +263,10 @@ export class DatabaseManager { /* ignore cleanup failures */ } } + + // Derived data is removed only after the primary database is gone. If deleting the main file fails, callers get + // the original filesystem error and can retry without losing paid topic snapshots or other cached results. + deleteSessionChatTopics(this.pathProvider.getUserDataDir(), sessionId, { nativeBinding: this.nativeBinding }) const cacheDir = this.pathProvider.getCacheDir() deleteSessionCache(sessionId, cacheDir) deleteSessionCache(sessionId, path.join(cacheDir, 'query')) diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index fcd9e67e..9be4c21f 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -149,6 +149,7 @@ export type { export { assertDataDirCompatible, DataDirCompatibilityError, + isRuntimeVersionAtLeast, raiseDataDirMinRuntimeVersion, readDataDirCompatibilityMeta, } from './data-dir-compat' diff --git a/packages/node-runtime/src/services/topics/index.ts b/packages/node-runtime/src/services/topics/index.ts index 158ced8b..92cd1581 100644 --- a/packages/node-runtime/src/services/topics/index.ts +++ b/packages/node-runtime/src/services/topics/index.ts @@ -31,7 +31,7 @@ export { loadTopicSourceDay, } from './source' export type { TopicChatType, TopicSourceBlock, TopicSourceDay, TopicSourceMessage } from './source' -export { ChatTopicStore, deleteSessionChatTopics } from './store' +export { ChatTopicStore, assertSessionChatTopicsIdle, deleteSessionChatTopics } from './store' export type { FinalizeTopicDayInput, TopicDayCheckpoint } from './store' export { assertValidTimezone, diff --git a/packages/node-runtime/src/services/topics/service.test.ts b/packages/node-runtime/src/services/topics/service.test.ts index 90387e92..b9fe5e39 100644 --- a/packages/node-runtime/src/services/topics/service.test.ts +++ b/packages/node-runtime/src/services/topics/service.test.ts @@ -7,6 +7,7 @@ import Database from 'better-sqlite3' import type { PathProvider } from '@openchatlab/core' import { CHAT_DB_SCHEMA } from '@openchatlab/core' import { DatabaseManager } from '../../database-manager' +import { assertDataDirCompatible, DataDirCompatibilityError, readDataDirCompatibilityMeta } from '../../data-dir-compat' import { createDatabaseManagerAdapter } from '../adapters' import type { ChatTopicModelClient } from './model-client' import { createChatTopicService, type ChatTopicService } from './service' @@ -69,10 +70,12 @@ function createHarness( let nextRunId = 1 createSession(root, messageCount, chatType) const paths = createPathProvider(root) - const manager = new DatabaseManager(paths, { nativeBinding, allowMissingRuntimeForTests: true }) + const runtimeIdentity = { version: '0.35.1', kind: 'cli' } as const + const manager = new DatabaseManager(paths, { nativeBinding, runtime: runtimeIdentity }) const service = createChatTopicService({ runtime: createDatabaseManagerAdapter(manager), - userDataDir: paths.getUserDataDir(), + pathProvider: paths, + runtimeIdentity, nativeBinding, getModelClient: () => modelClient, now: () => Date.parse('2026-08-09T12:00:00.000Z'), @@ -447,6 +450,47 @@ test('a paused model request resumes from the persisted day checkpoint', async ( } }) +test('a persisted paused run can be cancelled without blocking the next generation', async () => { + let calls = 0 + const modelClient: ChatTopicModelClient = { + modelId: 'test/model', + complete(prompts, options) { + calls += 1 + if (calls === 1) { + return new Promise((_, reject) => { + options.signal.addEventListener('abort', () => reject(new Error('aborted')), { once: true }) + }) + } + return Promise.resolve(successfulTopicResult(prompts)) + }, + } + const { service, manager } = createHarness(modelClient, 1) + + try { + const pausedRun = service.start('group', { + rangeKind: 'today', + timezone: 'Asia/Shanghai', + locale: 'zh-CN', + }) + await waitUntil(() => calls === 1) + service.pause('group', pausedRun.id) + await waitUntil(() => service.getRun('group', pausedRun.id)?.status === 'paused') + await new Promise((resolve) => setImmediate(resolve)) + + assert.equal(service.cancel('group', pausedRun.id).status, 'cancelled') + + const nextRun = service.start('group', { + rangeKind: 'today', + timezone: 'Asia/Shanghai', + locale: 'zh-CN', + }) + assert.equal((await waitForRun(service, 'group', nextRun.id, 'completed')).status, 'completed') + } finally { + service.close() + manager.closeAll() + } +}) + test('interactive AI work resumes topic generation after another session is deleted', async () => { let calls = 0 const modelClient: ChatTopicModelClient = { @@ -507,7 +551,148 @@ test('interactive AI work resumes topic generation after another session is dele } }) -test('session deletion waits for active topic work and removes its derived state', async () => { +test('a second runtime cannot recover or control a live topic execution', async () => { + let firstRuntimeCalled = false + const firstModelClient: ChatTopicModelClient = { + modelId: 'test/model', + complete(_prompts, options) { + firstRuntimeCalled = true + return new Promise((_, reject) => { + options.signal.addEventListener('abort', () => reject(new Error('first runtime stopped')), { once: true }) + }) + }, + } + const { root, service: firstService, manager: firstManager } = createHarness(firstModelClient, 1) + const paths = createPathProvider(root) + let secondManager: DatabaseManager | null = null + let secondService: ChatTopicService | null = null + + try { + const run = firstService.start('group', { + rangeKind: 'today', + timezone: 'Asia/Shanghai', + locale: 'zh-CN', + }) + await waitUntil(() => firstRuntimeCalled) + + const runtimeIdentity = { version: '0.35.1', kind: 'desktop' } as const + secondManager = new DatabaseManager(paths, { nativeBinding, runtime: runtimeIdentity }) + const runtimeService = createChatTopicService({ + runtime: createDatabaseManagerAdapter(secondManager), + pathProvider: paths, + runtimeIdentity, + nativeBinding, + getModelClient: () => firstModelClient, + now: () => Date.parse('2026-08-09T12:00:00.000Z'), + generateId: () => 'second-runtime-run', + }) + + assert.equal(secondService.getRun('group', run.id)?.status, 'running') + assert.throws( + () => secondService.pause('group', run.id), + (error: unknown) => (error as { statusCode?: number }).statusCode === 409 + ) + assert.throws( + () => secondService.cancel('group', run.id), + (error: unknown) => (error as { statusCode?: number }).statusCode === 409 + ) + assert.equal(firstService.getRun('group', run.id)?.status, 'running') + } finally { + firstService.close() + secondService?.close() + firstManager.closeAll() + secondManager?.closeAll() + } +}) + +test('session deletion preparation preserves completed topic data when primary deletion aborts', async () => { + const modelClient: ChatTopicModelClient = { + modelId: 'test/model', + async complete(prompt) { + return successfulTopicResult(prompt) + }, + } + const { service, manager } = createHarness(modelClient, 1) + + try { + const run = service.start('group', { rangeKind: 'today', timezone: 'Asia/Shanghai', locale: 'zh-CN' }) + await waitForRun(service, 'group', run.id, 'completed') + + await chatTopicWorkCoordinator.prepareSessionDelete('group') + + assert.equal(service.getRun('group', run.id)?.status, 'completed') + assert.equal(service.getDay('group', '2026-08-09', 'Asia/Shanghai')?.status, 'ready') + } finally { + service.close() + manager.closeAll() + } +}) + +test('topic storage rejects older runtimes while supported runtimes can delete all session data', async () => { + const modelClient: ChatTopicModelClient = { + modelId: 'test/model', + async complete(prompt) { + return successfulTopicResult(prompt) + }, + } + const { root, service, manager } = createHarness(modelClient, 1) + const pathProvider = createPathProvider(root) + + try { + const meta = readDataDirCompatibilityMeta(pathProvider.getUserDataDir()) + assert.equal(meta?.minRuntimeVersion, '0.35.1') + assert.equal(meta?.dataCompatibilityVersion, 2) + assert.ok(meta?.reasons.includes('chat-topics-store')) + assert.throws( + () => assertDataDirCompatible(pathProvider, { version: '0.35.0', kind: 'cli' }), + (error: unknown) => error instanceof DataDirCompatibilityError && error.code === 'DATA_DIR_REQUIRES_NEWER_RUNTIME' + ) + assert.doesNotThrow(() => assertDataDirCompatible(pathProvider, { version: '0.35.1', kind: 'desktop' })) + + const run = service.start('group', { rangeKind: 'today', timezone: 'Asia/Shanghai', locale: 'zh-CN' }) + await waitForRun(service, 'group', run.id, 'completed') + await chatTopicWorkCoordinator.prepareSessionDelete('group') + + assert.equal(manager.deleteSessionDatabaseFiles('group'), true) + assert.equal(service.getDay('group', '2026-08-09', 'Asia/Shanghai'), null) + } finally { + service.close() + manager.closeAll() + } +}) + +test('session deletion preparation cancels a persisted paused run', async () => { + let called = false + const modelClient: ChatTopicModelClient = { + modelId: 'test/model', + complete(_prompts, options) { + called = true + return new Promise((_, reject) => { + options.signal.addEventListener('abort', () => reject(new Error('paused before deletion')), { once: true }) + }) + }, + } + const { service, manager } = createHarness(modelClient, 1) + + try { + const run = service.start('group', { rangeKind: 'today', timezone: 'Asia/Shanghai', locale: 'zh-CN' }) + await waitUntil(() => called) + service.pause('group', run.id) + await waitUntil(() => service.getRun('group', run.id)?.status === 'paused') + await new Promise((resolve) => setImmediate(resolve)) + + await chatTopicWorkCoordinator.prepareSessionDelete('group') + + assert.equal(service.getRun('group', run.id)?.status, 'cancelled') + assert.equal(manager.deleteSessionDatabaseFiles('group'), true) + assert.equal(service.getRun('group', run.id), null) + } finally { + service.close() + manager.closeAll() + } +}) + +test('session deletion preparation waits for active topic work before primary deletion', async () => { let called = false const modelClient: ChatTopicModelClient = { modelId: 'test/model', @@ -524,6 +709,9 @@ test('session deletion waits for active topic work and removes its derived state const run = service.start('group', { rangeKind: 'today', timezone: 'Asia/Shanghai', locale: 'zh-CN' }) await waitUntil(() => called) await chatTopicWorkCoordinator.prepareSessionDelete('group') + + assert.equal(service.getRun('group', run.id)?.status, 'cancelled') + assert.equal(manager.deleteSessionDatabaseFiles('group'), true) assert.equal(service.getRun('group', run.id), null) assert.equal(service.getLatestRun('group'), null) } finally { diff --git a/packages/node-runtime/src/services/topics/service.ts b/packages/node-runtime/src/services/topics/service.ts index 9ae1a682..8122652f 100644 --- a/packages/node-runtime/src/services/topics/service.ts +++ b/packages/node-runtime/src/services/topics/service.ts @@ -1,6 +1,8 @@ import { randomUUID } from 'node:crypto' +import type { PathProvider } from '@openchatlab/core' import type { ChatTopicDay, ChatTopicPreflight, ChatTopicRun, CreateChatTopicsRequest } from '@openchatlab/shared-types' import { appLogger } from '../../logging/app-logger' +import { isRuntimeVersionAtLeast, raiseDataDirMinRuntimeVersion, type RuntimeIdentity } from '../../data-dir-compat' import type { SessionRuntimeAdapter } from '../adapters' import { applyTopicFinalization, @@ -28,11 +30,13 @@ import { chatTopicWorkCoordinator } from './work-coordinator' export interface ChatTopicServiceDeps { runtime: SessionRuntimeAdapter - userDataDir: string + pathProvider: PathProvider + runtimeIdentity: RuntimeIdentity getModelClient(): ChatTopicModelClient | null nativeBinding?: string now?: () => number generateId?: () => string + runtimeId?: string } export interface ChatTopicService { @@ -53,6 +57,7 @@ interface ActiveExecution { runId: string controller: AbortController requestedStatus: 'paused' | 'cancelled' | 'preempted' | null + leaseLost: boolean } interface PlannedDay { @@ -62,15 +67,30 @@ interface PlannedDay { skip: boolean } +const TOPIC_EXECUTION_LEASE_DURATION_MS = 30_000 +const TOPIC_EXECUTION_HEARTBEAT_MS = 10_000 +const CHAT_TOPICS_MIN_RUNTIME_VERSION = '0.35.1' +const CHAT_TOPICS_DATA_COMPATIBILITY_VERSION = 2 + export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicService { const now = deps.now ?? Date.now const generateId = deps.generateId ?? randomUUID - const store = new ChatTopicStore(getChatTopicsDbPath(deps.userDataDir), { nativeBinding: deps.nativeBinding }) + const runtimeId = deps.runtimeId ?? randomUUID() + const store = new ChatTopicStore(getChatTopicsDbPath(deps.pathProvider.getUserDataDir()), { + nativeBinding: deps.nativeBinding, + }) + try { + raiseChatTopicsCompatibilityGate(deps.pathProvider, deps.runtimeIdentity) + } catch (error) { + store.close() + throw error + } store.recoverInterruptedRuns(now()) let activeExecution: ActiveExecution | null = null let closed = false let storeClosed = false let preemptedRunId: string | null = null + let leaseHeartbeat: { runId: string; timer: ReturnType } | null = null const unsubscribeCoordinator = chatTopicWorkCoordinator.subscribe(handleInteractiveStateChange) const unsubscribeSessionDelete = chatTopicWorkCoordinator.subscribeSessionDelete(prepareSessionDelete) const executionWaiters: Array<{ runId: string; resolve: () => void }> = [] @@ -132,6 +152,7 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer totalBlocks: number modelId: string }): ChatTopicRun { + store.recoverInterruptedRuns(now()) const active = store.getActiveRun() if (active) { throw Object.assign(new Error(`A chat topic run is already ${active.status}`), { @@ -166,22 +187,36 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer updatedAt: timestamp, } store.createRun(run) - if (run.status === 'pending') launch(run.id) + if (run.status === 'pending') { + if (!acquireExecutionLease(run.id)) { + store.updateRun({ ...run, status: 'cancelled', lastError: 'Another runtime owns topic generation' }) + const competing = store.getActiveRun() + throw Object.assign(new Error('Another ChatLab runtime is already generating chat topics'), { + statusCode: 409, + activeRun: competing, + }) + } + launch(run.id) + } return run } function getRun(sessionId: string, runId: string): ChatTopicRun | null { + store.recoverInterruptedRuns(now()) const run = store.getRun(runId) return run?.sessionId === sessionId ? run : null } function getLatestRun(sessionId: string): ChatTopicRun | null { + store.recoverInterruptedRuns(now()) return store.getLatestRun(sessionId) } function pause(sessionId: string, runId: string): ChatTopicRun { const run = requireRun(sessionId, runId) if (run.status === 'pending' && activeExecution?.runId !== runId) { + requireExecutionLease(runId) + if (preemptedRunId === runId) preemptedRunId = null return updateRun(run, { status: 'paused' }) } if (run.status !== 'running' && run.status !== 'pending') return run @@ -191,6 +226,7 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer activeExecution.controller.abort() return run } + requireExecutionLease(runId) return updateRun(run, { status: 'paused' }) } @@ -202,9 +238,17 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer if (active && active.id !== runId) { throw Object.assign(new Error(`A chat topic run is already ${active.status}`), { statusCode: 409 }) } - const resumed = updateRun(run, { status: 'pending', lastError: null }) - launch(runId) - return resumed + if (!acquireExecutionLease(runId)) { + throw Object.assign(new Error('Another ChatLab runtime is already generating chat topics'), { statusCode: 409 }) + } + try { + const resumed = updateRun(run, { status: 'pending', lastError: null }) + launch(runId) + return resumed + } catch (error) { + releaseExecutionLease(runId) + throw error + } } function cancel(sessionId: string, runId: string): ChatTopicRun { @@ -214,6 +258,10 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer preemptedRunId = null activeExecution.requestedStatus = 'cancelled' activeExecution.controller.abort() + return run + } + if (!acquireExecutionLease(runId)) { + throw Object.assign(new Error('Another ChatLab runtime is already generating chat topics'), { statusCode: 409 }) } return updateRun(run, { status: 'cancelled' }) } @@ -240,12 +288,19 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer closed = true unsubscribeCoordinator() unsubscribeSessionDelete() - preemptedRunId = null if (activeExecution) { activeExecution.requestedStatus = 'paused' activeExecution.controller.abort() return } + if (preemptedRunId) { + const pending = store.getRun(preemptedRunId) + if (pending && store.ownsExecutionLease(pending.id, runtimeId, now())) { + updateRun(pending, { status: 'paused' }) + } + preemptedRunId = null + } + stopLeaseHeartbeat() closeStore() } @@ -255,7 +310,12 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer preemptedRunId = runId return } - const execution: ActiveExecution = { runId, controller: new AbortController(), requestedStatus: null } + const execution: ActiveExecution = { + runId, + controller: new AbortController(), + requestedStatus: null, + leaseLost: false, + } activeExecution = execution queueMicrotask(() => { void executeRun(execution).finally(() => { @@ -270,6 +330,7 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer async function executeRun(execution: ActiveExecution): Promise { let run = store.getRun(execution.runId) if (!run) return + if (execution.leaseLost) return if (execution.controller.signal.aborted) { updateRun(run, { status: execution.requestedStatus === 'preempted' ? 'pending' : (execution.requestedStatus ?? 'paused'), @@ -321,6 +382,12 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer modelCalls: run.modelCalls, }) } catch (error) { + if (execution.leaseLost || isExecutionLeaseLostError(error)) { + appLogger.warn('chat-topics', 'chat topic generation stopped after execution lease was lost', { + runId: execution.runId, + }) + return + } const latest = store.getRun(run.id) ?? run const requestedStatus = execution.requestedStatus const status = requestedStatus === 'preempted' ? 'pending' : (requestedStatus ?? 'failed') @@ -427,22 +494,25 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer if (currentSource.sourceSignature !== source.sourceSignature) { throw new Error('Chat messages changed during topic generation; retry is required') } - store.finalizeDay({ - sessionId: run.sessionId, - dayKey: source.dayKey, - timezone: source.timezone, - sourceSignature: source.sourceSignature, - sourceMessageCount: source.messages.length, - sourceFirstTs: source.messages[0]!.timestamp, - sourceLastTs: source.messages.at(-1)!.timestamp, - runId: run.id, - modelId: modelClient.modelId, - promptVersion: CHAT_TOPICS_PROMPT_VERSION, - algorithmVersion: CHAT_TOPICS_ALGORITHM_VERSION, - overview: finalResult.value.finalization.overview, - topics: materializeChatTopics(finalResult.value.ledger, source.messages), - generatedAt: now(), - }) + store.finalizeDay( + { + sessionId: run.sessionId, + dayKey: source.dayKey, + timezone: source.timezone, + sourceSignature: source.sourceSignature, + sourceMessageCount: source.messages.length, + sourceFirstTs: source.messages[0]!.timestamp, + sourceLastTs: source.messages.at(-1)!.timestamp, + runId: run.id, + modelId: modelClient.modelId, + promptVersion: CHAT_TOPICS_PROMPT_VERSION, + algorithmVersion: CHAT_TOPICS_ALGORITHM_VERSION, + overview: finalResult.value.finalization.overview, + topics: materializeChatTopics(finalResult.value.ledger, source.messages), + generatedAt: now(), + }, + executionLeaseGuard() + ) run = updateRun(run, { completedDays: run.completedDays + 1, currentDay: null, @@ -496,31 +566,34 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer status: TopicDayCheckpoint['status'], lastError: string | null = null ): void { - store.saveCheckpoint({ - sessionId: run.sessionId, - dayKey: source.dayKey, - timezone: source.timezone, - status, - sourceSignature: source.sourceSignature, - sourceMessageCount: source.messages.length, - sourceFirstTs: source.messages[0]!.timestamp, - sourceLastTs: source.messages.at(-1)!.timestamp, - runId: run.id, - totalBlocks: source.blocks.length, - completedBlockIndex, - ledgerJson: serializeTopicLedger(ledger), - modelId: run.modelId, - promptVersion: CHAT_TOPICS_PROMPT_VERSION, - algorithmVersion: CHAT_TOPICS_ALGORITHM_VERSION, - lastError, - updatedAt: now(), - }) + store.saveCheckpoint( + { + sessionId: run.sessionId, + dayKey: source.dayKey, + timezone: source.timezone, + status, + sourceSignature: source.sourceSignature, + sourceMessageCount: source.messages.length, + sourceFirstTs: source.messages[0]!.timestamp, + sourceLastTs: source.messages.at(-1)!.timestamp, + runId: run.id, + totalBlocks: source.blocks.length, + completedBlockIndex, + ledgerJson: serializeTopicLedger(ledger), + modelId: run.modelId, + promptVersion: CHAT_TOPICS_PROMPT_VERSION, + algorithmVersion: CHAT_TOPICS_ALGORITHM_VERSION, + lastError, + updatedAt: now(), + }, + executionLeaseGuard() + ) } function markCurrentCheckpointFailed(run: ChatTopicRun, lastError: string | null): void { const checkpoint = run.currentDay ? store.getCheckpoint(run.sessionId, run.currentDay) : null if (!checkpoint) return - store.saveCheckpoint({ ...checkpoint, status: 'failed', lastError, updatedAt: now() }) + store.saveCheckpoint({ ...checkpoint, status: 'failed', lastError, updatedAt: now() }, executionLeaseGuard()) } function isReusableCheckpoint( @@ -539,9 +612,57 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer ) } + function acquireExecutionLease(runId: string): boolean { + const acquired = store.tryAcquireExecutionLease(runId, runtimeId, now(), now() + TOPIC_EXECUTION_LEASE_DURATION_MS) + if (acquired) startLeaseHeartbeat(runId) + return acquired + } + + function startLeaseHeartbeat(runId: string): void { + stopLeaseHeartbeat() + const timer = setInterval(() => { + if (closed || storeClosed) return + const renewed = store.renewExecutionLease(runId, runtimeId, now() + TOPIC_EXECUTION_LEASE_DURATION_MS) + if (renewed) return + stopLeaseHeartbeat(runId) + if (preemptedRunId === runId) preemptedRunId = null + if (activeExecution?.runId === runId) { + activeExecution.leaseLost = true + activeExecution.controller.abort() + } + }, TOPIC_EXECUTION_HEARTBEAT_MS) + timer.unref?.() + leaseHeartbeat = { runId, timer } + } + + function stopLeaseHeartbeat(runId?: string): void { + if (!leaseHeartbeat || (runId && leaseHeartbeat.runId !== runId)) return + clearInterval(leaseHeartbeat.timer) + leaseHeartbeat = null + } + + function releaseExecutionLease(runId: string): void { + stopLeaseHeartbeat(runId) + store.releaseExecutionLease(runId, runtimeId) + } + + function requireExecutionLease(runId: string): void { + if (store.ownsExecutionLease(runId, runtimeId, now())) return + throw Object.assign(new Error('This chat topic run is active in another ChatLab runtime'), { statusCode: 409 }) + } + + function executionLeaseGuard() { + return { ownerId: runtimeId, now: now() } + } + function updateRun(run: ChatTopicRun, updates: Partial): ChatTopicRun { const updated = { ...run, ...updates, updatedAt: now() } - store.updateRun(updated) + if (!store.updateRunIfOwned(updated, executionLeaseGuard())) { + throw Object.assign(new Error('Chat topic execution lease was lost'), { code: 'TOPIC_EXECUTION_LEASE_LOST' }) + } + if (['completed', 'failed', 'paused', 'cancelled'].includes(updated.status)) { + releaseExecutionLease(updated.id) + } return updated } @@ -564,6 +685,7 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer function closeStore(): void { if (storeClosed) return + stopLeaseHeartbeat() storeClosed = true store.close() } @@ -585,7 +707,7 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer if (closed || chatTopicWorkCoordinator.isInteractiveActive || activeExecution || !preemptedRunId) return const runId = preemptedRunId const pending = store.getRun(runId) - if (pending?.status !== 'pending') { + if (pending?.status !== 'pending' || !store.ownsExecutionLease(runId, runtimeId, now())) { preemptedRunId = null return } @@ -605,8 +727,9 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer } } const activeRun = store.getActiveRun() - if (activeRun?.sessionId === sessionId) updateRun(activeRun, { status: 'cancelled' }) - store.deleteSession(sessionId) + if (activeRun?.sessionId === sessionId) { + cancel(sessionId, activeRun.id) + } } function waitForExecution(runId: string): Promise { @@ -638,6 +761,23 @@ export function createChatTopicService(deps: ChatTopicServiceDeps): ChatTopicSer } } +function raiseChatTopicsCompatibilityGate(pathProvider: PathProvider, runtime: RuntimeIdentity): void { + // The source branch still reports the last released version until the release workflow bumps package metadata. + // Do not make local development builds block themselves; the first publishable runtime is 0.35.1. + if (!isRuntimeVersionAtLeast(runtime.version, CHAT_TOPICS_MIN_RUNTIME_VERSION)) return + raiseDataDirMinRuntimeVersion(pathProvider, { + minRuntimeVersion: CHAT_TOPICS_MIN_RUNTIME_VERSION, + dataCompatibilityVersion: CHAT_TOPICS_DATA_COMPATIBILITY_VERSION, + reason: 'chat-topics-store', + runtime, + module: 'chat-topics', + }) +} + +function isExecutionLeaseLostError(error: unknown): boolean { + return (error as { code?: string } | null)?.code === 'TOPIC_EXECUTION_LEASE_LOST' +} + function validationErrorMessage(error: unknown): string { const message = error instanceof Error ? error.message : String(error) return message.replace(/[\r\n]+/g, ' ').slice(0, 500) diff --git a/packages/node-runtime/src/services/topics/store.test.ts b/packages/node-runtime/src/services/topics/store.test.ts index 671c8893..0a5a7699 100644 --- a/packages/node-runtime/src/services/topics/store.test.ts +++ b/packages/node-runtime/src/services/topics/store.test.ts @@ -223,6 +223,67 @@ test('deleting a session removes its snapshots, checkpoints and runs only', () = } }) +test('execution leases preserve live runs and recover them after expiry', () => { + const root = makeTempDir() + const store = new ChatTopicStore(getChatTopicsDbPath(root), { nativeBinding }) + const now = 1_786_205_000_000 + + try { + const run = createRun({ updatedAt: now }) + store.createRun(run) + assert.equal(store.tryAcquireExecutionLease(run.id, 'runtime-a', now, now + 30_000), true) + assert.equal(store.tryAcquireExecutionLease(run.id, 'runtime-b', now + 1, now + 30_001), false) + assert.equal(store.recoverInterruptedRuns(now + 10_000), 0) + assert.equal(store.getRun(run.id)?.status, 'running') + assert.equal( + store.updateRunIfOwned({ ...run, completedBlocks: 1 }, { ownerId: 'runtime-b', now: now + 10_000 }), + false + ) + + assert.equal(store.recoverInterruptedRuns(now + 30_001), 1) + assert.equal(store.getRun(run.id)?.status, 'paused') + assert.equal(store.tryAcquireExecutionLease(run.id, 'runtime-b', now + 30_001, now + 60_001), true) + assert.equal( + store.updateRunIfOwned({ ...run, status: 'running' }, { ownerId: 'runtime-a', now: now + 30_002 }), + false + ) + } finally { + store.close() + } +}) + +test('schema v2 stores add execution leases without losing existing runs', () => { + const root = makeTempDir() + const dbPath = getChatTopicsDbPath(root) + const initial = new ChatTopicStore(dbPath, { nativeBinding }) + initial.createRun(createRun({ status: 'completed' })) + initial.close() + + const legacy = new Database(dbPath, { nativeBinding }) + legacy.exec("UPDATE topic_meta SET value = '2' WHERE key = 'schema_version'; DROP TABLE topic_execution_lease;") + legacy.close() + + const migrated = new ChatTopicStore(dbPath, { nativeBinding }) + try { + assert.equal(migrated.getRun('run-1')?.status, 'completed') + const raw = new Database(dbPath, { nativeBinding }) + try { + assert.equal(raw.prepare("SELECT value FROM topic_meta WHERE key = 'schema_version'").pluck().get(), '3') + assert.equal( + raw + .prepare("SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'topic_execution_lease'") + .pluck() + .get(), + 1 + ) + } finally { + raw.close() + } + } finally { + migrated.close() + } +}) + test('schema v1 snapshots migrate without losing their legacy range fallback', () => { const root = makeTempDir() const dbPath = getChatTopicsDbPath(root) @@ -291,7 +352,14 @@ test('schema v1 snapshots migrate without losing their legacy range fallback', ( assert.deepEqual(topic?.messageIds, [10, 12]) const migrated = new Database(dbPath, { nativeBinding }) try { - assert.equal(migrated.prepare("SELECT value FROM topic_meta WHERE key = 'schema_version'").pluck().get(), '2') + assert.equal(migrated.prepare("SELECT value FROM topic_meta WHERE key = 'schema_version'").pluck().get(), '3') + assert.equal( + migrated + .prepare("SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'topic_execution_lease'") + .pluck() + .get(), + 1 + ) } finally { migrated.close() } diff --git a/packages/node-runtime/src/services/topics/store.ts b/packages/node-runtime/src/services/topics/store.ts index 7746562e..86b575d3 100644 --- a/packages/node-runtime/src/services/topics/store.ts +++ b/packages/node-runtime/src/services/topics/store.ts @@ -11,7 +11,7 @@ import type { import { openBetterSqliteDatabase } from '../../better-sqlite3-adapter' import { getChatTopicsDbPath } from './paths' -const CHAT_TOPICS_SCHEMA_VERSION = 2 +const CHAT_TOPICS_SCHEMA_VERSION = 3 interface TopicStoreOptions { nativeBinding?: string @@ -39,6 +39,11 @@ export interface TopicDayCheckpoint { export type StoredTopicDayCheckpoint = TopicDayCheckpoint +export interface TopicExecutionLeaseGuard { + ownerId: string + now: number +} + export interface FinalizeTopicDayInput { sessionId: string dayKey: string @@ -199,6 +204,105 @@ export class ChatTopicStore { ) } + updateRunIfOwned(run: ChatTopicRun, guard: TopicExecutionLeaseGuard): boolean { + return ( + this.db + .prepare( + `UPDATE topic_run SET + status = ?, total_days = ?, completed_days = ?, total_blocks = ?, completed_blocks = ?, + current_day = ?, current_block_index = ?, + model_id = ?, input_tokens = ?, output_tokens = ?, model_calls = ?, last_error = ?, updated_at = ? + WHERE id = ? + AND EXISTS ( + SELECT 1 FROM topic_execution_lease + WHERE singleton = 1 AND run_id = ? AND owner_id = ? AND expires_at > ? + )` + ) + .run( + run.status, + run.totalDays, + run.completedDays, + run.totalBlocks, + run.completedBlocks, + run.currentDay, + run.currentBlockIndex, + run.modelId, + run.inputTokens, + run.outputTokens, + run.modelCalls, + run.lastError, + run.updatedAt, + run.id, + run.id, + guard.ownerId, + guard.now + ).changes > 0 + ) + } + + tryAcquireExecutionLease(runId: string, ownerId: string, now: number, expiresAt: number): boolean { + return ( + this.db + .prepare( + `INSERT INTO topic_execution_lease (singleton, run_id, owner_id, expires_at) + VALUES (1, ?, ?, ?) + ON CONFLICT(singleton) DO UPDATE SET + run_id = excluded.run_id, + owner_id = excluded.owner_id, + expires_at = excluded.expires_at + WHERE topic_execution_lease.owner_id = excluded.owner_id + OR topic_execution_lease.expires_at <= ?` + ) + .run(runId, ownerId, expiresAt, now).changes > 0 + ) + } + + renewExecutionLease(runId: string, ownerId: string, expiresAt: number): boolean { + return ( + this.db + .prepare( + `UPDATE topic_execution_lease SET expires_at = ? + WHERE singleton = 1 AND run_id = ? AND owner_id = ?` + ) + .run(expiresAt, runId, ownerId).changes > 0 + ) + } + + ownsExecutionLease(runId: string, ownerId: string, now: number): boolean { + return Boolean( + this.db + .prepare( + `SELECT 1 FROM topic_execution_lease + WHERE singleton = 1 AND run_id = ? AND owner_id = ? AND expires_at > ?` + ) + .get(runId, ownerId, now) + ) + } + + hasLiveExecutionForSession(sessionId: string, now: number): boolean { + return Boolean( + this.db + .prepare( + `SELECT 1 + FROM topic_execution_lease lease + JOIN topic_run run ON run.id = lease.run_id + WHERE lease.singleton = 1 + AND lease.expires_at > ? + AND run.session_id = ? + AND run.status IN ('pending', 'running')` + ) + .get(now, sessionId) + ) + } + + releaseExecutionLease(runId: string, ownerId: string): boolean { + return ( + this.db + .prepare('DELETE FROM topic_execution_lease WHERE singleton = 1 AND run_id = ? AND owner_id = ?') + .run(runId, ownerId).changes > 0 + ) + } + getRun(runId: string): ChatTopicRun | null { const row = this.db .prepare( @@ -257,62 +361,88 @@ export class ChatTopicStore { this.db .prepare( `UPDATE topic_day_work SET status = 'failed', last_error = 'Generation interrupted by application restart', updated_at = ? - WHERE status IN ('pending', 'running')` + WHERE status IN ('pending', 'running') + AND NOT EXISTS ( + SELECT 1 FROM topic_execution_lease + WHERE singleton = 1 + AND run_id = topic_day_work.run_id + AND expires_at > ? + )` ) - .run(updatedAt) - return this.db + .run(updatedAt, updatedAt) + const recovered = this.db .prepare( `UPDATE topic_run SET status = 'paused', last_error = 'Generation interrupted by application restart', updated_at = ? - WHERE status IN ('pending', 'running')` + WHERE status IN ('pending', 'running') + AND NOT EXISTS ( + SELECT 1 FROM topic_execution_lease + WHERE singleton = 1 + AND run_id = topic_run.id + AND expires_at > ? + )` ) - .run(updatedAt).changes + .run(updatedAt, updatedAt).changes + this.db + .prepare( + `DELETE FROM topic_execution_lease + WHERE expires_at <= ? + OR NOT EXISTS ( + SELECT 1 FROM topic_run + WHERE id = topic_execution_lease.run_id AND status IN ('pending', 'running') + )` + ) + .run(updatedAt) + return recovered }) } - saveCheckpoint(checkpoint: TopicDayCheckpoint): void { - this.db - .prepare( - `INSERT INTO topic_day_work ( - session_id, day_key, timezone, status, source_signature, source_message_count, - source_first_ts, source_last_ts, run_id, total_blocks, completed_block_index, - ledger_json, model_id, prompt_version, algorithm_version, last_error, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(session_id, day_key) DO UPDATE SET - timezone = excluded.timezone, - status = excluded.status, - source_signature = excluded.source_signature, - source_message_count = excluded.source_message_count, - source_first_ts = excluded.source_first_ts, - source_last_ts = excluded.source_last_ts, - run_id = excluded.run_id, - total_blocks = excluded.total_blocks, - completed_block_index = excluded.completed_block_index, - ledger_json = excluded.ledger_json, - model_id = excluded.model_id, - prompt_version = excluded.prompt_version, - algorithm_version = excluded.algorithm_version, - last_error = excluded.last_error, - updated_at = excluded.updated_at` - ) - .run( - checkpoint.sessionId, - checkpoint.dayKey, - checkpoint.timezone, - checkpoint.status, - checkpoint.sourceSignature, - checkpoint.sourceMessageCount, - checkpoint.sourceFirstTs, - checkpoint.sourceLastTs, - checkpoint.runId, - checkpoint.totalBlocks, - checkpoint.completedBlockIndex, - checkpoint.ledgerJson, - checkpoint.modelId, - checkpoint.promptVersion, - checkpoint.algorithmVersion, - checkpoint.lastError ?? null, - checkpoint.updatedAt - ) + saveCheckpoint(checkpoint: TopicDayCheckpoint, guard?: TopicExecutionLeaseGuard): void { + this.db.transaction(() => { + if (guard) this.assertExecutionLease(checkpoint.runId, guard) + this.db + .prepare( + `INSERT INTO topic_day_work ( + session_id, day_key, timezone, status, source_signature, source_message_count, + source_first_ts, source_last_ts, run_id, total_blocks, completed_block_index, + ledger_json, model_id, prompt_version, algorithm_version, last_error, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(session_id, day_key) DO UPDATE SET + timezone = excluded.timezone, + status = excluded.status, + source_signature = excluded.source_signature, + source_message_count = excluded.source_message_count, + source_first_ts = excluded.source_first_ts, + source_last_ts = excluded.source_last_ts, + run_id = excluded.run_id, + total_blocks = excluded.total_blocks, + completed_block_index = excluded.completed_block_index, + ledger_json = excluded.ledger_json, + model_id = excluded.model_id, + prompt_version = excluded.prompt_version, + algorithm_version = excluded.algorithm_version, + last_error = excluded.last_error, + updated_at = excluded.updated_at` + ) + .run( + checkpoint.sessionId, + checkpoint.dayKey, + checkpoint.timezone, + checkpoint.status, + checkpoint.sourceSignature, + checkpoint.sourceMessageCount, + checkpoint.sourceFirstTs, + checkpoint.sourceLastTs, + checkpoint.runId, + checkpoint.totalBlocks, + checkpoint.completedBlockIndex, + checkpoint.ledgerJson, + checkpoint.modelId, + checkpoint.promptVersion, + checkpoint.algorithmVersion, + checkpoint.lastError ?? null, + checkpoint.updatedAt + ) + }) } getCheckpoint(sessionId: string, dayKey: string): StoredTopicDayCheckpoint | null { @@ -330,8 +460,9 @@ export class ChatTopicStore { return row ?? null } - finalizeDay(input: FinalizeTopicDayInput): void { + finalizeDay(input: FinalizeTopicDayInput, guard?: TopicExecutionLeaseGuard): void { this.db.transaction(() => { + if (guard) this.assertExecutionLease(input.runId, guard) this.db.prepare('DELETE FROM topic WHERE session_id = ? AND day_key = ?').run(input.sessionId, input.dayKey) // 成功快照与进行中的 checkpoint 分离。刷新失败时仍可继续展示上一次成功结果。 @@ -531,6 +662,17 @@ export class ChatTopicStore { }) } + private assertExecutionLease(runId: string, guard: TopicExecutionLeaseGuard): void { + const owned = this.db + .prepare( + `UPDATE topic_execution_lease SET expires_at = expires_at + WHERE singleton = 1 AND run_id = ? AND owner_id = ? AND expires_at > ?` + ) + .run(runId, guard.ownerId, guard.now).changes + if (owned > 0) return + throw Object.assign(new Error('Chat topic execution lease was lost'), { code: 'TOPIC_EXECUTION_LEASE_LOST' }) + } + private initialize(): void { this.db.exec(` CREATE TABLE IF NOT EXISTS topic_meta ( @@ -575,6 +717,14 @@ export class ChatTopicStore { updated_at INTEGER NOT NULL ); + CREATE TABLE IF NOT EXISTS topic_execution_lease ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + run_id TEXT NOT NULL, + owner_id TEXT NOT NULL, + expires_at INTEGER NOT NULL, + FOREIGN KEY (run_id) REFERENCES topic_run(id) ON DELETE CASCADE + ); + CREATE TABLE IF NOT EXISTS topic_day ( session_id TEXT NOT NULL, day_key TEXT NOT NULL, @@ -672,6 +822,11 @@ export class ChatTopicStore { ); CREATE INDEX IF NOT EXISTS idx_topic_message_message ON topic_message(message_id); `) + if (version < CHAT_TOPICS_SCHEMA_VERSION) { + this.db + .prepare("UPDATE topic_meta SET value = ? WHERE key = 'schema_version'") + .run(String(CHAT_TOPICS_SCHEMA_VERSION)) + } } } } @@ -691,6 +846,27 @@ export function deleteSessionChatTopics( } } +export function assertSessionChatTopicsIdle( + userDataDir: string, + sessionId: string, + options: TopicStoreOptions = {} +): void { + const dbPath = getChatTopicsDbPath(userDataDir) + if (!fs.existsSync(dbPath)) return + const store = new ChatTopicStore(dbPath, options) + try { + const timestamp = Date.now() + store.recoverInterruptedRuns(timestamp) + if (!store.hasLiveExecutionForSession(sessionId, timestamp)) return + throw Object.assign(new Error('Chat topics are being generated for this session in another runtime'), { + code: 'TOPIC_EXECUTION_IN_PROGRESS', + statusCode: 409, + }) + } finally { + store.close() + } +} + function parseStringArray(value: string): string[] { const parsed = JSON.parse(value) as unknown if (!Array.isArray(parsed) || parsed.some((item) => typeof item !== 'string')) {