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.
664 lines
22 KiB
JavaScript
664 lines
22 KiB
JavaScript
import { createRemoteWorkflowEngine } from './engine-registry.mjs';
|
|
import { WORKFLOW_ENGINE } from './contracts.mjs';
|
|
|
|
const SHADOW_EVENT_TYPES = Object.freeze([
|
|
'workflow_shadow_completed',
|
|
'workflow_shadow_failed',
|
|
'workflow_shadow_skipped',
|
|
]);
|
|
const EXECUTION_PLAN_EVENT_TYPE = 'workflow_execution_planned';
|
|
const MAX_METRIC_EVENTS = 5000;
|
|
const CANARY_READINESS_THRESHOLDS = Object.freeze({
|
|
hours: 24 * 7,
|
|
minObservations: 20,
|
|
minSuccessRate: 0.95,
|
|
maxSkipRate: 0,
|
|
maxP95LatencyMs: 2000,
|
|
minLatencyCoverageRate: 0.95,
|
|
minNativeSettledRate: 0.95,
|
|
minDistinctSessions: 5,
|
|
maxHoursSinceLastObservation: 24,
|
|
});
|
|
const NATIVE_TERMINAL_STATUSES = new Set(['succeeded', 'failed']);
|
|
const SYNTHETIC_TASK_TYPES = new Set(['orchestrator_shadow_smoke']);
|
|
|
|
function clampInteger(value, fallback, min, max) {
|
|
const parsed = Number(value);
|
|
if (!Number.isFinite(parsed)) return fallback;
|
|
return Math.min(max, Math.max(min, Math.floor(parsed)));
|
|
}
|
|
|
|
function parseJsonColumn(value, fallback = null) {
|
|
if (value == null || value === '') return fallback;
|
|
if (typeof value === 'object') return value;
|
|
try {
|
|
return JSON.parse(String(value));
|
|
} catch {
|
|
return fallback;
|
|
}
|
|
}
|
|
|
|
function percentile(values, ratio) {
|
|
if (!values.length) return null;
|
|
const sorted = [...values].sort((left, right) => left - right);
|
|
const index = Math.max(0, Math.ceil(sorted.length * ratio) - 1);
|
|
return sorted[index];
|
|
}
|
|
|
|
function normalizeRunId(value) {
|
|
const runId = String(value ?? '').trim();
|
|
return /^[a-zA-Z0-9_-]{1,128}$/.test(runId) ? runId : '';
|
|
}
|
|
|
|
function normalizeExecutorJobId(value) {
|
|
const jobId = String(value ?? '').trim();
|
|
return /^[a-zA-Z0-9:_-]{1,128}$/.test(jobId) ? jobId : '';
|
|
}
|
|
|
|
function projectShadowEventRow(row) {
|
|
const data = parseJsonColumn(row.data_json, {}) ?? {};
|
|
const succeeded = row.event_type === 'workflow_shadow_completed';
|
|
const skipped = row.event_type === 'workflow_shadow_skipped';
|
|
const latency = Number(data.latencyMs);
|
|
const taskType = data.taskType ?? null;
|
|
return {
|
|
eventId: row.event_id,
|
|
runId: row.run_id,
|
|
requestId: row.request_id,
|
|
userId: row.user_id,
|
|
sessionId: row.agent_session_id ?? null,
|
|
nativeStatus: row.native_status,
|
|
nativeAttempts: Number(row.native_attempts ?? 0),
|
|
shadowStatus: succeeded ? 'succeeded' : skipped ? 'skipped' : 'failed',
|
|
engine: data.engine ?? 'langgraph',
|
|
configVersion: data.configVersion == null ? null : Number(data.configVersion),
|
|
phase: data.phase ?? null,
|
|
taskType,
|
|
synthetic: data.synthetic === true || SYNTHETIC_TASK_TYPES.has(String(taskType ?? '')),
|
|
executorAdapter: data.executorAdapter ?? null,
|
|
latencyMs: Number.isFinite(latency) && latency >= 0 ? latency : null,
|
|
error: succeeded || skipped ? null : {
|
|
code: data.code ?? 'WORKFLOW_SHADOW_FAILED',
|
|
message: data.message ?? 'Shadow observation failed',
|
|
},
|
|
skipReason: skipped ? data.reason ?? 'shadow_queue_full' : null,
|
|
observedAt: Number(row.event_created_at ?? 0),
|
|
nativeCompletedAt: row.native_completed_at == null
|
|
? null
|
|
: Number(row.native_completed_at),
|
|
};
|
|
}
|
|
|
|
function projectUniqueShadowRuns(rows) {
|
|
const seen = new Set();
|
|
const runs = [];
|
|
for (const row of rows) {
|
|
const run = projectShadowEventRow(row);
|
|
if (seen.has(run.runId)) continue;
|
|
seen.add(run.runId);
|
|
runs.push(run);
|
|
}
|
|
return runs;
|
|
}
|
|
|
|
function projectExecutionPlanEventRow(row) {
|
|
const data = parseJsonColumn(row.data_json, {}) ?? {};
|
|
const bucket = Number(data.bucket);
|
|
return {
|
|
eventId: row.event_id,
|
|
runId: row.run_id,
|
|
requestId: row.request_id,
|
|
userId: row.user_id,
|
|
sessionId: row.agent_session_id ?? null,
|
|
nativeStatus: row.native_status,
|
|
nativeAttempts: Number(row.native_attempts ?? 0),
|
|
mode: data.mode ?? null,
|
|
candidateEngine: data.candidateEngine ?? 'native',
|
|
effectiveEngine: data.effectiveEngine ?? 'native',
|
|
fallbackEngine: data.fallbackEngine ?? 'native',
|
|
reason: data.reason ?? null,
|
|
candidateReason: data.candidateReason ?? null,
|
|
configVersion: data.configVersion == null ? null : Number(data.configVersion),
|
|
bucket: Number.isFinite(bucket) ? bucket : null,
|
|
taskType: data.taskType ?? null,
|
|
dryRun: data.dryRun === true,
|
|
handoffAllowed: data.handoffAllowed === true,
|
|
plannedAt: Number(row.event_created_at ?? 0),
|
|
nativeCompletedAt: row.native_completed_at == null
|
|
? null
|
|
: Number(row.native_completed_at),
|
|
};
|
|
}
|
|
|
|
function projectUniqueExecutionPlans(rows) {
|
|
const seen = new Set();
|
|
const plans = [];
|
|
for (const row of rows) {
|
|
const plan = projectExecutionPlanEventRow(row);
|
|
if (seen.has(plan.runId)) continue;
|
|
seen.add(plan.runId);
|
|
plans.push(plan);
|
|
}
|
|
return plans;
|
|
}
|
|
|
|
function ratio(numerator, denominator) {
|
|
return denominator > 0 ? numerator / denominator : null;
|
|
}
|
|
|
|
function buildCanaryReadiness(loaded, runtimeState, now) {
|
|
const thresholds = CANARY_READINESS_THRESHOLDS;
|
|
const eligibleRuns = loaded.runs.filter((run) => !run.synthetic);
|
|
const successes = eligibleRuns.filter((run) => run.shadowStatus === 'succeeded').length;
|
|
const failures = eligibleRuns.filter((run) => run.shadowStatus === 'failed').length;
|
|
const skipped = eligibleRuns.filter((run) => run.shadowStatus === 'skipped').length;
|
|
const latencies = eligibleRuns
|
|
.map((run) => run.latencyMs)
|
|
.filter((value) => Number.isFinite(value));
|
|
const nativeSettled = eligibleRuns.filter((run) =>
|
|
NATIVE_TERMINAL_STATUSES.has(run.nativeStatus)).length;
|
|
const distinctSessions = new Set(
|
|
eligibleRuns.map((run) => run.sessionId).filter(Boolean),
|
|
).size;
|
|
const lastObservedAt = eligibleRuns[0]?.observedAt ?? null;
|
|
const hoursSinceLastObservation = lastObservedAt == null
|
|
? null
|
|
: Math.max(0, (now - lastObservedAt) / (60 * 60 * 1000));
|
|
const successRate = ratio(successes, eligibleRuns.length);
|
|
const skipRate = ratio(skipped, eligibleRuns.length);
|
|
const latencyCoverageRate = ratio(latencies.length, eligibleRuns.length);
|
|
const nativeSettledRate = ratio(nativeSettled, eligibleRuns.length);
|
|
const latencyP95Ms = percentile(latencies, 0.95);
|
|
const serviceHealth = runtimeState?.serviceHealth ?? null;
|
|
const checkpoint = serviceHealth?.details?.checkpoint ?? null;
|
|
const executorStore = serviceHealth?.details?.executorGateway?.store ?? null;
|
|
|
|
const checks = [
|
|
{
|
|
id: 'shadow_mode',
|
|
passed: runtimeState?.config?.mode === 'shadow',
|
|
actual: runtimeState?.config?.mode ?? null,
|
|
target: 'shadow',
|
|
},
|
|
{
|
|
id: 'shadow_wiring_enabled',
|
|
passed: runtimeState?.runtime?.shadowObservation?.enabled === true,
|
|
actual: runtimeState?.runtime?.shadowObservation?.enabled ?? null,
|
|
target: true,
|
|
},
|
|
{
|
|
id: 'service_healthy',
|
|
passed: serviceHealth?.ok === true,
|
|
actual: serviceHealth?.status ?? null,
|
|
target: 'healthy',
|
|
},
|
|
{
|
|
id: 'durable_checkpoint',
|
|
passed: checkpoint?.durable === true,
|
|
actual: checkpoint?.durable ?? null,
|
|
target: true,
|
|
},
|
|
{
|
|
id: 'durable_executor_job_store',
|
|
passed: executorStore?.durable === true,
|
|
actual: executorStore?.durable ?? null,
|
|
target: true,
|
|
},
|
|
{
|
|
id: 'observe_only',
|
|
passed: serviceHealth?.details?.execution === 'observe-only',
|
|
actual: serviceHealth?.details?.execution ?? null,
|
|
target: 'observe-only',
|
|
},
|
|
{
|
|
id: 'sample_volume',
|
|
passed: eligibleRuns.length >= thresholds.minObservations,
|
|
actual: eligibleRuns.length,
|
|
target: thresholds.minObservations,
|
|
},
|
|
{
|
|
id: 'shadow_success_rate',
|
|
passed: successRate != null && successRate >= thresholds.minSuccessRate,
|
|
actual: successRate,
|
|
target: thresholds.minSuccessRate,
|
|
},
|
|
{
|
|
id: 'shadow_skip_rate',
|
|
passed: skipRate != null && skipRate <= thresholds.maxSkipRate,
|
|
actual: skipRate,
|
|
target: thresholds.maxSkipRate,
|
|
},
|
|
{
|
|
id: 'latency_coverage',
|
|
passed: latencyCoverageRate != null
|
|
&& latencyCoverageRate >= thresholds.minLatencyCoverageRate,
|
|
actual: latencyCoverageRate,
|
|
target: thresholds.minLatencyCoverageRate,
|
|
},
|
|
{
|
|
id: 'latency_p95',
|
|
passed: latencyP95Ms != null && latencyP95Ms <= thresholds.maxP95LatencyMs,
|
|
actual: latencyP95Ms,
|
|
target: thresholds.maxP95LatencyMs,
|
|
},
|
|
{
|
|
id: 'native_settled_rate',
|
|
passed: nativeSettledRate != null
|
|
&& nativeSettledRate >= thresholds.minNativeSettledRate,
|
|
actual: nativeSettledRate,
|
|
target: thresholds.minNativeSettledRate,
|
|
},
|
|
{
|
|
id: 'session_coverage',
|
|
passed: distinctSessions >= thresholds.minDistinctSessions,
|
|
actual: distinctSessions,
|
|
target: thresholds.minDistinctSessions,
|
|
},
|
|
{
|
|
id: 'sample_freshness',
|
|
passed: hoursSinceLastObservation != null
|
|
&& hoursSinceLastObservation <= thresholds.maxHoursSinceLastObservation,
|
|
actual: hoursSinceLastObservation,
|
|
target: thresholds.maxHoursSinceLastObservation,
|
|
},
|
|
{
|
|
id: 'complete_window',
|
|
passed: !loaded.capped,
|
|
actual: loaded.capped ? 'sampled' : 'complete',
|
|
target: 'complete',
|
|
},
|
|
];
|
|
const failureCounts = new Map();
|
|
for (const run of eligibleRuns) {
|
|
if (!run.error?.code) continue;
|
|
failureCounts.set(run.error.code, (failureCounts.get(run.error.code) ?? 0) + 1);
|
|
}
|
|
|
|
return {
|
|
generatedAt: now,
|
|
ready: checks.every((check) => check.passed),
|
|
recommendation: checks.every((check) => check.passed)
|
|
? 'manual_canary_review'
|
|
: 'keep_shadow',
|
|
window: {
|
|
hours: loaded.hours,
|
|
from: loaded.from,
|
|
},
|
|
thresholds,
|
|
samples: {
|
|
totalObservations: loaded.runs.length,
|
|
eligibleObservations: eligibleRuns.length,
|
|
excludedSynthetic: loaded.runs.length - eligibleRuns.length,
|
|
successes,
|
|
failures,
|
|
skipped,
|
|
successRate,
|
|
skipRate,
|
|
latencyCoverageRate,
|
|
latencyP95Ms,
|
|
nativeSettledRate,
|
|
distinctSessions,
|
|
lastObservedAt,
|
|
hoursSinceLastObservation,
|
|
sampled: loaded.capped,
|
|
},
|
|
service: {
|
|
status: serviceHealth?.status ?? null,
|
|
latencyMs: serviceHealth?.latencyMs ?? null,
|
|
checkpointKind: checkpoint?.kind ?? null,
|
|
checkpointDurable: checkpoint?.durable ?? null,
|
|
executorJobStoreKind: executorStore?.kind ?? null,
|
|
executorJobStoreDurable: executorStore?.durable ?? null,
|
|
execution: serviceHealth?.details?.execution ?? null,
|
|
},
|
|
checks,
|
|
blockers: checks.filter((check) => !check.passed).map((check) => check.id),
|
|
failureCodes: [...failureCounts.entries()]
|
|
.map(([code, count]) => ({ code, count }))
|
|
.sort((left, right) => right.count - left.count || left.code.localeCompare(right.code))
|
|
.slice(0, 10),
|
|
};
|
|
}
|
|
|
|
function summarizeRuns(runs, { capped = false } = {}) {
|
|
const successes = runs.filter((run) => run.shadowStatus === 'succeeded').length;
|
|
const failures = runs.filter((run) => run.shadowStatus === 'failed').length;
|
|
const skipped = runs.filter((run) => run.shadowStatus === 'skipped').length;
|
|
const latencies = runs
|
|
.map((run) => run.latencyMs)
|
|
.filter((value) => Number.isFinite(value));
|
|
return {
|
|
observations: runs.length,
|
|
successes,
|
|
failures,
|
|
skipped,
|
|
successRate: runs.length ? successes / runs.length : null,
|
|
failureRate: runs.length ? failures / runs.length : null,
|
|
skipRate: runs.length ? skipped / runs.length : null,
|
|
latencyP50Ms: percentile(latencies, 0.5),
|
|
latencyP95Ms: percentile(latencies, 0.95),
|
|
nativeSucceeded: runs.filter((run) => run.nativeStatus === 'succeeded').length,
|
|
nativeFailed: runs.filter((run) => run.nativeStatus === 'failed').length,
|
|
lastObservedAt: runs[0]?.observedAt ?? null,
|
|
sampled: capped,
|
|
};
|
|
}
|
|
|
|
function countValues(values) {
|
|
const counts = new Map();
|
|
for (const value of values) {
|
|
const normalized = String(value ?? '').trim() || 'unknown';
|
|
counts.set(normalized, (counts.get(normalized) ?? 0) + 1);
|
|
}
|
|
return [...counts.entries()]
|
|
.map(([value, count]) => ({ value, count }))
|
|
.sort((left, right) => right.count - left.count || left.value.localeCompare(right.value));
|
|
}
|
|
|
|
function summarizeExecutionPlans(plans, { capped = false } = {}) {
|
|
const candidateSelections = plans.filter(
|
|
(plan) => plan.candidateEngine !== WORKFLOW_ENGINE.NATIVE,
|
|
).length;
|
|
const nativeSettled = plans.filter((plan) =>
|
|
NATIVE_TERMINAL_STATUSES.has(plan.nativeStatus)).length;
|
|
return {
|
|
decisions: plans.length,
|
|
candidateSelections,
|
|
candidateSelectionRate: ratio(candidateSelections, plans.length),
|
|
nativeSelections: plans.length - candidateSelections,
|
|
nativeSucceeded: plans.filter((plan) => plan.nativeStatus === 'succeeded').length,
|
|
nativeFailed: plans.filter((plan) => plan.nativeStatus === 'failed').length,
|
|
nativeSettledRate: ratio(nativeSettled, plans.length),
|
|
handoffAllowed: plans.filter((plan) => plan.handoffAllowed).length,
|
|
distinctSessions: new Set(plans.map((plan) => plan.sessionId).filter(Boolean)).size,
|
|
lastPlannedAt: plans[0]?.plannedAt ?? null,
|
|
candidateReasons: countValues(plans.map((plan) => plan.candidateReason ?? plan.reason)),
|
|
taskTypes: countValues(plans.map((plan) => plan.taskType)),
|
|
sampled: capped,
|
|
};
|
|
}
|
|
|
|
function safeRemoteError(error) {
|
|
return {
|
|
code: String(error?.code ?? 'ORCHESTRATOR_UNAVAILABLE').slice(0, 128),
|
|
message: String(error instanceof Error ? error.message : error).slice(0, 1000),
|
|
};
|
|
}
|
|
|
|
export function createOrchestratorObservabilityService({
|
|
pool,
|
|
configService,
|
|
serviceToken = process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN,
|
|
fetchImpl = globalThis.fetch,
|
|
nowMs = Date.now,
|
|
} = {}) {
|
|
if (!pool?.query) throw new Error('Orchestrator observability requires a database pool');
|
|
if (!configService?.getRuntimeState) {
|
|
throw new Error('Orchestrator observability requires config service');
|
|
}
|
|
|
|
async function loadShadowEvents({ hours = 24 } = {}) {
|
|
const normalizedHours = clampInteger(hours, 24, 1, 24 * 30);
|
|
const from = nowMs() - normalizedHours * 60 * 60 * 1000;
|
|
const [rows] = await pool.query(
|
|
`SELECT
|
|
e.id AS event_id,
|
|
e.run_id,
|
|
e.event_type,
|
|
e.data_json,
|
|
e.created_at AS event_created_at,
|
|
r.request_id,
|
|
r.user_id,
|
|
r.agent_session_id,
|
|
r.status AS native_status,
|
|
r.attempts AS native_attempts,
|
|
r.completed_at AS native_completed_at
|
|
FROM h5_agent_run_events e
|
|
INNER JOIN h5_agent_runs r ON r.id = e.run_id
|
|
WHERE e.event_type IN (${SHADOW_EVENT_TYPES.map(() => '?').join(', ')})
|
|
AND e.created_at >= ?
|
|
ORDER BY e.created_at DESC, e.id DESC
|
|
LIMIT ?`,
|
|
[...SHADOW_EVENT_TYPES, from, MAX_METRIC_EVENTS],
|
|
);
|
|
return {
|
|
from,
|
|
hours: normalizedHours,
|
|
capped: rows.length >= MAX_METRIC_EVENTS,
|
|
runs: projectUniqueShadowRuns(rows),
|
|
};
|
|
}
|
|
|
|
async function loadExecutionPlanEvents({ hours = 24 } = {}) {
|
|
const normalizedHours = clampInteger(hours, 24, 1, 24 * 30);
|
|
const from = nowMs() - normalizedHours * 60 * 60 * 1000;
|
|
const [rows] = await pool.query(
|
|
`SELECT
|
|
e.id AS event_id,
|
|
e.run_id,
|
|
e.data_json,
|
|
e.created_at AS event_created_at,
|
|
r.request_id,
|
|
r.user_id,
|
|
r.agent_session_id,
|
|
r.status AS native_status,
|
|
r.attempts AS native_attempts,
|
|
r.completed_at AS native_completed_at
|
|
FROM h5_agent_run_events e
|
|
INNER JOIN h5_agent_runs r ON r.id = e.run_id
|
|
WHERE e.event_type = ?
|
|
AND e.created_at >= ?
|
|
ORDER BY e.created_at DESC, e.id DESC
|
|
LIMIT ?`,
|
|
[EXECUTION_PLAN_EVENT_TYPE, from, MAX_METRIC_EVENTS],
|
|
);
|
|
return {
|
|
from,
|
|
hours: normalizedHours,
|
|
capped: rows.length >= MAX_METRIC_EVENTS,
|
|
plans: projectUniqueExecutionPlans(rows),
|
|
};
|
|
}
|
|
|
|
return {
|
|
async getCanaryReadiness() {
|
|
const [loaded, runtimeState] = await Promise.all([
|
|
loadShadowEvents({ hours: CANARY_READINESS_THRESHOLDS.hours }),
|
|
configService.getRuntimeState({ probe: true }),
|
|
]);
|
|
return buildCanaryReadiness(loaded, runtimeState, nowMs());
|
|
},
|
|
|
|
async listShadowRuns({
|
|
hours = 24,
|
|
limit = 50,
|
|
status = 'all',
|
|
} = {}) {
|
|
const loaded = await loadShadowEvents({ hours });
|
|
const normalizedLimit = clampInteger(limit, 50, 1, 200);
|
|
const normalizedStatus = ['succeeded', 'failed', 'skipped'].includes(status)
|
|
? status
|
|
: 'all';
|
|
const filtered = normalizedStatus === 'all'
|
|
? loaded.runs
|
|
: loaded.runs.filter((run) => run.shadowStatus === normalizedStatus);
|
|
return {
|
|
generatedAt: nowMs(),
|
|
window: {
|
|
hours: loaded.hours,
|
|
from: loaded.from,
|
|
},
|
|
metrics: summarizeRuns(loaded.runs, { capped: loaded.capped }),
|
|
runs: filtered.slice(0, normalizedLimit),
|
|
};
|
|
},
|
|
|
|
async listExecutionPlans({
|
|
hours = 24,
|
|
limit = 50,
|
|
selection = 'all',
|
|
} = {}) {
|
|
const loaded = await loadExecutionPlanEvents({ hours });
|
|
const normalizedLimit = clampInteger(limit, 50, 1, 200);
|
|
const normalizedSelection = ['candidate', 'native'].includes(selection)
|
|
? selection
|
|
: 'all';
|
|
const filtered = normalizedSelection === 'all'
|
|
? loaded.plans
|
|
: loaded.plans.filter((plan) => (
|
|
normalizedSelection === 'candidate'
|
|
? plan.candidateEngine !== WORKFLOW_ENGINE.NATIVE
|
|
: plan.candidateEngine === WORKFLOW_ENGINE.NATIVE
|
|
));
|
|
return {
|
|
generatedAt: nowMs(),
|
|
window: {
|
|
hours: loaded.hours,
|
|
from: loaded.from,
|
|
},
|
|
metrics: summarizeExecutionPlans(loaded.plans, { capped: loaded.capped }),
|
|
plans: filtered.slice(0, normalizedLimit),
|
|
};
|
|
},
|
|
|
|
async getShadowRun(runId) {
|
|
const normalizedRunId = normalizeRunId(runId);
|
|
if (!normalizedRunId) return null;
|
|
const [runRows] = await pool.query(
|
|
`SELECT id, request_id, user_id, agent_session_id, status, attempts,
|
|
error_message, created_at, updated_at, started_at, completed_at
|
|
FROM h5_agent_runs
|
|
WHERE id = ?
|
|
LIMIT 1`,
|
|
[normalizedRunId],
|
|
);
|
|
if (!runRows[0]) return null;
|
|
const [eventRows] = await pool.query(
|
|
`SELECT id AS event_id, event_type, data_json, created_at
|
|
FROM h5_agent_run_events
|
|
WHERE run_id = ?
|
|
AND event_type IN (${SHADOW_EVENT_TYPES.map(() => '?').join(', ')})
|
|
ORDER BY created_at ASC, id ASC`,
|
|
[normalizedRunId, ...SHADOW_EVENT_TYPES],
|
|
);
|
|
const native = runRows[0];
|
|
const localEvents = eventRows.map((row) => ({
|
|
eventId: row.event_id,
|
|
type: row.event_type,
|
|
data: parseJsonColumn(row.data_json, null),
|
|
createdAt: Number(row.created_at),
|
|
}));
|
|
|
|
let remote = {
|
|
available: false,
|
|
state: null,
|
|
events: [],
|
|
error: null,
|
|
executorJob: {
|
|
available: false,
|
|
state: null,
|
|
events: [],
|
|
error: null,
|
|
},
|
|
};
|
|
try {
|
|
const runtimeState = await configService.getRuntimeState();
|
|
if (runtimeState.config.serviceUrl) {
|
|
const engine = createRemoteWorkflowEngine({
|
|
baseUrl: runtimeState.config.serviceUrl,
|
|
serviceToken,
|
|
timeoutMs: runtimeState.config.requestTimeoutMs,
|
|
fetchImpl,
|
|
});
|
|
const [state, events] = await Promise.all([
|
|
engine.getState(normalizedRunId),
|
|
(async () => {
|
|
const collected = [];
|
|
for await (const event of engine.streamEvents(normalizedRunId)) {
|
|
collected.push(event);
|
|
if (collected.length >= 500) break;
|
|
}
|
|
return collected;
|
|
})(),
|
|
]);
|
|
const executorJobId = normalizeExecutorJobId(state?.plan?.executorJob?.id);
|
|
let executorJob = {
|
|
available: false,
|
|
state: null,
|
|
events: [],
|
|
error: null,
|
|
};
|
|
if (executorJobId) {
|
|
try {
|
|
const [jobState, jobEvents] = await Promise.all([
|
|
engine.getExecutorJob(executorJobId),
|
|
(async () => {
|
|
const collected = [];
|
|
for await (const event of engine.streamExecutorJobEvents(executorJobId)) {
|
|
collected.push(event);
|
|
if (collected.length >= 500) break;
|
|
}
|
|
return collected;
|
|
})(),
|
|
]);
|
|
executorJob = {
|
|
available: true,
|
|
state: jobState,
|
|
events: jobEvents,
|
|
error: null,
|
|
};
|
|
} catch (error) {
|
|
executorJob.error = safeRemoteError(error);
|
|
}
|
|
}
|
|
remote = {
|
|
available: true,
|
|
state,
|
|
events,
|
|
error: null,
|
|
executorJob,
|
|
};
|
|
}
|
|
} catch (error) {
|
|
remote.error = safeRemoteError(error);
|
|
}
|
|
|
|
return {
|
|
native: {
|
|
runId: native.id,
|
|
requestId: native.request_id,
|
|
userId: native.user_id,
|
|
sessionId: native.agent_session_id ?? null,
|
|
status: native.status,
|
|
attempts: Number(native.attempts ?? 0),
|
|
error: native.error_message ?? null,
|
|
createdAt: Number(native.created_at ?? 0),
|
|
updatedAt: Number(native.updated_at ?? 0),
|
|
startedAt: native.started_at == null ? null : Number(native.started_at),
|
|
completedAt: native.completed_at == null ? null : Number(native.completed_at),
|
|
},
|
|
localEvents,
|
|
remote,
|
|
};
|
|
},
|
|
};
|
|
}
|
|
|
|
export const orchestratorObservabilityInternals = {
|
|
CANARY_READINESS_THRESHOLDS,
|
|
EXECUTION_PLAN_EVENT_TYPE,
|
|
MAX_METRIC_EVENTS,
|
|
SHADOW_EVENT_TYPES,
|
|
buildCanaryReadiness,
|
|
normalizeExecutorJobId,
|
|
normalizeRunId,
|
|
parseJsonColumn,
|
|
percentile,
|
|
projectShadowEventRow,
|
|
projectUniqueShadowRuns,
|
|
projectExecutionPlanEventRow,
|
|
projectUniqueExecutionPlans,
|
|
summarizeExecutionPlans,
|
|
summarizeRuns,
|
|
};
|