diff --git a/.env.example b/.env.example index cadea20..00e569e 100644 --- a/.env.example +++ b/.env.example @@ -20,6 +20,13 @@ H5_PUBLIC_BASE_URL=http://127.0.0.1:5173 # MEMIND_ANALYTICS_ID_SECRET= # MEMIND_ANALYTICS_DOMAINS=127.0.0.1,localhost +# Local Rybbit analytics (same-origin /rybbit proxy to rybbit.tkmind.cn). +# Create a Rybbit Site for localhost / 127.0.0.1, then set its numeric site_id. +# MEMIND_RYBBIT_ENABLED=true +# MEMIND_RYBBIT_URL=https://rybbit.tkmind.cn +# MEMIND_RYBBIT_SITE_ID= +# MEMIND_RYBBIT_ID_SECRET= + # 生产 H5 public base 当前临时切到 https://mm.tkmind.cn。 # 后续公网 H5 不再依赖 105 转发链路;m.tkmind.cn 仅作为 legacy/rollback 记录处理。 # H5_PUBLIC_BASE_URL=https://mm.tkmind.cn @@ -49,6 +56,9 @@ H5_PUBLIC_BASE_URL=http://127.0.0.1:5173 # MEMIND_AGENT_RUN_WORKER_BATCH_SIZE=1 # MEMIND_AGENT_RUN_HEARTBEAT_MS=30000 # MEMIND_AGENT_RUN_HEARTBEAT_STALE_MS=90000 +# MEMIND_AGENT_RUN_RECOVERY_INTERVAL_MS=30000 +# MEMIND_RUNTIME_BUILD_ID= +# MEMIND_AGENT_RUN_WORKER_ID= # MEMIND_AGENT_RUN_WORKER_EXPECT_RUNNING=0 # Runtime SLO 日报(默认每日 23:55 写 reports/runtime-slo)。 diff --git a/admin-bootstrap.mjs b/admin-bootstrap.mjs index 4119660..5f6803e 100644 --- a/admin-bootstrap.mjs +++ b/admin-bootstrap.mjs @@ -19,6 +19,8 @@ import { createLlmProviderService } from './llm-providers.mjs'; import { createAssetGatewayConfigService } from './asset-gateway.mjs'; import { createImageMakeAdminConfigService } from './image-make-admin-config.mjs'; import { createMemoryV2AdminConfigService } from './memory-v2-admin-config.mjs'; +import { createOrchestratorAdminConfigService } from './services/orchestrator/admin-config.mjs'; +import { createOrchestratorObservabilityService } from './services/orchestrator/observability.mjs'; import { createSkillRuntimeAdminConfigService } from './skill-runtime-admin-config.mjs'; import { createSystemDisclosurePolicyService } from './system-disclosure-policy.mjs'; import { createAgentCodeRunAdminConfigService } from './agent-code-run-admin-config.mjs'; @@ -110,6 +112,12 @@ export async function createAdminServices(env = {}) { }); await imageMakeAdminConfigService.ensureSchema(); const memoryV2ConfigService = createMemoryV2AdminConfigService(pool); + const orchestratorConfigService = createOrchestratorAdminConfigService(pool); + await orchestratorConfigService.ensureSchema(); + const orchestratorObservabilityService = createOrchestratorObservabilityService({ + pool, + configService: orchestratorConfigService, + }); const mindSearchConfigService = createMindSearchConfigService(pool); await mindSearchConfigService.ensureSchema(); const skillRuntimeConfigService = createSkillRuntimeAdminConfigService(pool, { h5Root }); @@ -157,6 +165,8 @@ export async function createAdminServices(env = {}) { assetGatewayConfigService, imageMakeAdminConfigService, memoryV2ConfigService, + orchestratorConfigService, + orchestratorObservabilityService, mindSearchConfigService, skillRuntimeConfigService, systemDisclosurePolicyService, diff --git a/admin-routes.mjs b/admin-routes.mjs index f8f3489..42ceead 100644 --- a/admin-routes.mjs +++ b/admin-routes.mjs @@ -31,6 +31,8 @@ function plazaRouteError(res, req, error) { * @param {object} deps.userAuth * @param {object|null} deps.llmProviderService * @param {object|null} deps.memoryV2ConfigService + * @param {object|null} deps.orchestratorConfigService + * @param {object|null} deps.orchestratorObservabilityService * @param {object|null} deps.skillRuntimeConfigService * @param {object|null} deps.systemDisclosurePolicyService * @param {object|null} deps.agentCodeRunPolicyService @@ -48,6 +50,8 @@ export function createAdminApi({ assetGatewayConfigService, imageMakeAdminConfigService, memoryV2ConfigService, + orchestratorConfigService, + orchestratorObservabilityService, mindSearchConfigService, skillRuntimeConfigService, systemDisclosurePolicyService, @@ -227,6 +231,72 @@ export function createAdminApi({ return res.json(result); }); + adminApi.get('/orchestrator/config', requireAdmin, async (_req, res) => { + if (!orchestratorConfigService?.getAdminConfig) { + return res.status(503).json({ message: 'Orchestrator 配置服务未启用' }); + } + return res.json(await orchestratorConfigService.getAdminConfig()); + }); + + const updateOrchestratorConfig = async (req, res) => { + if (!orchestratorConfigService?.updateAdminConfig) { + return res.status(503).json({ message: 'Orchestrator 配置服务未启用' }); + } + const result = await orchestratorConfigService.updateAdminConfig( + req.body?.config ?? req.body ?? {}, + { updatedBy: req.currentUser.id }, + ); + return res.json(result); + }; + + adminApi.put('/orchestrator/config', requireAdmin, updateOrchestratorConfig); + adminApi.patch('/orchestrator/config', requireAdmin, updateOrchestratorConfig); + + adminApi.get('/orchestrator/runtime', requireAdmin, async (_req, res) => { + if (!orchestratorConfigService?.getRuntimeState) { + return res.status(503).json({ message: 'Orchestrator 配置服务未启用' }); + } + return res.json(await orchestratorConfigService.getRuntimeState({ probe: true })); + }); + + adminApi.get('/orchestrator/shadow-runs', requireAdmin, async (req, res) => { + if (!orchestratorObservabilityService?.listShadowRuns) { + return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); + } + return res.json(await orchestratorObservabilityService.listShadowRuns({ + hours: req.query.hours, + limit: req.query.limit, + status: req.query.status, + })); + }); + + adminApi.get('/orchestrator/execution-plans', requireAdmin, async (req, res) => { + if (!orchestratorObservabilityService?.listExecutionPlans) { + return res.status(503).json({ message: 'Orchestrator Dry-run 观测服务未启用' }); + } + return res.json(await orchestratorObservabilityService.listExecutionPlans({ + hours: req.query.hours, + limit: req.query.limit, + selection: req.query.selection, + })); + }); + + adminApi.get('/orchestrator/canary-readiness', requireAdmin, async (_req, res) => { + if (!orchestratorObservabilityService?.getCanaryReadiness) { + return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); + } + return res.json(await orchestratorObservabilityService.getCanaryReadiness()); + }); + + adminApi.get('/orchestrator/shadow-runs/:runId', requireAdmin, async (req, res) => { + if (!orchestratorObservabilityService?.getShadowRun) { + return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); + } + const result = await orchestratorObservabilityService.getShadowRun(req.params.runId); + if (!result) return res.status(404).json({ message: 'Shadow run 不存在' }); + return res.json(result); + }); + adminApi.get('/skill-runtime/config', requireAdmin, async (_req, res) => { if (!skillRuntimeConfigService?.getAdminConfig) { return res.status(503).json({ message: 'Skill Runtime 配置服务未启用' }); diff --git a/admin-routes.test.mjs b/admin-routes.test.mjs index 18c0b90..bd84f0f 100644 --- a/admin-routes.test.mjs +++ b/admin-routes.test.mjs @@ -90,6 +90,162 @@ test('admin memory-v2 config routes expose config and runtime state', async () = } }); +test('admin orchestrator routes expose a versioned plug-in control plane', async () => { + const updates = []; + const state = { + config: { + mode: 'off', + primaryEngine: 'langgraph', + fallbackEngine: 'native', + serviceUrl: '', + requestTimeoutMs: 5000, + rolloutPercent: 0, + userAllowlist: [], + workflowAllowlist: ['code-run-v1'], + fallbackToNative: true, + requireHealthy: true, + }, + configVersion: 1, + runtime: { effective: false, reason: 'mode_off' }, + engines: [{ id: 'native', configured: true }, { id: 'langgraph', configured: false }], + }; + const router = createAdminApi({ + jsonBody: express.json(), + getToken() { + return 'token-admin'; + }, + userAuth: { + async getMe() { + return { id: 'admin-1', role: 'admin' }; + }, + }, + llmProviderService: null, + memoryV2ConfigService: null, + orchestratorConfigService: { + async getAdminConfig() { + return state; + }, + async updateAdminConfig(config, context) { + updates.push({ config, context }); + return { ...state, config: { ...state.config, ...config }, configVersion: 2 }; + }, + async getRuntimeState(options) { + assert.deepEqual(options, { probe: true }); + return { + ...state, + source: 'admin-db', + serviceHealth: { ok: true, status: 'healthy' }, + }; + }, + }, + orchestratorObservabilityService: { + async listExecutionPlans(options) { + assert.deepEqual(options, { + hours: '24', + limit: '50', + selection: 'candidate', + }); + return { + metrics: { decisions: 3, candidateSelections: 2 }, + plans: [{ runId: 'run-plan-1', candidateEngine: 'langgraph' }], + }; + }, + async getCanaryReadiness() { + return { + ready: false, + recommendation: 'keep_shadow', + blockers: ['sample_volume'], + samples: { eligibleObservations: 1 }, + }; + }, + async listShadowRuns(options) { + assert.deepEqual(options, { hours: '12', limit: '25', status: 'failed' }); + return { + metrics: { observations: 1, successes: 0, failures: 1 }, + runs: [{ runId: 'run-shadow-1', shadowStatus: 'failed' }], + }; + }, + async getShadowRun(runId) { + assert.equal(runId, 'run-shadow-1'); + return { + native: { runId, status: 'succeeded' }, + remote: { available: true, events: [{ type: 'workflow_validated' }] }, + }; + }, + }, + plazaPosts: null, + plazaOps: null, + wechatAdmin: null, + subscriptionService: null, + }); + + const server = await startTestServer(router); + try { + const configResponse = await fetch(`${server.baseUrl}/admin-api/orchestrator/config`, { + headers: { cookie: 'h5_user_session=token-admin' }, + }); + assert.equal(configResponse.status, 200); + assert.equal((await configResponse.json()).config.mode, 'off'); + + const updateResponse = await fetch(`${server.baseUrl}/admin-api/orchestrator/config`, { + method: 'PATCH', + headers: { + 'content-type': 'application/json', + cookie: 'h5_user_session=token-admin', + }, + body: JSON.stringify({ config: { mode: 'shadow', serviceUrl: 'http://127.0.0.1:8093' } }), + }); + assert.equal(updateResponse.status, 200); + assert.equal((await updateResponse.json()).configVersion, 2); + assert.deepEqual(updates, [{ + config: { mode: 'shadow', serviceUrl: 'http://127.0.0.1:8093' }, + context: { updatedBy: 'admin-1' }, + }]); + + const runtimeResponse = await fetch(`${server.baseUrl}/admin-api/orchestrator/runtime`, { + headers: { cookie: 'h5_user_session=token-admin' }, + }); + assert.equal(runtimeResponse.status, 200); + const runtimeBody = await runtimeResponse.json(); + assert.equal(runtimeBody.source, 'admin-db'); + assert.equal(runtimeBody.serviceHealth.status, 'healthy'); + + const shadowRunsResponse = await fetch( + `${server.baseUrl}/admin-api/orchestrator/shadow-runs?hours=12&limit=25&status=failed`, + { headers: { cookie: 'h5_user_session=token-admin' } }, + ); + assert.equal(shadowRunsResponse.status, 200); + assert.equal((await shadowRunsResponse.json()).metrics.failures, 1); + + const executionPlansResponse = await fetch( + `${server.baseUrl}/admin-api/orchestrator/execution-plans?hours=24&limit=50&selection=candidate`, + { headers: { cookie: 'h5_user_session=token-admin' } }, + ); + assert.equal(executionPlansResponse.status, 200); + const executionPlansBody = await executionPlansResponse.json(); + assert.equal(executionPlansBody.metrics.decisions, 3); + assert.equal(executionPlansBody.plans[0].candidateEngine, 'langgraph'); + + const readinessResponse = await fetch( + `${server.baseUrl}/admin-api/orchestrator/canary-readiness`, + { headers: { cookie: 'h5_user_session=token-admin' } }, + ); + assert.equal(readinessResponse.status, 200); + const readinessBody = await readinessResponse.json(); + assert.equal(readinessBody.ready, false); + assert.deepEqual(readinessBody.blockers, ['sample_volume']); + + const shadowRunResponse = await fetch( + `${server.baseUrl}/admin-api/orchestrator/shadow-runs/run-shadow-1`, + { headers: { cookie: 'h5_user_session=token-admin' } }, + ); + assert.equal(shadowRunResponse.status, 200); + assert.equal((await shadowRunResponse.json()).remote.available, true); + } finally { + await server.close(); + } +}); + test('admin system disclosure policy routes version config through the injected control-plane service', async () => { const updates = []; const config = { diff --git a/admin-server.mjs b/admin-server.mjs index 3f5960b..e3e4320 100644 --- a/admin-server.mjs +++ b/admin-server.mjs @@ -79,6 +79,8 @@ const CONSOLES = { assetGatewayConfigService: services.assetGatewayConfigService, imageMakeAdminConfigService: services.imageMakeAdminConfigService, memoryV2ConfigService: services.memoryV2ConfigService, + orchestratorConfigService: services.orchestratorConfigService, + orchestratorObservabilityService: services.orchestratorObservabilityService, mindSearchConfigService: services.mindSearchConfigService, skillRuntimeConfigService: services.skillRuntimeConfigService, systemDisclosurePolicyService: services.systemDisclosurePolicyService, diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index c68fd93..f696eb7 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -187,6 +187,24 @@ function positiveInteger(value, fallback) { return Math.floor(n); } +export function normalizeAgentRunWorkerIdentity(identity = {}) { + const runtimeRoot = path.resolve( + String(identity.runtimeRoot ?? process.cwd()).trim() || process.cwd(), + ); + const buildId = + String( + identity.buildId + ?? process.env.MEMIND_RUNTIME_BUILD_ID + ?? process.env.MEMIND_RELEASE_ID + ?? process.env.GIT_COMMIT + ?? 'dev', + ).trim() || 'dev'; + const workerId = + String(identity.workerId ?? '').trim() + || `${process.pid}:${crypto.randomUUID()}`; + return Object.freeze({ workerId, runtimeRoot, buildId }); +} + function timeoutError(ms) { const err = new Error(`agent run timed out after ${ms}ms`); err.code = 'AGENT_RUN_TIMEOUT'; @@ -377,8 +395,10 @@ export function createAgentRunGateway({ conversationMemoryService = null, syncUserPagesOnSuccess = null, observePersonalMemoryOnSuccess = null, + observeWorkflowRun = null, isSessionExternallyBusy = null, validateRunDeliverables = null, + quiesceSessionOnTerminal = null, retryDelaysMs = DEFAULT_RUN_RETRY_DELAYS_MS, autoDispatch = envFlag(process.env.MEMIND_AGENT_RUN_AUTODISPATCH, true), maxConcurrentRuns = positiveInteger( @@ -393,7 +413,9 @@ export function createAgentRunGateway({ process.env.MEMIND_AGENT_RUN_HEARTBEAT_MS, DEFAULT_RUN_HEARTBEAT_MS, ), + workerIdentity = null, }) { + const worker = normalizeAgentRunWorkerIdentity(workerIdentity ?? {}); const sessionStore = resolveSessionAccess({ userAuth, sessionAccess }); const inFlight = new Set(); const queuedDispatches = []; @@ -426,6 +448,105 @@ export function createAgentRunGateway({ ); } + async function quiesceTerminalSession(runId, { + userId, + sessionId, + requestId = null, + status, + }) { + if (!sessionId || typeof quiesceSessionOnTerminal !== 'function') return; + try { + const result = await quiesceSessionOnTerminal({ + runId, + userId, + sessionId, + requestId, + status, + }); + await appendEvent(runId, 'session_extensions_quiesced', { + sessionId, + status, + removed: result?.removed ?? [], + skipped: Boolean(result?.skipped), + cancellation: result?.cancellation ?? null, + }); + } catch (err) { + await appendEvent(runId, 'session_extension_quiesce_failed', { + sessionId, + status, + error: err instanceof Error ? err.message : String(err), + }).catch(() => {}); + console.warn( + `[AgentRun] failed to quiesce session ${sessionId}:`, + err instanceof Error ? err.message : err, + ); + } + } + + async function appendExecutionPlanEvent(input, source) { + const executionPlan = source?.executionPlan; + if (!executionPlan?.dryRun) return; + await appendEvent(input.runId, 'workflow_execution_planned', { + version: executionPlan.version ?? 'workflow-execution-decision-v1', + mode: source.mode ?? null, + candidateEngine: executionPlan.candidateEngine ?? null, + effectiveEngine: executionPlan.effectiveEngine ?? 'native', + fallbackEngine: executionPlan.fallbackEngine ?? 'native', + reason: executionPlan.reason ?? 'execution_gate_disabled', + candidateReason: executionPlan.candidateReason ?? null, + configVersion: executionPlan.configVersion ?? null, + bucket: executionPlan.bucket ?? null, + taskType: executionPlan.taskType ?? input.taskType ?? null, + dryRun: true, + handoffAllowed: false, + }).catch((error) => { + console.warn( + '[AgentRun] workflow execution plan event skipped:', + error instanceof Error ? error.message : error, + ); + }); + } + + function dispatchShadowObservation(input) { + if (typeof observeWorkflowRun !== 'function' || input.toolMode !== 'code') return; + setImmediate(() => { + const observationStartedAt = nowMs(); + void Promise.resolve() + .then(() => observeWorkflowRun(input)) + .then(async (result) => { + await appendExecutionPlanEvent(input, result); + if (!result?.observed) return; + await appendEvent(input.runId, 'workflow_shadow_completed', { + engine: result.engine ?? 'langgraph', + mode: result.mode ?? 'shadow', + configVersion: result.configVersion ?? null, + status: result.shadowRun?.status ?? null, + phase: result.shadowRun?.phase ?? null, + taskType: result.shadowRun?.plan?.taskType ?? input.taskType ?? null, + executorAdapter: result.shadowRun?.plan?.executorAdapter ?? null, + latencyMs: Math.max(0, nowMs() - observationStartedAt), + }); + }) + .catch(async (error) => { + await appendExecutionPlanEvent(input, error); + console.warn( + '[AgentRun] workflow shadow observation failed:', + error instanceof Error ? error.message : error, + ); + await appendEvent(input.runId, 'workflow_shadow_failed', { + code: String(error?.code ?? 'WORKFLOW_SHADOW_FAILED').slice(0, 128), + message: String(error instanceof Error ? error.message : error).slice(0, 1000), + latencyMs: Math.max(0, nowMs() - observationStartedAt), + }).catch((appendError) => { + console.warn( + '[AgentRun] workflow shadow failure event skipped:', + appendError instanceof Error ? appendError.message : appendError, + ); + }); + }); + }); + } + async function appendRunSnapshot(runId) { if (!isRunStreamReplayEnabled()) return; const row = await getRunById(runId); @@ -528,8 +649,11 @@ export function createAgentRunGateway({ await pool.query( `INSERT INTO h5_agent_runs (id, user_id, agent_session_id, request_id, status, attempts, - user_message_json, error_message, created_at, updated_at, started_at, completed_at) - VALUES (?, ?, ?, ?, 'queued', 0, ?, NULL, ?, ?, NULL, NULL)`, + user_message_json, error_message, created_at, updated_at, started_at, completed_at, + required_runtime_root, required_build_id, + claimed_worker_id, claimed_runtime_root, claimed_build_id) + VALUES (?, ?, ?, ?, 'queued', 0, ?, NULL, ?, ?, NULL, NULL, + ?, ?, NULL, NULL, NULL)`, [ runId, userId, @@ -538,6 +662,8 @@ export function createAgentRunGateway({ serializeMessage(runMessage), createdAt, createdAt, + worker.runtimeRoot, + worker.buildId, ], ); await appendEvent(runId, 'queued', { @@ -545,9 +671,22 @@ export function createAgentRunGateway({ toolMode: normalizedToolMode, taskType: normalizedTaskType, forceDeepReasoning: Boolean(forceDeepReasoning), + requiredRuntimeRoot: worker.runtimeRoot, + requiredBuildId: worker.buildId, + }); + const createdRun = projectRun(await getRunById(runId)); + dispatchShadowObservation({ + runId, + requestId: normalizedRequestId, + userId, + sessionId: sessionId || null, + workflowName: 'code-run-v1', + userMessage: runMessage, + toolMode: normalizedToolMode, + taskType: normalizedTaskType, }); if (autoDispatch) dispatchRun(runId); - return projectRun(await getRunById(runId)); + return createdRun; } async function prepareRunDeliverables({ userId, sessionId, runId, runStartedAtMs = null }) { @@ -623,6 +762,9 @@ export function createAgentRunGateway({ attempt, pid: process.pid, heartbeatMs, + workerId: worker.workerId, + runtimeRoot: worker.runtimeRoot, + buildId: worker.buildId, }); } catch (err) { console.error('[AgentRun] worker heartbeat failed:', err instanceof Error ? err.message : err); @@ -1188,12 +1330,32 @@ export function createAgentRunGateway({ const nextAttempt = Number(row.attempts ?? 0) + 1; const [claim] = await pool.query( `UPDATE h5_agent_runs - SET status = 'running', attempts = ?, started_at = COALESCE(started_at, ?), updated_at = ?, error_message = NULL - WHERE id = ? AND status IN ('queued', 'retryable')`, - [nextAttempt, nowMs(), nowMs(), runId], + SET status = 'running', attempts = ?, started_at = COALESCE(started_at, ?), + updated_at = ?, error_message = NULL, + claimed_worker_id = ?, claimed_runtime_root = ?, claimed_build_id = ? + WHERE id = ? + AND status IN ('queued', 'retryable') + AND (required_runtime_root IS NULL OR required_runtime_root = ?) + AND (required_build_id IS NULL OR required_build_id = ?)`, + [ + nextAttempt, + nowMs(), + nowMs(), + worker.workerId, + worker.runtimeRoot, + worker.buildId, + runId, + worker.runtimeRoot, + worker.buildId, + ], ); if (Number(claim?.affectedRows ?? 0) === 0) return; - await appendEvent(runId, 'running', { attempt: nextAttempt }); + await appendEvent(runId, 'running', { + attempt: nextAttempt, + workerId: worker.workerId, + runtimeRoot: worker.runtimeRoot, + buildId: worker.buildId, + }); const stopHeartbeat = startRunHeartbeat(runId, { attempt: nextAttempt }); try { @@ -1221,12 +1383,28 @@ export function createAgentRunGateway({ await markRun(runId, retryable ? 'retryable' : 'failed', { error_message: message, completed_at: retryable ? null : nowMs(), + ...(retryable + ? { + claimed_worker_id: null, + claimed_runtime_root: null, + claimed_build_id: null, + } + : {}), }, { expectedStatus: 'running' }); if (retryable && autoDispatch) { setTimeout(() => dispatchRun(runId), retryDelaysMs[nextAttempt - 1]); } } finally { stopHeartbeat(); + const terminalRun = await getRunById(runId).catch(() => null); + if (terminalRun && TERMINAL_STATUSES.has(terminalRun.status)) { + await quiesceTerminalSession(runId, { + userId: terminalRun.user_id ?? row.user_id, + sessionId: terminalRun.agent_session_id ?? row.agent_session_id ?? null, + requestId: terminalRun.request_id ?? row.request_id ?? null, + status: terminalRun.status, + }); + } } } @@ -1439,6 +1617,12 @@ export function createAgentRunGateway({ if (!marked) continue; item.status = 'succeeded'; item.recoveredAs = 'deliverables'; + await quiesceTerminalSession(row.id, { + userId: row.user_id, + sessionId: row.agent_session_id ?? null, + requestId: row.request_id ?? null, + status: 'succeeded', + }); recovered.push(item); continue; } @@ -1486,6 +1670,12 @@ export function createAgentRunGateway({ error: message, }); item.status = 'failed'; + await quiesceTerminalSession(row.id, { + userId: row.user_id, + sessionId: row.agent_session_id ?? null, + requestId: row.request_id ?? null, + status: 'failed', + }); } recovered.push(item); } @@ -1522,9 +1712,11 @@ export function createAgentRunGateway({ `SELECT id FROM h5_agent_runs WHERE status IN ('queued', 'retryable') + AND (required_runtime_root IS NULL OR required_runtime_root = ?) + AND (required_build_id IS NULL OR required_build_id = ?) ORDER BY updated_at ASC LIMIT ?`, - [take], + [worker.runtimeRoot, worker.buildId, take], ); for (const row of rows) { dispatchRun(row.id); diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 78fb062..8525bff 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -7,6 +7,7 @@ import test from 'node:test'; import { assertRequiredImageGenerationCompleted, createAgentRunGateway, + normalizeAgentRunWorkerIdentity, } from './agent-run-gateway.mjs'; test('required image generation cannot succeed without a verified raster image_make result', () => { @@ -32,6 +33,21 @@ test('required image generation cannot succeed without a verified raster image_m })); }); +test('normalizeAgentRunWorkerIdentity creates a stable normalized runtime boundary', () => { + assert.deepEqual( + normalizeAgentRunWorkerIdentity({ + workerId: 'worker-a', + runtimeRoot: '/tmp/runtime/../runtime/current', + buildId: 'build-123', + }), + { + workerId: 'worker-a', + runtimeRoot: '/tmp/runtime/current', + buildId: 'build-123', + }, + ); +}); + function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} } = {}) { const runs = new Map(); const events = []; @@ -164,9 +180,16 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} } }))]; } if (sql.includes('SELECT id') && sql.includes("status IN ('queued', 'retryable')")) { - const limit = Number(params[0] ?? 1); + const [runtimeRoot, buildId, rawLimit = 1] = params; + const limit = Number(rawLimit); return [[...runs.values()] - .filter((row) => ['queued', 'retryable'].includes(row.status)) + .filter((row) => ( + ['queued', 'retryable'].includes(row.status) && + (row.required_runtime_root == null || + row.required_runtime_root === runtimeRoot) && + (row.required_build_id == null || + row.required_build_id === buildId) + )) .sort((a, b) => Number(a.updated_at) - Number(b.updated_at)) .slice(0, limit) .map((row) => ({ id: row.id }))]; @@ -180,6 +203,8 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} } userMessageJson, createdAt, updatedAt, + requiredRuntimeRoot, + requiredBuildId, ] = params; runs.set(id, { id, @@ -194,6 +219,11 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} } updated_at: updatedAt, started_at: null, completed_at: null, + required_runtime_root: requiredRuntimeRoot, + required_build_id: requiredBuildId, + claimed_worker_id: null, + claimed_runtime_root: null, + claimed_build_id: null, }); return [{ affectedRows: 1 }]; } @@ -232,9 +262,26 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} } return [filtered.slice(0, limit)]; } if (sql.includes("SET status = 'running'")) { - const [attempts, startedAt, updatedAt, id] = params; + const [ + attempts, + startedAt, + updatedAt, + workerId, + runtimeRoot, + buildId, + id, + requiredRuntimeRoot, + requiredBuildId, + ] = params; const row = runs.get(id); - if (!row || !['queued', 'retryable'].includes(row.status)) { + if ( + !row || + !['queued', 'retryable'].includes(row.status) || + (row.required_runtime_root != null && + row.required_runtime_root !== requiredRuntimeRoot) || + (row.required_build_id != null && + row.required_build_id !== requiredBuildId) + ) { return [{ affectedRows: 0 }]; } Object.assign(row, { @@ -243,6 +290,9 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} } started_at: row.started_at ?? startedAt, updated_at: updatedAt, error_message: null, + claimed_worker_id: workerId, + claimed_runtime_root: runtimeRoot, + claimed_build_id: buildId, }); return [{ affectedRows: 1 }]; } @@ -561,6 +611,53 @@ test('agent run awaits session Finish before succeeding when proxy supports it', assert.equal(finishEvents.length, 1); }); +test('terminal agent run quiesces session extensions after preserving Finish', async () => { + const pool = createFakePool(); + const quiesced = []; + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async startSessionForUser() { + return { id: 'session-quiesce-1' }; + }, + async submitSessionReplyAndAwaitFinishForUser() { + return { ok: true, finishEvent: { type: 'Finish' } }; + }, + }, + quiesceSessionOnTerminal: async (input) => { + quiesced.push(input); + return { removed: ['sandbox-fs'], skipped: false }; + }, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-1', { + requestId: 'req-quiesce-1', + userMessage: { + role: 'user', + content: [{ type: 'text', text: 'hello' }], + }, + }); + + await waitFor(() => + pool.events.some( + (event) => + event.runId === run.id && + event.eventType === 'session_extensions_quiesced', + )); + assert.equal(pool.runs.get(run.id).status, 'succeeded'); + assert.deepEqual(quiesced, [ + { + runId: run.id, + userId: 'user-1', + sessionId: 'session-quiesce-1', + requestId: 'req-quiesce-1', + status: 'succeeded', + }, + ]); +}); + test('agent run replaces poisoned Goose session and retries with visible context', async () => { const pool = createFakePool(); const submitted = []; diff --git a/capabilities.mjs b/capabilities.mjs index 7fafc3e..8b0cbe3 100644 --- a/capabilities.mjs +++ b/capabilities.mjs @@ -87,6 +87,48 @@ export function resolveSandboxMcpUserDataPgUrl({ } } +export function resolveSandboxMcpControlDbEnv( + sourceEnv = {}, + { + containerized = false, + hostGateway = 'host.docker.internal', + } = {}, +) { + const resolved = {}; + for (const key of [ + 'DATABASE_URL', + 'MYSQL_HOST', + 'MYSQL_PORT', + 'MYSQL_USER', + 'MYSQL_PASSWORD', + 'MYSQL_DATABASE', + 'PRIVATE_DATA_MAX_BYTES', + ]) { + if (sourceEnv[key]) resolved[key] = String(sourceEnv[key]); + } + if (!containerized) return resolved; + + const gateway = String(hostGateway || 'host.docker.internal').trim(); + if ( + resolved.MYSQL_HOST && + LOOPBACK_PG_HOSTS.has(resolved.MYSQL_HOST.trim()) + ) { + resolved.MYSQL_HOST = gateway; + } + if (resolved.DATABASE_URL) { + try { + const parsed = new URL(resolved.DATABASE_URL); + if (LOOPBACK_PG_HOSTS.has(parsed.hostname)) { + parsed.hostname = gateway; + resolved.DATABASE_URL = parsed.toString(); + } + } catch { + // Keep the configured value so the MCP reports the real parse error. + } + } + return resolved; +} + export function resolveMindSearchMcpServerPath(overridePath, runtimeRoot) { return resolveBundledMcpServerPath( 'tkmind-search-mcp.mjs', @@ -454,17 +496,16 @@ function sandboxMcpEnvs(sandboxMcp, mcpTools) { envs.MINDSPACE_USERDATA_AUTO_PROVISION = String(autoProvision).trim(); } } - for (const key of [ - 'DATABASE_URL', - 'MYSQL_HOST', - 'MYSQL_PORT', - 'MYSQL_USER', - 'MYSQL_PASSWORD', - 'MYSQL_DATABASE', - 'PRIVATE_DATA_MAX_BYTES', - ]) { - if (process.env[key]) envs[key] = process.env[key]; - } + Object.assign( + envs, + resolveSandboxMcpControlDbEnv(process.env, { + containerized: Boolean(sandboxMcp.containerized), + hostGateway: + sandboxMcp.controlDbHostGateway ?? + process.env.MINDSPACE_CONTROL_DB_MCP_HOST ?? + 'host.docker.internal', + }), + ); return envs; } diff --git a/capabilities.test.mjs b/capabilities.test.mjs index d167b84..94ba67d 100644 --- a/capabilities.test.mjs +++ b/capabilities.test.mjs @@ -14,11 +14,51 @@ import { resolveSandboxMcpNodeExecPath, resolveSandboxMcpServerPath, resolveSandboxMcpUserDataPgUrl, + resolveSandboxMcpControlDbEnv, sandboxDeveloperTools, sandboxMcpTools, } from './capabilities.mjs'; import { applyPoliciesToCapabilities } from './policies.mjs'; +test('resolveSandboxMcpControlDbEnv rewrites loopback MySQL endpoints for container MCP', () => { + const resolved = resolveSandboxMcpControlDbEnv( + { + DATABASE_URL: 'mysql://portal:secret@127.0.0.1:3306/memind', + MYSQL_HOST: 'localhost', + MYSQL_PORT: '3306', + MYSQL_DATABASE: 'memind', + }, + { + containerized: true, + hostGateway: 'host.docker.internal', + }, + ); + + assert.equal( + resolved.DATABASE_URL, + 'mysql://portal:secret@host.docker.internal:3306/memind', + ); + assert.equal(resolved.MYSQL_HOST, 'host.docker.internal'); + assert.equal(resolved.MYSQL_PORT, '3306'); + assert.equal(resolved.MYSQL_DATABASE, 'memind'); +}); + +test('resolveSandboxMcpControlDbEnv preserves native-process database endpoints', () => { + assert.deepEqual( + resolveSandboxMcpControlDbEnv( + { + DATABASE_URL: 'mysql://portal:secret@127.0.0.1:3306/memind', + MYSQL_HOST: '127.0.0.1', + }, + { containerized: false }, + ), + { + DATABASE_URL: 'mysql://portal:secret@127.0.0.1:3306/memind', + MYSQL_HOST: '127.0.0.1', + }, + ); +}); + test('resolveSandboxMcpServerPath resolves to an existing MCP entry file', () => { const serverPath = resolveSandboxMcpServerPath(); assert.match(serverPath, /mindspace-sandbox-mcp\.mjs$/); diff --git a/chat-intent-router.test.mjs b/chat-intent-router.test.mjs index d4fac3f..345cc2f 100644 --- a/chat-intent-router.test.mjs +++ b/chat-intent-router.test.mjs @@ -128,6 +128,20 @@ test('classifyWithRules routes page data collection to page-data-collect', () => assert.equal(result.suggestedSkill, 'page-data-collect'); }); +test('classifyWithRules routes storefront ordering with product admin to page-data-collect', () => { + const text = '我要做一个页面,里面可以下单,后台可以上架产品'; + const result = classifyWithRules({ + text, + userMessage: { + role: 'user', + content: [{ type: 'text', text }], + metadata: { displayText: text }, + }, + }); + assert.equal(result.route, CHAT_INTENT_ROUTE.AGENT); + assert.equal(result.suggestedSkill, 'page-data-collect'); +}); + test('classifyWithRules routes ordinary household ledger language to page-data-collect', () => { const text = '帮我做一个家庭记账页面,我每天可以记录日期、收入、支出、分类和备注,日期默认当天。再做一个管理页面,可以查看所有记录和收支汇总。'; const result = classifyWithRules({ diff --git a/chat-skills.mjs b/chat-skills.mjs index 3c64f67..8eacfa2 100644 --- a/chat-skills.mjs +++ b/chat-skills.mjs @@ -49,6 +49,8 @@ const PAGE_DATA_INTENT_PATTERNS = [ /(?:记账|账本|收支|流水|日记|习惯打卡|每日打卡).{0,40}(?:页面|记录|管理|汇总|统计)/u, /(?:管理页面|管理页).{0,30}(?:查看|记录|汇总|统计|明细|删除|修改)/u, /(?:页面|网页|H5|h5).{0,80}(?:每天|每日|新增|添加|填写|记录).{0,80}(?:所有记录|历史记录|管理|汇总|统计)/u, + /(?:页面|网页|商城|店铺).{0,80}(?:下单|订单|购物车).{0,80}(?:后台|管理|上架|商品|产品|库存)/u, + /(?:后台|管理).{0,80}(?:上架|下架|商品|产品|库存).{0,80}(?:下单|订单|购物车|页面|网页|商城|店铺)/u, ]; const INTERACTIVE_PAGE_DATA_SUBJECT_PATTERN = /(?:便签|便利贴|备忘录|待办|清单)/u; diff --git a/chat-skills.test.mjs b/chat-skills.test.mjs index 6a7c741..f7309e7 100644 --- a/chat-skills.test.mjs +++ b/chat-skills.test.mjs @@ -150,6 +150,17 @@ test('isPageDataIntent matches survey and backend keywords', () => { assert.equal(isPageDataIntent('帮我做一个苏州攻略页面'), false); }); +test('isPageDataIntent treats storefront ordering plus product administration as Page Data', () => { + assert.equal( + isPageDataIntent('我要做一个页面,里面可以下单,后台可以上架产品'), + true, + ); + assert.equal( + isPageDataIntent('做一个后台上架和管理库存的商城,顾客可以购物车下单'), + true, + ); +}); + test('isPageDataIntent matches implicit interactive sticky-note app requests', () => { const text = '帮我设计一个便签提醒,可以写便签提交,时间轴来显示'; assert.equal(isPageDataIntent(text), true); diff --git a/db.mjs b/db.mjs index 9e7fcfa..d1e430b 100644 --- a/db.mjs +++ b/db.mjs @@ -335,13 +335,67 @@ export async function migrateSchema(pool) { updated_at BIGINT NOT NULL, started_at BIGINT NULL, completed_at BIGINT NULL, + required_runtime_root VARCHAR(1024) NULL, + required_build_id VARCHAR(128) NULL, + claimed_worker_id VARCHAR(255) NULL, + claimed_runtime_root VARCHAR(1024) NULL, + claimed_build_id VARCHAR(128) NULL, UNIQUE KEY uq_h5_agent_run_request (user_id, request_id), KEY idx_h5_agent_run_user_status (user_id, status, updated_at), KEY idx_h5_agent_run_session (agent_session_id, updated_at), + KEY idx_h5_agent_run_runtime_status (required_build_id, status, updated_at), CONSTRAINT fk_h5_agent_run_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE, CONSTRAINT fk_h5_agent_run_session FOREIGN KEY (agent_session_id) REFERENCES h5_user_sessions(agent_session_id) ON DELETE SET NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci `); + const agentRunRuntimeColumns = [ + ['required_runtime_root', 'VARCHAR(1024) NULL AFTER completed_at'], + ['required_build_id', 'VARCHAR(128) NULL AFTER required_runtime_root'], + ['claimed_worker_id', 'VARCHAR(255) NULL AFTER required_build_id'], + ['claimed_runtime_root', 'VARCHAR(1024) NULL AFTER claimed_worker_id'], + ['claimed_build_id', 'VARCHAR(128) NULL AFTER claimed_runtime_root'], + ]; + for (const [column, definition] of agentRunRuntimeColumns) { + if (!(await columnExists(pool, 'h5_agent_runs', column))) { + await pool.query(`ALTER TABLE h5_agent_runs ADD COLUMN \`${column}\` ${definition}`); + } + } + if (!(await indexExists(pool, 'h5_agent_runs', 'idx_h5_agent_run_runtime_status'))) { + await pool.query( + `ALTER TABLE h5_agent_runs + ADD KEY idx_h5_agent_run_runtime_status (required_build_id, status, updated_at)`, + ); + } + const [claimGuardRows] = await pool.query( + `SELECT 1 + FROM information_schema.TRIGGERS + WHERE TRIGGER_SCHEMA = DATABASE() + AND TRIGGER_NAME = 'trg_h5_agent_runs_claim_identity' + LIMIT 1`, + ); + if (claimGuardRows.length === 0) { + await pool.query(` + CREATE TRIGGER trg_h5_agent_runs_claim_identity + BEFORE UPDATE ON h5_agent_runs + FOR EACH ROW + BEGIN + IF NEW.status = 'running' + AND OLD.status IN ('queued', 'retryable') + AND NEW.required_runtime_root IS NOT NULL + AND ( + NEW.claimed_worker_id IS NULL + OR NEW.claimed_runtime_root IS NULL + OR NEW.claimed_build_id IS NULL + OR NEW.claimed_runtime_root <> NEW.required_runtime_root + OR NEW.claimed_build_id <> NEW.required_build_id + ) + THEN + SIGNAL SQLSTATE '45000' + SET MESSAGE_TEXT = 'agent run worker identity mismatch'; + END IF; + END + `); + } await pool.query(` CREATE TABLE IF NOT EXISTS h5_agent_run_events ( @@ -351,9 +405,16 @@ export async function migrateSchema(pool) { data_json JSON NULL, created_at BIGINT NOT NULL, KEY idx_h5_agent_run_event_run (run_id, created_at), + KEY idx_h5_agent_run_event_type_time (event_type, created_at), CONSTRAINT fk_h5_agent_run_event_run FOREIGN KEY (run_id) REFERENCES h5_agent_runs(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci `); + if (!(await indexExists(pool, 'h5_agent_run_events', 'idx_h5_agent_run_event_type_time'))) { + await pool.query( + `ALTER TABLE h5_agent_run_events + ADD KEY idx_h5_agent_run_event_type_time (event_type, created_at)`, + ); + } // MindSpace conversation packages: additive provenance tables for grouping // images, files, pages, and publications created during a chat session. diff --git a/deploy/orchestrator/.env.example b/deploy/orchestrator/.env.example new file mode 100644 index 0000000..cfa4f8a --- /dev/null +++ b/deploy/orchestrator/.env.example @@ -0,0 +1,15 @@ +ORCHESTRATOR_POSTGRES_PASSWORD=replace-with-a-long-random-password +MEMIND_ORCHESTRATOR_SERVICE_TOKEN=replace-with-a-long-random-service-token +MEMIND_ORCHESTRATOR_WORKER_TOKEN=replace-with-a-distinct-long-random-worker-token +MEMIND_ORCHESTRATOR_PORT=8093 + +# Closed by default. A local canary requires all three values plus the +# goosed-worker Compose profile. +MEMIND_ORCHESTRATOR_EXECUTION_ENABLED=0 +MEMIND_ORCHESTRATOR_ENABLED_EXECUTORS= +MEMIND_ORCHESTRATOR_USER_ALLOWLIST= +MEMIND_ORCHESTRATOR_TENANT_ALLOWLIST= +MEMIND_EXECUTOR_GOOSED_URL=https://host.docker.internal:18006 +MEMIND_EXECUTOR_GOOSED_TOKEN= +MEMIND_EXECUTOR_GOOSED_TLS_INSECURE=1 +MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON={} diff --git a/deploy/orchestrator/compose.executor-workers.example.yaml b/deploy/orchestrator/compose.executor-workers.example.yaml new file mode 100644 index 0000000..00bf170 --- /dev/null +++ b/deploy/orchestrator/compose.executor-workers.example.yaml @@ -0,0 +1,55 @@ +services: + worker-aider: + profiles: ["aider-worker"] + image: ${MEMIND_AIDER_WORKER_IMAGE:?set a dedicated image containing Node and Aider} + command: ["node", "/app/worker.mjs"] + restart: unless-stopped + environment: + MEMIND_ORCHESTRATOR_URL: http://orchestrator:8093 + MEMIND_ORCHESTRATOR_SERVICE_TOKEN: ${MEMIND_ORCHESTRATOR_SERVICE_TOKEN:?set service token} + MEMIND_ORCHESTRATOR_WORKER_TOKEN: ${MEMIND_ORCHESTRATOR_WORKER_TOKEN:?set worker token} + MEMIND_EXECUTOR_WORKER_ID: aider-1 + MEMIND_EXECUTOR_AIDER_COMMAND: /usr/local/bin/aider + MEMIND_EXECUTOR_WORKSPACE_ROOT: /workspaces + MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON: ${MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON:?use container paths below /workspaces} + volumes: + - ${MEMIND_EXECUTOR_HOST_WORKSPACE_ROOT:?set isolated host workspace root}:/workspaces:rw + read_only: true + tmpfs: + - /tmp:size=256m + cap_drop: [ALL] + security_opt: [no-new-privileges:true] + pids_limit: 256 + mem_limit: 4g + cpus: 2 + networks: [orchestrator_edge] + + worker-openhands: + profiles: ["openhands-worker"] + image: ${MEMIND_OPENHANDS_WORKER_IMAGE:?set a dedicated image containing Node and OpenHands} + command: ["node", "/app/worker.mjs"] + restart: unless-stopped + environment: + MEMIND_ORCHESTRATOR_URL: http://orchestrator:8093 + MEMIND_ORCHESTRATOR_SERVICE_TOKEN: ${MEMIND_ORCHESTRATOR_SERVICE_TOKEN:?set service token} + MEMIND_ORCHESTRATOR_WORKER_TOKEN: ${MEMIND_ORCHESTRATOR_WORKER_TOKEN:?set worker token} + MEMIND_EXECUTOR_WORKER_ID: openhands-1 + MEMIND_EXECUTOR_OPENHANDS_COMMAND: /usr/local/bin/openhands + MEMIND_EXECUTOR_WORKSPACE_ROOT: /workspaces + MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON: ${MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON:?use container paths below /workspaces} + volumes: + - ${MEMIND_EXECUTOR_HOST_WORKSPACE_ROOT:?set isolated host workspace root}:/workspaces:rw + read_only: true + tmpfs: + - /tmp:size=512m + cap_drop: [ALL] + security_opt: [no-new-privileges:true] + pids_limit: 512 + mem_limit: 8g + cpus: 4 + networks: [orchestrator_edge] + +networks: + orchestrator_edge: + external: true + name: orchestrator_orchestrator_edge diff --git a/deploy/orchestrator/compose.yaml b/deploy/orchestrator/compose.yaml new file mode 100644 index 0000000..58880eb --- /dev/null +++ b/deploy/orchestrator/compose.yaml @@ -0,0 +1,90 @@ +services: + postgres: + image: postgres:16-alpine + restart: unless-stopped + environment: + POSTGRES_DB: memind_orchestrator + POSTGRES_USER: memind_orchestrator + POSTGRES_PASSWORD: ${ORCHESTRATOR_POSTGRES_PASSWORD:?set ORCHESTRATOR_POSTGRES_PASSWORD} + healthcheck: + test: ["CMD-SHELL", "pg_isready -U memind_orchestrator -d memind_orchestrator"] + interval: 5s + timeout: 3s + retries: 20 + volumes: + - orchestrator_postgres_data:/var/lib/postgresql/data + networks: + - orchestrator_internal + + orchestrator: + image: memind/workflow-orchestrator:local + build: + context: ../../services/orchestrator + dockerfile: Dockerfile + restart: unless-stopped + depends_on: + postgres: + condition: service_healthy + environment: + MEMIND_ORCHESTRATOR_HOST: 0.0.0.0 + MEMIND_ORCHESTRATOR_PORT: 8093 + MEMIND_ORCHESTRATOR_CHECKPOINT_MODE: postgres + MEMIND_ORCHESTRATOR_DATABASE_URL: postgresql://memind_orchestrator:${ORCHESTRATOR_POSTGRES_PASSWORD}@postgres:5432/memind_orchestrator + MEMIND_ORCHESTRATOR_DATABASE_SCHEMA: memind_orchestrator + MEMIND_ORCHESTRATOR_SERVICE_TOKEN: ${MEMIND_ORCHESTRATOR_SERVICE_TOKEN:?set MEMIND_ORCHESTRATOR_SERVICE_TOKEN} + MEMIND_ORCHESTRATOR_WORKER_TOKEN: ${MEMIND_ORCHESTRATOR_WORKER_TOKEN:-local-worker-disabled} + MEMIND_ORCHESTRATOR_EXECUTION_ENABLED: ${MEMIND_ORCHESTRATOR_EXECUTION_ENABLED:-0} + MEMIND_ORCHESTRATOR_ENABLED_EXECUTORS: ${MEMIND_ORCHESTRATOR_ENABLED_EXECUTORS:-} + MEMIND_ORCHESTRATOR_TENANT_ALLOWLIST: ${MEMIND_ORCHESTRATOR_TENANT_ALLOWLIST:-} + MEMIND_ORCHESTRATOR_USER_ALLOWLIST: ${MEMIND_ORCHESTRATOR_USER_ALLOWLIST:-} + MEMIND_ORCHESTRATOR_WORKSPACE_KIND_ALLOWLIST: ${MEMIND_ORCHESTRATOR_WORKSPACE_KIND_ALLOWLIST:-workspace-alias} + MEMIND_ORCHESTRATOR_MAX_GLOBAL_ACTIVE: ${MEMIND_ORCHESTRATOR_MAX_GLOBAL_ACTIVE:-20} + MEMIND_ORCHESTRATOR_MAX_TENANT_ACTIVE: ${MEMIND_ORCHESTRATOR_MAX_TENANT_ACTIVE:-5} + MEMIND_ORCHESTRATOR_MAX_USER_ACTIVE: ${MEMIND_ORCHESTRATOR_MAX_USER_ACTIVE:-2} + MEMIND_ORCHESTRATOR_MAX_TENANT_PER_MINUTE: ${MEMIND_ORCHESTRATOR_MAX_TENANT_PER_MINUTE:-30} + MEMIND_ORCHESTRATOR_MAX_USER_PER_MINUTE: ${MEMIND_ORCHESTRATOR_MAX_USER_PER_MINUTE:-10} + ports: + - "127.0.0.1:${MEMIND_ORCHESTRATOR_PORT:-8093}:8093" + networks: + - orchestrator_internal + - orchestrator_edge + + worker-goosed: + profiles: ["goosed-worker"] + image: memind/workflow-orchestrator:local + restart: unless-stopped + command: ["node", "worker.mjs"] + depends_on: + orchestrator: + condition: service_started + environment: + MEMIND_ORCHESTRATOR_URL: http://orchestrator:8093 + MEMIND_ORCHESTRATOR_SERVICE_TOKEN: ${MEMIND_ORCHESTRATOR_SERVICE_TOKEN:?set MEMIND_ORCHESTRATOR_SERVICE_TOKEN} + MEMIND_ORCHESTRATOR_WORKER_TOKEN: ${MEMIND_ORCHESTRATOR_WORKER_TOKEN:-local-worker-disabled} + MEMIND_EXECUTOR_WORKER_ID: ${MEMIND_EXECUTOR_GOOSED_WORKER_ID:-colima-goosed-1} + MEMIND_EXECUTOR_GOOSED_URL: ${MEMIND_EXECUTOR_GOOSED_URL:-https://host.docker.internal:18006} + MEMIND_EXECUTOR_GOOSED_TOKEN: ${MEMIND_EXECUTOR_GOOSED_TOKEN:-} + MEMIND_EXECUTOR_ALLOW_REMOTE_GOOSED: "1" + MEMIND_EXECUTOR_GOOSED_TLS_INSECURE: ${MEMIND_EXECUTOR_GOOSED_TLS_INSECURE:-1} + MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON: ${MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON:-{}} + MEMIND_EXECUTOR_POLL_MS: ${MEMIND_EXECUTOR_POLL_MS:-1000} + MEMIND_EXECUTOR_HEARTBEAT_MS: ${MEMIND_EXECUTOR_HEARTBEAT_MS:-10000} + MEMIND_EXECUTOR_LEASE_MS: ${MEMIND_EXECUTOR_LEASE_MS:-30000} + read_only: true + tmpfs: + - /tmp:size=64m + cap_drop: + - ALL + security_opt: + - no-new-privileges:true + pids_limit: 128 + networks: + - orchestrator_edge + +networks: + orchestrator_internal: + internal: true + orchestrator_edge: + +volumes: + orchestrator_postgres_data: diff --git a/docker-compose.local.yml b/docker-compose.local.yml index ffa4e71..d7d6719 100644 --- a/docker-compose.local.yml +++ b/docker-compose.local.yml @@ -23,6 +23,12 @@ services: timeout: 5s retries: 3 start_period: 20s + # Each active session owns several stdio MCP pipes. Keep enough descriptor + # headroom for local soak tests; lifecycle cleanup is still required. + ulimits: + nofile: + soft: 65536 + hard: 65536 restart: unless-stopped networks: - local-dev @@ -48,6 +54,10 @@ services: timeout: 5s retries: 3 start_period: 20s + ulimits: + nofile: + soft: 65536 + hard: 65536 restart: unless-stopped networks: - local-dev diff --git a/docs/architecture/memind-orchestrator-boundary.md b/docs/architecture/memind-orchestrator-boundary.md new file mode 100644 index 0000000..d9c6960 --- /dev/null +++ b/docs/architecture/memind-orchestrator-boundary.md @@ -0,0 +1,237 @@ +# Memind Workflow Orchestrator Boundary + +## Decision + +The Orchestrator starts in the Memind repository but is designed as an +independently deployable bounded context. Source colocation does not authorize +in-process execution inside Portal. + +The stable boundary is: + +```text +Memind Control Plane + -> versioned Orchestrator API / events +Workflow Orchestrator + -> versioned Executor Gateway +Executor adapters + -> Goosed / Aider / OpenHands +``` + +LangGraph remains an internal workflow engine implementation. Memind must not +depend on LangGraph graph, checkpoint, command, or interrupt types. + +## Ownership + +| Data | Owner | +|---|---| +| User, authorization, billing, product run | Memind | +| Workflow node state, checkpoint, interrupt | Orchestrator | +| Goosed session | Goosed plus Session Broker mapping | +| Executor job | Executor Gateway / executor adapter | +| Workspace and deliverable | Workspace / Artifact / MindSpace services | +| User-visible progress and audit | Memind run-event projection | + +The Orchestrator must not read or write Memind user, billing, capability, or +provider-key tables. A shared PostgreSQL instance is acceptable during the +single-host phase, but the Orchestrator must use its own database and database +credentials. + +## Runtime modes + +memindadm owns the versioned routing configuration: + +| Mode | Behavior | +|---|---| +| `off` | Native Agent Run only | +| `shadow` | Native executes; LangGraph observes and compares decisions | +| `canary` | Explicit users and deterministic percentage may use LangGraph | +| `active` | Workflow-allowlisted tasks use the primary engine | + +All modes support an explicit Native fallback. The environment kill switch +`MEMIND_ORCHESTRATOR_KILL_SWITCH=1` overrides admin configuration and forces +Native selection. + +The initial workflow allowlist contains only `code-run-v1`. Ordinary chat and +existing Goosed session traffic remain outside the Orchestrator path. + +Phase 2 implemented only the `off` and `shadow` execution semantics. Canary and +Active were initially planning-only so an administrative configuration mistake +could not create two task executors. + +Phase 3 makes that restriction explicit in the contract. Routing returns both +`candidateEngine` and `engine`: Canary or Active may nominate LangGraph, while +`engine` remains Native. The decision is projected as +`workflow_execution_planned` with `dryRun=true` and `handoffAllowed=false`. +The execution adapter normalizes authorization, idempotency, timeout, +cancellation, and fallback controls. Phase 5 can transfer ownership only when +the memindadm switch, Portal environment gate, Orchestrator environment gate, +adapter enablement, worker health, authorization, and admission policy all +agree. Any missing gate keeps Native effective. + +The Dry-run projection records every code-run decision while the configured +mode is Shadow, Canary, or Active, including `canary_not_selected`. In Shadow, +the same allowlist and deterministic percentage are evaluated in parallel with +the remote observation, so collecting routing evidence does not interrupt +Shadow readiness samples. This prevents a selected-only dataset from reporting +a meaningless 100% candidate rate. The Memind control plane joins each decision +to the Native run terminal state and aggregates candidate rate, selection +reasons, task types, session coverage, and Native terminal coverage without +querying LangGraph. + +## Protocol + +The framework-neutral contracts are: + +- `orchestrator-run-v1` +- `orchestrator-event-v1` +- `workflow-execution-request-v1` +- `workflow-execution-decision-v1` +- `executor-job-request-v1` +- `executor-job-state-v1` +- `executor-dispatch-decision-v1` + +The Phase 3.2 Executor Gateway owns the adapter registry and executor job state +transition boundary. Requests contain workspace/artifact references, +authorization and side-effect policy, timeout, cancellation, fallback, and an +idempotency key. The store interface requires atomic create-if-absent semantics +so retries return the original job and conflicting payloads fail closed. + +Goosed, Aider, and OpenHands began as disabled `contract-only` adapters. The +reference in-memory store remains non-durable and is not a deployment backend. +In the current Phase 5 implementation the dispatch capability exists, but +execution and every adapter remain disabled by default. Shadow jobs still record +terminal `blocked` state without launching a process or touching a workspace. + +Phase 3.3 persists that state in the Orchestrator-owned PostgreSQL database. +LangGraph checkpoint tables and `executor_jobs` share the isolated +`memind_orchestrator` schema and credentials, but remain behind separate +storage interfaces and readiness probes. This is a deployment simplification, +not a domain ownership leak: neither store can access Memind business tables, +and the Executor Job Store can move to a separate database without changing +the graph or the Memind-facing protocol. + +The `build_plan` node creates the deterministic preview job +`:executor-preview` with idempotency key +`:build_plan:v1`. The stored result is terminal `blocked` state with +`executor_dispatch_not_implemented`; it is a durable audit and recovery +boundary, not an execution claim. If the run identifier is unsafe or would +overflow the 128-character Job id boundary, the graph substitutes a stable +SHA-256-derived opaque prefix. + +Phase 3.4 adds the append-only `executor-job-event-v1` contract. Job snapshots +remain in `executor_jobs`, while cursor-addressable events live in +`executor_job_events`. A transaction commits the initial snapshot and event +together; later state transitions lock the job row and commit the updated +snapshot and next event together. This prevents a visible state transition +without its corresponding audit event. + +The read boundary is deliberately narrower than an execution control API: + +```text +GET /v1/executor-jobs/:jobId +GET /v1/executor-jobs/:jobId/events?after=&limit= +``` + +Phase 3.4 had no HTTP submit, retry, claim, or executor cancel endpoint. +Phase 3.5 adds a separately authenticated worker protocol. The control-plane +read projection still contains no task instruction, secret, absolute path, lease +token, or executor SDK object; only a successful worker claim receives the +normalized request and lease token. + +The implemented internal API is: + +```text +POST /v1/runs +GET /v1/runs/:id +POST /v1/runs/:id/resume +POST /v1/runs/:id/cancel +GET /v1/runs/:id/events?after= +GET /v1/executor-jobs/:jobId +GET /v1/executor-jobs/:jobId/events?after=&limit= +``` + +External state changes must use an idempotency key derived from run, graph +version, node, and attempt. Graph state stores resource references and decisions, +not API keys, large logs, binary artifacts, or absolute production paths. + +The graph keeps three deterministic control-plane nodes: + +```text +validate_run -> build_plan -> finalize_run +``` + +Observe-only runs require `sideEffectsAllowed=false`, project the Native +boundary, and persist a blocked Executor Job. Explicit active runs require +`sideEffectsAllowed=true` plus every execution gate; they queue a job and enter +`waiting` until the worker-owned job reaches a terminal state. LangGraph never +mounts a workspace or imports an executor SDK. + +## Phase 5 worker and loop boundary + +The task loop is split across durable owners: + +```text +LangGraph run: validate -> queue -> wait -> project terminal +Executor Job: queued -> leased -> running -> terminal/retryable +Worker: claim -> heartbeat -> adapter -> artifact references +``` + +PostgreSQL row locks and lease tokens fence duplicate claims and late worker +updates. Expired leases recover to `retryable` until `maxAttempts` is exhausted, +then become `timed_out`. Worker heartbeats are persisted independently of job +heartbeats so readiness and autoscaling signals also work while the queue is +empty. + +Goosed is a remote HTTP/SSE adapter. Aider and OpenHands are process adapters +intended to run in separate worker containers with workspace aliases, an +allowlisted writable root, no shell, bounded environment/output, resource +limits, and no Docker socket. All adapter outputs cross the boundary as bounded +events and artifact references. + +Admission is defense in depth: tenant/user/workspace allowlists, global and +per-subject concurrency, and per-minute limits are evaluated before queueing. +The public metrics endpoint exposes queue states, claimable jobs, expired +leases, worker health, and execution-gate state without task contents. + +## Failure isolation + +Portal invokes Shadow only after the product run and its `queued` event have +been committed. The observation is scheduled without awaiting it. A timeout or +failure: + +1. cannot reject or delay `createRun`; +2. cannot mutate the Native run status; +3. is projected as `workflow_shadow_failed`; +4. remains removable by deleting the observer wiring and changing only the + service URL boundary. + +Successful observations are projected as `workflow_shadow_completed`; graph +checkpoints and blocked Executor Jobs stay in the Orchestrator-owned PostgreSQL +database. + +Both terminal projection events contain bounded observation latency. The +Memind-owned observability service aggregates those events and may join them to +the product run status. Only per-run detail calls cross the service boundary to +read LangGraph checkpoint state and node events. + +Canary readiness is also a Memind control-plane projection. It excludes +synthetic smoke runs and evaluates operational health, durable checkpoint state, +durable Executor Job storage, sample volume, session coverage, success rate, +latency coverage, P95 latency, Native terminal coverage, and sample freshness. +The result is advisory: +`manual_canary_review` never mutates routing configuration or transfers +execution ownership. + +## Deployment evolution + +1. Local native Orchestrator process with an explicit MemorySaver for debugging. +2. Colima Compose with an isolated PostgreSQL checkpoint database. +3. Approved server-side shadow service; no production task claim. +4. User-allowlisted code-run canary with Native rollback. +5. Isolated Aider/OpenHands executor workers. +6. Move the same service contract to Linux or Kubernetes when horizontal scaling + or independent ownership becomes necessary. + +The Orchestrator must not share the Goosed Compose lifecycle or mount the Docker +socket. It calls MindSpace through its service API and calls Goosed through the +existing proxy boundary. diff --git a/docs/local-analytics.md b/docs/local-analytics.md index 5d659a7..e053b6e 100644 --- a/docs/local-analytics.md +++ b/docs/local-analytics.md @@ -1,37 +1,51 @@ -# Local Memind + Umami analytics +# Local Memind analytics (Umami + Rybbit) -The integration is local-only by default. Memind proxies `/analytics/*` to the -local Umami service at `http://127.0.0.1:3100`; it does not contact 103/105. +## Umami (optional) -1. Start `/Users/john/Project/memind-analytics` and verify: +Memind can still proxy `/analytics/*` to a local Umami service at +`http://127.0.0.1:3100`. - ```bash - curl --fail http://127.0.0.1:3100/api/heartbeat - ``` +```dotenv +MEMIND_ANALYTICS_ENABLED=true +MEMIND_ANALYTICS_URL=http://127.0.0.1:3100 +MEMIND_ANALYTICS_WEBSITE_ID= +MEMIND_ANALYTICS_ID_SECRET= +MEMIND_ANALYTICS_DOMAINS=127.0.0.1,localhost +``` -2. Create one Umami Website for the local generated-page host. Do not create a - Website per page or per user. +## Rybbit (recommended for behavior analytics) -3. Put the Website ID and a local-only pseudonymization secret in Memind's - `.env`: +Rybbit runs on 105 as `https://rybbit.tkmind.cn`. Local Memind does not talk to +103 for analytics. It proxies same-origin `/rybbit/*` to Rybbit `/api/*` so the +browser script stays first-party: + +```text +Generated Page + -> /rybbit/script.js -> https://rybbit.tkmind.cn/api/script.js + -> /rybbit/track -> https://rybbit.tkmind.cn/api/track +``` + +1. In Rybbit, create one Site for the local host you use (`127.0.0.1` or + `localhost`). Do not create a Site per page or per user. +2. Put the numeric Site ID and a local pseudonymization secret in Memind `.env`: ```dotenv - MEMIND_ANALYTICS_ENABLED=true - MEMIND_ANALYTICS_URL=http://127.0.0.1:3100 - MEMIND_ANALYTICS_WEBSITE_ID= - MEMIND_ANALYTICS_ID_SECRET= - MEMIND_ANALYTICS_DOMAINS=127.0.0.1,localhost + MEMIND_RYBBIT_ENABLED=true + MEMIND_RYBBIT_URL=https://rybbit.tkmind.cn + MEMIND_RYBBIT_SITE_ID=2 + MEMIND_RYBBIT_ID_SECRET= ``` -4. Restart the local Memind server. Full generated HTML pages will receive a - same-origin `/analytics/script.js` tracker. The tracker identifies the - visitor with a stable pseudonymous owner ID before sending a standard Umami - page view, so Users and Pageviews are populated. The Identify properties - include the readable Memind username and current public page URL for - operational analytics. Click, form, scroll, and engagement events retain the - pseudonymous `owner_id`, `page_id`, and `channel` dimensions. +3. Restart the local Memind server. Full generated HTML pages receive a + same-origin `/rybbit/script.js` tracker. The tracker identifies the visitor + with a stable pseudonymous owner ID, then records engagement events such as + `page_click`, scroll depth, and dwell time. Initial pageviews come from + Rybbit's site setting `trackInitialPageView`. -The integration is fail-open: missing configuration, disabled analytics, or a -down Umami service leaves page generation and page delivery unchanged. User -facing analytics must be queried through a future Memind API that filters by -the authenticated owner; do not expose the Umami dashboard directly to users. +4. Open Rybbit from local `memind_adm` → Analytics 配置 →「打开 Rybbit 分析后台」 + (SSO). That path requires matching `MEMIND_RYBBIT_SSO_SECRET` / + `RYBBIT_SSO_EMAIL` with the 105 Rybbit deployment. + +Both integrations are fail-open: missing configuration or a down analytics +service leaves page generation and page delivery unchanged. Do not expose the +Rybbit or Umami dashboards directly to end users. diff --git a/docs/orchestrator-phase5-runbook.md b/docs/orchestrator-phase5-runbook.md new file mode 100644 index 0000000..ff25f89 --- /dev/null +++ b/docs/orchestrator-phase5-runbook.md @@ -0,0 +1,131 @@ +# Workflow Orchestrator Phase 5 Runbook + +## Scope + +Phase 5 makes the LangGraph control plane operationally deployable without +making it the default executor. The default remains: + +```text +memindadm executionEnabled=false +Portal MEMIND_ORCHESTRATOR_EXECUTION_HANDOFF_ENABLED=0 +Orchestrator MEMIND_ORCHESTRATOR_EXECUTION_ENABLED=0 +enabled executor list empty +Native Agent Run owns execution +``` + +An execution handoff requires all gates to agree: memindadm mode and explicit +execution switch, Portal environment gate, deterministic user/workflow rollout, +healthy Orchestrator storage, Orchestrator execution gate, enabled adapter, +authorization, workspace alias policy, quota admission, and a healthy worker. + +## Local Colima canary + +1. Copy `deploy/orchestrator/.env.example` to the ignored local `.env`. +2. Set distinct PostgreSQL, service, and worker secrets. +3. Keep execution disabled and start the durable control plane: + +```bash +docker compose -f deploy/orchestrator/compose.yaml up -d --build +curl -fsS http://127.0.0.1:8093/ready +curl -fsS http://127.0.0.1:8093/metrics +``` + +4. Configure a workspace alias understood by the target Goosed instance. +5. Enable only Goosed for a single local user and start the worker profile: + +```text +MEMIND_ORCHESTRATOR_EXECUTION_ENABLED=1 +MEMIND_ORCHESTRATOR_ENABLED_EXECUTORS=goosed +MEMIND_ORCHESTRATOR_USER_ALLOWLIST= +MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON={"canary":"/absolute/canary/path"} +``` + +```bash +docker compose -f deploy/orchestrator/compose.yaml --profile goosed-worker up -d +``` + +The Orchestrator `/ready` endpoint intentionally fails while execution is +enabled and no non-stale worker heartbeat exists. + +## Rollout order + +1. `off`: verify storage, metrics, backup, and worker registration. +2. `shadow`: collect Native versus LangGraph planning evidence. +3. `canary`, execution switch off: verify selection denominator and readiness. +4. `canary`, execution switch on: one explicit user, one workspace alias, zero + percentage rollout. +5. Increase percentage only after success rate, timeout, expired lease, queue + age, fallback, and Native rollback signals remain within the approved SLO. +6. `active` remains workflow-allowlisted and requires a separate release + approval. + +Never enable Aider or OpenHands in the Orchestrator service until its dedicated +worker is registered and healthy. Use the example worker Compose overlay as a +template; each tool gets a separate image, resource limits, writable workspace +root, read-only container root, no Docker socket, dropped capabilities, and a +distinct worker identity. + +## Immediate rollback + +Use any one of these independent controls: + +1. Set `MEMIND_ORCHESTRATOR_KILL_SWITCH=1` on Portal. +2. Clear the memindadm execution switch or set mode to `off`. +3. Set Portal `MEMIND_ORCHESTRATOR_EXECUTION_HANDOFF_ENABLED=0`. +4. Set Orchestrator `MEMIND_ORCHESTRATOR_EXECUTION_ENABLED=0`. +5. Remove an executor from `MEMIND_ORCHESTRATOR_ENABLED_EXECUTORS`. +6. Drain a worker by stopping it gracefully; the worker records `draining=true`. + +Queued jobs stay durable. Running jobs stop receiving heartbeats, and lease +recovery moves them to `retryable` or `timed_out` according to attempt limits. +Native fallback remains a Portal decision; the Orchestrator never launches a +second Native run by itself. + +## Monitoring + +Scrape `/metrics` and alert on: + +- `memind_orchestrator_executor_expired_leases > 0`; +- claimable jobs increasing while healthy workers are zero; +- repeated `retryable`, `failed`, or `timed_out` states; +- worker heartbeat age beyond 60 seconds; +- any execution-enabled interval without durable PostgreSQL readiness. + +Executor events contain bounded metadata and artifact references. They must not +contain provider keys, absolute host paths, full stdout/stderr, or binary data. + +## Backup and disaster recovery + +Create and verify a PostgreSQL custom-format backup: + +```bash +node scripts/orchestrator-dr.mjs backup --output /safe/path/orchestrator.dump +node scripts/orchestrator-dr.mjs verify --input /safe/path/orchestrator.dump +``` + +Restore only into a confirmed empty target: + +```bash +node scripts/orchestrator-dr.mjs restore \ + --input /safe/path/orchestrator.dump \ + --database-url postgresql://... \ + --confirm-empty-target +``` + +The restore command never uses `--clean` and refuses remote targets unless +`--allow-remote-target` is explicitly supplied. After restore, keep execution +disabled, start one worker, verify checkpoints/jobs/events/worker heartbeats, +then repeat the rollout order from `off`. + +## Production release gate + +This runbook does not authorize a production action. Before any server-side +deployment, follow `ENGINEERING_WORKFLOW_RULES.md`, +`PRODUCTION_RELEASE_RULES.md`, and run: + +```bash +bash scripts/check-release-ready.sh +``` + +Production still requires a clean commit on complete `main`, passing CI, an +explicit deployment approval, a current backup, and a tested rollback. diff --git a/mindspace-public-page-context.mjs b/mindspace-public-page-context.mjs index 1fdccd9..ba31976 100644 --- a/mindspace-public-page-context.mjs +++ b/mindspace-public-page-context.mjs @@ -2,6 +2,7 @@ import fs from 'node:fs'; import path from 'node:path'; import { PUBLISH_KEY_UUID, PUBLISH_ROOT_DIR } from './user-publish.mjs'; import { normalizeWorkspaceRelativePath } from './mindspace-pages.mjs'; +import { getPageDataPolicyIndexByWorkspacePath } from './page-data-policy-index.mjs'; const LOCAL_HOST_PATTERN = /^(localhost|127\.0\.0\.1|\[::1\]|192\.168\.|10\.|100\.)/i; @@ -102,6 +103,56 @@ export function injectPublishedPageDataContext(html, { pageSource = null, public }); } +export async function resolveMindSpacePageDataContext({ + pool = null, + pageService = null, + userId, + relativePath, + logger = console, +} = {}) { + const normalizedUserId = String(userId ?? '').trim().toLowerCase(); + const normalizedRelativePath = normalizeWorkspaceRelativePath(relativePath); + if (!normalizedUserId || !normalizedRelativePath) return null; + + // Page Data policy lives in the Portal control plane. In split-service mode the + // generic MindSpace page service is remote, so it may not see a local binding + // immediately. Resolve the local binding index first and retain the page service + // as the compatibility fallback for ordinary published pages. + const indexed = await getPageDataPolicyIndexByWorkspacePath( + pool, + normalizedUserId, + normalizedRelativePath, + ).catch((error) => { + logger?.warn?.( + `[MindSpace] failed to resolve local Page Data binding for ${normalizedUserId}/${normalizedRelativePath}: ${error?.message || error}`, + ); + return null; + }); + if (indexed?.pageId) { + return { + pageId: indexed.pageId, + accessMode: indexed.accessMode ?? null, + publicationId: indexed.publicationId ?? null, + }; + } + + if (typeof pageService?.findPageByRelativePath !== 'function') return null; + const page = await pageService + .findPageByRelativePath(normalizedUserId, normalizedRelativePath) + .catch((error) => { + logger?.warn?.( + `[MindSpace] failed to resolve remote page context for ${normalizedUserId}/${normalizedRelativePath}: ${error?.message || error}`, + ); + return null; + }); + if (!page?.id) return null; + return { + pageId: page.id, + accessMode: page.publicationAccessMode ?? null, + publicationId: page.currentPublishId ?? page.current_publish_id ?? null, + }; +} + export const mindspacePublicPageContextInternals = { LOCAL_HOST_PATTERN, }; diff --git a/mindspace-public-page-context.test.mjs b/mindspace-public-page-context.test.mjs index 5ac0233..35bfdbf 100644 --- a/mindspace-public-page-context.test.mjs +++ b/mindspace-public-page-context.test.mjs @@ -6,6 +6,7 @@ import { injectMindSpacePageDataContext, injectPublishedPageDataContext, parseMindSpacePublishFilePath, + resolveMindSpacePageDataContext, } from './mindspace-public-page-context.mjs'; test('parseMindSpacePublishFilePath extracts userId and relative path', () => { @@ -47,3 +48,65 @@ test('injectPublishedPageDataContext injects pageId from publication resolve res assert.match(next, /meta name="mindspace-page-data-page-id" content="2b201736-03ca-4172-affd-00f2a0f700b5"/); assert.match(next, /window\.__MINDSPACE_PAGE_DATA__=\{"pageId":"2b201736-03ca-4172-affd-00f2a0f700b5","accessMode":"public"\}/); }); + +test('resolveMindSpacePageDataContext prefers the Portal policy index in split-service mode', async () => { + const pool = { + async query() { + return [[{ + page_id: 'local-page', + owner_user_id: 'user-1', + access_mode: 'public', + dataset_count: 1, + scope_hash: 'scope', + updated_at: 123, + current_publish_id: 'local-publish', + }]]; + }, + }; + let remoteCalls = 0; + const result = await resolveMindSpacePageDataContext({ + pool, + pageService: { + async findPageByRelativePath() { + remoteCalls += 1; + return { id: 'remote-page', publicationAccessMode: 'password' }; + }, + }, + userId: 'USER-1', + relativePath: 'public/storefront.html', + }); + assert.deepEqual(result, { + pageId: 'local-page', + accessMode: 'public', + publicationId: 'local-publish', + }); + assert.equal(remoteCalls, 0); +}); + +test('resolveMindSpacePageDataContext falls back to the MindSpace page service', async () => { + const result = await resolveMindSpacePageDataContext({ + pool: { + async query() { + return [[]]; + }, + }, + pageService: { + async findPageByRelativePath(userId, relativePath) { + assert.equal(userId, 'user-1'); + assert.equal(relativePath, 'public/plain.html'); + return { + id: 'remote-page', + publicationAccessMode: 'internal', + currentPublishId: 'remote-publish', + }; + }, + }, + userId: 'USER-1', + relativePath: 'public/plain.html', + }); + assert.deepEqual(result, { + pageId: 'remote-page', + accessMode: 'internal', + publicationId: 'remote-publish', + }); +}); diff --git a/mindspace-rybbit.mjs b/mindspace-rybbit.mjs new file mode 100644 index 0000000..6b6f2de --- /dev/null +++ b/mindspace-rybbit.mjs @@ -0,0 +1,114 @@ +import crypto from 'node:crypto'; + +import { + pseudonymizeAnalyticsId, + resolveAnalyticsOwnerLabel, + resolveAnalyticsOwnerSegment, +} from './mindspace-analytics.mjs'; + +const RYBBIT_MARKER = 'data-memind-rybbit="1"'; + +export function resolveMindSpaceRybbitConfig(env = process.env) { + const enabled = String(env.MEMIND_RYBBIT_ENABLED ?? '').toLowerCase() === 'true'; + const siteId = String(env.MEMIND_RYBBIT_SITE_ID ?? '').trim(); + const secret = String(env.MEMIND_RYBBIT_ID_SECRET ?? env.MEMIND_ANALYTICS_ID_SECRET ?? '').trim(); + const rybbitUrl = String(env.MEMIND_RYBBIT_URL ?? env.RYBBIT_URL ?? 'https://rybbit.tkmind.cn').trim() + || 'https://rybbit.tkmind.cn'; + return { + enabled: enabled && Boolean(siteId) && Boolean(secret), + siteId, + idSecret: secret, + rybbitUrl: rybbitUrl.replace(/\/$/, ''), + // Same-origin proxy path. Script auto-discovers host by splitting on /script.js, + // so /rybbit/script.js makes events go to /rybbit/track → proxied to /api/track. + scriptPath: String(env.MEMIND_RYBBIT_SCRIPT_PATH ?? '/rybbit/script.js').trim() || '/rybbit/script.js', + hostPath: String(env.MEMIND_RYBBIT_HOST_PATH ?? '/rybbit').trim() || '/rybbit', + }; +} + +export { resolveAnalyticsOwnerLabel, resolveAnalyticsOwnerSegment, pseudonymizeAnalyticsId }; + +function jsonForInlineScript(value) { + return JSON.stringify(value) + .replaceAll('<', '\\u003c') + .replaceAll('>', '\\u003e') + .replaceAll('&', '\\u0026'); +} + +export function sendMindSpaceRybbitEvent({ + config, + eventName, + ownerId, + pageId = '', + publicationId = '', + agentRunId = '', + channel = 'h5', + ownerSegment = 'unknown', + ownerLabel = '未命名用户', + url = '', +} = {}) { + if (!config?.enabled || !config.siteId || !config.idSecret || !eventName) return Promise.resolve(false); + const owner = pseudonymizeAnalyticsId(ownerId, config.idSecret); + if (!owner) return Promise.resolve(false); + const endpoint = `${String(config.rybbitUrl || 'https://rybbit.tkmind.cn').replace(/\/$/, '')}/api/track`; + const payload = { + site_id: String(config.siteId), + type: 'custom_event', + event_name: String(eventName).slice(0, 256), + pathname: url || '/', + page_title: '', + referrer: '', + hostname: '127.0.0.1', + user_id: owner, + properties: JSON.stringify({ + owner_id: owner, + page_id: String(pageId || ''), + publication_id: String(publicationId || ''), + agent_run_id: String(agentRunId || ''), + channel, + owner_segment: String(ownerSegment || 'unknown'), + owner_label: resolveAnalyticsOwnerLabel({ displayName: ownerLabel }), + }), + }; + return fetch(endpoint, { + method: 'POST', + headers: { 'content-type': 'application/json', 'user-agent': 'Memind/local-rybbit' }, + body: JSON.stringify(payload), + signal: AbortSignal.timeout(1500), + }).then((response) => response.ok).catch(() => false); +} + +export function injectMindSpaceRybbit(html, { + ownerId, + pageId = '', + publicationId = '', + ownerSegment = 'unknown', + ownerLabel = '未命名用户', + channel = 'h5', + config = resolveMindSpaceRybbitConfig(), +} = {}) { + const source = String(html ?? ''); + if (!config?.enabled || !config.siteId || !/^\s*(`; + if (/<\/head>/i.test(source)) return source.replace(/<\/head>/i, `${block}`); + return source.replace(/ { + assert.equal(resolveMindSpaceRybbitConfig({ MEMIND_RYBBIT_ENABLED: 'true' }).enabled, false); + assert.equal(resolveMindSpaceRybbitConfig({ + MEMIND_RYBBIT_ENABLED: 'true', + MEMIND_RYBBIT_SITE_ID: '2', + }).enabled, false); + const config = resolveMindSpaceRybbitConfig({ + MEMIND_RYBBIT_ENABLED: 'true', + MEMIND_RYBBIT_SITE_ID: '2', + MEMIND_RYBBIT_ID_SECRET: 'local-secret', + MEMIND_RYBBIT_URL: 'https://rybbit.tkmind.cn', + }); + assert.equal(config.enabled, true); + assert.equal(config.siteId, '2'); + assert.equal(config.rybbitUrl, 'https://rybbit.tkmind.cn'); + assert.equal(config.hostPath, '/rybbit'); + assert.equal(config.scriptPath, '/rybbit/script.js'); +}); + +test('injects one local same-origin rybbit tracker with page dimensions', () => { + const html = 'Demo

