openclaw-voc-skill/claude-code/claude-code-qiwe-assistant/mcp/src/core/agent-workbench-db.js

const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const { DatabaseSync } = require('node:sqlite');

const now = () => new Date().toISOString();
const makeId = prefix => `${prefix}_${crypto.randomUUID()}`;
const json = value => JSON.stringify(value ?? null);
const parse = (value, fallback) => {
  try { return value ? JSON.parse(value) : fallback; } catch { return fallback; }
};

const normalizeKeyPart = value => String(value || '')
  .trim()
  .toLowerCase()
  .replace(/[\s\p{P}\p{S}]+/gu, '_')
  .replace(/^_+|_+$/g, '')
  .slice(0, 160);

function evidenceItems(value, fallbackText = '', sourceMessageId = null) {
  const rows = Array.isArray(value) ? value : [];
  const items = rows.map(item => typeof item === 'string'
    ? { text: item, sourceMessageId: null }
    : {
        text: String(item?.text || item?.evidence || '').trim(),
        sourceMessageId: item?.sourceMessageId || item?.source_message_id || null,
        createdAt: item?.createdAt || item?.created_at || null,
      }).filter(item => item.text);
  const text = String(fallbackText || '').trim();
  if (text) items.push({ text, sourceMessageId: sourceMessageId || null, createdAt: now() });
  const unique = new Map();
  for (const item of items) {
    const key = `${item.sourceMessageId || ''}\u0000${item.text}`;
    if (!unique.has(key)) unique.set(key, item);
  }
  return [...unique.values()].slice(-30);
}

class AgentWorkbenchDb {
  constructor(filePath, defaults = {}) {
    fs.mkdirSync(path.dirname(filePath), { recursive: true });
    this.filePath = filePath;
    this.db = new DatabaseSync(filePath);
    this.db.exec('PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON; PRAGMA busy_timeout=5000;');
    this.init(defaults);
  }

