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.
513 lines
16 KiB
JavaScript
513 lines
16 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import test from 'node:test';
|
|
import {
|
|
createOrchestratorObservabilityService,
|
|
orchestratorObservabilityInternals,
|
|
} from './observability.mjs';
|
|
|
|
function response(body, status = 200) {
|
|
return new Response(JSON.stringify(body), {
|
|
status,
|
|
headers: { 'content-type': 'application/json' },
|
|
});
|
|
}
|
|
|
|
test('observability aggregates shadow outcomes and latency percentiles', async () => {
|
|
const rows = [
|
|
{
|
|
event_id: 'event-3',
|
|
run_id: 'run-3',
|
|
event_type: 'workflow_shadow_completed',
|
|
data_json: { latencyMs: 300, engine: 'langgraph', taskType: 'refactor' },
|
|
event_created_at: 3000,
|
|
request_id: 'request-3',
|
|
user_id: 'user-3',
|
|
native_status: 'succeeded',
|
|
native_attempts: 1,
|
|
native_completed_at: 3100,
|
|
},
|
|
{
|
|
event_id: 'event-2',
|
|
run_id: 'run-2',
|
|
event_type: 'workflow_shadow_failed',
|
|
data_json: JSON.stringify({ latencyMs: 200, code: 'TIMEOUT', message: 'timed out' }),
|
|
event_created_at: 2000,
|
|
request_id: 'request-2',
|
|
user_id: 'user-2',
|
|
native_status: 'succeeded',
|
|
native_attempts: 1,
|
|
native_completed_at: 2200,
|
|
},
|
|
{
|
|
event_id: 'event-1',
|
|
run_id: 'run-1',
|
|
event_type: 'workflow_shadow_completed',
|
|
data_json: { latencyMs: 100, executorAdapter: 'native-agent-run' },
|
|
event_created_at: 1000,
|
|
request_id: 'request-1',
|
|
user_id: 'user-1',
|
|
native_status: 'failed',
|
|
native_attempts: 2,
|
|
native_completed_at: 1500,
|
|
},
|
|
{
|
|
event_id: 'event-skipped',
|
|
run_id: 'run-skipped',
|
|
event_type: 'workflow_shadow_skipped',
|
|
data_json: { reason: 'shadow_queue_full' },
|
|
event_created_at: 900,
|
|
request_id: 'request-skipped',
|
|
user_id: 'user-skipped',
|
|
native_status: 'succeeded',
|
|
native_attempts: 1,
|
|
native_completed_at: 1200,
|
|
},
|
|
];
|
|
const service = createOrchestratorObservabilityService({
|
|
pool: {
|
|
async query(sql) {
|
|
assert.match(sql, /INNER JOIN h5_agent_runs/);
|
|
return [rows];
|
|
},
|
|
},
|
|
configService: { getRuntimeState() {} },
|
|
});
|
|
|
|
const result = await service.listShadowRuns({ hours: 12, limit: 4 });
|
|
assert.equal(result.window.hours, 12);
|
|
assert.equal(result.metrics.observations, 4);
|
|
assert.equal(result.metrics.successes, 2);
|
|
assert.equal(result.metrics.failures, 1);
|
|
assert.equal(result.metrics.skipped, 1);
|
|
assert.equal(result.metrics.skipRate, 0.25);
|
|
assert.equal(result.metrics.latencyP50Ms, 200);
|
|
assert.equal(result.metrics.latencyP95Ms, 300);
|
|
assert.equal(result.metrics.nativeSucceeded, 3);
|
|
assert.equal(result.runs.length, 4);
|
|
assert.equal(result.runs[1].error.code, 'TIMEOUT');
|
|
assert.equal(result.runs[3].shadowStatus, 'skipped');
|
|
assert.equal(result.runs[3].skipReason, 'shadow_queue_full');
|
|
});
|
|
|
|
test('observability aggregates complete dry-run routing decisions and Native outcomes', async () => {
|
|
const rows = [
|
|
{
|
|
event_id: 'plan-event-3',
|
|
run_id: 'plan-run-3',
|
|
data_json: {
|
|
mode: 'canary',
|
|
candidateEngine: 'langgraph',
|
|
effectiveEngine: 'native',
|
|
fallbackEngine: 'native',
|
|
reason: 'execution_gate_disabled',
|
|
candidateReason: 'percentage',
|
|
configVersion: 4,
|
|
bucket: 12,
|
|
taskType: 'repo_refactor',
|
|
dryRun: true,
|
|
handoffAllowed: false,
|
|
},
|
|
event_created_at: 3000,
|
|
request_id: 'plan-request-3',
|
|
user_id: 'plan-user-3',
|
|
agent_session_id: 'plan-session-2',
|
|
native_status: 'failed',
|
|
native_attempts: 2,
|
|
native_completed_at: 3100,
|
|
},
|
|
{
|
|
event_id: 'plan-event-2',
|
|
run_id: 'plan-run-2',
|
|
data_json: JSON.stringify({
|
|
mode: 'canary',
|
|
candidateEngine: 'native',
|
|
effectiveEngine: 'native',
|
|
reason: 'canary_not_selected',
|
|
configVersion: 4,
|
|
bucket: 78,
|
|
taskType: 'code_analysis',
|
|
dryRun: true,
|
|
handoffAllowed: false,
|
|
}),
|
|
event_created_at: 2000,
|
|
request_id: 'plan-request-2',
|
|
user_id: 'plan-user-2',
|
|
agent_session_id: 'plan-session-1',
|
|
native_status: 'succeeded',
|
|
native_attempts: 1,
|
|
native_completed_at: 2100,
|
|
},
|
|
{
|
|
event_id: 'plan-event-1',
|
|
run_id: 'plan-run-1',
|
|
data_json: {
|
|
mode: 'canary',
|
|
candidateEngine: 'langgraph',
|
|
effectiveEngine: 'native',
|
|
reason: 'execution_gate_disabled',
|
|
candidateReason: 'user_allowlist',
|
|
taskType: 'repo_refactor',
|
|
dryRun: true,
|
|
handoffAllowed: false,
|
|
},
|
|
event_created_at: 1000,
|
|
request_id: 'plan-request-1',
|
|
user_id: 'plan-user-1',
|
|
agent_session_id: 'plan-session-1',
|
|
native_status: 'running',
|
|
native_attempts: 1,
|
|
native_completed_at: null,
|
|
},
|
|
];
|
|
const service = createOrchestratorObservabilityService({
|
|
pool: {
|
|
async query(sql, params) {
|
|
assert.match(sql, /e\.event_type = \?/);
|
|
assert.equal(params[0], 'workflow_execution_planned');
|
|
return [rows];
|
|
},
|
|
},
|
|
configService: { getRuntimeState() {} },
|
|
nowMs: () => 5000,
|
|
});
|
|
|
|
const result = await service.listExecutionPlans({
|
|
hours: 12,
|
|
limit: 10,
|
|
selection: 'candidate',
|
|
});
|
|
assert.equal(result.window.hours, 12);
|
|
assert.equal(result.metrics.decisions, 3);
|
|
assert.equal(result.metrics.candidateSelections, 2);
|
|
assert.equal(result.metrics.candidateSelectionRate, 2 / 3);
|
|
assert.equal(result.metrics.nativeSelections, 1);
|
|
assert.equal(result.metrics.nativeSucceeded, 1);
|
|
assert.equal(result.metrics.nativeFailed, 1);
|
|
assert.equal(result.metrics.nativeSettledRate, 2 / 3);
|
|
assert.equal(result.metrics.handoffAllowed, 0);
|
|
assert.equal(result.metrics.distinctSessions, 2);
|
|
assert.deepEqual(result.metrics.candidateReasons, [
|
|
{ value: 'canary_not_selected', count: 1 },
|
|
{ value: 'percentage', count: 1 },
|
|
{ value: 'user_allowlist', count: 1 },
|
|
]);
|
|
assert.deepEqual(result.metrics.taskTypes, [
|
|
{ value: 'repo_refactor', count: 2 },
|
|
{ value: 'code_analysis', count: 1 },
|
|
]);
|
|
assert.equal(result.plans.length, 2);
|
|
assert.equal(result.plans[0].candidateEngine, 'langgraph');
|
|
assert.equal(result.plans[0].effectiveEngine, 'native');
|
|
assert.equal(result.plans[0].taskType, 'repo_refactor');
|
|
});
|
|
|
|
test('observability detail joins Native state with remote LangGraph checkpoint events', async () => {
|
|
const queries = [];
|
|
const service = createOrchestratorObservabilityService({
|
|
pool: {
|
|
async query(sql, params) {
|
|
queries.push({ sql, params });
|
|
if (sql.includes('FROM h5_agent_runs')) {
|
|
return [[{
|
|
id: 'run-detail',
|
|
request_id: 'request-detail',
|
|
user_id: 'user-detail',
|
|
agent_session_id: 'session-detail',
|
|
status: 'succeeded',
|
|
attempts: 1,
|
|
error_message: null,
|
|
created_at: 100,
|
|
updated_at: 300,
|
|
started_at: 150,
|
|
completed_at: 300,
|
|
}]];
|
|
}
|
|
return [[{
|
|
event_id: 'local-event',
|
|
event_type: 'workflow_shadow_completed',
|
|
data_json: { latencyMs: 25 },
|
|
created_at: 250,
|
|
}]];
|
|
},
|
|
},
|
|
configService: {
|
|
async getRuntimeState() {
|
|
return {
|
|
config: {
|
|
serviceUrl: 'http://orchestrator.internal:8093',
|
|
requestTimeoutMs: 1000,
|
|
},
|
|
};
|
|
},
|
|
},
|
|
serviceToken: 'detail-token',
|
|
fetchImpl: async (url, init) => {
|
|
assert.equal(init.headers.authorization, 'Bearer detail-token');
|
|
if (String(url).includes('/executor-jobs/') && String(url).endsWith('/events')) {
|
|
return response({
|
|
events: [{ sequence: 1, type: 'executor_job_blocked' }],
|
|
});
|
|
}
|
|
if (String(url).includes('/executor-jobs/')) {
|
|
return response({
|
|
jobId: 'run-detail:executor-preview',
|
|
status: 'blocked',
|
|
reason: 'executor_dispatch_not_implemented',
|
|
});
|
|
}
|
|
if (String(url).endsWith('/events')) {
|
|
return response({
|
|
events: [{ sequence: 1, type: 'workflow_validated' }],
|
|
});
|
|
}
|
|
return response({
|
|
runId: 'run-detail',
|
|
status: 'succeeded',
|
|
plan: {
|
|
executorAdapter: 'native-agent-run',
|
|
executorJob: { id: 'run-detail:executor-preview' },
|
|
},
|
|
});
|
|
},
|
|
});
|
|
|
|
const detail = await service.getShadowRun('run-detail');
|
|
assert.equal(detail.native.status, 'succeeded');
|
|
assert.equal(detail.localEvents[0].data.latencyMs, 25);
|
|
assert.equal(detail.remote.available, true);
|
|
assert.equal(detail.remote.state.plan.executorAdapter, 'native-agent-run');
|
|
assert.equal(detail.remote.events[0].type, 'workflow_validated');
|
|
assert.equal(detail.remote.executorJob.available, true);
|
|
assert.equal(detail.remote.executorJob.state.status, 'blocked');
|
|
assert.equal(detail.remote.executorJob.events[0].type, 'executor_job_blocked');
|
|
assert.equal(queries.length, 2);
|
|
});
|
|
|
|
test('observability rejects unsafe run ids before querying', async () => {
|
|
let queries = 0;
|
|
const service = createOrchestratorObservabilityService({
|
|
pool: {
|
|
async query() {
|
|
queries += 1;
|
|
return [[]];
|
|
},
|
|
},
|
|
configService: { getRuntimeState() {} },
|
|
});
|
|
assert.equal(await service.getShadowRun('../unsafe'), null);
|
|
assert.equal(queries, 0);
|
|
});
|
|
|
|
test('observability counts only the newest terminal event for each run', () => {
|
|
const base = {
|
|
run_id: 'same-run',
|
|
request_id: 'same-request',
|
|
user_id: 'same-user',
|
|
native_status: 'succeeded',
|
|
native_attempts: 1,
|
|
};
|
|
const runs = orchestratorObservabilityInternals.projectUniqueShadowRuns([
|
|
{
|
|
...base,
|
|
event_id: 'newer-success',
|
|
event_type: 'workflow_shadow_completed',
|
|
data_json: { latencyMs: 80 },
|
|
event_created_at: 2000,
|
|
},
|
|
{
|
|
...base,
|
|
event_id: 'older-failure',
|
|
event_type: 'workflow_shadow_failed',
|
|
data_json: { latencyMs: 100, code: 'TIMEOUT' },
|
|
event_created_at: 1000,
|
|
},
|
|
]);
|
|
assert.equal(runs.length, 1);
|
|
assert.equal(runs[0].eventId, 'newer-success');
|
|
assert.equal(runs[0].shadowStatus, 'succeeded');
|
|
});
|
|
|
|
test('canary readiness excludes smoke runs and reports explicit blockers', async () => {
|
|
const now = 10_000_000;
|
|
const service = createOrchestratorObservabilityService({
|
|
pool: {
|
|
async query() {
|
|
return [[{
|
|
event_id: 'smoke-event',
|
|
run_id: 'smoke-run',
|
|
event_type: 'workflow_shadow_completed',
|
|
data_json: {
|
|
latencyMs: 50,
|
|
taskType: 'orchestrator_shadow_smoke',
|
|
},
|
|
event_created_at: now - 1000,
|
|
request_id: 'smoke-request',
|
|
user_id: 'smoke-user',
|
|
agent_session_id: null,
|
|
native_status: 'succeeded',
|
|
native_attempts: 0,
|
|
native_completed_at: now,
|
|
}]];
|
|
},
|
|
},
|
|
configService: {
|
|
async getRuntimeState(options) {
|
|
assert.deepEqual(options, { probe: true });
|
|
return {
|
|
config: { mode: 'shadow' },
|
|
runtime: {
|
|
shadowObservation: { enabled: true },
|
|
},
|
|
serviceHealth: {
|
|
ok: true,
|
|
status: 'healthy',
|
|
latencyMs: 5,
|
|
details: {
|
|
checkpoint: { kind: 'postgres', durable: true },
|
|
executorGateway: {
|
|
store: { kind: 'postgres', durable: true },
|
|
},
|
|
execution: 'observe-only',
|
|
},
|
|
},
|
|
};
|
|
},
|
|
},
|
|
nowMs: () => now,
|
|
});
|
|
|
|
const readiness = await service.getCanaryReadiness();
|
|
assert.equal(readiness.ready, false);
|
|
assert.equal(readiness.recommendation, 'keep_shadow');
|
|
assert.equal(readiness.samples.totalObservations, 1);
|
|
assert.equal(readiness.samples.eligibleObservations, 0);
|
|
assert.equal(readiness.samples.excludedSynthetic, 1);
|
|
assert.ok(readiness.blockers.includes('sample_volume'));
|
|
assert.ok(readiness.blockers.includes('session_coverage'));
|
|
assert.ok(!readiness.blockers.includes('service_healthy'));
|
|
});
|
|
|
|
test('canary readiness passes only when operational and sample gates all pass', async () => {
|
|
const now = 20_000_000;
|
|
let shadowObservationEnabled = true;
|
|
const rows = Array.from({ length: 20 }, (_, index) => ({
|
|
event_id: `event-${index}`,
|
|
run_id: `run-${index}`,
|
|
event_type: index === 19
|
|
? 'workflow_shadow_failed'
|
|
: 'workflow_shadow_completed',
|
|
data_json: {
|
|
latencyMs: 100 + index,
|
|
taskType: 'code_task',
|
|
...(index === 19 ? { code: 'TRANSIENT', message: 'retry later' } : {}),
|
|
},
|
|
event_created_at: now - index * 1000,
|
|
request_id: `request-${index}`,
|
|
user_id: `user-${index % 3}`,
|
|
agent_session_id: `session-${index % 5}`,
|
|
native_status: index % 2 ? 'succeeded' : 'failed',
|
|
native_attempts: 1,
|
|
native_completed_at: now,
|
|
}));
|
|
const service = createOrchestratorObservabilityService({
|
|
pool: {
|
|
async query() {
|
|
return [rows];
|
|
},
|
|
},
|
|
configService: {
|
|
async getRuntimeState() {
|
|
return {
|
|
config: { mode: 'shadow' },
|
|
runtime: {
|
|
shadowObservation: { enabled: shadowObservationEnabled },
|
|
},
|
|
serviceHealth: {
|
|
ok: true,
|
|
status: 'healthy',
|
|
latencyMs: 4,
|
|
details: {
|
|
checkpoint: { kind: 'postgres', durable: true },
|
|
executorGateway: {
|
|
store: { kind: 'postgres', durable: true },
|
|
},
|
|
execution: 'observe-only',
|
|
},
|
|
},
|
|
};
|
|
},
|
|
},
|
|
nowMs: () => now,
|
|
});
|
|
|
|
const readiness = await service.getCanaryReadiness();
|
|
assert.equal(readiness.ready, true);
|
|
assert.equal(readiness.recommendation, 'manual_canary_review');
|
|
assert.equal(readiness.window.hours, 168);
|
|
assert.equal(readiness.samples.eligibleObservations, 20);
|
|
assert.equal(readiness.samples.successRate, 0.95);
|
|
assert.equal(readiness.samples.skipped, 0);
|
|
assert.equal(readiness.samples.skipRate, 0);
|
|
assert.equal(readiness.samples.latencyCoverageRate, 1);
|
|
assert.equal(readiness.samples.nativeSettledRate, 1);
|
|
assert.equal(readiness.samples.distinctSessions, 5);
|
|
assert.equal(readiness.service.executorJobStoreDurable, true);
|
|
assert.equal(readiness.blockers.length, 0);
|
|
assert.deepEqual(readiness.failureCodes, [{ code: 'TRANSIENT', count: 1 }]);
|
|
|
|
shadowObservationEnabled = false;
|
|
const blockedReadiness = await service.getCanaryReadiness();
|
|
assert.equal(blockedReadiness.ready, false);
|
|
assert.ok(blockedReadiness.blockers.includes('shadow_wiring_enabled'));
|
|
});
|
|
|
|
test('canary readiness blocks when any Shadow observation was skipped', async () => {
|
|
const now = 30_000_000;
|
|
const rows = Array.from({ length: 20 }, (_, index) => ({
|
|
event_id: `event-skip-${index}`,
|
|
run_id: `run-skip-${index}`,
|
|
event_type: index === 0
|
|
? 'workflow_shadow_skipped'
|
|
: 'workflow_shadow_completed',
|
|
data_json: index === 0
|
|
? { reason: 'shadow_queue_full' }
|
|
: { latencyMs: 100, taskType: 'code_task' },
|
|
event_created_at: now - index * 1000,
|
|
request_id: `request-skip-${index}`,
|
|
user_id: `user-${index % 3}`,
|
|
agent_session_id: `session-${index % 5}`,
|
|
native_status: 'succeeded',
|
|
native_attempts: 1,
|
|
native_completed_at: now,
|
|
}));
|
|
const service = createOrchestratorObservabilityService({
|
|
pool: { async query() { return [rows]; } },
|
|
configService: {
|
|
async getRuntimeState() {
|
|
return {
|
|
config: { mode: 'shadow' },
|
|
runtime: {
|
|
shadowObservation: { enabled: true },
|
|
},
|
|
serviceHealth: {
|
|
ok: true,
|
|
status: 'healthy',
|
|
details: {
|
|
checkpoint: { kind: 'postgres', durable: true },
|
|
executorGateway: { store: { kind: 'postgres', durable: true } },
|
|
execution: 'observe-only',
|
|
},
|
|
},
|
|
};
|
|
},
|
|
},
|
|
nowMs: () => now,
|
|
});
|
|
|
|
const readiness = await service.getCanaryReadiness();
|
|
assert.equal(readiness.ready, false);
|
|
assert.equal(readiness.samples.skipped, 1);
|
|
assert.equal(readiness.samples.skipRate, 0.05);
|
|
assert.ok(readiness.blockers.includes('shadow_skip_rate'));
|
|
});
|