john
6df82818c5
Gate and bound Portal shadow observations while preserving Native execution. Add fail-closed service boundaries, terminal retention controls, Canary readiness telemetry, ops visibility, and isolated regression coverage.
519 lines
17 KiB
JavaScript
519 lines
17 KiB
JavaScript
import {
|
|
DEFAULT_ORCHESTRATED_WORKFLOWS,
|
|
ORCHESTRATOR_MODE,
|
|
WORKFLOW_ENGINE,
|
|
normalizeStringList,
|
|
normalizeWorkflowEngineId,
|
|
normalizeWorkflowName,
|
|
stableRolloutBucket,
|
|
} from './contracts.mjs';
|
|
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;
|
|
if (typeof value === 'boolean') return value;
|
|
const normalized = String(value).trim().toLowerCase();
|
|
if (['1', 'true', 'yes', 'on'].includes(normalized)) return true;
|
|
if (['0', 'false', 'no', 'off'].includes(normalized)) return false;
|
|
return fallback;
|
|
}
|
|
|
|
function clampInteger(value, fallback, min, max) {
|
|
const number = Number(value);
|
|
if (!Number.isFinite(number)) return fallback;
|
|
return Math.min(max, Math.max(min, Math.floor(number)));
|
|
}
|
|
|
|
function normalizeMode(value, fallback = ORCHESTRATOR_MODE.OFF) {
|
|
const normalized = String(value ?? '').trim().toLowerCase();
|
|
return Object.values(ORCHESTRATOR_MODE).includes(normalized) ? normalized : fallback;
|
|
}
|
|
|
|
function normalizeServiceUrl(value) {
|
|
const raw = String(value ?? '').trim().replace(/\/$/, '');
|
|
if (!raw) return '';
|
|
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.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;
|
|
try {
|
|
return JSON.parse(String(value));
|
|
} catch {
|
|
return fallback;
|
|
}
|
|
}
|
|
|
|
function clone(value) {
|
|
if (typeof structuredClone === 'function') return structuredClone(value);
|
|
return JSON.parse(JSON.stringify(value));
|
|
}
|
|
|
|
export function defaultOrchestratorConfig() {
|
|
return {
|
|
mode: ORCHESTRATOR_MODE.OFF,
|
|
primaryEngine: WORKFLOW_ENGINE.LANGGRAPH,
|
|
fallbackEngine: WORKFLOW_ENGINE.NATIVE,
|
|
serviceUrl: '',
|
|
requestTimeoutMs: 5000,
|
|
rolloutPercent: 0,
|
|
userAllowlist: [],
|
|
workflowAllowlist: [...DEFAULT_ORCHESTRATED_WORKFLOWS],
|
|
fallbackToNative: true,
|
|
requireHealthy: true,
|
|
executionEnabled: false,
|
|
};
|
|
}
|
|
|
|
export function normalizeOrchestratorConfig(input = {}, fallback = defaultOrchestratorConfig()) {
|
|
const source = input && typeof input === 'object' && !Array.isArray(input) ? input : {};
|
|
const base = fallback && typeof fallback === 'object' ? fallback : defaultOrchestratorConfig();
|
|
const workflowAllowlist = normalizeStringList(
|
|
source.workflowAllowlist ?? base.workflowAllowlist,
|
|
{ normalize: (item) => normalizeWorkflowName(item, ''), itemLimit: 128 },
|
|
);
|
|
return {
|
|
mode: normalizeMode(source.mode, base.mode),
|
|
primaryEngine: normalizeWorkflowEngineId(source.primaryEngine, base.primaryEngine),
|
|
fallbackEngine: normalizeWorkflowEngineId(source.fallbackEngine, base.fallbackEngine),
|
|
serviceUrl: normalizeServiceUrl(source.serviceUrl ?? base.serviceUrl),
|
|
requestTimeoutMs: clampInteger(
|
|
source.requestTimeoutMs,
|
|
base.requestTimeoutMs,
|
|
500,
|
|
60_000,
|
|
),
|
|
rolloutPercent: clampInteger(source.rolloutPercent, base.rolloutPercent, 0, 100),
|
|
userAllowlist: normalizeStringList(source.userAllowlist ?? base.userAllowlist, {
|
|
itemLimit: 128,
|
|
}),
|
|
workflowAllowlist: workflowAllowlist.length
|
|
? workflowAllowlist
|
|
: [...DEFAULT_ORCHESTRATED_WORKFLOWS],
|
|
fallbackToNative: normalizeBoolean(source.fallbackToNative, base.fallbackToNative),
|
|
requireHealthy: normalizeBoolean(source.requireHealthy, base.requireHealthy),
|
|
executionEnabled: normalizeBoolean(source.executionEnabled, base.executionEnabled),
|
|
};
|
|
}
|
|
|
|
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,
|
|
);
|
|
const configured = config.primaryEngine !== WORKFLOW_ENGINE.LANGGRAPH || Boolean(config.serviceUrl);
|
|
let reason = null;
|
|
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
|
|
&& executionModeSelected
|
|
&& config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH;
|
|
const executionHandoffEnabled = !reason
|
|
&& EXECUTION_HANDOFF_IMPLEMENTED
|
|
&& 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,
|
|
effective: !reason,
|
|
reason: reason ?? (
|
|
executionModeSelected
|
|
&& config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH
|
|
&& !executionHandoffEnabled
|
|
? 'execution_gate_disabled'
|
|
: null
|
|
),
|
|
executesLangGraph: executionHandoffEnabled,
|
|
plansLangGraph: !reason
|
|
&& executionModeSelected
|
|
&& 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,
|
|
enabled: executionHandoffEnabled,
|
|
reason: executionHandoffEnabled
|
|
? null
|
|
: !EXECUTION_HANDOFF_IMPLEMENTED
|
|
? 'execution_handoff_not_implemented'
|
|
: !environmentExecutionGate
|
|
? 'environment_execution_gate_disabled'
|
|
: 'execution_handoff_not_requested',
|
|
environmentGate: environmentExecutionGate,
|
|
},
|
|
};
|
|
}
|
|
|
|
function engineCatalog(config) {
|
|
return [
|
|
{
|
|
id: WORKFLOW_ENGINE.NATIVE,
|
|
label: 'Native Agent Run',
|
|
kind: 'built-in',
|
|
configured: true,
|
|
capabilities: ['existing-runtime', 'tool-gateway'],
|
|
},
|
|
{
|
|
id: WORKFLOW_ENGINE.LANGGRAPH,
|
|
label: 'LangGraph Orchestrator',
|
|
kind: 'remote',
|
|
configured: Boolean(config.serviceUrl),
|
|
capabilities: ['durable-execution', 'interrupt', 'streaming'],
|
|
},
|
|
];
|
|
}
|
|
|
|
async function probeServiceHealth(config, {
|
|
fetchImpl = globalThis.fetch,
|
|
} = {}) {
|
|
const checkedAt = Date.now();
|
|
if (!config.serviceUrl) {
|
|
return {
|
|
checkedAt,
|
|
ok: false,
|
|
status: 'unconfigured',
|
|
latencyMs: 0,
|
|
httpStatus: null,
|
|
details: null,
|
|
};
|
|
}
|
|
if (typeof fetchImpl !== 'function') {
|
|
return {
|
|
checkedAt,
|
|
ok: false,
|
|
status: 'fetch_unavailable',
|
|
latencyMs: 0,
|
|
httpStatus: null,
|
|
details: null,
|
|
};
|
|
}
|
|
const startedAt = Date.now();
|
|
const controller = new AbortController();
|
|
const timer = setTimeout(
|
|
() => controller.abort(),
|
|
Math.max(500, Number(config.requestTimeoutMs) || 5000),
|
|
);
|
|
try {
|
|
const response = await fetchImpl(`${config.serviceUrl}/ready`, {
|
|
headers: { accept: 'application/json' },
|
|
signal: controller.signal,
|
|
});
|
|
const body = await response.json().catch(() => null);
|
|
const healthy = response.ok && body?.status === 'ok';
|
|
return {
|
|
checkedAt,
|
|
ok: healthy,
|
|
status: healthy ? 'healthy' : 'unhealthy',
|
|
latencyMs: Math.max(0, Date.now() - startedAt),
|
|
httpStatus: response.status,
|
|
details: body && typeof body === 'object' ? {
|
|
service: body.service == null ? null : String(body.service).slice(0, 128),
|
|
checkpoint: body.checkpoint && typeof body.checkpoint === 'object' ? {
|
|
kind: body.checkpoint.kind == null ? null : String(body.checkpoint.kind).slice(0, 64),
|
|
durable: Boolean(body.checkpoint.durable),
|
|
} : null,
|
|
execution: body.execution == null ? null : String(body.execution).slice(0, 64),
|
|
executorGateway: body.executorGateway && typeof body.executorGateway === 'object' ? {
|
|
dispatchImplemented: body.executorGateway.dispatchImplemented === true,
|
|
executionEnabled: body.executorGateway.executionEnabled === true,
|
|
store: body.executorGateway.store && typeof body.executorGateway.store === 'object' ? {
|
|
kind: body.executorGateway.store.kind == null
|
|
? null
|
|
: String(body.executorGateway.store.kind).slice(0, 64),
|
|
durable: body.executorGateway.store.durable === true,
|
|
} : null,
|
|
} : null,
|
|
} : null,
|
|
};
|
|
} catch (error) {
|
|
return {
|
|
checkedAt,
|
|
ok: false,
|
|
status: error?.name === 'AbortError' ? 'timeout' : 'unreachable',
|
|
latencyMs: Math.max(0, Date.now() - startedAt),
|
|
httpStatus: null,
|
|
details: null,
|
|
};
|
|
} finally {
|
|
clearTimeout(timer);
|
|
}
|
|
}
|
|
|
|
async function ensureConfigTable(pool) {
|
|
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,
|
|
config_version INT NOT NULL DEFAULT 1,
|
|
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) {
|
|
await ensureConfigTable(pool);
|
|
const [rows] = await pool.query(
|
|
`SELECT config_json, config_version, updated_by, updated_at
|
|
FROM ${CONFIG_TABLE}
|
|
WHERE config_scope = ?
|
|
LIMIT 1`,
|
|
[CONFIG_SCOPE],
|
|
);
|
|
const row = rows[0];
|
|
if (!row) return null;
|
|
return {
|
|
config: normalizeOrchestratorConfig(parseJsonLike(row.config_json, {})),
|
|
configVersion: Math.max(1, Number(row.config_version ?? 1) || 1),
|
|
updatedBy: row.updated_by ?? null,
|
|
updatedAt: Number(row.updated_at ?? 0) || null,
|
|
};
|
|
}
|
|
|
|
export function createOrchestratorAdminConfigService(pool, {
|
|
env = process.env,
|
|
fetchImpl = globalThis.fetch,
|
|
} = {}) {
|
|
async function loadEffectiveState() {
|
|
const stored = await loadStoredState(pool);
|
|
if (stored) return { ...stored, source: 'admin-db' };
|
|
const config = normalizeOrchestratorConfig({
|
|
mode: env.MEMIND_ORCHESTRATOR_MODE,
|
|
serviceUrl: env.MEMIND_ORCHESTRATOR_URL,
|
|
});
|
|
const fromEnv = Boolean(env.MEMIND_ORCHESTRATOR_MODE || env.MEMIND_ORCHESTRATOR_URL);
|
|
return {
|
|
config,
|
|
configVersion: 1,
|
|
updatedBy: null,
|
|
updatedAt: null,
|
|
source: fromEnv ? 'env' : 'default',
|
|
};
|
|
}
|
|
|
|
return {
|
|
ensureSchema() {
|
|
return ensureConfigTable(pool);
|
|
},
|
|
|
|
async getAdminConfig() {
|
|
const state = await loadEffectiveState();
|
|
return {
|
|
...state,
|
|
runtime: runtimeState(state.config, env),
|
|
engines: engineCatalog(state.config),
|
|
executors: listPhase5ExecutorAdapters(),
|
|
};
|
|
},
|
|
|
|
async updateAdminConfig(patch = {}, { updatedBy = null } = {}) {
|
|
const stored = await loadStoredState(pool);
|
|
const current = stored?.config ?? defaultOrchestratorConfig();
|
|
const nextConfig = normalizeOrchestratorConfig(
|
|
{ ...current, ...(patch?.config ?? patch) },
|
|
current,
|
|
);
|
|
const configVersion = (stored?.configVersion ?? 0) + 1;
|
|
const now = Date.now();
|
|
await pool.query(
|
|
`INSERT INTO ${CONFIG_TABLE}
|
|
(config_scope, config_json, config_version, updated_by, updated_at)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
ON DUPLICATE KEY UPDATE
|
|
config_json = VALUES(config_json),
|
|
config_version = VALUES(config_version),
|
|
updated_by = VALUES(updated_by),
|
|
updated_at = VALUES(updated_at)`,
|
|
[CONFIG_SCOPE, JSON.stringify(nextConfig), configVersion, updatedBy, now],
|
|
);
|
|
return this.getAdminConfig();
|
|
},
|
|
|
|
async getRuntimeState({ probe = false } = {}) {
|
|
const state = await loadEffectiveState();
|
|
return {
|
|
...state,
|
|
runtime: runtimeState(state.config, env),
|
|
engines: engineCatalog(state.config),
|
|
executors: listPhase5ExecutorAdapters(),
|
|
...(probe ? { serviceHealth: await probeServiceHealth(state.config, { fetchImpl }) } : {}),
|
|
};
|
|
},
|
|
|
|
async selectEngine({
|
|
runId,
|
|
requestId,
|
|
userId = null,
|
|
workflowName,
|
|
} = {}) {
|
|
const state = await loadEffectiveState();
|
|
const config = state.config;
|
|
const runtime = runtimeState(config, env);
|
|
const normalizedWorkflow = normalizeWorkflowName(workflowName, '');
|
|
const workflowMatched = config.workflowAllowlist.includes(normalizedWorkflow);
|
|
const base = {
|
|
engine: WORKFLOW_ENGINE.NATIVE,
|
|
candidateEngine: WORKFLOW_ENGINE.NATIVE,
|
|
shadowEngine: null,
|
|
fallbackEngine: config.fallbackToNative ? config.fallbackEngine : null,
|
|
mode: config.mode,
|
|
workflowMatched,
|
|
configVersion: state.configVersion,
|
|
reason: runtime.reason,
|
|
candidateReason: null,
|
|
dryRun: false,
|
|
};
|
|
if (!runtime.effective || !workflowMatched) {
|
|
return { ...base, reason: runtime.reason ?? 'workflow_not_enabled' };
|
|
}
|
|
const rolloutKey = runId || requestId || `${userId ?? ''}:${normalizedWorkflow}`;
|
|
const bucket = stableRolloutBucket(rolloutKey);
|
|
const explicitlyAllowed = Boolean(userId && config.userAllowlist.includes(String(userId)));
|
|
const percentAllowed = bucket < config.rolloutPercent;
|
|
const rolloutSelected = explicitlyAllowed || percentAllowed;
|
|
const rolloutReason = explicitlyAllowed
|
|
? 'user_allowlist'
|
|
: percentAllowed
|
|
? 'percentage'
|
|
: 'canary_not_selected';
|
|
if (config.mode === ORCHESTRATOR_MODE.SHADOW) {
|
|
return {
|
|
...base,
|
|
shadowEngine: config.primaryEngine,
|
|
reason: 'shadow',
|
|
candidateEngine: rolloutSelected
|
|
? config.primaryEngine
|
|
: WORKFLOW_ENGINE.NATIVE,
|
|
candidateReason: rolloutReason,
|
|
bucket,
|
|
dryRun: true,
|
|
};
|
|
}
|
|
if (config.mode === ORCHESTRATOR_MODE.ACTIVE) {
|
|
const selected = {
|
|
...base,
|
|
candidateEngine: config.primaryEngine,
|
|
candidateReason: 'active',
|
|
};
|
|
if (config.primaryEngine === WORKFLOW_ENGINE.NATIVE) {
|
|
return { ...selected, reason: 'active' };
|
|
}
|
|
return runtime.executionHandoff.enabled
|
|
? { ...selected, engine: config.primaryEngine, reason: 'active' }
|
|
: { ...selected, reason: 'execution_gate_disabled', dryRun: true };
|
|
}
|
|
if (config.mode === ORCHESTRATOR_MODE.CANARY) {
|
|
if (!rolloutSelected) {
|
|
return { ...base, reason: 'canary_not_selected', bucket };
|
|
}
|
|
const selected = {
|
|
...base,
|
|
candidateEngine: config.primaryEngine,
|
|
candidateReason: rolloutReason,
|
|
bucket,
|
|
};
|
|
if (config.primaryEngine === WORKFLOW_ENGINE.NATIVE) {
|
|
return { ...selected, reason: rolloutReason };
|
|
}
|
|
return runtime.executionHandoff.enabled
|
|
? { ...selected, engine: config.primaryEngine, reason: rolloutReason }
|
|
: { ...selected, reason: 'execution_gate_disabled', dryRun: true };
|
|
}
|
|
return base;
|
|
},
|
|
};
|
|
}
|
|
|
|
export const orchestratorAdminConfigInternals = {
|
|
CONFIG_SCOPE,
|
|
CONFIG_TABLE,
|
|
engineCatalog,
|
|
EXECUTION_HANDOFF_IMPLEMENTED,
|
|
isServiceUrlAllowed,
|
|
normalizeServiceUrl,
|
|
probeServiceHealth,
|
|
runtimeState,
|
|
};
|