  init(defaults) {
    this.db.exec(`
      CREATE TABLE IF NOT EXISTS settings (
        key TEXT PRIMARY KEY,
        value TEXT NOT NULL,
        updated_at TEXT NOT NULL
      );
      CREATE TABLE IF NOT EXISTS conversations (
        id TEXT PRIMARY KEY,
        contact_id TEXT NOT NULL UNIQUE,
        contact_name TEXT NOT NULL DEFAULT '',
        mode TEXT NOT NULL DEFAULT 'review' CHECK(mode IN ('review','auto','human','paused')),
        last_message_at TEXT,
        created_at TEXT NOT NULL,
        updated_at TEXT NOT NULL
      );
      CREATE TABLE IF NOT EXISTS messages (
        id TEXT PRIMARY KEY,
        conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
        external_id TEXT UNIQUE,
        direction TEXT NOT NULL CHECK(direction IN ('inbound','outbound')),
        sender_type TEXT NOT NULL CHECK(sender_type IN ('customer','agent','human','system')),
        content TEXT NOT NULL,
        status TEXT NOT NULL DEFAULT 'received',
        created_at TEXT NOT NULL,
        raw_json TEXT
      );
      CREATE TABLE IF NOT EXISTS customer_profiles (
        conversation_id TEXT PRIMARY KEY REFERENCES conversations(id) ON DELETE CASCADE,
        profile_json TEXT NOT NULL DEFAULT '{}',
        tags_json TEXT NOT NULL DEFAULT '[]',
        updated_at TEXT NOT NULL
      );
      CREATE TABLE IF NOT EXISTS customer_tasks (
        id TEXT PRIMARY KEY,
        conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
        fingerprint TEXT NOT NULL,
        type TEXT NOT NULL DEFAULT 'follow_up',
        title TEXT NOT NULL,
        owner TEXT NOT NULL DEFAULT '',
        due_at TEXT NOT NULL DEFAULT '',
        priority TEXT NOT NULL DEFAULT 'medium',
        status TEXT NOT NULL DEFAULT 'open' CHECK(status IN ('open','in_progress','done','dismissed')),
        reason TEXT NOT NULL DEFAULT '',
        evidence TEXT NOT NULL DEFAULT '',
        source_message_id TEXT,
        created_at TEXT NOT NULL,
        updated_at TEXT NOT NULL,
        UNIQUE(conversation_id, fingerprint)
      );
      CREATE TABLE IF NOT EXISTS customer_alerts (
        id TEXT PRIMARY KEY,
        conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
        fingerprint TEXT NOT NULL,
        type TEXT NOT NULL DEFAULT 'attention',
        severity TEXT NOT NULL DEFAULT 'medium',
        title TEXT NOT NULL,
        detail TEXT NOT NULL DEFAULT '',
        evidence TEXT NOT NULL DEFAULT '',
        recommended_action TEXT NOT NULL DEFAULT '',
        status TEXT NOT NULL DEFAULT 'open' CHECK(status IN ('open','acknowledged','resolved','dismissed')),
        source_message_id TEXT,
        created_at TEXT NOT NULL,
        updated_at TEXT NOT NULL,
        UNIQUE(conversation_id, fingerprint)
      );
      CREATE TABLE IF NOT EXISTS customer_recommendations (
        id TEXT PRIMARY KEY,
        conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
        property_id TEXT NOT NULL,
        property_snapshot_json TEXT NOT NULL DEFAULT '{}',
        sources_json TEXT NOT NULL DEFAULT '[]',
        status TEXT NOT NULL DEFAULT 'recommended' CHECK(status IN ('candidate','recommended','interested','rejected','viewing','viewed','closed')),
        feedback_reason TEXT NOT NULL DEFAULT '',
        recommend_count INTEGER NOT NULL DEFAULT 1,
        first_recommended_at TEXT NOT NULL,
        last_recommended_at TEXT NOT NULL,
        updated_at TEXT NOT NULL,
        UNIQUE(conversation_id, property_id)
      );
      CREATE TABLE IF NOT EXISTS drafts (
        id TEXT PRIMARY KEY,
        conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
        inbound_message_id TEXT NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
        content TEXT NOT NULL,
        status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending','approved','rejected','sent','failed')),
        confidence REAL NOT NULL DEFAULT 0,
        intent TEXT NOT NULL DEFAULT '',
        reason TEXT NOT NULL DEFAULT '',
        requires_human INTEGER NOT NULL DEFAULT 1,
        citations_json TEXT NOT NULL DEFAULT '[]',
        tool_trace_json TEXT NOT NULL DEFAULT '[]',
        created_at TEXT NOT NULL,
        reviewed_at TEXT,
        reviewer TEXT,
        error TEXT,
        sent_message_id TEXT
      );
      CREATE TABLE IF NOT EXISTS audit_logs (
        id TEXT PRIMARY KEY,
        actor TEXT NOT NULL,
        action TEXT NOT NULL,
        conversation_id TEXT,
        entity_id TEXT,
        detail_json TEXT NOT NULL DEFAULT '{}',
        created_at TEXT NOT NULL
      );
      CREATE TABLE IF NOT EXISTS poll_state (
        key TEXT PRIMARY KEY,
        value TEXT NOT NULL,
        updated_at TEXT NOT NULL
      );
      CREATE INDEX IF NOT EXISTS idx_messages_conversation ON messages(conversation_id, created_at);
      CREATE INDEX IF NOT EXISTS idx_messages_content ON messages(conversation_id, direction, content, created_at);
      CREATE INDEX IF NOT EXISTS idx_drafts_status ON drafts(status, created_at);
      CREATE INDEX IF NOT EXISTS idx_customer_tasks_conversation ON customer_tasks(conversation_id, status, updated_at DESC);
      CREATE INDEX IF NOT EXISTS idx_customer_alerts_conversation ON customer_alerts(conversation_id, status, updated_at DESC);
      CREATE INDEX IF NOT EXISTS idx_customer_recommendations_conversation ON customer_recommendations(conversation_id, status, last_recommended_at DESC);
      CREATE INDEX IF NOT EXISTS idx_audit_created ON audit_logs(created_at DESC);
    `);
    this.ensureColumn('customer_tasks', 'business_key', "business_key TEXT NOT NULL DEFAULT ''");
    this.ensureColumn('customer_tasks', 'managed_by', "managed_by TEXT NOT NULL DEFAULT 'agent'");
    this.ensureColumn('customer_tasks', 'evidence_json', "evidence_json TEXT NOT NULL DEFAULT '[]'");
    this.ensureColumn('customer_tasks', 'resolution_reason', "resolution_reason TEXT NOT NULL DEFAULT ''");
    this.ensureColumn('customer_tasks', 'official_todo_id', "official_todo_id TEXT NOT NULL DEFAULT ''");
    this.ensureColumn('customer_tasks', 'official_sync_status', "official_sync_status TEXT NOT NULL DEFAULT ''");
    this.ensureColumn('customer_tasks', 'official_synced_at', 'official_synced_at TEXT');
    this.ensureColumn('customer_alerts', 'business_key', "business_key TEXT NOT NULL DEFAULT ''");
    this.ensureColumn('customer_alerts', 'managed_by', "managed_by TEXT NOT NULL DEFAULT 'agent'");
    this.ensureColumn('customer_alerts', 'evidence_json', "evidence_json TEXT NOT NULL DEFAULT '[]'");
    this.ensureColumn('customer_alerts', 'resolution_reason', "resolution_reason TEXT NOT NULL DEFAULT ''");
    this.lastIntelligenceMigration = this.migrateCustomerIntelligenceRecords();
    this.db.exec(`
      CREATE UNIQUE INDEX IF NOT EXISTS idx_customer_tasks_business_key ON customer_tasks(conversation_id, business_key);
      CREATE UNIQUE INDEX IF NOT EXISTS idx_customer_alerts_business_key ON customer_alerts(conversation_id, business_key);
    `);
    this.setDefault('global_paused', defaults.globalPaused ? 'true' : 'false');
    this.setDefault('default_mode', defaults.defaultMode || 'review');
    this.setDefault('auto_send_confidence', String(defaults.autoSendConfidence ?? 0.88));
    this.setDefault('agent_cutover_at', defaults.cutoverAt || now());
  }

  close() { this.db.close(); }

  ensureColumn(table, column, definition) {
    const exists = this.db.prepare(`PRAGMA table_info(${table})`).all().some(item => item.name === column);
    if (!exists) this.db.exec(`ALTER TABLE ${table} ADD COLUMN ${definition}`);
  }

  customerTaskBusinessKey(item = {}) {
    const title = String(item.title || '').trim();
    const type = normalizeKeyPart(item.type || 'follow_up');
    if (type === 'qualification' && /用途.*(购置)?时间/.test(title)) return 'qualification:purpose_and_timeline';
    if (type === 'recommendation' && /(筛选|发送).*(重点|方案)/.test(title)) return 'recommendation:shortlist';
    const explicit = item.businessKey || item.business_key || item.taskKey || item.key;
    if (explicit) return String(explicit).split(':').map(normalizeKeyPart).filter(Boolean).join(':');
    return `${type}:${normalizeKeyPart(title || 'task')}`;
  }

