修改后端接口 outbox 和 信息同步 worker

agent-core/src/db/database.ts

@@ -33,6 +33,29 @@ export function initDatabase(dbDir: string): Database.Database {
     );
 
     CREATE INDEX IF NOT EXISTS idx_llm_active ON llm_configs(is_active);
+
+    -- 同步队列（outbox）：本地变更等待推送到云端
+    CREATE TABLE IF NOT EXISTS sync_outbox (
+      id           TEXT PRIMARY KEY,
+      table_name   TEXT NOT NULL,          -- 变更的表名
+      record_id    TEXT NOT NULL,          -- 变更记录的主键 ID
+      operation    TEXT NOT NULL,          -- INSERT / UPDATE / DELETE
+      payload      TEXT NOT NULL,          -- 变更数据 JSON
+      status       TEXT DEFAULT 'pending', -- pending / syncing / done / failed
+      retry_count  INTEGER DEFAULT 0,      -- 重试次数
+      last_error   TEXT,                   -- 最后一次错误信息
+      created_at   TEXT DEFAULT (datetime('now')),
+      updated_at   TEXT DEFAULT (datetime('now'))
+    );
+
+    CREATE INDEX IF NOT EXISTS idx_outbox_status ON sync_outbox(status);
+    CREATE INDEX IF NOT EXISTS idx_outbox_table ON sync_outbox(table_name, record_id);
+
+    -- 同步元数据（如拉取游标）
+    CREATE TABLE IF NOT EXISTS sync_meta (
+      key   TEXT PRIMARY KEY,
+      value TEXT NOT NULL
+    );
   `)
 
   console.log(`[db] SQLite 已初始化: ${dbPath}`)

agent-core/src/index.ts

@@ -1,5 +1,6 @@
 import { initDatabase, closeDb } from './db/database'
 import { startWebSocketServer } from './transport/websocket'
+import { syncWorker } from './sync/worker'
 import { WorkflowEngine } from './workflow/engine'
 import { ToolRegistry, readFileTool } from './tools/registry'
 import type { AgentContext, LLMConfig, StreamCallbacks } from './types'
@@ -32,9 +33,13 @@ function main() {
   const wss = startWebSocketServer(WS_PORT)
   console.log(`[core] WebSocket 端口: ${WS_PORT}`)
 
+  // 3. 启动云端同步 Worker（本地变更 → outbox → 云端推送；云端变更 → 拉取合并）
+  syncWorker.start()
+
   // 优雅退出
   process.on('SIGINT', () => {
     console.log('\n[core] 正在关闭...')
+    syncWorker.stop()
     wss.close()
     closeDb()
     process.exit(0)

agent-core/src/llm/config.ts

@@ -1,5 +1,6 @@
 import { v4 as uuidv4 } from 'uuid'
 import { getDb } from '../db/database'
+import { outbox } from '../sync/outbox'
 import type { LLMConfig } from '../types'
 
 /**
@@ -38,7 +39,9 @@ export class LLMConfigManager {
          SET config = ?, updated_at = datetime('now')
          WHERE id = ?`
       ).run(JSON.stringify(config), existing.id)
-      return this.getById(existing.id)!
+      const record = this.getById(existing.id)!
+      outbox.enqueue('llm_configs', existing.id, 'UPDATE', record)
+      return record
     }
 
     // 新增
