memind/experience-service.mjs

import crypto from 'node:crypto';

/**
 * Shared experience store (etat C of the goose scale plan).
 *
 * All goosed Worker instances read/write the same learned experience so that
 * capability is not siloed per instance. This first implementation is backed by
 * MySQL (`h5_experience`) with a keyword + recency ranking. The retrieval scoring
 * is deliberately isolated in `rankRows` so that swapping to PostgreSQL + pgvector
 * later only replaces the query/scoring layer, not the calling contract
 * (record / search / reflect).
 *
 * @param {import('mysql2/promise').Pool} pool
 * @param {object} [options]
 * @param {() => number} [options.now] - injectable clock for tests
 */
export function createExperienceService(pool, options = {}) {
  const now = options.now ?? (() => Date.now());
  // Recency half-life: a record this old contributes half its keyword score.
  const halfLifeMs = Math.max(
    60_000,
    Number(options.recencyHalfLifeMs ?? 30 * 24 * 60 * 60 * 1000),
  );

  function normalizeTags(tags) {
    if (!Array.isArray(tags)) return [];
    return [...new Set(tags.map((t) => String(t).trim()).filter(Boolean))];
  }

  function rowToExperience(row) {
    return {
      id: row.id,
      scope: row.scope,
      kind: row.kind,
      title: row.title,
      body: row.body,
      tags: row.tags_json ? safeJsonArray(row.tags_json) : [],
      sourceSessionId: row.source_session_id ?? null,
      sourceUserId: row.source_user_id ?? null,
      useCount: Number(row.use_count ?? 0),
      createdAt: Number(row.created_at),
      updatedAt: Number(row.updated_at),
    };
  }

  function safeJsonArray(value) {
    if (Array.isArray(value)) return value;
    try {
      const parsed = JSON.parse(value);
      return Array.isArray(parsed) ? parsed : [];
    } catch {
      return [];
    }
  }

  // Keyword overlap × recency decay. Kept separate so the pgvector backend can
  // replace this with cosine similarity without touching search()'s shape.
  function rankRows(rows, queryTerms, nowMs) {
    const terms = queryTerms;
    return rows
      .map((row) => {
        const haystack = `${row.title}\n${row.body}`.toLowerCase();
        let keywordScore = 0;
        for (const term of terms) {
          if (haystack.includes(term)) keywordScore += 1;
        }
        const ageMs = Math.max(0, nowMs - Number(row.updated_at));
        const recency = Math.pow(0.5, ageMs / halfLifeMs);
        const score = keywordScore * recency;
        return { row, score };
      })
      .filter((entry) => entry.score > 0)
      .sort((a, b) => b.score - a.score);
  }

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

  /**
   * Persist a piece of experience. Returns the stored record.
   */
  async function record(input) {
    const title = String(input?.title ?? '').trim();
    const body = String(input?.body ?? '').trim();
    if (!title) throw experienceError('经验标题不能为空', 'invalid_experience_input');
    if (!body) throw experienceError('经验内容不能为空', 'invalid_experience_input');
    const id = crypto.randomUUID();
    const ts = now();
    const scope = String(input?.scope ?? 'global').trim() || 'global';
    const kind = String(input?.kind ?? 'lesson').trim() || 'lesson';
    const tags = normalizeTags(input?.tags);
    await pool.query(
      `INSERT INTO h5_experience
         (id, scope, kind, title, body, tags_json, source_session_id, source_user_id,
          use_count, created_at, updated_at)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`,
      [
        id,
        scope,
        kind,
        title,
        body,
        JSON.stringify(tags),
        input?.sourceSessionId ?? null,
        input?.sourceUserId ?? null,
        ts,
        ts,
      ],
    );
    return rowToExperience({
      id,
      scope,
      kind,
      title,
      body,
      tags_json: tags,
      source_session_id: input?.sourceSessionId ?? null,
      source_user_id: input?.sourceUserId ?? null,
      use_count: 0,
      created_at: ts,
      updated_at: ts,
    });
  }

  /**
   * Retrieve the most relevant experience for a query within a scope.
   * @returns {Promise<Array>} ranked experiences (best first), max `limit`.
   */
  async function search(query, { scope = 'global', limit = 5 } = {}) {
    const terms = tokenize(query);
    if (terms.length === 0) return [];
    // Pull a bounded candidate set by scope+recency, then rank in JS. The
    // candidate cap keeps this cheap on MySQL; pgvector would push ranking into SQL.
    const [rows] = await pool.query(
      `SELECT id, scope, kind, title, body, tags_json, source_session_id,
              source_user_id, use_count, created_at, updated_at
       FROM h5_experience
       WHERE scope = ?
       ORDER BY updated_at DESC
       LIMIT 500`,
      [scope],
    );
    const ranked = rankRows(rows, terms, now()).slice(0, Math.max(1, limit));
    if (ranked.length > 0) {
      // Best-effort usage bump so popular experience can be surfaced/weighted later.
      const ids = ranked.map((entry) => entry.row.id);
      const placeholders = ids.map(() => '?').join(',');
      await pool
        .query(
          `UPDATE h5_experience SET use_count = use_count + 1 WHERE id IN (${placeholders})`,
          ids,
        )
        .catch(() => {});
    }
    return ranked.map((entry) => rowToExperience(entry.row));
  }

  /**
   * Placeholder for the reflection pass that distills raw session traces into
   * durable lessons. Wired later once the record() trigger points are defined;
   * kept here so the calling contract is stable.
   */
  async function reflect() {
    return { ok: false, reason: 'not_implemented' };
  }

  return { record, search, reflect };
}

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