  customerAlertBusinessKey(item = {}) {
    const title = String(item.title || '').trim();
    const type = normalizeKeyPart(item.type || 'attention');
    if (type === 'high_intent' && /核心需求.*成形/.test(title)) return 'high_intent:core_demand_ready';
    if (type === 'complaint') return 'complaint:manual_takeover';
    if (type === 'time_sensitive') return 'time_sensitive:follow_up';
    const explicit = item.businessKey || item.business_key || item.alertKey || item.key;
    if (explicit) return String(explicit).split(':').map(normalizeKeyPart).filter(Boolean).join(':');
    return `${type}:${normalizeKeyPart(title || 'alert')}`;
  }

  migrateCustomerIntelligenceRecords() {
    const migrate = ({ table, keyFor, terminalStatus }) => {
      const rows = this.db.prepare(`SELECT * FROM ${table} ORDER BY created_at,id`).all();
      const groups = new Map();
      for (const row of rows) {
        const businessKey = keyFor.call(this, row);
        const groupKey = `${row.conversation_id}\u0000${businessKey}`;
        if (!groups.has(groupKey)) groups.set(groupKey, { businessKey, rows: [] });
        groups.get(groupKey).rows.push(row);
      }
      let removed = 0;
      for (const group of groups.values()) {
        const statusRank = terminalStatus === 'done'
          ? { dismissed: 0, open: 1, in_progress: 2, done: 3 }
          : { dismissed: 0, open: 1, acknowledged: 2, resolved: 3 };
        const canonical = [...group.rows].sort((a, b) =>
          (statusRank[b.status] || 0) - (statusRank[a.status] || 0) ||
          Date.parse(a.created_at) - Date.parse(b.created_at))[0];
        const evidence = evidenceItems(
          group.rows.flatMap(row => evidenceItems(parse(row.evidence_json, []), row.evidence, row.source_message_id)),
        );
        const duplicates = group.rows.filter(row => row.id !== canonical.id);
        for (const row of duplicates) {
          this.db.prepare(`DELETE FROM ${table} WHERE id=?`).run(row.id);
          removed += 1;
        }
        const fingerprint = this.intelligenceFingerprint(group.businessKey);
        this.db.prepare(`UPDATE ${table} SET business_key=?,fingerprint=?,evidence_json=?,evidence=?,updated_at=? WHERE id=?`)
          .run(group.businessKey, fingerprint, json(evidence), evidence.at(-1)?.text || canonical.evidence || '', canonical.updated_at || now(), canonical.id);
      }
      return { rows: rows.length, removed };
    };
    this.db.exec('BEGIN IMMEDIATE');
    try {
      const tasks = migrate({ table: 'customer_tasks', keyFor: this.customerTaskBusinessKey, terminalStatus: 'done' });
      const alerts = migrate({ table: 'customer_alerts', keyFor: this.customerAlertBusinessKey, terminalStatus: 'resolved' });
      this.db.exec('COMMIT');
      return { tasks, alerts };
    } catch (error) {
      this.db.exec('ROLLBACK');
      throw error;
    }
  }

  setDefault(key, value) {
    this.db.prepare('INSERT OR IGNORE INTO settings(key,value,updated_at) VALUES(?,?,?)').run(key, value, now());
  }

  getSetting(key, fallback = '') {
    return this.db.prepare('SELECT value FROM settings WHERE key=?').get(key)?.value ?? fallback;
  }

  setSetting(key, value) {
    this.db.prepare(`INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)
      ON CONFLICT(key) DO UPDATE SET value=excluded.value,updated_at=excluded.updated_at`).run(key, String(value), now());
  }

  globalState() {
    return {
      paused: this.getSetting('global_paused', 'true') === 'true',
      defaultMode: this.getSetting('default_mode', 'review'),
      autoSendConfidence: Number(this.getSetting('auto_send_confidence', '0.88')),
    };
  }

  ensureConversation(contactId, contactName = '') {
    const existing = this.db.prepare('SELECT * FROM conversations WHERE contact_id=?').get(String(contactId));
    if (existing) {
      if (contactName && existing.contact_name !== contactName) {
        this.db.prepare('UPDATE conversations SET contact_name=?,updated_at=? WHERE id=?').run(contactName, now(), existing.id);
      }
      return this.getConversation(existing.id);
    }
    const conversationId = makeId('conv');
    const timestamp = now();
    this.db.prepare(`INSERT INTO conversations(id,contact_id,contact_name,mode,created_at,updated_at)
      VALUES(?,?,?,?,?,?)`).run(conversationId, String(contactId), contactName, this.getSetting('default_mode', 'review'), timestamp, timestamp);
    this.db.prepare('INSERT INTO customer_profiles(conversation_id,updated_at) VALUES(?,?)').run(conversationId, timestamp);
    return this.getConversation(conversationId);
  }

  getConversationByContactId(contactId) {
    return this.db.prepare('SELECT * FROM conversations WHERE contact_id=?').get(String(contactId)) || null;
  }

  getConversation(conversationId) {
    return this.db.prepare('SELECT * FROM conversations WHERE id=?').get(conversationId) || null;
  }

