From 417aa9c78ba7fefaf84a6b3ff5cefaaf19d8e49b Mon Sep 17 00:00:00 2001 From: john Date: Sat, 25 Jul 2026 14:16:57 +0800 Subject: [PATCH] feat(orchestrator): observe Page Data validation --- agent-run-gateway.mjs | 157 +++++++++++++++++- agent-run-gateway.test.mjs | 147 ++++++++++++++++ .../memind-orchestrator-boundary.md | 12 ++ ops/src/api/admin.ts | 19 +++ .../pages/admin/OrchestratorShadowPanel.tsx | 13 ++ server/portal-gateway-services-bootstrap.mjs | 2 + ...portal-gateway-services-bootstrap.test.mjs | 4 + services/orchestrator/app.mjs | 13 ++ services/orchestrator/app.test.mjs | 28 +++- .../orchestrator/code-run-shadow-graph.mjs | 4 + services/orchestrator/contracts.mjs | 86 ++++++++++ services/orchestrator/contracts.test.mjs | 62 +++++++ services/orchestrator/engine-registry.mjs | 6 + services/orchestrator/runtime.mjs | 58 ++++++- services/orchestrator/runtime.test.mjs | 55 ++++++ services/orchestrator/shadow-observer.mjs | 101 +++++++++-- .../orchestrator/shadow-observer.test.mjs | 33 ++++ 17 files changed, 784 insertions(+), 16 deletions(-) diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index cbfaccd..dd5802a 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -111,6 +111,43 @@ function messageText(message) { .join('\n'); } +function isPageDataWorkflowRequested(userMessage, taskType = null) { + const metadata = userMessage?.metadata ?? {}; + const runMetadata = metadata?.[RUN_METADATA_KEY] ?? metadata?.agentRun ?? {}; + const normalizedTaskType = String( + taskType ?? runMetadata?.taskType ?? metadata?.taskType ?? '', + ).trim().toLowerCase(); + const selectedSkill = String( + runMetadata?.selectedChatSkill ?? metadata?.selectedChatSkill ?? '', + ).trim(); + const displayText = String(metadata?.displayText ?? '').trim() + || deriveUserFacingText(messageText(userMessage)); + return runMetadata?.pgRequired === true + || runMetadata?.pageDataAiderWorkflow === true + || normalizedTaskType === 'page_data_dev' + || normalizedTaskType === 'page_data_dev_complex' + || selectedSkill === 'page-data-collect' + || isPageDataIntent(displayText); +} + +function pageDataFailureCheckId(code) { + switch (String(code ?? '').trim().toUpperCase()) { + case 'PAGE_DATA_DELIVERY_FAILED': + return 'page_data_binding'; + case 'PAGE_DATA_DELIVERABLE_MISSING': + return 'page_data_deliverable'; + case 'DELIVERABLE_DATA_STORAGE_FORBIDDEN': + return 'page_data_storage_policy'; + case 'REQUIRED_REVIEW_EXECUTOR_MISMATCH': + case 'REQUIRED_EXECUTOR_UNAVAILABLE': + return 'independent_review'; + case 'TOOL_GATEWAY_VALIDATION_FAILED': + return 'code_validation'; + default: + return 'agent_run_completion'; + } +} + function buildFreshSessionContext(conversation, { maxMessages = 16, maxChars = 12_000 } = {}) { const visible = []; for (const message of Array.isArray(conversation) ? conversation : []) { @@ -394,6 +431,7 @@ function getRunOptionsFromMessage(userMessage) { ? String(runMetadata.reviewExecutor).trim().toLowerCase() : null, pageDataAiderWorkflow: runMetadata?.pageDataAiderWorkflow === true, + pgRequired: runMetadata?.pgRequired === true, forceDeepReasoning: runMetadata?.forceDeepReasoning === true || metadata?.forceDeepReasoning === true, validation: normalizeToolGatewayValidation(runMetadata?.validation ?? metadata?.toolGatewayValidation), sessionMessageCount: normalizeSessionMessageCount( @@ -475,6 +513,7 @@ export function createAgentRunGateway({ syncUserPagesOnSuccess = null, observePersonalMemoryOnSuccess = null, observeWorkflowRun = null, + observeWorkflowValidation = null, isSessionExternallyBusy = null, validateRunDeliverables = null, quiesceSessionOnTerminal = null, @@ -637,9 +676,78 @@ export function createAgentRunGateway({ logger: console, }); + const validationDispatcher = createShadowObservationDispatcher({ + observe: observeWorkflowValidation, + maxConcurrent: maxConcurrentShadowObservations, + maxQueued: maxQueuedShadowObservations, + onCompleted: async (input, result, context) => { + if (!result?.observed) { + await appendEvent(input.runId, 'workflow_validation_observation_skipped', { + reason: result?.reason ?? 'shadow_validation_not_selected', + }); + return; + } + await appendEvent(input.runId, 'workflow_validation_observation_completed', { + engine: result.engine ?? 'langgraph', + mode: result.mode ?? 'shadow', + configVersion: result.configVersion ?? null, + kind: result.validation?.kind ?? 'page-data-delivery', + verdict: result.validation?.verdict ?? null, + latencyMs: Math.max(0, nowMs() - context.enqueuedAt), + }); + }, + onFailed: async (input, error, context) => { + await appendEvent(input.runId, 'workflow_validation_observation_failed', { + code: String(error?.code ?? 'WORKFLOW_VALIDATION_OBSERVATION_FAILED').slice(0, 128), + message: String(error instanceof Error ? error.message : error).slice(0, 1000), + latencyMs: Math.max(0, nowMs() - context.enqueuedAt), + }).catch(() => {}); + }, + onSkipped: async (input, context) => { + await appendEvent(input.runId, 'workflow_validation_observation_skipped', { + reason: context.reason, + }); + }, + logger: console, + }); + function dispatchShadowObservation(input) { - if (input.toolMode !== 'code') return; - shadowDispatcher.dispatch(input); + const pageDataRequired = isPageDataWorkflowRequested(input.userMessage, input.taskType); + if (input.toolMode !== 'code' && !pageDataRequired) return; + shadowDispatcher.dispatch({ + ...input, + taskType: input.taskType || (pageDataRequired ? 'page_data_dev' : null), + pageDataRequired, + }); + } + + function dispatchPageDataValidationObservation({ + row, + runId, + checks, + metrics = null, + }) { + const userMessage = parseDbJsonColumn(row?.user_message_json, {}) ?? {}; + const runOptions = getRunOptionsFromMessage(userMessage); + if (!isPageDataWorkflowRequested(userMessage, runOptions.taskType)) return false; + return validationDispatcher.dispatch({ + runId, + requestId: row?.request_id ?? null, + userId: row?.user_id ?? null, + workflowName: 'code-run-v1', + observation: { + idempotencyKey: `${runId}:page-data-delivery:v1`, + taskType: runOptions.taskType || 'page_data_dev', + required: true, + checks, + metrics: { + pageCount: Number(metrics?.pageCount ?? 0), + publicationCount: Number(metrics?.publicationCount ?? 0), + }, + source: 'portal-agent-run', + observedAt: nowMs(), + }, + }); } async function appendRunSnapshot(runId) { @@ -1464,6 +1572,8 @@ export function createAgentRunGateway({ }, { expectedStatus: 'running' }); return; } + const userMessage = parseDbJsonColumn(row.user_message_json, {}) ?? {}; + const runOptions = getRunOptionsFromMessage(userMessage); assertRequiredImageGenerationCompleted(row, routing, toolEvidence); // `row` was loaded before this worker claimed the run, so its started_at // can still be null. Refresh it before scoping workspace files to the @@ -1490,7 +1600,8 @@ export function createAgentRunGateway({ } const runDisplayText = extractRunDisplayText(row); const selectedSkill = selectedRunSkill(row); - const pageDataIntent = isPageDataIntent(runDisplayText) + const pageDataIntent = isPageDataWorkflowRequested(userMessage, runOptions.taskType) + || isPageDataIntent(runDisplayText) || selectedSkill === 'page-data-collect' || routing?.suggestedSkill === 'page-data-collect'; const pageGenerationIntent = isPageGenerationIntent(runDisplayText) @@ -1557,6 +1668,35 @@ export function createAgentRunGateway({ error_message: null, }, { expectedStatus: 'running' }); if (!marked) return false; + if (pageDataIntent) { + dispatchPageDataValidationObservation({ + row, + runId, + checks: [ + { id: 'agent_run_completion', status: 'passed' }, + { + id: 'page_data_binding', + status: typeof syncUserPagesOnSuccess === 'function' ? 'passed' : 'skipped', + }, + { + id: 'page_data_storage_policy', + status: typeof validateRunDeliverables === 'function' ? 'passed' : 'skipped', + }, + { + id: 'page_data_deliverable', + status: requiresPageDeliverable ? 'passed' : 'skipped', + }, + { + id: 'independent_review', + status: runOptions.reviewExecutor ? 'passed' : 'skipped', + }, + ], + metrics: { + pageCount: deliverables?.pageCount ?? 0, + publicationCount: deliverables?.publicationCount ?? 0, + }, + }); + } if (typeof observePersonalMemoryOnSuccess === 'function') { await observePersonalMemoryOnSuccess({ userId: row.user_id, @@ -1639,6 +1779,17 @@ export function createAgentRunGateway({ } : {}), }, { expectedStatus: 'running' }); + if (!retryable) { + dispatchPageDataValidationObservation({ + row, + runId, + checks: [{ + id: pageDataFailureCheckId(err?.code), + status: 'failed', + codes: [String(err?.code ?? 'AGENT_RUN_FAILED')], + }], + }); + } if (retryable && autoDispatch) { setTimeout(() => dispatchRun(runId), retryDelaysMs[nextAttempt - 1]); } diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 3569e61..0a6a34f 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -462,6 +462,153 @@ test('agent run creation stores code tool metadata in the queued message', async }); }); +test('explicit Page Data selection enters Orchestrator Shadow even when the run starts in chat mode', async () => { + const pool = createFakePool(); + const observed = []; + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: {}, + autoDispatch: false, + observeWorkflowRun: async (input) => { + observed.push(input); + return { observed: false, reason: 'test-only' }; + }, + }); + + const run = await gateway.createRun('user-page-data-shadow', { + requestId: 'req-page-data-shadow', + userMessage: { + role: 'user', + content: [{ type: 'text', text: '解释一下这个数据空间' }], + metadata: { + displayText: '解释一下这个数据空间', + memindRun: { pgRequired: true }, + }, + }, + toolMode: 'chat', + }); + + await waitFor(() => observed.length === 1); + assert.equal(observed[0].runId, run.id); + assert.equal(observed[0].pageDataRequired, true); + assert.equal(observed[0].taskType, 'page_data_dev'); + assert.equal(pool.runs.get(run.id).status, 'queued'); +}); + +test('successful Page Data runs send bounded validation evidence without changing Native execution', async () => { + const pool = createFakePool(); + const validationInputs = []; + const submitted = []; + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async startSessionForUser() { + return { id: 'session-page-data-validation' }; + }, + async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { + submitted.push({ userId, sessionId, requestId, userMessage }); + }, + }, + observeWorkflowRun: async () => ({ observed: true }), + observeWorkflowValidation: async (input) => { + validationInputs.push(input); + return { + observed: true, + engine: 'langgraph', + mode: 'shadow', + validation: { + kind: 'page-data-delivery', + verdict: 'inconclusive', + }, + }; + }, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-page-data-validation', { + requestId: 'req-page-data-validation', + userMessage: { + role: 'user', + content: [{ type: 'text', text: '解释 Page Data API 的用途' }], + metadata: { + displayText: '解释 Page Data API 的用途', + memindRun: { pgRequired: true }, + }, + }, + toolMode: 'chat', + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); + await waitFor(() => validationInputs.length === 1); + assert.equal(submitted.length, 1); + assert.equal(validationInputs[0].runId, run.id); + assert.equal(validationInputs[0].observation.required, true); + assert.deepEqual( + validationInputs[0].observation.checks.map((check) => [check.id, check.status]), + [ + ['agent_run_completion', 'passed'], + ['page_data_binding', 'skipped'], + ['page_data_storage_policy', 'skipped'], + ['page_data_deliverable', 'skipped'], + ['independent_review', 'skipped'], + ], + ); + const projected = JSON.stringify(validationInputs[0].observation); + assert.equal(projected.includes('解释 Page Data API'), false); + assert.equal(pool.runs.get(run.id).status, 'succeeded'); + await waitFor(() => pool.events.some( + (event) => event.runId === run.id + && event.eventType === 'workflow_validation_observation_completed', + )); +}); + +test('Page Data validation observation failure stays isolated from the Native result', async () => { + const pool = createFakePool(); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async startSessionForUser() { + return { id: 'session-page-data-validation-failure' }; + }, + async submitSessionReplyForUser() {}, + }, + observeWorkflowRun: async () => ({ observed: true }), + observeWorkflowValidation: async () => { + throw Object.assign(new Error('validation observer unavailable'), { + code: 'VALIDATION_OBSERVER_UNAVAILABLE', + }); + }, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-page-data-validation-failure', { + requestId: 'req-page-data-validation-failure', + userMessage: { + role: 'user', + content: [{ type: 'text', text: '解释 Page Data API' }], + metadata: { + displayText: '解释 Page Data API', + memindRun: { pgRequired: true }, + }, + }, + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); + await waitFor(() => pool.events.some( + (event) => event.runId === run.id + && event.eventType === 'workflow_validation_observation_failed', + )); + assert.equal(pool.runs.get(run.id).status, 'succeeded'); + const failure = JSON.parse(pool.events.find( + (event) => event.runId === run.id + && event.eventType === 'workflow_validation_observation_failed', + ).dataJson); + assert.equal(failure.code, 'VALIDATION_OBSERVER_UNAVAILABLE'); +}); + test('agent run starts a session and marks submitted reply as succeeded', async () => { const pool = createFakePool(); const submitted = []; diff --git a/docs/architecture/memind-orchestrator-boundary.md b/docs/architecture/memind-orchestrator-boundary.md index 99f150b..c628e45 100644 --- a/docs/architecture/memind-orchestrator-boundary.md +++ b/docs/architecture/memind-orchestrator-boundary.md @@ -101,6 +101,7 @@ The framework-neutral contracts are: - `executor-job-request-v1` - `executor-job-state-v1` - `executor-dispatch-decision-v1` +- `page-data-validation-observation-v1` The Phase 3.2 Executor Gateway owns the adapter registry and executor job state transition boundary. Requests contain workspace/artifact references, @@ -155,6 +156,7 @@ The implemented internal API is: ```text POST /v1/runs GET /v1/runs/:id +POST /v1/runs/:id/validation-observations POST /v1/runs/:id/resume POST /v1/runs/:id/cancel DELETE /v1/runs/:id @@ -177,6 +179,16 @@ through `DELETE /v1/runs/:id`; deletion removes the LangGraph thread and its linked terminal Executor Job, whose events cascade in PostgreSQL. Active or waiting runs fail deletion with `409`. +Page Data code runs may append a post-run +`page-data-validation-observation-v1` after the Native delivery guards finish. +The observation contains only bounded check identifiers, pass/fail/skipped +states, error codes, and aggregate page/publication counts. User prompts, code, +workspace paths, dataset contents, credentials, and database identifiers must +not cross this boundary. The Orchestrator derives and checkpoints a +`passed`/`failed`/`inconclusive` verdict and emits +`workflow_validation_observed`. In the initial observation phase this verdict +does not mutate the Native run result or publication state. + The graph keeps three deterministic control-plane nodes: ```text diff --git a/ops/src/api/admin.ts b/ops/src/api/admin.ts index 3fea929..857eb47 100644 --- a/ops/src/api/admin.ts +++ b/ops/src/api/admin.ts @@ -460,6 +460,25 @@ export type OrchestratorShadowRunDetail = { phase?: string; plan?: Record; result?: Record; + validation?: { + version?: string; + kind?: string; + taskType?: string | null; + required?: boolean; + verdict?: 'passed' | 'failed' | 'inconclusive'; + checks?: Array<{ + id?: string; + status?: 'passed' | 'failed' | 'skipped'; + codes?: string[]; + }>; + metrics?: { + pageCount?: number; + publicationCount?: number; + }; + source?: string; + dataPolicy?: string; + observedAt?: number; + } | null; } | null; events: Array<{ sequence?: number; diff --git a/ops/src/pages/admin/OrchestratorShadowPanel.tsx b/ops/src/pages/admin/OrchestratorShadowPanel.tsx index 4ceb8cd..39e8578 100644 --- a/ops/src/pages/admin/OrchestratorShadowPanel.tsx +++ b/ops/src/pages/admin/OrchestratorShadowPanel.tsx @@ -328,6 +328,19 @@ export function OrchestratorShadowPanel() { {selected.remote.executorJob.error.message}

) : null} + {selected.remote.state?.validation ? ( +

+ Page Data 验证： + {selected.remote.state.validation.verdict ?? 'unknown'} + {' · '} + {selected.remote.state.validation.checks?.length ?? 0} 项检查 + {' · '} + {selected.remote.state.validation.dataPolicy ?? 'unknown-policy'} +

+ ) : null}
LangGraph checkpoint diff --git a/server/portal-gateway-services-bootstrap.mjs b/server/portal-gateway-services-bootstrap.mjs index 04ab9a4..f6be74b 100644 --- a/server/portal-gateway-services-bootstrap.mjs +++ b/server/portal-gateway-services-bootstrap.mjs @@ -329,6 +329,8 @@ export function bootstrapPortalGatewayServices({ sessionSnapshotService, conversationMemoryService, observeWorkflowRun: workflowShadowObserver, + observeWorkflowValidation: + workflowShadowObserver?.observeValidation ?? null, observePersonalMemoryOnSuccess: async ({ userId, sessionId, diff --git a/server/portal-gateway-services-bootstrap.test.mjs b/server/portal-gateway-services-bootstrap.test.mjs index 25e401d..dc79f14 100644 --- a/server/portal-gateway-services-bootstrap.test.mjs +++ b/server/portal-gateway-services-bootstrap.test.mjs @@ -167,6 +167,7 @@ test('preserves Proxy, Tool, and Agent gateway wiring', () => { assert.equal(captured.agentOptions.maxConcurrentRuns, 3); assert.equal(captured.agentOptions.runTimeoutMs, 9000); assert.equal(captured.agentOptions.observeWorkflowRun, null); + assert.equal(captured.agentOptions.observeWorkflowValidation, null); assert.equal(captured.agentOptions.maxConcurrentShadowObservations, 2); assert.equal(captured.agentOptions.maxQueuedShadowObservations, 100); }); @@ -174,6 +175,8 @@ test('preserves Proxy, Tool, and Agent gateway wiring', () => { test('wires Shadow observation only behind the explicit environment gate', () => { const configService = { id: 'orchestrator-config' }; const observer = async () => ({ observed: false }); + const validationObserver = async () => ({ observed: false }); + observer.observeValidation = validationObserver; let receivedObserverOptions = null; const setup = createSetup({ env: { @@ -197,6 +200,7 @@ test('wires Shadow observation only behind the explicit environment gate', () => assert.equal(receivedObserverOptions.configService, configService); assert.equal(receivedObserverOptions.serviceToken, 'shadow-token'); assert.equal(agentOptions.observeWorkflowRun, observer); + assert.equal(agentOptions.observeWorkflowValidation, validationObserver); assert.equal(agentOptions.maxConcurrentShadowObservations, 4); assert.equal(agentOptions.maxQueuedShadowObservations, 25); }); diff --git a/services/orchestrator/app.mjs b/services/orchestrator/app.mjs index 971bf5b..a34c05d 100644 --- a/services/orchestrator/app.mjs +++ b/services/orchestrator/app.mjs @@ -134,6 +134,19 @@ export function createOrchestratorApp({ } }); + app.post('/v1/runs/:runId/validation-observations', async (request, response, next) => { + try { + const state = await runtime.recordValidationObservation( + request.params.runId, + request.body, + ); + if (!state) return response.status(404).json({ error: { code: 'RUN_NOT_FOUND' } }); + return response.json(state); + } catch (error) { + return next(error); + } + }); + app.post('/v1/runs/:runId/resume', async (request, response, next) => { try { const state = await runtime.resume(request.params.runId, request.body); diff --git a/services/orchestrator/app.test.mjs b/services/orchestrator/app.test.mjs index 8d54ff7..4282d2d 100644 --- a/services/orchestrator/app.test.mjs +++ b/services/orchestrator/app.test.mjs @@ -76,12 +76,38 @@ test('orchestrator HTTP API exposes health and authenticated run endpoints', asy assert.equal(created.status, 202); assert.equal((await created.json()).status, 'succeeded'); + const validation = await fetch( + `${server.baseUrl}/v1/runs/http-shadow-1/validation-observations`, + { + method: 'POST', + headers: { + authorization: 'Bearer test-service-token', + 'content-type': 'application/json', + }, + body: JSON.stringify({ + idempotencyKey: 'http-shadow-1:page-data-delivery:v1', + taskType: 'page_data_dev', + required: true, + checks: [ + { id: 'agent_run_completion', status: 'passed' }, + { id: 'page_data_binding', status: 'passed' }, + { id: 'page_data_storage_policy', status: 'passed' }, + ], + metrics: { pageCount: 1, publicationCount: 1 }, + observedAt: 123, + }), + }, + ); + assert.equal(validation.status, 200); + assert.equal((await validation.json()).validation.verdict, 'passed'); + const events = await fetch(`${server.baseUrl}/v1/runs/http-shadow-1/events?after=1`, { headers: { authorization: 'Bearer test-service-token' }, }); const eventBody = await events.json(); assert.equal(events.status, 200); - assert.deepEqual(eventBody.events.map((event) => event.sequence), [2, 3]); + assert.deepEqual(eventBody.events.map((event) => event.sequence), [2, 3, 4]); + assert.equal(eventBody.events.at(-1).type, 'workflow_validation_observed'); const jobId = 'http-shadow-1:executor-preview'; const executorJob = await fetch( diff --git a/services/orchestrator/code-run-shadow-graph.mjs b/services/orchestrator/code-run-shadow-graph.mjs index 4181f78..0dc79e9 100644 --- a/services/orchestrator/code-run-shadow-graph.mjs +++ b/services/orchestrator/code-run-shadow-graph.mjs @@ -30,6 +30,10 @@ const ShadowState = Annotation.Root({ reducer: (_current, update) => update, default: () => null, }), + validation: Annotation({ + reducer: (_current, update) => update, + default: () => null, + }), events: Annotation({ reducer: (current, update) => [...current, ...update], default: () => [], diff --git a/services/orchestrator/contracts.mjs b/services/orchestrator/contracts.mjs index 62d2a61..8781293 100644 --- a/services/orchestrator/contracts.mjs +++ b/services/orchestrator/contracts.mjs @@ -16,8 +16,18 @@ export const DEFAULT_ORCHESTRATED_WORKFLOWS = Object.freeze([ 'code-run-v1', ]); +export const PAGE_DATA_VALIDATION_OBSERVATION_VERSION = + 'page-data-validation-observation-v1'; + const ENGINE_ID_PATTERN = /^[a-z][a-z0-9_-]{0,63}$/; const WORKFLOW_NAME_PATTERN = /^[a-z][a-z0-9_-]{0,127}$/; +const VALIDATION_CHECK_ID_PATTERN = /^[a-z][a-z0-9_-]{0,127}$/; +const VALIDATION_CHECK_STATUSES = new Set(['passed', 'failed', 'skipped']); +const REQUIRED_PAGE_DATA_CHECKS = Object.freeze([ + 'agent_run_completion', + 'page_data_binding', + 'page_data_storage_policy', +]); function normalizeIdentifier(value, pattern, fallback) { const normalized = String(value ?? '').trim().toLowerCase(); @@ -98,6 +108,82 @@ export function buildRunEvent({ }; } +function validationContractError(message) { + const error = new Error(message); + error.code = 'WORKFLOW_VALIDATION_INVALID'; + error.status = 422; + return error; +} + +function normalizeValidationCount(value) { + const parsed = Number(value); + if (!Number.isFinite(parsed) || parsed < 0) return 0; + return Math.min(1_000_000, Math.floor(parsed)); +} + +export function normalizePageDataValidationObservation(input = {}) { + const source = input && typeof input === 'object' && !Array.isArray(input) ? input : {}; + const idempotencyKey = String(source.idempotencyKey ?? '').trim().slice(0, 200); + if (!idempotencyKey) { + throw validationContractError('Page Data validation observation requires idempotencyKey'); + } + if (!Array.isArray(source.checks) || source.checks.length === 0) { + throw validationContractError('Page Data validation observation requires checks'); + } + const checks = source.checks.slice(0, 32).map((item) => { + const check = item && typeof item === 'object' && !Array.isArray(item) ? item : {}; + const id = normalizeIdentifier(check.id, VALIDATION_CHECK_ID_PATTERN, ''); + const status = String(check.status ?? '').trim().toLowerCase(); + if (!id || !VALIDATION_CHECK_STATUSES.has(status)) { + throw validationContractError('Page Data validation observation contains an invalid check'); + } + return { + id, + status, + codes: normalizeStringList(check.codes, { + limit: 32, + itemLimit: 128, + normalize: (value) => String(value ?? '').trim().toUpperCase(), + }), + }; + }); + const byId = new Map(checks.map((check) => [check.id, check])); + const failed = checks.some((check) => check.status === 'failed'); + const required = source.required === true; + const requiredChecksPassed = REQUIRED_PAGE_DATA_CHECKS.every( + (id) => byId.get(id)?.status === 'passed', + ); + const verdict = failed + ? 'failed' + : required && !requiredChecksPassed + ? 'inconclusive' + : checks.some((check) => check.status === 'passed') + ? 'passed' + : 'inconclusive'; + const taskType = normalizeWorkflowName(source.taskType, ''); + const observedAt = Number(source.observedAt); + return { + version: PAGE_DATA_VALIDATION_OBSERVATION_VERSION, + kind: 'page-data-delivery', + idempotencyKey, + taskType: taskType || null, + required, + verdict, + checks, + metrics: { + pageCount: normalizeValidationCount(source.metrics?.pageCount), + publicationCount: normalizeValidationCount(source.metrics?.publicationCount), + }, + source: normalizeIdentifier( + source.source, + VALIDATION_CHECK_ID_PATTERN, + 'portal-agent-run', + ), + dataPolicy: 'control-plane-only-v1', + observedAt: Number.isFinite(observedAt) && observedAt > 0 ? observedAt : Date.now(), + }; +} + export function stableRolloutBucket(value) { const hash = crypto.createHash('sha256').update(String(value ?? '')).digest(); return hash.readUInt32BE(0) % 100; diff --git a/services/orchestrator/contracts.test.mjs b/services/orchestrator/contracts.test.mjs index d53b0cf..b2d51b1 100644 --- a/services/orchestrator/contracts.test.mjs +++ b/services/orchestrator/contracts.test.mjs @@ -2,6 +2,7 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import { buildRunEvent, + normalizePageDataValidationObservation, normalizeRunSpec, stableRolloutBucket, } from './contracts.mjs'; @@ -33,6 +34,59 @@ test('orchestrator contracts normalize run specs without exposing framework type assert.equal(stableRolloutBucket('run-1'), stableRolloutBucket('run-1')); }); +test('Page Data validation observations derive a bounded control-plane verdict', () => { + const observation = normalizePageDataValidationObservation({ + idempotencyKey: 'run-1:page-data-delivery:v1', + taskType: 'page_data_dev_complex', + required: true, + checks: [ + { id: 'agent_run_completion', status: 'passed' }, + { id: 'page_data_binding', status: 'passed' }, + { id: 'page_data_storage_policy', status: 'passed' }, + { id: 'independent_review', status: 'skipped' }, + ], + metrics: { pageCount: 2, publicationCount: 1 }, + source: 'portal-agent-run', + observedAt: 123, + ignoredSensitivePayload: { + workspacePath: '/private/workspace', + instruction: 'must not be projected', + }, + }); + + assert.equal(observation.version, 'page-data-validation-observation-v1'); + assert.equal(observation.verdict, 'passed'); + assert.equal(observation.dataPolicy, 'control-plane-only-v1'); + assert.deepEqual(observation.metrics, { pageCount: 2, publicationCount: 1 }); + assert.equal('ignoredSensitivePayload' in observation, false); + assert.equal(JSON.stringify(observation).includes('/private/workspace'), false); + + const failed = normalizePageDataValidationObservation({ + idempotencyKey: 'run-2:page-data-delivery:v1', + required: true, + checks: [ + { + id: 'page_data_storage_policy', + status: 'failed', + codes: ['browser_storage_forbidden'], + }, + ], + }); + assert.equal(failed.verdict, 'failed'); + assert.deepEqual(failed.checks[0].codes, ['BROWSER_STORAGE_FORBIDDEN']); + + const inconclusive = normalizePageDataValidationObservation({ + idempotencyKey: 'run-3:page-data-delivery:v1', + required: true, + checks: [{ id: 'agent_run_completion', status: 'passed' }], + }); + assert.equal(inconclusive.verdict, 'inconclusive'); + assert.throws( + () => normalizePageDataValidationObservation({ idempotencyKey: 'missing-checks' }), + (error) => error.code === 'WORKFLOW_VALIDATION_INVALID' && error.status === 422, + ); +}); + test('workflow engine registry enforces the framework-neutral contract', () => { const engine = { id: 'native', @@ -71,6 +125,10 @@ test('remote workflow engine speaks only the versioned orchestrator HTTP contrac await engine.resume('run-1', { approvalId: 'approval-1', decision: 'approve' }); await engine.cancel('run-1'); await engine.getState('run-1'); + await engine.recordValidationObservation('run-1', { + idempotencyKey: 'run-1:page-data-delivery:v1', + checks: [{ id: 'agent_run_completion', status: 'passed' }], + }); const events = []; for await (const event of engine.streamEvents('run-1')) events.push(event); await engine.getExecutorJob('run-1:executor-preview'); @@ -86,6 +144,10 @@ test('remote workflow engine speaks only the versioned orchestrator HTTP contrac ['POST', 'http://orchestrator.internal/v1/runs/run-1/resume'], ['POST', 'http://orchestrator.internal/v1/runs/run-1/cancel'], ['GET', 'http://orchestrator.internal/v1/runs/run-1'], + [ + 'POST', + 'http://orchestrator.internal/v1/runs/run-1/validation-observations', + ], ['GET', 'http://orchestrator.internal/v1/runs/run-1/events'], ['GET', 'http://orchestrator.internal/v1/executor-jobs/run-1%3Aexecutor-preview'], [ diff --git a/services/orchestrator/engine-registry.mjs b/services/orchestrator/engine-registry.mjs index fad6c31..d739ba0 100644 --- a/services/orchestrator/engine-registry.mjs +++ b/services/orchestrator/engine-registry.mjs @@ -124,6 +124,12 @@ export function createRemoteWorkflowEngine({ async getState(runId) { return request(`/v1/runs/${encodeURIComponent(runId)}`); }, + async recordValidationObservation(runId, observation) { + return request(`/v1/runs/${encodeURIComponent(runId)}/validation-observations`, { + method: 'POST', + body: JSON.stringify(observation ?? {}), + }); + }, async *streamEvents(runId, cursor = null) { const query = cursor == null ? '' : `?after=${encodeURIComponent(cursor)}`; const result = await request(`/v1/runs/${encodeURIComponent(runId)}/events${query}`); diff --git a/services/orchestrator/runtime.mjs b/services/orchestrator/runtime.mjs index c65ec65..838991a 100644 --- a/services/orchestrator/runtime.mjs +++ b/services/orchestrator/runtime.mjs @@ -1,4 +1,8 @@ -import { buildRunEvent, normalizeRunSpec } from './contracts.mjs'; +import { + buildRunEvent, + normalizePageDataValidationObservation, + normalizeRunSpec, +} from './contracts.mjs'; import { CODE_RUN_SHADOW_WORKFLOW, createCodeRunShadowGraph, @@ -22,6 +26,12 @@ function graphConfig(runId) { }; } +function validationObservationFingerprint(observation) { + if (!observation || typeof observation !== 'object') return ''; + const { observedAt: _observedAt, ...stable } = observation; + return JSON.stringify(stable); +} + function projectSnapshot(runId, snapshot) { const values = snapshot?.values ?? {}; if (!values.spec) return null; @@ -34,6 +44,7 @@ function projectSnapshot(runId, snapshot) { phase: values.phase ?? null, plan: values.plan ?? null, result: values.result ?? null, + validation: values.validation ?? null, createdAt: values.events?.[0]?.timestamp ?? null, updatedAt: values.events?.at?.(-1)?.timestamp ?? null, }; @@ -136,6 +147,49 @@ export function createLangGraphOrchestratorRuntime({ ); } + async function recordValidationObservation(runId, input = {}) { + const normalizedRunId = String(runId ?? '').trim(); + if (!normalizedRunId) return null; + const snapshot = await getSnapshot(normalizedRunId); + const values = snapshot?.values ?? {}; + if (!values.spec) return null; + const projected = projectSnapshot(normalizedRunId, snapshot); + if (!TERMINAL_RUN_STATUSES.has(projected.status)) { + const error = new Error('Workflow run must be terminal before validation observation'); + error.code = 'WORKFLOW_VALIDATION_RUN_NOT_TERMINAL'; + error.status = 409; + throw error; + } + const observation = normalizePageDataValidationObservation(input); + const existing = values.validation ?? null; + if (existing?.idempotencyKey === observation.idempotencyKey) { + if ( + validationObservationFingerprint(existing) + !== validationObservationFingerprint(observation) + ) { + const error = new Error('Validation observation idempotency key conflicts with existing data'); + error.code = 'WORKFLOW_VALIDATION_IDEMPOTENCY_CONFLICT'; + error.status = 409; + throw error; + } + return projected; + } + await graph.updateState( + graphConfig(normalizedRunId), + { + validation: observation, + events: [buildRunEvent({ + runId: normalizedRunId, + sequence: (values.events?.length ?? 0) + 1, + type: 'workflow_validation_observed', + data: observation, + })], + }, + 'finalize_run', + ); + return getState(normalizedRunId); + } + async function deleteRun(runId) { const normalizedRunId = String(runId ?? '').trim(); if (!normalizedRunId) return null; @@ -204,6 +258,7 @@ export function createLangGraphOrchestratorRuntime({ }, getState, + recordValidationObservation, async listEvents(runId, { after = 0 } = {}) { const normalizedRunId = String(runId ?? '').trim(); @@ -397,4 +452,5 @@ export const orchestratorRuntimeInternals = { graphConfig, projectSnapshot, TERMINAL_RUN_STATUSES, + validationObservationFingerprint, }; diff --git a/services/orchestrator/runtime.test.mjs b/services/orchestrator/runtime.test.mjs index d6fb5ee..1a42ea8 100644 --- a/services/orchestrator/runtime.test.mjs +++ b/services/orchestrator/runtime.test.mjs @@ -87,6 +87,61 @@ test('LangGraph runtime is idempotent by run id', async () => { assert.equal((await runtime.listEvents('run-shadow-1')).events.length, 3); }); +test('LangGraph runtime records idempotent Page Data validation evidence after Shadow completion', async () => { + const runtime = createLangGraphOrchestratorRuntime({ + checkpointer: new MemorySaver(), + }); + await runtime.start(runSpec({ + input: { + instruction: '[shadow-control-plane-only]', + taskType: 'page_data_dev_complex', + pageDataRequired: true, + }, + })); + const input = { + idempotencyKey: 'run-shadow-1:page-data-delivery:v1', + taskType: 'page_data_dev_complex', + required: true, + checks: [ + { id: 'agent_run_completion', status: 'passed' }, + { id: 'page_data_binding', status: 'passed' }, + { id: 'page_data_storage_policy', status: 'passed' }, + ], + metrics: { pageCount: 2, publicationCount: 1 }, + observedAt: 123, + }; + + const recorded = await runtime.recordValidationObservation('run-shadow-1', input); + assert.equal(recorded.status, 'succeeded'); + assert.equal(recorded.validation.verdict, 'passed'); + assert.equal(recorded.validation.dataPolicy, 'control-plane-only-v1'); + assert.deepEqual( + (await runtime.listEvents('run-shadow-1')).events.map((event) => event.type), + [ + 'workflow_validated', + 'workflow_planned', + 'workflow_completed', + 'workflow_validation_observed', + ], + ); + + const duplicate = await runtime.recordValidationObservation('run-shadow-1', { + ...input, + observedAt: 456, + }); + assert.deepEqual(duplicate, recorded); + assert.equal((await runtime.listEvents('run-shadow-1')).events.length, 4); + await assert.rejects( + () => runtime.recordValidationObservation('run-shadow-1', { + ...input, + checks: [{ id: 'page_data_binding', status: 'failed' }], + }), + (error) => error.code === 'WORKFLOW_VALIDATION_IDEMPOTENCY_CONFLICT' + && error.status === 409, + ); + assert.equal(await runtime.recordValidationObservation('missing-run', input), null); +}); + test('LangGraph runtime deletes terminal checkpoints and linked Executor Jobs', async () => { const runtime = createLangGraphOrchestratorRuntime({ checkpointer: new MemorySaver(), diff --git a/services/orchestrator/shadow-observer.mjs b/services/orchestrator/shadow-observer.mjs index 82424a5..26fdb0f 100644 --- a/services/orchestrator/shadow-observer.mjs +++ b/services/orchestrator/shadow-observer.mjs @@ -1,9 +1,12 @@ import { WORKFLOW_ENGINE, + normalizePageDataValidationObservation, normalizeRunSpec, } from './contracts.mjs'; import { createRemoteWorkflowEngine } from './engine-registry.mjs'; +const VALIDATION_RETRY_DELAYS_MS = Object.freeze([25, 75, 225]); + function safeError(error) { return { code: String(error?.code ?? 'WORKFLOW_SHADOW_FAILED').slice(0, 128), @@ -11,6 +14,20 @@ function safeError(error) { }; } +async function recordValidationWithRetry(engine, runId, observation) { + let attempt = 0; + while (true) { + try { + return await engine.recordValidationObservation(runId, observation); + } catch (error) { + const delayMs = VALIDATION_RETRY_DELAYS_MS[attempt]; + if (error?.status !== 404 || delayMs == null) throw error; + attempt += 1; + await new Promise((resolve) => setTimeout(resolve, delayMs)); + } + } +} + export function createWorkflowShadowObserver({ configService, serviceToken = process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN, @@ -21,14 +38,43 @@ export function createWorkflowShadowObserver({ throw new Error('Workflow shadow observer requires orchestrator config service'); } - return async function observeWorkflowRun({ + async function selectShadowEngine({ + runId, + requestId, + userId, + workflowName, + }) { + const selection = await configService.selectEngine({ + runId, + requestId, + userId, + workflowName, + }); + if (selection.shadowEngine !== WORKFLOW_ENGINE.LANGGRAPH) { + return { selection, engine: null }; + } + const state = await configService.getRuntimeState(); + return { + selection, + engine: createRemoteWorkflowEngine({ + id: WORKFLOW_ENGINE.LANGGRAPH, + baseUrl: state.config.serviceUrl, + serviceToken, + timeoutMs: state.config.requestTimeoutMs, + fetchImpl, + }), + }; + } + + async function observeWorkflowRun({ runId, requestId, userId, workflowName = 'code-run-v1', taskType = null, + pageDataRequired = false, } = {}) { - const selection = await configService.selectEngine({ + const { selection, engine } = await selectShadowEngine({ runId, requestId, userId, @@ -49,7 +95,7 @@ export function createWorkflowShadowObserver({ dryRun: true, handoffAllowed: false, } : null; - if (selection.shadowEngine !== WORKFLOW_ENGINE.LANGGRAPH) { + if (!engine) { return { observed: false, reason: selection.reason, @@ -58,14 +104,6 @@ export function createWorkflowShadowObserver({ }; } - const state = await configService.getRuntimeState(); - const engine = createRemoteWorkflowEngine({ - id: WORKFLOW_ENGINE.LANGGRAPH, - baseUrl: state.config.serviceUrl, - serviceToken, - timeoutMs: state.config.requestTimeoutMs, - fetchImpl, - }); const spec = normalizeRunSpec({ runId, requestId, @@ -75,6 +113,7 @@ export function createWorkflowShadowObserver({ instruction: '[shadow-control-plane-only]', taskType, toolMode: 'code', + pageDataRequired: pageDataRequired === true, }, policy: { executionMode: 'observe-only', @@ -104,9 +143,49 @@ export function createWorkflowShadowObserver({ } throw error; } + } + + observeWorkflowRun.observeValidation = async function observeWorkflowValidation({ + runId, + requestId, + userId, + workflowName = 'code-run-v1', + observation, + } = {}) { + const { selection, engine } = await selectShadowEngine({ + runId, + requestId, + userId, + workflowName, + }); + if (!engine) { + return { + observed: false, + reason: selection.reason, + mode: selection.mode, + }; + } + const normalized = normalizePageDataValidationObservation(observation); + try { + const result = await recordValidationWithRetry(engine, runId, normalized); + return { + observed: true, + engine: WORKFLOW_ENGINE.LANGGRAPH, + mode: selection.mode, + configVersion: selection.configVersion, + validation: result.validation ?? normalized, + shadowRun: result, + }; + } catch (error) { + logger.warn('[orchestrator-shadow] validation observation failed:', safeError(error)); + throw error; + } }; + + return observeWorkflowRun; } export const workflowShadowObserverInternals = { + recordValidationWithRetry, safeError, }; diff --git a/services/orchestrator/shadow-observer.test.mjs b/services/orchestrator/shadow-observer.test.mjs index 79c4dbc..40eaae2 100644 --- a/services/orchestrator/shadow-observer.test.mjs +++ b/services/orchestrator/shadow-observer.test.mjs @@ -179,6 +179,7 @@ test('shadow observer sends a control-plane-only RunSpec without user content or userId: 'user-2', sessionId: 'session-2', taskType: 'code-change', + pageDataRequired: true, userMessage: { content: [{ type: 'text', text: 'Implement the service boundary' }], }, @@ -192,6 +193,7 @@ test('shadow observer sends a control-plane-only RunSpec without user content or assert.equal(body.policy.executionMode, 'observe-only'); assert.equal(body.policy.sideEffectsAllowed, false); assert.equal(body.input.instruction, '[shadow-control-plane-only]'); + assert.equal(body.input.pageDataRequired, true); assert.equal('sessionRef' in body.input, false); assert.deepEqual(body.subject, { tenantId: null, userId: null }); assert.equal(body.metadata.configVersion, 7); @@ -270,6 +272,37 @@ test('isolated Portal shadow flow reaches LangGraph over HTTP and removes all te assert.equal(checkpointJson.includes('private-user-id'), false); assert.equal(checkpointJson.includes('private-session-id'), false); + const validation = await observer.observeValidation({ + runId: 'isolated-shadow-run-1', + requestId: 'isolated-shadow-request-1', + userId: 'private-user-id', + workflowName: 'code-run-v1', + observation: { + idempotencyKey: 'isolated-shadow-run-1:page-data-delivery:v1', + taskType: 'page_data_dev_complex', + required: true, + checks: [ + { id: 'agent_run_completion', status: 'passed' }, + { id: 'page_data_binding', status: 'passed' }, + { id: 'page_data_storage_policy', status: 'passed' }, + ], + metrics: { pageCount: 2, publicationCount: 1 }, + source: 'portal-agent-run', + observedAt: 123, + sensitive: { + workspacePath: '/private/workspace', + instruction: secretUserContent, + }, + }, + }); + assert.equal(validation.observed, true); + assert.equal(validation.validation.verdict, 'passed'); + const validatedState = await runtime.getState('isolated-shadow-run-1'); + assert.equal(validatedState.validation.verdict, 'passed'); + const validatedJson = JSON.stringify(validatedState.validation); + assert.equal(validatedJson.includes('/private/workspace'), false); + assert.equal(validatedJson.includes(secretUserContent), false); + const deleted = await fetch(`${server.baseUrl}/v1/runs/isolated-shadow-run-1`, { method: 'DELETE', headers: { authorization: 'Bearer isolated-shadow-token' },