diff --git a/admin-routes.mjs b/admin-routes.mjs index bcb03dc..07462f6 100644 --- a/admin-routes.mjs +++ b/admin-routes.mjs @@ -252,7 +252,7 @@ export function createAdminApi({ if (!orchestratorConfigService?.getRuntimeState) { return res.status(503).json({ message: 'Orchestrator 配置服务未启用' }); } - return res.json(await orchestratorConfigService.getRuntimeState()); + return res.json(await orchestratorConfigService.getRuntimeState({ probe: true })); }); adminApi.get('/skill-runtime/config', requireAdmin, async (_req, res) => { diff --git a/admin-routes.test.mjs b/admin-routes.test.mjs index cdb2615..c1a5d40 100644 --- a/admin-routes.test.mjs +++ b/admin-routes.test.mjs @@ -129,8 +129,13 @@ test('admin orchestrator routes expose a versioned plug-in control plane', async updates.push({ config, context }); return { ...state, config: { ...state.config, ...config }, configVersion: 2 }; }, - async getRuntimeState() { - return { ...state, source: 'admin-db' }; + async getRuntimeState(options) { + assert.deepEqual(options, { probe: true }); + return { + ...state, + source: 'admin-db', + serviceHealth: { ok: true, status: 'healthy' }, + }; }, }, plazaPosts: null, @@ -166,7 +171,9 @@ test('admin orchestrator routes expose a versioned plug-in control plane', async headers: { cookie: 'h5_user_session=token-admin' }, }); assert.equal(runtimeResponse.status, 200); - assert.equal((await runtimeResponse.json()).source, 'admin-db'); + const runtimeBody = await runtimeResponse.json(); + assert.equal(runtimeBody.source, 'admin-db'); + assert.equal(runtimeBody.serviceHealth.status, 'healthy'); } finally { await server.close(); } diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index f465954..6426971 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -289,6 +289,7 @@ export function createAgentRunGateway({ conversationMemoryService = null, syncUserPagesOnSuccess = null, observePersonalMemoryOnSuccess = null, + observeWorkflowRun = null, isSessionExternallyBusy = null, validateRunDeliverables = null, retryDelaysMs = DEFAULT_RUN_RETRY_DELAYS_MS, @@ -338,6 +339,39 @@ export function createAgentRunGateway({ ); } + function dispatchShadowObservation(input) { + if (typeof observeWorkflowRun !== 'function' || input.toolMode !== 'code') return; + setImmediate(() => { + void Promise.resolve() + .then(() => observeWorkflowRun(input)) + .then(async (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, + }); + }) + .catch(async (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), + }).catch((appendError) => { + console.warn( + '[AgentRun] workflow shadow failure event skipped:', + appendError instanceof Error ? appendError.message : appendError, + ); + }); + }); + }); + } + async function appendRunSnapshot(runId) { if (!isRunStreamReplayEnabled()) return; const row = await getRunById(runId); @@ -443,8 +477,19 @@ export function createAgentRunGateway({ taskType: normalizedTaskType, forceDeepReasoning: Boolean(forceDeepReasoning), }); + const createdRun = projectRun(await getRunById(runId)); + dispatchShadowObservation({ + runId, + requestId: normalizedRequestId, + userId, + sessionId: sessionId || null, + workflowName: 'code-run-v1', + userMessage: runMessage, + toolMode: normalizedToolMode, + taskType: normalizedTaskType, + }); if (autoDispatch) dispatchRun(runId); - return projectRun(await getRunById(runId)); + return createdRun; } async function prepareRunDeliverables({ userId, sessionId, runId, runStartedAtMs = null }) { diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 6ec4ad4..33b28ab 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -374,6 +374,98 @@ test('agent run creation stores code tool metadata in the queued message', async }); }); +test('code run shadow observation is fire-and-forget and records completion separately', async () => { + const pool = createFakePool(); + let releaseObservation; + let observedInput = null; + const observation = new Promise((resolve) => { + releaseObservation = resolve; + }); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: {}, + autoDispatch: false, + observeWorkflowRun: async (input) => { + observedInput = input; + return observation; + }, + }); + + const run = await gateway.createRun('user-shadow', { + requestId: 'req-shadow', + sessionId: 'session-shadow', + userMessage: { role: 'user', content: [{ type: 'text', text: 'refactor this' }] }, + toolMode: 'code', + taskType: 'repo_refactor', + }); + + assert.equal(run.status, 'queued'); + await waitFor(() => observedInput != null); + assert.equal(observedInput.runId, run.id); + assert.equal(observedInput.workflowName, 'code-run-v1'); + assert.equal( + pool.events.some((event) => event.eventType === 'workflow_shadow_completed'), + false, + ); + + releaseObservation({ + observed: true, + engine: 'langgraph', + mode: 'shadow', + configVersion: 3, + shadowRun: { status: 'succeeded', phase: 'completed' }, + }); + await waitFor(() => pool.events.some( + (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', + }); +}); + +test('shadow observer failure cannot fail a Native run and chat runs are not observed', async () => { + const pool = createFakePool(); + let observations = 0; + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: {}, + autoDispatch: false, + observeWorkflowRun: async () => { + observations += 1; + const error = new Error('shadow unavailable'); + error.code = 'SHADOW_UNAVAILABLE'; + throw error; + }, + }); + + const codeRun = await gateway.createRun('user-shadow', { + requestId: 'req-shadow-failure', + userMessage: { role: 'user', content: 'change code' }, + toolMode: 'code', + }); + const chatRun = await gateway.createRun('user-shadow', { + requestId: 'req-chat-no-shadow', + userMessage: { role: 'user', content: 'hello' }, + toolMode: 'chat', + }); + + assert.equal(codeRun.status, 'queued'); + assert.equal(chatRun.status, 'queued'); + await waitFor(() => pool.events.some( + (event) => event.eventType === 'workflow_shadow_failed', + )); + assert.equal(observations, 1); + const failure = pool.events.find((event) => event.eventType === 'workflow_shadow_failed'); + assert.equal(JSON.parse(failure.dataJson).code, 'SHADOW_UNAVAILABLE'); +}); + test('agent run starts a session and marks submitted reply as succeeded', async () => { const pool = createFakePool(); const submitted = []; diff --git a/deploy/orchestrator/.env.example b/deploy/orchestrator/.env.example new file mode 100644 index 0000000..2f3fd24 --- /dev/null +++ b/deploy/orchestrator/.env.example @@ -0,0 +1,3 @@ +ORCHESTRATOR_POSTGRES_PASSWORD=replace-with-a-long-random-password +MEMIND_ORCHESTRATOR_SERVICE_TOKEN=replace-with-a-long-random-service-token +MEMIND_ORCHESTRATOR_PORT=8093 diff --git a/deploy/orchestrator/compose.yaml b/deploy/orchestrator/compose.yaml new file mode 100644 index 0000000..90be708 --- /dev/null +++ b/deploy/orchestrator/compose.yaml @@ -0,0 +1,46 @@ +services: + postgres: + image: postgres:17-alpine + restart: unless-stopped + environment: + POSTGRES_DB: memind_orchestrator + POSTGRES_USER: memind_orchestrator + POSTGRES_PASSWORD: ${ORCHESTRATOR_POSTGRES_PASSWORD:?set ORCHESTRATOR_POSTGRES_PASSWORD} + healthcheck: + test: ["CMD-SHELL", "pg_isready -U memind_orchestrator -d memind_orchestrator"] + interval: 5s + timeout: 3s + retries: 20 + volumes: + - orchestrator_postgres_data:/var/lib/postgresql/data + networks: + - orchestrator_internal + + orchestrator: + build: + context: ../../services/orchestrator + dockerfile: Dockerfile + restart: unless-stopped + depends_on: + postgres: + condition: service_healthy + environment: + MEMIND_ORCHESTRATOR_HOST: 0.0.0.0 + MEMIND_ORCHESTRATOR_PORT: 8093 + MEMIND_ORCHESTRATOR_CHECKPOINT_MODE: postgres + MEMIND_ORCHESTRATOR_DATABASE_URL: postgresql://memind_orchestrator:${ORCHESTRATOR_POSTGRES_PASSWORD}@postgres:5432/memind_orchestrator + MEMIND_ORCHESTRATOR_DATABASE_SCHEMA: memind_orchestrator + MEMIND_ORCHESTRATOR_SERVICE_TOKEN: ${MEMIND_ORCHESTRATOR_SERVICE_TOKEN:?set MEMIND_ORCHESTRATOR_SERVICE_TOKEN} + ports: + - "127.0.0.1:${MEMIND_ORCHESTRATOR_PORT:-8093}:8093" + networks: + - orchestrator_internal + - orchestrator_edge + +networks: + orchestrator_internal: + internal: true + orchestrator_edge: + +volumes: + orchestrator_postgres_data: diff --git a/docs/architecture/memind-orchestrator-boundary.md b/docs/architecture/memind-orchestrator-boundary.md index 22ba45a..61818e1 100644 --- a/docs/architecture/memind-orchestrator-boundary.md +++ b/docs/architecture/memind-orchestrator-boundary.md @@ -54,6 +54,11 @@ Native selection. The initial workflow allowlist contains only `code-run-v1`. Ordinary chat and existing Goosed session traffic remain outside the Orchestrator path. +Phase 2 implements only the `off` and `shadow` execution semantics. Although +Canary and Active can be configured and evaluated by the control plane, the +Portal does not hand execution ownership to LangGraph yet. This prevents an +administrative configuration mistake from creating two task executors. + ## Protocol The framework-neutral contracts are: @@ -61,7 +66,7 @@ The framework-neutral contracts are: - `orchestrator-run-v1` - `orchestrator-event-v1` -The eventual internal API is: +The implemented internal API is: ```text POST /v1/runs @@ -75,13 +80,38 @@ External state changes must use an idempotency key derived from run, graph version, node, and attempt. Graph state stores resource references and decisions, not API keys, large logs, binary artifacts, or absolute production paths. +The Phase 2 graph has three deterministic nodes: + +```text +validate_run -> build_plan -> finalize_run +``` + +It requires `policy.executionMode=observe-only` and +`policy.sideEffectsAllowed=false`. It projects the Native executor boundary but +cannot call Goosed, Aider, OpenHands, Tool Gateway, or the filesystem. + +## Failure isolation + +Portal invokes Shadow only after the product run and its `queued` event have +been committed. The observation is scheduled without awaiting it. A timeout or +failure: + +1. cannot reject or delay `createRun`; +2. cannot mutate the Native run status; +3. is projected as `workflow_shadow_failed`; +4. remains removable by deleting the observer wiring and changing only the + service URL boundary. + +Successful observations are projected as `workflow_shadow_completed`; graph +checkpoints stay in the Orchestrator-owned PostgreSQL database. + ## Deployment evolution -1. Local native Orchestrator process with a development checkpoint database. -2. 103 shadow LaunchAgent; no production task claim. -3. User-allowlisted code-run canary with Native rollback. -4. Isolated Aider/OpenHands executor workers. -5. Containerized Orchestrator in its own Compose project. +1. Local native Orchestrator process with an explicit MemorySaver for debugging. +2. Colima Compose with an isolated PostgreSQL checkpoint database. +3. Approved server-side shadow service; no production task claim. +4. User-allowlisted code-run canary with Native rollback. +5. Isolated Aider/OpenHands executor workers. 6. Move the same service contract to Linux or Kubernetes when horizontal scaling or independent ownership becomes necessary. diff --git a/ops/src/api/admin.ts b/ops/src/api/admin.ts index 428386b..ae910ad 100644 --- a/ops/src/api/admin.ts +++ b/ops/src/api/admin.ts @@ -240,6 +240,23 @@ export type OrchestratorConfigState = { engines: OrchestratorEngineDescriptor[]; }; +export type OrchestratorServiceHealth = { + checkedAt: number; + ok: boolean; + status: 'healthy' | 'unhealthy' | 'unconfigured' | 'timeout' | 'unreachable' | 'fetch_unavailable'; + latencyMs: number; + httpStatus: number | null; + details: { + service: string | null; + checkpoint: { kind?: string; durable?: boolean } | null; + execution: string | null; + } | null; +}; + +export type OrchestratorRuntimeState = OrchestratorConfigState & { + serviceHealth: OrchestratorServiceHealth; +}; + // ─── Summary ────────────────────────────────────────────────────────────────── export async function fetchAdminSummary() { @@ -272,6 +289,10 @@ export async function updateOrchestratorConfig(config: OrchestratorConfig) { }); } +export async function fetchOrchestratorRuntime() { + return adminFetch('/admin-api/orchestrator/runtime'); +} + // ─── Users ──────────────────────────────────────────────────────────────────── export async function fetchAdminUsers(params: { diff --git a/ops/src/pages/admin/OrchestratorPage.tsx b/ops/src/pages/admin/OrchestratorPage.tsx index 04e1054..ed7f325 100644 --- a/ops/src/pages/admin/OrchestratorPage.tsx +++ b/ops/src/pages/admin/OrchestratorPage.tsx @@ -1,9 +1,11 @@ import { useEffect, useMemo, useState } from 'react'; import { fetchOrchestratorConfig, + fetchOrchestratorRuntime, updateOrchestratorConfig, type OrchestratorConfig, type OrchestratorConfigState, + type OrchestratorServiceHealth, } from '../../api/admin'; function listToText(values: string[]) { @@ -31,6 +33,8 @@ export function OrchestratorPage() { const [workflowAllowlist, setWorkflowAllowlist] = useState(''); const [loading, setLoading] = useState(true); const [saving, setSaving] = useState(false); + const [checking, setChecking] = useState(false); + const [serviceHealth, setServiceHealth] = useState(null); const [error, setError] = useState(null); const [notice, setNotice] = useState(null); @@ -54,6 +58,19 @@ export function OrchestratorPage() { void load(); }, []); + const checkService = async () => { + setChecking(true); + setError(null); + try { + const result = await fetchOrchestratorRuntime(); + setServiceHealth(result.serviceHealth); + } catch (err) { + setError(err instanceof Error ? err.message : '服务检查失败'); + } finally { + setChecking(false); + } + }; + const normalizedDraft = useMemo(() => draft ? { ...draft, userAllowlist: textToList(userAllowlist), @@ -75,7 +92,7 @@ export function OrchestratorPage() { } if ( normalizedDraft.mode === 'active' - && !window.confirm('确认启用 Active？工作流白名单内的任务将由 LangGraph Orchestrator 接管。') + && !window.confirm('确认保存 Active 预配置？Phase 2 不会交出执行权，Native 仍是唯一执行者。') ) { return; } @@ -104,7 +121,7 @@ export function OrchestratorPage() { : state.runtime.shadowsLangGraph ? 'Shadow 已生效：Native 继续执行，LangGraph 只观察和生成决策。' : state.runtime.executesLangGraph - ? 'LangGraph 路由已生效。' + ? 'Canary / Active 路由决策已预配置；Phase 2 尚未交出执行权，Native 仍是唯一执行者。' : '配置有效。'; return ( @@ -158,8 +175,8 @@ export function OrchestratorPage() { > - - + +