  listConversations() {
    return this.db.prepare(`SELECT c.*,
      (SELECT content FROM messages m WHERE m.conversation_id=c.id ORDER BY m.created_at DESC LIMIT 1) AS last_content,
      (SELECT COUNT(*) FROM drafts d WHERE d.conversation_id=c.id AND d.status='pending') AS pending_count
      FROM conversations c ORDER BY COALESCE(c.last_message_at,c.created_at) DESC`).all();
  }

  setConversationMode(conversationId, mode) {
    if (!['review', 'auto', 'human', 'paused'].includes(mode)) throw new Error('不支持的会话模式');
    const result = this.db.prepare('UPDATE conversations SET mode=?,updated_at=? WHERE id=?').run(mode, now(), conversationId);
    if (!result.changes) throw new Error('会话不存在');
    return this.getConversation(conversationId);
  }

  insertMessage({ conversationId, externalId = null, direction, senderType, content, status = 'received', createdAt = now(), raw = null }) {
    if (externalId) {
      const existing = this.db.prepare('SELECT * FROM messages WHERE external_id=?').get(externalId);
      if (existing) return { message: existing, created: false };
    }
    const messageId = makeId('msg');
    this.db.prepare(`INSERT INTO messages(id,conversation_id,external_id,direction,sender_type,content,status,created_at,raw_json)
      VALUES(?,?,?,?,?,?,?,?,?)`).run(messageId, conversationId, externalId, direction, senderType, String(content), status, createdAt, raw ? json(raw) : null);
    this.db.prepare('UPDATE conversations SET last_message_at=?,updated_at=? WHERE id=?').run(createdAt, now(), conversationId);
    return { message: this.getMessage(messageId), created: true };
  }

  getMessage(messageId) { return this.db.prepare('SELECT * FROM messages WHERE id=?').get(messageId) || null; }

  getLatestInbound(conversationId) {
    return this.db.prepare("SELECT * FROM messages WHERE conversation_id=? AND direction='inbound' ORDER BY created_at DESC LIMIT 1").get(conversationId) || null;
  }

  findRecentInboundDuplicate(conversationId, content, createdAt, windowSeconds = 60) {
    const rows = this.db.prepare("SELECT * FROM messages WHERE conversation_id=? AND direction='inbound' AND content=? ORDER BY created_at DESC LIMIT 20").all(conversationId, String(content));
    const timestamp = Date.parse(createdAt);
    return rows.find(row => Number.isFinite(timestamp) && Math.abs(timestamp - Date.parse(row.created_at)) <= windowSeconds * 1000) || null;
  }

  cleanupInboundContentDuplicates(windowSeconds = 60) {
    const rows = this.db.prepare("SELECT * FROM messages WHERE direction='inbound' ORDER BY conversation_id,created_at,id").all();
    const lastSeen = new Map();
    const duplicateIds = [];
    for (const row of rows) {
      const key = `${row.conversation_id}\u0000${row.content}`;
      const previous = lastSeen.get(key);
      const timestamp = Date.parse(row.created_at);
      if (previous && Number.isFinite(timestamp) && Math.abs(timestamp - previous.timestamp) <= windowSeconds * 1000) duplicateIds.push(row.id);
      else lastSeen.set(key, { timestamp, id: row.id });
    }
    const remove = this.db.prepare('DELETE FROM messages WHERE id=?');
    for (const messageId of duplicateIds) remove.run(messageId);
    return duplicateIds.length;
  }

  listMessages(conversationId, limit = 100) {
    return this.db.prepare(`SELECT * FROM (
      SELECT * FROM messages WHERE conversation_id=? ORDER BY created_at DESC LIMIT ?
    ) ORDER BY created_at ASC`).all(conversationId, limit);
  }

  deleteImportedMessages(conversationId, source = 'manual_sync') {
    const pattern = `%\"source\":\"${String(source).replace(/[\"%]/g, '')}\"%`;
    const result = this.db.prepare('DELETE FROM messages WHERE conversation_id=? AND raw_json LIKE ?').run(conversationId, pattern);
    const latest = this.db.prepare('SELECT MAX(created_at) AS value FROM messages WHERE conversation_id=?').get(conversationId)?.value || null;
    this.db.prepare('UPDATE conversations SET last_message_at=?,updated_at=? WHERE id=?').run(latest, now(), conversationId);
    return Number(result.changes || 0);
  }

  getProfile(conversationId) {
    const row = this.db.prepare('SELECT * FROM customer_profiles WHERE conversation_id=?').get(conversationId);
    return row ? { profile: parse(row.profile_json, {}), tags: parse(row.tags_json, []), updatedAt: row.updated_at } : { profile: {}, tags: [] };
  }

  updateProfile(conversationId, profile, tags = []) {
    this.db.prepare(`INSERT INTO customer_profiles(conversation_id,profile_json,tags_json,updated_at) VALUES(?,?,?,?)
      ON CONFLICT(conversation_id) DO UPDATE SET profile_json=excluded.profile_json,tags_json=excluded.tags_json,updated_at=excluded.updated_at`)
      .run(conversationId, json(profile || {}), json(tags || []), now());
    return this.getProfile(conversationId);
  }

  mergeProfileByContactId(contactId, patch = {}, tags) {
    const conversation = this.getConversationByContactId(contactId);
    if (!conversation) return null;
    const current = this.getProfile(conversation.id);
    return this.updateProfile(
      conversation.id,
      { ...current.profile, ...(patch || {}) },
      tags === undefined ? current.tags : tags,
    );
  }