@@ -48,7 +51,9 @@ export class LLMConfigManager {
        VALUES (?, ?, ?, ?, 0)`
     ).run(id, name, provider, JSON.stringify(config))
 
-    return this.getById(id)!
+    const record = this.getById(id)!
+    outbox.enqueue('llm_configs', id, 'INSERT', record)
+    return record
   }
 
   /**
@@ -85,12 +90,22 @@ export class LLMConfigManager {
    */
   setActive(id: string): LLMConfigRecord | null {
     const db = getDb()
+    // 先找出当前激活的配置（需要写入 UPDATE outbox）
+    const prevActive = db.prepare("SELECT id FROM llm_configs WHERE is_active = 1 AND id != ?").get(id) as { id: string } | undefined
+
     const tx = db.transaction(() => {
       db.prepare('UPDATE llm_configs SET is_active = 0').run()
       db.prepare('UPDATE llm_configs SET is_active = 1 WHERE id = ?').run(id)
     })
     tx()
-    return this.getById(id)
+
+    const record = this.getById(id)
+    if (record) outbox.enqueue('llm_configs', id, 'UPDATE', record)
+    if (prevActive) {
+      const prevRecord = this.getById(prevActive.id)
+      if (prevRecord) outbox.enqueue('llm_configs', prevActive.id, 'UPDATE', prevRecord)
+    }
+    return record
   }
 
   /**
@@ -99,6 +114,9 @@ export class LLMConfigManager {
   delete(id: string): boolean {
     const db = getDb()
     const result = db.prepare('DELETE FROM llm_configs WHERE id = ?').run(id)
+    if (result.changes > 0) {
+      outbox.enqueue('llm_configs', id, 'DELETE', { id })
+    }
     return result.changes > 0
   }
 

agent-core/src/sync/cloud.ts

@@ -0,0 +1,116 @@
+import type { OutboxRecord } from './outbox'
+
+/**
+ * 云端同步客户端
+ *
+ * 云端地址通过环境变量配置：
+ *   CLOUD_SYNC_URL  - 云端同步服务地址（如 http://localhost:9000）
+ *   CLOUD_SYNC_TOKEN - 鉴权 Token（可选）
+ *
+ * 端口留空，等用户提供后通过环境变量传入即可。
+ */
+
+const CLOUD_URL = process.env.CLOUD_SYNC_URL || ''
+const CLOUD_TOKEN = process.env.CLOUD_SYNC_TOKEN || ''
+
+export interface CloudPushResult {
+  success: boolean
+  error?: string
+}
+
+export interface CloudPullResult {
+  /** 从云端拉取到的变更记录 */
+  changes: Array<{
+    table_name: string
+    record_id: string
+    operation: 'INSERT' | 'UPDATE' | 'DELETE'
+    payload: any
+    updated_at: string
+  }>
+  /** 拉取游标，下次拉取时传入以实现增量同步 */
+  cursor?: string
+}
+
+/**
+ * 推送一批 outbox 记录到云端
+ */
+export async function pushToCloud(records: OutboxRecord[]): Promise<Map<string, CloudPushResult>> {
+  const results = new Map<string, CloudPushResult>()
+
+  if (!CLOUD_URL) {
+    // 云端地址未配置，全部标记失败
+    for (const r of records) {
+      results.set(r.id, { success: false, error: 'CLOUD_SYNC_URL 未配置' })
+    }
+    return results
+  }
+
+  for (const r of records) {
+    try {
+      const res = await fetch(`${CLOUD_URL}/sync/push`, {
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+          ...(CLOUD_TOKEN ? { Authorization: `Bearer ${CLOUD_TOKEN}` } : {})
+        },
+        body: JSON.stringify({
+          table: r.table_name,
+          record_id: r.record_id,
+          operation: r.operation,
+          payload: JSON.parse(r.payload),
+          local_updated_at: r.updated_at
+        })
+      })
+
+      if (res.ok) {
+        results.set(r.id, { success: true })
+      } else {
+        const data = await res.json().catch(() => ({}))
+        results.set(r.id, {
+          success: false,
+          error: `HTTP ${res.status}: ${data?.message || res.statusText}`
+        })
+      }
+    } catch (e: any) {
+      results.set(r.id, { success: false, error: e?.message || String(e) })
+    }
+  }
+
+  return results
+}
+
+/**
+ * 从云端拉取变更记录（增量）
+ * @param cursor 上次拉取的游标，为空则全量拉取
+ */
+export async function pullFromCloud(cursor?: string): Promise<CloudPullResult> {
+  if (!CLOUD_URL) {
+    return { changes: [] }
+  }
+
+  try {
+    const url = new URL(`${CLOUD_URL}/sync/pull`)
+    if (cursor) url.searchParams.set('cursor', cursor)
+
+    const res = await fetch(url.toString(), {
+      method: 'GET',
+      headers: {
+        ...(CLOUD_TOKEN ? { Authorization: `Bearer ${CLOUD_TOKEN}` } : {})
+      }
+    })
+
+    if (!res.ok) {
+      console.error(`[cloud] 拉取失败: HTTP ${res.status}`)
+      return { changes: [] }
+    }
+
+    const data = await res.json()
+    return {
+      changes: data.changes || [],
+      cursor: data.cursor
+    }
+  } catch (e: any) {
+    console.error(`[cloud] 拉取异常: ${e?.message}`)
+    return { changes: [] }
+  }
+}

agent-core/src/sync/outbox.ts

@@ -0,0 +1,130 @@
+import { v4 as uuidv4 } from 'uuid'
+import { getDb } from '../db/database'
+
+export type SyncOperation = 'INSERT' | 'UPDATE' | 'DELETE'
+export type SyncStatus = 'pending' | 'syncing' | 'done' | 'failed'
+
+export interface OutboxRecord {
+  id: string
+  table_name: string
+  record_id: string
+  operation: SyncOperation
+  payload: string
+  status: SyncStatus
+  retry_count: number
+  last_error: string | null
+  created_at: string
+  updated_at: string
+}
+
+/**
+ * 同步队列（outbox）
+ * 本地数据变更后立即写入此表，由 SyncWorker 异步推送到云端。
+ * 用户读数据永远走本地 SQLite，不等待云端同步。
+ */
+export class Outbox {
+  /**
+   * 写入一条变更记录到同步队列
+   */
+  enqueue(tableName: string, recordId: string, operation: SyncOperation, payload: unknown): OutboxRecord {
+    const db = getDb()
+    const id = uuidv4()
+    db.prepare(
+      `INSERT INTO sync_outbox (id, table_name, record_id, operation, payload, status)
+       VALUES (?, ?, ?, ?, ?, 'pending')`
+    ).run(id, tableName, recordId, operation, JSON.stringify(payload))
+
+    return this.getById(id)!
+  }
+
+  /**
+   * 获取待同步的记录（pending 状态，按创建时间升序，最多 limit 条）
+   */
+  getPending(limit = 50): OutboxRecord[] {
+    const db = getDb()
+    const rows = db
+      .prepare(
+        `SELECT * FROM sync_outbox
+         WHERE status = 'pending'
+         ORDER BY created_at ASC
+         LIMIT ?`
+      )
+      .all(limit) as any[]
+    return rows.map((r) => this.mapRow(r))
+  }
+
+  /**
+   * 标记为同步中
+   */
+  markSyncing(id: string) {
+    const db = getDb()
+    db.prepare(
+      `UPDATE sync_outbox SET status = 'syncing', updated_at = datetime('now') WHERE id = ?`
+    ).run(id)
+  }
+
+  /**
+   * 标记为已同步
+   */
+  markDone(id: string) {
+    const db = getDb()
+    db.prepare(
+      `UPDATE sync_outbox SET status = 'done', updated_at = datetime('now') WHERE id = ?`
+    ).run(id)
+  }
+
+  /**
+   * 标记为失败（重试次数 +1，记录错误）
+   */
+  markFailed(id: string, error: string) {
+    const db = getDb()
+    db.prepare(
+      `UPDATE sync_outbox
+       SET status = 'failed', retry_count = retry_count + 1, last_error = ?, updated_at = datetime('now')
+       WHERE id = ?`
+    ).run(error, id)
+  }
+
+  /**
+   * 将失败的记录重新置为 pending（指数退避由 worker 控制调用时机）
+   */
+  resetToPending(id: string) {
+    const db = getDb()
+    db.prepare(
+      `UPDATE sync_outbox SET status = 'pending', updated_at = datetime('now') WHERE id = ?`
+    ).run(id)
+  }
+
+  /**
+   * 清理已同步的记录（保留最近 N 天）
+   */
+  cleanupDone(keepDays = 7) {
+    const db = getDb()
+    db.prepare(
+      `DELETE FROM sync_outbox WHERE status = 'done' AND created_at < datetime('now', ?)`
+    ).run(`-${keepDays} days`)
+  }
+
+  getById(id: string): OutboxRecord | null {
+    const db = getDb()
+    const row = db.prepare('SELECT * FROM sync_outbox WHERE id = ?').get(id) as any
+    return row ? this.mapRow(row) : null
+  }
+
+  private mapRow(row: any): OutboxRecord {
+    return {
+      id: row.id,
+      table_name: row.table_name,
+      record_id: row.record_id,
+      operation: row.operation,
+      payload: row.payload,
+      status: row.status,
+      retry_count: row.retry_count,
+      last_error: row.last_error,
+      created_at: row.created_at,
+      updated_at: row.updated_at
+    }
+  }
+}
+
+export const outbox = new Outbox()

agent-core/src/sync/worker.ts

@@ -0,0 +1,191 @@
+import { outbox } from './outbox'
+import { pushToCloud, pullFromCloud } from './cloud'
+import { getDb } from '../db/database'
+
+/**
+ * 同步 Worker
+ *
+ * 职责：
+ * 1. 推送：定期扫描 outbox 中 pending 记录 → 推送到云端
+ * 2. 拉取：定期从云端拉取其他设备的变更 → 合并到本地 SQLite
+ * 3. 重试：失败记录指数退避重试
+ *
+ * 用户读数据永远走本地 SQLite，同步完全异步。
+ */
+export class SyncWorker {
+  private pushTimer: NodeJS.Timeout | null = null
+  private pullTimer: NodeJS.Timeout | null = null
+  private running = false
+
+  // 配置（可通过环境变量调整）
+  private readonly pushInterval = Number(process.env.SYNC_PUSH_INTERVAL) || 5000      // 推送间隔 5s
+  private readonly pullInterval = Number(process.env.SYNC_PULL_INTERVAL) || 30000     // 拉取间隔 30s
+  private readonly maxRetry = Number(process.env.SYNC_MAX_RETRY) || 10                 // 最大重试次数
+  private readonly baseDelay = Number(process.env.SYNC_BASE_DELAY) || 1000             // 退避基数 1s
+  private readonly batchSize = Number(process.env.SYNC_BATCH_SIZE) || 20               // 每批推送条数
+
+  start() {
+    if (this.running) return
+    this.running = true
+    console.log('[sync] Worker 已启动')
+    console.log(`[sync] 推送间隔: ${this.pushInterval}ms, 拉取间隔: ${this.pullInterval}ms`)
+
+    // 启动时立即执行一次推送和拉取
+    this.pushLoop()
+    this.pullLoop()
+
+    this.pushTimer = setInterval(() => this.pushLoop(), this.pushInterval)
+    this.pullTimer = setInterval(() => this.pullLoop(), this.pullInterval)
+  }
+
+  stop() {
+    this.running = false
+    if (this.pushTimer) clearInterval(this.pushTimer)
+    if (this.pullTimer) clearInterval(this.pullTimer)
+    this.pushTimer = null
+    this.pullTimer = null
+    console.log('[sync] Worker 已停止')
+  }
+
+  /**
+   * 推送循环：扫描 outbox → 推送云端 → 更新状态
+   */
+  private async pushLoop() {
+    if (!this.running) return
+
+    try {
+      const records = outbox.getPending(this.batchSize)
+      if (records.length === 0) return
+
+      console.log(`[sync] 待推送 ${records.length} 条`)
+
+      // 先标记为 syncing
+      for (const r of records) {
+        outbox.markSyncing(r.id)
+      }
+
+      // 推送到云端
+      const results = await pushToCloud(records)
+
+      // 处理结果
+      for (const r of records) {
+        const result = results.get(r.id)
+        if (result?.success) {
+          outbox.markDone(r.id)
+          console.log(`[sync] ✓ 已推送: ${r.table_name}/${r.record_id} (${r.operation})`)
+        } else {
+          outbox.markFailed(r.id, result?.error || '未知错误')
+          console.warn(`[sync] ✗ 推送失败: ${r.table_name}/${r.record_id} - ${result?.error}`)
+
+          // 指数退避后重新入队
+          const retryCount = r.retry_count + 1
+          if (retryCount < this.maxRetry) {
+            const delay = Math.min(this.baseDelay * Math.pow(2, retryCount), 60000)
+            setTimeout(() => outbox.resetToPending(r.id), delay)
+          } else {
+            console.error(`[sync] 超过最大重试次数: ${r.table_name}/${r.record_id}`)
+          }
+        }
+      }
+
+      // 清理已同步的旧记录
+      outbox.cleanupDone(7)
+    } catch (e: any) {
+      console.error('[sync] 推送循环异常:', e?.message)
+    }
+  }
+
+  /**
+   * 拉取循环：从云端拉取变更 → 合并到本地 SQLite
+   */
+  private async pullLoop() {
+    if (!this.running) return
+
+    try {
+      const cursor = this.getPullCursor()
+      const result = await pullFromCloud(cursor)
+
+      if (result.changes.length === 0) return
+
+      console.log(`[sync] 拉取到 ${result.changes.length} 条云端变更`)
+
+      for (const change of result.changes) {
+        this.mergeChange(change)
+      }
+
+      if (result.cursor) {
+        this.savePullCursor(result.cursor)
+      }
+    } catch (e: any) {
+      console.error('[sync] 拉取循环异常:', e?.message)
+    }
+  }
+
+  /**
+   * 合并云端变更到本地 SQLite
+   * 以 updated_at 为依据：云端版本更新才覆盖本地
+   */
+  private mergeChange(change: {
+    table_name: string
+    record_id: string
+    operation: 'INSERT' | 'UPDATE' | 'DELETE'
+    payload: any
+    updated_at: string
+  }) {
+    const db = getDb()
+
+    // 检查本地记录的更新时间，只合并更新的版本
+    const local = db
+      .prepare(`SELECT updated_at FROM ${change.table_name} WHERE id = ?`)
+      .get(change.record_id) as { updated_at: string } | undefined
+
+    if (local && local.updated_at >= change.updated_at) {
+      // 本地版本更新或相同，跳过
+      return
+    }
+
+    try {
+      if (change.operation === 'DELETE') {
+        db.prepare(`DELETE FROM ${change.table_name} WHERE id = ?`).run(change.record_id)
+      } else {
+        // INSERT 或 UPDATE：用 UPSERT
+        const fields = Object.keys(change.payload)
+        const placeholders = fields.map(() => '?').join(', ')
+        const updateSet = fields.map((f) => `${f} = ?`).join(', ')
+        const values = fields.map((f) =>
+          typeof change.payload[f] === 'object' ? JSON.stringify(change.payload[f]) : change.payload[f]
+        )
+
+        db.prepare(
+          `INSERT INTO ${change.table_name} (id, ${fields.join(', ')})
+           VALUES (?, ${placeholders})
+           ON CONFLICT(id) DO UPDATE SET ${updateSet}`
+        ).run(change.record_id, ...values, ...values)
+      }
+      console.log(`[sync] 已合并云端变更: ${change.table_name}/${change.record_id} (${change.operation})`)
+    } catch (e: any) {
+      console.error(`[sync] 合并失败: ${change.table_name}/${change.record_id} - ${e?.message}`)
+    }
+  }
+
+  /**
+   * 拉取游标持久化（存在 SQLite 的 sync_meta 表中）
+   */
+  private getPullCursor(): string | undefined {
+    const db = getDb()
+    const row = db
+      .prepare("SELECT value FROM sync_meta WHERE key = 'pull_cursor'")
+      .get() as { value: string } | undefined
+    return row?.value
+  }
+
+  private savePullCursor(cursor: string) {
+    const db = getDb()
+    db.prepare(
+      `INSERT INTO sync_meta (key, value) VALUES ('pull_cursor', ?)
+       ON CONFLICT(key) DO UPDATE SET value = excluded.value`
+    ).run(cursor)
+  }
+}
+
+export const syncWorker = new SyncWorker()

agent-core/test-ws.mjs

@@ -1,62 +0,0 @@
-// 测试 agent-core WebSocket 通信
-import WebSocket from 'ws'
-
-const ws = new WebSocket('ws://localhost:8787')
-
-ws.on('open', () => {
-  console.log('[测试] 已连接')
-
-  // 测试 1: 保存 LLM 配置
-  ws.send(JSON.stringify({
-    type: 'llm:save',
-    data: {
-      name: '文心千帆测试',
-      provider: 'qianfan_v2',
-      config: {
-        baseUrl: 'https://qianfan.baidubce.com/v2/chat/completions',
-        authorization: 'Bearer test-token',
-        model: 'ernie-4.0',
-        temperature: 0.7
-      }
-    },
-    id: '1'
-  }))
-})
-
-ws.on('message', (data) => {
-  const msg = JSON.parse(data.toString())
-  console.log('[收到]', JSON.stringify(msg, null, 2))
-
-  // 收到保存响应后，测试获取列表
-  if (msg.type === 'llm:saved') {
-    ws.send(JSON.stringify({ type: 'llm:list', id: '2' }))
-  }
-
-  if (msg.type === 'llm:list') {
-    // 测试设置激活
-    if (msg.data && msg.data[0]) {
-      ws.send(JSON.stringify({ type: 'llm:setActive', data: { id: msg.data[0].id }, id: '3' }))
-    }
-  }
-
-  if (msg.type === 'llm:activeSet') {
-    // 测试获取激活
-    ws.send(JSON.stringify({ type: 'llm:getActive', id: '4' }))
-  }
-
-  if (msg.type === 'llm:active') {
-    console.log('\n=== 所有测试完成 ===')
-    ws.close()
-    process.exit(0)
-  }
-})
-
-ws.on('error', (err) => {
-  console.error('[错误]', err.message)
-  process.exit(1)
-})
-
-setTimeout(() => {
-  console.log('超时退出')
-  process.exit(1)
-}, 10000)