feat(runtime): add global insight service and worker

apps/cli/tsup.config.ts

@@ -7,6 +7,7 @@ export default defineConfig({
     'semantic-index-worker': '../../packages/node-runtime/src/semantic-index/worker-thread-entry.ts',
     'contacts-worker': '../../packages/node-runtime/src/services/contacts/worker-entry.ts',
     'people-relationships-worker': '../../packages/node-runtime/src/services/people/relationships/worker-entry.ts',
+    'global-insight-worker': '../../packages/node-runtime/src/services/global-insight/worker-entry.ts',
   },
   format: ['esm'],
   outDir: 'dist',

apps/desktop/electron.vite.config.ts

@@ -34,6 +34,10 @@ export default defineConfig({
             rootDir,
             'packages/node-runtime/src/services/people/relationships/worker-entry.ts'
           ),
+          'global-insight-worker': resolve(
+            rootDir,
+            'packages/node-runtime/src/services/global-insight/worker-entry.ts'
+          ),
         },
       },
     },

apps/desktop/main/database/core.test.ts

@@ -44,6 +44,9 @@ test('openDatabase keeps readonly validation read-only for DELETE journal databa
         contactsService: {
           getContactsFactsCacheDir: () => path.join(root, 'cache', 'contacts'),
         },
+        globalInsightService: {
+          getGlobalInsightFactsCacheDir: () => path.join(root, 'cache', 'global-insight'),
+        },
         peopleRelationshipsService: {
           getPeopleRelationshipsFactsCacheDir: () => path.join(root, 'cache', 'people-relationships'),
         },

apps/desktop/main/database/core.ts

@@ -16,6 +16,7 @@ import {
   BetterSqliteAdapter,
   writeParseResultToDb,
   contactsService,
+  globalInsightService,
   peopleRelationshipsService,
 } from '@openchatlab/node-runtime'
 import type { RuntimeIdentity } from '@openchatlab/node-runtime/src/data-dir-compat'
@@ -171,6 +172,10 @@ export function deleteSession(sessionId: string): boolean {
     deleteSessionCache(sessionId, cacheDir)
     deleteSessionCache(sessionId, path.join(cacheDir, 'query'))
     deleteSessionCache(sessionId, contactsService.getContactsFactsCacheDir(getPathProvider().getUserDataDir()))
+    deleteSessionCache(
+      sessionId,
+      globalInsightService.getGlobalInsightFactsCacheDir(getPathProvider().getUserDataDir())
+    )
     deleteSessionCache(
       sessionId,
       peopleRelationshipsService.getPeopleRelationshipsFactsCacheDir(getPathProvider().getUserDataDir())

packages/node-runtime/src/index.ts

@@ -376,6 +376,7 @@ export * as exportService from './services/export-service'
 export * as ownerProfileService from './services/owner-profile-service'
 export * as contactsService from './services/contacts'
 export * as peopleRelationshipsService from './services/people/relationships'
+export * as globalInsightService from './services/global-insight'
 // Semantic index (Phase 1 vector search) — independent of legacy ai/rag
 export * as semanticIndex from './semantic-index'
 export {
@@ -407,9 +408,11 @@ export type {
 export {
   CONTACTS_ALGORITHM_VERSION,
   PEOPLE_RELATIONSHIPS_ALGORITHM_VERSION,
+  ANNUAL_SUMMARY_ALGORITHM_VERSION,
   MergeSessionCache,
   createContactsService,
   createPeopleRelationshipsService,
+  createGlobalInsightService,
   createDatabaseManagerAdapter,
   pushImport,
 } from './services'
@@ -429,6 +432,10 @@ export type {
   PeopleRelationshipsService,
   PeopleRelationshipsServiceDeps,
   PeopleRelationshipsServiceOptions,
+  AnnualSummaryComputeRunner,
+  GlobalInsightService,
+  GlobalInsightServiceDeps,
+  GlobalInsightServiceOptions,
   PushImportPayload,
   PushImportResult,
   PushImportOutcome,

packages/node-runtime/src/services/global-insight/index.ts

@@ -0,0 +1,5 @@
+export { createGlobalInsightService } from './service'
+export type { GlobalInsightService, GlobalInsightServiceDeps, GlobalInsightServiceOptions } from './service'
+export { ANNUAL_SUMMARY_ALGORITHM_VERSION } from './types'
+export type { AnnualSummaryComputeRunner, AnnualSummaryRunnerOptions } from './worker-runner'
+export { getGlobalInsightDir, getGlobalInsightFactsCacheDir } from './paths'

packages/node-runtime/src/services/global-insight/service.test.ts

@@ -0,0 +1,166 @@
+import assert from 'node:assert/strict'
+import fs from 'node:fs'
+import os from 'node:os'
+import path from 'node:path'
+import test from 'node:test'
+import type { AnnualSummaryRange } from '@openchatlab/shared-types'
+import type { SessionRuntimeAdapter } from '../adapters'
+import { createGlobalInsightService } from './service'
+import type { AnnualSummarySnapshot } from './types'
+import type { AnnualSummaryComputeRunner } from './worker-runner'
+
+function makeSnapshot(range: AnnualSummaryRange, signature: string, count = 3): AnnualSummarySnapshot {
+  return {
+    algorithmVersion: 'annual-summary-v1',
+    signature,
+    computedAt: 100,
+    range,
+    availableDataYears: [2026],
+    latestDataYear: 2026,
+    metrics: {
+      sentMessageCount: count,
+      activeDayCount: 1,
+      directContactCount: 1,
+      averageMessagesPerDay: count,
+      averageDirectContactsPerDay: 1,
+    },
+    monthlyActivity: [],
+    dailyActivity: [],
+    messageTypes: [],
+    textLength: { textMessageCount: 0, median: null, p90: null, buckets: [] },
+    coverage: {
+      totalSessions: 1,
+      analyzedSessions: 1,
+      missingOwnerSessions: 0,
+      unresolvedOwnerSessions: 0,
+      failedSessions: 0,
+    },
+    workerStats: { durationMs: 1, totalSessions: 1, processedSessions: 1, cacheHits: 0, cacheMisses: 1 },
+  }
+}
+
+function createEnv(runner: AnnualSummaryComputeRunner) {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-global-insight-service-'))
+  const dbPath = path.join(dir, 'chat.db')
+  fs.writeFileSync(dbPath, 'db')
+  const adapter = {
+    listSessionIds: () => ['chat-1'],
+    getDbPath: () => dbPath,
+  } as unknown as SessionRuntimeAdapter
+  const service = createGlobalInsightService({ adapter, userDataDir: dir, runner, now: () => Date.UTC(2026, 0, 2) })
+  return { dir, dbPath, adapter, service }
+}
+
+async function flushTasks(): Promise<void> {
+  await new Promise((resolve) => setTimeout(resolve, 0))
+}
+
+test('starts one background task on a missing snapshot and then serves it fresh', async (t) => {
+  let calls = 0
+  const env = createEnv(async ({ signature, range }) => {
+    calls++
+    return makeSnapshot(range, signature)
+  })
+  t.after(async () => {
+    await env.service.close()
+    fs.rmSync(env.dir, { recursive: true, force: true })
+  })
+
+  const first = env.service.getAnnualSummary({ mode: 'year', year: 2026 })
+  const duplicate = env.service.getAnnualSummary({ mode: 'year', year: 2026 })
+  assert.equal(first.cache.status, 'missing')
+  assert.equal(first.metrics, null)
+  assert.equal(duplicate.task.status, 'running')
+  assert.equal(calls, 1)
+
+  await flushTasks()
+  const fresh = env.service.getAnnualSummary({ mode: 'year', year: 2026 })
+  assert.equal(fresh.cache.status, 'fresh')
+  assert.equal(fresh.metrics?.sentMessageCount, 3)
+})
+
+test('serves stale data while recomputing after the DB signature changes', async (t) => {
+  let resolveRunner: ((snapshot: AnnualSummarySnapshot) => void) | undefined
+  const runner: AnnualSummaryComputeRunner = ({ signature, range }) =>
+    new Promise((resolve) => {
+      resolveRunner = () => resolve(makeSnapshot(range, signature, 9))
+    })
+  const env = createEnv(runner)
+  t.after(async () => {
+    await env.service.close()
+    fs.rmSync(env.dir, { recursive: true, force: true })
+  })
+  const range = env.service.normalizeRange({ mode: 'year', year: 2026 })
+  env.service.replaceSnapshotForTests(makeSnapshot(range, 'old-signature'))
+
+  const stale = env.service.getAnnualSummary({ mode: 'year', year: 2026, acceptStale: true })
+  assert.equal(stale.cache.status, 'stale')
+  assert.equal(stale.metrics?.sentMessageCount, 3)
+  assert.equal(stale.task.status, 'running')
+
+  resolveRunner?.(makeSnapshot(range, stale.cache.signature ?? '', 9))
+})
+
+test('preserves failed state until explicit recompute', async (t) => {
+  let calls = 0
+  const env = createEnv(async ({ signature, range }) => {
+    calls++
+    if (calls === 1) throw new Error('worker failed')
+    return makeSnapshot(range, signature)
+  })
+  t.after(async () => {
+    await env.service.close()
+    fs.rmSync(env.dir, { recursive: true, force: true })
+  })
+
+  env.service.getAnnualSummary({ mode: 'year', year: 2026 })
+  await flushTasks()
+  assert.equal(env.service.getAnnualSummary({ mode: 'year', year: 2026 }).task.status, 'failed')
+  assert.equal(calls, 1)
+
+  env.service.startRecompute({ mode: 'year', year: 2026 })
+  await flushTasks()
+  assert.equal(calls, 2)
+  assert.equal(env.service.getAnnualSummary({ mode: 'year', year: 2026 }).cache.status, 'fresh')
+})
+
+test('discards a worker result when the DB signature changes during compute', async (t) => {
+  let finish: (() => void) | undefined
+  const env = createEnv(
+    ({ signature, range }) =>
+      new Promise((resolve) => {
+        finish = () => resolve(makeSnapshot(range, signature))
+      })
+  )
+  t.after(async () => {
+    await env.service.close()
+    fs.rmSync(env.dir, { recursive: true, force: true })
+  })
+  env.service.getAnnualSummary({ mode: 'year', year: 2026 })
+  fs.appendFileSync(env.dbPath, 'changed')
+  finish?.()
+  await flushTasks()
+
+  const response = env.service.getAnnualSummary({ mode: 'year', year: 2026 })
+  assert.equal(response.cache.status, 'missing')
+  assert.equal(response.task.status, 'running')
+})
+
+test('close aborts an in-flight task', async (t) => {
+  let aborted = false
+  const env = createEnv(
+    ({ signal }) =>
+      new Promise((_resolve, reject) => {
+        signal.addEventListener('abort', () => {
+          aborted = true
+          reject(new Error('aborted'))
+        })
+      })
+  )
+  t.after(() => fs.rmSync(env.dir, { recursive: true, force: true }))
+  env.service.getAnnualSummary({ mode: 'year', year: 2026 })
+
+  await env.service.close()
+
+  assert.equal(aborted, true)
+})

packages/node-runtime/src/services/global-insight/service.ts

@@ -0,0 +1,262 @@
+import type { PathProvider } from '@openchatlab/core'
+import type { AnnualSummaryRange, AnnualSummaryResponse, AnnualSummaryTaskState } from '@openchatlab/shared-types'
+import type { RuntimeIdentity } from '../../data-dir-compat'
+import { appLogger } from '../../logging/app-logger'
+import type { SessionRuntimeAdapter } from '../adapters'
+import { getGlobalInsightDir } from './paths'
+import { buildAnnualSummarySignature } from './signature'
+import {
+  cleanupAnnualSummarySnapshotTempFiles,
+  readAnnualSummarySnapshot,
+  writeAnnualSummarySnapshot,
+} from './snapshot'
+import { normalizeAnnualSummaryRange, toAnnualSummaryRangeKey, type AnnualSummaryRangeInput } from './time-range'
+import { createAnnualSummaryWorkerRunner, type AnnualSummaryComputeRunner } from './worker-runner'
+import type { AnnualSummarySnapshot } from './types'
+
+export interface GlobalInsightServiceOptions extends AnnualSummaryRangeInput {
+  acceptStale?: boolean
+  forceRecompute?: boolean
+}
+
+export interface GlobalInsightServiceDeps {
+  adapter: SessionRuntimeAdapter
+  pathProvider?: PathProvider
+  userDataDir?: string
+  runtimeIdentity?: RuntimeIdentity
+  nativeBinding?: string
+  workerEntryUrl?: string | URL
+  runner?: AnnualSummaryComputeRunner
+  now?: () => number
+}
+
+export interface GlobalInsightService {
+  getAnnualSummary(options?: GlobalInsightServiceOptions): AnnualSummaryResponse
+  startRecompute(options?: GlobalInsightServiceOptions): AnnualSummaryResponse
+  invalidateCache(): void
+  close(): Promise<void>
+  normalizeRange(input?: AnnualSummaryRangeInput): AnnualSummaryRange
+  replaceSnapshotForTests(snapshot: AnnualSummarySnapshot): void
+}
+
+interface InFlightTask {
+  id: string
+  signature: string
+  promise: Promise<AnnualSummarySnapshot>
+  abortController: AbortController
+}
+
+export function createGlobalInsightService(deps: GlobalInsightServiceDeps): GlobalInsightService {
+  return new DefaultGlobalInsightService(deps)
+}
+
+class DefaultGlobalInsightService implements GlobalInsightService {
+  private readonly snapshots = new Map<string, AnnualSummarySnapshot | null>()
+  private readonly snapshotDir: string
+  private readonly runner: AnnualSummaryComputeRunner
+  private inFlight: InFlightTask | null = null
+  private task: AnnualSummaryTaskState = createIdleTaskState()
+
+  constructor(private readonly deps: GlobalInsightServiceDeps) {
+    this.snapshotDir = deps.userDataDir
+      ? getGlobalInsightDir(deps.userDataDir)
+      : getGlobalInsightDir(requirePathProvider(deps).getUserDataDir())
+    cleanupAnnualSummarySnapshotTempFiles(this.snapshotDir)
+    this.runner =
+      deps.runner ??
+      createAnnualSummaryWorkerRunner({
+        pathProvider: requirePathProvider(deps),
+        runtimeIdentity: deps.runtimeIdentity,
+        nativeBinding: deps.nativeBinding,
+        workerEntryUrl: deps.workerEntryUrl,
+      })
+  }
+
+  getAnnualSummary(options: GlobalInsightServiceOptions = {}): AnnualSummaryResponse {
+    const range = this.normalizeRange(options)
+    const signature = buildAnnualSummarySignature(this.deps.adapter, range)
+    const status = this.getCacheStatus(signature, range)
+    if (this.shouldStartTask(options, status)) this.ensureTaskStarted(signature, range)
+    return this.toResponse(signature, range, options.acceptStale === true)
+  }
+
+  startRecompute(options: GlobalInsightServiceOptions = {}): AnnualSummaryResponse {
+    const range = this.normalizeRange(options)
+    const signature = buildAnnualSummarySignature(this.deps.adapter, range)
+    this.ensureTaskStarted(signature, range)
+    return this.toResponse(signature, range, true)
+  }
+
+  invalidateCache(): void {
+    this.snapshots.clear()
+  }
+
+  normalizeRange(input: AnnualSummaryRangeInput = {}): AnnualSummaryRange {
+    return normalizeAnnualSummaryRange(input, new Date(this.now()))
+  }
+
+  replaceSnapshotForTests(snapshot: AnnualSummarySnapshot): void {
+    this.snapshots.set(toAnnualSummaryRangeKey(snapshot.range), snapshot)
+  }
+
+  async close(): Promise<void> {
+    const inFlight = this.inFlight
+    if (!inFlight) return
+    this.inFlight = null
+    inFlight.abortController.abort()
+    this.task = {
+      ...this.task,
+      status: 'failed',
+      finishedAt: this.now(),
+      lastError: 'annual summary task aborted',
+    }
+  }
+
+  private shouldStartTask(
+    options: GlobalInsightServiceOptions,
+    status: AnnualSummaryResponse['cache']['status']
+  ): boolean {
+    if (options.forceRecompute) return true
+    if (status === 'fresh') return false
+    return this.task.status !== 'failed'
+  }
+
+  private ensureTaskStarted(signature: string, range: AnnualSummaryRange): void {
+    if (this.inFlight) return
+    const id = `annual_summary_${this.now()}_${Math.random().toString(36).slice(2)}`
+    this.task = {
+      id,
+      status: 'running',
+      startedAt: this.now(),
+      finishedAt: null,
+      processedSessions: 0,
+      totalSessions: this.deps.adapter.listSessionIds().length,
+    }
+    const abortController = new AbortController()
+    const promise = this.runner({
+      signature,
+      range,
+      signal: abortController.signal,
+      onProgress: (progress) => {
+        if (this.task.id !== id || this.task.status !== 'running') return
+        this.task = { ...this.task, ...progress }
+      },
+    })
+    this.inFlight = { id, signature, promise, abortController }
+    promise
+      .then((snapshot) => this.handleTaskSuccess(id, signature, snapshot))
+      .catch((error) => this.handleTaskFailure(id, error))
+  }
+
+  private handleTaskSuccess(id: string, inputSignature: string, snapshot: AnnualSummarySnapshot): void {
+    if (this.inFlight?.id !== id) return
+    this.inFlight = null
+    const latestSignature = buildAnnualSummarySignature(this.deps.adapter, snapshot.range)
+    const finishedAt = this.now()
+    if (inputSignature !== latestSignature || snapshot.signature !== latestSignature) {
+      this.task = { ...this.task, status: 'superseded', finishedAt, currentSessionId: undefined }
+      appLogger.info('global-insight', 'annual summary worker result discarded because signature changed')
+      return
+    }
+    try {
+      writeAnnualSummarySnapshot(this.snapshotDir, snapshot)
+      this.snapshots.set(toAnnualSummaryRangeKey(snapshot.range), snapshot)
+      this.task = {
+        ...this.task,
+        status: 'succeeded',
+        finishedAt,
+        processedSessions: snapshot.workerStats.processedSessions,
+        totalSessions: snapshot.workerStats.totalSessions,
+        currentSessionId: undefined,
+      }
+      appLogger.info('global-insight', 'annual summary snapshot persisted', {
+        analyzedSessions: snapshot.coverage.analyzedSessions,
+        durationMs: snapshot.workerStats.durationMs,
+      })
+    } catch (error) {
+      this.handleTaskFailure(id, error)
+    }
+  }
+
+  private handleTaskFailure(id: string, error: unknown): void {
+    if (this.inFlight?.id !== id) return
+    this.inFlight = null
+    this.task = {
+      ...this.task,
+      status: 'failed',
+      finishedAt: this.now(),
+      currentSessionId: undefined,
+      lastError: error instanceof Error ? error.message : String(error),
+    }
+    appLogger.error('global-insight', 'annual summary worker failed', error)
+  }
+
+  private toResponse(signature: string, range: AnnualSummaryRange, acceptStale: boolean): AnnualSummaryResponse {
+    const snapshot = this.getSnapshot(range)
+    const status = this.getCacheStatus(signature, range)
+    const includeSnapshot = status === 'fresh' || (status === 'stale' && acceptStale)
+    return {
+      range,
+      availableDataYears: includeSnapshot ? (snapshot?.availableDataYears ?? []) : [],
+      latestDataYear: includeSnapshot ? (snapshot?.latestDataYear ?? null) : null,
+      metrics: includeSnapshot ? (snapshot?.metrics ?? null) : null,
+      monthlyActivity: includeSnapshot ? (snapshot?.monthlyActivity ?? []) : [],
+      dailyActivity: includeSnapshot ? (snapshot?.dailyActivity ?? []) : [],
+      messageTypes: includeSnapshot ? (snapshot?.messageTypes ?? []) : [],
+      textLength: includeSnapshot ? (snapshot?.textLength ?? null) : null,
+      coverage: includeSnapshot ? (snapshot?.coverage ?? emptyCoverage()) : emptyCoverage(),
+      cache: {
+        status,
+        computedAt: snapshot?.computedAt ?? null,
+        signature: snapshot?.signature,
+        staleReason: status === 'stale' ? 'signature_changed' : undefined,
+      },
+      task: this.task,
+    }
+  }
+
+  private getCacheStatus(signature: string, range: AnnualSummaryRange): AnnualSummaryResponse['cache']['status'] {
+    const snapshot = this.getSnapshot(range)
+    if (!snapshot) return 'missing'
+    return snapshot.signature === signature ? 'fresh' : 'stale'
+  }
+
+  private getSnapshot(range: AnnualSummaryRange): AnnualSummarySnapshot | null {
+    const key = toAnnualSummaryRangeKey(range)
+    if (!this.snapshots.has(key)) {
+      this.snapshots.set(key, readAnnualSummarySnapshot(this.snapshotDir, range, { now: this.deps.now }))
+    }
+    return this.snapshots.get(key) ?? null
+  }
+
+  private now(): number {
+    return this.deps.now?.() ?? Date.now()
+  }
+}
+
+function requirePathProvider(deps: GlobalInsightServiceDeps): PathProvider {
+  if (!deps.pathProvider)
+    throw new Error('GlobalInsightService requires pathProvider unless userDataDir and runner are provided')
+  return deps.pathProvider
+}
+
+function createIdleTaskState(): AnnualSummaryTaskState {
+  return {
+    id: null,
+    status: 'idle',
+    startedAt: null,
+    finishedAt: null,
+    processedSessions: 0,
+    totalSessions: 0,
+  }
+}
+
+function emptyCoverage(): AnnualSummaryResponse['coverage'] {
+  return {
+    totalSessions: 0,
+    analyzedSessions: 0,
+    missingOwnerSessions: 0,
+    unresolvedOwnerSessions: 0,
+    failedSessions: 0,
+  }
+}

packages/node-runtime/src/services/global-insight/worker-entry.ts

@@ -0,0 +1,43 @@
+import { parentPort, workerData } from 'node:worker_threads'
+import type { AnnualSummaryRange } from '@openchatlab/shared-types'
+import { DatabaseManager } from '../../database-manager'
+import type { RuntimeIdentity } from '../../data-dir-compat'
+import { initAppLogger } from '../../logging/app-logger'
+import { StaticPathProvider, type StaticPathProviderSnapshot } from '../../semantic-index/static-path-provider'
+import { createDatabaseManagerAdapter } from '../adapters'
+import { computeAnnualSummarySnapshot, type ComputeAnnualSummarySnapshotOptions } from './compute'
+import { getGlobalInsightFactsCacheDir } from './paths'
+
+interface StartupOptions {
+  paths: StaticPathProviderSnapshot
+  runtimeIdentity?: RuntimeIdentity
+  nativeBinding?: string
+  signature: string
+  range: AnnualSummaryRange
+}
+
+async function main(): Promise<void> {
+  if (!parentPort) throw new Error('annual summary worker requires parentPort')
+  const options = workerData as StartupOptions
+  initAppLogger(options.paths.logsDir)
+  const pathProvider = new StaticPathProvider(options.paths)
+  const dbManager = new DatabaseManager(pathProvider, {
+    nativeBinding: options.nativeBinding,
+    runtime: options.runtimeIdentity,
+    allowMissingRuntimeForTests: !options.runtimeIdentity,
+  })
+  const adapter = createDatabaseManagerAdapter(dbManager)
+  const computeOptions: ComputeAnnualSummarySnapshotOptions = {
+    adapter,
+    signature: options.signature,
+    range: options.range,
+    factsCacheDir: getGlobalInsightFactsCacheDir(pathProvider.getUserDataDir()),
+    onProgress: (progress) => parentPort?.postMessage({ type: 'progress', progress }),
+  }
+  const snapshot = computeAnnualSummarySnapshot(computeOptions)
+  parentPort.postMessage({ type: 'success', snapshot })
+}
+
+main().catch((error) => {
+  parentPort?.postMessage({ type: 'error', error: error instanceof Error ? error.message : String(error) })
+})

packages/node-runtime/src/services/global-insight/worker-runner.test.ts

@@ -0,0 +1,26 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+import { resolveDefaultAnnualSummaryWorkerEntryUrl } from './worker-runner'
+
+test('resolves TypeScript development worker beside the runner', () => {
+  assert.equal(
+    resolveDefaultAnnualSummaryWorkerEntryUrl(
+      'file:///repo/packages/node-runtime/src/services/global-insight/worker-runner.ts'
+    ).href,
+    'file:///repo/packages/node-runtime/src/services/global-insight/worker-entry.ts'
+  )
+})
+
+test('resolves bundled CLI worker beside the mjs bundle', () => {
+  assert.equal(
+    resolveDefaultAnnualSummaryWorkerEntryUrl('file:///app/cli/dist/global-insight-worker.mjs').href,
+    'file:///app/cli/dist/global-insight-worker.mjs'
+  )
+})
+
+test('falls back to the desktop bundled worker when no sibling entry exists', () => {
+  assert.equal(
+    resolveDefaultAnnualSummaryWorkerEntryUrl('file:///app/dist/main/index.js', () => false).href,
+    'file:///app/dist/main/global-insight-worker.js'
+  )
+})

packages/node-runtime/src/services/global-insight/worker-runner.ts

@@ -0,0 +1,119 @@
+import { existsSync } from 'node:fs'
+import { Worker, type WorkerOptions } from 'node:worker_threads'
+import type { PathProvider } from '@openchatlab/core'
+import type { AnnualSummaryRange } from '@openchatlab/shared-types'
+import type { RuntimeIdentity } from '../../data-dir-compat'
+import { snapshotPathProvider } from '../../semantic-index/static-path-provider'
+import type { AnnualSummaryComputeProgress, AnnualSummarySnapshot } from './types'
+
+type ModuleWorkerOptions = WorkerOptions & { type: 'module' }
+type EntryExists = (url: URL) => boolean
+
+export interface AnnualSummaryRunnerOptions {
+  signature: string
+  range: AnnualSummaryRange
+  onProgress: (progress: AnnualSummaryComputeProgress) => void
+  signal: AbortSignal
+}
+
+export type AnnualSummaryComputeRunner = (options: AnnualSummaryRunnerOptions) => Promise<AnnualSummarySnapshot>
+
+export interface AnnualSummaryWorkerRunnerOptions {
+  pathProvider: PathProvider
+  runtimeIdentity?: RuntimeIdentity
+  nativeBinding?: string
+  workerEntryUrl?: string | URL
+}
+
+interface WorkerMessage {
+  type: 'progress' | 'success' | 'error'
+  progress?: AnnualSummaryComputeProgress
+  snapshot?: AnnualSummarySnapshot
+  error?: string
+}
+
+export function resolveDefaultAnnualSummaryWorkerEntryUrl(
+  currentModuleUrl: string | URL = import.meta.url,
+  entryExists: EntryExists = (url) => existsSync(url)
+): URL {
+  const moduleUrl = typeof currentModuleUrl === 'string' ? currentModuleUrl : currentModuleUrl.href
+  if (moduleUrl.endsWith('.ts')) return new URL('./worker-entry.ts', moduleUrl)
+  if (moduleUrl.endsWith('.mjs')) return new URL('./global-insight-worker.mjs', moduleUrl)
+  const sibling = new URL('./worker-entry.js', moduleUrl)
+  return entryExists(sibling) ? sibling : new URL('./global-insight-worker.js', moduleUrl)
+}
+
+export function createAnnualSummaryWorkerRunner(options: AnnualSummaryWorkerRunnerOptions): AnnualSummaryComputeRunner {
+  return ({ signature, range, signal, onProgress }) =>
+    new Promise<AnnualSummarySnapshot>((resolve, reject) => {
+      if (signal.aborted) {
+        reject(new Error('annual summary worker aborted'))
+        return
+      }
+      const worker = createWorker(
+        {
+          paths: snapshotPathProvider(options.pathProvider),
+          runtimeIdentity: options.runtimeIdentity,
+          nativeBinding: options.nativeBinding,
+          signature,
+          range,
+        },
+        options.workerEntryUrl
+      )
+      let settled = false
+      const abort = () => {
+        if (settled) return
+        settled = true
+        void worker.terminate()
+        reject(new Error('annual summary worker aborted'))
+      }
+      signal.addEventListener('abort', abort, { once: true })
+      worker.on('message', (message: WorkerMessage) => {
+        if (message.type === 'progress' && message.progress) {
+          onProgress(message.progress)
+          return
+        }
+        if (message.type === 'success' && message.snapshot) {
+          settled = true
+          signal.removeEventListener('abort', abort)
+          resolve(message.snapshot)
+          void worker.terminate()
+          return
+        }
+        if (message.type === 'error') {
+          settled = true
+          signal.removeEventListener('abort', abort)
+          reject(new Error(message.error ?? 'annual summary worker failed'))
+          void worker.terminate()
+        }
+      })
+      worker.on('error', (error) => {
+        if (settled) return
+        settled = true
+        signal.removeEventListener('abort', abort)
+        reject(error)
+      })
+      worker.on('exit', (code) => {
+        if (settled || code === 0) return
+        settled = true
+        signal.removeEventListener('abort', abort)
+        reject(new Error(`annual summary worker exited with code ${code}`))
+      })
+    })
+}
+
+function createWorker(workerData: unknown, entryUrlInput?: string | URL): Worker {
+  const entryUrl = entryUrlInput
+    ? typeof entryUrlInput === 'string'
+      ? new URL(entryUrlInput)
+      : entryUrlInput
+    : resolveDefaultAnnualSummaryWorkerEntryUrl()
+  if (!entryUrl.href.endsWith('.ts')) return new Worker(entryUrl, { workerData })
+  const bootstrap = `
+    import { register } from 'tsx/esm/api';
+    register();
+    await import(${JSON.stringify(entryUrl.href)});
+  `
+  const workerOptions: ModuleWorkerOptions = { eval: true, type: 'module', workerData, execArgv: [] }
+  return new Worker(bootstrap, workerOptions)
+}

packages/node-runtime/src/services/index.ts

@@ -68,6 +68,15 @@ export type {
   PeopleRelationshipsServiceOptions,
 } from './people/relationships'
 
+// Global insight service
+export { ANNUAL_SUMMARY_ALGORITHM_VERSION, createGlobalInsightService } from './global-insight'
+export type {
+  AnnualSummaryComputeRunner,
+  GlobalInsightService,
+  GlobalInsightServiceDeps,
+  GlobalInsightServiceOptions,
+} from './global-insight'
+
 // Merge cache
 export { MergeSessionCache } from './merge-cache'