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