fix(relationships): release session databases during graph compute

packages/node-runtime/src/services/people/relationships/compute.test.ts

@@ -72,6 +72,7 @@ class TestEnv {
 
     this.adapter = {
       listSessionIds: () => [...this.dbPaths.keys()],
+      listSessionCandidateIds: () => [...this.dbPaths.keys()],
       openReadonly: (id) => open(id, true),
       openWritable: (id) => open(id, false),
       closeSession: () => {},
@@ -189,6 +190,64 @@ function makeGraphEdge(sourceKey: string, targetKey: string, weight: number): Pe
   }
 }
 
+test('processes database candidates without retaining all session connections', (t) => {
+  const env = new TestEnv()
+  t.after(() => env.cleanup())
+
+  for (const id of ['private-a', 'private-b']) {
+    env.seed({
+      id,
+      platform: 'weixin',
+      type: 'private',
+      ownerId: 'owner',
+      members: [
+        { id: 1, platformId: 'owner', accountName: 'Me' },
+        { id: 2, platformId: `friend-${id}`, accountName: `Friend ${id}` },
+      ],
+      messages: [
+        { id: 1, senderId: 1, ts: 1704103200 },
+        { id: 2, senderId: 2, ts: 1704103201 },
+      ],
+    })
+  }
+
+  const baseAdapter = env.adapter
+  const activeDbs = new Map<string, DatabaseAdapter>()
+  const closedSessionIds = new Set<string>()
+  let maxActiveDbs = 0
+  const adapter: SessionRuntimeAdapter = {
+    ...baseAdapter,
+    listSessionIds: () => {
+      throw new Error('relationship computation must use database candidates')
+    },
+    openReadonly: (sessionId) => {
+      const db = baseAdapter.openReadonly(sessionId)
+      if (db) {
+        activeDbs.set(sessionId, db)
+        maxActiveDbs = Math.max(maxActiveDbs, activeDbs.size)
+      }
+      return db
+    },
+    closeSession: (sessionId) => {
+      const db = activeDbs.get(sessionId)
+      if (db) db.close()
+      activeDbs.delete(sessionId)
+      closedSessionIds.add(sessionId)
+    },
+  }
+
+  const snapshot = computePeopleRelationshipsSnapshot({
+    adapter,
+    signature: 'sig-candidates',
+    timeRangePreset: 'all',
+  })
+
+  assert.equal(snapshot.workerStats.totalSessions, 2)
+  assert.equal(maxActiveDbs, 1)
+  assert.equal(activeDbs.size, 0)
+  assert.deepEqual([...closedSessionIds].sort(), ['private-a', 'private-b'])
+})
+
 test('computes a cropped relationship galaxy from private contacts and group interaction edges', (t) => {
   const env = new TestEnv()
   t.after(() => env.cleanup())

packages/node-runtime/src/services/people/relationships/compute.ts

@@ -193,7 +193,7 @@ export function computePeopleRelationshipsSnapshot(
   options: ComputePeopleRelationshipsSnapshotOptions
 ): PeopleRelationshipsSnapshot {
   const startedAt = options.now?.() ?? Date.now()
-  const sessionIds = options.adapter.listSessionIds()
+  const sessionIds = options.adapter.listSessionCandidateIds?.() ?? options.adapter.listSessionIds()
   const limits = normalizeLimits(options.limits)
   const factsCache = options.factsCacheDir ? createFactsCacheContext(options.factsCacheDir) : null
   const timeRange = resolvePeopleRelationshipsTimeRange(
@@ -437,6 +437,7 @@ function computePeopleRelationships(options: {
       diagnostics.skippedFailedSessions++
       appLogger.error('people-relationships', `failed to process people relationship session: ${sessionId}`, error)
     } finally {
+      options.adapter.closeSession(sessionId)
       processedSessions++
       options.onProgress?.({ processedSessions, totalSessions: options.sessionIds.length, currentSessionId: sessionId })
     }
@@ -477,6 +478,8 @@ function findGlobalLatestMessageTs(
         `failed to inspect people relationship session range: ${sessionId}`,
         error
       )
+    } finally {
+      adapter.closeSession(sessionId)
     }
   }
   return latest

packages/node-runtime/src/services/people/relationships/service.test.ts

@@ -599,11 +599,22 @@ test('starts a background task when snapshot is missing and returns task state i
   try {
     let runnerCalls = 0
     let releaseRunner!: (snapshot: PeopleRelationshipsSnapshot) => void
+    let runnerSignature = ''
+    const baseAdapter = makeAdapter()
+    const adapter: SessionRuntimeAdapter = {
+      ...baseAdapter,
+      listSessionIds: () => {
+        throw new Error('relationship service must use database candidates')
+      },
+      listSessionCandidateIds: () => ['session-a', 'session-b'],
+      getDbPath: (sessionId) => path.join('/tmp', `chatlab-${sessionId}.db`),
+    }
     const service = createPeopleRelationshipsService({
-      adapter: makeAdapter(),
+      adapter,
       systemDir: dir,
-      runner: () => {
+      runner: ({ signature }) => {
         runnerCalls++
+        runnerSignature = signature
         return new Promise<PeopleRelationshipsSnapshot>((resolve) => {
           releaseRunner = resolve
         })
@@ -615,8 +626,12 @@ test('starts a background task when snapshot is missing and returns task state i
 
     assert.equal(response.cache.status, 'missing')
     assert.equal(response.task?.status, 'running')
+    assert.equal(response.task?.totalSessions, 2)
     assert.equal(runnerCalls, 1)
-    releaseRunner(makeSnapshot(makeFreshSignature()))
+    releaseRunner({
+      ...makeSnapshot(runnerSignature),
+      workerStats: { durationMs: 1, totalSessions: 2, processedSessions: 2, skippedFailedSessions: 0 },
+    })
     await new Promise((resolve) => setTimeout(resolve, 0))
     await service.close()
   } finally {

packages/node-runtime/src/services/people/relationships/service.ts

@@ -186,6 +186,7 @@ class DefaultPeopleRelationshipsService implements PeopleRelationshipsService {
   private ensureTaskStarted(signature: string, timeRangePreset: ContactsTimeRangePreset): void {
     if (this.inFlight) return
 
+    const sessionIds = this.deps.adapter.listSessionCandidateIds?.() ?? this.deps.adapter.listSessionIds()
     const taskId = `people_relationships_${this.now()}_${Math.random().toString(36).slice(2)}`
     this.task = {
       id: taskId,
@@ -193,7 +194,7 @@ class DefaultPeopleRelationshipsService implements PeopleRelationshipsService {
       startedAt: this.now(),
       finishedAt: null,
       processedSessions: 0,
-      totalSessions: this.deps.adapter.listSessionIds().length,
+      totalSessions: sessionIds.length,
       timeRangePreset,
     }
 

packages/node-runtime/src/services/people/relationships/signature.test.ts

@@ -0,0 +1,50 @@
+/**
+ * Run: pnpm test -- packages/node-runtime/src/services/people/relationships/signature.test.ts
+ */
+
+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 { SessionRuntimeAdapter } from '../../adapters'
+import { PEOPLE_RELATIONSHIPS_ALGORITHM_VERSION } from './compute'
+import { buildPeopleRelationshipsSignature } from './signature'
+
+test('builds the signature from database candidates without opening every session', (t) => {
+  const baseDir = process.env.CHATLAB_TEST_TMPDIR ?? (fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir())
+  const dir = fs.mkdtempSync(path.join(baseDir, 'chatlab-relationships-signature-'))
+  t.after(() => fs.rmSync(dir, { recursive: true, force: true }))
+
+  const pathsById = new Map([
+    ['session-b', path.join(dir, 'b.db')],
+    ['session-a', path.join(dir, 'a.db')],
+  ])
+  for (const dbPath of pathsById.values()) fs.writeFileSync(dbPath, 'db')
+
+  const adapter: SessionRuntimeAdapter = {
+    listSessionIds: () => {
+      throw new Error('signature must not validate or open every database')
+    },
+    listSessionCandidateIds: () => [...pathsById.keys()],
+    openReadonly: () => {
+      throw new Error('signature must not open databases')
+    },
+    openWritable: () => null,
+    closeSession: () => {},
+    getDbPath: (sessionId) => pathsById.get(sessionId) ?? '',
+    deleteSessionFile: () => false,
+    ensureReadonly: () => {
+      throw new Error('signature must not open databases')
+    },
+    ensureWritable: () => {
+      throw new Error('signature must not open databases')
+    },
+  }
+
+  const signature = buildPeopleRelationshipsSignature(adapter)
+
+  assert.match(signature, new RegExp(`algorithm:${PEOPLE_RELATIONSHIPS_ALGORITHM_VERSION}`))
+  assert.match(signature, /range:1y/)
+  assert.ok(signature.indexOf('session-a:') < signature.indexOf('session-b:'))
+})

packages/node-runtime/src/services/people/relationships/signature.ts

@@ -12,7 +12,8 @@ export function buildPeopleRelationshipsSignature(
     `algorithm:${PEOPLE_RELATIONSHIPS_ALGORITHM_VERSION}`,
     `range:${normalizePeopleRelationshipsTimeRangePreset(timeRangePreset)}`,
   ]
-  for (const sessionId of [...adapter.listSessionIds()].sort()) {
+  const sessionIds = adapter.listSessionCandidateIds?.() ?? adapter.listSessionIds()
+  for (const sessionId of [...sessionIds].sort()) {
     const dbPath = adapter.getDbPath(sessionId)
     parts.push(`${sessionId}:${getDbFileVersion(dbPath)}`)
   }