diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index 80679f4..68ba987 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -49,6 +49,7 @@ import { isDirectEscalationContextEnabledForUser, resolveDirectEscalationContextPolicy, } from './chat-task-intent-config.mjs'; +import { extractAgentRunExperience } from './experience-extractor.mjs'; const DEFAULT_RUN_RETRY_DELAYS_MS = [1500, 5000, 15000]; const TERMINAL_STATUSES = new Set(['succeeded', 'failed']); @@ -849,6 +850,7 @@ export function createAgentRunGateway({ conversationMemoryService = null, syncUserPagesOnSuccess = null, observePersonalMemoryOnSuccess = null, + experienceService = null, observeWorkflowRun = null, observeWorkflowValidation = null, isSessionExternallyBusy = null, @@ -905,6 +907,7 @@ export function createAgentRunGateway({ const queuedDispatches = []; const queuedDispatchSet = new Set(); const suggestionRepairTriggered = new Set(); + const runExecutionContext = new Map(); function enqueueRun(runId) { if (!runId || inFlight.has(runId) || queuedDispatchSet.has(runId)) return false; @@ -2566,6 +2569,7 @@ export function createAgentRunGateway({ try { const execution = await runWithTimeout(runId, () => executeRun(row, runId)); + runExecutionContext.set(runId, execution); await finalizeSuccessfulRun(runId, row, execution.sessionId, execution); } catch (err) { const latest = await getRunById(runId); @@ -2669,6 +2673,26 @@ export function createAgentRunGateway({ stopHeartbeat(); const terminalRun = await getRunById(runId).catch(() => null); if (terminalRun && TERMINAL_STATUSES.has(terminalRun.status)) { + const executionContext = runExecutionContext.get(runId) ?? null; + runExecutionContext.delete(runId); + if (experienceService?.record) { + await extractAgentRunExperience({ + experienceService, + pool, + row: terminalRun, + runId, + status: terminalRun.status, + sessionId: terminalRun.agent_session_id ?? row.agent_session_id ?? null, + errorMessage: terminalRun.error_message ?? null, + toolEvidence: executionContext?.toolEvidence ?? null, + routing: executionContext?.routing ?? null, + }).catch((err) => { + console.warn( + '[AgentRun] experience extraction failed:', + err instanceof Error ? err.message : err, + ); + }); + } if (goalRunService?.onAgentRunCompleted && terminalRun.goal_checkpoint_id) { await goalRunService.onAgentRunCompleted({ agentRunId: runId, diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 3ef4953..02e069e 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -1151,6 +1151,91 @@ test('agent run awaits session Finish before succeeding when proxy supports it', assert.equal(finishEvents.length, 1); }); +test('terminal agent run records structured experience on success', async () => { + const pool = createFakePool(); + const recorded = []; + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async startSessionForUser() { + return { id: 'session-exp-1' }; + }, + async submitSessionReplyAndAwaitFinishForUser() { + return { + ok: true, + finishEvent: { type: 'Finish' }, + toolEvidence: { calls: ['edit_file'] }, + }; + }, + }, + experienceService: { + async record(payload) { + recorded.push(payload); + return { id: 'exp-1', ...payload }; + }, + }, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-1', { + requestId: 'req-exp-1', + userMessage: { + role: 'user', + content: [{ type: 'text', text: '修复 SSE 断线' }], + metadata: { displayText: '修复 SSE 断线' }, + }, + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); + assert.equal(recorded.length, 1); + assert.equal(recorded[0].kind, 'task_outcome'); + assert.equal(recorded[0].problem, '修复 SSE 断线'); + assert.equal(recorded[0].evidence.provenance.run_id, run.id); + assert.match(recorded[0].body, /edit_file/); +}); + +test('terminal agent run records structured experience on failure', async () => { + const pool = createFakePool(); + const recorded = []; + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async startSessionForUser() { + return { id: 'session-exp-fail' }; + }, + async submitSessionReplyAndAwaitFinishForUser() { + const error = new Error('goosed unavailable'); + error.code = 'GOOSED_UNAVAILABLE'; + error.retryable = false; + throw error; + }, + }, + experienceService: { + async record(payload) { + recorded.push(payload); + return { id: 'exp-fail', ...payload }; + }, + }, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-1', { + requestId: 'req-exp-fail', + userMessage: { + role: 'user', + content: [{ type: 'text', text: '部署 headscale' }], + metadata: { displayText: '部署 headscale' }, + }, + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); + assert.equal(recorded.length, 1); + assert.equal(recorded[0].result, 'failure'); + assert.match(recorded[0].body, /goosed unavailable/); +}); + test('terminal agent run quiesces session extensions after preserving Finish', async () => { const pool = createFakePool(); const quiesced = []; diff --git a/db.mjs b/db.mjs index a4835db..3fc5442 100644 --- a/db.mjs +++ b/db.mjs @@ -334,13 +334,47 @@ export async function migrateSchema(pool) { source_user_id CHAR(36) NULL, use_count BIGINT NOT NULL DEFAULT 0, embedding JSON NULL, + problem VARCHAR(512) NULL, + environment_json JSON NULL, + hypothesis VARCHAR(512) NULL, + action_json JSON NULL, + result ENUM('success','partial','failure','unknown') NULL, + confidence DECIMAL(4,3) NULL, + evidence_json JSON NULL, + parent_experience_id CHAR(36) NULL, + supersedes_id CHAR(36) NULL, + status ENUM('active','archived','deleted') NOT NULL DEFAULT 'active', created_at BIGINT NOT NULL, updated_at BIGINT NOT NULL, KEY idx_h5_experience_scope_updated (scope, updated_at), + KEY idx_h5_experience_status_updated (status, updated_at), FULLTEXT KEY ftx_h5_experience (title, body) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci `); + const experienceV1Columns = [ + ['problem', 'VARCHAR(512) NULL'], + ['environment_json', 'JSON NULL'], + ['hypothesis', 'VARCHAR(512) NULL'], + ['action_json', 'JSON NULL'], + ["result", "ENUM('success','partial','failure','unknown') NULL"], + ['confidence', 'DECIMAL(4,3) NULL'], + ['evidence_json', 'JSON NULL'], + ['parent_experience_id', 'CHAR(36) NULL'], + ['supersedes_id', 'CHAR(36) NULL'], + ['status', "ENUM('active','archived','deleted') NOT NULL DEFAULT 'active'"], + ]; + for (const [column, definition] of experienceV1Columns) { + if (!(await columnExists(pool, 'h5_experience', column))) { + await pool.query(`ALTER TABLE h5_experience ADD COLUMN \`${column}\` ${definition}`); + } + } + if (!(await indexExists(pool, 'h5_experience', 'idx_h5_experience_status_updated'))) { + await pool.query( + 'ALTER TABLE h5_experience ADD KEY idx_h5_experience_status_updated (status, updated_at)', + ); + } + // Per-user conversation memory: raw dialogue rows plus lightweight extracted // memory items. This is intentionally additive and can be disabled by env. await pool.query(` diff --git a/experience-extractor.mjs b/experience-extractor.mjs new file mode 100644 index 0000000..f599867 --- /dev/null +++ b/experience-extractor.mjs @@ -0,0 +1,221 @@ +import { deriveUserFacingText } from './conversation-display.mjs'; +import { normalizeExperienceRecordInput } from './experience-schema.mjs'; + +const RUN_METADATA_KEY = 'memindRun'; + +function safeJsonParse(value, fallback = null) { + if (value == null || value === '') return fallback; + if (typeof value === 'object') return value; + try { + return JSON.parse(String(value)); + } catch { + return fallback; + } +} + +function extractRunMessageText(userMessage) { + if (typeof userMessage?.content === 'string') return userMessage.content; + if (!Array.isArray(userMessage?.content)) return ''; + return userMessage.content + .filter((item) => item?.type === 'text') + .map((item) => String(item.text ?? '').trim()) + .filter(Boolean) + .join('\n'); +} + +export function extractAgentRunProblem(row) { + const userMessage = safeJsonParse(row?.user_message_json, {}) ?? {}; + const displayText = String(userMessage?.metadata?.displayText ?? '').trim(); + const problem = displayText || deriveUserFacingText(extractRunMessageText(userMessage)); + return problem.trim(); +} + +function resolveRunOptions(userMessage) { + const metadata = userMessage?.metadata ?? {}; + const runMetadata = metadata?.[RUN_METADATA_KEY] ?? metadata?.agentRun ?? {}; + return { + toolMode: String(runMetadata?.toolMode ?? metadata?.toolMode ?? 'chat').trim().toLowerCase(), + taskType: String(runMetadata?.taskType ?? metadata?.taskType ?? '').trim(), + requiredExecutor: String(runMetadata?.executor ?? runMetadata?.requiredExecutor ?? '').trim().toLowerCase(), + suggestedSkill: String( + runMetadata?.selectedChatSkill ?? metadata?.selectedChatSkill ?? '', + ).trim(), + }; +} + +function summarizeToolEvidence(toolEvidence) { + if (!toolEvidence) return ''; + const calls = Array.isArray(toolEvidence.calls) + ? toolEvidence.calls.filter(Boolean) + : []; + if (calls.length > 0) { + return `工具调用: ${calls.join(', ')}`; + } + if (toolEvidence.generateImage?.succeeded === true) { + return '图片生成完成'; + } + return ''; +} + +function summarizeRunEvents(events = []) { + for (const event of events) { + const data = safeJsonParse(event?.data_json ?? event?.dataJson, {}) ?? {}; + if (event.event_type === 'tool_gateway_result' || event.eventType === 'tool_gateway_result') { + const stdout = String(data.stdoutTail ?? '').trim(); + const stderr = String(data.stderrTail ?? '').trim(); + const executor = String(data.executor ?? '').trim(); + const parts = []; + if (executor) parts.push(`执行器 ${executor}`); + if (stdout) parts.push(stdout); + if (stderr) parts.push(`stderr: ${stderr}`); + if (parts.length > 0) return parts.join('\n').slice(0, 2000); + } + if (event.event_type === 'direct_chat_completed' || event.eventType === 'direct_chat_completed') { + const model = String(data.model ?? '').trim(); + return model ? `Direct chat 完成 (${model})` : 'Direct chat 完成'; + } + if (event.event_type === 'session_finished' || event.eventType === 'session_finished') { + const calls = Array.isArray(data.toolCalls) ? data.toolCalls.filter(Boolean) : []; + if (calls.length > 0) return `会话完成,工具: ${calls.join(', ')}`; + return 'Goose 会话完成'; + } + } + return ''; +} + +export function buildAgentRunExperienceBody({ + status, + errorMessage = null, + toolEvidence = null, + eventSummary = null, + routing = null, +}) { + if (status === 'failed') { + const message = String(errorMessage ?? '').trim(); + if (message) return message.slice(0, 4000); + return 'Agent run failed'; + } + const parts = [ + summarizeToolEvidence(toolEvidence), + String(eventSummary ?? '').trim(), + routing?.suggestedSkill + ? `技能: ${routing.suggestedSkill}` + : routing?.route + ? `路由: ${routing.route}` + : '', + 'Agent run completed successfully.', + ].filter(Boolean); + return parts.join('\n').slice(0, 4000); +} + +export function buildAgentRunExperiencePayload({ + row, + runId, + status, + sessionId = null, + errorMessage = null, + toolEvidence = null, + routing = null, + eventSummary = null, +} = {}) { + const terminalStatus = String(status ?? '').trim().toLowerCase(); + if (!['succeeded', 'failed'].includes(terminalStatus)) return null; + + const problem = extractAgentRunProblem(row); + const body = buildAgentRunExperienceBody({ + status: terminalStatus, + errorMessage, + toolEvidence, + eventSummary, + routing, + }); + if (!problem || !body) return null; + + const userMessage = safeJsonParse(row?.user_message_json, {}) ?? {}; + const runOptions = resolveRunOptions(userMessage); + const executor = runOptions.requiredExecutor + || (routing?.route === 'direct_chat' ? 'direct-chat' : 'goose'); + const components = [ + routing?.route ?? null, + runOptions.toolMode !== 'chat' ? runOptions.toolMode : null, + runOptions.requiredExecutor || null, + runOptions.suggestedSkill || routing?.suggestedSkill || null, + 'portal', + ].filter(Boolean); + + return normalizeExperienceRecordInput({ + kind: 'task_outcome', + title: problem.slice(0, 200), + body, + problem, + result: terminalStatus === 'succeeded' ? 'success' : 'failure', + confidence: terminalStatus === 'succeeded' ? 0.7 : 0.6, + sourceSessionId: sessionId, + sourceUserId: row?.user_id ?? null, + environment: { + runtime: 'agent-run-gateway', + components, + }, + action: { + executor, + steps: Array.isArray(toolEvidence?.calls) ? toolEvidence.calls : ['agent_run_complete'], + artifacts: [], + }, + evidence: { + sources: [{ + source_id: `run:${runId}`, + source_type: 'agent_run', + actor: executor, + timestamp_ms: Date.now(), + modality: terminalStatus === 'failed' ? 'run_failure' : 'run_result', + }], + provenance: { + run_id: runId, + session_id: sessionId, + user_id: row?.user_id ?? null, + }, + }, + }); +} + +export async function loadAgentRunEventSummary(pool, runId) { + if (!pool?.query || !runId) return ''; + const [rows] = await pool.query( + `SELECT event_type, data_json + FROM h5_agent_run_events + WHERE run_id = ? + ORDER BY created_at DESC + LIMIT 20`, + [runId], + ); + return summarizeRunEvents(Array.isArray(rows) ? rows : []); +} + +export async function extractAgentRunExperience({ + experienceService, + pool = null, + row, + runId, + status, + sessionId = null, + errorMessage = null, + toolEvidence = null, + routing = null, +}) { + if (!experienceService || typeof experienceService.record !== 'function') { + return null; + } + const eventSummary = pool ? await loadAgentRunEventSummary(pool, runId).catch(() => '') : ''; + const payload = buildAgentRunExperiencePayload({ + row, + runId, + status, + sessionId, + errorMessage, + toolEvidence, + routing, + eventSummary, + }); + if (!payload) return null; + return experienceService.record(payload); +} diff --git a/experience-extractor.test.mjs b/experience-extractor.test.mjs new file mode 100644 index 0000000..2735d72 --- /dev/null +++ b/experience-extractor.test.mjs @@ -0,0 +1,97 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + buildAgentRunExperienceBody, + buildAgentRunExperiencePayload, + extractAgentRunProblem, + extractAgentRunExperience, +} from './experience-extractor.mjs'; + +function sampleRow(overrides = {}) { + return { + user_id: 'user-1', + user_message_json: JSON.stringify({ + role: 'user', + content: [{ type: 'text', text: '帮我修复 SSE 断线问题' }], + metadata: { + displayText: '帮我修复 SSE 断线问题', + memindRun: { + toolMode: 'chat', + selectedChatSkill: 'page-data-collect', + }, + }, + }), + ...overrides, + }; +} + +test('extractAgentRunProblem prefers displayText', () => { + assert.equal( + extractAgentRunProblem(sampleRow()), + '帮我修复 SSE 断线问题', + ); +}); + +test('buildAgentRunExperiencePayload maps succeeded runs with evidence provenance', () => { + const payload = buildAgentRunExperiencePayload({ + row: sampleRow(), + runId: 'run-1', + status: 'succeeded', + sessionId: 'sess-1', + routing: { route: 'agent_orchestration', suggestedSkill: 'page-data-collect' }, + toolEvidence: { calls: ['edit_file'] }, + }); + assert.equal(payload.kind, 'task_outcome'); + assert.equal(payload.result, 'success'); + assert.equal(payload.problem, '帮我修复 SSE 断线问题'); + assert.equal(payload.sourceSessionId, 'sess-1'); + assert.equal(payload.evidence.provenance.run_id, 'run-1'); + assert.match(payload.body, /edit_file/); +}); + +test('buildAgentRunExperiencePayload maps failed runs to failure result', () => { + const payload = buildAgentRunExperiencePayload({ + row: sampleRow(), + runId: 'run-2', + status: 'failed', + errorMessage: 'delivery failed', + }); + assert.equal(payload.result, 'failure'); + assert.equal(payload.body, 'delivery failed'); +}); + +test('buildAgentRunExperienceBody falls back for successful runs without tool evidence', () => { + const body = buildAgentRunExperienceBody({ + status: 'succeeded', + routing: { route: 'direct_chat' }, + }); + assert.match(body, /Agent run completed successfully/); +}); + +test('extractAgentRunExperience records through experience service', async () => { + const recorded = []; + const pool = { + async query() { + return [[{ + event_type: 'session_finished', + data_json: JSON.stringify({ toolCalls: ['write_file'] }), + }]]; + }, + }; + const saved = await extractAgentRunExperience({ + experienceService: { + async record(payload) { + recorded.push(payload); + return { id: 'exp-1', ...payload }; + }, + }, + pool, + row: sampleRow(), + runId: 'run-3', + status: 'succeeded', + sessionId: 'sess-3', + }); + assert.equal(recorded.length, 1); + assert.equal(saved.id, 'exp-1'); + assert.match(recorded[0].body, /write_file/); +}); diff --git a/experience-schema.mjs b/experience-schema.mjs new file mode 100644 index 0000000..9ad919a --- /dev/null +++ b/experience-schema.mjs @@ -0,0 +1,206 @@ +export const EXPERIENCE_RESULTS = Object.freeze(['success', 'partial', 'failure', 'unknown']); +export const EXPERIENCE_STATUSES = Object.freeze(['active', 'archived', 'deleted']); +export const EXPERIENCE_KINDS = Object.freeze([ + 'lesson', + 'task_outcome', + 'execution_pattern', + 'skill_candidate', +]); + +export const EXPERIENCE_V1_MYSQL_DDL = Object.freeze([ + 'problem VARCHAR(512) NULL', + 'environment_json JSON NULL', + 'hypothesis VARCHAR(512) NULL', + 'action_json JSON NULL', + "result ENUM('success','partial','failure','unknown') NULL", + 'confidence DECIMAL(4,3) NULL', + 'evidence_json JSON NULL', + 'parent_experience_id CHAR(36) NULL', + 'supersedes_id CHAR(36) NULL', + "status ENUM('active','archived','deleted') NOT NULL DEFAULT 'active'", +]); + +function nonEmptyString(value, maxLen = null) { + const text = String(value ?? '').trim(); + if (!text) return null; + return maxLen ? text.slice(0, maxLen) : text; +} + +function normalizeConfidence(value) { + if (value == null || value === '') return null; + const parsed = Number(value); + if (!Number.isFinite(parsed)) return null; + return Math.min(1, Math.max(0, parsed)); +} + +function normalizeResult(value) { + const text = String(value ?? '').trim().toLowerCase(); + if (!text) return null; + return EXPERIENCE_RESULTS.includes(text) ? text : null; +} + +function normalizeStatus(value) { + const text = String(value ?? 'active').trim().toLowerCase(); + return EXPERIENCE_STATUSES.includes(text) ? text : 'active'; +} + +function normalizeKind(value) { + const text = String(value ?? 'lesson').trim().toLowerCase(); + return EXPERIENCE_KINDS.includes(text) ? text : 'lesson'; +} + +export function serializeExperienceJson(value) { + if (value == null) return null; + if (typeof value === 'object') return JSON.stringify(value); + const trimmed = String(value).trim(); + if (!trimmed) return null; + JSON.parse(trimmed); + return trimmed; +} + +export function parseExperienceJson(value, fallback = null) { + if (value == null) return fallback; + if (typeof value === 'object') return value; + try { + return JSON.parse(String(value)); + } catch { + return fallback; + } +} + +/** + * Normalize caller input into stable DB-bound fields. Title/body remain required + * at the service layer; structured fields are optional. + */ +export function normalizeExperienceRecordInput(input = {}) { + return { + scope: nonEmptyString(input.scope, 64) ?? 'global', + kind: normalizeKind(input.kind), + title: nonEmptyString(input.title, 255), + body: nonEmptyString(input.body), + tags: Array.isArray(input.tags) + ? [...new Set(input.tags.map((tag) => String(tag).trim()).filter(Boolean))] + : [], + sourceSessionId: nonEmptyString(input.sourceSessionId, 128), + sourceUserId: nonEmptyString(input.sourceUserId, 36), + problem: nonEmptyString(input.problem, 512), + environment: parseExperienceJson(input.environment ?? input.environmentJson, null), + hypothesis: nonEmptyString(input.hypothesis, 512), + action: parseExperienceJson(input.action ?? input.actionJson, null), + result: normalizeResult(input.result), + confidence: normalizeConfidence(input.confidence), + evidence: parseExperienceJson(input.evidence ?? input.evidenceJson, null), + parentExperienceId: nonEmptyString(input.parentExperienceId, 36), + supersedesId: nonEmptyString(input.supersedesId, 36), + status: normalizeStatus(input.status), + }; +} + +export function mapExperienceRow(row) { + return { + id: row.id, + scope: row.scope, + kind: row.kind, + title: row.title, + body: row.body, + tags: parseExperienceJson(row.tags_json ?? row.tags, []) ?? [], + 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), + problem: row.problem ?? null, + environment: parseExperienceJson(row.environment_json, null), + hypothesis: row.hypothesis ?? null, + action: parseExperienceJson(row.action_json, null), + result: row.result ?? null, + confidence: row.confidence == null ? null : Number(row.confidence), + evidence: parseExperienceJson(row.evidence_json, null), + parentExperienceId: row.parent_experience_id ?? null, + supersedesId: row.supersedes_id ?? null, + status: row.status ?? 'active', + }; +} + +export function experienceSearchHaystack(row) { + const environmentText = row.environment_json + ? JSON.stringify(parseExperienceJson(row.environment_json, {})) + : ''; + return [ + row.title, + row.body, + row.problem, + environmentText, + ] + .filter(Boolean) + .join('\n') + .toLowerCase(); +} + +export function formatExperienceInjectionBlock(hits = []) { + if (!Array.isArray(hits) || hits.length === 0) return ''; + const lines = hits.map((hit) => formatExperienceInjectionLine(hit)); + return `# 相关经验(供参考,来自历史任务)\n${lines.join('\n')}`; +} + +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 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(' · ')}`); + } + return parts.join('\n'); +} + +/** + * Build a structured task_outcome payload for MindSpace agent jobs. + */ +export function buildMindspaceJobExperiencePayload({ claim, sessionId, parsed, jobId = null } = {}) { + const title = String(parsed?.title ?? claim?.instruction ?? '').trim().slice(0, 200); + const summary = String(parsed?.summary ?? '').trim(); + if (!title || !summary) return null; + + const runRef = jobId ? String(jobId) : null; + return normalizeExperienceRecordInput({ + kind: 'task_outcome', + title, + body: summary, + problem: claim?.instruction ?? title, + result: parsed?.status === 'failed' ? 'failure' : 'success', + sourceSessionId: sessionId ?? null, + sourceUserId: claim?.userId ?? null, + environment: { + runtime: 'mindspace-agent-job', + components: ['goose', 'portal'], + }, + action: { + executor: 'goose', + steps: ['agent_job_complete'], + artifacts: Array.isArray(parsed?.sourceAssetIds) ? parsed.sourceAssetIds : [], + }, + evidence: { + sources: runRef + ? [{ + source_id: `job:${runRef}`, + source_type: 'agent_job', + actor: 'goose', + timestamp_ms: Date.now(), + modality: 'task_result', + }] + : [], + provenance: { + session_id: sessionId ?? null, + user_id: claim?.userId ?? null, + job_id: runRef, + }, + }, + }); +} diff --git a/experience-schema.test.mjs b/experience-schema.test.mjs new file mode 100644 index 0000000..5e82e52 --- /dev/null +++ b/experience-schema.test.mjs @@ -0,0 +1,110 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + buildMindspaceJobExperiencePayload, + formatExperienceInjectionBlock, + formatExperienceInjectionLine, + mapExperienceRow, + normalizeExperienceRecordInput, + serializeExperienceJson, +} from './experience-schema.mjs'; + +test('normalizeExperienceRecordInput dedupes tags and defaults kind/status', () => { + const normalized = normalizeExperienceRecordInput({ + title: ' 标题 ', + body: '正文', + tags: ['a', 'a', ' b ', ''], + kind: 'TASK_OUTCOME', + status: 'ACTIVE', + }); + assert.equal(normalized.kind, 'task_outcome'); + assert.equal(normalized.status, 'active'); + assert.deepEqual(normalized.tags, ['a', 'b']); +}); + +test('normalizeExperienceRecordInput clamps confidence and validates result', () => { + const normalized = normalizeExperienceRecordInput({ + title: 't', + body: 'b', + confidence: 1.5, + result: 'SUCCESS', + }); + assert.equal(normalized.confidence, 1); + assert.equal(normalized.result, 'success'); +}); + +test('serializeExperienceJson accepts objects and valid JSON strings', () => { + assert.equal(serializeExperienceJson({ a: 1 }), '{"a":1}'); + assert.equal(serializeExperienceJson('{"a":1}'), '{"a":1}'); + assert.equal(serializeExperienceJson(''), null); +}); + +test('mapExperienceRow parses JSON columns', () => { + const mapped = mapExperienceRow({ + id: 'id-1', + scope: 'global', + kind: 'lesson', + title: 't', + body: 'b', + tags_json: '["x"]', + use_count: 2, + environment_json: '{"runtime":"portal"}', + evidence_json: '{"sources":[]}', + status: 'active', + created_at: 1, + updated_at: 2, + }); + assert.deepEqual(mapped.tags, ['x']); + assert.deepEqual(mapped.environment, { runtime: 'portal' }); + assert.equal(mapped.useCount, 2); +}); + +test('formatExperienceInjectionLine includes structured hints', () => { + const line = formatExperienceInjectionLine({ + title: '部署', + body: '摘要', + problem: '公网访问失败', + result: 'success', + confidence: 0.9, + environment: { runtime: '103', components: ['caddy', 'portal'] }, + }); + assert.match(line, /部署: 摘要/); + assert.match(line, /问题: 公网访问失败/); + assert.match(line, /结果: success, conf=0.9/); + assert.match(line, /环境: 103 · caddy \+ portal/); +}); + +test('formatExperienceInjectionBlock wraps lines with header', () => { + const block = formatExperienceInjectionBlock([ + { title: 'A', body: 'one' }, + { title: 'B', body: 'two' }, + ]); + assert.match(block, /^# 相关经验/); + assert.match(block, /- A: one/); + assert.match(block, /- B: two/); +}); + +test('buildMindspaceJobExperiencePayload maps failed jobs to failure result', () => { + const payload = buildMindspaceJobExperiencePayload({ + claim: { instruction: '生成页面', userId: 'u1' }, + sessionId: 's1', + parsed: { title: '页面标题', summary: '完成摘要', status: 'failed' }, + jobId: 'job-42', + }); + assert.equal(payload.kind, 'task_outcome'); + assert.equal(payload.result, 'failure'); + assert.equal(payload.problem, '生成页面'); + assert.equal(payload.sourceSessionId, 's1'); + assert.equal(payload.evidence.provenance.job_id, 'job-42'); +}); + +test('buildMindspaceJobExperiencePayload returns null when title or summary missing', () => { + assert.equal( + buildMindspaceJobExperiencePayload({ + claim: { instruction: 'x' }, + sessionId: 's', + parsed: { title: '', summary: 'ok' }, + }), + null, + ); +}); diff --git a/experience-service-pg.mjs b/experience-service-pg.mjs index 360bcef..03e8d35 100644 --- a/experience-service-pg.mjs +++ b/experience-service-pg.mjs @@ -1,5 +1,12 @@ import crypto from 'node:crypto'; +import { + experienceSearchHaystack, + mapExperienceRow, + normalizeExperienceRecordInput, + serializeExperienceJson, +} from './experience-schema.mjs'; + /** * PostgreSQL + pgvector backed experience store (etat C, polyglot mode). * @@ -51,10 +58,35 @@ export async function createPgExperienceService(options = {}) { 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)', ); @@ -66,80 +98,88 @@ export async function createPgExperienceService(options = {}) { // ivfflat needs the extension + may warn on empty table; non-fatal. }); - function normalizeTags(tags) { - if (!Array.isArray(tags)) return []; - return [...new Set(tags.map((t) => String(t).trim()).filter(Boolean))]; - } - function toVectorLiteral(vec) { // pgvector accepts a string like '[0.1,0.2,...]' return `[${vec.map((n) => Number(n)).join(',')}]`; } function rowToExperience(row) { - return { - id: row.id, - scope: row.scope, - kind: row.kind, - title: row.title, - body: row.body, - tags: Array.isArray(row.tags) ? row.tags : [], - 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), - }; + return mapExperienceRow({ + ...row, + tags_json: row.tags, + }); } 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 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 scope = String(input?.scope ?? 'global').trim() || 'global'; - const kind = String(input?.kind ?? 'lesson').trim() || 'lesson'; - const tags = normalizeTags(input?.tags); + const environmentJson = normalized.environment; + const actionJson = normalized.action; + const evidenceJson = normalized.evidence; let embedding = null; if (embed) { try { - const vec = await embed(`${title}\n${body}`); + const vec = await embed(`${normalized.title}\n${normalized.body}`); if (Array.isArray(vec) && vec.length === embeddingDim) embedding = toVectorLiteral(vec); } catch { - embedding = null; // embedding failure must not block recording + embedding = null; } } await pool.query( `INSERT INTO h5_experience (id, scope, kind, title, body, tags, source_session_id, source_user_id, - use_count, embedding, created_at, updated_at) - VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7,$8,0,$9::vector,$10,$11)`, + 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, - scope, - kind, - title, - body, - JSON.stringify(tags), - input?.sourceSessionId ?? null, - input?.sourceUserId ?? null, + 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, ], ); return rowToExperience({ id, - scope, - kind, - title, - body, - tags, - source_session_id: input?.sourceSessionId ?? null, - source_user_id: input?.sourceUserId ?? null, + 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, }); @@ -157,9 +197,11 @@ export async function createPgExperienceService(options = {}) { 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, created_at, updated_at + 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 embedding IS NOT NULL + 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], @@ -181,9 +223,11 @@ export async function createPgExperienceService(options = {}) { const params = [scope, ...terms.map((t) => `%${t}%`), max]; const { rows } = await pool.query( `SELECT id, scope, kind, title, body, tags, source_session_id, - source_user_id, use_count, created_at, updated_at + 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 (${likeClauses.join(' OR ')}) + WHERE scope = $1 AND status = 'active' AND (${likeClauses.join(' OR ')}) ORDER BY updated_at DESC LIMIT $${terms.length + 2}`, params, diff --git a/experience-service.mjs b/experience-service.mjs index 5e4c3eb..e8e98f6 100644 --- a/experience-service.mjs +++ b/experience-service.mjs @@ -1,5 +1,12 @@ import crypto from 'node:crypto'; +import { + experienceSearchHaystack, + mapExperienceRow, + normalizeExperienceRecordInput, + serializeExperienceJson, +} from './experience-schema.mjs'; + /** * Shared experience store (etat C of the goose scale plan). * @@ -16,58 +23,22 @@ import crypto from 'node:crypto'; */ 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(); + const haystack = experienceSearchHaystack(row); let keywordScore = 0; - for (const term of terms) { + 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); - const score = keywordScore * recency; - return { row, score }; + return { row, score: keywordScore * recency }; }) .filter((entry) => entry.score > 0) .sort((a, b) => b.score - a.score); @@ -89,43 +60,66 @@ export function createExperienceService(pool, options = {}) { * 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 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 scope = String(input?.scope ?? 'global').trim() || 'global'; - const kind = String(input?.kind ?? 'lesson').trim() || 'lesson'; - const tags = normalizeTags(input?.tags); + 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, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`, + 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, - scope, - kind, - title, - body, - JSON.stringify(tags), - input?.sourceSessionId ?? null, - input?.sourceUserId ?? null, + 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, ], ); - return rowToExperience({ + + return mapExperienceRow({ id, - scope, - kind, - title, - body, - tags_json: tags, - source_session_id: input?.sourceSessionId ?? null, - source_user_id: input?.sourceUserId ?? null, + 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, }); @@ -138,20 +132,19 @@ export function createExperienceService(pool, options = {}) { 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 + 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 = ? + 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) { - // 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 @@ -161,7 +154,7 @@ export function createExperienceService(pool, options = {}) { ) .catch(() => {}); } - return ranked.map((entry) => rowToExperience(entry.row)); + return ranked.map((entry) => mapExperienceRow(entry.row)); } /** @@ -179,3 +172,5 @@ export function createExperienceService(pool, options = {}) { function experienceError(message, code) { return Object.assign(new Error(message), { code }); } + +export { formatExperienceInjectionBlock, formatExperienceInjectionLine } from './experience-schema.mjs'; diff --git a/experience-service.test.mjs b/experience-service.test.mjs index 06836f2..628fcfb 100644 --- a/experience-service.test.mjs +++ b/experience-service.test.mjs @@ -17,6 +17,16 @@ function createMockPool() { tagsJson, sourceSessionId, sourceUserId, + problem, + environmentJson, + hypothesis, + actionJson, + result, + confidence, + evidenceJson, + parentExperienceId, + supersedesId, + status, createdAt, updatedAt, ] = params; @@ -30,6 +40,16 @@ function createMockPool() { source_session_id: sourceSessionId, source_user_id: sourceUserId, use_count: 0, + problem, + environment_json: environmentJson, + hypothesis, + action_json: actionJson, + result, + confidence, + evidence_json: evidenceJson, + parent_experience_id: parentExperienceId, + supersedes_id: supersedesId, + status, created_at: createdAt, updated_at: updatedAt, }); @@ -38,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) + .filter((r) => r.scope === scope && r.status === 'active') .sort((a, b) => b.updated_at - a.updated_at); return [matched.map((r) => ({ ...r }))]; } @@ -67,6 +87,28 @@ test('record persists an experience with normalized fields', async () => { assert.equal(saved.kind, 'lesson'); assert.deepEqual(saved.tags, ['headscale', '网络']); assert.equal(saved.useCount, 0); + assert.equal(saved.status, 'active'); +}); + +test('record persists V1 structured fields', async () => { + const { pool } = createMockPool(); + const svc = createExperienceService(pool, { now: () => 2000 }); + const saved = await svc.record({ + kind: 'task_outcome', + title: 'SSE 断线恢复', + body: 'Portal replay 需映射 Goose Last-Event-ID', + problem: 'H5 SSE 断线后消息丢失', + environment: { runtime: 'portal', components: ['goose', 'sse'] }, + action: { executor: 'goose', steps: ['replay'] }, + result: 'success', + confidence: 0.85, + evidence: { sources: [{ source_id: 'job:abc', source_type: 'agent_job' }] }, + }); + assert.equal(saved.kind, 'task_outcome'); + assert.equal(saved.problem, 'H5 SSE 断线后消息丢失'); + assert.deepEqual(saved.environment, { runtime: 'portal', components: ['goose', 'sse'] }); + assert.equal(saved.result, 'success'); + assert.equal(saved.confidence, 0.85); }); test('record rejects empty title or body', async () => { @@ -89,6 +131,28 @@ test('search ranks keyword matches and ignores empty queries', async () => { assert.equal(hits[0].title, 'Headscale 部署'); }); +test('search matches problem and environment haystack', async () => { + const { pool } = createMockPool(); + const svc = createExperienceService(pool, { now: () => 3000 }); + await svc.record({ + title: '通用标题', + body: '正文不含关键词', + problem: 'MemFuse recall 排序偏低', + environment: { runtime: 'memory-v2-pgvector' }, + }); + const hits = await svc.search('memfuse recall'); + assert.equal(hits.length, 1); + assert.equal(hits[0].problem, 'MemFuse recall 排序偏低'); +}); + +test('search excludes non-active rows', async () => { + const { pool } = createMockPool(); + const svc = createExperienceService(pool, { now: () => 4000 }); + await svc.record({ title: '已归档', body: 'archived keyword match', status: 'archived' }); + const hits = await svc.search('archived keyword'); + assert.equal(hits.length, 0); +}); + test('search applies recency decay so newer wins on equal keyword score', async () => { const { pool } = createMockPool(); let clock = 0; @@ -98,7 +162,7 @@ test('search applies recency decay so newer wins on equal keyword score', async }); clock = 0; await svc.record({ title: '部署指南 A', body: 'deploy 部署 指南' }); - clock = 10_000; // 10 half-lives newer + clock = 10_000; await svc.record({ title: '部署指南 B', body: 'deploy 部署 指南' }); clock = 10_000; const hits = await svc.search('部署 deploy 指南', { limit: 2 }); diff --git a/mindspace-agent-runner.mjs b/mindspace-agent-runner.mjs index a7db5bf..6281560 100644 --- a/mindspace-agent-runner.mjs +++ b/mindspace-agent-runner.mjs @@ -6,6 +6,8 @@ import { jsonrepair } from 'jsonrepair'; import { reconcileAgentSession } from './session-reconcile.mjs'; import { resolveSessionAccess } from './session-broker.mjs'; import { resolveBillingTokenState } from './billing-token-state.mjs'; +import { buildMindspaceJobExperiencePayload } from './experience-schema.mjs'; +import { formatExperienceInjectionBlock } from './experience-service.mjs'; const insecureDispatcher = new Agent({ connect: { rejectUnauthorized: false }, @@ -429,7 +431,7 @@ export function createMindSpaceAgentRunner({ sourceAssetIds: claim.allowedAssets.map((asset) => asset.assetId), }); // Record the outcome as shared experience so other instances benefit. - await recordExperience(claim, sessionId, parsed); + await recordExperience(jobId, claim, sessionId, parsed); return completed; } catch (error) { if (claim?.jobToken) { @@ -452,27 +454,24 @@ export function createMindSpaceAgentRunner({ try { const hits = await experienceService.search(instruction, { limit: 3 }); if (!hits.length) return ''; - const lines = hits.map((h) => `- ${h.title}: ${h.body}`).join('\n'); - return `# 相关经验(供参考,来自历史任务)\n${lines}`; + return formatExperienceInjectionBlock(hits); } catch { return ''; } } // Persists a one-line lesson from a completed job. Best-effort; never throws. - async function recordExperience(claim, sessionId, parsed) { + async function recordExperience(jobId, claim, sessionId, parsed) { if (!experienceService) return; try { - const title = String(parsed?.title ?? claim.instruction ?? '').slice(0, 200); - const summary = String(parsed?.summary ?? '').trim(); - if (!title || !summary) return; - await experienceService.record({ - kind: 'task_outcome', - title, - body: summary, - sourceSessionId: sessionId, - sourceUserId: claim.userId, + const payload = buildMindspaceJobExperiencePayload({ + claim, + sessionId, + parsed, + jobId, }); + if (!payload) return; + await experienceService.record(payload); } catch { // swallow — experience recording is non-critical } diff --git a/package.json b/package.json index 1e842f3..54eae49 100644 --- a/package.json +++ b/package.json @@ -93,6 +93,9 @@ "dev:page-data-aider-loop": "node scripts/page-data-aider-dev-loop.mjs", "ci:page-data-dev-loop-smoke": "node scripts/ci-page-data-dev-loop-smoke.mjs", "migrate:agent-code-run-config": "node scripts/migrate-agent-code-run-config-from-env.mjs", + "migrate:h5-experience-v1": "node scripts/migrate-h5-experience-v1.mjs", + "verify:experience-v1": "node --test experience-schema.test.mjs experience-service.test.mjs experience-extractor.test.mjs", + "verify:experience-agent-run-local": "node scripts/verify-experience-agent-run-local.mjs", "test:scenario:john4-diet": "node scripts/run-scenario-test.mjs --scenario john4-children-hobby-diet-update", "test": "node --test api-core-retry.test.mjs auth.test.mjs asr-proxy.test.mjs billing.test.mjs billing-token-state.test.mjs billing-recharge.test.mjs wechat-pay.test.mjs wechat-oauth.test.mjs wechat-voice-reco.test.mjs wechat-mp.test.mjs wechat-media.test.mjs wechat/image-generation-policy.test.mjs wechat/verify/generated-thumbnail.test.mjs schedule-intent.test.mjs schedule-reminder-worker.test.mjs scheduled-task-intent.test.mjs scheduled-task-service.test.mjs scheduled-task-executor.test.mjs scheduled-task-worker.test.mjs scheduled-task-worker-config.test.mjs wechat/handlers/scheduled-task.test.mjs capabilities.test.mjs policies.test.mjs server/portal-api-auth-middleware.test.mjs server/portal-config-routes.test.mjs server/portal-plaza-discovery-routes.test.mjs server/portal-runtime-routes.test.mjs server/portal-gateway-services-bootstrap.test.mjs chat-skills.test.mjs chat-intent-router.test.mjs chat-finish-sync.test.mjs chat-agent-run-gate.test.mjs conversation-display.test.mjs user-publish.test.mjs user-memory-profile.test.mjs skills-registry.test.mjs skill-runtime-policy.test.mjs excel-analyst.test.mjs agent-run-gateway.test.mjs agent-run-routes.test.mjs session-broker.test.mjs sse-event-taxonomy.test.mjs goosed-proxy-boundary.test.mjs agent-run-stream.test.mjs mindspace-h5-html-finish-guard.test.mjs admin-routes.test.mjs services/orchestrator/admin-config.test.mjs services/orchestrator/contracts.test.mjs services/orchestrator/checkpoint.test.mjs services/orchestrator/runtime.test.mjs services/orchestrator/app.test.mjs services/orchestrator/server.test.mjs services/orchestrator/shadow-dispatcher.test.mjs services/orchestrator/shadow-observer.test.mjs services/orchestrator/observability.test.mjs services/orchestrator/executor-gateway.test.mjs services/orchestrator/executor-job-store.test.mjs image-make-admin-config.test.mjs asset-gateway.test.mjs image-make-client.test.mjs mindspace-image-generation.test.mjs mindspace-image-generation-routes.test.mjs mindspace-image-review.test.mjs mindspace-run-public-html-scope.test.mjs direct-chat-service.test.mjs tool-gateway.test.mjs mindspace.test.mjs mindspace-scan.test.mjs mindspace-assets.test.mjs mindspace-local-runtime-services.test.mjs mindspace-local-server-adapter.test.mjs mindspace-public-asset-token.test.mjs mindspace-remote-server-adapter.test.mjs mindspace-server-adapter.test.mjs mindspace-pages.test.mjs mindspace-page-sync-service.test.mjs public-site-bases.test.mjs mindspace-html-download-links.test.mjs mindspace-long-image.test.mjs mindspace-page-purge.test.mjs mindspace-public-delivery.test.mjs mindspace-public-page-context.test.mjs mindspace-published-page-csp.test.mjs mindspace-published-script-localize.test.mjs agent-run-deliverable-check.test.mjs mindspace-public-route.test.mjs mindspace-publications.test.mjs mindspace-public-links.test.mjs mindspace-chat-save.test.mjs mindspace-chat-save-service.test.mjs mindspace-chat-docx-package.test.mjs mindspace-public-finish-sync.test.mjs mindspace-wechat-html-delivery.test.mjs mindspace-chat-context.test.mjs mindspace-canonical-url.test.mjs mindspace-conversation-package.test.mjs mindspace-conversation-package-artifact-service.test.mjs mindspace-conversation-package-audit.test.mjs mindspace-conversation-package-backfill.test.mjs mindspace-conversation-package-public-html.test.mjs mindspace-conversation-package-verify.test.mjs mindspace-conversation-package-registry.test.mjs mindspace-conversation-package-routes.test.mjs mindspace-conversation-package-store.test.mjs mindspace-conversation-schema.test.mjs mindspace-runtime-config.test.mjs mindspace-config.test.mjs mindspace-analytics.test.mjs mindspace-service.test.mjs mindspace-storage-adapter.test.mjs mindspace-content-scan.test.mjs mindspace-html-localize.test.mjs mindspace-visual-editor.test.mjs mindspace-cleanup.test.mjs mindspace-thumbnails.test.mjs mindspace-workspace-thumbnails.test.mjs mindspace-workspace-sync.test.mjs mindspace-workspace-publication-delivery-service.test.mjs mindspace-workspace-tool-service.test.mjs mindspace-mcp-scoped-token.test.mjs mindspace-asset-preview.test.mjs mindspace-agent-jobs.test.mjs mindspace-agent-runner.test.mjs mindspace-sandbox-mcp.test.mjs mindspace-userdata-postgres.test.mjs postgres-user-data-space-service.test.mjs user-data-space-service.test.mjs page-data-routes.test.mjs page-access-policy.test.mjs page-access-visitor.test.mjs page-data-public-service.test.mjs page-data-integration.test.mjs page-data-log-store.test.mjs page-data-ops.test.mjs page-data-session-store.test.mjs page-data-browser-client.test.mjs page-data-policy-index.test.mjs message-stream.test.mjs mindspace-service/mindspace-rpc-server.test.mjs plaza-posts.test.mjs plaza-interactions.test.mjs plaza-algorithm.test.mjs plaza-seo.test.mjs plaza-ops.test.mjs user-auth.test.mjs llm-providers.test.mjs admin-guard.test.mjs user-feedback.test.mjs memory-v2.test.mjs memory-v2-admin-config.test.mjs memory-v2-lifecycle.test.mjs memory-v2-adapter-scaffold.test.mjs memory-v2-backend-contract.test.mjs memory-v2-health.test.mjs memory-v2-runtime.test.mjs memory-v2-plugin-backends.test.mjs memory-v2-pgvector.test.mjs memory-v2-pgvector-schema.test.mjs memory-v2-pgvector-backfill.test.mjs memory-v2-pgvector-smoke.test.mjs memory-v2-qdrant.test.mjs memory-v2-weaviate.test.mjs memory-v2-mem0.test.mjs memory-v2-letta.test.mjs memory-v2-external-adapters.test.mjs scripts/embed-memory-v2-local-hash.test.mjs scripts/check-memory-v2-app-canary.test.mjs scripts/check-memory-v2-config-gaps.test.mjs scripts/check-memory-v2-contracts.test.mjs scripts/check-memory-v2-health.test.mjs scripts/check-memory-v2-session-flow.test.mjs scripts/check-memory-v2-stack.test.mjs scripts/setup-memory-v2-pgvector-schema.test.mjs scripts/backfill-memory-v2-pgvector.test.mjs scripts/scaffold-memory-v2-backend.test.mjs scripts/smoke-memory-v2-pgvector.test.mjs scripts/smoke-memory-v2-qdrant.test.mjs scripts/smoke-memory-v2-external.test.mjs scripts/mock-memory-v2-services.test.mjs", "test:episodic-memory": "node --test episodic-memory.test.mjs direct-chat-service.test.mjs chat-intent-router.test.mjs", diff --git a/schema.sql b/schema.sql index 4b4dc21..bd130f0 100644 --- a/schema.sql +++ b/schema.sql @@ -1311,9 +1311,20 @@ CREATE TABLE IF NOT EXISTS h5_experience ( source_user_id CHAR(36) NULL, use_count BIGINT NOT NULL DEFAULT 0, embedding JSON NULL, + problem VARCHAR(512) NULL, + environment_json JSON NULL, + hypothesis VARCHAR(512) NULL, + action_json JSON NULL, + result ENUM('success','partial','failure','unknown') NULL, + confidence DECIMAL(4,3) NULL, + evidence_json JSON NULL, + parent_experience_id CHAR(36) NULL, + supersedes_id CHAR(36) NULL, + status ENUM('active','archived','deleted') NOT NULL DEFAULT 'active', created_at BIGINT NOT NULL, updated_at BIGINT NOT NULL, KEY idx_h5_experience_scope_updated (scope, updated_at), + KEY idx_h5_experience_status_updated (status, updated_at), FULLTEXT KEY ftx_h5_experience (title, body) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/scripts/agent-run-worker.mjs b/scripts/agent-run-worker.mjs index 7e240cf..cd99bb0 100644 --- a/scripts/agent-run-worker.mjs +++ b/scripts/agent-run-worker.mjs @@ -18,6 +18,7 @@ import { createDirectChatService } from '../direct-chat-service.mjs'; import { createSessionAccess } from '../session-broker.mjs'; import { createOrchestratorAdminConfigService } from '../services/orchestrator/admin-config.mjs'; import { createWorkflowShadowObserver } from '../services/orchestrator/shadow-observer.mjs'; +import { createExperienceService } from '../experience-service.mjs'; const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..'); @@ -196,6 +197,7 @@ async function bootstrapWorker() { apiSecret: process.env.TKMIND_SERVER__SECRET_KEY ?? 'local-dev-secret', userAuth, }); + const experienceService = createExperienceService(pool); const gateway = createAgentRunGateway({ pool, userAuth, @@ -206,6 +208,7 @@ async function bootstrapWorker() { sessionSnapshotService, conversationMemoryService, chatIntentRouter, + experienceService, conversationMemoryService, autoDispatch: false, maxConcurrentRuns: Number(process.env.MEMIND_AGENT_RUN_QUEUE_CONCURRENCY ?? 1), diff --git a/scripts/migrate-h5-experience-v1.mjs b/scripts/migrate-h5-experience-v1.mjs new file mode 100644 index 0000000..e20d437 --- /dev/null +++ b/scripts/migrate-h5-experience-v1.mjs @@ -0,0 +1,47 @@ +#!/usr/bin/env node +/** + * Idempotent migration: add h5_experience V1 columns (problem, environment, evidence, …). + * + * Usage: + * node scripts/migrate-h5-experience-v1.mjs + * node scripts/migrate-h5-experience-v1.mjs --dry-run + */ +import { loadH5Environment } from './load-env.mjs'; +import { createDbPool, migrateSchema } from '../db.mjs'; + +function parseArgs(argv) { + const options = { dryRun: false }; + for (const arg of argv) { + if (arg === '--dry-run') options.dryRun = true; + else if (arg === '-h' || arg === '--help') { + console.log('Usage: node scripts/migrate-h5-experience-v1.mjs [--dry-run]'); + process.exit(0); + } else { + throw new Error(`未知参数: ${arg}`); + } + } + return options; +} + +async function main() { + loadH5Environment(import.meta.dirname); + const options = parseArgs(process.argv.slice(2)); + + if (options.dryRun) { + console.log('[migrate-h5-experience-v1] dry-run: would run migrateSchema() (includes h5_experience V1 ALTERs)'); + return; + } + + const pool = createDbPool(); + try { + await migrateSchema(pool); + console.log('[migrate-h5-experience-v1] ok'); + } finally { + await pool.end(); + } +} + +main().catch((error) => { + console.error('[migrate-h5-experience-v1] failed:', error?.message ?? error); + process.exit(1); +}); diff --git a/scripts/verify-experience-agent-run-local.mjs b/scripts/verify-experience-agent-run-local.mjs new file mode 100644 index 0000000..a9fd8a4 --- /dev/null +++ b/scripts/verify-experience-agent-run-local.mjs @@ -0,0 +1,252 @@ +#!/usr/bin/env node +/** + * Local E2E: agent run terminal → h5_experience via Experience Extractor. + * + * Usage: + * node scripts/verify-experience-agent-run-local.mjs + * + * Requires: .env MySQL, goosed or direct-chat LLM path. + */ +import crypto from 'node:crypto'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { loadH5Environment } from './load-env.mjs'; +import { createAgentRunGateway } from '../agent-run-gateway.mjs'; +import { createManagedChatIntentRouter } from '../chat-intent-router.mjs'; +import { createConversationMemoryService } from '../conversation-memory.mjs'; +import { createDbPool } from '../db.mjs'; +import { createEpisodicMemoryService } from '../episodic-memory.mjs'; +import { createExperienceService } from '../experience-service.mjs'; +import { createLlmProviderService } from '../llm-providers.mjs'; +import { createMemoryV2AdminConfigService } from '../memory-v2-admin-config.mjs'; +import { createManagedMemoryV2Runtime } from '../memory-v2-runtime.mjs'; +import { createTkmindProxy } from '../tkmind-proxy.mjs'; +import { createToolGateway } from '../tool-gateway.mjs'; +import { createUserAuth } from '../user-auth.mjs'; +import { createSessionSnapshotService } from '../session-snapshot.mjs'; +import { createDirectChatService } from '../direct-chat-service.mjs'; +import { createSessionAccess } from '../session-broker.mjs'; + +const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..'); +const PORTAL = `http://127.0.0.1:${process.env.H5_PORT ?? 8081}`; +const USERNAME = process.env.MEMIND_E2E_USERNAME ?? 'john'; +const PASSWORD = process.env.JOHN_PASSWORD ?? process.env.H5_ACCESS_PASSWORD ?? '888888'; +const VERIFY_MESSAGE = process.env.EXPERIENCE_E2E_MESSAGE + ?? 'Experience E2E 验证:请用一句话回复收到即可'; +const MAX_WAIT_MS = Number(process.env.EXPERIENCE_E2E_WAIT_MS ?? 180_000); + +loadH5Environment(import.meta.dirname); + +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function loginViaApi(pool) { + const response = await fetch(`${PORTAL}/auth/login`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ username: USERNAME, password: PASSWORD }), + }); + const body = await response.json().catch(() => ({})); + if (response.ok && body?.authenticated && body.user?.id) { + return { userId: body.user.id }; + } + + const auth = createUserAuth(pool, { + usersRoot: process.env.H5_USERS_ROOT ?? path.join(root, 'users'), + h5Root: root, + }); + const result = await auth.login({ username: USERNAME, password: PASSWORD, ip: '127.0.0.1' }); + if (!result.ok || !result.user?.id) { + throw new Error(`登录失败: ${result.message ?? body?.message ?? 'unknown'}`); + } + return { userId: result.user.id }; +} + +async function resolveUserId(pool) { + const forcedUserId = String(process.env.EXPERIENCE_E2E_USER_ID ?? '').trim(); + if (forcedUserId) return forcedUserId; + + try { + const { userId } = await loginViaApi(pool); + if (userId) return userId; + } catch (error) { + console.warn('[experience-e2e] API/DB login skipped:', error instanceof Error ? error.message : error); + } + + const [rows] = await pool.query( + 'SELECT id FROM h5_users WHERE username = ? LIMIT 1', + [USERNAME], + ); + if (rows[0]?.id) { + console.log(`[experience-e2e] using user id from DB lookup: ${USERNAME}`); + return rows[0].id; + } + throw new Error(`找不到用户 ${USERNAME},可设置 EXPERIENCE_E2E_USER_ID`); +} + +async function waitForTerminal(pool, runId, timeoutMs) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const [rows] = await pool.query( + 'SELECT status, error_message FROM h5_agent_runs WHERE id = ? LIMIT 1', + [runId], + ); + const status = rows[0]?.status; + if (status === 'succeeded' || status === 'failed') { + return rows[0]; + } + await sleep(1000); + } + throw new Error(`run ${runId} 未在 ${timeoutMs}ms 内进入终态`); +} + +async function findExperienceForRun(pool, runId) { + const [rows] = await pool.query( + `SELECT id, kind, title, problem, result, status, + JSON_EXTRACT(evidence_json, '$.provenance.run_id') AS run_id, + created_at + FROM h5_experience + WHERE JSON_UNQUOTE(JSON_EXTRACT(evidence_json, '$.provenance.run_id')) = ? + ORDER BY created_at DESC + LIMIT 1`, + [runId], + ); + return rows[0] ?? null; +} + +async function bootstrapGateway(pool) { + const userAuth = createUserAuth(pool, { + usersRoot: process.env.H5_USERS_ROOT ?? path.join(root, 'users'), + h5Root: root, + }); + const llmProviderService = createLlmProviderService(pool, { + apiTarget: process.env.TKMIND_API_TARGET ?? 'https://127.0.0.1:18006', + apiSecret: process.env.TKMIND_SERVER__SECRET_KEY ?? 'local-dev-secret', + }); + const memoryV2ConfigService = createMemoryV2AdminConfigService(pool); + const getEffectiveEnv = async () => { + const state = await memoryV2ConfigService.getRuntimeState().catch(() => null); + return { ...process.env, ...(state?.overrides ?? {}) }; + }; + const conversationMemoryService = createConversationMemoryService(pool, { + llmProviderService, + getEffectiveEnv, + }); + const memoryV2 = await createManagedMemoryV2Runtime({ + legacyMemoryService: conversationMemoryService, + configService: memoryV2ConfigService, + mysqlPool: pool, + }); + const episodicMemoryService = createEpisodicMemoryService(pool, { + getEffectiveEnv, + logger: console, + }); + const sessionSnapshotService = createSessionSnapshotService(pool, { + conversationMemoryService, + memoryV2, + episodicMemoryService, + }); + const sessionAccess = createSessionAccess({ userAuth, enabled: false }); + const directChatService = createDirectChatService({ + userAuth, + sessionAccess, + llmProviderService, + sessionSnapshotService, + memoryV2, + conversationMemoryService, + episodicMemoryService, + }); + const chatIntentRouter = createManagedChatIntentRouter({ + llmProviderService, + memoryV2, + conversationMemoryService, + episodicMemoryService, + configService: memoryV2ConfigService, + }); + const tkmindProxy = createTkmindProxy({ + apiTarget: process.env.TKMIND_API_TARGET ?? 'https://127.0.0.1:18006', + apiTargets: String(process.env.TKMIND_API_TARGETS ?? '').split(',').filter(Boolean).length + ? String(process.env.TKMIND_API_TARGETS).split(',').map((s) => s.trim()).filter(Boolean) + : [process.env.TKMIND_API_TARGET ?? 'https://127.0.0.1:18006'], + apiSecret: process.env.TKMIND_SERVER__SECRET_KEY ?? 'local-dev-secret', + userAuth, + }); + const experienceService = createExperienceService(pool); + return createAgentRunGateway({ + pool, + userAuth, + sessionAccess, + tkmindProxy, + toolGateway: createToolGateway({ llmProviderService }), + directChatService, + sessionSnapshotService, + conversationMemoryService, + chatIntentRouter, + experienceService, + autoDispatch: false, + retryDelaysMs: [], + }); +} + +async function main() { + const pool = createDbPool(); + try { + console.log('[experience-e2e] login...'); + const userId = await resolveUserId(pool); + + const gateway = await bootstrapGateway(pool); + + console.log('[experience-e2e] create + dispatch run (Experience Extractor gateway)...'); + const run = await gateway.createRun(userId, { + requestId: crypto.randomUUID(), + userMessage: { + id: crypto.randomUUID(), + role: 'user', + content: [{ type: 'text', text: VERIFY_MESSAGE }], + metadata: { + userVisible: true, + displayText: VERIFY_MESSAGE, + }, + }, + }); + const runId = run.id; + console.log('[experience-e2e] run created:', runId, 'status:', run.status); + + gateway.dispatchRun(runId); + await sleep(500); + while ((await gateway.getQueueStatus()).inFlight > 0) { + await sleep(500); + } + + const terminal = await waitForTerminal(pool, runId, MAX_WAIT_MS); + console.log('[experience-e2e] run terminal:', terminal.status, terminal.error_message ?? ''); + + const experience = await findExperienceForRun(pool, runId); + if (!experience) { + console.error('[experience-e2e] FAIL: no h5_experience row for run', runId); + process.exit(1); + } + + console.log('[experience-e2e] PASS'); + console.log(JSON.stringify({ + runId, + runStatus: terminal.status, + experience: { + id: experience.id, + kind: experience.kind, + problem: experience.problem, + result: experience.result, + status: experience.status, + run_id: experience.run_id, + }, + }, null, 2)); + } finally { + await pool.end(); + } +} + +main().catch((error) => { + console.error('[experience-e2e] failed:', error instanceof Error ? error.message : error); + process.exit(1); +}); diff --git a/server.mjs b/server.mjs index 357ad9a..0a863f8 100644 --- a/server.mjs +++ b/server.mjs @@ -450,6 +450,8 @@ async function bootstrapUserAuth() { mindSpaceAgentRunner = agentServices.mindSpaceAgentRunner; mindSpaceAudit = agentServices.mindSpaceAudit; + const experienceService = + agentServices.experienceService; llmProviderService = agentServices.llmProviderService; assetGatewayConfigService = @@ -525,6 +527,7 @@ async function bootstrapUserAuth() { chatIntentRouter, syncUserGeneratedPages, isSessionPageDeliveryActive, + experienceService, }); tkmindProxy = gatewayServices.tkmindProxy; toolGateway = gatewayServices.toolGateway; diff --git a/server/portal-gateway-services-bootstrap.mjs b/server/portal-gateway-services-bootstrap.mjs index a908e72..b17373a 100644 --- a/server/portal-gateway-services-bootstrap.mjs +++ b/server/portal-gateway-services-bootstrap.mjs @@ -142,6 +142,7 @@ export function bootstrapPortalGatewayServices({ chatIntentRouter, syncUserGeneratedPages, isSessionPageDeliveryActive, + experienceService = null, createTkmindProxyFn = createTkmindProxy, createToolGatewayFn = createToolGateway, createAgentRunGatewayFn = createAgentRunGateway, @@ -254,6 +255,7 @@ export function bootstrapPortalGatewayServices({ sessionSnapshotService, conversationMemoryService, goalRunService, + experienceService, observeWorkflowRun: workflowShadowObserver, observeWorkflowValidation: workflowShadowObserver?.observeValidation ?? null, diff --git a/server/portal-gateway-services-bootstrap.test.mjs b/server/portal-gateway-services-bootstrap.test.mjs index 4818aba..9c019df 100644 --- a/server/portal-gateway-services-bootstrap.test.mjs +++ b/server/portal-gateway-services-bootstrap.test.mjs @@ -263,6 +263,15 @@ test('preserves local asset reads and optional asset absence', async () => { ); }); +test('passes experienceService into agent run gateway', () => { + const setup = createSetup({ + experienceService: { id: 'experience' }, + }); + bootstrapPortalGatewayServices(setup.options); + const { agentOptions } = setup.getCaptured(); + assert.equal(agentOptions.experienceService.id, 'experience'); +}); + test('preserves memory, page sync, and busy callbacks', async () => { const setup = createSetup(); bootstrapPortalGatewayServices(setup.options);