diff --git a/admin-bootstrap.mjs b/admin-bootstrap.mjs index 01fc55f..41679cc 100644 --- a/admin-bootstrap.mjs +++ b/admin-bootstrap.mjs @@ -20,6 +20,7 @@ import { createAssetGatewayConfigService } from './asset-gateway.mjs'; import { createImageMakeAdminConfigService } from './image-make-admin-config.mjs'; import { createMemoryV2AdminConfigService } from './memory-v2-admin-config.mjs'; import { createOrchestratorAdminConfigService } from './services/orchestrator/admin-config.mjs'; +import { createOrchestratorObservabilityService } from './services/orchestrator/observability.mjs'; import { createSkillRuntimeAdminConfigService } from './skill-runtime-admin-config.mjs'; import { createSystemDisclosurePolicyService } from './system-disclosure-policy.mjs'; import { createMindSearchConfigService } from './mindsearch-config.mjs'; @@ -112,6 +113,10 @@ export async function createAdminServices(env = {}) { const memoryV2ConfigService = createMemoryV2AdminConfigService(pool); const orchestratorConfigService = createOrchestratorAdminConfigService(pool); await orchestratorConfigService.ensureSchema(); + const orchestratorObservabilityService = createOrchestratorObservabilityService({ + pool, + configService: orchestratorConfigService, + }); const mindSearchConfigService = createMindSearchConfigService(pool); await mindSearchConfigService.ensureSchema(); const skillRuntimeConfigService = createSkillRuntimeAdminConfigService(pool, { h5Root }); @@ -159,6 +164,7 @@ export async function createAdminServices(env = {}) { imageMakeAdminConfigService, memoryV2ConfigService, orchestratorConfigService, + orchestratorObservabilityService, mindSearchConfigService, skillRuntimeConfigService, systemDisclosurePolicyService, diff --git a/admin-routes.mjs b/admin-routes.mjs index 07462f6..f62e9a4 100644 --- a/admin-routes.mjs +++ b/admin-routes.mjs @@ -32,6 +32,7 @@ function plazaRouteError(res, req, error) { * @param {object|null} deps.llmProviderService * @param {object|null} deps.memoryV2ConfigService * @param {object|null} deps.orchestratorConfigService + * @param {object|null} deps.orchestratorObservabilityService * @param {object|null} deps.skillRuntimeConfigService * @param {object|null} deps.systemDisclosurePolicyService * @param {object|null} deps.adminSystemTestService @@ -49,6 +50,7 @@ export function createAdminApi({ imageMakeAdminConfigService, memoryV2ConfigService, orchestratorConfigService, + orchestratorObservabilityService, mindSearchConfigService, skillRuntimeConfigService, systemDisclosurePolicyService, @@ -255,6 +257,26 @@ export function createAdminApi({ return res.json(await orchestratorConfigService.getRuntimeState({ probe: true })); }); + adminApi.get('/orchestrator/shadow-runs', requireAdmin, async (req, res) => { + if (!orchestratorObservabilityService?.listShadowRuns) { + return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); + } + return res.json(await orchestratorObservabilityService.listShadowRuns({ + hours: req.query.hours, + limit: req.query.limit, + status: req.query.status, + })); + }); + + adminApi.get('/orchestrator/shadow-runs/:runId', requireAdmin, async (req, res) => { + if (!orchestratorObservabilityService?.getShadowRun) { + return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); + } + const result = await orchestratorObservabilityService.getShadowRun(req.params.runId); + if (!result) return res.status(404).json({ message: 'Shadow run 不存在' }); + return res.json(result); + }); + adminApi.get('/skill-runtime/config', requireAdmin, async (_req, res) => { if (!skillRuntimeConfigService?.getAdminConfig) { return res.status(503).json({ message: 'Skill Runtime 配置服务未启用' }); diff --git a/admin-routes.test.mjs b/admin-routes.test.mjs index c1a5d40..d601980 100644 --- a/admin-routes.test.mjs +++ b/admin-routes.test.mjs @@ -138,6 +138,22 @@ test('admin orchestrator routes expose a versioned plug-in control plane', async }; }, }, + orchestratorObservabilityService: { + async listShadowRuns(options) { + assert.deepEqual(options, { hours: '12', limit: '25', status: 'failed' }); + return { + metrics: { observations: 1, successes: 0, failures: 1 }, + runs: [{ runId: 'run-shadow-1', shadowStatus: 'failed' }], + }; + }, + async getShadowRun(runId) { + assert.equal(runId, 'run-shadow-1'); + return { + native: { runId, status: 'succeeded' }, + remote: { available: true, events: [{ type: 'workflow_validated' }] }, + }; + }, + }, plazaPosts: null, plazaOps: null, wechatAdmin: null, @@ -174,6 +190,20 @@ test('admin orchestrator routes expose a versioned plug-in control plane', async const runtimeBody = await runtimeResponse.json(); assert.equal(runtimeBody.source, 'admin-db'); assert.equal(runtimeBody.serviceHealth.status, 'healthy'); + + const shadowRunsResponse = await fetch( + `${server.baseUrl}/admin-api/orchestrator/shadow-runs?hours=12&limit=25&status=failed`, + { headers: { cookie: 'h5_user_session=token-admin' } }, + ); + assert.equal(shadowRunsResponse.status, 200); + assert.equal((await shadowRunsResponse.json()).metrics.failures, 1); + + const shadowRunResponse = await fetch( + `${server.baseUrl}/admin-api/orchestrator/shadow-runs/run-shadow-1`, + { headers: { cookie: 'h5_user_session=token-admin' } }, + ); + assert.equal(shadowRunResponse.status, 200); + assert.equal((await shadowRunResponse.json()).remote.available, true); } finally { await server.close(); } diff --git a/admin-server.mjs b/admin-server.mjs index 8b623a8..fe39d6b 100644 --- a/admin-server.mjs +++ b/admin-server.mjs @@ -80,6 +80,7 @@ const CONSOLES = { imageMakeAdminConfigService: services.imageMakeAdminConfigService, memoryV2ConfigService: services.memoryV2ConfigService, orchestratorConfigService: services.orchestratorConfigService, + orchestratorObservabilityService: services.orchestratorObservabilityService, mindSearchConfigService: services.mindSearchConfigService, systemDisclosurePolicyService: services.systemDisclosurePolicyService, adminSystemTestService: services.adminSystemTestService, diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index 6426971..445d0e8 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -342,6 +342,7 @@ export function createAgentRunGateway({ function dispatchShadowObservation(input) { if (typeof observeWorkflowRun !== 'function' || input.toolMode !== 'code') return; setImmediate(() => { + const observationStartedAt = nowMs(); void Promise.resolve() .then(() => observeWorkflowRun(input)) .then(async (result) => { @@ -352,6 +353,9 @@ export function createAgentRunGateway({ 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) => { @@ -362,6 +366,7 @@ export function createAgentRunGateway({ 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:', diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 33b28ab..7eae8fa 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -420,13 +420,15 @@ test('code run shadow observation is fire-and-forget and records completion sepa (event) => event.eventType === 'workflow_shadow_completed', )); const event = pool.events.find((item) => item.eventType === 'workflow_shadow_completed'); - assert.deepEqual(JSON.parse(event.dataJson), { - engine: 'langgraph', - mode: 'shadow', - configVersion: 3, - status: 'succeeded', - phase: 'completed', - }); + const eventData = JSON.parse(event.dataJson); + assert.equal(eventData.engine, 'langgraph'); + assert.equal(eventData.mode, 'shadow'); + assert.equal(eventData.configVersion, 3); + assert.equal(eventData.status, 'succeeded'); + assert.equal(eventData.phase, 'completed'); + assert.equal(eventData.taskType, 'repo_refactor'); + assert.equal(eventData.executorAdapter, null); + assert.equal(Number.isFinite(eventData.latencyMs), true); }); test('shadow observer failure cannot fail a Native run and chat runs are not observed', async () => { diff --git a/db.mjs b/db.mjs index 9e7fcfa..f143402 100644 --- a/db.mjs +++ b/db.mjs @@ -351,9 +351,16 @@ export async function migrateSchema(pool) { data_json JSON NULL, created_at BIGINT NOT NULL, KEY idx_h5_agent_run_event_run (run_id, created_at), + KEY idx_h5_agent_run_event_type_time (event_type, created_at), CONSTRAINT fk_h5_agent_run_event_run FOREIGN KEY (run_id) REFERENCES h5_agent_runs(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci `); + if (!(await indexExists(pool, 'h5_agent_run_events', 'idx_h5_agent_run_event_type_time'))) { + await pool.query( + `ALTER TABLE h5_agent_run_events + ADD KEY idx_h5_agent_run_event_type_time (event_type, created_at)`, + ); + } // MindSpace conversation packages: additive provenance tables for grouping // images, files, pages, and publications created during a chat session. diff --git a/deploy/orchestrator/compose.yaml b/deploy/orchestrator/compose.yaml index 90be708..849680d 100644 --- a/deploy/orchestrator/compose.yaml +++ b/deploy/orchestrator/compose.yaml @@ -1,6 +1,6 @@ services: postgres: - image: postgres:17-alpine + image: postgres:16-alpine restart: unless-stopped environment: POSTGRES_DB: memind_orchestrator @@ -17,6 +17,7 @@ services: - orchestrator_internal orchestrator: + image: memind/workflow-orchestrator:local build: context: ../../services/orchestrator dockerfile: Dockerfile diff --git a/docs/architecture/memind-orchestrator-boundary.md b/docs/architecture/memind-orchestrator-boundary.md index 61818e1..e9e36c8 100644 --- a/docs/architecture/memind-orchestrator-boundary.md +++ b/docs/architecture/memind-orchestrator-boundary.md @@ -105,6 +105,11 @@ failure: Successful observations are projected as `workflow_shadow_completed`; graph checkpoints stay in the Orchestrator-owned PostgreSQL database. +Both terminal projection events contain bounded observation latency. The +Memind-owned observability service aggregates those events and may join them to +the product run status. Only per-run detail calls cross the service boundary to +read LangGraph checkpoint state and node events. + ## Deployment evolution 1. Local native Orchestrator process with an explicit MemorySaver for debugging. diff --git a/ops/src/api/admin.ts b/ops/src/api/admin.ts index ae910ad..4c8a983 100644 --- a/ops/src/api/admin.ts +++ b/ops/src/api/admin.ts @@ -257,6 +257,85 @@ export type OrchestratorRuntimeState = OrchestratorConfigState & { serviceHealth: OrchestratorServiceHealth; }; +export type OrchestratorShadowMetrics = { + observations: number; + successes: number; + failures: number; + successRate: number | null; + failureRate: number | null; + latencyP50Ms: number | null; + latencyP95Ms: number | null; + nativeSucceeded: number; + nativeFailed: number; + lastObservedAt: number | null; + sampled: boolean; +}; + +export type OrchestratorShadowRun = { + eventId: string; + runId: string; + requestId: string; + userId: string; + sessionId: string | null; + nativeStatus: string; + nativeAttempts: number; + shadowStatus: 'succeeded' | 'failed'; + engine: string; + configVersion: number | null; + phase: string | null; + taskType: string | null; + executorAdapter: string | null; + latencyMs: number | null; + error: { code: string; message: string } | null; + observedAt: number; + nativeCompletedAt: number | null; +}; + +export type OrchestratorShadowRunList = { + generatedAt: number; + window: { hours: number; from: number }; + metrics: OrchestratorShadowMetrics; + runs: OrchestratorShadowRun[]; +}; + +export type OrchestratorShadowRunDetail = { + native: { + runId: string; + requestId: string; + userId: string; + sessionId: string | null; + status: string; + attempts: number; + error: string | null; + createdAt: number; + updatedAt: number; + startedAt: number | null; + completedAt: number | null; + }; + localEvents: Array<{ + eventId: string; + type: string; + data: Record | null; + createdAt: number; + }>; + remote: { + available: boolean; + state: { + status?: string; + phase?: string; + plan?: Record; + result?: Record; + } | null; + events: Array<{ + sequence?: number; + type?: string; + timestamp?: number; + data?: Record | null; + }>; + error: { code: string; message: string } | null; + }; +}; + // ─── Summary ────────────────────────────────────────────────────────────────── export async function fetchAdminSummary() { @@ -293,6 +372,26 @@ export async function fetchOrchestratorRuntime() { return adminFetch('/admin-api/orchestrator/runtime'); } +export async function fetchOrchestratorShadowRuns(params: { + hours?: number; + limit?: number; + status?: 'all' | 'succeeded' | 'failed'; +} = {}) { + const query = new URLSearchParams(); + if (params.hours) query.set('hours', String(params.hours)); + if (params.limit) query.set('limit', String(params.limit)); + if (params.status) query.set('status', params.status); + return adminFetch( + `/admin-api/orchestrator/shadow-runs?${query}`, + ); +} + +export async function fetchOrchestratorShadowRun(runId: string) { + return adminFetch( + `/admin-api/orchestrator/shadow-runs/${encodeURIComponent(runId)}`, + ); +} + // ─── Users ──────────────────────────────────────────────────────────────────── export async function fetchAdminUsers(params: { diff --git a/ops/src/pages/admin/OrchestratorPage.tsx b/ops/src/pages/admin/OrchestratorPage.tsx index ed7f325..f3afaf2 100644 --- a/ops/src/pages/admin/OrchestratorPage.tsx +++ b/ops/src/pages/admin/OrchestratorPage.tsx @@ -7,6 +7,7 @@ import { type OrchestratorConfigState, type OrchestratorServiceHealth, } from '../../api/admin'; +import { OrchestratorShadowPanel } from './OrchestratorShadowPanel'; function listToText(values: string[]) { return values.join('\n'); @@ -160,6 +161,8 @@ export function OrchestratorPage() {

+ +

路由模式

diff --git a/ops/src/pages/admin/OrchestratorShadowPanel.tsx b/ops/src/pages/admin/OrchestratorShadowPanel.tsx new file mode 100644 index 0000000..a8fa2a8 --- /dev/null +++ b/ops/src/pages/admin/OrchestratorShadowPanel.tsx @@ -0,0 +1,229 @@ +import { useEffect, useState } from 'react'; +import { + fetchOrchestratorShadowRun, + fetchOrchestratorShadowRuns, + type OrchestratorShadowRunDetail, + type OrchestratorShadowRunList, +} from '../../api/admin'; + +function formatTime(value: number | null | undefined) { + return value ? new Date(value).toLocaleString('zh-CN', { hour12: false }) : '—'; +} + +function formatPercent(value: number | null) { + return value == null ? '—' : `${(value * 100).toFixed(1)}%`; +} + +function formatLatency(value: number | null) { + return value == null ? '—' : `${value} ms`; +} + +function shortId(value: string) { + return value.length > 16 ? `${value.slice(0, 8)}…${value.slice(-6)}` : value; +} + +const metricCardStyle = { + border: '1px solid #e4e9e6', + borderRadius: 10, + padding: 12, + minWidth: 140, +} as const; + +export function OrchestratorShadowPanel() { + const [hours, setHours] = useState(24); + const [status, setStatus] = useState<'all' | 'succeeded' | 'failed'>('all'); + const [data, setData] = useState(null); + const [selected, setSelected] = useState(null); + const [loading, setLoading] = useState(true); + const [detailLoading, setDetailLoading] = useState(false); + const [error, setError] = useState(null); + + const load = async () => { + setLoading(true); + setError(null); + try { + setData(await fetchOrchestratorShadowRuns({ hours, status, limit: 100 })); + } catch (err) { + setError(err instanceof Error ? err.message : 'Shadow 指标加载失败'); + } finally { + setLoading(false); + } + }; + + useEffect(() => { + void load(); + }, [hours, status]); + + const openDetail = async (runId: string) => { + setDetailLoading(true); + setError(null); + try { + setSelected(await fetchOrchestratorShadowRun(runId)); + } catch (err) { + setError(err instanceof Error ? err.message : 'Shadow run 详情加载失败'); + } finally { + setDetailLoading(false); + } + }; + + const metrics = data?.metrics; + + return ( +
+
+
+

Shadow 运行观测

+

+ Native 执行结果与 LangGraph 观察结果并排展示，不改变任务执行权。 +

+
+
+ + + +
+
+ + {error ?

{error}

: null} + +
+
+ 观察请求 +
{metrics?.observations ?? '—'}
+
+
+ Shadow 成功率 +
{formatPercent(metrics?.successRate ?? null)}
+
+
+ 失败数 +
{metrics?.failures ?? '—'}
+
+
+ 延迟 P50 +
{formatLatency(metrics?.latencyP50Ms ?? null)}
+
+
+ 延迟 P95 +
{formatLatency(metrics?.latencyP95Ms ?? null)}
+
+
+ Native 成功 / 失败 +
+ {metrics ? `${metrics.nativeSucceeded} / ${metrics.nativeFailed}` : '—'} +
+
+
+ + {metrics?.sampled ? ( +

当前指标基于最近 5000 条 Shadow 结果采样。

+ ) : null} + +
+ + + + + + + + + + + + + + {(data?.runs ?? []).map((run) => ( + + + + + + + + + + ))} + {!loading && !data?.runs.length ? ( + + + + ) : null} + +
时间RunShadowNative延迟任务 / Adapter错误
{formatTime(run.observedAt)} + + + {run.shadowStatus} + {run.nativeStatus}{formatLatency(run.latencyMs)} + {run.taskType ?? '—'} + + {run.executorAdapter ?? '—'} + + + {run.error ? `${run.error.code}: ${run.error.message}` : '—'} +
+ 当前时间范围内没有 Shadow 结果。 +
+
+ + {selected ? ( +
+
+
+

Run 详情：{selected.native.runId}

+

+ Native {selected.native.status} · attempts {selected.native.attempts} + {' · '} + LangGraph {selected.remote.available ? (selected.remote.state?.status ?? 'available') : 'unavailable'} +

+
+ +
+ {selected.remote.error ? ( +

+ {selected.remote.error.code}: {selected.remote.error.message} +

+ ) : null} +
+
+ LangGraph checkpoint + 
+                {JSON.stringify(selected.remote.state, null, 2)}
+
+
+
+ LangGraph 节点事件 + 
+                {JSON.stringify(selected.remote.events, null, 2)}
+
+
+
+
+ ) : null} +
+ ); +} diff --git a/package.json b/package.json index aa97e7e..4e592a3 100644 --- a/package.json +++ b/package.json @@ -56,13 +56,14 @@ "smoke:memory-v2-pgvector": "node scripts/smoke-memory-v2-pgvector.mjs", "smoke:memory-v2-qdrant": "node scripts/smoke-memory-v2-qdrant.mjs", "smoke:memory-v2-external": "node scripts/smoke-memory-v2-external.mjs", + "smoke:orchestrator-shadow": "node scripts/smoke-orchestrator-shadow.mjs", "mock:memory-v2-services": "node scripts/mock-memory-v2-services.mjs", "test:memind": "node scripts/run-memind-tests.mjs", "pretest": "node --test episodic-memory.test.mjs", "test:scenario": "node scripts/run-scenario-test.mjs", "verify:children-hobby-diet-survey": "node scripts/verify-children-hobby-diet-survey.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 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 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-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 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-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/schema.sql b/schema.sql index b7468d4..5305eaf 100644 --- a/schema.sql +++ b/schema.sql @@ -422,6 +422,7 @@ CREATE TABLE IF NOT EXISTS h5_agent_run_events ( data_json JSON NULL, created_at BIGINT NOT NULL, KEY idx_h5_agent_run_event_run (run_id, created_at), + KEY idx_h5_agent_run_event_type_time (event_type, created_at), CONSTRAINT fk_h5_agent_run_event_run FOREIGN KEY (run_id) REFERENCES h5_agent_runs(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/scripts/smoke-orchestrator-shadow.mjs b/scripts/smoke-orchestrator-shadow.mjs new file mode 100644 index 0000000..a318048 --- /dev/null +++ b/scripts/smoke-orchestrator-shadow.mjs @@ -0,0 +1,148 @@ +#!/usr/bin/env node +import crypto from 'node:crypto'; +import { createAgentRunGateway } from '../agent-run-gateway.mjs'; +import { createDbPool, isDatabaseConfigured } from '../db.mjs'; +import { createOrchestratorAdminConfigService } from '../services/orchestrator/admin-config.mjs'; +import { createOrchestratorObservabilityService } from '../services/orchestrator/observability.mjs'; +import { createWorkflowShadowObserver } from '../services/orchestrator/shadow-observer.mjs'; +import { loadMemindEnvFiles } from './memind-runtime-profile.mjs'; + +loadMemindEnvFiles(new URL('..', import.meta.url).pathname); + +const args = new Set(process.argv.slice(2)); +const cleanup = args.has('--cleanup'); +const restoreConfig = args.has('--restore-config'); +const serviceUrl = String( + process.env.MEMIND_ORCHESTRATOR_URL ?? 'http://127.0.0.1:8093', +).trim().replace(/\/$/, ''); +const serviceToken = String(process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN ?? '').trim(); + +function assertLocalServiceUrl(value) { + const url = new URL(value); + if (!['127.0.0.1', 'localhost', '::1'].includes(url.hostname) && !args.has('--allow-remote')) { + throw new Error('Shadow smoke defaults to a loopback Orchestrator; pass --allow-remote explicitly'); + } +} + +async function waitForShadowEvent(pool, runId, timeoutMs = 15_000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const [rows] = await pool.query( + `SELECT event_type, data_json, created_at + FROM h5_agent_run_events + WHERE run_id = ? + AND event_type IN ('workflow_shadow_completed', 'workflow_shadow_failed') + ORDER BY created_at DESC + LIMIT 1`, + [runId], + ); + if (rows[0]) return rows[0]; + await new Promise((resolve) => setTimeout(resolve, 100)); + } + throw new Error(`Timed out waiting for Shadow projection for run ${runId}`); +} + +async function main() { + if (!isDatabaseConfigured()) { + throw new Error('Shadow smoke requires the local Memind MySQL configuration'); + } + if (!serviceToken) { + throw new Error('MEMIND_ORCHESTRATOR_SERVICE_TOKEN is required'); + } + assertLocalServiceUrl(serviceUrl); + const ready = await fetch(`${serviceUrl}/ready`); + if (!ready.ok) throw new Error(`Orchestrator readiness failed with HTTP ${ready.status}`); + + const pool = createDbPool(); + const configService = createOrchestratorAdminConfigService(pool); + const previous = await configService.getAdminConfig(); + let runId = null; + try { + await configService.updateAdminConfig({ + ...previous.config, + mode: 'shadow', + serviceUrl, + primaryEngine: 'langgraph', + workflowAllowlist: ['code-run-v1'], + }); + const [users] = await pool.query( + `SELECT id + FROM h5_users + ORDER BY created_at ASC + LIMIT 1`, + ); + if (!users[0]?.id) throw new Error('Shadow smoke requires at least one local user'); + + const observer = createWorkflowShadowObserver({ + configService, + serviceToken, + }); + const gateway = createAgentRunGateway({ + pool, + tkmindProxy: {}, + autoDispatch: false, + observeWorkflowRun: observer, + }); + const requestId = `orchestrator-shadow-smoke-${Date.now()}-${crypto.randomUUID()}`; + const run = await gateway.createRun(users[0].id, { + requestId, + toolMode: 'code', + taskType: 'orchestrator_shadow_smoke', + userMessage: { + role: 'user', + content: [{ type: 'text', text: 'Observe this local smoke run without executing tools.' }], + }, + }); + runId = run.id; + const event = await waitForShadowEvent(pool, runId); + if (event.event_type !== 'workflow_shadow_completed') { + throw new Error(`Shadow smoke failed: ${JSON.stringify(event.data_json)}`); + } + + const completedAt = Date.now(); + await pool.query( + `UPDATE h5_agent_runs + SET status = 'succeeded', updated_at = ?, completed_at = ? + WHERE id = ? AND status = 'queued'`, + [completedAt, completedAt, runId], + ); + + const observability = createOrchestratorObservabilityService({ + pool, + configService, + serviceToken, + }); + const [summary, detail] = await Promise.all([ + observability.listShadowRuns({ hours: 1, limit: 10 }), + observability.getShadowRun(runId), + ]); + if (!detail?.remote?.available || detail.remote.state?.status !== 'succeeded') { + throw new Error('Shadow checkpoint detail was not recoverable through the remote API'); + } + + console.log(JSON.stringify({ + ok: true, + runId, + mode: 'shadow', + shadowEvent: event.event_type, + metrics: summary.metrics, + checkpointStatus: detail.remote.state.status, + graphEvents: detail.remote.events.map((item) => item.type), + cleanup, + restoreConfig, + }, null, 2)); + } finally { + if (cleanup && runId) { + await pool.query('DELETE FROM h5_agent_runs WHERE id = ?', [runId]); + } + if (restoreConfig) { + await configService.updateAdminConfig(previous.config); + } + await pool.end(); + } +} + +main().catch((error) => { + console.error(error instanceof Error ? error.message : error); + process.exitCode = 1; +}); diff --git a/services/orchestrator/README.md b/services/orchestrator/README.md index 3898ec7..a33987e 100644 --- a/services/orchestrator/README.md +++ b/services/orchestrator/README.md @@ -48,6 +48,18 @@ Phase 2 adds: - live health probing from memindadm; - a separate Colima/Compose deployment artifact. +Phase 2.5 adds a Memind-owned Shadow observability projection: + +- success/failure totals and rates; +- P50/P95 observation latency; +- Native terminal status beside the Shadow result; +- recent failure codes and messages; +- per-run LangGraph checkpoint and node-event inspection through the versioned + remote API. + +The projection reads `h5_agent_run_events` from the Memind control plane. +LangGraph still has no access to Memind business tables. + The default memindadm mode remains `off`. Canary and Active routing are not wired to the LangGraph executor in Phase 2. Native Agent Run remains the sole executor, including in Shadow mode. @@ -84,6 +96,13 @@ 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`. +memindadm exposes the projection through: + +```text +GET /admin-api/orchestrator/shadow-runs +GET /admin-api/orchestrator/shadow-runs/:runId +``` + ## Colima deployment Colima is the recommended first container host on macOS because this service and diff --git a/services/orchestrator/observability.mjs b/services/orchestrator/observability.mjs new file mode 100644 index 0000000..0ac3377 --- /dev/null +++ b/services/orchestrator/observability.mjs @@ -0,0 +1,254 @@ +import { createRemoteWorkflowEngine } from './engine-registry.mjs'; + +const SHADOW_EVENT_TYPES = Object.freeze([ + 'workflow_shadow_completed', + 'workflow_shadow_failed', +]); +const MAX_METRIC_EVENTS = 5000; + +function clampInteger(value, fallback, min, max) { + const parsed = Number(value); + if (!Number.isFinite(parsed)) return fallback; + return Math.min(max, Math.max(min, Math.floor(parsed))); +} + +function parseJsonColumn(value, fallback = null) { + if (value == null || value === '') return fallback; + if (typeof value === 'object') return value; + try { + return JSON.parse(String(value)); + } catch { + return fallback; + } +} + +function percentile(values, ratio) { + if (!values.length) return null; + const sorted = [...values].sort((left, right) => left - right); + const index = Math.max(0, Math.ceil(sorted.length * ratio) - 1); + return sorted[index]; +} + +function normalizeRunId(value) { + const runId = String(value ?? '').trim(); + return /^[a-zA-Z0-9_-]{1,128}$/.test(runId) ? runId : ''; +} + +function projectShadowEventRow(row) { + const data = parseJsonColumn(row.data_json, {}) ?? {}; + const succeeded = row.event_type === 'workflow_shadow_completed'; + const latency = Number(data.latencyMs); + return { + eventId: row.event_id, + runId: row.run_id, + requestId: row.request_id, + userId: row.user_id, + sessionId: row.agent_session_id ?? null, + nativeStatus: row.native_status, + nativeAttempts: Number(row.native_attempts ?? 0), + shadowStatus: succeeded ? 'succeeded' : 'failed', + engine: data.engine ?? 'langgraph', + configVersion: data.configVersion == null ? null : Number(data.configVersion), + phase: data.phase ?? null, + taskType: data.taskType ?? null, + executorAdapter: data.executorAdapter ?? null, + latencyMs: Number.isFinite(latency) && latency >= 0 ? latency : null, + error: succeeded ? null : { + code: data.code ?? 'WORKFLOW_SHADOW_FAILED', + message: data.message ?? 'Shadow observation failed', + }, + observedAt: Number(row.event_created_at ?? 0), + nativeCompletedAt: row.native_completed_at == null + ? null + : Number(row.native_completed_at), + }; +} + +function summarizeRuns(runs, { capped = false } = {}) { + const successes = runs.filter((run) => run.shadowStatus === 'succeeded').length; + const failures = runs.length - successes; + const latencies = runs + .map((run) => run.latencyMs) + .filter((value) => Number.isFinite(value)); + return { + observations: runs.length, + successes, + failures, + successRate: runs.length ? successes / runs.length : null, + failureRate: runs.length ? failures / runs.length : null, + latencyP50Ms: percentile(latencies, 0.5), + latencyP95Ms: percentile(latencies, 0.95), + nativeSucceeded: runs.filter((run) => run.nativeStatus === 'succeeded').length, + nativeFailed: runs.filter((run) => run.nativeStatus === 'failed').length, + lastObservedAt: runs[0]?.observedAt ?? null, + sampled: capped, + }; +} + +function safeRemoteError(error) { + return { + code: String(error?.code ?? 'ORCHESTRATOR_UNAVAILABLE').slice(0, 128), + message: String(error instanceof Error ? error.message : error).slice(0, 1000), + }; +} + +export function createOrchestratorObservabilityService({ + pool, + configService, + serviceToken = process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN, + fetchImpl = globalThis.fetch, +} = {}) { + if (!pool?.query) throw new Error('Orchestrator observability requires a database pool'); + if (!configService?.getRuntimeState) { + throw new Error('Orchestrator observability requires config service'); + } + + async function loadShadowEvents({ hours = 24 } = {}) { + const normalizedHours = clampInteger(hours, 24, 1, 24 * 30); + const from = Date.now() - normalizedHours * 60 * 60 * 1000; + const [rows] = await pool.query( + `SELECT + e.id AS event_id, + e.run_id, + e.event_type, + e.data_json, + e.created_at AS event_created_at, + r.request_id, + r.user_id, + r.agent_session_id, + r.status AS native_status, + r.attempts AS native_attempts, + 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 (?, ?) + AND e.created_at >= ? + ORDER BY e.created_at DESC, e.id DESC + LIMIT ?`, + [...SHADOW_EVENT_TYPES, from, MAX_METRIC_EVENTS], + ); + return { + from, + hours: normalizedHours, + capped: rows.length >= MAX_METRIC_EVENTS, + runs: rows.map(projectShadowEventRow), + }; + } + + return { + async listShadowRuns({ + hours = 24, + limit = 50, + status = 'all', + } = {}) { + const loaded = await loadShadowEvents({ hours }); + const normalizedLimit = clampInteger(limit, 50, 1, 200); + const normalizedStatus = ['succeeded', 'failed'].includes(status) ? status : 'all'; + const filtered = normalizedStatus === 'all' + ? loaded.runs + : loaded.runs.filter((run) => run.shadowStatus === normalizedStatus); + return { + generatedAt: Date.now(), + window: { + hours: loaded.hours, + from: loaded.from, + }, + metrics: summarizeRuns(loaded.runs, { capped: loaded.capped }), + runs: filtered.slice(0, normalizedLimit), + }; + }, + + async getShadowRun(runId) { + const normalizedRunId = normalizeRunId(runId); + if (!normalizedRunId) return null; + const [runRows] = await pool.query( + `SELECT id, request_id, user_id, agent_session_id, status, attempts, + error_message, created_at, updated_at, started_at, completed_at + FROM h5_agent_runs + WHERE id = ? + LIMIT 1`, + [normalizedRunId], + ); + if (!runRows[0]) return null; + 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 (?, ?) + ORDER BY created_at ASC, id ASC`, + [normalizedRunId, ...SHADOW_EVENT_TYPES], + ); + const native = runRows[0]; + const localEvents = eventRows.map((row) => ({ + eventId: row.event_id, + type: row.event_type, + data: parseJsonColumn(row.data_json, null), + createdAt: Number(row.created_at), + })); + + let remote = { + available: false, + state: null, + events: [], + error: null, + }; + try { + const runtimeState = await configService.getRuntimeState(); + if (runtimeState.config.serviceUrl) { + const engine = createRemoteWorkflowEngine({ + baseUrl: runtimeState.config.serviceUrl, + serviceToken, + timeoutMs: runtimeState.config.requestTimeoutMs, + fetchImpl, + }); + const [state, events] = await Promise.all([ + engine.getState(normalizedRunId), + (async () => { + const collected = []; + for await (const event of engine.streamEvents(normalizedRunId)) { + collected.push(event); + if (collected.length >= 500) break; + } + return collected; + })(), + ]); + remote = { + available: true, + state, + events, + error: null, + }; + } + } catch (error) { + remote.error = safeRemoteError(error); + } + + return { + native: { + runId: native.id, + requestId: native.request_id, + userId: native.user_id, + sessionId: native.agent_session_id ?? null, + status: native.status, + attempts: Number(native.attempts ?? 0), + error: native.error_message ?? null, + createdAt: Number(native.created_at ?? 0), + updatedAt: Number(native.updated_at ?? 0), + startedAt: native.started_at == null ? null : Number(native.started_at), + completedAt: native.completed_at == null ? null : Number(native.completed_at), + }, + localEvents, + remote, + }; + }, + }; +} + +export const orchestratorObservabilityInternals = { + MAX_METRIC_EVENTS, + SHADOW_EVENT_TYPES, + normalizeRunId, + parseJsonColumn, + percentile, + projectShadowEventRow, + summarizeRuns, +}; diff --git a/services/orchestrator/observability.test.mjs b/services/orchestrator/observability.test.mjs new file mode 100644 index 0000000..5f4c99f --- /dev/null +++ b/services/orchestrator/observability.test.mjs @@ -0,0 +1,150 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { createOrchestratorObservabilityService } from './observability.mjs'; + +function response(body, status = 200) { + return new Response(JSON.stringify(body), { + status, + headers: { 'content-type': 'application/json' }, + }); +} + +test('observability aggregates shadow outcomes and latency percentiles', async () => { + const rows = [ + { + event_id: 'event-3', + run_id: 'run-3', + event_type: 'workflow_shadow_completed', + data_json: { latencyMs: 300, engine: 'langgraph', taskType: 'refactor' }, + event_created_at: 3000, + request_id: 'request-3', + user_id: 'user-3', + native_status: 'succeeded', + native_attempts: 1, + native_completed_at: 3100, + }, + { + event_id: 'event-2', + run_id: 'run-2', + event_type: 'workflow_shadow_failed', + data_json: JSON.stringify({ latencyMs: 200, code: 'TIMEOUT', message: 'timed out' }), + event_created_at: 2000, + request_id: 'request-2', + user_id: 'user-2', + native_status: 'succeeded', + native_attempts: 1, + native_completed_at: 2200, + }, + { + event_id: 'event-1', + run_id: 'run-1', + event_type: 'workflow_shadow_completed', + data_json: { latencyMs: 100, executorAdapter: 'native-agent-run' }, + event_created_at: 1000, + request_id: 'request-1', + user_id: 'user-1', + native_status: 'failed', + native_attempts: 2, + native_completed_at: 1500, + }, + ]; + const service = createOrchestratorObservabilityService({ + pool: { + async query(sql) { + assert.match(sql, /INNER JOIN h5_agent_runs/); + return [rows]; + }, + }, + configService: { getRuntimeState() {} }, + }); + + const result = await service.listShadowRuns({ hours: 12, limit: 2 }); + assert.equal(result.window.hours, 12); + assert.equal(result.metrics.observations, 3); + assert.equal(result.metrics.successes, 2); + assert.equal(result.metrics.failures, 1); + 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.runs[1].error.code, 'TIMEOUT'); +}); + +test('observability detail joins Native state with remote LangGraph checkpoint events', async () => { + const queries = []; + const service = createOrchestratorObservabilityService({ + pool: { + async query(sql, params) { + queries.push({ sql, params }); + if (sql.includes('FROM h5_agent_runs')) { + return [[{ + id: 'run-detail', + request_id: 'request-detail', + user_id: 'user-detail', + agent_session_id: 'session-detail', + status: 'succeeded', + attempts: 1, + error_message: null, + created_at: 100, + updated_at: 300, + started_at: 150, + completed_at: 300, + }]]; + } + return [[{ + event_id: 'local-event', + event_type: 'workflow_shadow_completed', + data_json: { latencyMs: 25 }, + created_at: 250, + }]]; + }, + }, + configService: { + async getRuntimeState() { + return { + config: { + serviceUrl: 'http://orchestrator.internal:8093', + requestTimeoutMs: 1000, + }, + }; + }, + }, + serviceToken: 'detail-token', + fetchImpl: async (url, init) => { + assert.equal(init.headers.authorization, 'Bearer detail-token'); + if (String(url).endsWith('/events')) { + return response({ + events: [{ sequence: 1, type: 'workflow_validated' }], + }); + } + return response({ + runId: 'run-detail', + status: 'succeeded', + plan: { executorAdapter: 'native-agent-run' }, + }); + }, + }); + + const detail = await service.getShadowRun('run-detail'); + assert.equal(detail.native.status, 'succeeded'); + assert.equal(detail.localEvents[0].data.latencyMs, 25); + assert.equal(detail.remote.available, true); + assert.equal(detail.remote.state.plan.executorAdapter, 'native-agent-run'); + assert.equal(detail.remote.events[0].type, 'workflow_validated'); + assert.equal(queries.length, 2); +}); + +test('observability rejects unsafe run ids before querying', async () => { + let queries = 0; + const service = createOrchestratorObservabilityService({ + pool: { + async query() { + queries += 1; + return [[]]; + }, + }, + configService: { getRuntimeState() {} }, + }); + assert.equal(await service.getShadowRun('../unsafe'), null); + assert.equal(queries, 0); +});