  intelligenceFingerprint(...parts) {
    return crypto.createHash('sha256').update(parts.map(item => String(item || '').trim().toLowerCase()).join('\u0000')).digest('hex').slice(0, 24);
  }

  upsertCustomerTasks(conversationId, tasks = [], sourceMessageId = null) {
    const results = [];
    const find = this.db.prepare('SELECT * FROM customer_tasks WHERE conversation_id=? AND business_key=?');
    const insert = this.db.prepare(`INSERT INTO customer_tasks(id,conversation_id,fingerprint,business_key,managed_by,type,title,owner,due_at,priority,status,reason,evidence,evidence_json,source_message_id,created_at,updated_at)
      VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`);
    const update = this.db.prepare(`UPDATE customer_tasks SET managed_by=?,type=?,title=?,owner=CASE WHEN owner='' THEN ? ELSE owner END,due_at=CASE WHEN due_at='' THEN ? ELSE due_at END,priority=?,reason=?,evidence=?,evidence_json=?,source_message_id=COALESCE(?,source_message_id),updated_at=? WHERE id=?`);
    for (const item of tasks) {
      const title = String(item?.title || '').trim();
      if (!title) continue;
      const type = String(item.type || 'follow_up').trim();
      const evidence = String(item.evidence || '').trim();
      const businessKey = this.customerTaskBusinessKey(item);
      const fingerprint = this.intelligenceFingerprint(businessKey);
      const managedBy = String(item.managedBy || item.managed_by || 'agent').trim();
      const itemSourceMessageId = Object.prototype.hasOwnProperty.call(item, 'sourceMessageId')
        ? item.sourceMessageId
        : sourceMessageId;
      const existing = find.get(conversationId, businessKey);
      if (existing) {
        const evidences = evidenceItems(parse(existing.evidence_json, []), itemSourceMessageId ? evidence : '', itemSourceMessageId);
        update.run(managedBy, type, title, String(item.owner || '').trim().replace(/^待分配$/, ''), String(item.dueAt || item.due_at || '').trim(), String(item.priority || 'medium').trim(), String(item.reason || '').trim(), evidence || existing.evidence || '', json(evidences), itemSourceMessageId, now(), existing.id);
        results.push(this.db.prepare('SELECT * FROM customer_tasks WHERE id=?').get(existing.id));
      } else {
        const id = makeId('task');
        const evidences = evidenceItems([], evidence, itemSourceMessageId);
        insert.run(id, conversationId, fingerprint, businessKey, managedBy, type, title, String(item.owner || '').trim().replace(/^待分配$/, ''), String(item.dueAt || item.due_at || '').trim(), String(item.priority || 'medium').trim(), 'open', String(item.reason || '').trim(), evidence, json(evidences), itemSourceMessageId, now(), now());
        results.push(this.db.prepare('SELECT * FROM customer_tasks WHERE id=?').get(id));
      }
    }
    return results;
  }

  reconcileCustomerTasks(conversationId, tasks = [], sourceMessageId = null) {
    const results = this.upsertCustomerTasks(conversationId, tasks, sourceMessageId);
    const activeRuleKeys = new Set(tasks
      .filter(item => String(item.managedBy || item.managed_by || 'agent') === 'rule')
      .map(item => this.customerTaskBusinessKey(item)));
    const existingRules = this.db.prepare("SELECT * FROM customer_tasks WHERE conversation_id=? AND managed_by='rule' AND status IN ('open','in_progress')").all(conversationId);
    const resolved = [];
    for (const task of existingRules) {
      if (activeRuleKeys.has(task.business_key)) continue;
      this.db.prepare(`UPDATE customer_tasks SET status='done',resolution_reason='profile_condition_resolved',
        official_sync_status=CASE WHEN official_todo_id<>'' THEN 'completion_pending' ELSE official_sync_status END,updated_at=? WHERE id=?`).run(now(), task.id);
      resolved.push(task.id);
    }
    return { tasks: results, resolved };
  }

  listCustomerTasks(conversationId, limit = 100) {
    return this.db.prepare(`SELECT * FROM customer_tasks WHERE conversation_id=?
      ORDER BY CASE status WHEN 'open' THEN 0 WHEN 'in_progress' THEN 1 ELSE 2 END,
      CASE priority WHEN 'urgent' THEN 0 WHEN 'high' THEN 1 WHEN 'medium' THEN 2 ELSE 3 END, updated_at DESC LIMIT ?`).all(conversationId, limit);
  }

  getCustomerTask(taskId) {
    return this.db.prepare('SELECT * FROM customer_tasks WHERE id=?').get(taskId) || null;
  }

  completeCustomerTaskByBusinessKey(conversationId, businessKey, reason = 'business_action_completed') {
    const key = this.customerTaskBusinessKey({ businessKey });
    const task = this.db.prepare('SELECT * FROM customer_tasks WHERE conversation_id=? AND business_key=?').get(conversationId, key);
    if (!task || !['open', 'in_progress'].includes(task.status)) return task || null;
    this.db.prepare(`UPDATE customer_tasks SET status='done',resolution_reason=?,
      official_sync_status=CASE WHEN official_todo_id<>'' THEN 'completion_pending' ELSE official_sync_status END,updated_at=? WHERE id=?`).run(reason, now(), task.id);
    return this.getCustomerTask(task.id);
  }

