QIWEI-skill/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 CONVERSATION_MODES = ['review', 'auto', 'autopilot', 'human', 'paused'];
const PERSONAL_INTAKE_MODES = ['allowlist_only', 'auto_enroll_review', 'auto_enroll_autopilot'];
const WELCOME_SEND_MODES = ['draft', 'send'];
const WELCOME_STATES = ['skipped', 'draft_pending', 'send_pending', 'sending', 'sent', 'failed', 'delivery_unknown'];

const CONTACT_NAME_PLACEHOLDERS = new Set([
  '企微客户',
  '白名单企微联系人',
  '未知客户',
  '未命名客户',
]);

function normalizeContactName(value) {
  return String(value || '').trim().slice(0, 160);
}

function isMeaningfulContactName(value) {
  const name = normalizeContactName(value);
  if (!name || CONTACT_NAME_PLACEHOLDERS.has(name) || name.includes('\uFFFD')) return false;
  return !/^[?？\s\p{P}\p{S}]+$/u.test(name);
}

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','autopilot','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,
        content_type TEXT NOT NULL DEFAULT 'text',
        payload_json TEXT,
        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 customer_memory_items (
        id TEXT PRIMARY KEY,
        conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
        memory_key TEXT NOT NULL,
        type TEXT NOT NULL DEFAULT 'fact' CHECK(type IN ('fact','preference','constraint','event','hypothesis')),
        content TEXT NOT NULL,
        confidence REAL NOT NULL DEFAULT 1,
        importance REAL NOT NULL DEFAULT 0.5,
        status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active','superseded','rejected')),
        source_message_ids_json TEXT NOT NULL DEFAULT '[]',
        direction TEXT NOT NULL DEFAULT 'customer_to_agent',
        created_by TEXT NOT NULL DEFAULT 'rule',
        expires_at TEXT,
        created_at TEXT NOT NULL,
        updated_at TEXT NOT NULL,
        UNIQUE(conversation_id, memory_key)
      );
      CREATE TABLE IF NOT EXISTS customer_memory_snapshots (
        id TEXT PRIMARY KEY,
        conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
        version INTEGER NOT NULL,
        compact_text TEXT NOT NULL DEFAULT '',
        content_hash TEXT NOT NULL,
        created_at TEXT NOT NULL,
        UNIQUE(conversation_id, version)
      );
      CREATE TABLE IF NOT EXISTS memory_extraction_jobs (
        id TEXT PRIMARY KEY,
        conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
        message_id TEXT NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
        profile_updates_json TEXT NOT NULL DEFAULT '{}',
        status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending','processing','completed','failed')),
        attempts INTEGER NOT NULL DEFAULT 0,
        max_attempts INTEGER NOT NULL DEFAULT 3,
        next_attempt_at TEXT NOT NULL,
        error TEXT,
        result_json TEXT NOT NULL DEFAULT '{}',
        created_at TEXT NOT NULL,
        updated_at TEXT NOT NULL,
        completed_at TEXT,
        UNIQUE(conversation_id, message_id)
      );
      CREATE TABLE IF NOT EXISTS customer_memory_revisions (
        id TEXT PRIMARY KEY,
        memory_id TEXT NOT NULL REFERENCES customer_memory_items(id) ON DELETE CASCADE,
        conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
        memory_key TEXT NOT NULL,
        reason TEXT NOT NULL,
        previous_json TEXT NOT NULL,
        next_json TEXT NOT NULL,
        created_at TEXT NOT NULL
      );
      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 TABLE IF NOT EXISTS contact_onboarding (
        account_key TEXT NOT NULL,
        contact_id TEXT NOT NULL,
        conversation_id TEXT,
        first_message_id TEXT,
        contact_name TEXT NOT NULL DEFAULT '',
        source TEXT NOT NULL DEFAULT 'polling',
        policy TEXT NOT NULL,
        welcome_state TEXT NOT NULL DEFAULT 'skipped' CHECK(welcome_state IN ('skipped','draft_pending','send_pending','sending','sent','failed','delivery_unknown')),
        welcome_text TEXT NOT NULL DEFAULT '',
        welcome_draft_id TEXT,
        welcome_message_id TEXT,
        attempt_count INTEGER NOT NULL DEFAULT 0,
        last_error TEXT NOT NULL DEFAULT '',
        last_attempt_at TEXT,
        discovered_at TEXT NOT NULL,
        enrolled_at TEXT NOT NULL,
        sent_at TEXT,
        updated_at TEXT NOT NULL,
        PRIMARY KEY(account_key, contact_id)
      );
      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_customer_memory_conversation ON customer_memory_items(conversation_id, status, importance DESC, updated_at DESC);
      CREATE INDEX IF NOT EXISTS idx_customer_memory_snapshots ON customer_memory_snapshots(conversation_id, version DESC);
      CREATE INDEX IF NOT EXISTS idx_memory_extraction_jobs_pending ON memory_extraction_jobs(status, next_attempt_at, created_at);
      CREATE INDEX IF NOT EXISTS idx_customer_memory_revisions ON customer_memory_revisions(memory_id, created_at DESC);
      CREATE INDEX IF NOT EXISTS idx_audit_created ON audit_logs(created_at DESC);
      CREATE INDEX IF NOT EXISTS idx_contact_onboarding_state ON contact_onboarding(account_key, welcome_state, updated_at DESC);
    `);
    this.migrateConversationModes();
    this.ensureColumn('messages', 'content_type', "content_type TEXT NOT NULL DEFAULT 'text'");
    this.ensureColumn('messages', 'payload_json', 'payload_json TEXT');
    this.db.prepare(`UPDATE memory_extraction_jobs SET status='pending',next_attempt_at=?,updated_at=? WHERE status='processing'`)
      .run(now(), now());
    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', CONVERSATION_MODES.includes(defaults.defaultMode) ? defaults.defaultMode : 'review');
    this.setDefault('auto_send_confidence', String(defaults.autoSendConfidence ?? 0.88));
    this.setDefault('agent_cutover_at', defaults.cutoverAt || now());
    this.setDefault('personal_intake_mode', PERSONAL_INTAKE_MODES.includes(defaults.personalIntakeMode) ? defaults.personalIntakeMode : 'allowlist_only');
    this.setDefault('welcome_enabled', defaults.welcomeEnabled ? 'true' : 'false');
    this.setDefault('welcome_text', String(defaults.welcomeText || '您好，已经收到您的消息，我会尽快为您处理。'));
    this.setDefault('welcome_send_mode', WELCOME_SEND_MODES.includes(defaults.welcomeSendMode) ? defaults.welcomeSendMode : 'draft');
    this.setDefault('personal_intake_autopilot_confirmation', '');
    this.setDefault('personal_intake_autopilot_confirmed_at', '');
    this.setDefault('personal_intake_autopilot_confirmed_by', '');
    this.setDefault('auto_enrolled_contact_ids', '[]');
  }

  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());
  }

  migrateConversationModes() {
    const schema = this.db.prepare("SELECT sql FROM sqlite_master WHERE type='table' AND name='conversations'").get()?.sql || '';
    if (schema.includes("'autopilot'")) return;
    this.db.exec('PRAGMA foreign_keys=OFF;');
    try {
      this.db.exec(`
        BEGIN IMMEDIATE;
        CREATE TABLE conversations_mode_migration (
          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','autopilot','human','paused')),
          last_message_at TEXT,
          created_at TEXT NOT NULL,
          updated_at TEXT NOT NULL
        );
        INSERT INTO conversations_mode_migration(id,contact_id,contact_name,mode,last_message_at,created_at,updated_at)
          SELECT id,contact_id,contact_name,mode,last_message_at,created_at,updated_at FROM conversations;
        DROP TABLE conversations;
        ALTER TABLE conversations_mode_migration RENAME TO conversations;
        COMMIT;
      `);
    } catch (error) {
      try { this.db.exec('ROLLBACK;'); } catch {}
      throw error;
    } finally {
      this.db.exec('PRAGMA foreign_keys=ON;');
    }
  }

  intakePolicy() {
    const mode = this.getSetting('personal_intake_mode', 'allowlist_only');
    const welcomeSendMode = this.getSetting('welcome_send_mode', 'draft');
    return {
      mode: PERSONAL_INTAKE_MODES.includes(mode) ? mode : 'allowlist_only',
      welcomeEnabled: this.getSetting('welcome_enabled', 'false') === 'true',
      welcomeText: this.getSetting('welcome_text', ''),
      welcomeSendMode: WELCOME_SEND_MODES.includes(welcomeSendMode) ? welcomeSendMode : 'draft',
      autopilotConfirmation: this.getSetting('personal_intake_autopilot_confirmation', ''),
      autopilotConfirmedAt: this.getSetting('personal_intake_autopilot_confirmed_at', ''),
      autopilotConfirmedBy: this.getSetting('personal_intake_autopilot_confirmed_by', ''),
    };
  }

  setIntakePolicy(input = {}) {
    if (input.mode !== undefined) {
      if (!PERSONAL_INTAKE_MODES.includes(input.mode)) throw new Error('不支持的个人消息接入模式');
      this.setSetting('personal_intake_mode', input.mode);
    }
    if (input.welcomeEnabled !== undefined) this.setSetting('welcome_enabled', input.welcomeEnabled ? 'true' : 'false');
    if (input.welcomeText !== undefined) this.setSetting('welcome_text', String(input.welcomeText || '').trim());
    if (input.welcomeSendMode !== undefined) {
      if (!WELCOME_SEND_MODES.includes(input.welcomeSendMode)) throw new Error('不支持的欢迎语发送模式');
      this.setSetting('welcome_send_mode', input.welcomeSendMode);
    }
    if (input.autopilotConfirmation !== undefined) this.setSetting('personal_intake_autopilot_confirmation', input.autopilotConfirmation);
    if (input.autopilotConfirmedAt !== undefined) this.setSetting('personal_intake_autopilot_confirmed_at', input.autopilotConfirmedAt);
    if (input.autopilotConfirmedBy !== undefined) this.setSetting('personal_intake_autopilot_confirmed_by', input.autopilotConfirmedBy);
    return this.intakePolicy();
  }

  getAutoEnrolledContactIds() {
    const ids = parse(this.getSetting('auto_enrolled_contact_ids', '[]'), []);
    return [...new Set((Array.isArray(ids) ? ids : []).map(String).map(item => item.trim()).filter(Boolean))];
  }

  addAutoEnrolledContactId(contactId) {
    const ids = [...new Set([...this.getAutoEnrolledContactIds(), String(contactId || '').trim()].filter(Boolean))];
    this.setSetting('auto_enrolled_contact_ids', json(ids));
    return ids;
  }

  getOnboarding(accountKey, contactId) {
    return this.db.prepare('SELECT * FROM contact_onboarding WHERE account_key=? AND contact_id=?')
      .get(String(accountKey), String(contactId)) || null;
  }

  listOnboardings(accountKey, { states = [], limit = 100 } = {}) {
    const selectedStates = (Array.isArray(states) ? states : []).filter(state => WELCOME_STATES.includes(state));
    let sql = 'SELECT * FROM contact_onboarding WHERE account_key=?';
    const params = [String(accountKey)];
    if (selectedStates.length) {
      sql += ` AND welcome_state IN (${selectedStates.map(() => '?').join(',')})`;
      params.push(...selectedStates);
    }
    sql += ' ORDER BY updated_at DESC, contact_id ASC LIMIT ?';
    params.push(Math.max(1, Math.min(500, Number(limit) || 100)));
    return this.db.prepare(sql).all(...params);
  }

  upsertOnboarding(input = {}) {
    const accountKey = String(input.accountKey || '').trim();
    const contactId = String(input.contactId || '').trim();
    if (!accountKey || !contactId) throw new Error('欢迎语幂等键缺少账号或联系人');
    const timestamp = now();
    const state = WELCOME_STATES.includes(input.welcomeState) ? input.welcomeState : 'skipped';
    this.db.prepare(`INSERT INTO contact_onboarding(
      account_key,contact_id,conversation_id,first_message_id,contact_name,source,policy,welcome_state,welcome_text,
      welcome_draft_id,welcome_message_id,attempt_count,last_error,last_attempt_at,discovered_at,enrolled_at,sent_at,updated_at
    ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
    ON CONFLICT(account_key,contact_id) DO UPDATE SET
      conversation_id=COALESCE(excluded.conversation_id,contact_onboarding.conversation_id),
      first_message_id=COALESCE(contact_onboarding.first_message_id,excluded.first_message_id),
      contact_name=CASE WHEN excluded.contact_name<>'' THEN excluded.contact_name ELSE contact_onboarding.contact_name END,
      source=excluded.source,policy=excluded.policy,updated_at=excluded.updated_at`)
      .run(accountKey, contactId, input.conversationId || null, input.firstMessageId || null, String(input.contactName || ''), String(input.source || 'polling'), String(input.policy || 'allowlist_only'), state, String(input.welcomeText || ''), input.welcomeDraftId || null, input.welcomeMessageId || null, Number(input.attemptCount) || 0, String(input.lastError || ''), input.lastAttemptAt || null, input.discoveredAt || timestamp, input.enrolledAt || timestamp, input.sentAt || null, timestamp);
    return this.getOnboarding(accountKey, contactId);
  }

  updateOnboarding(accountKey, contactId, fields = {}) {
    const allowed = ['conversation_id', 'first_message_id', 'policy', 'welcome_state', 'welcome_text', 'welcome_draft_id', 'welcome_message_id', 'attempt_count', 'last_error', 'last_attempt_at', 'sent_at'];
    const entries = Object.entries(fields).filter(([key, value]) => allowed.includes(key) && value !== undefined);
    if (!entries.length) return this.getOnboarding(accountKey, contactId);
    if (fields.welcome_state !== undefined && !WELCOME_STATES.includes(fields.welcome_state)) throw new Error('不支持的欢迎语状态');
    const assignments = [...entries.map(([key]) => `${key}=?`), 'updated_at=?'].join(',');
    this.db.prepare(`UPDATE contact_onboarding SET ${assignments} WHERE account_key=? AND contact_id=?`)
      .run(...entries.map(([, value]) => value), now(), String(accountKey), String(contactId));
    return this.getOnboarding(accountKey, contactId);
  }

  claimWelcomeSend(accountKey, contactId) {
    const timestamp = now();
    const result = this.db.prepare(`UPDATE contact_onboarding SET welcome_state='sending',attempt_count=attempt_count+1,
      last_error='',last_attempt_at=?,updated_at=? WHERE account_key=? AND contact_id=? AND welcome_state IN ('send_pending','failed')`)
      .run(timestamp, timestamp, String(accountKey), String(contactId));
    return Number(result.changes || 0) === 1 ? this.getOnboarding(accountKey, contactId) : null;
  }

  markWelcomeSent(accountKey, contactId, messageId) {
    const timestamp = now();
    this.db.prepare(`UPDATE contact_onboarding SET welcome_state='sent',welcome_message_id=?,last_error='',sent_at=?,updated_at=?
      WHERE account_key=? AND contact_id=? AND welcome_state='sending'`)
      .run(messageId || null, timestamp, timestamp, String(accountKey), String(contactId));
    return this.getOnboarding(accountKey, contactId);
  }

  markWelcomeFailed(accountKey, contactId, error) {
    this.db.prepare(`UPDATE contact_onboarding SET welcome_state='failed',last_error=?,updated_at=?
      WHERE account_key=? AND contact_id=? AND welcome_state='sending'`)
      .run(String(error || '欢迎语发送失败').slice(0, 500), now(), String(accountKey), String(contactId));
    return this.getOnboarding(accountKey, contactId);
  }

  markStaleWelcomeSending(accountKey, cutoff) {
    return Number(this.db.prepare(`UPDATE contact_onboarding SET welcome_state='delivery_unknown',
      last_error='发送结果待人工确认',updated_at=? WHERE account_key=? AND welcome_state='sending'
      AND COALESCE(last_attempt_at,'')<>'' AND last_attempt_at<?`)
      .run(now(), String(accountKey), String(cutoff)).changes || 0);
  }

  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));
    const normalizedName = normalizeContactName(contactName);
    const safeName = isMeaningfulContactName(normalizedName) ? normalizedName : '';
    if (existing) {
      if (safeName && existing.contact_name !== safeName) {
        this.db.prepare('UPDATE conversations SET contact_name=?,updated_at=? WHERE id=?').run(safeName, now(), existing.id);
      }
      return this.getConversation(existing.id);
    }
    const conversationId = makeId('conv');
    const timestamp = now();
    const configuredMode = this.getSetting('default_mode', 'review');
    const defaultMode = this.getSetting('global_paused', 'true') === 'true'
      ? 'paused'
      : CONVERSATION_MODES.includes(configuredMode) ? configuredMode : 'review';
    this.db.prepare(`INSERT INTO conversations(id,contact_id,contact_name,mode,created_at,updated_at)
      VALUES(?,?,?,?,?,?)`).run(conversationId, String(contactId), safeName, defaultMode, 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 CASE WHEN NULLIF(TRIM(c.last_message_at),'') IS NULL THEN 1 ELSE 0 END,
        c.last_message_at DESC,c.created_at DESC,c.id ASC`).all();
  }

  setConversationMode(conversationId, mode) {
    if (!CONVERSATION_MODES.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);
  }

  setAllConversationModes(mode) {
    if (!CONVERSATION_MODES.includes(mode)) throw new Error('Unsupported conversation mode');
    const result = this.db.prepare('UPDATE conversations SET mode=?,updated_at=? WHERE mode<>?').run(mode, now(), mode);
    return Number(result.changes || 0);
  }

  insertMessage({ conversationId, externalId = null, direction, senderType, content, contentType = 'text', payload = null, 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,content_type,payload_json,status,created_at,raw_json)
      VALUES(?,?,?,?,?,?,?,?,?,?,?)`).run(messageId, conversationId, externalId, direction, senderType, String(content), String(contentType || 'text'), payload ? json(payload) : null, 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; }

  updateMessageRaw(messageId, raw) {
    const result = this.db.prepare('UPDATE messages SET raw_json=? WHERE id=?').run(json(raw || {}), messageId);
    if (!result.changes) throw new Error('消息不存在');
    return this.getMessage(messageId);
  }

  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);
  }

  upsertCustomerMemory(conversationId, item = {}) {
    const memoryKey = String(item.memoryKey || item.memory_key || '').trim().slice(0, 240);
    const content = String(item.content || '').trim().slice(0, 1200);
    if (!memoryKey || !content) return null;
    const type = ['fact', 'preference', 'constraint', 'event', 'hypothesis'].includes(item.type) ? item.type : 'fact';
    const sourceMessageIds = [...new Set((item.sourceMessageIds || item.source_message_ids || []).map(String).filter(Boolean))].slice(-20);
    const existing = this.db.prepare('SELECT * FROM customer_memory_items WHERE conversation_id=? AND memory_key=?').get(conversationId, memoryKey);
    const timestamp = now();
    if (existing) {
      const mergedSources = [...new Set([...parse(existing.source_message_ids_json, []), ...sourceMessageIds])].slice(-20);
      const nextState = {
        ...existing,
        type,
        content,
        confidence: Math.max(0, Math.min(1, Number(item.confidence ?? existing.confidence ?? 1))),
        importance: Math.max(0, Math.min(1, Number(item.importance ?? existing.importance ?? 0.5))),
        status: 'active',
        source_message_ids_json: json(mergedSources),
        direction: String(item.direction || existing.direction || 'customer_to_agent'),
        created_by: String(item.createdBy || item.created_by || existing.created_by || 'rule'),
        expires_at: item.expiresAt || item.expires_at || null,
      };
      if (['type', 'content', 'confidence', 'importance', 'status', 'source_message_ids_json', 'expires_at'].some(key => String(existing[key] ?? '') !== String(nextState[key] ?? ''))) {
        this.recordCustomerMemoryRevision(existing, nextState, item.revisionReason || (existing.content !== content ? 'superseded_by_new_evidence' : 'evidence_or_metadata_updated'));
      }
      this.db.prepare(`UPDATE customer_memory_items SET type=?,content=?,confidence=?,importance=?,status='active',
        source_message_ids_json=?,direction=?,created_by=?,expires_at=?,updated_at=? WHERE id=?`)
        .run(nextState.type, nextState.content, nextState.confidence, nextState.importance, nextState.source_message_ids_json,
          nextState.direction, nextState.created_by, nextState.expires_at, timestamp, existing.id);
      return this.getCustomerMemory(existing.id);
    }
    const id = makeId('memory');
    this.db.prepare(`INSERT INTO customer_memory_items(id,conversation_id,memory_key,type,content,confidence,importance,status,
      source_message_ids_json,direction,created_by,expires_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`)
      .run(id, conversationId, memoryKey, type, content, Math.max(0, Math.min(1, Number(item.confidence ?? 1))),
        Math.max(0, Math.min(1, Number(item.importance ?? 0.5))), 'active', json(sourceMessageIds),
        String(item.direction || 'customer_to_agent'), String(item.createdBy || item.created_by || 'rule'),
        item.expiresAt || item.expires_at || null, timestamp, timestamp);
    return this.getCustomerMemory(id);
  }

  getCustomerMemory(memoryId) {
    const row = this.db.prepare('SELECT * FROM customer_memory_items WHERE id=?').get(memoryId);
    return row ? { ...row, source_message_ids: parse(row.source_message_ids_json, []) } : null;
  }

  getCustomerMemoryByKey(conversationId, memoryKey) {
    const row = this.db.prepare('SELECT * FROM customer_memory_items WHERE conversation_id=? AND memory_key=?').get(conversationId, String(memoryKey || ''));
    return row ? { ...row, source_message_ids: parse(row.source_message_ids_json, []) } : null;
  }

  listCustomerMemories(conversationId, { status = 'active', limit = 100 } = {}) {
    const rows = status
      ? this.db.prepare('SELECT * FROM customer_memory_items WHERE conversation_id=? AND status=? ORDER BY importance DESC,updated_at DESC LIMIT ?').all(conversationId, status, limit)
      : this.db.prepare('SELECT * FROM customer_memory_items WHERE conversation_id=? ORDER BY importance DESC,updated_at DESC LIMIT ?').all(conversationId, limit);
    return rows.map(row => ({ ...row, source_message_ids: parse(row.source_message_ids_json, []) }));
  }

  updateCustomerMemory(memoryId, fields = {}) {
    const current = this.getCustomerMemory(memoryId);
    if (!current) throw new Error('客户记忆不存在');
    const allowedTypes = new Set(['fact', 'preference', 'constraint', 'event', 'hypothesis']);
    const allowedStatuses = new Set(['active', 'superseded', 'rejected']);
    const updates = {
      content: fields.content === undefined ? current.content : String(fields.content || '').trim().slice(0, 1200),
      type: fields.type === undefined ? current.type : String(fields.type),
      status: fields.status === undefined ? current.status : String(fields.status),
      confidence: fields.confidence === undefined ? current.confidence : Math.max(0, Math.min(1, Number(fields.confidence))),
      importance: fields.importance === undefined ? current.importance : Math.max(0, Math.min(1, Number(fields.importance))),
      expiresAt: fields.expiresAt === undefined && fields.expires_at === undefined ? current.expires_at : (fields.expiresAt || fields.expires_at || null),
      createdBy: fields.createdBy === undefined && fields.created_by === undefined ? current.created_by : String(fields.createdBy || fields.created_by || 'human'),
    };
    if (!updates.content) throw new Error('客户记忆内容不能为空');
    if (!allowedTypes.has(updates.type)) throw new Error('不支持的客户记忆类型');
    if (!allowedStatuses.has(updates.status)) throw new Error('不支持的客户记忆状态');
    const nextState = {
      ...current,
      content: updates.content,
      type: updates.type,
      status: updates.status,
      confidence: updates.confidence,
      importance: updates.importance,
      expires_at: updates.expiresAt,
      created_by: updates.createdBy,
    };
    if (['content', 'type', 'status', 'confidence', 'importance', 'expires_at', 'created_by'].some(key => String(current[key] ?? '') !== String(nextState[key] ?? ''))) {
      this.recordCustomerMemoryRevision(current, nextState, fields.revisionReason || 'memory_updated');
    }
    this.db.prepare(`UPDATE customer_memory_items SET content=?,type=?,status=?,confidence=?,importance=?,expires_at=?,created_by=?,updated_at=? WHERE id=?`)
      .run(updates.content, updates.type, updates.status, updates.confidence, updates.importance, updates.expiresAt, updates.createdBy, now(), memoryId);
    return this.getCustomerMemory(memoryId);
  }

  forgetCustomerMemory(memoryId) {
    const current = this.getCustomerMemory(memoryId);
    if (!current) throw new Error('客户记忆不存在');
    this.db.prepare('DELETE FROM customer_memory_items WHERE id=?').run(memoryId);
    return current;
  }

  expireCustomerMemories(conversationId, timestamp = now()) {
    const expired = this.db.prepare(`SELECT * FROM customer_memory_items
      WHERE conversation_id=? AND status='active' AND expires_at IS NOT NULL AND expires_at<>'' AND expires_at<=?`)
      .all(conversationId, timestamp);
    for (const row of expired) {
      this.recordCustomerMemoryRevision(row, { ...row, status: 'superseded' }, 'expired');
    }
    if (expired.length) {
      this.db.prepare(`UPDATE customer_memory_items SET status='superseded',updated_at=?
        WHERE conversation_id=? AND status='active' AND expires_at IS NOT NULL AND expires_at<>'' AND expires_at<=?`)
        .run(timestamp, conversationId, timestamp);
    }
    return expired.length;
  }

  recordCustomerMemoryRevision(previous, next, reason = 'memory_updated') {
    if (!previous?.id) return null;
    const id = makeId('memory_revision');
    this.db.prepare(`INSERT INTO customer_memory_revisions(id,memory_id,conversation_id,memory_key,reason,previous_json,next_json,created_at)
      VALUES(?,?,?,?,?,?,?,?)`).run(id, previous.id, previous.conversation_id, previous.memory_key, String(reason || 'memory_updated'),
      json(previous), json(next || {}), now());
    return id;
  }

  listCustomerMemoryRevisions(memoryId, limit = 100) {
    return this.db.prepare('SELECT * FROM customer_memory_revisions WHERE memory_id=? ORDER BY created_at DESC LIMIT ?')
      .all(memoryId, limit).map(row => ({ ...row, previous: parse(row.previous_json, {}), next: parse(row.next_json, {}) }));
  }

  latestMemorySnapshot(conversationId) {
    return this.db.prepare('SELECT * FROM customer_memory_snapshots WHERE conversation_id=? ORDER BY version DESC LIMIT 1').get(conversationId) || null;
  }

  ensureMemorySnapshot(conversationId, compactText) {
    const text = String(compactText || '').trim();
    const contentHash = crypto.createHash('sha256').update(text).digest('hex');
    const latest = this.latestMemorySnapshot(conversationId);
    if (latest?.content_hash === contentHash) return latest;
    const version = Number(latest?.version || 0) + 1;
    const id = makeId('memory_snapshot');
    this.db.prepare('INSERT INTO customer_memory_snapshots(id,conversation_id,version,compact_text,content_hash,created_at) VALUES(?,?,?,?,?,?)')
      .run(id, conversationId, version, text, contentHash, now());
    return this.latestMemorySnapshot(conversationId);
  }

  enqueueMemoryExtraction({ conversationId, messageId, profileUpdates = {}, maxAttempts = 3 }) {
    const timestamp = now();
    const id = makeId('memory_job');
    this.db.prepare(`INSERT INTO memory_extraction_jobs(id,conversation_id,message_id,profile_updates_json,status,attempts,max_attempts,
      next_attempt_at,error,result_json,created_at,updated_at) VALUES(?,?,?,?, 'pending',0,?,?,?,?,?,?)
      ON CONFLICT(conversation_id,message_id) DO UPDATE SET profile_updates_json=excluded.profile_updates_json,
        status=CASE WHEN memory_extraction_jobs.status='completed' THEN 'completed' ELSE 'pending' END,
        next_attempt_at=excluded.next_attempt_at,error=NULL,updated_at=excluded.updated_at`)
      .run(id, conversationId, messageId, json(profileUpdates || {}), Math.max(1, Math.min(10, Number(maxAttempts) || 3)),
        timestamp, null, json({}), timestamp, timestamp);
    return this.db.prepare('SELECT * FROM memory_extraction_jobs WHERE conversation_id=? AND message_id=?').get(conversationId, messageId);
  }

  claimMemoryExtractionJob(timestamp = now()) {
    this.db.exec('BEGIN IMMEDIATE');
    try {
      const row = this.db.prepare(`SELECT * FROM memory_extraction_jobs WHERE status='pending' AND next_attempt_at<=?
        ORDER BY created_at ASC LIMIT 1`).get(timestamp);
      if (!row) {
        this.db.exec('COMMIT');
        return null;
      }
      this.db.prepare(`UPDATE memory_extraction_jobs SET status='processing',attempts=attempts+1,updated_at=? WHERE id=?`)
        .run(timestamp, row.id);
      this.db.exec('COMMIT');
      const claimed = this.db.prepare('SELECT * FROM memory_extraction_jobs WHERE id=?').get(row.id);
      return { ...claimed, profileUpdates: parse(claimed.profile_updates_json, {}) };
    } catch (error) {
      this.db.exec('ROLLBACK');
      throw error;
    }
  }

  completeMemoryExtractionJob(jobId, result = {}) {
    const timestamp = now();
    this.db.prepare(`UPDATE memory_extraction_jobs SET status='completed',result_json=?,error=NULL,completed_at=?,updated_at=? WHERE id=?`)
      .run(json(result || {}), timestamp, timestamp, jobId);
    return this.getMemoryExtractionJob(jobId);
  }

  failMemoryExtractionJob(jobId, error, retryDelayMs = 1000) {
    const current = this.getMemoryExtractionJob(jobId);
    if (!current) return null;
    const terminal = Number(current.attempts || 0) >= Number(current.max_attempts || 3);
    const timestamp = now();
    const nextAttemptAt = new Date(Date.now() + Math.max(100, Number(retryDelayMs) || 1000)).toISOString();
    this.db.prepare(`UPDATE memory_extraction_jobs SET status=?,next_attempt_at=?,error=?,updated_at=? WHERE id=?`)
      .run(terminal ? 'failed' : 'pending', nextAttemptAt, String(error?.message || error || '').slice(0, 500), timestamp, jobId);
    return this.getMemoryExtractionJob(jobId);
  }

  getMemoryExtractionJob(jobId) {
    const row = this.db.prepare('SELECT * FROM memory_extraction_jobs WHERE id=?').get(jobId);
    return row ? { ...row, profileUpdates: parse(row.profile_updates_json, {}), result: parse(row.result_json, {}) } : null;
  }

  listMemoryExtractionJobs({ status = '', limit = 100 } = {}) {
    const rows = status
      ? this.db.prepare('SELECT * FROM memory_extraction_jobs WHERE status=? ORDER BY created_at DESC LIMIT ?').all(status, limit)
      : this.db.prepare('SELECT * FROM memory_extraction_jobs ORDER BY created_at DESC LIMIT ?').all(limit);
    return rows.map(row => ({ ...row, profileUpdates: parse(row.profile_updates_json, {}), result: parse(row.result_json, {}) }));
  }

  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','autopilot_message_sent','autopilot_send_failed','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'
        ? '客户消息无需回复'
        : row.action === 'autopilot_message_sent' ? '全自动接管已直接发送 Agent 回复' : 'Agent 上游不可用'),
      entityId: row.entity_id,
      createdAt: row.created_at,
    };
  }

  latestAgentState(conversationId) {
    const outcome = this.latestAgentOutcome(conversationId);
    if (!outcome || ['draft_created', 'autopilot_message_sent', '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', 'customer_memory_items', 'customer_memory_snapshots', 'memory_extraction_jobs', 'customer_memory_revisions', 'drafts', 'audit_logs', 'poll_state', 'contact_onboarding'];
    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, isMeaningfulContactName, normalizeContactName };