memind/experience-service.mjs

import crypto from 'node:crypto';

import {
  experienceSearchHaystack,
  mapExperienceRow,
  normalizeExperienceRecordInput,
  serializeExperienceJson,
} 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 EXPERIENCE_SELECT_SQL = `SELECT id, scope, kind, title, body, tags_json, 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`;

/**
 * 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
 * @param {import('mysql2/promise').Pool} [options.productEventsPool] - MySQL pool for product events
 */
export function createExperienceService(pool, options = {}) {
  const now = options.now ?? (() => Date.now());
  const productEventsPool = options.productEventsPool ?? pool;
  const halfLifeMs = Math.max(
    60_000,
    Number(options.recencyHalfLifeMs ?? 30 * 24 * 60 * 60 * 1000),
  );

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

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

  /**
   * Persist a piece of experience. Returns the stored record.
   */
  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 = serializeExperienceJson(normalized.environment);
    const actionJson = serializeExperienceJson(normalized.action);
    const evidenceJson = serializeExperienceJson(normalized.evidence);

    await pool.query(
      `INSERT INTO h5_experience
         (id, scope, kind, title, body, tags_json, 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)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
      [
        id,
        normalized.scope,
        normalized.kind,
        normalized.title,
        normalized.body,
        JSON.stringify(normalized.tags),
        normalized.sourceSessionId,
        normalized.sourceUserId,
        normalized.problem,
        environmentJson,
        normalized.hypothesis,
        actionJson,
        normalized.result,
        normalized.confidence,
        evidenceJson,
        normalized.parentExperienceId,
        normalized.supersedesId,
        normalized.status,
        ts,
        ts,
      ],
    );

    const saved = mapExperienceRow({
      id,
      scope: normalized.scope,
      kind: normalized.kind,
      title: normalized.title,
      body: normalized.body,
      tags_json: JSON.stringify(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;
  }

  /**
   * 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 [];
    const [rows] = await pool.query(
      `${EXPERIENCE_SELECT_SQL}
       WHERE scope = ? AND status = 'active'
       ORDER BY updated_at DESC
       LIMIT 500`,
      [scope],
    );
    const ranked = rankRows(rows, terms, now()).slice(0, Math.max(1, limit));
    if (ranked.length > 0) {
      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(() => {});
    }
    const hits = ranked.map((entry) => mapExperienceRow(entry.row));
    void emitExperienceRecalled({ query, hits, scope });
    return hits;
  }

  /**
   * Rule-based reflection: aggregate repeated task_outcome rows into one
   * execution_pattern. No LLM; idempotent per problem fingerprint.
   */
  async function reflect({
    scope = 'global',
    minGroupSize = DEFAULT_REFLECT_MIN_GROUP_SIZE,
    dryRun = false,
  } = {}) {
    const [rows] = await pool.query(
      `${EXPERIENCE_SELECT_SQL}
       WHERE scope = ?
       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',
      });
      const placeholders = plan.sourceIds.map(() => '?').join(',');
      await pool.query(
        `UPDATE h5_experience
         SET status = 'archived', parent_experience_id = ?, updated_at = ?
         WHERE id IN (${placeholders})`,
        [saved.id, now(), ...plan.sourceIds],
      );
      created += 1;
      archived += plan.sourceIds.length;
      patterns.push(saved);
    }

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

  return { record, search, reflect };
}

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

export { formatExperienceInjectionBlock, formatExperienceInjectionLine } from './experience-schema.mjs';