diff --git a/.env.example b/.env.example index 00e569e..b311761 100644 --- a/.env.example +++ b/.env.example @@ -61,6 +61,24 @@ H5_PUBLIC_BASE_URL=http://127.0.0.1:5173 # MEMIND_AGENT_RUN_WORKER_ID= # MEMIND_AGENT_RUN_WORKER_EXPECT_RUNNING=0 +# LangGraph Orchestrator Shadow（默认完全不接入 Agent Run 请求路径）。 +# 仅在独立 Orchestrator 已健康、service token 已配置后，先显式打开观察 wiring， +# 再通过 memindadm 选择 shadow。关闭 wiring 时不查询编排配置、不排队、不传用户数据。 +# MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED=0 +# MEMIND_ORCHESTRATOR_URL=http://127.0.0.1:8093 +# MEMIND_ORCHESTRATOR_SERVICE_TOKEN= +# MEMIND_ORCHESTRATOR_SHADOW_MAX_CONCURRENCY=2 +# MEMIND_ORCHESTRATOR_SHADOW_MAX_QUEUE=100 +# 非 loopback URL 必须精确列入；逗号分隔，只接受 origin。 +# MEMIND_ORCHESTRATOR_ALLOWED_ORIGINS=https://orchestrator.internal +# 真实执行交接是另一道独立闸门，默认保持关闭。 +# MEMIND_ORCHESTRATOR_EXECUTION_HANDOFF_ENABLED=0 +# MEMIND_ORCHESTRATOR_KILL_SWITCH=0 +# Orchestrator 终态数据保留清理默认关闭；在独立服务环境配置，建议真实 Canary 前明确周期。 +# MEMIND_ORCHESTRATOR_RETENTION_DAYS=0 +# MEMIND_ORCHESTRATOR_RETENTION_SWEEP_INTERVAL_MS=86400000 +# MEMIND_ORCHESTRATOR_RETENTION_SWEEP_LIMIT=100 + # Runtime SLO 日报（默认每日 23:55 写 reports/runtime-slo）。 # MEMIND_RUNTIME_REPORT_DIR=/Users/john/Project/Memind/reports/runtime-slo # MEMIND_RUNTIME_SLO_HOUR=23 diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index f696eb7..51abcd9 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -15,6 +15,7 @@ import { deriveUserFacingText, } from './conversation-display.mjs'; import { ensureGooseUserMessageMetadata } from './goose-message.mjs'; +import { createShadowObservationDispatcher } from './services/orchestrator/shadow-dispatcher.mjs'; import { prepareAndDetectSessionDeliverables, SESSION_FINISHED_STALE_GRACE_MS, @@ -413,6 +414,14 @@ export function createAgentRunGateway({ process.env.MEMIND_AGENT_RUN_HEARTBEAT_MS, DEFAULT_RUN_HEARTBEAT_MS, ), + maxConcurrentShadowObservations = positiveInteger( + process.env.MEMIND_ORCHESTRATOR_SHADOW_MAX_CONCURRENCY, + 2, + ), + maxQueuedShadowObservations = positiveInteger( + process.env.MEMIND_ORCHESTRATOR_SHADOW_MAX_QUEUE, + 100, + ), workerIdentity = null, }) { const worker = normalizeAgentRunWorkerIdentity(workerIdentity ?? {}); @@ -507,44 +516,52 @@ export function createAgentRunGateway({ }); } + const shadowDispatcher = createShadowObservationDispatcher({ + observe: observeWorkflowRun, + maxConcurrent: maxConcurrentShadowObservations, + maxQueued: maxQueuedShadowObservations, + onCompleted: async (input, result, context) => { + 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() - context.enqueuedAt), + }); + }, + onFailed: async (input, error, context) => { + 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() - context.enqueuedAt), + }).catch((appendError) => { + console.warn( + '[AgentRun] workflow shadow failure event skipped:', + appendError instanceof Error ? appendError.message : appendError, + ); + }); + }, + onSkipped: async (input, context) => { + await appendEvent(input.runId, 'workflow_shadow_skipped', { + reason: context.reason, + }); + }, + logger: console, + }); + 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, - ); - }); - }); - }); + if (input.toolMode !== 'code') return; + shadowDispatcher.dispatch(input); } async function appendRunSnapshot(runId) { diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 8525bff..1a18751 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -457,6 +457,105 @@ test('agent run starts a session and marks submitted reply as succeeded', async assert.equal(pool.runs.get(run.id).attempts, 1); }); +test('workflow shadow failure cannot change the native agent run result', async () => { + const pool = createFakePool(); + const submitted = []; + const shadowError = Object.assign(new Error('orchestrator unavailable'), { + code: 'ORCHESTRATOR_UNAVAILABLE', + }); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async startSessionForUser() { + return { id: 'session-shadow-failure' }; + }, + async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { + submitted.push({ userId, sessionId, requestId, userMessage }); + }, + }, + observeWorkflowRun: async () => { + throw shadowError; + }, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-shadow-failure', { + requestId: 'req-shadow-failure', + userMessage: { role: 'user', content: [{ type: 'text', text: 'native must win' }] }, + toolMode: 'code', + taskType: 'repo_refactor', + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); + await waitFor(() => pool.events.some( + (event) => event.runId === run.id && event.eventType === 'workflow_shadow_failed', + )); + + assert.equal(submitted.length, 1); + assert.equal(pool.runs.get(run.id).status, 'succeeded'); + assert.equal(pool.runs.get(run.id).attempts, 1); + const failureEvent = pool.events.find( + (event) => event.runId === run.id && event.eventType === 'workflow_shadow_failed', + ); + const failureData = JSON.parse(failureEvent.dataJson); + assert.equal(failureData.code, 'ORCHESTRATOR_UNAVAILABLE'); + assert.equal(failureData.message, 'orchestrator unavailable'); + assert.ok(Number.isFinite(failureData.latencyMs)); + assert.ok(failureData.latencyMs >= 0); +}); + +test('workflow shadow queue overflow skips observation without changing native queued runs', async () => { + const pool = createFakePool(); + const observedRunIds = []; + let releaseFirstObservation; + const firstObservationBlocked = new Promise((resolve) => { + releaseFirstObservation = resolve; + }); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: {}, + autoDispatch: false, + maxConcurrentShadowObservations: 1, + maxQueuedShadowObservations: 1, + observeWorkflowRun: async (input) => { + observedRunIds.push(input.runId); + if (observedRunIds.length === 1) await firstObservationBlocked; + return { observed: false, reason: 'test_observation_complete' }; + }, + }); + + const createCodeRun = (suffix) => gateway.createRun(`user-shadow-${suffix}`, { + requestId: `req-shadow-${suffix}`, + userMessage: { role: 'user', content: [{ type: 'text', text: `run ${suffix}` }] }, + toolMode: 'code', + taskType: 'repo_refactor', + }); + + const first = await createCodeRun('first'); + await waitFor(() => observedRunIds.length === 1); + const second = await createCodeRun('second'); + const third = await createCodeRun('third'); + + await waitFor(() => pool.events.some( + (event) => event.runId === third.id && event.eventType === 'workflow_shadow_skipped', + )); + assert.deepEqual( + [first, second, third].map((run) => pool.runs.get(run.id).status), + ['queued', 'queued', 'queued'], + ); + assert.deepEqual(JSON.parse(pool.events.find( + (event) => event.runId === third.id && event.eventType === 'workflow_shadow_skipped', + ).dataJson), { + reason: 'shadow_queue_full', + }); + + releaseFirstObservation(); + await waitFor(() => observedRunIds.length === 2); + assert.deepEqual(observedRunIds, [first.id, second.id]); +}); + test('agent run policy allow path preserves existing routing and submission behavior', async () => { const pool = createFakePool(); const submitted = []; diff --git a/deploy/orchestrator/.env.example b/deploy/orchestrator/.env.example index cfa4f8a..cf1a213 100644 --- a/deploy/orchestrator/.env.example +++ b/deploy/orchestrator/.env.example @@ -9,6 +9,10 @@ MEMIND_ORCHESTRATOR_EXECUTION_ENABLED=0 MEMIND_ORCHESTRATOR_ENABLED_EXECUTORS= MEMIND_ORCHESTRATOR_USER_ALLOWLIST= MEMIND_ORCHESTRATOR_TENANT_ALLOWLIST= +# Terminal checkpoint/job retention is disabled until explicitly approved. +MEMIND_ORCHESTRATOR_RETENTION_DAYS=0 +MEMIND_ORCHESTRATOR_RETENTION_SWEEP_INTERVAL_MS=86400000 +MEMIND_ORCHESTRATOR_RETENTION_SWEEP_LIMIT=100 MEMIND_EXECUTOR_GOOSED_URL=https://host.docker.internal:18006 MEMIND_EXECUTOR_GOOSED_TOKEN= MEMIND_EXECUTOR_GOOSED_TLS_INSECURE=1 diff --git a/docs/architecture/memind-orchestrator-boundary.md b/docs/architecture/memind-orchestrator-boundary.md index d9c6960..99f150b 100644 --- a/docs/architecture/memind-orchestrator-boundary.md +++ b/docs/architecture/memind-orchestrator-boundary.md @@ -54,6 +54,18 @@ Native selection. The initial workflow allowlist contains only `code-run-v1`. Ordinary chat and existing Goosed session traffic remain outside the Orchestrator path. +Portal also has an independent wiring gate: +`MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED=1`. It is off by default. +When it is off, Portal does not construct the Shadow observer, query routing +configuration from the Agent Run path, schedule background observations, or +send data to the Orchestrator. Changing memindadm mode alone therefore cannot +put existing Native traffic on the Shadow path. + +memindadm projects the requested and effective wiring states separately. Canary +readiness requires `shadow_wiring_enabled=true`, so a saved Shadow mode with a +closed Portal gate is visible and cannot be mistaken for a collecting Shadow +deployment. + 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. @@ -145,6 +157,8 @@ POST /v1/runs GET /v1/runs/:id POST /v1/runs/:id/resume POST /v1/runs/:id/cancel +DELETE /v1/runs/:id +POST /v1/maintenance/purge-terminal-runs GET /v1/runs/:id/events?after= GET /v1/executor-jobs/:jobId GET /v1/executor-jobs/:jobId/events?after=&limit= @@ -154,6 +168,15 @@ 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. +Shadow uses the stricter `control-plane-only-v1` data policy. Portal evaluates +the rollout locally but sends no user message, user identifier, session +identifier, provider key, or workspace path to LangGraph. The remote RunSpec +contains only product run/request identifiers, workflow/task type, configuration +version, and a fixed non-user instruction marker. Terminal runs can be removed +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`. + The graph keeps three deterministic control-plane nodes: ```text @@ -205,10 +228,29 @@ failure: 4. remains removable by deleting the observer wiring and changing only the service URL boundary. +The observer uses a bounded in-process dispatcher. Concurrency defaults to `2` +and the waiting queue to `100`; overflow records `workflow_shadow_skipped` and +never blocks, rejects, or cancels the Native run. Skipped observations are +included in the denominator, shown separately in memindadm, and any non-zero +skip rate blocks Canary readiness. + Successful observations are projected as `workflow_shadow_completed`; graph checkpoints and blocked Executor Jobs stay in the Orchestrator-owned PostgreSQL database. +Non-loopback Orchestrator binding requires a service token. Enabling Executor +Jobs requires both distinct service and worker tokens even on loopback. +Non-loopback service URLs are rejected by Portal unless their origin is +declared through `MEMIND_ORCHESTRATOR_URL` or +`MEMIND_ORCHESTRATOR_ALLOWED_ORIGINS`; loopback origins remain allowed for +local validation. + +Terminal retention is independently disabled by default. Setting +`MEMIND_ORCHESTRATOR_RETENTION_DAYS` to a positive value starts a bounded +periodic sweep after service startup; it uses the same terminal-only deletion +path and never deletes waiting/running workflows. The authenticated maintenance +endpoint defaults to dry-run and requires `apply=true` for mutation. + 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 @@ -218,6 +260,7 @@ 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. +It also requires the independent Portal Shadow wiring gate to be effective. The result is advisory: `manual_canary_review` never mutates routing configuration or transfers execution ownership. diff --git a/docs/orchestrator-phase5-runbook.md b/docs/orchestrator-phase5-runbook.md index ff25f89..4703e1e 100644 --- a/docs/orchestrator-phase5-runbook.md +++ b/docs/orchestrator-phase5-runbook.md @@ -7,6 +7,7 @@ making it the default executor. The default remains: ```text memindadm executionEnabled=false +Portal MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED=0 Portal MEMIND_ORCHESTRATOR_EXECUTION_HANDOFF_ENABLED=0 Orchestrator MEMIND_ORCHESTRATOR_EXECUTION_ENABLED=0 enabled executor list empty @@ -50,7 +51,11 @@ 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. +2. Set Portal `MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED=1`, restart only + the local/target Portal instance, then select `shadow` to collect Native + versus LangGraph planning evidence. Confirm memindadm shows `Shadow wiring: + 已开启`; Canary readiness must include a passing `shadow_wiring_enabled` + check before evidence collection is considered valid. 3. `canary`, execution switch off: verify selection denominator and readiness. 4. `canary`, execution switch on: one explicit user, one workspace alias, zero percentage rollout. @@ -71,10 +76,11 @@ 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`. +3. Set Portal `MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED=0`. +4. Set Portal `MEMIND_ORCHESTRATOR_EXECUTION_HANDOFF_ENABLED=0`. +5. Set Orchestrator `MEMIND_ORCHESTRATOR_EXECUTION_ENABLED=0`. +6. Remove an executor from `MEMIND_ORCHESTRATOR_ENABLED_EXECUTORS`. +7. 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. @@ -88,12 +94,52 @@ 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; +- any `workflow_shadow_skipped` event or non-zero Shadow skip rate; +- `shadow_wiring_enabled` becoming false while memindadm mode remains Shadow; - 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. +## Data lifecycle + +Shadow uses `control-plane-only-v1`: user content, user ID, and session ID are +not copied into Orchestrator storage. The product run ID remains the deletion +join key. + +Before deleting the corresponding Memind run or completing a user-data purge, +call the authenticated endpoint for every linked run: + +```text +DELETE /v1/runs/:runId +``` + +Only terminal runs can be deleted. The endpoint deletes the LangGraph +checkpoint thread and linked terminal Executor Job; PostgreSQL cascades its job +events. A `409` means execution is still active and must first be cancelled or +allowed to reach a terminal state. This endpoint is internal and must never be +exposed without the service-token boundary. + +Bulk retention remains off unless explicitly configured: + +```text +MEMIND_ORCHESTRATOR_RETENTION_DAYS=30 +MEMIND_ORCHESTRATOR_RETENTION_SWEEP_INTERVAL_MS=86400000 +MEMIND_ORCHESTRATOR_RETENTION_SWEEP_LIMIT=100 +``` + +Before enabling it, preview the exact terminal candidates through the +authenticated endpoint. Mutation requires `apply=true`; omitting it is always +dry-run: + +```text +POST /v1/maintenance/purge-terminal-runs +{"before": , "limit": 100} +``` + +The periodic sweep never runs when retention days is empty or `0`. + ## Backup and disaster recovery Create and verify a PostgreSQL custom-format backup: diff --git a/ops/src/api/admin.ts b/ops/src/api/admin.ts index 004c552..3fea929 100644 --- a/ops/src/api/admin.ts +++ b/ops/src/api/admin.ts @@ -222,6 +222,12 @@ export type OrchestratorRuntime = { executesLangGraph: boolean; plansLangGraph: boolean; shadowsLangGraph: boolean; + shadowObservation: { + requested: boolean; + enabled: boolean; + reason: string | null; + environmentGate: boolean; + }; executionHandoff: { implemented: boolean; requested: boolean; @@ -286,8 +292,10 @@ export type OrchestratorShadowMetrics = { observations: number; successes: number; failures: number; + skipped: number; successRate: number | null; failureRate: number | null; + skipRate: number | null; latencyP50Ms: number | null; latencyP95Ms: number | null; nativeSucceeded: number; @@ -304,7 +312,7 @@ export type OrchestratorShadowRun = { sessionId: string | null; nativeStatus: string; nativeAttempts: number; - shadowStatus: 'succeeded' | 'failed'; + shadowStatus: 'succeeded' | 'failed' | 'skipped'; engine: string; configVersion: number | null; phase: string | null; @@ -313,6 +321,7 @@ export type OrchestratorShadowRun = { executorAdapter: string | null; latencyMs: number | null; error: { code: string; message: string } | null; + skipReason: string | null; observedAt: number; nativeCompletedAt: number | null; }; @@ -386,6 +395,7 @@ export type OrchestratorCanaryReadiness = { hours: number; minObservations: number; minSuccessRate: number; + maxSkipRate: number; maxP95LatencyMs: number; minLatencyCoverageRate: number; minNativeSettledRate: number; @@ -398,7 +408,9 @@ export type OrchestratorCanaryReadiness = { excludedSynthetic: number; successes: number; failures: number; + skipped: number; successRate: number | null; + skipRate: number | null; latencyCoverageRate: number | null; latencyP95Ms: number | null; nativeSettledRate: number | null; @@ -521,7 +533,7 @@ export async function fetchOrchestratorRuntime() { export async function fetchOrchestratorShadowRuns(params: { hours?: number; limit?: number; - status?: 'all' | 'succeeded' | 'failed'; + status?: 'all' | 'succeeded' | 'failed' | 'skipped'; } = {}) { const query = new URLSearchParams(); if (params.hours) query.set('hours', String(params.hours)); diff --git a/ops/src/pages/admin/OrchestratorPage.tsx b/ops/src/pages/admin/OrchestratorPage.tsx index 56f2a79..2df94a6 100644 --- a/ops/src/pages/admin/OrchestratorPage.tsx +++ b/ops/src/pages/admin/OrchestratorPage.tsx @@ -25,6 +25,7 @@ function formatTime(value?: number | null) { const RUNTIME_REASON: Record = { mode_off: '当前为关闭模式，所有任务继续由 Native Agent Run 执行。', service_url_missing: '尚未配置 Orchestrator 服务地址，LangGraph 不会接管任务。', + service_url_not_allowed: 'Orchestrator 地址不在环境 origin 白名单中，Portal 不会访问该地址。', kill_switch: '环境级紧急熔断已开启，强制回退 Native。', execution_gate_disabled: 'Canary / Active 仅生成 Dry-run 候选决策；执行交接闸门仍被硬锁定，Native 是唯一执行者。', environment_execution_gate_disabled: 'memindadm 已请求执行交接，但环境级执行闸门仍关闭；当前继续 Dry-run。', @@ -122,11 +123,19 @@ export function OrchestratorPage() { const runtimeMessage = state.runtime.reason ? (RUNTIME_REASON[state.runtime.reason] ?? state.runtime.reason) - : state.runtime.shadowsLangGraph + : state.runtime.shadowObservation.requested + && !state.runtime.shadowObservation.enabled + ? 'Shadow 配置已保存，但 Portal 环境 wiring 仍关闭；现有 Agent Run 不会查询或调用 Orchestrator。' + : state.runtime.shadowsLangGraph ? 'Shadow 已生效：Native 继续执行，LangGraph 只观察和生成决策。' : state.runtime.executesLangGraph ? '执行交接已开启。' : '配置有效。'; + const shadowWiringLabel = state.runtime.shadowObservation.enabled + ? '已开启' + : state.runtime.shadowObservation.requested + ? '环境闸门关闭' + : '未请求'; return (
@@ -164,6 +173,15 @@ export function OrchestratorPage() { {state.runtime.executionHandoff.enabled ? '已开启' : 'Dry-run 锁定'}

+
+ Shadow wiring +

+ {shadowWiringLabel} +

+

{runtimeMessage} diff --git a/ops/src/pages/admin/OrchestratorShadowPanel.tsx b/ops/src/pages/admin/OrchestratorShadowPanel.tsx index 4a9c5ea..4ceb8cd 100644 --- a/ops/src/pages/admin/OrchestratorShadowPanel.tsx +++ b/ops/src/pages/admin/OrchestratorShadowPanel.tsx @@ -27,12 +27,14 @@ function shortId(value: string) { const readinessCheckLabels: Record = { shadow_mode: '当前保持 Shadow', + shadow_wiring_enabled: 'Portal Shadow wiring', service_healthy: '服务健康', durable_checkpoint: '持久化 checkpoint', durable_executor_job_store: '持久化 Executor Job Store', observe_only: '仅观察不执行', sample_volume: '有效样本量', shadow_success_rate: 'Shadow 成功率', + shadow_skip_rate: 'Shadow 跳过率', latency_coverage: '延迟数据覆盖率', latency_p95: '延迟 P95', native_settled_rate: 'Native 终态覆盖率', @@ -44,7 +46,7 @@ const readinessCheckLabels: Record = { function formatReadinessValue(check: OrchestratorCanaryReadinessCheck) { const value = check.actual; if (value == null) return '无数据'; - if (['shadow_success_rate', 'latency_coverage', 'native_settled_rate'].includes(check.id)) { + if (['shadow_success_rate', 'shadow_skip_rate', 'latency_coverage', 'native_settled_rate'].includes(check.id)) { return formatPercent(Number(value)); } if (check.id === 'latency_p95') return formatLatency(Number(value)); @@ -56,6 +58,7 @@ function formatReadinessValue(check: OrchestratorCanaryReadinessCheck) { function formatReadinessTarget(check: OrchestratorCanaryReadinessCheck) { const target = check.target; if (target == null) return ''; + if (check.id === 'shadow_skip_rate') return `≤ ${formatPercent(Number(target))}`; if (['shadow_success_rate', 'latency_coverage', 'native_settled_rate'].includes(check.id)) { return `≥ ${formatPercent(Number(target))}`; } @@ -75,7 +78,7 @@ const metricCardStyle = { export function OrchestratorShadowPanel() { const [hours, setHours] = useState(24); - const [status, setStatus] = useState<'all' | 'succeeded' | 'failed'>('all'); + const [status, setStatus] = useState<'all' | 'succeeded' | 'failed' | 'skipped'>('all'); const [data, setData] = useState(null); const [readiness, setReadiness] = useState(null); const [selected, setSelected] = useState(null); @@ -141,6 +144,7 @@ export function OrchestratorShadowPanel() { + - + {run.shadowStatus} {run.nativeStatus} @@ -267,7 +284,9 @@ export function OrchestratorShadowPanel() { - {run.error ? `${run.error.code}: ${run.error.message}` : '—'} + {run.error + ? `${run.error.code}: ${run.error.message}` + : run.skipReason ?? '—'} ))} diff --git a/package.json b/package.json index ebd2031..392e4a8 100644 --- a/package.json +++ b/package.json @@ -66,7 +66,7 @@ "ci:page-data-dev-loop-smoke": "node scripts/ci-page-data-dev-loop-smoke.mjs", "migrate:agent-code-run-config": "node scripts/migrate-agent-code-run-config-from-env.mjs", "test:scenario:john4-diet": "node scripts/run-scenario-test.mjs --scenario john4-children-hobby-diet-update", - "test": "node --test auth.test.mjs asr-proxy.test.mjs billing.test.mjs billing-token-state.test.mjs billing-recharge.test.mjs wechat-pay.test.mjs wechat-oauth.test.mjs wechat-mp.test.mjs wechat-media.test.mjs wechat/image-generation-policy.test.mjs wechat/verify/generated-thumbnail.test.mjs schedule-intent.test.mjs schedule-reminder-worker.test.mjs capabilities.test.mjs policies.test.mjs server/portal-api-auth-middleware.test.mjs server/portal-config-routes.test.mjs server/portal-plaza-discovery-routes.test.mjs server/portal-runtime-routes.test.mjs chat-skills.test.mjs chat-intent-router.test.mjs chat-finish-sync.test.mjs chat-agent-run-gate.test.mjs conversation-display.test.mjs user-publish.test.mjs user-memory-profile.test.mjs skills-registry.test.mjs skill-runtime-policy.test.mjs excel-analyst.test.mjs agent-run-gateway.test.mjs agent-run-routes.test.mjs session-broker.test.mjs sse-event-taxonomy.test.mjs goosed-proxy-boundary.test.mjs agent-run-stream.test.mjs mindspace-h5-html-finish-guard.test.mjs admin-routes.test.mjs services/orchestrator/admin-config.test.mjs services/orchestrator/contracts.test.mjs services/orchestrator/checkpoint.test.mjs services/orchestrator/runtime.test.mjs services/orchestrator/app.test.mjs services/orchestrator/shadow-observer.test.mjs services/orchestrator/observability.test.mjs image-make-admin-config.test.mjs asset-gateway.test.mjs image-make-client.test.mjs mindspace-image-generation.test.mjs mindspace-image-generation-routes.test.mjs mindspace-image-review.test.mjs mindspace-run-public-html-scope.test.mjs direct-chat-service.test.mjs tool-gateway.test.mjs mindspace.test.mjs mindspace-scan.test.mjs mindspace-assets.test.mjs mindspace-local-runtime-services.test.mjs mindspace-local-server-adapter.test.mjs mindspace-public-asset-token.test.mjs mindspace-remote-server-adapter.test.mjs mindspace-server-adapter.test.mjs mindspace-pages.test.mjs mindspace-page-sync-service.test.mjs public-site-bases.test.mjs mindspace-html-download-links.test.mjs mindspace-page-purge.test.mjs mindspace-public-delivery.test.mjs mindspace-public-page-context.test.mjs mindspace-published-page-csp.test.mjs mindspace-published-script-localize.test.mjs agent-run-deliverable-check.test.mjs mindspace-public-route.test.mjs mindspace-publications.test.mjs mindspace-public-links.test.mjs mindspace-chat-save.test.mjs mindspace-chat-docx-package.test.mjs mindspace-public-finish-sync.test.mjs mindspace-chat-context.test.mjs mindspace-canonical-url.test.mjs mindspace-conversation-package.test.mjs mindspace-conversation-package-backfill.test.mjs mindspace-conversation-package-public-html.test.mjs mindspace-conversation-package-verify.test.mjs mindspace-conversation-package-registry.test.mjs mindspace-conversation-package-routes.test.mjs mindspace-conversation-package-store.test.mjs mindspace-conversation-schema.test.mjs mindspace-runtime-config.test.mjs mindspace-config.test.mjs mindspace-analytics.test.mjs mindspace-rybbit.test.mjs mindspace-service.test.mjs mindspace-storage-adapter.test.mjs mindspace-content-scan.test.mjs mindspace-html-localize.test.mjs mindspace-visual-editor.test.mjs mindspace-cleanup.test.mjs mindspace-thumbnails.test.mjs mindspace-workspace-thumbnails.test.mjs mindspace-workspace-sync.test.mjs mindspace-asset-preview.test.mjs mindspace-agent-jobs.test.mjs mindspace-agent-runner.test.mjs mindspace-sandbox-mcp.test.mjs mindspace-userdata-postgres.test.mjs postgres-user-data-space-service.test.mjs user-data-space-service.test.mjs page-data-routes.test.mjs page-access-policy.test.mjs page-access-visitor.test.mjs page-data-public-service.test.mjs page-data-integration.test.mjs page-data-log-store.test.mjs page-data-ops.test.mjs page-data-session-store.test.mjs page-data-browser-client.test.mjs page-data-policy-index.test.mjs message-stream.test.mjs mindspace-service/mindspace-rpc-server.test.mjs plaza-posts.test.mjs plaza-interactions.test.mjs plaza-algorithm.test.mjs plaza-seo.test.mjs plaza-ops.test.mjs user-auth.test.mjs llm-providers.test.mjs admin-guard.test.mjs user-feedback.test.mjs memory-v2.test.mjs memory-v2-admin-config.test.mjs memory-v2-lifecycle.test.mjs memory-v2-adapter-scaffold.test.mjs memory-v2-backend-contract.test.mjs memory-v2-health.test.mjs memory-v2-runtime.test.mjs memory-v2-plugin-backends.test.mjs memory-v2-pgvector.test.mjs memory-v2-pgvector-schema.test.mjs memory-v2-pgvector-backfill.test.mjs memory-v2-pgvector-smoke.test.mjs memory-v2-qdrant.test.mjs memory-v2-weaviate.test.mjs memory-v2-mem0.test.mjs memory-v2-letta.test.mjs memory-v2-external-adapters.test.mjs scripts/embed-memory-v2-local-hash.test.mjs scripts/check-memory-v2-app-canary.test.mjs scripts/check-memory-v2-config-gaps.test.mjs scripts/check-memory-v2-contracts.test.mjs scripts/check-memory-v2-health.test.mjs scripts/check-memory-v2-session-flow.test.mjs scripts/check-memory-v2-stack.test.mjs scripts/setup-memory-v2-pgvector-schema.test.mjs scripts/backfill-memory-v2-pgvector.test.mjs scripts/scaffold-memory-v2-backend.test.mjs scripts/smoke-memory-v2-pgvector.test.mjs scripts/smoke-memory-v2-qdrant.test.mjs scripts/smoke-memory-v2-external.test.mjs scripts/mock-memory-v2-services.test.mjs", + "test": "node --test auth.test.mjs asr-proxy.test.mjs billing.test.mjs billing-token-state.test.mjs billing-recharge.test.mjs wechat-pay.test.mjs wechat-oauth.test.mjs wechat-mp.test.mjs wechat-media.test.mjs wechat/image-generation-policy.test.mjs wechat/verify/generated-thumbnail.test.mjs schedule-intent.test.mjs schedule-reminder-worker.test.mjs capabilities.test.mjs policies.test.mjs server/portal-api-auth-middleware.test.mjs server/portal-config-routes.test.mjs server/portal-plaza-discovery-routes.test.mjs server/portal-runtime-routes.test.mjs server/portal-gateway-services-bootstrap.test.mjs chat-skills.test.mjs chat-intent-router.test.mjs chat-finish-sync.test.mjs chat-agent-run-gate.test.mjs conversation-display.test.mjs user-publish.test.mjs user-memory-profile.test.mjs skills-registry.test.mjs skill-runtime-policy.test.mjs excel-analyst.test.mjs agent-run-gateway.test.mjs agent-run-routes.test.mjs session-broker.test.mjs sse-event-taxonomy.test.mjs goosed-proxy-boundary.test.mjs agent-run-stream.test.mjs mindspace-h5-html-finish-guard.test.mjs admin-routes.test.mjs services/orchestrator/admin-config.test.mjs services/orchestrator/contracts.test.mjs services/orchestrator/checkpoint.test.mjs services/orchestrator/runtime.test.mjs services/orchestrator/app.test.mjs services/orchestrator/server.test.mjs services/orchestrator/shadow-dispatcher.test.mjs services/orchestrator/shadow-observer.test.mjs services/orchestrator/observability.test.mjs services/orchestrator/executor-gateway.test.mjs services/orchestrator/executor-job-store.test.mjs image-make-admin-config.test.mjs asset-gateway.test.mjs image-make-client.test.mjs mindspace-image-generation.test.mjs mindspace-image-generation-routes.test.mjs mindspace-image-review.test.mjs mindspace-run-public-html-scope.test.mjs direct-chat-service.test.mjs tool-gateway.test.mjs mindspace.test.mjs mindspace-scan.test.mjs mindspace-assets.test.mjs mindspace-local-runtime-services.test.mjs mindspace-local-server-adapter.test.mjs mindspace-public-asset-token.test.mjs mindspace-remote-server-adapter.test.mjs mindspace-server-adapter.test.mjs mindspace-pages.test.mjs mindspace-page-sync-service.test.mjs public-site-bases.test.mjs mindspace-html-download-links.test.mjs mindspace-page-purge.test.mjs mindspace-public-delivery.test.mjs mindspace-public-page-context.test.mjs mindspace-published-page-csp.test.mjs mindspace-published-script-localize.test.mjs agent-run-deliverable-check.test.mjs mindspace-public-route.test.mjs mindspace-publications.test.mjs mindspace-public-links.test.mjs mindspace-chat-save.test.mjs mindspace-chat-docx-package.test.mjs mindspace-public-finish-sync.test.mjs mindspace-chat-context.test.mjs mindspace-canonical-url.test.mjs mindspace-conversation-package.test.mjs mindspace-conversation-package-backfill.test.mjs mindspace-conversation-package-public-html.test.mjs mindspace-conversation-package-verify.test.mjs mindspace-conversation-package-registry.test.mjs mindspace-conversation-package-routes.test.mjs mindspace-conversation-package-store.test.mjs mindspace-conversation-schema.test.mjs mindspace-runtime-config.test.mjs mindspace-config.test.mjs mindspace-analytics.test.mjs mindspace-rybbit.test.mjs mindspace-service.test.mjs mindspace-storage-adapter.test.mjs mindspace-content-scan.test.mjs mindspace-html-localize.test.mjs mindspace-visual-editor.test.mjs mindspace-cleanup.test.mjs mindspace-thumbnails.test.mjs mindspace-workspace-thumbnails.test.mjs mindspace-workspace-sync.test.mjs mindspace-asset-preview.test.mjs mindspace-agent-jobs.test.mjs mindspace-agent-runner.test.mjs mindspace-sandbox-mcp.test.mjs mindspace-userdata-postgres.test.mjs postgres-user-data-space-service.test.mjs user-data-space-service.test.mjs page-data-routes.test.mjs page-access-policy.test.mjs page-access-visitor.test.mjs page-data-public-service.test.mjs page-data-integration.test.mjs page-data-log-store.test.mjs page-data-ops.test.mjs page-data-session-store.test.mjs page-data-browser-client.test.mjs page-data-policy-index.test.mjs message-stream.test.mjs mindspace-service/mindspace-rpc-server.test.mjs plaza-posts.test.mjs plaza-interactions.test.mjs plaza-algorithm.test.mjs plaza-seo.test.mjs plaza-ops.test.mjs user-auth.test.mjs llm-providers.test.mjs admin-guard.test.mjs user-feedback.test.mjs memory-v2.test.mjs memory-v2-admin-config.test.mjs memory-v2-lifecycle.test.mjs memory-v2-adapter-scaffold.test.mjs memory-v2-backend-contract.test.mjs memory-v2-health.test.mjs memory-v2-runtime.test.mjs memory-v2-plugin-backends.test.mjs memory-v2-pgvector.test.mjs memory-v2-pgvector-schema.test.mjs memory-v2-pgvector-backfill.test.mjs memory-v2-pgvector-smoke.test.mjs memory-v2-qdrant.test.mjs memory-v2-weaviate.test.mjs memory-v2-mem0.test.mjs memory-v2-letta.test.mjs memory-v2-external-adapters.test.mjs scripts/embed-memory-v2-local-hash.test.mjs scripts/check-memory-v2-app-canary.test.mjs scripts/check-memory-v2-config-gaps.test.mjs scripts/check-memory-v2-contracts.test.mjs scripts/check-memory-v2-health.test.mjs scripts/check-memory-v2-session-flow.test.mjs scripts/check-memory-v2-stack.test.mjs scripts/setup-memory-v2-pgvector-schema.test.mjs scripts/backfill-memory-v2-pgvector.test.mjs scripts/scaffold-memory-v2-backend.test.mjs scripts/smoke-memory-v2-pgvector.test.mjs scripts/smoke-memory-v2-qdrant.test.mjs scripts/smoke-memory-v2-external.test.mjs scripts/mock-memory-v2-services.test.mjs", "test:episodic-memory": "node --test episodic-memory.test.mjs direct-chat-service.test.mjs chat-intent-router.test.mjs", "test:deep-search": "node --test deep-search.test.mjs mindsearch.test.mjs", "test:image-review": "node --test mindspace-image-review.test.mjs mindspace-image-generation.test.mjs", diff --git a/scripts/smoke-orchestrator-shadow.mjs b/scripts/smoke-orchestrator-shadow.mjs index a318048..175b80e 100644 --- a/scripts/smoke-orchestrator-shadow.mjs +++ b/scripts/smoke-orchestrator-shadow.mjs @@ -42,6 +42,22 @@ async function waitForShadowEvent(pool, runId, timeoutMs = 15_000) { throw new Error(`Timed out waiting for Shadow projection for run ${runId}`); } +async function deleteOrchestratorRun(runId) { + const response = await fetch(`${serviceUrl}/v1/runs/${encodeURIComponent(runId)}`, { + method: 'DELETE', + headers: { + authorization: `Bearer ${serviceToken}`, + accept: 'application/json', + }, + }); + if (response.ok) return; + if (response.status === 404) { + const body = await response.json().catch(() => null); + if (body?.error?.code === 'RUN_NOT_FOUND') return; + } + throw new Error(`Orchestrator run cleanup failed with HTTP ${response.status}`); +} + async function main() { if (!isDatabaseConfigured()) { throw new Error('Shadow smoke requires the local Memind MySQL configuration'); @@ -132,13 +148,23 @@ async function main() { restoreConfig, }, null, 2)); } finally { - if (cleanup && runId) { - await pool.query('DELETE FROM h5_agent_runs WHERE id = ?', [runId]); + let cleanupError = null; + try { + if (cleanup && runId) { + await deleteOrchestratorRun(runId); + await pool.query('DELETE FROM h5_agent_runs WHERE id = ?', [runId]); + } + } catch (error) { + cleanupError = error; } - if (restoreConfig) { - await configService.updateAdminConfig(previous.config); + try { + if (restoreConfig) { + await configService.updateAdminConfig(previous.config); + } + } finally { + await pool.end(); } - await pool.end(); + if (cleanupError) throw cleanupError; } } diff --git a/server/portal-gateway-services-bootstrap.mjs b/server/portal-gateway-services-bootstrap.mjs index 5f2b76e..04ab9a4 100644 --- a/server/portal-gateway-services-bootstrap.mjs +++ b/server/portal-gateway-services-bootstrap.mjs @@ -285,12 +285,18 @@ export function bootstrapPortalGatewayServices({ }); const validateRunDeliverables = createRunDeliverablesValidatorFn({ h5Root }); - const workflowShadowObserver = - createWorkflowShadowObserverFn({ - configService: - createOrchestratorAdminConfigServiceFn(pool), - logger: console, - }); + const workflowShadowEnabled = isEnabledFlag( + env.MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED, + ); + const workflowShadowObserver = workflowShadowEnabled + ? createWorkflowShadowObserverFn({ + configService: + createOrchestratorAdminConfigServiceFn(pool), + serviceToken: + env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN, + logger: console, + }) + : null; let runtimeRoot; try { runtimeRoot = fs.realpathSync(h5Root); @@ -389,6 +395,14 @@ export function bootstrapPortalGatewayServices({ env.MEMIND_AGENT_RUN_TIMEOUT_MS ?? 15 * 60 * 1000, ), + maxConcurrentShadowObservations: Number( + env.MEMIND_ORCHESTRATOR_SHADOW_MAX_CONCURRENCY ?? + 2, + ), + maxQueuedShadowObservations: Number( + env.MEMIND_ORCHESTRATOR_SHADOW_MAX_QUEUE ?? + 100, + ), workerIdentity, }); const agentRunRecoveryTimer = diff --git a/server/portal-gateway-services-bootstrap.test.mjs b/server/portal-gateway-services-bootstrap.test.mjs index f69752c..25e401d 100644 --- a/server/portal-gateway-services-bootstrap.test.mjs +++ b/server/portal-gateway-services-bootstrap.test.mjs @@ -166,6 +166,39 @@ test('preserves Proxy, Tool, and Agent gateway wiring', () => { assert.equal(captured.agentOptions.autoDispatch, true); assert.equal(captured.agentOptions.maxConcurrentRuns, 3); assert.equal(captured.agentOptions.runTimeoutMs, 9000); + assert.equal(captured.agentOptions.observeWorkflowRun, null); + assert.equal(captured.agentOptions.maxConcurrentShadowObservations, 2); + assert.equal(captured.agentOptions.maxQueuedShadowObservations, 100); +}); + +test('wires Shadow observation only behind the explicit environment gate', () => { + const configService = { id: 'orchestrator-config' }; + const observer = async () => ({ observed: false }); + let receivedObserverOptions = null; + const setup = createSetup({ + env: { + MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED: '1', + MEMIND_ORCHESTRATOR_SERVICE_TOKEN: 'shadow-token', + MEMIND_ORCHESTRATOR_SHADOW_MAX_CONCURRENCY: '4', + MEMIND_ORCHESTRATOR_SHADOW_MAX_QUEUE: '25', + }, + createOrchestratorAdminConfigServiceFn(receivedPool) { + assert.equal(receivedPool, setup.pool); + return configService; + }, + createWorkflowShadowObserverFn(options) { + receivedObserverOptions = options; + return observer; + }, + }); + + bootstrapPortalGatewayServices(setup.options); + const { agentOptions } = setup.getCaptured(); + assert.equal(receivedObserverOptions.configService, configService); + assert.equal(receivedObserverOptions.serviceToken, 'shadow-token'); + assert.equal(agentOptions.observeWorkflowRun, observer); + assert.equal(agentOptions.maxConcurrentShadowObservations, 4); + assert.equal(agentOptions.maxQueuedShadowObservations, 25); }); test('preserves local asset reads and optional asset absence', async () => { diff --git a/services/orchestrator/README.md b/services/orchestrator/README.md index 861a637..cb3ac5b 100644 --- a/services/orchestrator/README.md +++ b/services/orchestrator/README.md @@ -197,7 +197,15 @@ MEMIND_ORCHESTRATOR_CHECKPOINT_MODE=memory pnpm dev:orchestrator ``` The service binds to `127.0.0.1:8093` by default. Configure the same URL and -service token in Portal, then enable `shadow` in `/ops/admin/orchestrator`. +service token in Portal, set +`MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED=1` on that Portal instance, +then enable `shadow` in `/ops/admin/orchestrator`. Without the Portal gate, +Native code runs do not construct or call the observer. + +The admin runtime projection distinguishes requested Shadow mode from effective +Portal wiring. Canary readiness requires `shadow_wiring_enabled=true`; a closed +environment gate therefore blocks promotion even when older successful samples +are still present. memindadm exposes the projection through: @@ -223,8 +231,14 @@ POST /v1/workers/jobs/:jobId/complete POST /v1/workers/jobs/:jobId/fail POST /v1/workers/recover-expired GET /v1/workers/stats +DELETE /v1/runs/:runId +POST /v1/maintenance/purge-terminal-runs ``` +Shadow RunSpecs use the `control-plane-only-v1` policy and do not copy user +messages, user IDs, or session IDs into Orchestrator storage. Deleting a +terminal run removes its checkpoint thread and linked terminal Executor Job. + ## Colima deployment Colima is the recommended first container host on macOS because this service and @@ -278,6 +292,23 @@ requires the repository release gates and a separately approved deployment. | `MEMIND_ORCHESTRATOR_CHECKPOINT_MODE` | `postgres` or explicit `memory` | `postgres` | | `MEMIND_ORCHESTRATOR_DATABASE_URL` | Dedicated Orchestrator PostgreSQL URL | required | | `MEMIND_ORCHESTRATOR_DATABASE_SCHEMA` | Checkpoint and Executor Job schema | `memind_orchestrator` | +| `MEMIND_ORCHESTRATOR_RETENTION_DAYS` | Delete terminal workflow data older than this age; `0` disables | `0` | +| `MEMIND_ORCHESTRATOR_RETENTION_SWEEP_INTERVAL_MS` | Retention sweep interval, minimum one hour | `86400000` | +| `MEMIND_ORCHESTRATOR_RETENTION_SWEEP_LIMIT` | Maximum terminal candidates per sweep | `100` | + +Portal-side variables: + +| Variable | Purpose | Default | +|---|---|---| +| `MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED` | Construct and schedule the Shadow observer | `false` | +| `MEMIND_ORCHESTRATOR_SHADOW_MAX_CONCURRENCY` | Maximum simultaneous observations | `2` | +| `MEMIND_ORCHESTRATOR_SHADOW_MAX_QUEUE` | Maximum waiting observations before skip | `100` | +| `MEMIND_ORCHESTRATOR_ALLOWED_ORIGINS` | Additional exact non-loopback Orchestrator origins | empty | + +An Orchestrator bound beyond loopback refuses to start without the service +token. Execution refuses to start without both service and worker tokens. +Retention is disabled by default; the maintenance endpoint is dry-run unless +the authenticated caller sends `apply=true`. ## Extraction test diff --git a/services/orchestrator/admin-config.mjs b/services/orchestrator/admin-config.mjs index dedff21..fa764b1 100644 --- a/services/orchestrator/admin-config.mjs +++ b/services/orchestrator/admin-config.mjs @@ -12,6 +12,7 @@ import { listPhase5ExecutorAdapters } from './executor-gateway.mjs'; const CONFIG_TABLE = 'h5_orchestrator_admin_config'; const CONFIG_SCOPE = 'global'; const EXECUTION_HANDOFF_IMPLEMENTED = true; +const CONFIG_SCHEMA_PROMISES = new WeakMap(); function normalizeBoolean(value, fallback = false) { if (value == null || value === '') return fallback; @@ -39,15 +40,45 @@ function normalizeServiceUrl(value) { try { const url = new URL(raw); if (!['http:', 'https:'].includes(url.protocol)) return ''; + if ((url.pathname && url.pathname !== '/') || url.search) return ''; url.username = ''; url.password = ''; url.hash = ''; - return url.toString().replace(/\/$/, '').slice(0, 2048); + return url.origin.slice(0, 2048); } catch { return ''; } } +function isLoopbackServiceUrl(value) { + try { + const hostname = new URL(value).hostname.toLowerCase(); + return hostname === 'localhost' + || hostname === '::1' + || hostname === '[::1]' + || /^127(?:\.\d{1,3}){3}$/.test(hostname); + } catch { + return false; + } +} + +function allowedServiceOrigins(env = process.env) { + const configured = [ + env.MEMIND_ORCHESTRATOR_URL, + ...String(env.MEMIND_ORCHESTRATOR_ALLOWED_ORIGINS ?? '').split(','), + ] + .map(normalizeServiceUrl) + .filter(Boolean); + return new Set(configured); +} + +function isServiceUrlAllowed(value, env = process.env) { + const normalized = normalizeServiceUrl(value); + if (!normalized) return false; + return isLoopbackServiceUrl(normalized) + || allowedServiceOrigins(env).has(normalized); +} + function parseJsonLike(value, fallback) { if (value == null || value === '') return fallback; if (typeof value === 'object') return value; @@ -112,6 +143,10 @@ export function normalizeOrchestratorConfig(input = {}, fallback = defaultOrches function runtimeState(config, env = process.env) { const killSwitch = normalizeBoolean(env.MEMIND_ORCHESTRATOR_KILL_SWITCH, false); + const environmentShadowObservationGate = normalizeBoolean( + env.MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED, + false, + ); const environmentExecutionGate = normalizeBoolean( env.MEMIND_ORCHESTRATOR_EXECUTION_HANDOFF_ENABLED, false, @@ -121,6 +156,10 @@ function runtimeState(config, env = process.env) { if (killSwitch) reason = 'kill_switch'; else if (config.mode === ORCHESTRATOR_MODE.OFF) reason = 'mode_off'; else if (!configured) reason = 'service_url_missing'; + else if ( + config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH + && !isServiceUrlAllowed(config.serviceUrl, env) + ) reason = 'service_url_not_allowed'; const executionModeSelected = [ORCHESTRATOR_MODE.CANARY, ORCHESTRATOR_MODE.ACTIVE] .includes(config.mode); const executionHandoffRequested = config.executionEnabled @@ -131,6 +170,11 @@ function runtimeState(config, env = process.env) { && environmentExecutionGate && executionHandoffRequested && config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH; + const shadowObservationRequested = config.mode === ORCHESTRATOR_MODE.SHADOW + && config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH; + const shadowObservationEnabled = !reason + && environmentShadowObservationGate + && shadowObservationRequested; return { killSwitch, configured, @@ -146,9 +190,19 @@ function runtimeState(config, env = process.env) { plansLangGraph: !reason && executionModeSelected && config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH, - shadowsLangGraph: !reason - && config.mode === ORCHESTRATOR_MODE.SHADOW - && config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH, + shadowsLangGraph: shadowObservationEnabled, + shadowObservation: { + requested: shadowObservationRequested, + enabled: shadowObservationEnabled, + reason: shadowObservationEnabled + ? null + : !shadowObservationRequested + ? 'shadow_observation_not_requested' + : reason + ? reason + : 'environment_shadow_observation_gate_disabled', + environmentGate: environmentShadowObservationGate, + }, executionHandoff: { implemented: EXECUTION_HANDOFF_IMPLEMENTED, requested: executionHandoffRequested, @@ -261,7 +315,9 @@ async function probeServiceHealth(config, { } async function ensureConfigTable(pool) { - await pool.query(` + const existing = CONFIG_SCHEMA_PROMISES.get(pool); + if (existing) return existing; + const initializing = pool.query(` CREATE TABLE IF NOT EXISTS ${CONFIG_TABLE} ( config_scope VARCHAR(32) PRIMARY KEY, config_json JSON NOT NULL, @@ -269,7 +325,12 @@ async function ensureConfigTable(pool) { updated_by CHAR(36) NULL, updated_at BIGINT NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci - `); + `).catch((error) => { + CONFIG_SCHEMA_PROMISES.delete(pool); + throw error; + }); + CONFIG_SCHEMA_PROMISES.set(pool, initializing); + return initializing; } async function loadStoredState(pool) { @@ -450,6 +511,8 @@ export const orchestratorAdminConfigInternals = { CONFIG_TABLE, engineCatalog, EXECUTION_HANDOFF_IMPLEMENTED, + isServiceUrlAllowed, + normalizeServiceUrl, probeServiceHealth, runtimeState, }; diff --git a/services/orchestrator/admin-config.test.mjs b/services/orchestrator/admin-config.test.mjs index 02a2722..5081120 100644 --- a/services/orchestrator/admin-config.test.mjs +++ b/services/orchestrator/admin-config.test.mjs @@ -8,9 +8,16 @@ import { function createPool() { let row = null; + let createTableCalls = 0; return { + get createTableCalls() { + return createTableCalls; + }, async query(sql, params = []) { - if (sql.includes('CREATE TABLE')) return [[], []]; + if (sql.includes('CREATE TABLE')) { + createTableCalls += 1; + return [[], []]; + } if (sql.includes('SELECT config_json')) return [[...(row ? [row] : [])], []]; if (sql.includes('INSERT INTO')) { row = { @@ -33,6 +40,12 @@ test('orchestrator config defaults to a disabled native-safe runtime', async () assert.equal(state.config.primaryEngine, 'langgraph'); assert.equal(state.runtime.effective, false); assert.equal(state.runtime.reason, 'mode_off'); + assert.deepEqual(state.runtime.shadowObservation, { + requested: false, + enabled: false, + reason: 'shadow_observation_not_requested', + environmentGate: false, + }); assert.equal(state.engines.find((engine) => engine.id === 'native').configured, true); assert.equal(state.engines.find((engine) => engine.id === 'langgraph').configured, false); assert.deepEqual( @@ -59,6 +72,48 @@ test('orchestrator config normalizes unsafe values and keeps a workflow allowlis assert.equal(normalized.requestTimeoutMs, 500); assert.equal(normalized.rolloutPercent, 100); assert.deepEqual(normalized.workflowAllowlist, ['code-run-v1']); + assert.equal( + normalizeOrchestratorConfig({ + serviceUrl: 'https://orchestrator.example/internal?token=secret', + }).serviceUrl, + '', + ); +}); + +test('orchestrator config initializes its schema once per shared pool', async () => { + const pool = createPool(); + const first = createOrchestratorAdminConfigService(pool, { env: {} }); + const second = createOrchestratorAdminConfigService(pool, { env: {} }); + await first.ensureSchema(); + await first.getRuntimeState(); + await second.getAdminConfig(); + assert.equal(pool.createTableCalls, 1); +}); + +test('orchestrator blocks non-loopback service origins unless explicitly allowed', async () => { + const pool = createPool(); + const blockedService = createOrchestratorAdminConfigService(pool, { env: {} }); + const blocked = await blockedService.updateAdminConfig({ + mode: 'shadow', + serviceUrl: 'https://orchestrator.example', + }); + assert.equal(blocked.runtime.effective, false); + assert.equal(blocked.runtime.reason, 'service_url_not_allowed'); + assert.deepEqual(blocked.runtime.shadowObservation, { + requested: true, + enabled: false, + reason: 'service_url_not_allowed', + environmentGate: false, + }); + + const allowedService = createOrchestratorAdminConfigService(pool, { + env: { + MEMIND_ORCHESTRATOR_ALLOWED_ORIGINS: 'https://orchestrator.example', + }, + }); + const allowed = await allowedService.getRuntimeState(); + assert.equal(allowed.runtime.effective, true); + assert.equal(allowed.runtime.reason, null); }); test('orchestrator config persists versioned admin updates and selects canary users', async () => { @@ -194,6 +249,32 @@ test('Shadow mode evaluates the Canary rollout while Native remains effective', assert.equal(native.candidateEngine, 'native'); assert.equal(native.candidateReason, 'canary_not_selected'); assert.equal(native.dryRun, true); + const runtime = await service.getRuntimeState(); + assert.equal(runtime.runtime.shadowsLangGraph, false); + assert.deepEqual(runtime.runtime.shadowObservation, { + requested: true, + enabled: false, + reason: 'environment_shadow_observation_gate_disabled', + environmentGate: false, + }); +}); + +test('Shadow runtime reports the independent Portal observation wiring gate', async () => { + const service = createOrchestratorAdminConfigService(createPool(), { + env: { MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED: '1' }, + }); + await service.updateAdminConfig({ + mode: 'shadow', + serviceUrl: 'http://127.0.0.1:8093', + }); + const runtime = await service.getRuntimeState(); + assert.equal(runtime.runtime.shadowsLangGraph, true); + assert.deepEqual(runtime.runtime.shadowObservation, { + requested: true, + enabled: true, + reason: null, + environmentGate: true, + }); }); test('orchestrator emergency kill switch always forces native selection', async () => { diff --git a/services/orchestrator/app.mjs b/services/orchestrator/app.mjs index 14319b9..971bf5b 100644 --- a/services/orchestrator/app.mjs +++ b/services/orchestrator/app.mjs @@ -154,6 +154,28 @@ export function createOrchestratorApp({ } }); + app.delete('/v1/runs/:runId', async (request, response, next) => { + try { + const result = await runtime.deleteRun(request.params.runId); + if (!result) return response.status(404).json({ error: { code: 'RUN_NOT_FOUND' } }); + return response.json(result); + } catch (error) { + return next(error); + } + }); + + app.post('/v1/maintenance/purge-terminal-runs', async (request, response, next) => { + try { + return response.json(await runtime.purgeTerminalRuns({ + before: request.body?.before, + limit: request.body?.limit, + dryRun: request.body?.apply !== true, + })); + } catch (error) { + return next(error); + } + }); + app.get('/v1/executor-jobs/:jobId', async (request, response, next) => { try { const job = await runtime.getExecutorJob(request.params.jobId); diff --git a/services/orchestrator/app.test.mjs b/services/orchestrator/app.test.mjs index d8f422f..8d54ff7 100644 --- a/services/orchestrator/app.test.mjs +++ b/services/orchestrator/app.test.mjs @@ -102,6 +102,39 @@ test('orchestrator HTTP API exposes health and authenticated run endpoints', asy ['executor_job_blocked'], ); assert.equal(executorEventBody.nextCursor, 1); + + const purgePreview = await fetch( + `${server.baseUrl}/v1/maintenance/purge-terminal-runs`, + { + method: 'POST', + headers: { + authorization: 'Bearer test-service-token', + 'content-type': 'application/json', + }, + body: JSON.stringify({ before: Date.now() + 1000 }), + }, + ); + assert.equal(purgePreview.status, 200); + const purgePreviewBody = await purgePreview.json(); + assert.equal(purgePreviewBody.dryRun, true); + assert.equal(purgePreviewBody.candidates.length, 1); + + const deleted = await fetch(`${server.baseUrl}/v1/runs/http-shadow-1`, { + method: 'DELETE', + headers: { authorization: 'Bearer test-service-token' }, + }); + assert.equal(deleted.status, 200); + assert.deepEqual(await deleted.json(), { + runId: 'http-shadow-1', + deleted: true, + executorJobDeleted: true, + }); + assert.equal( + (await fetch(`${server.baseUrl}/v1/runs/http-shadow-1`, { + headers: { authorization: 'Bearer test-service-token' }, + })).status, + 404, + ); }); test('orchestrator HTTP API reports missing runs and observe-only violations', async (t) => { diff --git a/services/orchestrator/code-run-shadow-graph.mjs b/services/orchestrator/code-run-shadow-graph.mjs index 19f0beb..4181f78 100644 --- a/services/orchestrator/code-run-shadow-graph.mjs +++ b/services/orchestrator/code-run-shadow-graph.mjs @@ -127,6 +127,7 @@ async function planNode(state, executorGateway) { metadata: { source: 'langgraph-shadow-plan', workflow: state.spec.workflow, + workflowRunId: state.spec.runId, }, }); const plan = { diff --git a/services/orchestrator/executor-gateway.mjs b/services/orchestrator/executor-gateway.mjs index 8a48009..2f51f59 100644 --- a/services/orchestrator/executor-gateway.mjs +++ b/services/orchestrator/executor-gateway.mjs @@ -79,6 +79,11 @@ function gatewayError(code, message, status = 409) { return error; } +function normalizeWorkflowRunId(value) { + const runId = String(value ?? '').trim(); + return /^[a-zA-Z0-9_-]{1,128}$/.test(runId) ? runId : null; +} + export function buildExecutorJobEvent({ eventId = crypto.randomUUID(), jobId, @@ -366,6 +371,36 @@ export function createInMemoryExecutorJobStore() { const value = jobId ? byId.get(jobId) : null; return value ? clone(value) : null; }, + async deleteById(jobId) { + const normalizedJobId = String(jobId ?? '').trim(); + const existing = byId.get(normalizedJobId); + if (!existing) return false; + byId.delete(normalizedJobId); + byIdempotencyKey.delete(existing.idempotencyKey); + eventsByJobId.delete(normalizedJobId); + return true; + }, + async listTerminalBefore({ before, limit = 100 } = {}) { + const cutoff = Number(before); + if (!Number.isFinite(cutoff)) return []; + return [...byId.values()] + .filter((record) => ( + TERMINAL_JOB_STATUSES.has(record.status) + && record.workflowRunId + && Number(record.completedAt ?? 0) > 0 + && Number(record.completedAt) <= cutoff + )) + .sort((left, right) => ( + Number(left.completedAt) - Number(right.completedAt) + || String(left.jobId).localeCompare(String(right.jobId)) + )) + .slice(0, clampInteger(limit, 100, 1, 500)) + .map((record) => ({ + jobId: record.jobId, + workflowRunId: record.workflowRunId, + completedAt: Number(record.completedAt), + })); + }, async createIfAbsent(record, { initialEvent = null } = {}) { const existingId = byIdempotencyKey.get(record.idempotencyKey); if (existingId) return { created: false, record: clone(byId.get(existingId)) }; @@ -553,6 +588,8 @@ function validateJobStore(store) { 'getById', 'getByIdempotencyKey', 'createIfAbsent', + 'deleteById', + 'listTerminalBefore', 'update', 'listEvents', ]) { @@ -674,6 +711,7 @@ export function createExecutorGateway({ fallbackRegistered: decision.fallbackRegistered, fallbackAvailable: decision.fallbackAvailable, decision, + workflowRunId: normalizeWorkflowRunId(request.metadata?.workflowRunId), request: dispatchAllowed ? request : null, createdAt: now, updatedAt: now, @@ -741,10 +779,28 @@ export function createExecutorGateway({ return store.update(updated, { event: eventForCancellation(updated) }); } + async function deleteJob(jobId) { + const current = await store.getById(jobId); + if (!current) return false; + if (!TERMINAL_JOB_STATUSES.has(current.status)) { + throw gatewayError( + 'EXECUTOR_JOB_NOT_TERMINAL', + `Executor job ${current.jobId} must be terminal before deletion`, + ); + } + return store.deleteById(current.jobId); + } + + async function listRetentionCandidates(options = {}) { + return store.listTerminalBefore(options); + } + return { preview, createJob, cancel, + deleteJob, + listRetentionCandidates, async getJob(jobId, { includeRequest = false } = {}) { const record = await store.getById(jobId); return includeRequest ? record : projectExecutorJobForRead(record); @@ -793,6 +849,7 @@ export const executorGatewayInternals = { PHASE3_EXECUTOR_DESCRIPTORS, TERMINAL_JOB_STATUSES, fingerprint, + normalizeWorkflowRunId, projectExecutorJobForRead, stableValue, }; diff --git a/services/orchestrator/executor-gateway.test.mjs b/services/orchestrator/executor-gateway.test.mjs index 8f6af2f..beb4287 100644 --- a/services/orchestrator/executor-gateway.test.mjs +++ b/services/orchestrator/executor-gateway.test.mjs @@ -162,6 +162,53 @@ test('Executor Gateway cancellation is idempotent and never invokes a disabled a assert.equal(await gateway.listEvents('missing-job'), null); }); +test('Executor Gateway deletes only terminal jobs and their private request state', async () => { + const gateway = createExecutorGateway(); + await gateway.createJob(request()); + assert.equal(await gateway.deleteJob('job-1'), true); + assert.equal(await gateway.getJob('job-1'), null); + assert.equal(await gateway.listEvents('job-1'), null); + assert.equal(await gateway.deleteJob('job-1'), false); + + const activeGateway = createExecutorGateway({ + registry: createPhase5ExecutorAdapterRegistry({ enabledExecutors: ['aider'] }), + executionEnabled: true, + }); + await activeGateway.createJob(request()); + await assert.rejects( + () => activeGateway.deleteJob('job-1'), + (error) => error.code === 'EXECUTOR_JOB_NOT_TERMINAL' && error.status === 409, + ); +}); + +test('Executor Gateway exposes only terminal workflow-linked retention candidates', async () => { + let now = 1000; + const gateway = createExecutorGateway({ nowMs: () => now }); + await gateway.createJob(request({ + metadata: { workflowRunId: 'run-retention-1' }, + })); + assert.deepEqual( + await gateway.listRetentionCandidates({ before: 999 }), + [], + ); + assert.deepEqual( + await gateway.listRetentionCandidates({ before: 1000 }), + [{ + jobId: 'job-1', + workflowRunId: 'run-retention-1', + completedAt: 1000, + }], + ); + + now = 2000; + await gateway.createJob(request({ + jobId: 'job-without-link', + idempotencyKey: 'without-link', + metadata: {}, + })); + assert.equal((await gateway.listRetentionCandidates({ before: 3000 })).length, 1); +}); + test('Executor Gateway reports implemented capability while execution remains disabled', () => { const status = createExecutorGateway().status(); assert.equal(status.version, 'executor-gateway-status-v1'); diff --git a/services/orchestrator/executor-job-store.mjs b/services/orchestrator/executor-job-store.mjs index 21eb85d..e174e4d 100644 --- a/services/orchestrator/executor-job-store.mjs +++ b/services/orchestrator/executor-job-store.mjs @@ -192,6 +192,41 @@ export function createPostgresExecutorJobStore({ return projectRow(result.rows?.[0]); }, + async deleteById(jobId) { + const result = await pool.query( + `DELETE FROM ${table} + WHERE job_id = $1 + RETURNING job_id`, + [String(jobId ?? '').trim()], + ); + return Boolean(result.rows?.[0]); + }, + + async listTerminalBefore({ before, limit = 100 } = {}) { + const cutoff = Number(before); + if (!Number.isFinite(cutoff)) return []; + const pageSize = Math.min(500, Math.max(1, Number(limit) || 100)); + const result = await pool.query( + `SELECT + state_json->>'jobId' AS job_id, + state_json->>'workflowRunId' AS workflow_run_id, + (state_json->>'completedAt')::BIGINT AS completed_at + FROM ${table} + WHERE state_json->>'status' = ANY($1::text[]) + AND state_json->>'workflowRunId' IS NOT NULL + AND COALESCE((state_json->>'completedAt')::BIGINT, 0) > 0 + AND (state_json->>'completedAt')::BIGINT <= $2 + ORDER BY (state_json->>'completedAt')::BIGINT ASC, job_id ASC + LIMIT $3`, + [['succeeded', 'failed', 'cancelled', 'timed_out', 'blocked'], cutoff, pageSize], + ); + return (result.rows ?? []).map((row) => ({ + jobId: row.job_id, + workflowRunId: row.workflow_run_id, + completedAt: Number(row.completed_at), + })); + }, + async createIfAbsent(record, { initialEvent = null } = {}) { const created = await withTransaction(pool, async (client) => { const inserted = await client.query( diff --git a/services/orchestrator/executor-job-store.test.mjs b/services/orchestrator/executor-job-store.test.mjs index ae9c890..77c308c 100644 --- a/services/orchestrator/executor-job-store.test.mjs +++ b/services/orchestrator/executor-job-store.test.mjs @@ -137,6 +137,51 @@ test('PostgreSQL Executor Job Store returns the existing idempotent record after ); }); +test('PostgreSQL Executor Job Store deletes a job and cascades its events', async () => { + const calls = []; + const store = createPostgresExecutorJobStore({ + pool: { + async query(sql, params) { + calls.push({ sql, params }); + return sql.includes('DELETE FROM') + ? { rows: [{ job_id: 'job-1' }] } + : { rows: [] }; + }, + }, + }); + assert.equal(await store.deleteById('job-1'), true); + assert.match(calls[0].sql, /DELETE FROM .*executor_jobs/); + assert.deepEqual(calls[0].params, ['job-1']); +}); + +test('PostgreSQL Executor Job Store lists bounded terminal retention candidates', async () => { + const calls = []; + const store = createPostgresExecutorJobStore({ + pool: { + async query(sql, params) { + calls.push({ sql, params }); + return { + rows: [{ + job_id: 'job-1', + workflow_run_id: 'run-1', + completed_at: '1000', + }], + }; + }, + }, + }); + assert.deepEqual( + await store.listTerminalBefore({ before: 2000, limit: 10 }), + [{ jobId: 'job-1', workflowRunId: 'run-1', completedAt: 1000 }], + ); + assert.match(calls[0].sql, /workflowRunId/); + assert.deepEqual(calls[0].params, [ + ['succeeded', 'failed', 'cancelled', 'timed_out', 'blocked'], + 2000, + 10, + ]); +}); + test('PostgreSQL Executor Job Store rolls back job creation when its initial event fails', async () => { const calls = []; const eventError = new Error('event insert unavailable'); diff --git a/services/orchestrator/observability.mjs b/services/orchestrator/observability.mjs index 5ea5477..659c4c2 100644 --- a/services/orchestrator/observability.mjs +++ b/services/orchestrator/observability.mjs @@ -4,6 +4,7 @@ import { WORKFLOW_ENGINE } from './contracts.mjs'; const SHADOW_EVENT_TYPES = Object.freeze([ 'workflow_shadow_completed', 'workflow_shadow_failed', + 'workflow_shadow_skipped', ]); const EXECUTION_PLAN_EVENT_TYPE = 'workflow_execution_planned'; const MAX_METRIC_EVENTS = 5000; @@ -11,6 +12,7 @@ const CANARY_READINESS_THRESHOLDS = Object.freeze({ hours: 24 * 7, minObservations: 20, minSuccessRate: 0.95, + maxSkipRate: 0, maxP95LatencyMs: 2000, minLatencyCoverageRate: 0.95, minNativeSettledRate: 0.95, @@ -56,6 +58,7 @@ function normalizeExecutorJobId(value) { function projectShadowEventRow(row) { const data = parseJsonColumn(row.data_json, {}) ?? {}; const succeeded = row.event_type === 'workflow_shadow_completed'; + const skipped = row.event_type === 'workflow_shadow_skipped'; const latency = Number(data.latencyMs); const taskType = data.taskType ?? null; return { @@ -66,7 +69,7 @@ function projectShadowEventRow(row) { sessionId: row.agent_session_id ?? null, nativeStatus: row.native_status, nativeAttempts: Number(row.native_attempts ?? 0), - shadowStatus: succeeded ? 'succeeded' : 'failed', + shadowStatus: succeeded ? 'succeeded' : skipped ? 'skipped' : 'failed', engine: data.engine ?? 'langgraph', configVersion: data.configVersion == null ? null : Number(data.configVersion), phase: data.phase ?? null, @@ -74,10 +77,11 @@ function projectShadowEventRow(row) { synthetic: data.synthetic === true || SYNTHETIC_TASK_TYPES.has(String(taskType ?? '')), executorAdapter: data.executorAdapter ?? null, latencyMs: Number.isFinite(latency) && latency >= 0 ? latency : null, - error: succeeded ? null : { + error: succeeded || skipped ? null : { code: data.code ?? 'WORKFLOW_SHADOW_FAILED', message: data.message ?? 'Shadow observation failed', }, + skipReason: skipped ? data.reason ?? 'shadow_queue_full' : null, observedAt: Number(row.event_created_at ?? 0), nativeCompletedAt: row.native_completed_at == null ? null @@ -146,6 +150,8 @@ function buildCanaryReadiness(loaded, runtimeState, now) { const thresholds = CANARY_READINESS_THRESHOLDS; const eligibleRuns = loaded.runs.filter((run) => !run.synthetic); const successes = eligibleRuns.filter((run) => run.shadowStatus === 'succeeded').length; + const failures = eligibleRuns.filter((run) => run.shadowStatus === 'failed').length; + const skipped = eligibleRuns.filter((run) => run.shadowStatus === 'skipped').length; const latencies = eligibleRuns .map((run) => run.latencyMs) .filter((value) => Number.isFinite(value)); @@ -159,6 +165,7 @@ function buildCanaryReadiness(loaded, runtimeState, now) { ? null : Math.max(0, (now - lastObservedAt) / (60 * 60 * 1000)); const successRate = ratio(successes, eligibleRuns.length); + const skipRate = ratio(skipped, eligibleRuns.length); const latencyCoverageRate = ratio(latencies.length, eligibleRuns.length); const nativeSettledRate = ratio(nativeSettled, eligibleRuns.length); const latencyP95Ms = percentile(latencies, 0.95); @@ -173,6 +180,12 @@ function buildCanaryReadiness(loaded, runtimeState, now) { actual: runtimeState?.config?.mode ?? null, target: 'shadow', }, + { + id: 'shadow_wiring_enabled', + passed: runtimeState?.runtime?.shadowObservation?.enabled === true, + actual: runtimeState?.runtime?.shadowObservation?.enabled ?? null, + target: true, + }, { id: 'service_healthy', passed: serviceHealth?.ok === true, @@ -209,6 +222,12 @@ function buildCanaryReadiness(loaded, runtimeState, now) { actual: successRate, target: thresholds.minSuccessRate, }, + { + id: 'shadow_skip_rate', + passed: skipRate != null && skipRate <= thresholds.maxSkipRate, + actual: skipRate, + target: thresholds.maxSkipRate, + }, { id: 'latency_coverage', passed: latencyCoverageRate != null @@ -271,8 +290,10 @@ function buildCanaryReadiness(loaded, runtimeState, now) { eligibleObservations: eligibleRuns.length, excludedSynthetic: loaded.runs.length - eligibleRuns.length, successes, - failures: eligibleRuns.length - successes, + failures, + skipped, successRate, + skipRate, latencyCoverageRate, latencyP95Ms, nativeSettledRate, @@ -301,7 +322,8 @@ function buildCanaryReadiness(loaded, runtimeState, now) { function summarizeRuns(runs, { capped = false } = {}) { const successes = runs.filter((run) => run.shadowStatus === 'succeeded').length; - const failures = runs.length - successes; + const failures = runs.filter((run) => run.shadowStatus === 'failed').length; + const skipped = runs.filter((run) => run.shadowStatus === 'skipped').length; const latencies = runs .map((run) => run.latencyMs) .filter((value) => Number.isFinite(value)); @@ -309,8 +331,10 @@ function summarizeRuns(runs, { capped = false } = {}) { observations: runs.length, successes, failures, + skipped, successRate: runs.length ? successes / runs.length : null, failureRate: runs.length ? failures / runs.length : null, + skipRate: runs.length ? skipped / runs.length : null, latencyP50Ms: percentile(latencies, 0.5), latencyP95Ms: percentile(latencies, 0.95), nativeSucceeded: runs.filter((run) => run.nativeStatus === 'succeeded').length, @@ -391,7 +415,7 @@ export function createOrchestratorObservabilityService({ r.completed_at AS native_completed_at FROM h5_agent_run_events e INNER JOIN h5_agent_runs r ON r.id = e.run_id - WHERE e.event_type IN (?, ?) + WHERE e.event_type IN (${SHADOW_EVENT_TYPES.map(() => '?').join(', ')}) AND e.created_at >= ? ORDER BY e.created_at DESC, e.id DESC LIMIT ?`, @@ -452,7 +476,9 @@ export function createOrchestratorObservabilityService({ } = {}) { const loaded = await loadShadowEvents({ hours }); const normalizedLimit = clampInteger(limit, 50, 1, 200); - const normalizedStatus = ['succeeded', 'failed'].includes(status) ? status : 'all'; + const normalizedStatus = ['succeeded', 'failed', 'skipped'].includes(status) + ? status + : 'all'; const filtered = normalizedStatus === 'all' ? loaded.runs : loaded.runs.filter((run) => run.shadowStatus === normalizedStatus); @@ -510,7 +536,8 @@ export function createOrchestratorObservabilityService({ const [eventRows] = await pool.query( `SELECT id AS event_id, event_type, data_json, created_at FROM h5_agent_run_events - WHERE run_id = ? AND event_type IN (?, ?) + WHERE run_id = ? + AND event_type IN (${SHADOW_EVENT_TYPES.map(() => '?').join(', ')}) ORDER BY created_at ASC, id ASC`, [normalizedRunId, ...SHADOW_EVENT_TYPES], ); diff --git a/services/orchestrator/observability.test.mjs b/services/orchestrator/observability.test.mjs index b201ffc..d7de02b 100644 --- a/services/orchestrator/observability.test.mjs +++ b/services/orchestrator/observability.test.mjs @@ -50,6 +50,18 @@ test('observability aggregates shadow outcomes and latency percentiles', async ( native_attempts: 2, native_completed_at: 1500, }, + { + event_id: 'event-skipped', + run_id: 'run-skipped', + event_type: 'workflow_shadow_skipped', + data_json: { reason: 'shadow_queue_full' }, + event_created_at: 900, + request_id: 'request-skipped', + user_id: 'user-skipped', + native_status: 'succeeded', + native_attempts: 1, + native_completed_at: 1200, + }, ]; const service = createOrchestratorObservabilityService({ pool: { @@ -61,16 +73,20 @@ test('observability aggregates shadow outcomes and latency percentiles', async ( configService: { getRuntimeState() {} }, }); - const result = await service.listShadowRuns({ hours: 12, limit: 2 }); + const result = await service.listShadowRuns({ hours: 12, limit: 4 }); assert.equal(result.window.hours, 12); - assert.equal(result.metrics.observations, 3); + assert.equal(result.metrics.observations, 4); assert.equal(result.metrics.successes, 2); assert.equal(result.metrics.failures, 1); + assert.equal(result.metrics.skipped, 1); + assert.equal(result.metrics.skipRate, 0.25); assert.equal(result.metrics.latencyP50Ms, 200); assert.equal(result.metrics.latencyP95Ms, 300); - assert.equal(result.metrics.nativeSucceeded, 2); - assert.equal(result.runs.length, 2); + assert.equal(result.metrics.nativeSucceeded, 3); + assert.equal(result.runs.length, 4); assert.equal(result.runs[1].error.code, 'TIMEOUT'); + assert.equal(result.runs[3].shadowStatus, 'skipped'); + assert.equal(result.runs[3].skipReason, 'shadow_queue_full'); }); test('observability aggregates complete dry-run routing decisions and Native outcomes', async () => { @@ -339,6 +355,9 @@ test('canary readiness excludes smoke runs and reports explicit blockers', async assert.deepEqual(options, { probe: true }); return { config: { mode: 'shadow' }, + runtime: { + shadowObservation: { enabled: true }, + }, serviceHealth: { ok: true, status: 'healthy', @@ -370,6 +389,7 @@ test('canary readiness excludes smoke runs and reports explicit blockers', async test('canary readiness passes only when operational and sample gates all pass', async () => { const now = 20_000_000; + let shadowObservationEnabled = true; const rows = Array.from({ length: 20 }, (_, index) => ({ event_id: `event-${index}`, run_id: `run-${index}`, @@ -399,6 +419,9 @@ test('canary readiness passes only when operational and sample gates all pass', async getRuntimeState() { return { config: { mode: 'shadow' }, + runtime: { + shadowObservation: { enabled: shadowObservationEnabled }, + }, serviceHealth: { ok: true, status: 'healthy', @@ -423,10 +446,67 @@ test('canary readiness passes only when operational and sample gates all pass', assert.equal(readiness.window.hours, 168); assert.equal(readiness.samples.eligibleObservations, 20); assert.equal(readiness.samples.successRate, 0.95); + assert.equal(readiness.samples.skipped, 0); + assert.equal(readiness.samples.skipRate, 0); assert.equal(readiness.samples.latencyCoverageRate, 1); assert.equal(readiness.samples.nativeSettledRate, 1); assert.equal(readiness.samples.distinctSessions, 5); assert.equal(readiness.service.executorJobStoreDurable, true); assert.equal(readiness.blockers.length, 0); assert.deepEqual(readiness.failureCodes, [{ code: 'TRANSIENT', count: 1 }]); + + shadowObservationEnabled = false; + const blockedReadiness = await service.getCanaryReadiness(); + assert.equal(blockedReadiness.ready, false); + assert.ok(blockedReadiness.blockers.includes('shadow_wiring_enabled')); +}); + +test('canary readiness blocks when any Shadow observation was skipped', async () => { + const now = 30_000_000; + const rows = Array.from({ length: 20 }, (_, index) => ({ + event_id: `event-skip-${index}`, + run_id: `run-skip-${index}`, + event_type: index === 0 + ? 'workflow_shadow_skipped' + : 'workflow_shadow_completed', + data_json: index === 0 + ? { reason: 'shadow_queue_full' } + : { latencyMs: 100, taskType: 'code_task' }, + event_created_at: now - index * 1000, + request_id: `request-skip-${index}`, + user_id: `user-${index % 3}`, + agent_session_id: `session-${index % 5}`, + native_status: 'succeeded', + native_attempts: 1, + native_completed_at: now, + })); + const service = createOrchestratorObservabilityService({ + pool: { async query() { return [rows]; } }, + configService: { + async getRuntimeState() { + return { + config: { mode: 'shadow' }, + runtime: { + shadowObservation: { enabled: true }, + }, + serviceHealth: { + ok: true, + status: 'healthy', + details: { + checkpoint: { kind: 'postgres', durable: true }, + executorGateway: { store: { kind: 'postgres', durable: true } }, + execution: 'observe-only', + }, + }, + }; + }, + }, + nowMs: () => now, + }); + + const readiness = await service.getCanaryReadiness(); + assert.equal(readiness.ready, false); + assert.equal(readiness.samples.skipped, 1); + assert.equal(readiness.samples.skipRate, 0.05); + assert.ok(readiness.blockers.includes('shadow_skip_rate')); }); diff --git a/services/orchestrator/runtime.mjs b/services/orchestrator/runtime.mjs index 63734ce..c65ec65 100644 --- a/services/orchestrator/runtime.mjs +++ b/services/orchestrator/runtime.mjs @@ -11,6 +11,8 @@ import { import { createExecutorWorkerCoordinator } from './executor-worker-protocol.mjs'; import { createExecutorAdmissionPolicy } from './executor-admission-policy.mjs'; +const TERMINAL_RUN_STATUSES = new Set(['succeeded', 'failed', 'cancelled']); + function graphConfig(runId) { return { configurable: { @@ -134,6 +136,35 @@ export function createLangGraphOrchestratorRuntime({ ); } + async function deleteRun(runId) { + const normalizedRunId = String(runId ?? '').trim(); + if (!normalizedRunId) return null; + const state = await getState(normalizedRunId); + if (!state) return null; + if (!TERMINAL_RUN_STATUSES.has(state.status)) { + const error = new Error('Workflow run must be terminal before deletion'); + error.code = 'WORKFLOW_RUN_NOT_TERMINAL'; + error.status = 409; + throw error; + } + if (typeof checkpointer.deleteThread !== 'function') { + const error = new Error('Checkpoint backend does not support run deletion'); + error.code = 'WORKFLOW_DELETE_UNSUPPORTED'; + error.status = 501; + throw error; + } + const executorJobId = state.plan?.executorJob?.id ?? null; + const executorJobDeleted = executorJobId + ? await executorGateway.deleteJob(executorJobId) + : false; + await checkpointer.deleteThread(normalizedRunId); + return { + runId: normalizedRunId, + deleted: true, + executorJobDeleted, + }; + } + function health() { return { status: 'ok', @@ -215,6 +246,77 @@ export function createLangGraphOrchestratorRuntime({ throw error; }, + deleteRun, + + async purgeTerminalRuns({ + before, + limit = 100, + dryRun = true, + } = {}) { + const cutoff = Number(before); + if (!Number.isFinite(cutoff) || cutoff <= 0) { + const error = new Error('Retention purge requires a positive before timestamp'); + error.code = 'WORKFLOW_PURGE_CUTOFF_REQUIRED'; + error.status = 422; + throw error; + } + const pageSize = Math.min(500, Math.max(1, Number(limit) || 100)); + const candidates = await executorGateway.listRetentionCandidates({ + before: cutoff, + limit: pageSize, + }); + const uniqueCandidates = [...new Map( + candidates.map((candidate) => [candidate.workflowRunId, candidate]), + ).values()]; + if (dryRun) { + return { + version: 'orchestrator-retention-purge-v1', + dryRun: true, + before: cutoff, + candidates: uniqueCandidates, + deleted: [], + failures: [], + }; + } + const deleted = []; + const failures = []; + for (const candidate of uniqueCandidates) { + try { + const result = await deleteRun(candidate.workflowRunId); + if (result) { + deleted.push({ + runId: candidate.workflowRunId, + jobId: candidate.jobId, + }); + continue; + } + const executorJobDeleted = await executorGateway.deleteJob(candidate.jobId); + if (executorJobDeleted) { + deleted.push({ + runId: candidate.workflowRunId, + jobId: candidate.jobId, + orphanedCheckpoint: true, + }); + } + } catch (error) { + failures.push({ + runId: candidate.workflowRunId, + jobId: candidate.jobId, + code: String(error?.code ?? 'WORKFLOW_PURGE_FAILED').slice(0, 128), + message: String(error instanceof Error ? error.message : error).slice(0, 500), + }); + } + } + return { + version: 'orchestrator-retention-purge-v1', + dryRun: false, + before: cutoff, + candidates: uniqueCandidates, + deleted, + failures, + }; + }, + getExecutorJob(jobId) { return executorGateway.getJob(jobId); }, @@ -294,4 +396,5 @@ export function createLangGraphOrchestratorRuntime({ export const orchestratorRuntimeInternals = { graphConfig, projectSnapshot, + TERMINAL_RUN_STATUSES, }; diff --git a/services/orchestrator/runtime.test.mjs b/services/orchestrator/runtime.test.mjs index 95d1860..d6fb5ee 100644 --- a/services/orchestrator/runtime.test.mjs +++ b/services/orchestrator/runtime.test.mjs @@ -87,6 +87,45 @@ test('LangGraph runtime is idempotent by run id', async () => { assert.equal((await runtime.listEvents('run-shadow-1')).events.length, 3); }); +test('LangGraph runtime deletes terminal checkpoints and linked Executor Jobs', async () => { + const runtime = createLangGraphOrchestratorRuntime({ + checkpointer: new MemorySaver(), + }); + await runtime.start(runSpec()); + assert.deepEqual(await runtime.deleteRun('run-shadow-1'), { + runId: 'run-shadow-1', + deleted: true, + executorJobDeleted: true, + }); + assert.equal(await runtime.getState('run-shadow-1'), null); + assert.equal(await runtime.getExecutorJob('run-shadow-1:executor-preview'), null); + assert.equal(await runtime.deleteRun('run-shadow-1'), null); +}); + +test('LangGraph runtime retention purge is dry-run by default and deletes only on apply', async () => { + const runtime = createLangGraphOrchestratorRuntime({ + checkpointer: new MemorySaver(), + }); + await runtime.start(runSpec({ runId: 'run-retention-1' })); + const before = Date.now() + 1000; + const preview = await runtime.purgeTerminalRuns({ before }); + assert.equal(preview.dryRun, true); + assert.equal(preview.candidates.length, 1); + assert.equal(await runtime.getState('run-retention-1') != null, true); + + const applied = await runtime.purgeTerminalRuns({ + before, + dryRun: false, + }); + assert.equal(applied.deleted.length, 1); + assert.equal(applied.failures.length, 0); + assert.equal(await runtime.getState('run-retention-1'), null); + await assert.rejects( + () => runtime.purgeTerminalRuns({ before: 0 }), + (error) => error.code === 'WORKFLOW_PURGE_CUTOFF_REQUIRED', + ); +}); + test('LangGraph runtime persists a durable blocked Executor Job and probes both stores', async () => { const checkpointProbeCalls = []; const executorProbeCalls = []; @@ -179,6 +218,10 @@ test('LangGraph active workflow waits on a leased Executor Job and converges ter assert.equal(state.status, 'waiting'); assert.equal(state.phase, 'executor_queued'); assert.equal(state.plan.executorJob.status, 'queued'); + await assert.rejects( + () => runtime.deleteRun('run-active-1'), + (error) => error.code === 'WORKFLOW_RUN_NOT_TERMINAL' && error.status === 409, + ); const claim = await runtime.claimExecutorJob({ workerId: 'worker-1', diff --git a/services/orchestrator/server.mjs b/services/orchestrator/server.mjs index 51cb36a..42de433 100644 --- a/services/orchestrator/server.mjs +++ b/services/orchestrator/server.mjs @@ -25,10 +25,112 @@ function csvList(value) { )]; } +function isLoopbackHost(value) { + const host = String(value ?? '').trim().toLowerCase(); + return host === 'localhost' + || host === '::1' + || host === '[::1]' + || /^127(?:\.\d{1,3}){3}$/.test(host); +} + +function assertSecureServerConfig({ + host, + executionEnabled, + serviceToken, + workerToken, +} = {}) { + const hasServiceToken = Boolean(String(serviceToken ?? '').trim()); + const hasWorkerToken = Boolean(String(workerToken ?? '').trim()); + if (!isLoopbackHost(host) && !hasServiceToken) { + const error = new Error( + 'MEMIND_ORCHESTRATOR_SERVICE_TOKEN is required for non-loopback binding', + ); + error.code = 'ORCHESTRATOR_SERVICE_TOKEN_REQUIRED'; + throw error; + } + if (executionEnabled && (!hasServiceToken || !hasWorkerToken)) { + const error = new Error( + 'Execution requires both MEMIND_ORCHESTRATOR_SERVICE_TOKEN and MEMIND_ORCHESTRATOR_WORKER_TOKEN', + ); + error.code = 'ORCHESTRATOR_EXECUTION_TOKENS_REQUIRED'; + throw error; + } + if ( + executionEnabled + && String(serviceToken).trim() === String(workerToken).trim() + ) { + const error = new Error('Execution service and worker tokens must be distinct'); + error.code = 'ORCHESTRATOR_EXECUTION_TOKENS_NOT_DISTINCT'; + throw error; + } +} + +function startRetentionSweep(runtime, { + retentionDays, + intervalMs = 24 * 60 * 60 * 1000, + limit = 100, + nowMs = Date.now, + setIntervalFn = setInterval, + clearIntervalFn = clearInterval, + logger = console, +} = {}) { + const days = Number(retentionDays); + if (!Number.isFinite(days) || days <= 0) return null; + const boundedDays = Math.min(3650, Math.max(1, Math.floor(days))); + const configuredInterval = Number(intervalMs); + const boundedInterval = Number.isFinite(configuredInterval) + ? Math.max(60 * 60 * 1000, configuredInterval) + : 24 * 60 * 60 * 1000; + let active = false; + const sweep = async () => { + if (active) return null; + active = true; + try { + const result = await runtime.purgeTerminalRuns({ + before: nowMs() - boundedDays * 24 * 60 * 60 * 1000, + limit, + dryRun: false, + }); + if (result.deleted.length || result.failures.length) { + logger.log( + `[orchestrator-retention] deleted=${result.deleted.length} failures=${result.failures.length}`, + ); + } + return result; + } catch (error) { + logger.warn( + '[orchestrator-retention] sweep failed:', + error instanceof Error ? error.message : error, + ); + return null; + } finally { + active = false; + } + }; + const timer = setIntervalFn(() => void sweep(), boundedInterval); + timer?.unref?.(); + return { + retentionDays: boundedDays, + intervalMs: boundedInterval, + sweep, + stop() { + clearIntervalFn(timer); + }, + }; +} + export async function startOrchestratorServer({ env = process.env, logger = console, } = {}) { + const host = String(env.MEMIND_ORCHESTRATOR_HOST ?? '127.0.0.1').trim() || '127.0.0.1'; + const executionEnabled = envFlag(env.MEMIND_ORCHESTRATOR_EXECUTION_ENABLED, false); + assertSecureServerConfig({ + host, + executionEnabled, + serviceToken: env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN, + workerToken: env.MEMIND_ORCHESTRATOR_WORKER_TOKEN, + }); const checkpoint = await createOrchestratorCheckpoint({ mode: env.MEMIND_ORCHESTRATOR_CHECKPOINT_MODE, connectionString: env.MEMIND_ORCHESTRATOR_DATABASE_URL, @@ -52,7 +154,7 @@ export async function startOrchestratorServer({ checkpointProbe: checkpoint.probe, executorJobStore: executorJobs.store, executorJobProbe: executorJobs.probe, - executionEnabled: envFlag(env.MEMIND_ORCHESTRATOR_EXECUTION_ENABLED, false), + executionEnabled, enabledExecutors: csvList(env.MEMIND_ORCHESTRATOR_ENABLED_EXECUTORS), admissionConfig: { tenantAllowlist: csvList(env.MEMIND_ORCHESTRATOR_TENANT_ALLOWLIST), @@ -72,8 +174,8 @@ export async function startOrchestratorServer({ serviceToken: env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN, workerToken: env.MEMIND_ORCHESTRATOR_WORKER_TOKEN, }); - const host = String(env.MEMIND_ORCHESTRATOR_HOST ?? '127.0.0.1').trim() || '127.0.0.1'; const port = positivePort(env.MEMIND_ORCHESTRATOR_PORT, 8093); + let retentionSweep = null; let server; try { server = await new Promise((resolve, reject) => { @@ -84,9 +186,16 @@ export async function startOrchestratorServer({ await Promise.allSettled([executorJobs.close(), checkpoint.close()]); throw error; } + retentionSweep = startRetentionSweep(runtime, { + retentionDays: env.MEMIND_ORCHESTRATOR_RETENTION_DAYS, + intervalMs: env.MEMIND_ORCHESTRATOR_RETENTION_SWEEP_INTERVAL_MS, + limit: env.MEMIND_ORCHESTRATOR_RETENTION_SWEEP_LIMIT, + logger, + }); logger.log(`[orchestrator] listening on http://${host}:${port} (${checkpoint.kind})`); async function close() { + retentionSweep?.stop(); let serverCloseError = null; try { await new Promise((resolve, reject) => { @@ -110,6 +219,7 @@ export async function startOrchestratorServer({ runtime, checkpoint, executorJobs, + retentionSweep, close, }; } @@ -129,7 +239,10 @@ if (isEntrypoint) { } export const orchestratorServerInternals = { + assertSecureServerConfig, csvList, envFlag, + isLoopbackHost, positivePort, + startRetentionSweep, }; diff --git a/services/orchestrator/server.test.mjs b/services/orchestrator/server.test.mjs index 93d0fde..bb7df7a 100644 --- a/services/orchestrator/server.test.mjs +++ b/services/orchestrator/server.test.mjs @@ -1,7 +1,10 @@ import assert from 'node:assert/strict'; import net from 'node:net'; import test from 'node:test'; -import { startOrchestratorServer } from './server.mjs'; +import { + orchestratorServerInternals, + startOrchestratorServer, +} from './server.mjs'; async function reservePort() { const server = net.createServer(); @@ -46,3 +49,79 @@ test('orchestrator server owns checkpoint and Executor Job Store lifecycle', asy closed = true; assert.equal(running.server.listening, false); }); + +test('orchestrator server fails closed for exposed or executable configurations without tokens', () => { + assert.throws( + () => orchestratorServerInternals.assertSecureServerConfig({ + host: '0.0.0.0', + executionEnabled: false, + }), + (error) => error.code === 'ORCHESTRATOR_SERVICE_TOKEN_REQUIRED', + ); + assert.throws( + () => orchestratorServerInternals.assertSecureServerConfig({ + host: '127.0.0.1', + executionEnabled: true, + serviceToken: 'service-token', + }), + (error) => error.code === 'ORCHESTRATOR_EXECUTION_TOKENS_REQUIRED', + ); + assert.throws( + () => orchestratorServerInternals.assertSecureServerConfig({ + host: '127.0.0.1', + executionEnabled: true, + serviceToken: 'shared-token', + workerToken: 'shared-token', + }), + (error) => error.code === 'ORCHESTRATOR_EXECUTION_TOKENS_NOT_DISTINCT', + ); + assert.doesNotThrow( + () => orchestratorServerInternals.assertSecureServerConfig({ + host: '0.0.0.0', + executionEnabled: true, + serviceToken: 'service-token', + workerToken: 'worker-token', + }), + ); +}); + +test('orchestrator retention sweep is disabled by default and applies an explicit cutoff', async () => { + const calls = []; + let intervalCallback = null; + let cleared = false; + const runtime = { + async purgeTerminalRuns(options) { + calls.push(options); + return { deleted: [], failures: [] }; + }, + }; + assert.equal( + orchestratorServerInternals.startRetentionSweep(runtime, { + retentionDays: 0, + }), + null, + ); + const sweep = orchestratorServerInternals.startRetentionSweep(runtime, { + retentionDays: 30, + intervalMs: 60 * 60 * 1000, + nowMs: () => 40 * 24 * 60 * 60 * 1000, + setIntervalFn(callback) { + intervalCallback = callback; + return { unref() {} }; + }, + clearIntervalFn() { + cleared = true; + }, + logger: { log() {}, warn() {} }, + }); + assert.equal(sweep.retentionDays, 30); + intervalCallback(); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(calls, [{ + before: 10 * 24 * 60 * 60 * 1000, + limit: 100, + dryRun: false, + }]); + sweep.stop(); + assert.equal(cleared, true); +}); diff --git a/services/orchestrator/shadow-dispatcher.mjs b/services/orchestrator/shadow-dispatcher.mjs new file mode 100644 index 0000000..729c39e --- /dev/null +++ b/services/orchestrator/shadow-dispatcher.mjs @@ -0,0 +1,93 @@ +function positiveInteger(value, fallback) { + const parsed = Number(value); + return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback; +} + +export function createShadowObservationDispatcher({ + observe, + onCompleted = async () => {}, + onFailed = async () => {}, + onSkipped = async () => {}, + maxConcurrent = 2, + maxQueued = 100, + schedule = setImmediate, + now = Date.now, + logger = console, +} = {}) { + const enabled = typeof observe === 'function'; + const concurrency = positiveInteger(maxConcurrent, 2); + const queueLimit = positiveInteger(maxQueued, 100); + const queue = []; + let active = 0; + let drainScheduled = false; + + function reportCallbackFailure(label, error) { + logger.warn( + `[orchestrator-shadow] ${label} callback failed:`, + error instanceof Error ? error.message : error, + ); + } + + function scheduleDrain() { + if (!enabled || drainScheduled || !queue.length || active >= concurrency) return; + drainScheduled = true; + schedule(drain); + } + + function runObservation(entry) { + active += 1; + const context = { + enqueuedAt: entry.enqueuedAt, + startedAt: now(), + }; + void Promise.resolve() + .then(() => observe(entry.input)) + .then((result) => onCompleted(entry.input, result, context)) + .catch((error) => onFailed(entry.input, error, context)) + .catch((error) => reportCallbackFailure('failure', error)) + .finally(() => { + active -= 1; + scheduleDrain(); + }); + } + + function drain() { + drainScheduled = false; + while (active < concurrency && queue.length) { + runObservation(queue.shift()); + } + } + + return { + dispatch(input) { + if (!enabled) return false; + if (queue.length >= queueLimit) { + const context = { + enqueuedAt: now(), + reason: 'shadow_queue_full', + }; + void Promise.resolve() + .then(() => onSkipped(input, context)) + .catch((error) => reportCallbackFailure('skip', error)); + return false; + } + queue.push({ input, enqueuedAt: now() }); + scheduleDrain(); + return true; + }, + + status() { + return { + enabled, + active, + queued: queue.length, + maxConcurrent: concurrency, + maxQueued: queueLimit, + }; + }, + }; +} + +export const shadowDispatcherInternals = { + positiveInteger, +}; diff --git a/services/orchestrator/shadow-dispatcher.test.mjs b/services/orchestrator/shadow-dispatcher.test.mjs new file mode 100644 index 0000000..b248529 --- /dev/null +++ b/services/orchestrator/shadow-dispatcher.test.mjs @@ -0,0 +1,71 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { createShadowObservationDispatcher } from './shadow-dispatcher.mjs'; + +async function flush() { + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); +} + +test('shadow dispatcher is a no-op when observation is disabled', () => { + const dispatcher = createShadowObservationDispatcher(); + assert.equal(dispatcher.dispatch({ runId: 'run-disabled' }), false); + assert.deepEqual(dispatcher.status(), { + enabled: false, + active: 0, + queued: 0, + maxConcurrent: 2, + maxQueued: 100, + }); +}); + +test('shadow dispatcher bounds concurrency and drops overflow without blocking callers', async () => { + const releases = []; + const started = []; + const skipped = []; + const dispatcher = createShadowObservationDispatcher({ + maxConcurrent: 1, + maxQueued: 2, + observe: async ({ runId }) => { + started.push(runId); + await new Promise((resolve) => releases.push(resolve)); + return { runId }; + }, + onSkipped: async ({ runId }, context) => { + skipped.push([runId, context.reason]); + }, + logger: { warn() {} }, + }); + + assert.equal(dispatcher.dispatch({ runId: 'run-1' }), true); + assert.equal(dispatcher.dispatch({ runId: 'run-2' }), true); + assert.equal(dispatcher.dispatch({ runId: 'run-3' }), false); + await flush(); + assert.deepEqual(started, ['run-1']); + assert.deepEqual(skipped, [['run-3', 'shadow_queue_full']]); + + releases.shift()(); + await flush(); + assert.deepEqual(started, ['run-1', 'run-2']); + releases.shift()(); + await flush(); + assert.equal(dispatcher.status().active, 0); +}); + +test('shadow dispatcher isolates observer and callback failures', async () => { + const failures = []; + const dispatcher = createShadowObservationDispatcher({ + observe: async () => { + throw Object.assign(new Error('remote unavailable'), { code: 'REMOTE_DOWN' }); + }, + onFailed: async ({ runId }, error) => { + failures.push([runId, error.code]); + }, + logger: { warn() {} }, + }); + + assert.equal(dispatcher.dispatch({ runId: 'run-failed' }), true); + await flush(); + assert.deepEqual(failures, [['run-failed', 'REMOTE_DOWN']]); + assert.equal(dispatcher.status().active, 0); +}); diff --git a/services/orchestrator/shadow-observer.mjs b/services/orchestrator/shadow-observer.mjs index 5eeb376..82424a5 100644 --- a/services/orchestrator/shadow-observer.mjs +++ b/services/orchestrator/shadow-observer.mjs @@ -4,19 +4,6 @@ import { } from './contracts.mjs'; import { createRemoteWorkflowEngine } from './engine-registry.mjs'; -const MAX_INSTRUCTION_CHARACTERS = 16_000; - -function extractMessageText(message) { - if (typeof message === 'string') return message; - if (!message || typeof message !== 'object') return ''; - if (typeof message.content === 'string') return message.content; - if (!Array.isArray(message.content)) return ''; - return message.content - .filter((part) => part?.type === 'text') - .map((part) => String(part.text ?? '')) - .join('\n'); -} - function safeError(error) { return { code: String(error?.code ?? 'WORKFLOW_SHADOW_FAILED').slice(0, 128), @@ -38,9 +25,7 @@ export function createWorkflowShadowObserver({ runId, requestId, userId, - sessionId = null, workflowName = 'code-run-v1', - userMessage, taskType = null, } = {}) { const selection = await configService.selectEngine({ @@ -81,19 +66,15 @@ export function createWorkflowShadowObserver({ timeoutMs: state.config.requestTimeoutMs, fetchImpl, }); - const instruction = extractMessageText(userMessage).slice(0, MAX_INSTRUCTION_CHARACTERS); const spec = normalizeRunSpec({ runId, requestId, workflow: { name: workflowName, version: 1 }, - subject: { userId }, + subject: {}, input: { - instruction, + instruction: '[shadow-control-plane-only]', taskType, toolMode: 'code', - sessionRef: sessionId - ? { kind: 'goose-session', id: String(sessionId) } - : null, }, policy: { executionMode: 'observe-only', @@ -102,6 +83,7 @@ export function createWorkflowShadowObserver({ metadata: { source: 'memind-agent-run', configVersion: selection.configVersion, + dataPolicy: 'control-plane-only-v1', }, }); try { @@ -126,7 +108,5 @@ export function createWorkflowShadowObserver({ } export const workflowShadowObserverInternals = { - MAX_INSTRUCTION_CHARACTERS, - extractMessageText, safeError, }; diff --git a/services/orchestrator/shadow-observer.test.mjs b/services/orchestrator/shadow-observer.test.mjs index 548a641..79c4dbc 100644 --- a/services/orchestrator/shadow-observer.test.mjs +++ b/services/orchestrator/shadow-observer.test.mjs @@ -1,5 +1,8 @@ import assert from 'node:assert/strict'; import test from 'node:test'; +import { MemorySaver } from '@langchain/langgraph'; +import { createOrchestratorApp } from './app.mjs'; +import { createLangGraphOrchestratorRuntime } from './runtime.mjs'; import { createWorkflowShadowObserver } from './shadow-observer.mjs'; function jsonResponse(body, status = 200) { @@ -9,6 +12,19 @@ function jsonResponse(body, status = 200) { }); } +async function listen(app) { + const server = await new Promise((resolve) => { + const candidate = app.listen(0, '127.0.0.1', () => resolve(candidate)); + }); + const address = server.address(); + return { + baseUrl: `http://127.0.0.1:${address.port}`, + close: () => new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }), + }; +} + test('shadow observer skips without creating a remote client when mode does not select shadow', async () => { let fetchCalls = 0; const observer = createWorkflowShadowObserver({ @@ -127,7 +143,7 @@ test('shadow boundary records non-selected canary decisions for a complete rate assert.equal(result.executionPlan.taskType, 'code_analysis'); }); -test('shadow observer sends a bounded observe-only RunSpec to LangGraph service', async () => { +test('shadow observer sends a control-plane-only RunSpec without user content or identity', async () => { let capturedUrl = null; let capturedInit = null; const observer = createWorkflowShadowObserver({ @@ -175,6 +191,98 @@ test('shadow observer sends a bounded observe-only RunSpec to LangGraph service' assert.equal(body.version, 'orchestrator-run-v1'); assert.equal(body.policy.executionMode, 'observe-only'); assert.equal(body.policy.sideEffectsAllowed, false); - assert.deepEqual(body.input.sessionRef, { kind: 'goose-session', id: 'session-2' }); + assert.equal(body.input.instruction, '[shadow-control-plane-only]'); + assert.equal('sessionRef' in body.input, false); + assert.deepEqual(body.subject, { tenantId: null, userId: null }); assert.equal(body.metadata.configVersion, 7); + assert.equal(body.metadata.dataPolicy, 'control-plane-only-v1'); + assert.equal( + capturedInit.body.includes('Implement the service boundary'), + false, + ); + assert.equal(capturedInit.body.includes('user-2'), false); + assert.equal(capturedInit.body.includes('session-2'), false); +}); + +test('isolated Portal shadow flow reaches LangGraph over HTTP and removes all terminal state', async (t) => { + const checkpointer = new MemorySaver(); + const runtime = createLangGraphOrchestratorRuntime({ + checkpointer, + checkpointKind: 'memory', + durable: false, + }); + const server = await listen(createOrchestratorApp({ + runtime, + serviceToken: 'isolated-shadow-token', + })); + t.after(server.close); + + const observer = createWorkflowShadowObserver({ + configService: { + async selectEngine() { + return { + shadowEngine: 'langgraph', + reason: 'shadow', + mode: 'shadow', + configVersion: 11, + }; + }, + async getRuntimeState() { + return { + config: { + serviceUrl: server.baseUrl, + requestTimeoutMs: 2000, + }, + }; + }, + }, + serviceToken: 'isolated-shadow-token', + }); + + const secretUserContent = 'private user content must never cross the shadow boundary'; + const result = await observer({ + runId: 'isolated-shadow-run-1', + requestId: 'isolated-shadow-request-1', + userId: 'private-user-id', + sessionId: 'private-session-id', + taskType: 'repo_refactor', + userMessage: { + role: 'user', + content: [{ type: 'text', text: secretUserContent }], + }, + }); + + assert.equal(result.observed, true); + assert.equal(result.shadowRun.status, 'succeeded'); + assert.equal(result.shadowRun.result.executed, false); + const tuple = await checkpointer.getTuple({ + configurable: { + thread_id: 'isolated-shadow-run-1', + checkpoint_ns: '', + }, + }); + assert.equal( + tuple.checkpoint.channel_values.spec.input.instruction, + '[shadow-control-plane-only]', + ); + const checkpointJson = JSON.stringify(tuple.checkpoint.channel_values.spec); + assert.equal(checkpointJson.includes(secretUserContent), false); + assert.equal(checkpointJson.includes('private-user-id'), false); + assert.equal(checkpointJson.includes('private-session-id'), false); + + const deleted = await fetch(`${server.baseUrl}/v1/runs/isolated-shadow-run-1`, { + method: 'DELETE', + headers: { authorization: 'Bearer isolated-shadow-token' }, + }); + assert.equal(deleted.status, 200); + assert.deepEqual(await deleted.json(), { + runId: 'isolated-shadow-run-1', + deleted: true, + executorJobDeleted: true, + }); + assert.equal(await runtime.getState('isolated-shadow-run-1'), null); + assert.equal( + await runtime.getExecutorJob('isolated-shadow-run-1:executor-preview'), + null, + ); });