  updateCustomerTask(taskId, fields = {}) {
    const allowed = ['status', 'owner', 'due_at', 'priority', 'reason', 'resolution_reason', 'official_todo_id', 'official_sync_status', 'official_synced_at'];
    const entries = Object.entries(fields).filter(([key]) => allowed.includes(key));
    if (entries.length) {
      const assignments = entries.map(([key]) => `${key}=?`).join(',');
      this.db.prepare(`UPDATE customer_tasks SET ${assignments},updated_at=? WHERE id=?`).run(...entries.map(([, value]) => value), now(), taskId);
    }
    return this.db.prepare('SELECT * FROM customer_tasks WHERE id=?').get(taskId) || null;
  }

  upsertCustomerAlerts(conversationId, alerts = [], sourceMessageId = null) {
    const results = [];
    const find = this.db.prepare('SELECT * FROM customer_alerts WHERE conversation_id=? AND business_key=?');
    const insert = this.db.prepare(`INSERT INTO customer_alerts(id,conversation_id,fingerprint,business_key,managed_by,type,severity,title,detail,evidence,evidence_json,recommended_action,status,source_message_id,created_at,updated_at)
      VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`);
    const update = this.db.prepare(`UPDATE customer_alerts SET managed_by=?,type=?,severity=?,title=?,detail=?,evidence=?,evidence_json=?,recommended_action=?,source_message_id=COALESCE(?,source_message_id),updated_at=? WHERE id=?`);
    for (const item of alerts) {
      const title = String(item?.title || '').trim();
      if (!title) continue;
      const type = String(item.type || 'attention').trim();
      const evidence = String(item.evidence || '').trim();
      const businessKey = this.customerAlertBusinessKey(item);
      const fingerprint = this.intelligenceFingerprint(businessKey);
      const managedBy = String(item.managedBy || item.managed_by || 'agent').trim();
      const itemSourceMessageId = Object.prototype.hasOwnProperty.call(item, 'sourceMessageId')
        ? item.sourceMessageId
        : sourceMessageId;
      const existing = find.get(conversationId, businessKey);
      if (existing) {
        const evidences = evidenceItems(parse(existing.evidence_json, []), itemSourceMessageId ? evidence : '', itemSourceMessageId);
        update.run(managedBy, type, String(item.severity || 'medium').trim(), title, String(item.detail || '').trim(), evidence || existing.evidence || '', json(evidences), String(item.recommendedAction || item.recommended_action || '').trim(), itemSourceMessageId, now(), existing.id);
        results.push(this.db.prepare('SELECT * FROM customer_alerts WHERE id=?').get(existing.id));
      } else {
        const id = makeId('alert');
        const evidences = evidenceItems([], evidence, itemSourceMessageId);
        insert.run(id, conversationId, fingerprint, businessKey, managedBy, type, String(item.severity || 'medium').trim(), title, String(item.detail || '').trim(), evidence, json(evidences), String(item.recommendedAction || item.recommended_action || '').trim(), 'open', itemSourceMessageId, now(), now());
        results.push(this.db.prepare('SELECT * FROM customer_alerts WHERE id=?').get(id));
      }
    }
    return results;
  }

  reconcileCustomerAlerts(conversationId, alerts = [], sourceMessageId = null) {
    const results = this.upsertCustomerAlerts(conversationId, alerts, sourceMessageId);
    const activeRuleKeys = new Set(alerts
      .filter(item => String(item.managedBy || item.managed_by || 'agent') === 'rule')
      .map(item => this.customerAlertBusinessKey(item)));
    const existingRules = this.db.prepare("SELECT * FROM customer_alerts WHERE conversation_id=? AND managed_by='rule' AND status IN ('open','acknowledged')").all(conversationId);
    const resolved = [];
    for (const alert of existingRules) {
      if (activeRuleKeys.has(alert.business_key)) continue;
      this.db.prepare("UPDATE customer_alerts SET status='resolved',resolution_reason='profile_condition_resolved',updated_at=? WHERE id=?").run(now(), alert.id);
      resolved.push(alert.id);
    }
    return { alerts: results, resolved };
  }

  listCustomerAlerts(conversationId, limit = 100) {
    return this.db.prepare(`SELECT * FROM customer_alerts WHERE conversation_id=?
      ORDER BY CASE status WHEN 'open' THEN 0 WHEN 'acknowledged' THEN 1 ELSE 2 END,
      CASE severity WHEN 'critical' THEN 0 WHEN 'high' THEN 1 WHEN 'medium' THEN 2 ELSE 3 END, updated_at DESC LIMIT ?`).all(conversationId, limit);
  }

  updateCustomerAlert(alertId, fields = {}) {
    const allowed = ['status', 'severity', 'recommended_action'];
    const entries = Object.entries(fields).filter(([key]) => allowed.includes(key));
    if (entries.length) {
      const assignments = entries.map(([key]) => `${key}=?`).join(',');
      this.db.prepare(`UPDATE customer_alerts SET ${assignments},updated_at=? WHERE id=?`).run(...entries.map(([, value]) => value), now(), alertId);
    }
    return this.db.prepare('SELECT * FROM customer_alerts WHERE id=?').get(alertId) || null;
  }

  hydrateCustomerRecommendation(row) {
    return row ? {
      ...row,
      property_snapshot: parse(row.property_snapshot_json, {}),
      sources: parse(row.sources_json, []),
    } : null;
  }

