memind/experience-service-pg.mjs

import crypto from 'node:crypto';

import {
  mapExperienceRow,
  normalizeExperienceRecordInput,
} from './experience-schema.mjs';
import {
  DEFAULT_REFLECT_MIN_GROUP_SIZE,
  planExperienceReflection,
  scoreExperienceRow,
} from './experience-reflect.mjs';
import {
  MEMORY_V2_PRODUCT_EVENT_TYPES,
  recordMemoryV2ProductEvent,
} from './memory-v2-product-events.mjs';

const PG_EXPERIENCE_SELECT = `SELECT id, scope, kind, title, body, tags, source_session_id,
              source_user_id, use_count, problem, environment_json, hypothesis,
              action_json, result, confidence, evidence_json, parent_experience_id,
              supersedes_id, status, created_at, updated_at
       FROM h5_experience`;

/**
 * PostgreSQL + pgvector backed experience store (etat C, polyglot mode).
 *
 * Same contract as the MySQL `createExperienceService` (record / search /
 * reflect) so callers (mindspace-agent-runner) are backend-agnostic. The MySQL
 * business DB is untouched; only this shared-experience workload lives on PG.
 *
 * Semantic search via pgvector is used when an `embed(text) => number[]` function
 * is supplied; otherwise it degrades to keyword (ILIKE) + recency, identical in
 * spirit to the MySQL version, so it is useful before an embedding provider is wired.
 *
 * The `pg` driver is imported dynamically so this module loads even when the
 * dependency is not yet installed; it is only required once the service is built.
 *
 * @param {object} options
 * @param {string} options.connectionString - e.g. process.env.EXPERIENCE_PG_URL
 * @param {(text: string) => Promise<number[]>} [options.embed] - embedding fn
 * @param {number} [options.embeddingDim=1536] - vector dimension (must match embed)
 * @param {() => number} [options.now]
 * @param {number} [options.recencyHalfLifeMs]
 * @param {import('mysql2/promise').Pool} [options.productEventsPool] - MySQL pool for product events
 */
