From 2ec6fdbe11df4a409059014de725e95a81eb8c30 Mon Sep 17 00:00:00 2001 From: john Date: Wed, 2 Sep 2026 14:30:59 +0800 Subject: [PATCH] Implement Experience V1 reflect, retrieval boost, and product metrics. Add rule-based reflect() to aggregate task outcomes, environment-aware search scoring, experience_saved/recalled events, and structured injection blocks per the V1 schema. Co-authored-by: Cursor --- experience-reflect.mjs | 186 ++++++++++++++++++ experience-reflect.test.mjs | 90 +++++++++ experience-schema.mjs | 38 +++- experience-schema.test.mjs | 19 +- experience-service-pg.mjs | 147 ++++++++++++-- experience-service.mjs | 138 ++++++++++--- experience-service.test.mjs | 71 ++++++- memory-v2-product-events.mjs | 35 ++++ memory-v2-product-events.test.mjs | 2 + .../mindspace-service-bootstrap.mjs | 2 +- server/portal-agent-services-bootstrap.mjs | 5 +- 11 files changed, 675 insertions(+), 58 deletions(-) create mode 100644 experience-reflect.mjs create mode 100644 experience-reflect.test.mjs diff --git a/experience-reflect.mjs b/experience-reflect.mjs new file mode 100644 index 0000000..8a28f84 --- /dev/null +++ b/experience-reflect.mjs @@ -0,0 +1,186 @@ +import { + mapExperienceRow, + normalizeExperienceRecordInput, + parseExperienceJson, +} from './experience-schema.mjs'; + +export const DEFAULT_REFLECT_MIN_GROUP_SIZE = 3; + +export function problemFingerprint(problem) { + const text = String(problem ?? '') + .toLowerCase() + .replace(/[^\p{L}\p{N}]+/gu, ' ') + .trim() + .replace(/\s+/g, ' '); + if (!text) return ''; + return text.slice(0, 120); +} + +export function environmentMatchBoost(row, terms) { + if (!terms.length) return 0; + const environment = parseExperienceJson(row.environment_json ?? row.environment, null); + if (!environment) return 0; + const haystack = [ + environment.runtime, + ...(Array.isArray(environment.components) ? environment.components : []), + environment.region, + ] + .filter(Boolean) + .join(' ') + .toLowerCase(); + if (!haystack) return 0; + let boost = 0; + for (const term of terms) { + if (haystack.includes(term)) boost += 0.5; + } + return boost; +} + +export function scoreExperienceRow(row, terms, nowMs, halfLifeMs) { + const haystack = [ + row.title, + row.body, + row.problem, + row.environment_json ? JSON.stringify(parseExperienceJson(row.environment_json, {})) : '', + ] + .filter(Boolean) + .join('\n') + .toLowerCase(); + + let keywordScore = 0; + for (const term of terms) { + if (haystack.includes(term)) keywordScore += 1; + } + keywordScore += environmentMatchBoost(row, terms); + if (keywordScore <= 0) return { row, score: 0 }; + + const ageMs = Math.max(0, nowMs - Number(row.updated_at ?? row.updatedAt ?? 0)); + const recency = Math.pow(0.5, ageMs / halfLifeMs); + return { row, score: keywordScore * recency }; +} + +export function groupTaskOutcomesByProblem(rows, minGroupSize = DEFAULT_REFLECT_MIN_GROUP_SIZE) { + const buckets = new Map(); + for (const row of rows) { + if (String(row.kind ?? '').toLowerCase() !== 'task_outcome') continue; + if (row.parent_experience_id ?? row.parentExperienceId) continue; + if (String(row.status ?? 'active').toLowerCase() !== 'active') continue; + const fp = problemFingerprint(row.problem || row.title); + if (!fp) continue; + if (!buckets.has(fp)) buckets.set(fp, []); + buckets.get(fp).push(row); + } + return [...buckets.values()].filter((group) => group.length >= minGroupSize); +} + +export function findExistingExecutionPattern(rows, fingerprint) { + for (const row of rows) { + if (String(row.kind ?? '').toLowerCase() !== 'execution_pattern') continue; + if (String(row.status ?? 'active').toLowerCase() !== 'active') continue; + const fp = problemFingerprint(row.problem || row.title); + if (fp === fingerprint) return row; + } + return null; +} + +export function buildExecutionPatternAggregate(rows) { + const sorted = [...rows].sort( + (a, b) => Number(b.updated_at ?? b.updatedAt ?? 0) - Number(a.updated_at ?? a.updatedAt ?? 0), + ); + const problem = sorted[0].problem || sorted[0].title; + const results = sorted.map((row) => row.result).filter(Boolean); + const successCount = results.filter((value) => value === 'success').length; + const result = + successCount >= results.length / 2 + ? 'success' + : results.every((value) => value === 'failure') + ? 'failure' + : 'partial'; + + const confidences = sorted + .map((row) => (row.confidence == null ? null : Number(row.confidence))) + .filter((value) => Number.isFinite(value)); + const confidence = + confidences.length > 0 + ? Number((confidences.reduce((sum, value) => sum + value, 0) / confidences.length).toFixed(3)) + : null; + + const components = new Set(); + let runtime = null; + const steps = new Set(); + let executor = null; + const sources = []; + const seenSourceIds = new Set(); + + for (const row of sorted) { + const environment = parseExperienceJson(row.environment_json ?? row.environment, null); + if (environment?.runtime) runtime = environment.runtime; + for (const component of environment?.components ?? []) { + if (component) components.add(String(component)); + } + + const action = parseExperienceJson(row.action_json ?? row.action, null); + if (action?.executor) executor = action.executor; + for (const step of action?.steps ?? []) { + if (step) steps.add(String(step)); + } + + const evidence = parseExperienceJson(row.evidence_json ?? row.evidence, null); + for (const source of evidence?.sources ?? []) { + const sourceId = String(source?.source_id ?? '').trim(); + if (sourceId && seenSourceIds.has(sourceId)) continue; + if (sourceId) seenSourceIds.add(sourceId); + sources.push(source); + } + } + + const bodyLines = sorted.slice(0, 5).map((row) => `- ${row.title}: ${row.body}`); + if (sorted.length > 5) { + bodyLines.push(`- … 另有 ${sorted.length - 5} 条同类任务记录`); + } + + return normalizeExperienceRecordInput({ + kind: 'execution_pattern', + title: String(problem).slice(0, 200), + body: `聚合 ${sorted.length} 次同类任务经验：\n${bodyLines.join('\n')}`, + problem, + result, + confidence, + environment: { + runtime, + components: [...components], + }, + action: { + executor, + steps: [...steps], + }, + evidence: { + sources, + provenance: { + aggregated_from: sorted.map((row) => row.id), + }, + }, + tags: ['execution_pattern', 'reflect'], + }); +} + +export function planExperienceReflection(rows, { minGroupSize = DEFAULT_REFLECT_MIN_GROUP_SIZE } = {}) { + const activeRows = rows.filter((row) => String(row.status ?? 'active').toLowerCase() === 'active'); + const groups = groupTaskOutcomesByProblem(activeRows, minGroupSize); + const plans = []; + for (const group of groups) { + const fingerprint = problemFingerprint(group[0].problem || group[0].title); + if (!fingerprint) continue; + if (findExistingExecutionPattern(activeRows, fingerprint)) continue; + plans.push({ + fingerprint, + sourceIds: group.map((row) => row.id), + payload: buildExecutionPatternAggregate(group), + }); + } + return plans; +} + +export function mapReflectResult(savedRow) { + return mapExperienceRow(savedRow); +} diff --git a/experience-reflect.test.mjs b/experience-reflect.test.mjs new file mode 100644 index 0000000..36729ba --- /dev/null +++ b/experience-reflect.test.mjs @@ -0,0 +1,90 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + buildExecutionPatternAggregate, + environmentMatchBoost, + groupTaskOutcomesByProblem, + planExperienceReflection, + problemFingerprint, +} from './experience-reflect.mjs'; + +test('problemFingerprint normalizes punctuation and spacing', () => { + assert.equal(problemFingerprint(' SSE connection!!! '), 'sse connection'); +}); + +test('groupTaskOutcomesByProblem groups active task outcomes by problem', () => { + const groups = groupTaskOutcomesByProblem([ + { id: '1', kind: 'task_outcome', status: 'active', problem: 'SSE 断线', title: 'A' }, + { id: '2', kind: 'task_outcome', status: 'active', problem: 'SSE 断线', title: 'B' }, + { id: '3', kind: 'task_outcome', status: 'active', problem: 'SSE 断线', title: 'C' }, + { id: '4', kind: 'task_outcome', status: 'active', problem: 'Other', title: 'D' }, + { id: '5', kind: 'lesson', status: 'active', problem: 'SSE 断线', title: 'E' }, + ], 3); + assert.equal(groups.length, 1); + assert.equal(groups[0].length, 3); +}); + +test('environmentMatchBoost rewards component matches', () => { + const boost = environmentMatchBoost( + { environment_json: JSON.stringify({ runtime: 'goosed', components: ['docker', 'postgres'] }) }, + ['goosed', 'docker'], + ); + assert.equal(boost, 1); +}); + +test('buildExecutionPatternAggregate merges structured fields', () => { + const payload = buildExecutionPatternAggregate([ + { + id: '1', + title: 'SSE 修复', + body: '改 server.rs', + problem: 'SSE connection instability', + result: 'success', + confidence: 0.8, + updated_at: 100, + environment_json: JSON.stringify({ runtime: 'goosed', components: ['docker'] }), + action_json: JSON.stringify({ executor: 'goose', steps: ['edit server.rs'] }), + evidence_json: JSON.stringify({ sources: [{ source_id: 'run:1' }] }), + }, + { + id: '2', + title: 'SSE 再修', + body: '补 replay 映射', + problem: 'SSE connection instability', + result: 'success', + confidence: 0.9, + updated_at: 200, + environment_json: JSON.stringify({ components: ['portal'] }), + action_json: JSON.stringify({ executor: 'goose', steps: ['replay mapping'] }), + evidence_json: JSON.stringify({ sources: [{ source_id: 'run:2' }] }), + }, + { + id: '3', + title: 'SSE 第三次', + body: '终态恢复', + problem: 'SSE connection instability', + result: 'partial', + confidence: 0.7, + updated_at: 300, + environment_json: JSON.stringify({ components: ['sse'] }), + action_json: JSON.stringify({ executor: 'goose', steps: ['finish sync'] }), + evidence_json: JSON.stringify({ sources: [{ source_id: 'run:1' }] }), + }, + ]); + assert.equal(payload.kind, 'execution_pattern'); + assert.equal(payload.result, 'success'); + assert.equal(payload.confidence, 0.8); + assert.deepEqual(payload.environment.components.sort(), ['docker', 'portal', 'sse']); + assert.equal(payload.evidence.sources.length, 2); + assert.deepEqual(payload.evidence.provenance.aggregated_from, ['3', '2', '1']); +}); + +test('planExperienceReflection skips groups with existing execution_pattern', () => { + const plans = planExperienceReflection([ + { id: '1', kind: 'task_outcome', status: 'active', problem: 'SSE 断线', title: 'A' }, + { id: '2', kind: 'task_outcome', status: 'active', problem: 'SSE 断线', title: 'B' }, + { id: '3', kind: 'task_outcome', status: 'active', problem: 'SSE 断线', title: 'C' }, + { id: 'p1', kind: 'execution_pattern', status: 'active', problem: 'SSE 断线', title: 'pattern' }, + ], { minGroupSize: 3 }); + assert.equal(plans.length, 0); +}); diff --git a/experience-schema.mjs b/experience-schema.mjs index 9ad919a..89fcc7d 100644 --- a/experience-schema.mjs +++ b/experience-schema.mjs @@ -137,6 +137,24 @@ export function experienceSearchHaystack(row) { .toLowerCase(); } +export function formatExperienceMonth(timestampMs) { + const ts = Number(timestampMs); + if (!Number.isFinite(ts) || ts <= 0) return 'unknown'; + const date = new Date(ts); + const year = date.getUTCFullYear(); + const month = String(date.getUTCMonth() + 1).padStart(2, '0'); + return `${year}-${month}`; +} + +export function formatExperienceActionSummary(action) { + if (!action || typeof action !== 'object') return ''; + const steps = Array.isArray(action.steps) ? action.steps.filter(Boolean) : []; + if (steps.length > 0) return steps.slice(0, 3).join(' → '); + const artifacts = Array.isArray(action.artifacts) ? action.artifacts.filter(Boolean) : []; + if (artifacts.length > 0) return `artifacts: ${artifacts.slice(0, 3).join(', ')}`; + return action.executor ? String(action.executor) : ''; +} + export function formatExperienceInjectionBlock(hits = []) { if (!Array.isArray(hits) || hits.length === 0) return ''; const lines = hits.map((hit) => formatExperienceInjectionLine(hit)); @@ -144,19 +162,25 @@ export function formatExperienceInjectionBlock(hits = []) { } export function formatExperienceInjectionLine(hit) { - const parts = [`- ${hit.title}: ${hit.body}`]; - if (hit.problem) parts.push(` 问题: ${hit.problem}`); - if (hit.result) { - const conf = hit.confidence == null ? '' : `, conf=${hit.confidence}`; - parts.push(` 结果: ${hit.result}${conf}`); - } + const month = formatExperienceMonth(hit.createdAt); + const headline = hit.problem || hit.title; + const resultPart = hit.result + ? ` (${hit.result}${hit.confidence == null ? '' : `, conf=${hit.confidence}`})` + : ''; + const parts = [`- [${month}] ${headline}${resultPart}`]; + const runtime = hit.environment?.runtime; const components = Array.isArray(hit.environment?.components) ? hit.environment.components.filter(Boolean).join(' + ') : ''; if (runtime || components) { - parts.push(` 环境: ${[runtime, components].filter(Boolean).join(' · ')}`); + parts.push(` 环境: ${[runtime, components].filter(Boolean).join(' + ')}`); } + + const actionSummary = formatExperienceActionSummary(hit.action); + if (actionSummary) parts.push(` 处理: ${actionSummary}`); + + parts.push(` 摘要: ${hit.body}`); return parts.join('\n'); } diff --git a/experience-schema.test.mjs b/experience-schema.test.mjs index 5e82e52..c09f52f 100644 --- a/experience-schema.test.mjs +++ b/experience-schema.test.mjs @@ -66,22 +66,25 @@ test('formatExperienceInjectionLine includes structured hints', () => { problem: '公网访问失败', result: 'success', confidence: 0.9, + createdAt: Date.parse('2026-09-15T00:00:00Z'), environment: { runtime: '103', components: ['caddy', 'portal'] }, + action: { executor: 'goose', steps: ['edit server.rs', 'reload caddy'] }, }); - assert.match(line, /部署: 摘要/); - assert.match(line, /问题: 公网访问失败/); - assert.match(line, /结果: success, conf=0.9/); - assert.match(line, /环境: 103 · caddy \+ portal/); + assert.match(line, /\[2026-09\] 公网访问失败 \(success, conf=0.9\)/); + assert.match(line, /环境: 103 \+ caddy \+ portal/); + assert.match(line, /处理: edit server.rs → reload caddy/); + assert.match(line, /摘要: 摘要/); }); test('formatExperienceInjectionBlock wraps lines with header', () => { const block = formatExperienceInjectionBlock([ - { title: 'A', body: 'one' }, - { title: 'B', body: 'two' }, + { title: 'A', body: 'one', createdAt: Date.parse('2026-01-01T00:00:00Z') }, + { title: 'B', body: 'two', createdAt: Date.parse('2026-02-01T00:00:00Z') }, ]); assert.match(block, /^# 相关经验/); - assert.match(block, /- A: one/); - assert.match(block, /- B: two/); + assert.match(block, /\[2026-01\] A/); + assert.match(block, /摘要: one/); + assert.match(block, /\[2026-02\] B/); }); test('buildMindspaceJobExperiencePayload maps failed jobs to failure result', () => { diff --git a/experience-service-pg.mjs b/experience-service-pg.mjs index 03e8d35..2629d85 100644 --- a/experience-service-pg.mjs +++ b/experience-service-pg.mjs @@ -1,11 +1,24 @@ 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 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). @@ -27,6 +40,7 @@ import { * @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; @@ -34,6 +48,7 @@ export async function createPgExperienceService(options = {}) { 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( @@ -110,6 +125,45 @@ export async function createPgExperienceService(options = {}) { }); } + 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'); @@ -160,7 +214,7 @@ export async function createPgExperienceService(options = {}) { ts, ], ); - return rowToExperience({ + const saved = rowToExperience({ id, scope: normalized.scope, kind: normalized.kind, @@ -183,6 +237,8 @@ export async function createPgExperienceService(options = {}) { created_at: ts, updated_at: ts, }); + void emitExperienceSaved(saved, input); + return saved; } async function search(query, { scope = 'global', limit = 5 } = {}) { @@ -208,7 +264,9 @@ export async function createPgExperienceService(options = {}) { ); if (rows.length > 0) { await bumpUseCount(rows.map((r) => r.id)); - return rows.map(rowToExperience); + const hits = rows.map(rowToExperience); + void emitExperienceRecalled({ query: text, hits, scope }); + return hits; } } } catch { @@ -216,24 +274,25 @@ export async function createPgExperienceService(options = {}) { } } - // Keyword fallback: ILIKE any term, ordered by recency. + // 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})`); - const params = [scope, ...terms.map((t) => `%${t}%`), max]; + 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( - `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 + `${PG_EXPERIENCE_SELECT} WHERE scope = $1 AND status = 'active' AND (${likeClauses.join(' OR ')}) ORDER BY updated_at DESC - LIMIT $${terms.length + 2}`, + LIMIT 500`, params, ); - if (rows.length > 0) await bumpUseCount(rows.map((r) => r.id)); - return rows.map(rowToExperience); + 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) { @@ -255,8 +314,62 @@ export async function createPgExperienceService(options = {}) { ]; } - async function reflect() { - return { ok: false, reason: 'not_implemented' }; + 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() { diff --git a/experience-service.mjs b/experience-service.mjs index e8e98f6..a314123 100644 --- a/experience-service.mjs +++ b/experience-service.mjs @@ -6,6 +6,21 @@ import { 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). @@ -20,9 +35,11 @@ import { * @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), @@ -30,16 +47,7 @@ export function createExperienceService(pool, options = {}) { function rankRows(rows, queryTerms, nowMs) { return rows - .map((row) => { - const haystack = experienceSearchHaystack(row); - let keywordScore = 0; - for (const term of queryTerms) { - if (haystack.includes(term)) keywordScore += 1; - } - const ageMs = Math.max(0, nowMs - Number(row.updated_at)); - const recency = Math.pow(0.5, ageMs / halfLifeMs); - return { row, score: keywordScore * recency }; - }) + .map((row) => scoreExperienceRow(row, queryTerms, nowMs, halfLifeMs)) .filter((entry) => entry.score > 0) .sort((a, b) => b.score - a.score); } @@ -56,6 +64,38 @@ export function createExperienceService(pool, options = {}) { ]; } + 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. */ @@ -100,7 +140,7 @@ export function createExperienceService(pool, options = {}) { ], ); - return mapExperienceRow({ + const saved = mapExperienceRow({ id, scope: normalized.scope, kind: normalized.kind, @@ -123,6 +163,8 @@ export function createExperienceService(pool, options = {}) { created_at: ts, updated_at: ts, }); + void emitExperienceSaved(saved, input); + return saved; } /** @@ -133,11 +175,7 @@ export function createExperienceService(pool, options = {}) { const terms = tokenize(query); if (terms.length === 0) return []; const [rows] = await pool.query( - `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 + `${EXPERIENCE_SELECT_SQL} WHERE scope = ? AND status = 'active' ORDER BY updated_at DESC LIMIT 500`, @@ -154,16 +192,72 @@ export function createExperienceService(pool, options = {}) { ) .catch(() => {}); } - return ranked.map((entry) => mapExperienceRow(entry.row)); + const hits = ranked.map((entry) => mapExperienceRow(entry.row)); + void emitExperienceRecalled({ query, hits, scope }); + return hits; } /** - * 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. + * Rule-based reflection: aggregate repeated task_outcome rows into one + * execution_pattern. No LLM; idempotent per problem fingerprint. */ - async function reflect() { - return { ok: false, reason: 'not_implemented' }; + 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 }; diff --git a/experience-service.test.mjs b/experience-service.test.mjs index 628fcfb..1469c44 100644 --- a/experience-service.test.mjs +++ b/experience-service.test.mjs @@ -58,7 +58,7 @@ function createMockPool() { if (sql.includes('FROM h5_experience') && sql.includes('WHERE scope = ?')) { const [scope] = params; const matched = rows - .filter((r) => r.scope === scope && r.status === 'active') + .filter((r) => r.scope === scope && (sql.includes("status = 'active'") ? r.status === 'active' : true)) .sort((a, b) => b.updated_at - a.updated_at); return [matched.map((r) => ({ ...r }))]; } @@ -69,6 +69,18 @@ function createMockPool() { } return [{ affectedRows: params.length }]; } + if (sql.includes("SET status = 'archived'")) { + const [parentId, updatedAt, ...ids] = params; + for (const id of ids) { + const row = rows.find((r) => r.id === id); + if (row) { + row.status = 'archived'; + row.parent_experience_id = parentId; + row.updated_at = updatedAt; + } + } + return [{ affectedRows: ids.length }]; + } throw new Error(`unexpected SQL: ${sql}`); }; return { pool: { query }, rows }; @@ -176,3 +188,60 @@ test('search bumps use_count for returned rows', async () => { await svc.search('caddy 灰度'); assert.equal(rows[0].use_count, 1); }); + +test('search boosts environment component matches', async () => { + const { pool } = createMockPool(); + const svc = createExperienceService(pool, { now: () => 10_000 }); + await svc.record({ + title: '通用标题', + body: '正文', + environment: { runtime: 'goosed', components: ['docker'] }, + }); + await svc.record({ + title: '另一标题', + body: '正文', + problem: 'goosed docker 部署', + }); + const hits = await svc.search('goosed docker', { limit: 1 }); + assert.equal(hits.length, 1); + assert.equal(hits[0].title, '通用标题'); +}); + +test('reflect aggregates repeated task outcomes into execution_pattern', async () => { + const { pool, rows } = createMockPool(); + const svc = createExperienceService(pool, { now: () => 9000 }); + for (let i = 0; i < 3; i += 1) { + await svc.record({ + kind: 'task_outcome', + title: `SSE 修复 ${i}`, + body: `body ${i}`, + problem: 'SSE connection instability', + result: 'success', + confidence: 0.8, + }); + } + const result = await svc.reflect({ minGroupSize: 3 }); + assert.equal(result.ok, true); + assert.equal(result.created, 1); + assert.equal(result.archived, 3); + assert.equal(rows.filter((row) => row.kind === 'execution_pattern').length, 1); + assert.equal(rows.filter((row) => row.status === 'archived').length, 3); +}); + +test('reflect dryRun reports eligible groups without writing', async () => { + const { pool, rows } = createMockPool(); + const svc = createExperienceService(pool, { now: () => 11_000 }); + for (let i = 0; i < 3; i += 1) { + await svc.record({ + kind: 'task_outcome', + title: `Deploy ${i}`, + body: `body ${i}`, + problem: 'headscale deploy', + result: 'success', + }); + } + const preview = await svc.reflect({ minGroupSize: 3, dryRun: true }); + assert.equal(preview.dryRun, true); + assert.equal(preview.groupsEligible, 1); + assert.equal(rows.filter((row) => row.kind === 'execution_pattern').length, 0); +}); diff --git a/memory-v2-product-events.mjs b/memory-v2-product-events.mjs index 9506e20..2677543 100644 --- a/memory-v2-product-events.mjs +++ b/memory-v2-product-events.mjs @@ -5,6 +5,8 @@ export const MEMORY_V2_PRODUCT_EVENT_TYPES = Object.freeze({ PROMOTED: 'memory_promoted', RESOLVED_INJECTED: 'memory_resolved_injected', RECALL_HIT: 'memory_recall_hit', + EXPERIENCE_SAVED: 'experience_saved', + EXPERIENCE_RECALLED: 'experience_recalled', }); const TABLE = 'h5_memory_v2_product_events'; @@ -170,6 +172,21 @@ async function countMemoryItemsCreated(pool, { sinceMs, userId }) { return Number(rows[0]?.count ?? 0); } +async function countExperienceRowsCreated(pool, { sinceMs, userId }) { + if (!(await tableExists(pool, 'h5_experience'))) return 0; + const clauses = ['created_at >= ?', "status = 'active'"]; + const params = [sinceMs]; + if (userId) { + clauses.push('source_user_id = ?'); + params.push(String(userId)); + } + const [rows] = await pool.query( + `SELECT COUNT(*) AS count FROM h5_experience WHERE ${clauses.join(' AND ')}`, + params, + ); + return Number(rows[0]?.count ?? 0); +} + export async function aggregateMemoryV2ProductMetrics( pool, { since = '7d', userId = null, now = Date.now() } = {}, @@ -183,10 +200,13 @@ export async function aggregateMemoryV2ProductMetrics( promotedEvents, injectedEvents, recallHitEvents, + experienceSavedEvents, + experienceRecalledEvents, candidateSavedFallback, promotedFallback, injectedFallback, recallHitFallback, + experienceSavedFallback, ] = await Promise.all([ countProductEvents(pool, { sinceMs, @@ -208,10 +228,21 @@ export async function aggregateMemoryV2ProductMetrics( userId, eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.RECALL_HIT, }), + countProductEvents(pool, { + sinceMs, + userId, + eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.EXPERIENCE_SAVED, + }), + countProductEvents(pool, { + sinceMs, + userId, + eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.EXPERIENCE_RECALLED, + }), countCandidatesCreated(pool, { sinceMs, userId }), countMemoryItemsCreated(pool, { sinceMs, userId }), countAgentMemoryEvents(pool, { sinceMs, userId, injectionOnly: true }), countAgentRecallHits(pool, { sinceMs, userId }), + countExperienceRowsCreated(pool, { sinceMs, userId }), ]); const events = { @@ -219,6 +250,8 @@ export async function aggregateMemoryV2ProductMetrics( memory_promoted: promotedEvents || promotedFallback, memory_resolved_injected: injectedEvents || injectedFallback, memory_recall_hit: recallHitEvents || recallHitFallback, + experience_saved: experienceSavedEvents || experienceSavedFallback, + experience_recalled: experienceRecalledEvents, }; return { @@ -234,6 +267,8 @@ export async function aggregateMemoryV2ProductMetrics( memory_promoted: promotedEvents > 0 ? 'product_events' : 'memory_items_table', memory_resolved_injected: injectedEvents > 0 ? 'product_events' : 'agent_run_events', memory_recall_hit: recallHitEvents > 0 ? 'product_events' : 'agent_run_events', + experience_saved: experienceSavedEvents > 0 ? 'product_events' : 'h5_experience', + experience_recalled: experienceRecalledEvents > 0 ? 'product_events' : 'none', }, }; } diff --git a/memory-v2-product-events.test.mjs b/memory-v2-product-events.test.mjs index e8abca4..46a3231 100644 --- a/memory-v2-product-events.test.mjs +++ b/memory-v2-product-events.test.mjs @@ -51,5 +51,7 @@ test('aggregateMemoryV2ProductMetrics falls back to legacy tables', async () => assert.equal(metrics.events.memory_promoted, 8); assert.equal(metrics.events.memory_resolved_injected, 7); assert.equal(metrics.events.memory_recall_hit, 5); + assert.equal(metrics.events.experience_saved, 0); + assert.equal(metrics.events.experience_recalled, 0); assert.equal(metrics.sources.memory_candidate_saved, 'candidates_table'); }); diff --git a/mindspace-service/mindspace-service-bootstrap.mjs b/mindspace-service/mindspace-service-bootstrap.mjs index 6d10ac4..307b8a7 100644 --- a/mindspace-service/mindspace-service-bootstrap.mjs +++ b/mindspace-service/mindspace-service-bootstrap.mjs @@ -176,7 +176,7 @@ export async function bootstrapMindSpaceService({ const apiSecret = env.TKMIND_SERVER__SECRET_KEY ?? 'local-dev-secret'; let experienceService = null; if (runtime.experienceEnabled) { - experienceService = createExperienceService(pool); + experienceService = createExperienceService(pool, { productEventsPool: pool }); } const agentRunner = createMindSpaceAgentRunner({ apiTarget, diff --git a/server/portal-agent-services-bootstrap.mjs b/server/portal-agent-services-bootstrap.mjs index 9296658..d39d402 100644 --- a/server/portal-agent-services-bootstrap.mjs +++ b/server/portal-agent-services-bootstrap.mjs @@ -82,6 +82,7 @@ export async function bootstrapPortalAgentServices({ experienceService = await createPgExperienceServiceFn({ connectionString: env.EXPERIENCE_PG_URL, + productEventsPool: pool, }); logger.log( 'Experience store: PostgreSQL + pgvector', @@ -91,10 +92,10 @@ export async function bootstrapPortalAgentServices({ 'Experience PG init failed, falling back to MySQL store:', error instanceof Error ? error.message : error, ); - experienceService = createExperienceServiceFn(pool); + experienceService = createExperienceServiceFn(pool, { productEventsPool: pool }); } } else { - experienceService = createExperienceServiceFn(pool); + experienceService = createExperienceServiceFn(pool, { productEventsPool: pool }); } }