john
133 lines
3.9 KiB
JavaScript
133 lines
3.9 KiB
JavaScript
import {
|
|
WORKFLOW_ENGINE,
|
|
normalizeRunSpec,
|
|
} from './contracts.mjs';
|
|
import { createRemoteWorkflowEngine } from './engine-registry.mjs';
|
|
|
|
const MAX_INSTRUCTION_CHARACTERS = 16_000;
|
|
|
|
function extractMessageText(message) {
|
|
if (typeof message === 'string') return message;
|
|
if (!message || typeof message !== 'object') return '';
|
|
if (typeof message.content === 'string') return message.content;
|
|
if (!Array.isArray(message.content)) return '';
|
|
return message.content
|
|
.filter((part) => part?.type === 'text')
|
|
.map((part) => String(part.text ?? ''))
|
|
.join('\n');
|
|
}
|
|
|
|
function safeError(error) {
|
|
return {
|
|
code: String(error?.code ?? 'WORKFLOW_SHADOW_FAILED').slice(0, 128),
|
|
message: String(error instanceof Error ? error.message : error).slice(0, 1000),
|
|
};
|
|
}
|
|
|
|
export function createWorkflowShadowObserver({
|
|
configService,
|
|
serviceToken = process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN,
|
|
fetchImpl = globalThis.fetch,
|
|
logger = console,
|
|
} = {}) {
|
|
if (!configService?.selectEngine || !configService?.getRuntimeState) {
|
|
throw new Error('Workflow shadow observer requires orchestrator config service');
|
|
}
|
|
|
|
return async function observeWorkflowRun({
|
|
runId,
|
|
requestId,
|
|
userId,
|
|
sessionId = null,
|
|
workflowName = 'code-run-v1',
|
|
userMessage,
|
|
taskType = null,
|
|
} = {}) {
|
|
const selection = await configService.selectEngine({
|
|
runId,
|
|
requestId,
|
|
userId,
|
|
workflowName,
|
|
});
|
|
const planningMode = ['shadow', 'canary', 'active'].includes(selection.mode);
|
|
const executionPlan = planningMode
|
|
&& (selection.dryRun || ['canary', 'active'].includes(selection.mode)) ? {
|
|
version: 'workflow-execution-decision-v1',
|
|
candidateEngine: selection.candidateEngine ?? WORKFLOW_ENGINE.NATIVE,
|
|
effectiveEngine: selection.engine ?? WORKFLOW_ENGINE.NATIVE,
|
|
fallbackEngine: selection.fallbackEngine ?? WORKFLOW_ENGINE.NATIVE,
|
|
reason: selection.reason,
|
|
candidateReason: selection.candidateReason ?? null,
|
|
configVersion: selection.configVersion ?? null,
|
|
bucket: selection.bucket ?? null,
|
|
taskType,
|
|
dryRun: true,
|
|
handoffAllowed: false,
|
|
} : null;
|
|
if (selection.shadowEngine !== WORKFLOW_ENGINE.LANGGRAPH) {
|
|
return {
|
|
observed: false,
|
|
reason: selection.reason,
|
|
mode: selection.mode,
|
|
executionPlan,
|
|
};
|
|
}
|
|
|
|
const state = await configService.getRuntimeState();
|
|
const engine = createRemoteWorkflowEngine({
|
|
id: WORKFLOW_ENGINE.LANGGRAPH,
|
|
baseUrl: state.config.serviceUrl,
|
|
serviceToken,
|
|
timeoutMs: state.config.requestTimeoutMs,
|
|
fetchImpl,
|
|
});
|
|
const instruction = extractMessageText(userMessage).slice(0, MAX_INSTRUCTION_CHARACTERS);
|
|
const spec = normalizeRunSpec({
|
|
runId,
|
|
requestId,
|
|
workflow: { name: workflowName, version: 1 },
|
|
subject: { userId },
|
|
input: {
|
|
instruction,
|
|
taskType,
|
|
toolMode: 'code',
|
|
sessionRef: sessionId
|
|
? { kind: 'goose-session', id: String(sessionId) }
|
|
: null,
|
|
},
|
|
policy: {
|
|
executionMode: 'observe-only',
|
|
sideEffectsAllowed: false,
|
|
},
|
|
metadata: {
|
|
source: 'memind-agent-run',
|
|
configVersion: selection.configVersion,
|
|
},
|
|
});
|
|
try {
|
|
const result = await engine.start(spec);
|
|
return {
|
|
observed: true,
|
|
engine: WORKFLOW_ENGINE.LANGGRAPH,
|
|
mode: selection.mode,
|
|
configVersion: selection.configVersion,
|
|
executionPlan,
|
|
shadowRun: result,
|
|
};
|
|
} catch (error) {
|
|
logger.warn('[orchestrator-shadow] observation failed:', safeError(error));
|
|
if (error && typeof error === 'object') {
|
|
error.executionPlan = executionPlan;
|
|
error.mode = selection.mode;
|
|
}
|
|
throw error;
|
|
}
|
|
};
|
|
}
|
|
|
|
export const workflowShadowObserverInternals = {
|
|
MAX_INSTRUCTION_CHARACTERS,
|
|
extractMessageText,
|
|
safeError,
|
|
};
|