Demo

'; + const out = injectMindSpaceRybbit(html, { + ownerId: 'user-123', + ownerLabel: '张三', + pageId: 'page-1', + publicationId: 'pub-1', + config: { + enabled: true, + siteId: '2', + idSecret: 'secret', + scriptPath: '/rybbit/script.js', + hostPath: '/rybbit', + }, + }); + assert.match(out, /src="\/rybbit\/script\.js"/); + assert.match(out, /data-site-id="2"/); + assert.match(out, /data-memind-rybbit="1"/); + assert.match(out, /r\.identify\(d\.owner_id,/); + assert.match(out, /r\.event\(n,p\)/); + assert.match(out, /page_click/); + assert.match(out, /page_form_submit/); + assert.match(out, /page_scroll_/); + assert.match(out, /page_engaged_10s/); + assert.match(out, /"username":"张三"/); + assert.doesNotMatch(out, /user-123/); + assert.equal(injectMindSpaceRybbit(out, { + ownerId: 'user-123', + config: { enabled: true, siteId: '2', idSecret: 'secret' }, + }), out); +}); + +test('identifies the pseudonymous owner once rybbit is ready', () => { + const out = injectMindSpaceRybbit('', { + ownerId: 'user-123', + ownerSegment: 'plan:pro', + ownerLabel: '张三', + channel: 'h5', + config: { + enabled: true, + siteId: '2', + idSecret: 'secret', + scriptPath: '/rybbit/script.js', + hostPath: '/rybbit', + }, + }); + const inlineScript = out.match(/