export async function createPgExperienceService(options = {}) {
  const connectionString = options.connectionString;
  if (!connectionString) {
    throw new Error('createPgExperienceService 需要 connectionString（建议用 EXPERIENCE_PG_URL）');
  }
  const now = options.now ?? (() => Date.now());
  const productEventsPool = options.productEventsPool ?? null;
  const embed = options.embed ?? null;
  const embeddingDim = Math.max(1, Number(options.embeddingDim ?? 1536));
  const halfLifeMs = Math.max(
    60_000,
    Number(options.recencyHalfLifeMs ?? 30 * 24 * 60 * 60 * 1000),
  );

  const { default: pg } = await import('pg');
  const pool = new pg.Pool({ connectionString, max: Number(options.poolMax ?? 10) });

  // Idempotent setup. pgvector must be available on the server (CREATE EXTENSION).
  await pool.query('CREATE EXTENSION IF NOT EXISTS vector');
  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_experience (
      id UUID PRIMARY KEY,
      scope TEXT NOT NULL DEFAULT 'global',
      kind TEXT NOT NULL DEFAULT 'lesson',
      title TEXT NOT NULL,
      body TEXT NOT NULL,
      tags JSONB NOT NULL DEFAULT '[]'::jsonb,
      source_session_id TEXT,
      source_user_id TEXT,
      use_count BIGINT NOT NULL DEFAULT 0,
      embedding vector(${embeddingDim}),
      problem TEXT,
      environment_json JSONB,
      hypothesis TEXT,
      action_json JSONB,
      result TEXT,
      confidence DOUBLE PRECISION,
      evidence_json JSONB,
      parent_experience_id UUID,
      supersedes_id UUID,
      status TEXT NOT NULL DEFAULT 'active',
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL
    )
  `);
  const pgExperienceColumns = [
    ['problem', 'TEXT'],
    ['environment_json', 'JSONB'],
    ['hypothesis', 'TEXT'],
    ['action_json', 'JSONB'],
    ['result', 'TEXT'],
    ['confidence', 'DOUBLE PRECISION'],
    ['evidence_json', 'JSONB'],
    ['parent_experience_id', 'UUID'],
    ['supersedes_id', 'UUID'],
    ["status", "TEXT NOT NULL DEFAULT 'active'"],
  ];
  for (const [column, definition] of pgExperienceColumns) {
    await pool.query(`ALTER TABLE h5_experience ADD COLUMN IF NOT EXISTS ${column} ${definition}`);
  }
  await pool.query(
    'CREATE INDEX IF NOT EXISTS idx_h5_experience_scope_updated ON h5_experience (scope, updated_at DESC)',
  );
  // Trigram-ish keyword fallback index (btree on lower(title)) is cheap; the
  // ANN vector index is created lazily once rows have embeddings.
  await pool.query(
    "CREATE INDEX IF NOT EXISTS idx_h5_experience_embedding ON h5_experience USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100)",
  ).catch(() => {
    // ivfflat needs the extension + may warn on empty table; non-fatal.
  });

  function toVectorLiteral(vec) {
    // pgvector accepts a string like '[0.1,0.2,...]'
    return `[${vec.map((n) => Number(n)).join(',')}]`;
  }

  function rowToExperience(row) {
    return mapExperienceRow({
      ...row,
      tags_json: row.tags,
    });
  }

  async function emitExperienceSaved(saved, input) {
    if (!productEventsPool?.query) return;
    await recordMemoryV2ProductEvent(productEventsPool, {
      eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.EXPERIENCE_SAVED,
      userId: saved.sourceUserId,
      sessionId: saved.sourceSessionId,
      data: {
        experienceId: saved.id,
        kind: saved.kind,
        scope: saved.scope,
        result: saved.result ?? null,
        source: input?.source ?? null,
      },
      createdAt: saved.createdAt,
    }).catch(() => {});
  }

  async function emitExperienceRecalled({ query, hits, scope }) {
    if (!productEventsPool?.query || !hits.length) return;
    await recordMemoryV2ProductEvent(productEventsPool, {
      eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.EXPERIENCE_RECALLED,
      userId: hits[0]?.sourceUserId ?? null,
      data: {
        query: String(query ?? '').slice(0, 500),
        scope,
        hitCount: hits.length,
        experienceIds: hits.map((hit) => hit.id),
        kinds: hits.map((hit) => hit.kind),
      },
    }).catch(() => {});
  }

  function rankRows(rows, queryTerms, nowMs) {
    return rows
      .map((row) => scoreExperienceRow(row, queryTerms, nowMs, halfLifeMs))
      .filter((entry) => entry.score > 0)
      .sort((a, b) => b.score - a.score);
  }

  async function record(input) {
    const normalized = normalizeExperienceRecordInput(input);
    if (!normalized.title) throw experienceError('经验标题不能为空', 'invalid_experience_input');
    if (!normalized.body) throw experienceError('经验内容不能为空', 'invalid_experience_input');
    const id = crypto.randomUUID();
    const ts = now();
    const environmentJson = normalized.environment;
    const actionJson = normalized.action;
    const evidenceJson = normalized.evidence;
    let embedding = null;
    if (embed) {
      try {
        const vec = await embed(`${normalized.title}\n${normalized.body}`);
        if (Array.isArray(vec) && vec.length === embeddingDim) embedding = toVectorLiteral(vec);
      } catch {
        embedding = null;
      }
    }
    await pool.query(
      `INSERT INTO h5_experience
         (id, scope, kind, title, body, tags, source_session_id, source_user_id,
          use_count, embedding, problem, environment_json, hypothesis, action_json,
          result, confidence, evidence_json, parent_experience_id, supersedes_id,
          status, created_at, updated_at)
       VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7,$8,0,$9::vector,$10,$11::jsonb,$12,$13::jsonb,
               $14,$15,$16::jsonb,$17::uuid,$18::uuid,$19,$20,$21)`,
      [
        id,
        normalized.scope,
        normalized.kind,
        normalized.title,
        normalized.body,
        JSON.stringify(normalized.tags),
        normalized.sourceSessionId,
        normalized.sourceUserId,
        embedding,
        normalized.problem,
        environmentJson == null ? null : JSON.stringify(environmentJson),
        normalized.hypothesis,
        actionJson == null ? null : JSON.stringify(actionJson),
        normalized.result,
        normalized.confidence,
        evidenceJson == null ? null : JSON.stringify(evidenceJson),
        normalized.parentExperienceId,
        normalized.supersedesId,
        normalized.status,
        ts,
        ts,
      ],
    );
    const saved = rowToExperience({
      id,
      scope: normalized.scope,
      kind: normalized.kind,
      title: normalized.title,
      body: normalized.body,
      tags: normalized.tags,
      source_session_id: normalized.sourceSessionId,
      source_user_id: normalized.sourceUserId,
      use_count: 0,
      problem: normalized.problem,
      environment_json: environmentJson,
      hypothesis: normalized.hypothesis,
      action_json: actionJson,
      result: normalized.result,
      confidence: normalized.confidence,
      evidence_json: evidenceJson,
      parent_experience_id: normalized.parentExperienceId,
      supersedes_id: normalized.supersedesId,
      status: normalized.status,
      created_at: ts,
      updated_at: ts,
    });
    void emitExperienceSaved(saved, input);
    return saved;
  }

  async function search(query, { scope = 'global', limit = 5 } = {}) {
    const text = String(query ?? '').trim();
    if (!text) return [];
    const max = Math.max(1, limit);

    // Vector path: cosine distance ordering, recency as tiebreaker.
    if (embed) {
      try {
        const vec = await embed(text);
        if (Array.isArray(vec) && vec.length === embeddingDim) {
          const { rows } = await pool.query(
            `SELECT id, scope, kind, title, body, tags, source_session_id,
                    source_user_id, use_count, problem, environment_json, hypothesis,
                    action_json, result, confidence, evidence_json, parent_experience_id,
                    supersedes_id, status, created_at, updated_at
             FROM h5_experience
             WHERE scope = $1 AND status = 'active' AND embedding IS NOT NULL
             ORDER BY embedding <=> $2::vector, updated_at DESC
             LIMIT $3`,
            [scope, toVectorLiteral(vec), max],
          );
          if (rows.length > 0) {
            await bumpUseCount(rows.map((r) => r.id));
            const hits = rows.map(rowToExperience);
            void emitExperienceRecalled({ query: text, hits, scope });
            return hits;
          }
        }
      } catch {
        // fall through to keyword path
      }
    }

    // Keyword fallback: fetch candidate rows, rank with environment/problem boost.
    const terms = tokenize(text);
    if (terms.length === 0) return [];
    const likeClauses = terms.map(
      (_, i) => `(title ILIKE $${i + 2} OR body ILIKE $${i + 2} OR problem ILIKE $${i + 2} OR environment_json::text ILIKE $${i + 2})`,
    );
    const params = [scope, ...terms.map((t) => `%${t}%`)];
    const { rows } = await pool.query(
      `${PG_EXPERIENCE_SELECT}
       WHERE scope = $1 AND status = 'active' AND (${likeClauses.join(' OR ')})
       ORDER BY updated_at DESC
       LIMIT 500`,
      params,
    );
    const ranked = rankRows(rows, terms, now()).slice(0, max);
    if (ranked.length > 0) await bumpUseCount(ranked.map((entry) => entry.row.id));
    const hits = ranked.map((entry) => rowToExperience(entry.row));
    void emitExperienceRecalled({ query: text, hits, scope });
    return hits;
  }

  async function bumpUseCount(ids) {
    if (!ids.length) return;
    await pool
      .query('UPDATE h5_experience SET use_count = use_count + 1 WHERE id = ANY($1::uuid[])', [ids])
      .catch(() => {});
  }

  function tokenize(query) {
    return [
      ...new Set(
        String(query ?? '')
          .toLowerCase()
          .split(/[^\p{L}\p{N}]+/u)
          .map((t) => t.trim())
          .filter((t) => t.length >= 2),
      ),
    ];
  }

  async function reflect({
    scope = 'global',
    minGroupSize = DEFAULT_REFLECT_MIN_GROUP_SIZE,
    dryRun = false,
  } = {}) {
    const { rows } = await pool.query(
      `${PG_EXPERIENCE_SELECT}
       WHERE scope = $1
       ORDER BY updated_at DESC
       LIMIT 2000`,
      [scope],
    );
    const plans = planExperienceReflection(rows, { minGroupSize });
    if (dryRun) {
      return {
        ok: true,
        dryRun: true,
        scope,
        groupsEligible: plans.length,
        wouldArchive: plans.reduce((sum, plan) => sum + plan.sourceIds.length, 0),
        previews: plans.map((plan) => ({
          fingerprint: plan.fingerprint,
          sourceCount: plan.sourceIds.length,
          title: plan.payload.title,
        })),
      };
    }

    let created = 0;
    let archived = 0;
    const patterns = [];
    for (const plan of plans) {
      const saved = await record({
        ...plan.payload,
        scope,
        source: 'reflect',
      });
      await pool.query(
        `UPDATE h5_experience
         SET status = 'archived', parent_experience_id = $1::uuid, updated_at = $2
         WHERE id = ANY($3::uuid[])`,
        [saved.id, now(), plan.sourceIds],
      );
      created += 1;
      archived += plan.sourceIds.length;
      patterns.push(saved);
    }

    return {
      ok: true,
      dryRun: false,
      scope,
      created,
      archived,
      patterns,
    };
  }

  async function close() {
    await pool.end();
  }

  return { record, search, reflect, close };
}

function experienceError(message, code) {
  return Object.assign(new Error(message), { code });
}