feat: add pluggable workflow orchestrator controls
This commit is contained in:
john
@@ -0,0 +1,296 @@
|
||||
import {
|
||||
DEFAULT_ORCHESTRATED_WORKFLOWS,
|
||||
ORCHESTRATOR_MODE,
|
||||
WORKFLOW_ENGINE,
|
||||
normalizeStringList,
|
||||
normalizeWorkflowEngineId,
|
||||
normalizeWorkflowName,
|
||||
stableRolloutBucket,
|
||||
} from './contracts.mjs';
|
||||
|
||||
const CONFIG_TABLE = 'h5_orchestrator_admin_config';
|
||||
const CONFIG_SCOPE = 'global';
|
||||
|
||||
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 '';
|
||||
url.username = '';
|
||||
url.password = '';
|
||||
url.hash = '';
|
||||
return url.toString().replace(/\/$/, '').slice(0, 2048);
|
||||
} catch {
|
||||
return '';
|
||||
}
|
||||
}
|
||||
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
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),
|
||||
};
|
||||
}
|
||||
|
||||
function runtimeState(config, env = process.env) {
|
||||
const killSwitch = normalizeBoolean(env.MEMIND_ORCHESTRATOR_KILL_SWITCH, 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';
|
||||
return {
|
||||
killSwitch,
|
||||
configured,
|
||||
effective: !reason,
|
||||
reason,
|
||||
executesLangGraph: !reason
|
||||
&& [ORCHESTRATOR_MODE.CANARY, ORCHESTRATOR_MODE.ACTIVE].includes(config.mode)
|
||||
&& config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH,
|
||||
shadowsLangGraph: !reason
|
||||
&& config.mode === ORCHESTRATOR_MODE.SHADOW
|
||||
&& config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH,
|
||||
};
|
||||
}
|
||||
|
||||
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 ensureConfigTable(pool) {
|
||||
await 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
|
||||
`);
|
||||
}
|
||||
|
||||
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 } = {}) {
|
||||
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),
|
||||
};
|
||||
},
|
||||
|
||||
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() {
|
||||
const state = await loadEffectiveState();
|
||||
return {
|
||||
...state,
|
||||
runtime: runtimeState(state.config, env),
|
||||
engines: engineCatalog(state.config),
|
||||
};
|
||||
},
|
||||
|
||||
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,
|
||||
shadowEngine: null,
|
||||
fallbackEngine: config.fallbackToNative ? config.fallbackEngine : null,
|
||||
mode: config.mode,
|
||||
workflowMatched,
|
||||
configVersion: state.configVersion,
|
||||
reason: runtime.reason,
|
||||
};
|
||||
if (!runtime.effective || !workflowMatched) {
|
||||
return { ...base, reason: runtime.reason ?? 'workflow_not_enabled' };
|
||||
}
|
||||
if (config.mode === ORCHESTRATOR_MODE.SHADOW) {
|
||||
return {
|
||||
...base,
|
||||
shadowEngine: config.primaryEngine,
|
||||
reason: 'shadow',
|
||||
};
|
||||
}
|
||||
if (config.mode === ORCHESTRATOR_MODE.ACTIVE) {
|
||||
return { ...base, engine: config.primaryEngine, reason: 'active' };
|
||||
}
|
||||
if (config.mode === ORCHESTRATOR_MODE.CANARY) {
|
||||
const explicitlyAllowed = Boolean(userId && config.userAllowlist.includes(String(userId)));
|
||||
const rolloutKey = runId || requestId || `${userId ?? ''}:${normalizedWorkflow}`;
|
||||
const bucket = stableRolloutBucket(rolloutKey);
|
||||
const percentAllowed = bucket < config.rolloutPercent;
|
||||
return explicitlyAllowed || percentAllowed
|
||||
? { ...base, engine: config.primaryEngine, reason: explicitlyAllowed ? 'user_allowlist' : 'percentage', bucket }
|
||||
: { ...base, reason: 'canary_not_selected', bucket };
|
||||
}
|
||||
return base;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export const orchestratorAdminConfigInternals = {
|
||||
CONFIG_SCOPE,
|
||||
CONFIG_TABLE,
|
||||
engineCatalog,
|
||||
runtimeState,
|
||||
};
|
||||
Reference in New Issue
Block a user