  upsertCustomerRecommendations(conversationId, items = [], source = {}) {
    const find = this.db.prepare('SELECT * FROM customer_recommendations WHERE conversation_id=? AND property_id=?');
    const insert = this.db.prepare(`INSERT INTO customer_recommendations(id,conversation_id,property_id,property_snapshot_json,sources_json,status,feedback_reason,recommend_count,first_recommended_at,last_recommended_at,updated_at)
      VALUES(?,?,?,?,?,?,?,?,?,?,?)`);
    const update = this.db.prepare(`UPDATE customer_recommendations SET property_snapshot_json=?,sources_json=?,status=CASE WHEN status IN ('interested','rejected','viewing','viewed','closed') THEN status ELSE ? END,recommend_count=?,last_recommended_at=?,updated_at=? WHERE id=?`);
    const results = [];
    for (const item of items) {
      const propertyId = String(item?.id || item?.propertyId || item?.property_id || '').trim();
      if (!propertyId) continue;
      const timestamp = String(source.createdAt || item.recommendedAt || now());
      const sourceItem = {
        type: String(source.type || 'unknown'),
        entityId: source.entityId || null,
        evidence: String(source.evidence || '').trim(),
        createdAt: timestamp,
      };
      const existing = find.get(conversationId, propertyId);
      const requestedStatus = ['candidate', 'recommended'].includes(String(source.status || '')) ? String(source.status) : 'recommended';
      if (!existing) {
        const id = makeId('recommendation');
        insert.run(id, conversationId, propertyId, json(item), json([sourceItem]), requestedStatus, '', 1, timestamp, timestamp, timestamp);
        results.push(this.hydrateCustomerRecommendation(this.db.prepare('SELECT * FROM customer_recommendations WHERE id=?').get(id)));
        continue;
      }
      const sources = parse(existing.sources_json, []);
      const sourceKey = `${sourceItem.type}\u0000${sourceItem.entityId || ''}`;
      const isNewSource = !sources.some(entry => `${entry.type || ''}\u0000${entry.entityId || ''}` === sourceKey);
      if (isNewSource) sources.push(sourceItem);
      update.run(json({ ...parse(existing.property_snapshot_json, {}), ...item }), json(sources.slice(-30)), requestedStatus, Number(existing.recommend_count || 0) + (isNewSource ? 1 : 0), isNewSource ? timestamp : existing.last_recommended_at, now(), existing.id);
      results.push(this.hydrateCustomerRecommendation(this.db.prepare('SELECT * FROM customer_recommendations WHERE id=?').get(existing.id)));
    }
    return results;
  }

  listCustomerRecommendations(conversationId, limit = 100) {
    return this.db.prepare(`SELECT * FROM customer_recommendations WHERE conversation_id=? ORDER BY last_recommended_at DESC LIMIT ?`).all(conversationId, limit).map(row => this.hydrateCustomerRecommendation(row));
  }

  updateCustomerRecommendation(conversationId, recommendationId, fields = {}) {
    const allowedStatuses = new Set(['candidate', 'recommended', 'interested', 'rejected', 'viewing', 'viewed', 'closed']);
    const status = String(fields.status || '');
    const entries = [];
    if (allowedStatuses.has(status)) entries.push(['status', status]);
    if (fields.feedback_reason !== undefined || fields.feedbackReason !== undefined) entries.push(['feedback_reason', String(fields.feedback_reason ?? fields.feedbackReason ?? '').trim()]);
    if (entries.length) {
      const assignments = entries.map(([key]) => `${key}=?`).join(',');
      this.db.prepare(`UPDATE customer_recommendations SET ${assignments},updated_at=? WHERE id=? AND conversation_id=?`).run(...entries.map(([, value]) => value), now(), recommendationId, conversationId);
    }
    return this.hydrateCustomerRecommendation(this.db.prepare('SELECT * FROM customer_recommendations WHERE id=? AND conversation_id=?').get(recommendationId, conversationId));
  }

  clearCustomerIntelligence() {
    const alerts = Number(this.db.prepare('DELETE FROM customer_alerts').run().changes || 0);
    const tasks = Number(this.db.prepare('DELETE FROM customer_tasks').run().changes || 0);
    return { tasks, alerts };
  }

  createDraft({ conversationId, inboundMessageId, content, confidence, intent, reason, requiresHuman, citations, toolTrace }) {
    const draftId = makeId('draft');
    this.db.prepare(`INSERT INTO drafts(id,conversation_id,inbound_message_id,content,confidence,intent,reason,requires_human,citations_json,tool_trace_json,created_at)
      VALUES(?,?,?,?,?,?,?,?,?,?,?)`).run(draftId, conversationId, inboundMessageId, String(content), Number(confidence) || 0, intent || '', reason || '', requiresHuman ? 1 : 0, json(citations || []), json(toolTrace || []), now());
    return this.getDraft(draftId);
  }

  getDraft(draftId) {
    const row = this.db.prepare('SELECT * FROM drafts WHERE id=?').get(draftId);
    return row ? this.hydrateDraft(row) : null;
  }

  hydrateDraft(row) {
    return { ...row, requires_human: Boolean(row.requires_human), citations: parse(row.citations_json, []), tool_trace: parse(row.tool_trace_json, []) };
  }

  listDrafts({ status = '', conversationId = '', limit = 100 } = {}) {
    let sql = 'SELECT * FROM drafts WHERE 1=1';
    const params = [];
    if (status) { sql += ' AND status=?'; params.push(status); }
    if (conversationId) { sql += ' AND conversation_id=?'; params.push(conversationId); }
    sql += ' ORDER BY created_at DESC LIMIT ?';
    params.push(limit);
    return this.db.prepare(sql).all(...params).map(row => this.hydrateDraft(row));
  }

