fix(topics): coordinate session lifecycle across runtimes

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<string, DatabaseAdapter>([['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()

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()

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<RuntimeRouteContext, 'sessionAdapter' | 'pathProvider' | 'nativeBinding'> &
+type ChatTopicRouteContext = Pick<
+  RuntimeRouteContext,
+  'sessionAdapter' | 'pathProvider' | 'runtimeIdentity' | 'nativeBinding'
+> &
   Pick<AiRouteContext, 'llmConfigStore'>
 
 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()

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<RuntimeRouteContext, 'dbManager' | 'sessionAdapter' | 'pathProvider' | 'nativeBinding'> &
+  Pick<RuntimeRouteContext, 'dbManager' | 'sessionAdapter' | 'pathProvider' | 'runtimeIdentity' | 'nativeBinding'> &
   Pick<StorageRouteContext, 'showInFolder'>
 
 export interface AiRouteOptions {

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) => {

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

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()
+  }
+})

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'))

packages/node-runtime/src/index.ts

@@ -149,6 +149,7 @@ export type {
 export {
   assertDataDirCompatible,
   DataDirCompatibilityError,
+  isRuntimeVersionAtLeast,
   raiseDataDirMinRuntimeVersion,
   readDataDirCompatibilityMeta,
 } from './data-dir-compat'

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,

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<void>((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<void>((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 {

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<typeof setInterval> } | 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<void> {
     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>): 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<void> {
@@ -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)

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()
     }

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')) {