Codex
324 lines
14 KiB
TypeScript
324 lines
14 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { test } from "bun:test";
|
|
|
|
import { nodeWebObserveRunnerRealtimeSource } from "./hwlab-node-web-observe-runner-realtime-source";
|
|
|
|
test("realtime fanout keeps the YAML terminal timeout when durationMs is absent", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(
|
|
"boundedInteger",
|
|
"sanitize",
|
|
`${source}\nreturn realtimeFanoutEffectiveProfile;`,
|
|
) as (
|
|
boundedInteger: (value: unknown, fallback: number, min: number, max: number) => number,
|
|
sanitize: <T>(value: T) => T,
|
|
) => (profile: Record<string, unknown>, durationMs: unknown) => Record<string, unknown>;
|
|
const effectiveProfile = build(
|
|
(value, fallback, min, max) => Math.min(max, Math.max(min, Number.isFinite(Number(value)) ? Number(value) : fallback)),
|
|
(value) => value,
|
|
);
|
|
const profile = {
|
|
subscriberCount: 2,
|
|
debugStreams: ["stdio", "agentrun", "hwlab"],
|
|
fromBeginning: false,
|
|
barrierTimeoutMs: 45_000,
|
|
terminalTimeoutMs: 480_000,
|
|
terminalQuietMs: 2_000,
|
|
reconnectQuietMs: 3_000,
|
|
expectedProductSse: {
|
|
deliverySemantics: "kafka-retention-then-live",
|
|
liveOnly: false,
|
|
replay: true,
|
|
replaySupported: true,
|
|
lossPossible: false,
|
|
refreshHandoff: true,
|
|
replayPriorTraceOnReconnect: true,
|
|
},
|
|
forbiddenRequestPaths: [],
|
|
requiredEventTypes: { agentrun: ["terminal_status"], hwlab: ["terminal"] },
|
|
expectedKafka: {
|
|
topics: { stdio: "codex-stdio.raw.v1", agentrun: "agentrun.event.v1", hwlab: "hwlab.event.v1" },
|
|
groups: { directPublish: "direct", transactionalProjector: "projector", liveSse: "live" },
|
|
capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: true },
|
|
},
|
|
};
|
|
|
|
assert.equal(effectiveProfile(profile, null).terminalTimeoutMs, 480_000);
|
|
assert.equal(effectiveProfile(profile, undefined).terminalTimeoutMs, 480_000);
|
|
assert.equal(effectiveProfile(profile, 5_000).terminalTimeoutMs, 5_000);
|
|
});
|
|
|
|
test("realtime fanout retries a YAML-bounded transient health navigation", async () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(
|
|
"sleep",
|
|
"isRetryableNavigationError",
|
|
"baseUrl",
|
|
`${source}\nreturn realtimeGotoHealth;`,
|
|
) as (
|
|
sleep: (milliseconds: number) => Promise<void>,
|
|
isRetryableNavigationError: (message: string) => boolean,
|
|
baseUrl: string,
|
|
) => (page: { goto: () => Promise<{ status: () => number }> }, timeoutMs: number, profile: Record<string, number>, label: string) => Promise<Record<string, unknown>>;
|
|
const gotoHealth = build(
|
|
async () => {},
|
|
(message) => /ERR_NETWORK_CHANGED/u.test(message),
|
|
"https://hwlab.example.test",
|
|
);
|
|
let calls = 0;
|
|
const result = await gotoHealth({
|
|
async goto() {
|
|
calls += 1;
|
|
if (calls === 1) throw new Error("page.goto: net::ERR_NETWORK_CHANGED");
|
|
return { status: () => 200 };
|
|
},
|
|
}, 45_000, { navigationMaxAttempts: 4, navigationRetryDelayMs: 1_000 }, "subscriber-a");
|
|
|
|
assert.equal(result.ok, true);
|
|
assert.equal(calls, 2);
|
|
assert.equal((result.attempts as unknown[]).length, 2);
|
|
});
|
|
|
|
test("realtime fanout validates stdio against the scoped AgentRun session lineage", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn realtimeExpectedStreamSessionId;`) as () => (
|
|
stream: string,
|
|
hwlabSessionId: string,
|
|
) => string;
|
|
const expectedStreamSessionId = build();
|
|
const hwlabSessionId = "ses_5ec4e141-6abc-466d-9afe-049f7c0ac105";
|
|
|
|
assert.equal(expectedStreamSessionId("stdio", hwlabSessionId), "ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105");
|
|
assert.equal(expectedStreamSessionId("agentrun", hwlabSessionId), hwlabSessionId);
|
|
assert.equal(expectedStreamSessionId("hwlab", hwlabSessionId), hwlabSessionId);
|
|
});
|
|
|
|
test("realtime fanout validates YAML-owned refresh and legacy live-only connected contracts", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn realtimeAssertConnectedContract;`) as () => (
|
|
connected: Record<string, unknown>,
|
|
sessionId: string,
|
|
label: string,
|
|
profile: Record<string, unknown>,
|
|
options?: Record<string, unknown>,
|
|
) => void;
|
|
const assertConnected = build();
|
|
const sessionId = "ses_refresh_contract";
|
|
const expectedKafka = {
|
|
topics: { hwlab: "hwlab.event.v1" },
|
|
capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: true },
|
|
};
|
|
const refreshProfile = {
|
|
expectedKafka,
|
|
expectedProductSse: {
|
|
deliverySemantics: "kafka-retention-then-live",
|
|
liveOnly: false,
|
|
replay: true,
|
|
replaySupported: true,
|
|
lossPossible: false,
|
|
refreshHandoff: true,
|
|
replayPriorTraceOnReconnect: true,
|
|
},
|
|
};
|
|
const refreshConnected = {
|
|
realtimeSource: "hwlab.event.v1",
|
|
deliverySemantics: "kafka-retention-then-live",
|
|
liveOnly: false,
|
|
replay: true,
|
|
replaySupported: true,
|
|
lossPossible: false,
|
|
filters: { sessionId, traceId: null },
|
|
capabilities: expectedKafka.capabilities,
|
|
refreshReplay: {
|
|
phase: "live",
|
|
topic: "hwlab.event.v1",
|
|
topicPartitions: [0],
|
|
barrier: [{ partition: 0, endOffset: "18" }],
|
|
counts: { matched: 8, replayed: 8, buffered: 1, bufferedDelivered: 1, liveDelivered: 0, deduplicated: 0 },
|
|
},
|
|
};
|
|
|
|
assert.doesNotThrow(() => assertConnected(refreshConnected, sessionId, "refresh", refreshProfile, { requireRetainedEvents: true }));
|
|
assert.throws(
|
|
() => assertConnected({ ...refreshConnected, deliverySemantics: "live-only" }, sessionId, "refresh", refreshProfile),
|
|
/differs from the owning YAML/u,
|
|
);
|
|
assert.throws(
|
|
() => assertConnected({ ...refreshConnected, refreshReplay: { ...refreshConnected.refreshReplay, counts: { ...refreshConnected.refreshReplay.counts, replayed: 9 } } }, sessionId, "refresh", refreshProfile),
|
|
/replayed count exceeds matched count/u,
|
|
);
|
|
|
|
const liveProfile = {
|
|
expectedKafka: { topics: { hwlab: "hwlab.event.v1" }, capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: false } },
|
|
expectedProductSse: {
|
|
deliverySemantics: "live-only",
|
|
liveOnly: true,
|
|
replay: false,
|
|
replaySupported: false,
|
|
lossPossible: true,
|
|
refreshHandoff: false,
|
|
replayPriorTraceOnReconnect: false,
|
|
},
|
|
};
|
|
assert.doesNotThrow(() => assertConnected({
|
|
realtimeSource: "hwlab.event.v1",
|
|
deliverySemantics: "live-only",
|
|
liveOnly: true,
|
|
replay: false,
|
|
replaySupported: false,
|
|
lossPossible: true,
|
|
filters: { sessionId, traceId: null },
|
|
capabilities: liveProfile.expectedKafka.capabilities,
|
|
}, sessionId, "live", liveProfile));
|
|
});
|
|
|
|
test("existing session refresh rejects duplicate or reordered Workbench identities", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn realtimeAssertSameWorkbenchDom;`) as () => (
|
|
before: Record<string, unknown>, after: Record<string, unknown>,
|
|
) => void;
|
|
const assertSame = build();
|
|
const before = { messageCount: 2, identities: ["user:m1:", "agent:m2:t1"], missingIdentityRows: [], duplicateIdentities: [] };
|
|
assert.doesNotThrow(() => assertSame(before, { ...before }));
|
|
assert.throws(() => assertSame(before, { ...before, identities: [...before.identities].reverse() }), /identity or order changed/u);
|
|
assert.throws(() => assertSame(before, { ...before, duplicateIdentities: ["agent:m2:t1"] }), /duplicate message identities/u);
|
|
assert.throws(() => assertSame(before, { ...before, missingIdentityRows: [1] }), /without a stable message identity/u);
|
|
});
|
|
|
|
test("existing session refresh failures preserve typed phase and code", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`function errorSummary(error) { return { message: error.message }; }\n${source}\nreturn { realtimeExistingRefreshFailure, realtimeExistingRefreshStreamError };`) as () => {
|
|
realtimeExistingRefreshFailure: (phase: string, error: Error & { details?: Record<string, unknown> }) => Record<string, unknown>;
|
|
realtimeExistingRefreshStreamError: (value: Record<string, unknown>) => Error & { details?: Record<string, unknown> };
|
|
};
|
|
const helpers = build();
|
|
const error = helpers.realtimeExistingRefreshStreamError({
|
|
phase: "flushing",
|
|
error: { code: "workbench_kafka_refresh_barrier_gap", message: "Kafka retention barrier gap", retryable: true },
|
|
fallback: false,
|
|
});
|
|
const failure = helpers.realtimeExistingRefreshFailure("retention-barrier", error);
|
|
assert.equal(failure.phase, "retention-barrier");
|
|
assert.equal(failure.code, "retention_barrier_failed");
|
|
assert.equal(failure.streamPhase, "flushing");
|
|
assert.equal(failure.streamCode, "workbench_kafka_refresh_barrier_gap");
|
|
assert.equal((failure.streamFailure as Record<string, unknown>).fallback, false);
|
|
assert.deepEqual(failure.valuesRedacted, true);
|
|
});
|
|
|
|
test("existing session refresh preserves rejected connected evidence and rejects missing browser replay delivery", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(
|
|
"sanitize",
|
|
`${source}\nreturn { realtimeExistingRefreshConnectedEvidence, realtimeAssertExistingRefreshDelivery };`,
|
|
) as (sanitize: <T>(value: T) => T) => {
|
|
realtimeExistingRefreshConnectedEvidence: (connected: Record<string, any>, sessionId: string, profile: Record<string, any>) => Record<string, unknown>;
|
|
realtimeAssertExistingRefreshDelivery: (browser: Record<string, unknown>, refreshReplay: Record<string, any>) => void;
|
|
};
|
|
const helpers = build((value) => value);
|
|
const sessionId = "ses_existing";
|
|
const refreshReplay = {
|
|
phase: "live",
|
|
topic: "hwlab.event.v1",
|
|
topicPartitions: [0],
|
|
barrier: [{ partition: 0, endOffset: "18" }],
|
|
counts: { matched: 8, replayed: 8, buffered: 1, bufferedDelivered: 1, liveDelivered: 0, deduplicated: 0 },
|
|
};
|
|
const profile = {
|
|
expectedKafka: { topics: { hwlab: "hwlab.event.v1" }, capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: true } },
|
|
expectedProductSse: { deliverySemantics: "kafka-retention-then-live", liveOnly: false, replay: true, replaySupported: true, lossPossible: false, refreshHandoff: true },
|
|
};
|
|
const connected = {
|
|
realtimeSource: "hwlab.event.v1",
|
|
deliverySemantics: "invalid",
|
|
liveOnly: false,
|
|
replay: true,
|
|
replaySupported: true,
|
|
lossPossible: false,
|
|
filters: { sessionId, traceId: null },
|
|
capabilities: profile.expectedKafka.capabilities,
|
|
refreshReplay,
|
|
};
|
|
assert.throws(
|
|
() => helpers.realtimeExistingRefreshConnectedEvidence(connected, sessionId, profile),
|
|
(error: Error & { details?: Record<string, any> }) => {
|
|
assert.deepEqual(error.details?.refreshReplay?.barrier, refreshReplay.barrier);
|
|
assert.equal(error.details?.refreshReplay?.counts?.replayed, 8);
|
|
return true;
|
|
},
|
|
);
|
|
assert.doesNotThrow(() => helpers.realtimeAssertExistingRefreshDelivery({ businessEventCount: 9 }, refreshReplay));
|
|
assert.throws(
|
|
() => helpers.realtimeAssertExistingRefreshDelivery({ businessEventCount: 8 }, refreshReplay),
|
|
/fewer than replayed plus buffered delivery 9/u,
|
|
);
|
|
});
|
|
|
|
test("realtime fanout proves one formal user event through AgentRun, HWLAB Kafka, and product SSE", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn realtimeValidateTurnEvidence;`) as () => (input: Record<string, unknown>) => void;
|
|
const validate = build();
|
|
const traceId = "trc_user_passthrough";
|
|
const sessionId = "ses_user_passthrough";
|
|
const runId = "run_user_passthrough";
|
|
const commandId = "cmd_user_passthrough";
|
|
const userMessageId = "msg_user_passthrough";
|
|
const base = { traceId, runId, commandId, terminal: false };
|
|
const userHwlab = {
|
|
...base,
|
|
partition: 0,
|
|
offset: "0",
|
|
hwlabSessionId: sessionId,
|
|
eventId: "hwlab:evt_user",
|
|
sourceEventId: "evt_user",
|
|
sourceSeq: 1,
|
|
eventType: "user",
|
|
userMessageId,
|
|
envelopeFingerprint: "fingerprint:user",
|
|
};
|
|
const terminalHwlab = {
|
|
...base,
|
|
partition: 0,
|
|
offset: "1",
|
|
hwlabSessionId: sessionId,
|
|
eventId: "hwlab:evt_terminal",
|
|
sourceEventId: "evt_terminal",
|
|
sourceSeq: 2,
|
|
eventType: "terminal",
|
|
userMessageId: null,
|
|
terminal: true,
|
|
envelopeFingerprint: "fingerprint:terminal",
|
|
};
|
|
const debug = {
|
|
connected: {
|
|
stdio: { consumerReady: true, topic: "codex-stdio.raw.v1" },
|
|
agentrun: { consumerReady: true, topic: "agentrun.event.v1" },
|
|
hwlab: { consumerReady: true, topic: "hwlab.event.v1" },
|
|
},
|
|
openCount: { stdio: 1, agentrun: 1, hwlab: 1 },
|
|
connectedEventCount: { stdio: 1, agentrun: 1, hwlab: 1 },
|
|
errors: { stdio: [], agentrun: [], hwlab: [] },
|
|
records: {
|
|
stdio: [{ ...base, partition: 0, offset: "0", frameSeq: 1, hwlabSessionId: "ses_agentrun_user_passthrough" }],
|
|
agentrun: [
|
|
{ ...base, partition: 0, offset: "0", hwlabSessionId: sessionId, eventId: "evt_user", eventType: "user_message", userMessageId },
|
|
{ ...base, partition: 0, offset: "1", hwlabSessionId: sessionId, eventId: "evt_terminal", eventType: "terminal_status", terminal: true },
|
|
],
|
|
hwlab: [userHwlab, terminalHwlab],
|
|
},
|
|
};
|
|
const product = { openCount: 1, connectedEventCount: 1, errors: [], records: [{ ...userHwlab }, { ...terminalHwlab }] };
|
|
const profile = {
|
|
debugStreams: ["stdio", "agentrun", "hwlab"],
|
|
requiredEventTypes: { agentrun: ["user_message", "terminal_status"], hwlab: ["user", "terminal"] },
|
|
expectedKafka: { topics: { stdio: "codex-stdio.raw.v1", agentrun: "agentrun.event.v1", hwlab: "hwlab.event.v1" } },
|
|
};
|
|
|
|
assert.doesNotThrow(() => validate({ turn: 1, traceId, sessionId, runId, commandId, debug, terminalSnapshots: [product], preTerminalSnapshots: [], profile }));
|
|
const missingUserProduct = { ...product, records: [terminalHwlab] };
|
|
assert.throws(
|
|
() => validate({ turn: 1, traceId, sessionId, runId, commandId, debug, terminalSnapshots: [missingUserProduct], preTerminalSnapshots: [], profile }),
|
|
/product subscriber lacks pre-terminal or terminal event|product SSE lacks the formal user event/u,
|
|
);
|
|
});
|