  updateDraft(draftId, fields) {
    const allowed = ['content', 'status', 'reviewed_at', 'reviewer', 'error', 'sent_message_id'];
    const entries = Object.entries(fields).filter(([key]) => allowed.includes(key));
    if (!entries.length) return this.getDraft(draftId);
    const assignments = entries.map(([key]) => `${key}=?`).join(',');
    this.db.prepare(`UPDATE drafts SET ${assignments} WHERE id=?`).run(...entries.map(([, value]) => value), draftId);
    return this.getDraft(draftId);
  }

  audit({ actor = 'system', action, conversationId = null, entityId = null, detail = {} }) {
    const auditId = makeId('audit');
    this.db.prepare('INSERT INTO audit_logs(id,actor,action,conversation_id,entity_id,detail_json,created_at) VALUES(?,?,?,?,?,?,?)')
      .run(auditId, actor, action, conversationId, entityId, json(detail), now());
    return auditId;
  }

  listAudit(limit = 200, conversationId = '') {
    const rows = conversationId
      ? this.db.prepare('SELECT * FROM audit_logs WHERE conversation_id=? ORDER BY created_at DESC LIMIT ?').all(conversationId, limit)
      : this.db.prepare('SELECT * FROM audit_logs ORDER BY created_at DESC LIMIT ?').all(limit);
    return rows.map(row => ({ ...row, detail: parse(row.detail_json, {}) }));
  }

  latestAgentOutcome(conversationId) {
    const row = this.db.prepare(`SELECT * FROM audit_logs
      WHERE conversation_id=? AND action IN ('draft_created','agent_failed','agent_not_configured','agent_no_reply_needed')
      ORDER BY created_at DESC LIMIT 1`).get(conversationId);
    if (!row) return null;
    const detail = parse(row.detail_json, {});
    return {
      action: row.action,
      message: detail.message || (row.action === 'agent_no_reply_needed' ? '客户消息无需回复' : 'Agent 上游不可用'),
      entityId: row.entity_id,
      createdAt: row.created_at,
    };
  }

  latestAgentState(conversationId) {
    const outcome = this.latestAgentOutcome(conversationId);
    if (!outcome || ['draft_created', 'agent_no_reply_needed'].includes(outcome.action)) return null;
    return outcome;
  }

  getPollState(key, fallback = '') {
    return this.db.prepare('SELECT value FROM poll_state WHERE key=?').get(key)?.value ?? fallback;
  }

  setPollState(key, value) {
    this.db.prepare(`INSERT INTO poll_state(key,value,updated_at) VALUES(?,?,?)
      ON CONFLICT(key) DO UPDATE SET value=excluded.value,updated_at=excluded.updated_at`).run(key, String(value), now());
  }

  importCompatibleDatabase(sourcePath) {
    if (!sourcePath || !fs.existsSync(sourcePath) || path.resolve(sourcePath) === path.resolve(this.filePath)) return { imported: false, reason: 'source_missing' };
    if (this.listConversations().length) return { imported: false, reason: 'target_not_empty' };
    const source = new DatabaseSync(sourcePath, { readOnly: true });
    const tableOrder = ['settings', 'conversations', 'messages', 'customer_profiles', 'customer_tasks', 'customer_alerts', 'customer_recommendations', 'drafts', 'audit_logs', 'poll_state'];
    let rowsImported = 0;
    this.db.exec(`
      DROP INDEX IF EXISTS idx_customer_tasks_business_key;
      DROP INDEX IF EXISTS idx_customer_alerts_business_key;
    `);
    this.db.exec('BEGIN IMMEDIATE');
    try {
      for (const table of tableOrder) {
        const exists = source.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name=?").get(table);
        if (!exists) continue;
        const columns = source.prepare(`PRAGMA table_info(${table})`).all().map(item => item.name);
        if (!columns.length) continue;
        const placeholders = columns.map(() => '?').join(',');
        const insert = this.db.prepare(`INSERT OR IGNORE INTO ${table}(${columns.join(',')}) VALUES(${placeholders})`);
        for (const row of source.prepare(`SELECT ${columns.join(',')} FROM ${table}`).all()) {
          rowsImported += Number(insert.run(...columns.map(column => row[column])).changes || 0);
        }
      }
      this.db.exec('COMMIT');
    } catch (error) {
      this.db.exec('ROLLBACK');
      this.db.exec(`
        CREATE UNIQUE INDEX IF NOT EXISTS idx_customer_tasks_business_key ON customer_tasks(conversation_id, business_key);
        CREATE UNIQUE INDEX IF NOT EXISTS idx_customer_alerts_business_key ON customer_alerts(conversation_id, business_key);
      `);
      source.close();
      throw error;
    }
    source.close();
    const intelligenceMigration = this.migrateCustomerIntelligenceRecords();
    this.db.exec(`
      CREATE UNIQUE INDEX IF NOT EXISTS idx_customer_tasks_business_key ON customer_tasks(conversation_id, business_key);
      CREATE UNIQUE INDEX IF NOT EXISTS idx_customer_alerts_business_key ON customer_alerts(conversation_id, business_key);
    `);
    this.audit({ actor: 'migration', action: 'legacy_workbench_imported', detail: { source: path.basename(sourcePath), rowsImported, intelligenceMigration } });
    return { imported: true, rowsImported, intelligenceMigration };
  }
}

module.exports = { AgentWorkbenchDb };