From 2596c8f2c6dde320bdb89b80eae4fc8e85919423 Mon Sep 17 00:00:00 2001 From: john Date: Sat, 25 Jul 2026 07:12:53 +0800 Subject: [PATCH] feat: add orchestrator controls to memindadm --- server/app.mjs | 67 +++++ server/bootstrap.mjs | 16 ++ server/index.mjs | 4 + src/App.tsx | 3 + src/admin/AdminNav.tsx | 1 + src/admin/pages/OrchestratorPage.tsx | 383 +++++++++++++++++++++++++++ src/api/client.ts | 66 +++++ src/index.css | 206 ++++++++++++++ src/types/orchestrator.ts | 237 +++++++++++++++++ 9 files changed, 983 insertions(+) create mode 100644 src/admin/pages/OrchestratorPage.tsx create mode 100644 src/types/orchestrator.ts diff --git a/server/app.mjs b/server/app.mjs index 2169717..24d9281 100644 --- a/server/app.mjs +++ b/server/app.mjs @@ -100,6 +100,8 @@ export function createAdminApp(services) { updateMindSpaceConfig, memoryV2ConfigService, mindSearchConfigService, + orchestratorConfigService, + orchestratorObservabilityService, personalMemoryCandidateStore, skillRuntimeConfigService, wechatScheduleLlmConfigService, @@ -426,6 +428,71 @@ export function createAdminApp(services) { } }); + adminApi.get('/orchestrator/config', requireAdmin, async (_req, res) => { + if (!orchestratorConfigService?.getAdminConfig) { + return res.status(503).json({ message: 'Orchestrator 配置服务未启用' }); + } + return res.json(await orchestratorConfigService.getAdminConfig()); + }); + + const updateOrchestratorConfig = async (req, res) => { + if (!orchestratorConfigService?.updateAdminConfig) { + return res.status(503).json({ message: 'Orchestrator 配置服务未启用' }); + } + return res.json(await orchestratorConfigService.updateAdminConfig( + req.body?.config ?? req.body ?? {}, + { updatedBy: req.currentUser.id }, + )); + }; + + adminApi.put('/orchestrator/config', requireAdmin, updateOrchestratorConfig); + adminApi.patch('/orchestrator/config', requireAdmin, updateOrchestratorConfig); + + adminApi.get('/orchestrator/runtime', requireAdmin, async (_req, res) => { + if (!orchestratorConfigService?.getRuntimeState) { + return res.status(503).json({ message: 'Orchestrator 配置服务未启用' }); + } + return res.json(await orchestratorConfigService.getRuntimeState({ probe: true })); + }); + + adminApi.get('/orchestrator/shadow-runs', requireAdmin, async (req, res) => { + if (!orchestratorObservabilityService?.listShadowRuns) { + return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); + } + return res.json(await orchestratorObservabilityService.listShadowRuns({ + hours: req.query.hours, + limit: req.query.limit, + status: req.query.status, + })); + }); + + adminApi.get('/orchestrator/execution-plans', requireAdmin, async (req, res) => { + if (!orchestratorObservabilityService?.listExecutionPlans) { + return res.status(503).json({ message: 'Orchestrator Dry-run 观测服务未启用' }); + } + return res.json(await orchestratorObservabilityService.listExecutionPlans({ + hours: req.query.hours, + limit: req.query.limit, + selection: req.query.selection, + })); + }); + + adminApi.get('/orchestrator/canary-readiness', requireAdmin, async (_req, res) => { + if (!orchestratorObservabilityService?.getCanaryReadiness) { + return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); + } + return res.json(await orchestratorObservabilityService.getCanaryReadiness()); + }); + + adminApi.get('/orchestrator/shadow-runs/:runId', requireAdmin, async (req, res) => { + if (!orchestratorObservabilityService?.getShadowRun) { + return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); + } + const result = await orchestratorObservabilityService.getShadowRun(req.params.runId); + if (!result) return res.status(404).json({ message: 'Shadow run 不存在' }); + return res.json(result); + }); + adminApi.get('/memory-v2/status', requireAdmin, async (_req, res) => { const portalBaseUrl = `http://127.0.0.1:${process.env.H5_PORT ?? 8081}`; try { diff --git a/server/bootstrap.mjs b/server/bootstrap.mjs index 7f259e3..bd74dcc 100644 --- a/server/bootstrap.mjs +++ b/server/bootstrap.mjs @@ -31,6 +31,12 @@ export async function bootstrapAdminServices() { } = await importMemind('mindspace-config.mjs'); const { createMemoryV2AdminConfigService } = await importMemind('memory-v2-admin-config.mjs'); const { createMindSearchConfigService } = await importMemind('mindsearch-config.mjs'); + const { createOrchestratorAdminConfigService } = await importMemind( + 'services/orchestrator/admin-config.mjs', + ); + const { createOrchestratorObservabilityService } = await importMemind( + 'services/orchestrator/observability.mjs', + ); const { createPersonalMemoryCandidateStore } = await importMemind('memory-v2-personal-store.mjs'); const { createSkillRuntimeAdminConfigService } = await importMemind('skill-runtime-admin-config.mjs'); const { createAssetGatewayConfigService } = await importMemind('asset-gateway.mjs'); @@ -99,6 +105,14 @@ export async function bootstrapAdminServices() { }); const mindSearchConfigService = createMindSearchConfigService(pool, { env: process.env }); await mindSearchConfigService.ensureSchema(); + const orchestratorConfigService = createOrchestratorAdminConfigService(pool, { + env: process.env, + }); + await orchestratorConfigService.ensureSchema(); + const orchestratorObservabilityService = createOrchestratorObservabilityService({ + pool, + configService: orchestratorConfigService, + }); const personalMemoryCandidateStore = createPersonalMemoryCandidateStore(pool); const skillRuntimeConfigService = createSkillRuntimeAdminConfigService(pool, { env: process.env, @@ -165,6 +179,8 @@ export async function bootstrapAdminServices() { updateMindSpaceConfig, memoryV2ConfigService, mindSearchConfigService, + orchestratorConfigService, + orchestratorObservabilityService, personalMemoryCandidateStore, skillRuntimeConfigService, wechatScheduleLlmConfigService, diff --git a/server/index.mjs b/server/index.mjs index 9f314d2..b4e3868 100644 --- a/server/index.mjs +++ b/server/index.mjs @@ -106,6 +106,8 @@ ready updateMindSpaceConfig, memoryV2ConfigService, mindSearchConfigService, + orchestratorConfigService, + orchestratorObservabilityService, personalMemoryCandidateStore, skillRuntimeConfigService, wechatScheduleLlmConfigService, @@ -132,6 +134,8 @@ ready updateMindSpaceConfig, memoryV2ConfigService, mindSearchConfigService, + orchestratorConfigService, + orchestratorObservabilityService, personalMemoryCandidateStore, skillRuntimeConfigService, wechatScheduleLlmConfigService, diff --git a/src/App.tsx b/src/App.tsx index 3b09e1e..188ee53 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -19,6 +19,7 @@ import { AssetGatewayPage } from './admin/pages/AssetGatewayPage'; import { BlockedWordsPage } from './admin/pages/BlockedWordsPage'; import { AnalyticsConfigPage } from './admin/pages/AnalyticsConfigPage'; import { SkillRuntimePage } from './admin/pages/SkillRuntimePage'; +import { OrchestratorPage } from './admin/pages/OrchestratorPage'; import { defaultHomePath } from './lib/routes'; import { OpsLayout } from './ops/components/OpsLayout'; import { AnalyticsPage } from './ops/pages/AnalyticsPage'; @@ -127,6 +128,7 @@ function AdminApp({ user, onLogout }: { user: PortalUser; onLogout: () => void } } /> } /> } /> + } /> } /> } /> } /> @@ -175,6 +177,7 @@ function loginRedirectPath(pathname: string, role: string | undefined) { || pathname.startsWith('/mindspace') || pathname.startsWith('/memory-v2') || pathname.startsWith('/skill-runtime') + || pathname.startsWith('/orchestrator') || pathname.startsWith('/providers') || pathname.startsWith('/wechat') || pathname.startsWith('/asset-gateway') diff --git a/src/admin/AdminNav.tsx b/src/admin/AdminNav.tsx index 122bf45..3b15503 100644 --- a/src/admin/AdminNav.tsx +++ b/src/admin/AdminNav.tsx @@ -32,6 +32,7 @@ const NAV_SECTIONS: NavSection[] = [ { to: '/memory-v2', label: 'Memory V2' }, { to: '/mindsearch', label: 'MindSearch' }, { to: '/skill-runtime', label: 'Skill Runtime' }, + { to: '/orchestrator', label: '任务编排' }, { to: '/system-tests', label: '系统测试验证' }, { to: '/capabilities', label: '能力' }, { to: '/skills', label: '技能' }, diff --git a/src/admin/pages/OrchestratorPage.tsx b/src/admin/pages/OrchestratorPage.tsx new file mode 100644 index 0000000..7cf9476 --- /dev/null +++ b/src/admin/pages/OrchestratorPage.tsx @@ -0,0 +1,383 @@ +import { useEffect, useMemo, useState } from 'react'; +import { + getOrchestratorCanaryReadiness, + getOrchestratorConfig, + getOrchestratorExecutionPlans, + getOrchestratorRuntime, + getOrchestratorShadowRun, + getOrchestratorShadowRuns, + updateOrchestratorConfig, +} from '../../api/client'; +import type { + OrchestratorCanaryReadiness, + OrchestratorConfig, + OrchestratorConfigState, + OrchestratorExecutionPlanList, + OrchestratorServiceHealth, + OrchestratorShadowRunDetail, + OrchestratorShadowRunList, +} from '../../types/orchestrator'; + +const runtimeReasons: Record = { + mode_off: '当前为关闭模式，所有任务继续由 Native Agent Run 执行。', + service_url_missing: '尚未配置 Orchestrator 服务地址，LangGraph 不会接管任务。', + kill_switch: '环境级紧急熔断已开启，强制回退 Native。', + execution_gate_disabled: '执行交接闸门被硬锁定，当前仅生成 Dry-run 决策。', + environment_execution_gate_disabled: 'memindadm 已请求交接，但环境闸门仍关闭。', +}; + +const readinessLabels: Record = { + shadow_mode: '保持 Shadow', + 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 终态覆盖率', + session_coverage: '真实会话覆盖', + sample_freshness: '最近样本新鲜度', + complete_window: '统计窗口完整', +}; + +function listToText(values: string[]) { + return values.join('\n'); +} + +function textToList(value: string) { + return [...new Set(value.split(/\r?\n|,/).map((item) => item.trim()).filter(Boolean))]; +} + +function formatTime(value?: number | null) { + 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; +} + +function readinessValue(check: OrchestratorCanaryReadiness['checks'][number]) { + if (check.actual == null) return '无数据'; + if (['shadow_success_rate', 'latency_coverage', 'native_settled_rate'].includes(check.id)) { + return formatPercent(Number(check.actual)); + } + if (check.id === 'latency_p95') return formatLatency(Number(check.actual)); + if (check.id === 'sample_freshness') return `${Number(check.actual).toFixed(1)} 小时`; + if (typeof check.actual === 'boolean') return check.actual ? '是' : '否'; + return String(check.actual); +} + +export function OrchestratorPage() { + const [state, setState] = useState(null); + const [draft, setDraft] = useState(null); + const [workflowAllowlist, setWorkflowAllowlist] = useState(''); + const [userAllowlist, setUserAllowlist] = useState(''); + const [shadow, setShadow] = useState(null); + const [plans, setPlans] = useState(null); + const [readiness, setReadiness] = useState(null); + const [health, setHealth] = useState(null); + const [detail, setDetail] = useState(null); + const [loading, setLoading] = useState(true); + const [saving, setSaving] = useState(false); + const [checking, setChecking] = useState(false); + const [error, setError] = useState(null); + const [message, setMessage] = useState(null); + + const load = async () => { + setLoading(true); + setError(null); + try { + const [configState, shadowState, planState, readinessState] = await Promise.all([ + getOrchestratorConfig(), + getOrchestratorShadowRuns({ hours: 24, limit: 100 }), + getOrchestratorExecutionPlans({ hours: 24, limit: 100 }), + getOrchestratorCanaryReadiness(), + ]); + setState(configState); + setDraft(configState.config); + setWorkflowAllowlist(listToText(configState.config.workflowAllowlist)); + setUserAllowlist(listToText(configState.config.userAllowlist)); + setShadow(shadowState); + setPlans(planState); + setReadiness(readinessState); + } catch (err) { + setError(err instanceof Error ? err.message : 'Orchestrator 数据加载失败'); + } finally { + setLoading(false); + } + }; + + useEffect(() => { + void load(); + }, []); + + const normalizedDraft = useMemo(() => draft ? { + ...draft, + workflowAllowlist: textToList(workflowAllowlist), + userAllowlist: textToList(userAllowlist), + } : null, [draft, workflowAllowlist, userAllowlist]); + + const dirty = Boolean( + state && normalizedDraft && JSON.stringify(state.config) !== JSON.stringify(normalizedDraft), + ); + + const save = async () => { + if (!normalizedDraft) return; + if (['canary', 'active'].includes(normalizedDraft.mode) && !normalizedDraft.serviceUrl) { + setError('Canary / Active 模式必须先配置 Orchestrator URL。'); + return; + } + if ( + normalizedDraft.executionEnabled + && !window.confirm('确认请求执行交接？环境闸门关闭时仍只会 Dry-run。') + ) { + return; + } + setSaving(true); + setError(null); + setMessage(null); + try { + const result = await updateOrchestratorConfig(normalizedDraft); + setState(result); + setDraft(result.config); + setWorkflowAllowlist(listToText(result.config.workflowAllowlist)); + setUserAllowlist(listToText(result.config.userAllowlist)); + setMessage(`Orchestrator 配置 v${result.configVersion} 已保存。`); + } catch (err) { + setError(err instanceof Error ? err.message : '保存失败'); + } finally { + setSaving(false); + } + }; + + const checkHealth = async () => { + setChecking(true); + setError(null); + try { + const result = await getOrchestratorRuntime(); + setHealth(result.serviceHealth); + } catch (err) { + setError(err instanceof Error ? err.message : '服务检查失败'); + } finally { + setChecking(false); + } + }; + + const openDetail = async (runId: string) => { + setError(null); + try { + setDetail(await getOrchestratorShadowRun(runId)); + } catch (err) { + setError(err instanceof Error ? err.message : 'Run 详情加载失败'); + } + }; + + if (loading) return

正在加载 Workflow Orchestrator…

; + if (!state || !draft) { + return

{error ?? '配置不可用'}

; + } + + const runtimeMessage = state.runtime.reason + ? (runtimeReasons[state.runtime.reason] ?? state.runtime.reason) + : state.runtime.shadowsLangGraph + ? 'Shadow 已生效：Native 继续执行，LangGraph 只观察。' + : state.runtime.executesLangGraph + ? '执行交接已开启。' + : '配置有效。'; + + return ( +
+
+

Workflow Orchestrator

+

+ LangGraph 以可插拔服务接入；Off 与 Shadow 不改变现有 Native Agent Run 执行路径。 +

+
+ + {error &&

{error}

} + {message &&

{message}

} + +
+
+
配置版本v{state.configVersion}
+
运行状态{state.runtime.effective ? '有效' : '安全回退'}
+
配置来源{state.source}
+
执行交接{state.runtime.executionHandoff.enabled ? '已开启' : 'Dry-run 锁定'}
+
更新时间{formatTime(state.updatedAt)}
+
+

+ {runtimeMessage} +

+
+ +
+
+
+

Executor Gateway

+

执行器默认关闭，最终状态以 Orchestrator 健康检查为准。

+
+
+
+ {state.executors.map((executor) => ( +
+
{executor.label}{executor.status}
+

{executor.capabilities.join(' · ') || '无能力声明'}

+ dispatch {executor.dispatchImplemented ? 'implemented' : 'disabled'} +
+ ))} +
+
+ +
+
+
+

Shadow 运行观测

+

最近 24 小时 Native 与 LangGraph 观察结果，不改变执行权。

+
+ +
+ + {readiness && ( +
+
+ Canary 准入评估 + {readiness.ready ? '达到门槛，等待人工审批' : '继续保持 Shadow'} +
+

+ 有效样本 {readiness.samples.eligibleObservations} / {readiness.thresholds.minObservations} + {' · '}排除 smoke {readiness.samples.excludedSynthetic} + {' · '}真实会话 {readiness.samples.distinctSessions} / {readiness.thresholds.minDistinctSessions} +

+
+ {readiness.checks.map((check) => ( + + {check.passed ? '通过' : '阻塞'} · {readinessLabels[check.id] ?? check.id}：{readinessValue(check)} + + ))} +
+
+ )} + +
+
观察请求{shadow?.metrics.observations ?? '—'}
+
成功率{formatPercent(shadow?.metrics.successRate)}
+
失败数{shadow?.metrics.failures ?? '—'}
+
延迟 P95{formatLatency(shadow?.metrics.latencyP95Ms)}
+
Native 成功 / 失败{shadow ? `${shadow.metrics.nativeSucceeded} / ${shadow.metrics.nativeFailed}` : '—'}
+
+ +
+ + + + {(shadow?.runs ?? []).map((run) => ( + + + + + + + + + ))} + {!shadow?.runs.length && } + +
时间RunShadowNative延迟任务 / Adapter
{formatTime(run.observedAt)}{run.shadowStatus}{run.nativeStatus}{formatLatency(run.latencyMs)}{run.taskType ?? '—'}{run.synthetic ? '（smoke）' : ''}{run.executorAdapter ?? '—'}
当前没有 Shadow 结果。
+
+ + {detail && ( +
+
+
Run 详情：{detail.native.runId}

Native {detail.native.status} · LangGraph {detail.remote.available ? 'available' : 'unavailable'}

+ +
+
+ 
{JSON.stringify(detail.remote.state, null, 2)}
+ 
{JSON.stringify(detail.remote.events, null, 2)}
+
+
+ )} +
+ +
+
+

执行路由 Dry-run

候选路由仅观测，实际执行引擎必须保持 Native。

+
+ {(plans?.metrics.handoffAllowed ?? 0) > 0 && ( +

检测到 handoffAllowed=true，请立即检查执行闸门。

+ )} +
+
路由决策{plans?.metrics.decisions ?? '—'}
+
候选命中率{formatPercent(plans?.metrics.candidateSelectionRate)}
+
LangGraph / Native{plans ? `${plans.metrics.candidateSelections} / ${plans.metrics.nativeSelections}` : '—'}
+
Native 终态覆盖{formatPercent(plans?.metrics.nativeSettledRate)}
+
真实会话{plans?.metrics.distinctSessions ?? '—'}
+
+
+ + + + {(plans?.plans ?? []).map((plan) => ( + + + + + + + + + ))} + {!plans?.plans.length && } + +
时间Run模式候选 → 实际原因Native
{formatTime(plan.plannedAt)}{shortId(plan.runId)}{plan.mode ?? '—'}{plan.candidateEngine} → {plan.effectiveEngine}{plan.candidateReason ?? plan.reason ?? '—'}{plan.nativeStatus}
当前没有 Dry-run 路由决策。
+
+
+ +
+

路由与服务配置

+
+ + + + + +
+
+ {health ? `${health.ok ? '健康' : '不可用'}（${health.status}，${health.latencyMs} ms）` : '尚未检查服务连接'} + +
+
+ + + +
+
+