fix: 增量导入缺失的 Hermes 会话

packages/server/src/db/hermes/session-store.ts

@@ -129,14 +129,16 @@ function mapMessageRow(row: Record<string, unknown>): HermesMessageRow {
 export function createSession(data: {
   id: string
   profile?: string
+  source?: string
   model?: string
   title?: string
   workspace?: string
 }): HermesSessionRow {
   const now = Math.floor(Date.now() / 1000)
+  const source = data.source || 'api_server'
   if (!isSqliteAvailable()) {
     return {
-      id: data.id, profile: data.profile || 'default', source: 'api_server',
+      id: data.id, profile: data.profile || 'default', source,
       user_id: null, model: data.model || '', title: data.title || null,
       started_at: now, ended_at: null, end_reason: null,
       message_count: 0, tool_call_count: 0,
@@ -148,8 +150,8 @@ export function createSession(data: {
   const db = getDb()!
   db.prepare(
     `INSERT INTO ${SESSIONS_TABLE} (id, profile, source, model, title, started_at, last_active, workspace)
-     VALUES (?, ?, 'api_server', ?, ?, ?, ?, ?)`,
-  ).run(data.id, data.profile || 'default', data.model || '', data.title || null, now, now, data.workspace || null)
+     VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+  ).run(data.id, data.profile || 'default', source, data.model || '', data.title || null, now, now, data.workspace || null)
   return getSession(data.id)!
 }
 

packages/server/src/services/hermes/session-sync.ts

@@ -1,39 +1,32 @@
 /**
- * Sync Hermes sessions from all profiles on startup.
- * Reads api_server sessions from Hermes state.db and imports into local DB.
- * Only runs when local DB is empty (first startup).
+ * Incrementally sync Hermes sessions from all profiles on startup.
+ * Reads durable Hermes state.db sessions and mirrors missing chat-visible
+ * sessions into WebUI's local session DB.
  *
  * Uses sessions-db.ts query logic to properly aggregate session chains.
  */
 import { readdirSync, existsSync } from 'fs'
-import { resolve, join } from 'path'
-import { homedir } from 'os'
-import { randomBytes } from 'crypto'
-import { getProfileDir } from './hermes-profile'
-import { createSession, addMessage, updateSession } from '../../db/hermes/session-store'
+import { join } from 'path'
+import {
+  createSession,
+  addMessages,
+  updateSession,
+  deleteSession,
+} from '../../db/hermes/session-store'
 import { getDb } from '../../db/index'
 import { logger } from '../logger'
-import { listSessionSummaries as listHermesSessionSummaries } from '../../db/hermes/sessions-db'
+import {
+  listSessionSummaries as listHermesSessionSummaries,
+  getSessionDetailFromDbWithProfile,
+} from '../../db/hermes/sessions-db'
 import { detectHermesHome } from './hermes-path'
 
 const HERMES_BASE = detectHermesHome()
 const PROFILES_DIR = join(HERMES_BASE, 'profiles')
+const SYNC_SOURCES = ['api_server', 'webui'] as const
 
-/**
- * Generate a UUID v4 without external dependencies
- */
-function generateUuid(): string {
-  const bytes = randomBytes(16)
-  bytes[6] = (bytes[6]! & 0x0f) | 0x40 // Version 4
-  bytes[8] = (bytes[8]! & 0x3f) | 0x80 // Variant 10
-  return [
-    bytes.subarray(0, 4).toString('hex'),
-    bytes.subarray(4, 6).toString('hex'),
-    bytes.subarray(6, 8).toString('hex'),
-    bytes.subarray(8, 10).toString('hex'),
-    bytes.subarray(10, 16).toString('hex'),
-  ].join('-')
-}
+type SyncSource = typeof SYNC_SOURCES[number]
+type HermesSessionSummary = Awaited<ReturnType<typeof listHermesSessionSummaries>>[number]
 
 /**
  * Get all available profile names including 'default'
@@ -51,129 +44,176 @@ function getAllProfiles(): string[] {
   return profiles
 }
 
+function localSessionExists(profile: string, hermesSession: HermesSessionSummary, source: SyncSource): boolean {
+  const db = getDb()
+  if (!db) return false
+
+  const canonical = db.prepare('SELECT id FROM sessions WHERE id = ? LIMIT 1').get(hermesSession.id)
+  if (canonical) return true
+
+  // Older startup imports used a random local UUID. Detect those rows by their
+  // stable summary fingerprint so the first incremental sync does not create a
+  // duplicate copy under the canonical Hermes session id.
+  const legacy = db.prepare(`
+    SELECT id FROM sessions
+    WHERE profile = ?
+      AND source = ?
+      AND started_at = ?
+      AND last_active = ?
+      AND COALESCE(title, '') = ?
+      AND COALESCE(preview, '') = ?
+    LIMIT 1
+  `).get(
+    profile,
+    hermesSession.source || source,
+    hermesSession.started_at,
+    hermesSession.last_active,
+    hermesSession.title || '',
+    hermesSession.preview || '',
+  )
+
+  return Boolean(legacy)
+}
+
 /**
- * Sync api_server sessions from a single profile.
+ * Sync chat-visible sessions from a single profile.
  * Uses sessions-db.ts query logic to properly aggregate session chains.
  */
 async function syncProfileSessions(profile: string): Promise<{
   synced: number
+  skipped: number
   errors: string[]
 }> {
-  const result = { synced: 0, errors: [] as string[] }
-
-  try {
-    // Use listSessionSummaries to get aggregated session chains
-    // This returns only root sessions with aggregated stats from the entire chain
-    const summaries = await listHermesSessionSummaries('api_server', 10000, profile)
-
-    logger.info(`[session-sync] profile '${profile}': found ${summaries.length} aggregated session chains`)
-
-    for (const hermesSession of summaries) {
-      // Skip ephemeral sessions (created internally by chat-run-socket)
-      if (hermesSession.id.startsWith('eph_')) continue
-      try {
-        // Generate new session ID for local DB
-        const newSessionId = generateUuid()
-
-        // Create session in local DB
-        createSession({
-          id: newSessionId,
-          profile,
-          model: hermesSession.model,
-          title: hermesSession.title || undefined,
-        })
-
-        // Get full detail including all messages from the session chain
-        const { getSessionDetailFromDbWithProfile } = await import('../../db/hermes/sessions-db')
-        const detail = await getSessionDetailFromDbWithProfile(hermesSession.id, profile)
-
-        if (!detail || !detail.messages) {
-          result.errors.push(`session ${hermesSession.id}: failed to load messages`)
-          logger.warn(`[session-sync] failed to load messages for session ${hermesSession.id}`)
+  const result = { synced: 0, skipped: 0, errors: [] as string[] }
+
+  for (const source of SYNC_SOURCES) {
+    try {
+      // Use listSessionSummaries to get aggregated session chains.
+      // This returns only root sessions with aggregated stats from the entire chain.
+      const summaries = await listHermesSessionSummaries(source, 10000, profile)
+
+      logger.info(`[session-sync] profile '${profile}' source '${source}': found ${summaries.length} aggregated session chains`)
+
+      for (const hermesSession of summaries) {
+        // Skip ephemeral sessions created internally by chat-run-socket.
+        if (hermesSession.id.startsWith('eph_')) {
+          result.skipped++
           continue
         }
 
-        // Insert all messages from the entire chain
-        for (const msg of detail.messages) {
-          addMessage({
-            session_id: newSessionId,
-            role: msg.role,
-            content: msg.content,
-            tool_call_id: msg.tool_call_id,
-            tool_calls: msg.tool_calls,
-            tool_name: msg.tool_name,
-            timestamp: msg.timestamp,
-            token_count: msg.token_count,
-            finish_reason: msg.finish_reason,
-            reasoning: msg.reasoning,
-            reasoning_details: msg.reasoning_details,
-            reasoning_content: msg.reasoning_content,
-          })
+        if (localSessionExists(profile, hermesSession, source)) {
+          result.skipped++
+          continue
         }
 
-        // Update session with aggregated stats from Hermes
-        updateSession(newSessionId, {
-          started_at: hermesSession.started_at,
-          ended_at: hermesSession.ended_at,
-          end_reason: hermesSession.end_reason,
-          input_tokens: hermesSession.input_tokens,
-          output_tokens: hermesSession.output_tokens,
-          cache_read_tokens: hermesSession.cache_read_tokens,
-          cache_write_tokens: hermesSession.cache_write_tokens,
-          reasoning_tokens: hermesSession.reasoning_tokens,
-          estimated_cost_usd: hermesSession.estimated_cost_usd,
-          last_active: hermesSession.last_active,
-          preview: hermesSession.preview,
-        })
-
-        result.synced++
-        logger.info(`[session-sync] synced Hermes session ${hermesSession.id} -> ${newSessionId} (${detail.messages.length} messages, thread_session_count=${detail.thread_session_count})`)
-      } catch (err: any) {
-        result.errors.push(`session ${hermesSession.id}: ${err.message}`)
-        logger.warn(err, `[session-sync] failed to sync session ${hermesSession.id}`)
+        try {
+          // Get full detail including all messages from the session chain before
+          // creating the local row. That avoids partial imports when detail load fails.
+          const detail = await getSessionDetailFromDbWithProfile(hermesSession.id, profile)
+
+          if (!detail || !detail.messages) {
+            result.errors.push(`session ${hermesSession.id}: failed to load messages`)
+            logger.warn(`[session-sync] failed to load messages for session ${hermesSession.id}`)
+            continue
+          }
+
+          let createdLocalSession = false
+          try {
+            createSession({
+              id: hermesSession.id,
+              profile,
+              source: hermesSession.source || source,
+              model: hermesSession.model,
+              title: hermesSession.title || undefined,
+            })
+            createdLocalSession = true
+
+            addMessages(detail.messages.map(msg => ({
+              session_id: hermesSession.id,
+              role: msg.role,
+              content: msg.content,
+              tool_call_id: msg.tool_call_id,
+              tool_calls: msg.tool_calls,
+              tool_name: msg.tool_name,
+              timestamp: msg.timestamp,
+              token_count: msg.token_count,
+              finish_reason: msg.finish_reason,
+              reasoning: msg.reasoning,
+              reasoning_details: msg.reasoning_details,
+              reasoning_content: msg.reasoning_content,
+            })))
+
+            // Update session with aggregated stats from Hermes.
+            updateSession(hermesSession.id, {
+              source: hermesSession.source || source,
+              user_id: hermesSession.user_id,
+              started_at: hermesSession.started_at,
+              ended_at: hermesSession.ended_at,
+              end_reason: hermesSession.end_reason,
+              message_count: hermesSession.message_count,
+              tool_call_count: hermesSession.tool_call_count,
+              input_tokens: hermesSession.input_tokens,
+              output_tokens: hermesSession.output_tokens,
+              cache_read_tokens: hermesSession.cache_read_tokens,
+              cache_write_tokens: hermesSession.cache_write_tokens,
+              reasoning_tokens: hermesSession.reasoning_tokens,
+              billing_provider: hermesSession.billing_provider,
+              estimated_cost_usd: hermesSession.estimated_cost_usd,
+              actual_cost_usd: hermesSession.actual_cost_usd,
+              cost_status: hermesSession.cost_status,
+              preview: hermesSession.preview,
+              last_active: hermesSession.last_active,
+            })
+          } catch (err) {
+            if (createdLocalSession) {
+              deleteSession(hermesSession.id)
+            }
+            throw err
+          }
+
+          result.synced++
+          logger.info(`[session-sync] synced Hermes session ${hermesSession.id} (${detail.messages.length} messages, thread_session_count=${detail.thread_session_count})`)
+        } catch (err: any) {
+          result.errors.push(`session ${hermesSession.id}: ${err.message}`)
+          logger.warn(err, `[session-sync] failed to sync session ${hermesSession.id}`)
+        }
+      }
+    } catch (err: any) {
+      if (!err.message.includes('state.db not found')) {
+        result.errors.push(`${source}: ${err.message}`)
+        logger.warn(err, `[session-sync] failed to open state.db for profile '${profile}', source '${source}'`)
       }
-    }
-  } catch (err: any) {
-    if (!err.message.includes('state.db not found')) {
-      result.errors.push(err.message)
-      logger.warn(err, `[session-sync] failed to open state.db for profile '${profile}'`)
     }
   }
 
   return result
 }
 
 /**
- * Main entry point: sync all profiles on startup
- * Only runs if local DB is empty (first startup or after DB reset)
+ * Main entry point: sync all profiles on startup.
+ * Runs incrementally so sessions created outside the local WebUI DB are imported
+ * after every restart without duplicating sessions already mirrored locally.
  */
 export async function syncAllHermesSessionsOnStartup(): Promise<void> {
-  // Check if local DB has any sessions - only sync if completely empty
   const db = getDb()
   if (!db) {
     logger.info('[session-sync] SQLite not available, skipping Hermes sync')
     return
   }
 
-  const countResult = db.prepare('SELECT COUNT(*) as count FROM sessions').get() as { count: number } | undefined
-  const hasExistingSessions = countResult && countResult.count > 0
-
-  if (hasExistingSessions) {
-    logger.info('[session-sync] local DB has %d sessions, skipping Hermes sync', countResult!.count)
-    return
-  }
-
-  logger.info('[session-sync] local DB is empty, starting Hermes session sync...')
+  logger.info('[session-sync] starting incremental Hermes session sync...')
 
   const profiles = getAllProfiles()
   logger.info(`[session-sync] found ${profiles.length} profiles: ${profiles.join(', ')}`)
 
   let totalSynced = 0
+  let totalSkipped = 0
   let totalErrors = 0
 
   for (const profile of profiles) {
     const result = await syncProfileSessions(profile)
     totalSynced += result.synced
+    totalSkipped += result.skipped
     totalErrors += result.errors.length
 
     if (result.errors.length > 0) {
@@ -187,5 +227,5 @@ export async function syncAllHermesSessionsOnStartup(): Promise<void> {
     }
   }
 
-  logger.info(`[session-sync] sync complete: synced=${totalSynced}, errors=${totalErrors}`)
+  logger.info(`[session-sync] sync complete: synced=${totalSynced}, skipped=${totalSkipped}, errors=${totalErrors}`)
 }