feat(web-probe): add typed Kafka live fanout validation
This commit is contained in:
Codex
+49
-4
@@ -63,6 +63,7 @@ export interface CheckOptions {
|
||||
recoveryGuardrails: boolean;
|
||||
rust: boolean;
|
||||
scriptsTypecheckTimeoutMs: number;
|
||||
scriptsTypecheckPath: string | null;
|
||||
checkHeartbeatMs: number;
|
||||
}
|
||||
|
||||
@@ -79,6 +80,7 @@ const defaultCheckOptions: CheckOptions = {
|
||||
recoveryGuardrails: false,
|
||||
rust: false,
|
||||
scriptsTypecheckTimeoutMs: defaultScriptsTypecheckTimeoutMs,
|
||||
scriptsTypecheckPath: null,
|
||||
checkHeartbeatMs: defaultCheckHeartbeatMs,
|
||||
};
|
||||
|
||||
@@ -95,7 +97,7 @@ export function checkHelp(): Record<string, unknown> {
|
||||
ok: true,
|
||||
command: "check",
|
||||
usage: [
|
||||
"bun scripts/cli.ts check [--syntax-only|--full|--files|--scripts-typecheck|--scripts-typecheck-timeout-ms N|--check-heartbeat-ms N|--components|--compose|--logs|--recovery-guardrails|--rust]",
|
||||
"bun scripts/cli.ts check [--syntax-only|--full|--files|--scripts-typecheck|--scripts-typecheck-path PREFIX|--scripts-typecheck-timeout-ms N|--check-heartbeat-ms N|--components|--compose|--logs|--recovery-guardrails|--rust]",
|
||||
"bun scripts/cli.ts check recovery-guardrails",
|
||||
],
|
||||
defaultMode: "syntax/config only; Rust is never compiled on the master server by default",
|
||||
@@ -104,6 +106,7 @@ export function checkHelp(): Record<string, unknown> {
|
||||
{ name: "--full", description: "Enable all non-Rust checks." },
|
||||
{ name: "--files", description: "Verify required entrypoint files, including backend-core Cargo files." },
|
||||
{ name: "--scripts-typecheck", description: "Run scripts TypeScript typecheck through the observed checker." },
|
||||
{ name: "--scripts-typecheck-path PREFIX", description: "Keep the full typecheck gate, but return bounded primary diagnostics only for a selected path prefix." },
|
||||
{ name: "--scripts-typecheck-timeout-ms N", description: `Bound scripts TypeScript typecheck duration; default ${defaultScriptsTypecheckTimeoutMs}.` },
|
||||
{ name: "--check-heartbeat-ms N", description: `Emit unidesk.check.progress JSON lines for running command checks; default ${defaultCheckHeartbeatMs}.` },
|
||||
{ name: "--components", description: "Run component TypeScript typecheck." },
|
||||
@@ -139,6 +142,12 @@ export function parseCheckOptions(args: string[]): CheckOptions {
|
||||
} else if (arg === "--scripts-typecheck-timeout-ms") {
|
||||
options.scriptsTypecheckTimeoutMs = positiveIntegerOption(arg, args[index + 1], 600_000);
|
||||
index += 1;
|
||||
} else if (arg === "--scripts-typecheck-path") {
|
||||
const value = args[index + 1];
|
||||
if (!value || value.includes("\0") || value.length > 300) throw new Error(`${arg} requires a 1-300 character non-NUL prefix`);
|
||||
options.scriptsTypecheck = true;
|
||||
options.scriptsTypecheckPath = value;
|
||||
index += 1;
|
||||
} else if (arg === "--check-heartbeat-ms") {
|
||||
options.checkHeartbeatMs = positiveIntegerOption(arg, args[index + 1], 60_000);
|
||||
index += 1;
|
||||
@@ -180,7 +189,7 @@ function emitCheckProgress(detail: Record<string, unknown>): void {
|
||||
console.error(JSON.stringify({ event: "unidesk.check.progress", ...detail }));
|
||||
}
|
||||
|
||||
async function commandItem(name: string, command: string[], timeoutMs = 30_000, env: NodeJS.ProcessEnv = process.env, heartbeatMs = defaultCheckHeartbeatMs): Promise<CheckItem> {
|
||||
async function commandItem(name: string, command: string[], timeoutMs = 30_000, env: NodeJS.ProcessEnv = process.env, heartbeatMs = defaultCheckHeartbeatMs, diagnosticPath: string | null = null): Promise<CheckItem> {
|
||||
const startedAt = Date.now();
|
||||
emitCheckProgress({ phase: "started", name, command, timeoutMs, heartbeatMs });
|
||||
const result = await runCommandObserved(command, repoRoot, {
|
||||
@@ -213,12 +222,48 @@ async function commandItem(name: string, command: string[], timeoutMs = 30_000,
|
||||
stderrBytes: result.stderrBytes,
|
||||
stdoutTruncated: result.stdoutTruncated,
|
||||
stderrTruncated: result.stderrTruncated,
|
||||
stdoutTail: result.stdout.slice(-4000),
|
||||
...(name === "typescript:scripts" ? { diagnostics: typescriptDiagnosticSummary(result.stdout, diagnosticPath) } : {}),
|
||||
stdoutTail: name === "typescript:scripts" ? "" : result.stdout.slice(-4000),
|
||||
stderrTail: result.stderr.slice(-4000),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function typescriptDiagnosticSummary(stdout: string, pathPrefix: string | null): Record<string, unknown> {
|
||||
const rows = stdout.split(/\r?\n/u).flatMap((line) => {
|
||||
const match = line.match(/^(.+?)\((\d+),(\d+)\): error (TS\d+): (.*)$/u);
|
||||
if (match === null) return [];
|
||||
return [{
|
||||
path: match[1],
|
||||
line: Number(match[2]),
|
||||
column: Number(match[3]),
|
||||
code: match[4],
|
||||
message: (match[5] ?? "").slice(0, 240),
|
||||
}];
|
||||
});
|
||||
const byPath = new Map<string, number>();
|
||||
for (const row of rows) {
|
||||
byPath.set(row.path, (byPath.get(row.path) ?? 0) + 1);
|
||||
}
|
||||
const selected = pathPrefix === null ? rows : rows.filter((row) => row.path.startsWith(pathPrefix));
|
||||
const selectedPathCounts = [...byPath.entries()].filter(([path]) => pathPrefix === null || path.startsWith(pathPrefix));
|
||||
const itemLimit = pathPrefix === null ? 5 : 8;
|
||||
const pathLimit = pathPrefix === null ? 20 : 12;
|
||||
return {
|
||||
count: rows.length,
|
||||
pathCount: byPath.size,
|
||||
pathPrefix,
|
||||
matchedCount: selected.length,
|
||||
paths: selectedPathCounts.slice(0, pathLimit).map(([path, count]) => ({ path, count })),
|
||||
items: selected.slice(0, itemLimit),
|
||||
truncated: selected.length > itemLimit || selectedPathCounts.length > pathLimit,
|
||||
next: selected.length > itemLimit || pathPrefix === null
|
||||
? "bun scripts/cli.ts check --scripts-typecheck --scripts-typecheck-path <narrower-path-prefix>"
|
||||
: null,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
|
||||
function emitCommandHeartbeat(name: string, command: string[], progress: CommandProgress): void {
|
||||
emitCheckProgress({
|
||||
phase: "running",
|
||||
@@ -425,7 +470,7 @@ export async function runChecks(config: UniDeskConfig, options: CheckOptions = d
|
||||
items.push(skippedItem("files:required-entrypoints", "required file presence scan is opt-in", "--files or --full"));
|
||||
}
|
||||
if (options.scriptsTypecheck) {
|
||||
items.push(await commandItem("typescript:scripts", ["bun", "--bun", "tsc", "-p", "scripts/tsconfig.json", "--noEmit", "--pretty", "false"], options.scriptsTypecheckTimeoutMs, process.env, options.checkHeartbeatMs));
|
||||
items.push(await commandItem("typescript:scripts", ["bun", "--bun", "tsc", "-p", "scripts/tsconfig.json", "--noEmit", "--pretty", "false"], options.scriptsTypecheckTimeoutMs, process.env, options.checkHeartbeatMs, options.scriptsTypecheckPath));
|
||||
} else {
|
||||
items.push(skippedItem("typescript:scripts", "scripts TypeScript typecheck is opt-in", "--scripts-typecheck or --full"));
|
||||
}
|
||||
|
||||
@@ -60,6 +60,8 @@ export function hwlabNodeWebProbeHelp(): Record<string, unknown> {
|
||||
"bun scripts/cli.ts web-probe observe start --node <node> --lane <lane> --target-path /workbench --sample-interval-ms 5000",
|
||||
"bun scripts/cli.ts web-probe observe start --node <node> --lane <lane> --target-path /projects/mdtodo --sample-interval-ms 5000",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type sendPrompt --text 'ping'",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type validateRealtimeFanout --profile pure-kafka-live --provider gpt.pika --text '<first-turn>' --second-text '<second-turn>'",
|
||||
"bun scripts/cli.ts web-probe observe status webobs-xxxx --command-id cmd-xxxx",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type closeMdtodoSourceConfig",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type selectMdtodoFile --filename 20260609_频率判断_用户反馈.md",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type openMdtodoReportPreview --task R1 --link R1.1",
|
||||
@@ -100,10 +102,12 @@ export function hwlabNodeWebProbeHelp(): Record<string, unknown> {
|
||||
notes: [
|
||||
"Default URL, browser proxy mode, observe/analyze thresholds, and project-management command allowlist come from config/hwlab-node-lanes.yaml webProbe.",
|
||||
"`web-probe script` is an ad-hoc exploration escape hatch; repeated/high-frequency workflows must become `web-probe observe command` types or repo-owned web-probe commands.",
|
||||
"Every script result starts with machine-readable `UNIDESK_WEB_PROBE_COMMAND_PROMOTION_HINT <json>` guidance; repo-owned generated commands report reuse instead of ad-hoc promotion.",
|
||||
"`web-probe opencode-smoke` is the repo-owned OpenCode smoke; prefer it over repeating one-off OpenCode Playwright snippets.",
|
||||
"observe is passive by default; user actions must be explicit observe command entries in control.jsonl.",
|
||||
"observe gc keeps manifest, heartbeat, control/error logs and analysis reports, and only removes dead-run raw samples/browser/network/screenshot artifacts after YAML-configured retention.",
|
||||
"After observe start, prefer observe status|command|stop|collect|analyze <id> instead of repeating --node/--lane/--state-dir.",
|
||||
"validateRealtimeFanout is a YAML-profiled background command: it gates submit on two live product subscribers plus three Kafka debug consumers, validates two turns and disconnect/reconnect without replay, and is polled by exact command id.",
|
||||
"collect views render bounded summaries from existing artifacts and do not create a second source of truth.",
|
||||
"performanceCapture records an explicit bounded Chrome CPU profile and drains LongTask/LoAF/event-loop-gap artifacts; performance-summary reads existing artifacts only.",
|
||||
"analyze is offline-only: it reads artifact JSONL and writes analysis/report.md plus analysis/report.json.",
|
||||
|
||||
@@ -144,6 +144,27 @@ export interface HwlabRuntimeWebProbeSpec {
|
||||
readonly alertThresholds?: HwlabRuntimeWebProbeAlertThresholdsSpec;
|
||||
readonly browserFreezePolicy?: HwlabRuntimeWebProbeBrowserFreezePolicySpec;
|
||||
readonly projectManagement?: HwlabRuntimeWebProbeProjectManagementSpec;
|
||||
readonly realtimeFanoutProfiles?: Readonly<Record<string, HwlabRuntimeWebProbeRealtimeFanoutProfileSpec>>;
|
||||
}
|
||||
|
||||
export interface HwlabRuntimeWebProbeRealtimeFanoutProfileSpec {
|
||||
readonly subscriberCount: 2;
|
||||
readonly debugStreams: readonly ("stdio" | "agentrun" | "hwlab")[];
|
||||
readonly productEventsPath: string;
|
||||
readonly debugEventsPath: string;
|
||||
readonly productReadyEvent: string;
|
||||
readonly debugReadyEvent: string;
|
||||
readonly businessEvent: string;
|
||||
readonly fromBeginning: false;
|
||||
readonly barrierTimeoutMs: number;
|
||||
readonly terminalTimeoutMs: number;
|
||||
readonly terminalQuietMs: number;
|
||||
readonly reconnectQuietMs: number;
|
||||
readonly forbiddenRequestPaths: readonly string[];
|
||||
readonly requiredEventTypes: Readonly<{
|
||||
agentrun: readonly string[];
|
||||
hwlab: readonly string[];
|
||||
}>;
|
||||
}
|
||||
|
||||
export interface HwlabRuntimeWebProbeAuthLoginSpec {
|
||||
@@ -427,6 +448,7 @@ export interface HwlabRuntimeKafkaEventBridgeSpec {
|
||||
};
|
||||
readonly configRef: string;
|
||||
readonly bootstrapServers: string;
|
||||
readonly stdioTopic: string;
|
||||
readonly agentRunEventTopic: string;
|
||||
readonly hwlabEventTopic: string;
|
||||
readonly clientId: string;
|
||||
@@ -1050,6 +1072,7 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun
|
||||
},
|
||||
configRef: stringField(raw, "configRef", path),
|
||||
bootstrapServers: stringField(raw, "bootstrapServers", path),
|
||||
stdioTopic: stringField(raw, "stdioTopic", path),
|
||||
agentRunEventTopic: stringField(raw, "agentRunEventTopic", path),
|
||||
hwlabEventTopic: stringField(raw, "hwlabEventTopic", path),
|
||||
clientId: stringField(raw, "clientId", path),
|
||||
@@ -1250,6 +1273,63 @@ function webProbeConfig(value: unknown, path: string): HwlabRuntimeWebProbeSpec
|
||||
...(raw.alertThresholds === undefined ? {} : { alertThresholds: webProbeAlertThresholdsConfig(raw.alertThresholds, `${path}.alertThresholds`) }),
|
||||
...(raw.browserFreezePolicy === undefined ? {} : { browserFreezePolicy: webProbeBrowserFreezePolicyConfig(raw.browserFreezePolicy, `${path}.browserFreezePolicy`) }),
|
||||
...(raw.projectManagement === undefined ? {} : { projectManagement: webProbeProjectManagementConfig(raw.projectManagement, `${path}.projectManagement`) }),
|
||||
...(raw.realtimeFanoutProfiles === undefined ? {} : { realtimeFanoutProfiles: webProbeRealtimeFanoutProfilesConfig(raw.realtimeFanoutProfiles, `${path}.realtimeFanoutProfiles`) }),
|
||||
};
|
||||
}
|
||||
|
||||
function webProbeRealtimeFanoutProfilesConfig(value: unknown, path: string): Readonly<Record<string, HwlabRuntimeWebProbeRealtimeFanoutProfileSpec>> {
|
||||
const raw = asRecord(value, path);
|
||||
const entries = Object.entries(raw).map(([id, item]) => {
|
||||
if (!/^[A-Za-z0-9_.-]+$/u.test(id)) throw new Error(`${path} profile id ${id} must use safe identifier characters`);
|
||||
return [id, webProbeRealtimeFanoutProfileConfig(item, `${path}.${id}`)] as const;
|
||||
});
|
||||
if (entries.length === 0) throw new Error(`${path} must declare at least one profile`);
|
||||
return Object.fromEntries(entries);
|
||||
}
|
||||
|
||||
function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): HwlabRuntimeWebProbeRealtimeFanoutProfileSpec {
|
||||
const raw = asRecord(value, path);
|
||||
const subscriberCount = numberField(raw, "subscriberCount", path);
|
||||
if (subscriberCount !== 2) throw new Error(`${path}.subscriberCount must be 2`);
|
||||
const debugStreams = nonEmptyStringArrayField(raw, "debugStreams", path);
|
||||
if (debugStreams.length !== 3 || new Set(debugStreams).size !== 3 || debugStreams.some((stream) => !["stdio", "agentrun", "hwlab"].includes(stream))) {
|
||||
throw new Error(`${path}.debugStreams must contain stdio, agentrun, and hwlab exactly once`);
|
||||
}
|
||||
const fromBeginning = booleanField(raw, "fromBeginning", path);
|
||||
if (fromBeginning !== false) throw new Error(`${path}.fromBeginning must be false for live-only validation`);
|
||||
const requiredEventTypes = asRecord(raw.requiredEventTypes, `${path}.requiredEventTypes`);
|
||||
const forbiddenRequestPaths = nonEmptyStringArrayField(raw, "forbiddenRequestPaths", path);
|
||||
if (new Set(forbiddenRequestPaths).size !== forbiddenRequestPaths.length) throw new Error(`${path}.forbiddenRequestPaths must not contain duplicates`);
|
||||
for (const requestPath of forbiddenRequestPaths) {
|
||||
if (!requestPath.startsWith("/")) throw new Error(`${path}.forbiddenRequestPaths entries must be absolute paths`);
|
||||
}
|
||||
const productEventsPath = stringField(raw, "productEventsPath", path);
|
||||
const debugEventsPath = stringField(raw, "debugEventsPath", path);
|
||||
for (const [field, routePath] of [["productEventsPath", productEventsPath], ["debugEventsPath", debugEventsPath]] as const) {
|
||||
if (!routePath.startsWith("/")) throw new Error(`${path}.${field} must be an absolute path`);
|
||||
}
|
||||
const agentrunEventTypes = nonEmptyStringArrayField(requiredEventTypes, "agentrun", `${path}.requiredEventTypes`);
|
||||
const hwlabEventTypes = nonEmptyStringArrayField(requiredEventTypes, "hwlab", `${path}.requiredEventTypes`);
|
||||
if (new Set(agentrunEventTypes).size !== agentrunEventTypes.length) throw new Error(`${path}.requiredEventTypes.agentrun must not contain duplicates`);
|
||||
if (new Set(hwlabEventTypes).size !== hwlabEventTypes.length) throw new Error(`${path}.requiredEventTypes.hwlab must not contain duplicates`);
|
||||
return {
|
||||
subscriberCount: 2,
|
||||
debugStreams: debugStreams as ("stdio" | "agentrun" | "hwlab")[],
|
||||
productEventsPath,
|
||||
debugEventsPath,
|
||||
productReadyEvent: stringField(raw, "productReadyEvent", path),
|
||||
debugReadyEvent: stringField(raw, "debugReadyEvent", path),
|
||||
businessEvent: stringField(raw, "businessEvent", path),
|
||||
fromBeginning: false,
|
||||
barrierTimeoutMs: boundedIntegerField(raw, "barrierTimeoutMs", path, 1_000, 120_000),
|
||||
terminalTimeoutMs: boundedIntegerField(raw, "terminalTimeoutMs", path, 1_000, 600_000),
|
||||
terminalQuietMs: boundedIntegerField(raw, "terminalQuietMs", path, 250, 10_000),
|
||||
reconnectQuietMs: boundedIntegerField(raw, "reconnectQuietMs", path, 250, 30_000),
|
||||
forbiddenRequestPaths,
|
||||
requiredEventTypes: {
|
||||
agentrun: agentrunEventTypes,
|
||||
hwlab: hwlabEventTypes,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -101,6 +101,7 @@ async function processCommand(command) {
|
||||
throwOnActionMismatch: true,
|
||||
expectedActionWaitMs: command.expectedActionWaitMs,
|
||||
}), "sendPrompt");
|
||||
case "validateRealtimeFanout": return validateRealtimeFanout(command);
|
||||
case "steer": return withObserverSync(await sendPrompt(String(command.text || ""), { expectedAction: "steer", responsePath: "/v1/agent/chat/steer", noActiveReason: "steer-no-active-turn", expectedActionWaitMs: command.expectedActionWaitMs }), "steer");
|
||||
case "cancel": return withObserverSync(await cancelRunningTurn(), "cancel");
|
||||
case "selectProvider": return withObserverSync(await selectProvider(String(command.provider || command.value || command.text || "")), "selectProvider");
|
||||
|
||||
@@ -0,0 +1,971 @@
|
||||
// SPEC: PJ2026-0104010803 Workbench realtime visibility.
|
||||
// Responsibility: Long-running, typed realtime fanout validation inside the persistent Web observer.
|
||||
|
||||
export function nodeWebObserveRunnerRealtimeSource(): string {
|
||||
return String.raw`
|
||||
function parseRealtimeFanoutProfiles(raw) {
|
||||
if (!raw) return {};
|
||||
try {
|
||||
const parsed = JSON.parse(raw);
|
||||
return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? parsed : {};
|
||||
} catch {
|
||||
return {};
|
||||
}
|
||||
}
|
||||
|
||||
async function validateRealtimeFanout(command) {
|
||||
const profileId = String(command.profile || "").trim();
|
||||
const profile = realtimeFanoutProfiles[profileId];
|
||||
if (!profile || typeof profile !== "object") throw new Error("validateRealtimeFanout profile is not declared by the owning YAML: " + profileId);
|
||||
const provider = String(command.provider || "").trim();
|
||||
const firstPrompt = String(command.text || "").trim();
|
||||
const secondPrompt = String(command.secondText || "").trim();
|
||||
if (!provider) throw new Error("validateRealtimeFanout requires provider");
|
||||
if (!firstPrompt || !secondPrompt) throw new Error("validateRealtimeFanout requires first and second prompt");
|
||||
const effective = realtimeFanoutEffectiveProfile(profile, command.durationMs);
|
||||
const reportDir = path.join(stateDir, "artifacts", "realtime-fanout", safeId(command.id));
|
||||
await mkdir(reportDir, { recursive: true, mode: 0o700 });
|
||||
const network = [];
|
||||
const requestListener = (request) => {
|
||||
let url;
|
||||
try { url = new URL(request.url()); } catch { return; }
|
||||
if (url.origin !== new URL(baseUrl).origin) return;
|
||||
network.push({
|
||||
at: Date.now(),
|
||||
method: request.method(),
|
||||
path: url.pathname,
|
||||
sessionId: url.searchParams.get("sessionId"),
|
||||
traceId: url.searchParams.get("traceId"),
|
||||
afterSeq: url.searchParams.get("afterSeq"),
|
||||
lastEventId: request.headers()["last-event-id"] || null,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
if (network.length > 1000) network.splice(0, network.length - 1000);
|
||||
};
|
||||
page.on("request", requestListener);
|
||||
let subscriberA = null;
|
||||
let subscriberB = null;
|
||||
let subscriberB2 = null;
|
||||
let debug = null;
|
||||
let createdSessionId = null;
|
||||
const report = {
|
||||
ok: false,
|
||||
contract: "web-probe-realtime-fanout-v1",
|
||||
commandId: command.id,
|
||||
profile: profileId,
|
||||
provider,
|
||||
startedAt: new Date().toISOString(),
|
||||
promptEvidence: {
|
||||
first: { hash: sha256Text(firstPrompt), bytes: Buffer.byteLength(firstPrompt) },
|
||||
second: { hash: sha256Text(secondPrompt), bytes: Buffer.byteLength(secondPrompt) },
|
||||
valuesRedacted: true,
|
||||
},
|
||||
valuesRedacted: true,
|
||||
};
|
||||
try {
|
||||
const providerSelection = await selectProvider(provider);
|
||||
const session = await createSessionFromUi();
|
||||
const sessionId = session.sessionId;
|
||||
createdSessionId = sessionId;
|
||||
report.sessionId = sessionId;
|
||||
report.providerSelection = sanitize(providerSelection);
|
||||
const storageState = await context.storageState();
|
||||
subscriberA = await realtimeNewSubscriber(storageState, sessionId, "subscriber-a", effective);
|
||||
subscriberB = await realtimeNewSubscriber(storageState, sessionId, "subscriber-b", effective);
|
||||
debug = await realtimeNewDebugContext(storageState);
|
||||
const connectedA = await realtimeWaitProductConnected(subscriberA, effective.barrierTimeoutMs);
|
||||
const connectedB = await realtimeWaitProductConnected(subscriberB, effective.barrierTimeoutMs);
|
||||
realtimeAssertLiveConnected(connectedA, sessionId, "subscriber-a", effective);
|
||||
realtimeAssertLiveConnected(connectedB, sessionId, "subscriber-b", effective);
|
||||
await appendJsonl(files.control, eventRecord("realtime-fanout-product-ready", {
|
||||
profile: profileId,
|
||||
sessionId,
|
||||
subscriberCount: 2,
|
||||
liveOnly: true,
|
||||
replay: false,
|
||||
valuesRedacted: true,
|
||||
}));
|
||||
|
||||
const first = await realtimeRunTurn({
|
||||
command,
|
||||
turn: 1,
|
||||
prompt: firstPrompt,
|
||||
sessionId,
|
||||
subscribers: [subscriberA, subscriberB],
|
||||
debug,
|
||||
profile: effective,
|
||||
});
|
||||
const bBeforeDisconnect = await realtimeProductSnapshot(subscriberB);
|
||||
realtimeAssertStableProductTransport(bBeforeDisconnect, "subscriber-b-before-disconnect");
|
||||
if (bBeforeDisconnect.records.some((item) => item.traceId === first.traceId && item.terminal === true)) {
|
||||
throw new Error("first turn reached terminal before subscriber B could be disconnected; use a prompt with an observable live interval");
|
||||
}
|
||||
const bBeforeShot = await realtimeCaptureSubscriberScreenshot(subscriberB.page, reportDir, "subscriber-b-before-disconnect.png", {
|
||||
title: "Subscriber B before disconnect",
|
||||
sessionId,
|
||||
traceId: first.traceId,
|
||||
eventCount: bBeforeDisconnect.records.length,
|
||||
terminalSeen: bBeforeDisconnect.records.some((item) => item.terminal === true),
|
||||
valuesRedacted: true,
|
||||
});
|
||||
await subscriberB.context.close();
|
||||
subscriberB = null;
|
||||
const firstTerminal = await realtimeWaitTraceEvent(subscriberA, first.traceId, true, effective.terminalTimeoutMs);
|
||||
await realtimeWaitUiTerminal(first.traceId, effective.terminalTimeoutMs);
|
||||
const firstStable = await realtimeWaitTurnStable(debug, first.debugKey, first.traceId, [subscriberA], effective);
|
||||
const firstDebug = firstStable.debug;
|
||||
realtimeEnrichTurnExecutionIds(first, firstDebug);
|
||||
const firstProductA = firstStable.products[0];
|
||||
realtimeAssertStableProductTransport(firstProductA, "subscriber-a-first-terminal");
|
||||
realtimeValidateTurnEvidence({
|
||||
turn: 1,
|
||||
traceId: first.traceId,
|
||||
sessionId,
|
||||
runId: first.runId,
|
||||
commandId: first.commandId,
|
||||
debug: firstDebug,
|
||||
terminalSnapshots: [firstProductA],
|
||||
preTerminalSnapshots: [bBeforeDisconnect],
|
||||
profile: effective,
|
||||
});
|
||||
await realtimeCloseDebugStreams(debug, first.debugKey);
|
||||
|
||||
subscriberB2 = await realtimeNewSubscriber(storageState, sessionId, "subscriber-b-reconnected", effective);
|
||||
const bReconnected = await realtimeWaitProductConnected(subscriberB2, effective.barrierTimeoutMs);
|
||||
realtimeAssertLiveConnected(bReconnected, sessionId, "subscriber-b-reconnected", effective);
|
||||
await sleep(effective.reconnectQuietMs);
|
||||
const reconnectBeforeSecond = await realtimeProductSnapshot(subscriberB2);
|
||||
realtimeAssertStableProductTransport(reconnectBeforeSecond, "subscriber-b-reconnected-before-second-turn");
|
||||
if (reconnectBeforeSecond.records.some((item) => item.traceId === first.traceId)) {
|
||||
throw new Error("reconnected subscriber replayed first-turn events");
|
||||
}
|
||||
|
||||
const second = await realtimeRunTurn({
|
||||
command,
|
||||
turn: 2,
|
||||
prompt: secondPrompt,
|
||||
sessionId,
|
||||
subscribers: [subscriberA, subscriberB2],
|
||||
debug,
|
||||
profile: effective,
|
||||
});
|
||||
if (second.traceId === first.traceId) throw new Error("second turn reused the first business traceId");
|
||||
const secondTerminalA = await realtimeWaitTraceEvent(subscriberA, second.traceId, true, effective.terminalTimeoutMs);
|
||||
const secondTerminalB = await realtimeWaitTraceEvent(subscriberB2, second.traceId, true, effective.terminalTimeoutMs);
|
||||
await realtimeWaitUiTerminal(second.traceId, effective.terminalTimeoutMs);
|
||||
const secondStable = await realtimeWaitTurnStable(debug, second.debugKey, second.traceId, [subscriberA, subscriberB2], effective);
|
||||
const secondDebug = secondStable.debug;
|
||||
realtimeEnrichTurnExecutionIds(second, secondDebug);
|
||||
const [secondProductA, secondProductB] = secondStable.products;
|
||||
realtimeAssertStableProductTransport(secondProductA, "subscriber-a-second-terminal");
|
||||
realtimeAssertStableProductTransport(secondProductB, "subscriber-b-reconnected-second-terminal");
|
||||
if (secondProductB.records.some((item) => item.traceId === first.traceId)) {
|
||||
throw new Error("reconnected subscriber received first-turn events after the second submit");
|
||||
}
|
||||
realtimeValidateTurnEvidence({
|
||||
turn: 2,
|
||||
traceId: second.traceId,
|
||||
sessionId,
|
||||
runId: second.runId,
|
||||
commandId: second.commandId,
|
||||
debug: secondDebug,
|
||||
terminalSnapshots: [secondProductA, secondProductB],
|
||||
preTerminalSnapshots: [],
|
||||
profile: effective,
|
||||
});
|
||||
await realtimeCloseDebugStreams(debug, second.debugKey);
|
||||
|
||||
const productA = await realtimeProductSnapshot(subscriberA);
|
||||
const productB2 = await realtimeProductSnapshot(subscriberB2);
|
||||
realtimeAssertStableProductTransport(productA, "subscriber-a-final");
|
||||
realtimeAssertStableProductTransport(productB2, "subscriber-b-reconnected-final");
|
||||
const mainStreams = network.filter((item) => item.path === effective.productEventsPath && item.sessionId === sessionId);
|
||||
const forbidden = network.filter((item) => realtimeForbiddenRequest(item, effective.forbiddenRequestPaths));
|
||||
if (mainStreams.length !== 1) throw new Error("main Workbench session SSE connection count is " + mainStreams.length + ", expected 1");
|
||||
if (mainStreams[0].traceId !== null) throw new Error("main Workbench switched to trace-scoped SSE");
|
||||
if (mainStreams[0].afterSeq !== null || mainStreams[0].lastEventId !== null) throw new Error("main Workbench requested replay cursor");
|
||||
if (forbidden.length > 0) throw new Error("forbidden recovery or polling requests observed: " + forbidden.map((item) => item.path).join(","));
|
||||
|
||||
const summaryScreenshot = await realtimeCaptureSubscriberScreenshot(debug.page, reportDir, "realtime-fanout-summary.png", {
|
||||
title: "Pure Kafka live fanout validated",
|
||||
profile: profileId,
|
||||
sessionId,
|
||||
traceIds: [first.traceId, second.traceId],
|
||||
productSubscriberCount: 2,
|
||||
debugStreams: effective.debugStreams,
|
||||
forbiddenRequestCount: forbidden.length,
|
||||
replayedEventCount: 0,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
const subscriberAScreenshot = await realtimeCaptureSubscriberScreenshot(subscriberA.page, reportDir, "subscriber-a-final.png", {
|
||||
title: "Subscriber A continuous live connection",
|
||||
sessionId,
|
||||
traceIds: [first.traceId, second.traceId],
|
||||
eventCount: productA.records.length,
|
||||
terminalCount: productA.records.filter((item) => item.terminal === true).length,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
const subscriberB2Screenshot = await realtimeCaptureSubscriberScreenshot(subscriberB2.page, reportDir, "subscriber-b-reconnected-final.png", {
|
||||
title: "Subscriber B reconnected live-only",
|
||||
sessionId,
|
||||
traceIds: [second.traceId],
|
||||
eventCount: productB2.records.length,
|
||||
firstTraceReplayCount: productB2.records.filter((item) => item.traceId === first.traceId).length,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
const turns = [
|
||||
realtimeTurnSummary(first, firstDebug, productA, firstTerminal),
|
||||
realtimeTurnSummary(second, secondDebug, productB2, secondTerminalB),
|
||||
];
|
||||
Object.assign(report, {
|
||||
ok: true,
|
||||
status: "passed",
|
||||
completedAt: new Date().toISOString(),
|
||||
turns,
|
||||
productSse: {
|
||||
mainSessionConnectionCount: mainStreams.length,
|
||||
sessionScoped: true,
|
||||
afterSeqCount: 0,
|
||||
lastEventIdCount: 0,
|
||||
subscriberAEventCount: productA.records.length,
|
||||
subscriberBReconnectEventCount: productB2.records.length,
|
||||
subscriberBFirstTraceReplayCount: 0,
|
||||
valuesRedacted: true,
|
||||
},
|
||||
liveContract: {
|
||||
expectedKafka: effective.expectedKafka,
|
||||
subscriberA: realtimeConnectedSummary(connectedA),
|
||||
subscriberBBeforeDisconnect: realtimeConnectedSummary(connectedB),
|
||||
subscriberBAfterReconnect: realtimeConnectedSummary(bReconnected),
|
||||
valuesRedacted: true,
|
||||
},
|
||||
forbiddenRequests: { count: forbidden.length, rows: forbidden, valuesRedacted: true },
|
||||
screenshots: [bBeforeShot, summaryScreenshot, subscriberAScreenshot, subscriberB2Screenshot],
|
||||
warmRunnerReused: first.runnerId && second.runnerId ? first.runnerId === second.runnerId : null,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
const artifacts = await realtimeWriteEvidenceArtifacts(reportDir, network, [firstDebug, secondDebug]);
|
||||
report.artifacts = artifacts;
|
||||
const reportArtifact = await realtimeWriteReport(reportDir, report);
|
||||
await appendJsonl(files.control, eventRecord("realtime-fanout-validated", {
|
||||
profile: profileId,
|
||||
sessionId,
|
||||
traceIds: turns.map((item) => item.traceId),
|
||||
runIds: turns.map((item) => item.runId),
|
||||
commandIds: turns.map((item) => item.commandId),
|
||||
reportPath: reportArtifact.path,
|
||||
reportSha256: reportArtifact.sha256,
|
||||
valuesRedacted: true,
|
||||
}));
|
||||
const observer = await syncObserverPageToControlSession("validateRealtimeFanout", sessionId);
|
||||
return {
|
||||
ok: true,
|
||||
status: "passed",
|
||||
profile: profileId,
|
||||
sessionId,
|
||||
traceIds: turns.map((item) => item.traceId),
|
||||
runIds: turns.map((item) => item.runId),
|
||||
commandIds: turns.map((item) => item.commandId),
|
||||
warmRunnerReused: report.warmRunnerReused,
|
||||
forbiddenRequestCount: 0,
|
||||
replayedEventCount: 0,
|
||||
reportPath: reportArtifact.path,
|
||||
reportSha256: reportArtifact.sha256,
|
||||
screenshots: report.screenshots,
|
||||
observer,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
} catch (error) {
|
||||
Object.assign(report, {
|
||||
ok: false,
|
||||
status: "failed",
|
||||
completedAt: new Date().toISOString(),
|
||||
failure: errorSummary(error),
|
||||
valuesRedacted: true,
|
||||
});
|
||||
const partialDebug = debug ? await realtimeAllDebugSnapshots(debug).catch(() => []) : [];
|
||||
const partialProducts = await realtimePartialProductEvidence([subscriberA, subscriberB, subscriberB2]);
|
||||
const partialArtifacts = await realtimeWriteEvidenceArtifacts(reportDir, network, partialDebug).catch(() => null);
|
||||
report.partialEvidence = { products: partialProducts, debug: partialDebug.map(realtimeDebugSnapshotSummary), valuesRedacted: true };
|
||||
if (partialArtifacts) report.artifacts = partialArtifacts;
|
||||
const reportArtifact = await realtimeWriteReport(reportDir, report).catch(() => null);
|
||||
if (createdSessionId) {
|
||||
await syncObserverPageToControlSession("validateRealtimeFanout-failed", createdSessionId)
|
||||
.catch((syncError) => appendJsonl(files.errors, eventRecord("realtime-fanout-observer-sync-error", { commandId: command.id, error: errorSummary(syncError), valuesRedacted: true })));
|
||||
}
|
||||
const wrapped = error instanceof Error ? error : new Error(String(error));
|
||||
wrapped.details = {
|
||||
...(wrapped.details || {}),
|
||||
profile: profileId,
|
||||
reportPath: reportArtifact?.path || null,
|
||||
reportSha256: reportArtifact?.sha256 || null,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
throw wrapped;
|
||||
} finally {
|
||||
page.off("request", requestListener);
|
||||
if (debug) await realtimeCloseDebugStreams(debug, null).catch(() => {});
|
||||
if (debug) await debug.context.close().catch(() => {});
|
||||
if (subscriberA) await subscriberA.context.close().catch(() => {});
|
||||
if (subscriberB) await subscriberB.context.close().catch(() => {});
|
||||
if (subscriberB2) await subscriberB2.context.close().catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
function realtimeFanoutEffectiveProfile(profile, durationMs) {
|
||||
const debugStreams = Array.isArray(profile.debugStreams) ? profile.debugStreams.map(String) : [];
|
||||
if (Number(profile.subscriberCount) !== 2) throw new Error("realtime fanout profile subscriberCount must be 2");
|
||||
if (debugStreams.length !== 3 || new Set(debugStreams).size !== 3 || !["stdio", "agentrun", "hwlab"].every((item) => debugStreams.includes(item))) {
|
||||
throw new Error("realtime fanout profile debugStreams must be stdio, agentrun, and hwlab");
|
||||
}
|
||||
if (profile.fromBeginning !== false) throw new Error("realtime fanout profile fromBeginning must be false");
|
||||
const expectedKafka = profile.expectedKafka && typeof profile.expectedKafka === "object" ? profile.expectedKafka : null;
|
||||
if (!expectedKafka?.topics || !expectedKafka?.groups || !expectedKafka?.capabilities) throw new Error("realtime fanout profile lacks YAML-derived Kafka expectations");
|
||||
if (expectedKafka.capabilities.directPublish !== true || expectedKafka.capabilities.liveKafkaSse !== true) {
|
||||
throw new Error("realtime fanout profile requires YAML-enabled directPublish and liveKafkaSse capabilities");
|
||||
}
|
||||
const terminalTimeoutMs = Number.isFinite(Number(durationMs))
|
||||
? Math.min(Number(profile.terminalTimeoutMs), Math.max(1000, Number(durationMs)))
|
||||
: Number(profile.terminalTimeoutMs);
|
||||
return {
|
||||
...profile,
|
||||
debugStreams,
|
||||
barrierTimeoutMs: boundedInteger(profile.barrierTimeoutMs, 45000, 1000, 120000),
|
||||
terminalTimeoutMs: boundedInteger(terminalTimeoutMs, 480000, 1000, 600000),
|
||||
terminalQuietMs: boundedInteger(profile.terminalQuietMs, 2000, 250, 10000),
|
||||
reconnectQuietMs: boundedInteger(profile.reconnectQuietMs, 3000, 250, 30000),
|
||||
forbiddenRequestPaths: Array.isArray(profile.forbiddenRequestPaths) ? profile.forbiddenRequestPaths.map(String) : [],
|
||||
requiredEventTypes: profile.requiredEventTypes && typeof profile.requiredEventTypes === "object" ? profile.requiredEventTypes : {},
|
||||
expectedKafka: sanitize(expectedKafka),
|
||||
};
|
||||
}
|
||||
|
||||
async function realtimeNewSubscriber(storageState, sessionId, label, profile) {
|
||||
const subscriberContext = await browser.newContext({ storageState, viewport });
|
||||
const subscriberPage = await subscriberContext.newPage();
|
||||
await subscriberPage.goto(new URL("/health/live", baseUrl).toString(), { waitUntil: "domcontentloaded", timeout: profile.barrierTimeoutMs });
|
||||
const key = label + "-" + sessionId;
|
||||
await subscriberPage.evaluate(({ key: streamKey, sessionId: scopedSessionId, profile: inputProfile }) => {
|
||||
const root = window.__unideskRealtimeFanoutProducts ??= {};
|
||||
const state = { connected: null, connectedEventCount: 0, openCount: 0, records: [], errors: [], source: null };
|
||||
root[streamKey] = state;
|
||||
const firstText = (...values) => {
|
||||
for (const value of values) if (typeof value === "string" && value.trim()) return value.trim();
|
||||
return null;
|
||||
};
|
||||
const fingerprint = (text) => {
|
||||
const bytes = new TextEncoder().encode(String(text));
|
||||
let hash = 2166136261;
|
||||
for (const byte of bytes) hash = Math.imul((hash ^ byte) >>> 0, 16777619) >>> 0;
|
||||
return "fnv1a32:" + hash.toString(16).padStart(8, "0") + ":" + bytes.length;
|
||||
};
|
||||
const source = new EventSource(inputProfile.productEventsPath + "?sessionId=" + encodeURIComponent(scopedSessionId), { withCredentials: true });
|
||||
state.source = source;
|
||||
source.onopen = () => { state.openCount += 1; };
|
||||
const record = (eventName, raw) => {
|
||||
let value;
|
||||
try { value = JSON.parse(raw); } catch { return; }
|
||||
const nested = value?.event && typeof value.event === "object" ? value.event : {};
|
||||
const context = value?.context && typeof value.context === "object" ? value.context : {};
|
||||
const eventType = firstText(nested.eventType, nested.type, value.eventType, value.type);
|
||||
state.records.push({
|
||||
eventName,
|
||||
at: Date.now(),
|
||||
schema: firstText(value.schema),
|
||||
eventId: firstText(value.eventId, nested.id),
|
||||
sourceEventId: firstText(value.sourceEventId, nested.sourceEventId),
|
||||
eventType,
|
||||
sourceSeq: Number.isFinite(Number(value.sourceSeq)) ? Number(value.sourceSeq) : Number.isFinite(Number(nested.sourceSeq)) ? Number(nested.sourceSeq) : null,
|
||||
traceId: firstText(value.traceId, nested.traceId, context.traceId),
|
||||
hwlabSessionId: firstText(value.hwlabSessionId, value.sessionId, nested.hwlabSessionId, nested.sessionId, context.sessionId),
|
||||
runId: firstText(value.runId, nested.runId, context.runId),
|
||||
commandId: firstText(value.commandId, nested.commandId, context.commandId),
|
||||
terminal: value.terminal === true || nested.terminal === true || eventType === "terminal" || eventType === "terminal_status",
|
||||
status: firstText(nested.status, value.status),
|
||||
envelopeFingerprint: fingerprint(raw),
|
||||
valuesRedacted: true,
|
||||
});
|
||||
if (state.records.length > 800) state.records.splice(0, state.records.length - 800);
|
||||
};
|
||||
source.addEventListener(inputProfile.productReadyEvent, (event) => {
|
||||
state.connectedEventCount += 1;
|
||||
try { state.connected = JSON.parse(event.data); } catch { state.connected = { parseError: true }; }
|
||||
});
|
||||
source.addEventListener(inputProfile.businessEvent, (event) => record(inputProfile.businessEvent, event.data));
|
||||
source.addEventListener("workbench.error", (event) => state.errors.push({ at: Date.now(), event: "workbench.error", dataBytes: String(event.data || "").length }));
|
||||
source.onerror = () => state.errors.push({ at: Date.now(), event: "transport.error" });
|
||||
}, { key, sessionId, profile: { productEventsPath: profile.productEventsPath, productReadyEvent: profile.productReadyEvent, businessEvent: profile.businessEvent } });
|
||||
return { label, key, context: subscriberContext, page: subscriberPage };
|
||||
}
|
||||
|
||||
async function realtimeNewDebugContext(storageState) {
|
||||
const debugContext = await browser.newContext({ storageState, viewport });
|
||||
const debugPage = await debugContext.newPage();
|
||||
await debugPage.goto(new URL("/health/live", baseUrl).toString(), { waitUntil: "domcontentloaded", timeout: 30000 });
|
||||
return { context: debugContext, page: debugPage };
|
||||
}
|
||||
|
||||
async function realtimeRunTurn(input) {
|
||||
const turn = input.turn;
|
||||
const prompt = input.prompt;
|
||||
await page.locator("#command-input").fill(prompt);
|
||||
await page.locator("#command-send").waitFor({ state: "visible", timeout: input.profile.barrierTimeoutMs });
|
||||
await page.waitForFunction(() => {
|
||||
const button = document.querySelector("#command-send");
|
||||
return Boolean(button && !button.disabled && button.getAttribute("data-action") === "turn");
|
||||
}, null, { timeout: input.profile.barrierTimeoutMs });
|
||||
let capturedResolve;
|
||||
let releaseResolve;
|
||||
let releasedToServer = false;
|
||||
const capturedPromise = new Promise((resolve) => { capturedResolve = resolve; });
|
||||
const releasePromise = new Promise((resolve) => { releaseResolve = resolve; });
|
||||
const handler = async (route) => {
|
||||
const request = route.request();
|
||||
let body = {};
|
||||
try { body = request.postDataJSON(); } catch {}
|
||||
capturedResolve({ traceId: String(body?.traceId || ""), sessionId: String(body?.sessionId || ""), provider: String(body?.providerProfile || "") });
|
||||
const releaseAction = await releasePromise;
|
||||
if (releaseAction === "continue") await route.continue();
|
||||
else await route.abort("aborted");
|
||||
};
|
||||
await page.route("**/v1/agent/chat", handler);
|
||||
try {
|
||||
const admissionPromise = page.waitForResponse((response) => {
|
||||
let url;
|
||||
try { url = new URL(response.url()); } catch { return false; }
|
||||
return response.request().method() === "POST" && url.pathname === "/v1/agent/chat";
|
||||
}, { timeout: Math.min(input.profile.terminalTimeoutMs, input.profile.barrierTimeoutMs + 30000) })
|
||||
.then((response) => ({ response, error: null }))
|
||||
.catch((error) => ({ response: null, error }));
|
||||
await page.locator("#command-send").click();
|
||||
const captured = await withHardTimeout(capturedPromise, input.profile.barrierTimeoutMs, "realtime turn request capture timed out");
|
||||
if (!/^trc_[A-Za-z0-9_-]+$/u.test(captured.traceId)) throw new Error("realtime turn did not pre-generate a traceId");
|
||||
if (captured.sessionId !== input.sessionId) throw new Error("realtime turn sessionId mismatch");
|
||||
if (captured.provider !== String(input.command.provider || "")) throw new Error("realtime turn provider mismatch");
|
||||
const debugKey = "turn-" + turn + "-" + captured.traceId;
|
||||
await realtimeOpenDebugStreams(input.debug, debugKey, captured.traceId, input.profile);
|
||||
const debugConnected = await realtimeWaitDebugConnected(input.debug, debugKey, input.profile);
|
||||
for (const stream of input.profile.debugStreams) {
|
||||
const connected = debugConnected[stream];
|
||||
if (connected?.consumerReady !== true) throw new Error(stream + " debug consumer is not ready");
|
||||
if (!String(connected?.groupId || "").includes("-debug-sse-")) throw new Error(stream + " debug group is not isolated");
|
||||
if (connected?.topic !== input.profile.expectedKafka.topics[stream]) throw new Error(stream + " debug topic does not match the owning YAML");
|
||||
}
|
||||
for (const subscriber of input.subscribers) {
|
||||
const connected = await realtimeWaitProductConnected(subscriber, input.profile.barrierTimeoutMs);
|
||||
realtimeAssertLiveConnected(connected, input.sessionId, subscriber.label, input.profile);
|
||||
}
|
||||
await appendJsonl(files.control, eventRecord("realtime-fanout-ready-barrier", {
|
||||
turn,
|
||||
traceId: captured.traceId,
|
||||
productSubscriberCount: input.subscribers.length,
|
||||
debugStreams: input.profile.debugStreams,
|
||||
topics: Object.fromEntries(input.profile.debugStreams.map((stream) => [stream, debugConnected[stream]?.topic || null])),
|
||||
groups: Object.fromEntries(input.profile.debugStreams.map((stream) => [stream, debugConnected[stream]?.groupId || null])),
|
||||
valuesRedacted: true,
|
||||
}));
|
||||
releasedToServer = true;
|
||||
releaseResolve("continue");
|
||||
const admissionOutcome = await admissionPromise;
|
||||
if (!admissionOutcome.response) throw new Error("realtime turn admission response failed: " + String(admissionOutcome.error?.message || admissionOutcome.error || "unknown"));
|
||||
const admission = admissionOutcome.response;
|
||||
if (admission.status() < 200 || admission.status() >= 300) throw new Error("realtime turn admission returned HTTP " + admission.status());
|
||||
let admissionBody = {};
|
||||
try { admissionBody = await admission.json(); } catch {}
|
||||
const canonicalTraceId = realtimeFirstText(admissionBody?.traceId, admissionBody?.result?.traceId, captured.traceId);
|
||||
if (canonicalTraceId !== captured.traceId) throw new Error("canonical trace changed after ready barrier");
|
||||
const nonTerminal = [];
|
||||
for (const subscriber of input.subscribers) {
|
||||
nonTerminal.push(await realtimeWaitTraceEvent(subscriber, canonicalTraceId, false, input.profile.terminalTimeoutMs));
|
||||
}
|
||||
const ids = await realtimeWaitTurnIds(input.debug, debugKey, canonicalTraceId, input.profile.terminalTimeoutMs);
|
||||
await appendJsonl(files.control, eventRecord("realtime-fanout-pre-terminal", {
|
||||
turn,
|
||||
traceId: canonicalTraceId,
|
||||
runId: ids.runId,
|
||||
commandId: ids.commandId,
|
||||
subscriberCount: nonTerminal.length,
|
||||
valuesRedacted: true,
|
||||
}));
|
||||
return { turn, traceId: canonicalTraceId, sessionId: input.sessionId, runId: ids.runId, commandId: ids.commandId, runnerId: ids.runnerId, attemptId: ids.attemptId, debugKey, admissionStatus: admission.status(), valuesRedacted: true };
|
||||
} finally {
|
||||
if (!releasedToServer) releaseResolve("abort");
|
||||
await page.unroute("**/v1/agent/chat", handler).catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
async function realtimeOpenDebugStreams(debug, key, traceId, profile) {
|
||||
await debug.page.evaluate(({ key: streamKey, traceId: scopedTraceId, profile: inputProfile }) => {
|
||||
const root = window.__unideskRealtimeFanoutDebug ??= {};
|
||||
const state = { connected: {}, connectedEventCount: {}, openCount: {}, records: {}, errors: {}, sources: {} };
|
||||
root[streamKey] = state;
|
||||
const firstText = (...values) => {
|
||||
for (const value of values) if (typeof value === "string" && value.trim()) return value.trim();
|
||||
return null;
|
||||
};
|
||||
const finite = (...values) => {
|
||||
for (const value of values) {
|
||||
const number = Number(value);
|
||||
if (Number.isFinite(number)) return number;
|
||||
}
|
||||
return null;
|
||||
};
|
||||
const fingerprint = (text) => {
|
||||
const bytes = new TextEncoder().encode(String(text));
|
||||
let hash = 2166136261;
|
||||
for (const byte of bytes) hash = Math.imul((hash ^ byte) >>> 0, 16777619) >>> 0;
|
||||
return "fnv1a32:" + hash.toString(16).padStart(8, "0") + ":" + bytes.length;
|
||||
};
|
||||
for (const stream of inputProfile.debugStreams) {
|
||||
state.records[stream] = [];
|
||||
state.errors[stream] = [];
|
||||
state.connectedEventCount[stream] = 0;
|
||||
state.openCount[stream] = 0;
|
||||
const params = new URLSearchParams({ stream, traceId: scopedTraceId });
|
||||
if (inputProfile.fromBeginning === true) params.set("fromBeginning", "true");
|
||||
const source = new EventSource(inputProfile.debugEventsPath + "?" + params.toString(), { withCredentials: true });
|
||||
state.sources[stream] = source;
|
||||
source.onopen = () => { state.openCount[stream] += 1; };
|
||||
source.addEventListener(inputProfile.debugReadyEvent, (event) => {
|
||||
state.connectedEventCount[stream] += 1;
|
||||
try { state.connected[stream] = JSON.parse(event.data); } catch { state.connected[stream] = { parseError: true }; }
|
||||
});
|
||||
source.addEventListener("hwlab.kafka.event", (event) => {
|
||||
let outer;
|
||||
try { outer = JSON.parse(event.data); } catch { return; }
|
||||
const value = outer?.value && typeof outer.value === "object" ? outer.value : {};
|
||||
const nested = value?.event && typeof value.event === "object" ? value.event : {};
|
||||
const payload = nested?.payload && typeof nested.payload === "object" ? nested.payload : {};
|
||||
const context = value?.context && typeof value.context === "object" ? value.context : {};
|
||||
const run = value?.run && typeof value.run === "object" ? value.run : {};
|
||||
const command = value?.command && typeof value.command === "object" ? value.command : {};
|
||||
const eventType = stream === "agentrun"
|
||||
? firstText(nested.type, value.eventType)
|
||||
: stream === "hwlab"
|
||||
? firstText(nested.eventType, nested.type, value.eventType)
|
||||
: firstText(value.eventType, value.stdio?.method);
|
||||
state.records[stream].push({
|
||||
stream,
|
||||
at: Date.now(),
|
||||
topic: firstText(outer.topic),
|
||||
partition: finite(outer.partition),
|
||||
offset: firstText(outer.offset),
|
||||
schema: firstText(value.schema),
|
||||
eventId: firstText(value.eventId, nested.id),
|
||||
sourceEventId: firstText(value.sourceEventId, nested.sourceEventId),
|
||||
eventType,
|
||||
sourceSeq: finite(value.sourceSeq, nested.seq, nested.sourceSeq),
|
||||
frameSeq: finite(value.frameSeq, value.stdio?.frameSeq),
|
||||
traceId: firstText(value.traceId, context.traceId, payload.traceId, nested.traceId),
|
||||
hwlabSessionId: firstText(value.hwlabSessionId, value.sessionId, run.hwlabSessionId, run.sessionId, context.sessionId, payload.hwlabSessionId, payload.sessionId, nested.hwlabSessionId, nested.sessionId),
|
||||
runId: firstText(value.runId, run.runId, context.runId, nested.runId, payload.runId),
|
||||
commandId: firstText(value.commandId, command.commandId, context.commandId, payload.commandId, nested.commandId),
|
||||
runnerId: firstText(value.runnerId, run.runnerId, command.runnerId, context.runnerId, payload.runnerId, nested.runnerId),
|
||||
attemptId: firstText(value.attemptId, run.attemptId, command.attemptId, context.attemptId, payload.attemptId, nested.attemptId),
|
||||
terminal: value.terminal === true || nested.terminal === true || eventType === "terminal" || eventType === "terminal_status",
|
||||
direction: firstText(value.stdio?.direction),
|
||||
method: firstText(value.stdio?.method),
|
||||
envelopeFingerprint: stream === "hwlab" ? fingerprint(JSON.stringify(value)) : null,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
if (state.records[stream].length > 1000) state.records[stream].splice(0, state.records[stream].length - 1000);
|
||||
});
|
||||
source.addEventListener("hwlab.kafka.error", (event) => state.errors[stream].push({ at: Date.now(), dataBytes: String(event.data || "").length }));
|
||||
source.onerror = () => state.errors[stream].push({ at: Date.now(), transport: true });
|
||||
}
|
||||
}, { key, traceId, profile: {
|
||||
debugStreams: profile.debugStreams,
|
||||
debugEventsPath: profile.debugEventsPath,
|
||||
debugReadyEvent: profile.debugReadyEvent,
|
||||
fromBeginning: profile.fromBeginning,
|
||||
} });
|
||||
}
|
||||
|
||||
async function realtimeWaitProductConnected(subscriber, timeoutMs) {
|
||||
return realtimePoll(async () => (await realtimeProductSnapshot(subscriber)).connected, timeoutMs, "product connected");
|
||||
}
|
||||
|
||||
async function realtimeWaitDebugConnected(debug, key, profile) {
|
||||
return realtimePoll(async () => {
|
||||
const snapshot = await realtimeDebugSnapshot(debug, key);
|
||||
return profile.debugStreams.every((stream) => snapshot.connected[stream]?.consumerReady === true) ? snapshot.connected : null;
|
||||
}, profile.barrierTimeoutMs, "debug consumers ready");
|
||||
}
|
||||
|
||||
async function realtimeWaitTurnIds(debug, key, traceId, timeoutMs) {
|
||||
return realtimePoll(async () => {
|
||||
const snapshot = await realtimeDebugSnapshot(debug, key);
|
||||
const rows = Object.values(snapshot.records).flat().filter((item) => item.traceId === traceId);
|
||||
const runId = rows.map((item) => item.runId).find(Boolean) || null;
|
||||
const commandId = rows.map((item) => item.commandId).find(Boolean) || null;
|
||||
const runnerId = rows.map((item) => item.runnerId).find(Boolean) || null;
|
||||
const attemptId = rows.map((item) => item.attemptId).find(Boolean) || null;
|
||||
return runId && commandId ? { runId, commandId, runnerId, attemptId } : null;
|
||||
}, timeoutMs, "realtime run and command ids");
|
||||
}
|
||||
|
||||
async function realtimeWaitDebugTurnComplete(debug, key, traceId, profile) {
|
||||
return realtimePoll(async () => {
|
||||
const snapshot = await realtimeDebugSnapshot(debug, key);
|
||||
if (!profile.debugStreams.every((stream) => (snapshot.records[stream] || []).some((item) => item.traceId === traceId))) return null;
|
||||
const agentrunTypes = new Set((snapshot.records.agentrun || []).filter((item) => item.traceId === traceId).map((item) => item.eventType));
|
||||
const hwlabTypes = new Set((snapshot.records.hwlab || []).filter((item) => item.traceId === traceId).map((item) => item.eventType));
|
||||
if (!(profile.requiredEventTypes.agentrun || []).every((type) => agentrunTypes.has(type))) return null;
|
||||
if (!(profile.requiredEventTypes.hwlab || []).every((type) => hwlabTypes.has(type))) return null;
|
||||
return snapshot;
|
||||
}, profile.terminalTimeoutMs, "realtime debug terminal event families");
|
||||
}
|
||||
|
||||
async function realtimeWaitTurnStable(debug, key, traceId, subscribers, profile) {
|
||||
await realtimeWaitDebugTurnComplete(debug, key, traceId, profile);
|
||||
const deadline = Date.now() + profile.terminalTimeoutMs;
|
||||
let previousSignature = null;
|
||||
let stableSince = Date.now();
|
||||
while (Date.now() < deadline) {
|
||||
const debugSnapshot = await realtimeDebugSnapshot(debug, key);
|
||||
const productSnapshots = await Promise.all(subscribers.map((subscriber) => realtimeProductSnapshot(subscriber)));
|
||||
const signature = JSON.stringify({
|
||||
debug: Object.fromEntries(profile.debugStreams.map((stream) => {
|
||||
const rows = (debugSnapshot.records[stream] || []).filter((item) => item.traceId === traceId);
|
||||
return [stream, { count: rows.length, lastOffset: rows.at(-1)?.offset || null, lastFingerprint: rows.at(-1)?.envelopeFingerprint || null }];
|
||||
})),
|
||||
products: productSnapshots.map((snapshot) => {
|
||||
const rows = snapshot.records.filter((item) => item.traceId === traceId);
|
||||
return { count: rows.length, lastSourceSeq: rows.at(-1)?.sourceSeq || null, lastFingerprint: rows.at(-1)?.envelopeFingerprint || null };
|
||||
}),
|
||||
});
|
||||
if (signature !== previousSignature) {
|
||||
previousSignature = signature;
|
||||
stableSince = Date.now();
|
||||
} else if (Date.now() - stableSince >= profile.terminalQuietMs) {
|
||||
return { debug: debugSnapshot, products: productSnapshots };
|
||||
}
|
||||
await sleep(250);
|
||||
}
|
||||
throw new Error("realtime debug/product streams did not become quiet after terminal");
|
||||
}
|
||||
|
||||
async function realtimeWaitTraceEvent(subscriber, traceId, terminal, timeoutMs) {
|
||||
return realtimePoll(async () => {
|
||||
const snapshot = await realtimeProductSnapshot(subscriber);
|
||||
return snapshot.records.find((item) => item.traceId === traceId && item.terminal === terminal) || null;
|
||||
}, timeoutMs, terminal ? "terminal product event" : "pre-terminal product event");
|
||||
}
|
||||
|
||||
async function realtimeWaitUiTerminal(traceId, timeoutMs) {
|
||||
await page.waitForFunction((expectedTraceId) => {
|
||||
const card = document.querySelector(".message-card[data-role='agent'][data-trace-id='" + expectedTraceId + "']");
|
||||
if (!card) return false;
|
||||
const status = card.getAttribute("data-status");
|
||||
const body = card.querySelector(".message-text")?.textContent?.trim() || "";
|
||||
return status === "completed" && body.length > 0;
|
||||
}, traceId, { timeout: timeoutMs });
|
||||
}
|
||||
|
||||
async function realtimeProductSnapshot(subscriber) {
|
||||
return subscriber.page.evaluate((key) => {
|
||||
const state = window.__unideskRealtimeFanoutProducts?.[key];
|
||||
if (!state) return { connected: null, connectedEventCount: 0, openCount: 0, records: [], errors: [{ missing: true }] };
|
||||
return { connected: state.connected ? JSON.parse(JSON.stringify(state.connected)) : null, connectedEventCount: state.connectedEventCount, openCount: state.openCount, records: JSON.parse(JSON.stringify(state.records)), errors: JSON.parse(JSON.stringify(state.errors)) };
|
||||
}, subscriber.key);
|
||||
}
|
||||
|
||||
async function realtimeDebugSnapshot(debug, key) {
|
||||
return debug.page.evaluate((streamKey) => {
|
||||
const state = window.__unideskRealtimeFanoutDebug?.[streamKey];
|
||||
if (!state) return { connected: {}, connectedEventCount: {}, openCount: {}, records: { stdio: [], agentrun: [], hwlab: [] }, errors: { stdio: [], agentrun: [], hwlab: [] } };
|
||||
return { connected: JSON.parse(JSON.stringify(state.connected)), connectedEventCount: JSON.parse(JSON.stringify(state.connectedEventCount)), openCount: JSON.parse(JSON.stringify(state.openCount)), records: JSON.parse(JSON.stringify(state.records)), errors: JSON.parse(JSON.stringify(state.errors)) };
|
||||
}, key);
|
||||
}
|
||||
|
||||
async function realtimeAllDebugSnapshots(debug) {
|
||||
return debug.page.evaluate(() => {
|
||||
const root = window.__unideskRealtimeFanoutDebug || {};
|
||||
return Object.entries(root).map(([key, state]) => ({
|
||||
key,
|
||||
connected: JSON.parse(JSON.stringify(state.connected || {})),
|
||||
connectedEventCount: JSON.parse(JSON.stringify(state.connectedEventCount || {})),
|
||||
openCount: JSON.parse(JSON.stringify(state.openCount || {})),
|
||||
records: JSON.parse(JSON.stringify(state.records || {})),
|
||||
errors: JSON.parse(JSON.stringify(state.errors || {})),
|
||||
}));
|
||||
});
|
||||
}
|
||||
|
||||
async function realtimeCloseDebugStreams(debug, key) {
|
||||
if (!debug?.page || debug.page.isClosed()) return;
|
||||
await debug.page.evaluate((streamKey) => {
|
||||
const root = window.__unideskRealtimeFanoutDebug || {};
|
||||
const keys = streamKey ? [streamKey] : Object.keys(root);
|
||||
for (const current of keys) {
|
||||
for (const source of Object.values(root[current]?.sources || {})) source?.close?.();
|
||||
}
|
||||
}, key);
|
||||
}
|
||||
|
||||
function realtimeAssertLiveConnected(connected, sessionId, label, profile) {
|
||||
if (connected?.realtimeSource !== profile.expectedKafka.topics.hwlab) throw new Error(label + " realtime source mismatch");
|
||||
if (connected?.deliverySemantics !== "live-only" || connected?.liveOnly !== true) throw new Error(label + " is not live-only");
|
||||
if (connected?.replay !== false || connected?.replaySupported !== false || connected?.lossPossible !== true) throw new Error(label + " replay/loss disclosure mismatch");
|
||||
if (connected?.filters?.sessionId !== sessionId) throw new Error(label + " session filter mismatch");
|
||||
const capabilities = connected?.capabilities || {};
|
||||
for (const [name, expected] of Object.entries(profile.expectedKafka.capabilities)) {
|
||||
if (capabilities[name] !== expected) throw new Error(label + " capability " + name + " differs from the owning YAML");
|
||||
}
|
||||
}
|
||||
|
||||
function realtimeConnectedSummary(connected) {
|
||||
return {
|
||||
realtimeSource: connected?.realtimeSource || null,
|
||||
deliverySemantics: connected?.deliverySemantics || null,
|
||||
liveOnly: connected?.liveOnly === true,
|
||||
replay: connected?.replay === true,
|
||||
replaySupported: connected?.replaySupported === true,
|
||||
lossPossible: connected?.lossPossible === true,
|
||||
filters: { sessionId: connected?.filters?.sessionId || null, traceId: connected?.filters?.traceId || null },
|
||||
capabilities: sanitize(connected?.capabilities || {}),
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
|
||||
function realtimeAssertStableProductTransport(snapshot, label) {
|
||||
if (snapshot.openCount !== 1) throw new Error(label + " product SSE open count is " + snapshot.openCount + ", expected 1");
|
||||
if (snapshot.connectedEventCount !== 1) throw new Error(label + " product connected event count is " + snapshot.connectedEventCount + ", expected 1");
|
||||
if ((snapshot.errors || []).length !== 0) throw new Error(label + " product SSE observed a transport or business error");
|
||||
}
|
||||
|
||||
function realtimeEnrichTurnExecutionIds(turn, debug) {
|
||||
const rows = Object.values(debug.records || {}).flat().filter((item) => item.traceId === turn.traceId);
|
||||
turn.runnerId ||= rows.map((item) => item.runnerId).find(Boolean) || null;
|
||||
turn.attemptId ||= rows.map((item) => item.attemptId).find(Boolean) || null;
|
||||
}
|
||||
|
||||
function realtimeValidateTurnEvidence(input) {
|
||||
const profile = input.profile;
|
||||
for (const stream of profile.debugStreams) {
|
||||
if (input.debug.connected[stream]?.consumerReady !== true) throw new Error("turn " + input.turn + " " + stream + " consumer not ready");
|
||||
if (input.debug.connected[stream]?.topic !== profile.expectedKafka.topics[stream]) throw new Error("turn " + input.turn + " " + stream + " topic mismatch");
|
||||
if (input.debug.openCount[stream] !== 1 || input.debug.connectedEventCount[stream] !== 1) throw new Error("turn " + input.turn + " " + stream + " debug SSE reconnected");
|
||||
const rows = (input.debug.records[stream] || []).filter((item) => item.traceId === input.traceId);
|
||||
if (rows.length === 0) throw new Error("turn " + input.turn + " " + stream + " has no trace records");
|
||||
if (!realtimeMonotonicOffsets(rows)) throw new Error("turn " + input.turn + " " + stream + " offsets are not monotonic");
|
||||
if (rows.some((item) => item.hwlabSessionId !== input.sessionId)) throw new Error("turn " + input.turn + " " + stream + " session missing or mismatched");
|
||||
if (rows.some((item) => item.runId !== input.runId)) throw new Error("turn " + input.turn + " " + stream + " runId missing or mismatched");
|
||||
if (rows.some((item) => item.commandId && item.commandId !== input.commandId)) throw new Error("turn " + input.turn + " " + stream + " commandId mismatched");
|
||||
const commandBoundRows = stream === "stdio"
|
||||
? rows
|
||||
: rows.filter((item) => (profile.requiredEventTypes[stream] || []).includes(item.eventType));
|
||||
if (commandBoundRows.length === 0 || commandBoundRows.some((item) => item.commandId !== input.commandId)) {
|
||||
throw new Error("turn " + input.turn + " " + stream + " command-bound event lacks the current commandId");
|
||||
}
|
||||
if ((input.debug.errors[stream] || []).length !== 0) throw new Error("turn " + input.turn + " " + stream + " debug SSE error");
|
||||
}
|
||||
const stdio = input.debug.records.stdio.filter((item) => item.traceId === input.traceId);
|
||||
const frameSeqs = stdio.map((item) => item.frameSeq).filter(Number.isFinite);
|
||||
if (frameSeqs.length === 0 || !realtimeStrictlyIncreasing(frameSeqs)) throw new Error("turn " + input.turn + " stdio frameSeq is not strictly increasing");
|
||||
const agentrunTypes = new Set(input.debug.records.agentrun.filter((item) => item.traceId === input.traceId).map((item) => item.eventType));
|
||||
const hwlabTypes = new Set(input.debug.records.hwlab.filter((item) => item.traceId === input.traceId).map((item) => item.eventType));
|
||||
for (const type of profile.requiredEventTypes.agentrun || []) if (!agentrunTypes.has(type)) throw new Error("turn " + input.turn + " AgentRun lacks " + type);
|
||||
for (const type of profile.requiredEventTypes.hwlab || []) if (!hwlabTypes.has(type)) throw new Error("turn " + input.turn + " HWLAB lacks " + type);
|
||||
const identityRows = Object.values(input.debug.records).flat().filter((item) => item.traceId === input.traceId);
|
||||
if (!input.runId || new Set(identityRows.map((item) => item.runId)).size !== 1) throw new Error("turn " + input.turn + " runId missing or mismatched across streams");
|
||||
const commandIds = new Set(identityRows.map((item) => item.commandId).filter(Boolean));
|
||||
if (!input.commandId || commandIds.size !== 1 || !commandIds.has(input.commandId)) throw new Error("turn " + input.turn + " commandId missing or mismatched across command-bound events");
|
||||
const hwlabRows = input.debug.records.hwlab.filter((item) => item.traceId === input.traceId);
|
||||
for (const snapshot of input.terminalSnapshots) {
|
||||
const rows = snapshot.records.filter((item) => item.traceId === input.traceId);
|
||||
if (!rows.some((item) => item.terminal !== true) || !rows.some((item) => item.terminal === true)) throw new Error("turn " + input.turn + " product subscriber lacks pre-terminal or terminal event");
|
||||
realtimeAssertEnvelopePassthrough(hwlabRows, rows, true, "turn " + input.turn + " terminal subscriber");
|
||||
}
|
||||
for (const snapshot of input.preTerminalSnapshots) {
|
||||
const rows = snapshot.records.filter((item) => item.traceId === input.traceId);
|
||||
if (!rows.some((item) => item.terminal !== true)) throw new Error("turn " + input.turn + " disconnected subscriber lacks pre-terminal event");
|
||||
realtimeAssertEnvelopePassthrough(hwlabRows, rows, false, "turn " + input.turn + " pre-terminal subscriber");
|
||||
}
|
||||
}
|
||||
|
||||
function realtimeAssertEnvelopePassthrough(kafkaRows, productRows, requireAllKafkaRows, label) {
|
||||
const sameEnvelope = (left, right) => left.envelopeFingerprint && left.envelopeFingerprint === right.envelopeFingerprint
|
||||
&& left.eventId === right.eventId
|
||||
&& left.sourceEventId === right.sourceEventId
|
||||
&& left.sourceSeq === right.sourceSeq
|
||||
&& left.eventType === right.eventType
|
||||
&& left.traceId === right.traceId
|
||||
&& left.hwlabSessionId === right.hwlabSessionId
|
||||
&& left.runId === right.runId
|
||||
&& left.commandId === right.commandId
|
||||
&& left.terminal === right.terminal;
|
||||
for (const productRow of productRows) {
|
||||
if (!kafkaRows.some((kafkaRow) => sameEnvelope(kafkaRow, productRow))) throw new Error(label + " received a product SSE envelope absent from hwlab.event.v1");
|
||||
}
|
||||
if (requireAllKafkaRows) {
|
||||
if (productRows.length !== kafkaRows.length) throw new Error(label + " envelope count differs from hwlab.event.v1");
|
||||
for (const kafkaRow of kafkaRows) {
|
||||
if (!productRows.some((productRow) => sameEnvelope(kafkaRow, productRow))) throw new Error(label + " did not receive an unchanged hwlab.event.v1 envelope");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function realtimeMonotonicOffsets(rows) {
|
||||
const partitions = new Map();
|
||||
for (const row of rows) {
|
||||
let offset;
|
||||
try { offset = BigInt(String(row.offset)); } catch { return false; }
|
||||
if (!Number.isFinite(row.partition)) return false;
|
||||
const partition = row.partition;
|
||||
const previous = partitions.get(partition);
|
||||
if (previous !== undefined && offset <= previous) return false;
|
||||
partitions.set(partition, offset);
|
||||
}
|
||||
return partitions.size > 0;
|
||||
}
|
||||
|
||||
function realtimeStrictlyIncreasing(values) {
|
||||
for (let index = 1; index < values.length; index += 1) if (!(values[index] > values[index - 1])) return false;
|
||||
return true;
|
||||
}
|
||||
|
||||
function realtimeForbiddenRequest(item, forbiddenPaths) {
|
||||
const pathValue = String(item.path || "").toLowerCase();
|
||||
return forbiddenPaths.some((entry) => pathValue.includes(String(entry).toLowerCase()));
|
||||
}
|
||||
|
||||
function realtimeTurnSummary(turn, debug, product, terminal) {
|
||||
const streams = {};
|
||||
for (const [stream, rowsRaw] of Object.entries(debug.records)) {
|
||||
const rows = rowsRaw.filter((item) => item.traceId === turn.traceId);
|
||||
streams[stream] = {
|
||||
topic: debug.connected[stream]?.topic || rows[0]?.topic || null,
|
||||
groupId: debug.connected[stream]?.groupId || null,
|
||||
count: rows.length,
|
||||
partitions: [...new Set(rows.map((item) => item.partition).filter(Number.isFinite))],
|
||||
firstOffset: rows[0]?.offset || null,
|
||||
lastOffset: rows.at(-1)?.offset || null,
|
||||
eventTypes: [...new Set(rows.map((item) => item.eventType).filter(Boolean))],
|
||||
firstFrameSeq: rows.map((item) => item.frameSeq).find(Number.isFinite) || null,
|
||||
lastFrameSeq: rows.map((item) => item.frameSeq).filter(Number.isFinite).at(-1) || null,
|
||||
openCount: debug.openCount?.[stream] || 0,
|
||||
connectedEventCount: debug.connectedEventCount?.[stream] || 0,
|
||||
errorCount: (debug.errors?.[stream] || []).length,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
const productRows = product.records.filter((item) => item.traceId === turn.traceId);
|
||||
return {
|
||||
turn: turn.turn,
|
||||
traceId: turn.traceId,
|
||||
runId: turn.runId,
|
||||
commandId: turn.commandId,
|
||||
runnerId: turn.runnerId,
|
||||
attemptId: turn.attemptId,
|
||||
sessionId: turn.sessionId,
|
||||
admissionStatus: turn.admissionStatus,
|
||||
terminalStatus: terminal?.status || null,
|
||||
streams,
|
||||
product: { count: productRows.length, openCount: product.openCount, connectedEventCount: product.connectedEventCount, errorCount: product.errors.length, preTerminalSeen: productRows.some((item) => item.terminal !== true), terminalSeen: productRows.some((item) => item.terminal === true), valuesRedacted: true },
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
|
||||
async function realtimeCaptureSubscriberScreenshot(targetPage, reportDir, name, summary) {
|
||||
await targetPage.evaluate((value) => {
|
||||
document.documentElement.style.background = "#0b1020";
|
||||
document.body.innerHTML = "";
|
||||
const pre = document.createElement("pre");
|
||||
pre.style.cssText = "color:#d7e3ff;background:#111a30;padding:32px;border-radius:16px;font:18px/1.55 ui-monospace,monospace;white-space:pre-wrap;margin:40px;";
|
||||
pre.textContent = JSON.stringify(value, null, 2);
|
||||
document.body.appendChild(pre);
|
||||
}, summary);
|
||||
const file = path.join(reportDir, name);
|
||||
await targetPage.screenshot({ path: file, fullPage: true, animations: "disabled", timeout: screenshotCaptureTimeoutMs });
|
||||
const meta = await fileMeta(file);
|
||||
const artifact = { kind: "realtime-fanout-screenshot", path: file, name, byteCount: meta.byteCount, sha256: meta.sha256, commandId: activeCommandId, valuesRedacted: true };
|
||||
await appendJsonl(files.artifacts, { ts: new Date().toISOString(), ...artifact });
|
||||
return artifact;
|
||||
}
|
||||
|
||||
async function realtimePartialProductEvidence(subscribers) {
|
||||
const out = [];
|
||||
for (const subscriber of subscribers.filter(Boolean)) {
|
||||
const snapshot = await realtimeProductSnapshot(subscriber).catch(() => null);
|
||||
if (!snapshot) continue;
|
||||
out.push({
|
||||
label: subscriber.label,
|
||||
openCount: snapshot.openCount,
|
||||
connectedEventCount: snapshot.connectedEventCount,
|
||||
errorCount: snapshot.errors.length,
|
||||
eventCount: snapshot.records.length,
|
||||
traceIds: [...new Set(snapshot.records.map((item) => item.traceId).filter(Boolean))].slice(0, 4),
|
||||
terminalCount: snapshot.records.filter((item) => item.terminal === true).length,
|
||||
connected: realtimeConnectedSummary(snapshot.connected),
|
||||
valuesRedacted: true,
|
||||
});
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
function realtimeDebugSnapshotSummary(snapshot) {
|
||||
return {
|
||||
key: snapshot.key || null,
|
||||
streams: Object.fromEntries(Object.entries(snapshot.records || {}).map(([stream, rows]) => [stream, {
|
||||
topic: snapshot.connected?.[stream]?.topic || null,
|
||||
groupId: snapshot.connected?.[stream]?.groupId || null,
|
||||
openCount: snapshot.openCount?.[stream] || 0,
|
||||
connectedEventCount: snapshot.connectedEventCount?.[stream] || 0,
|
||||
errorCount: (snapshot.errors?.[stream] || []).length,
|
||||
eventCount: rows.length,
|
||||
traceIds: [...new Set(rows.map((item) => item.traceId).filter(Boolean))].slice(0, 4),
|
||||
valuesRedacted: true,
|
||||
}])),
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
|
||||
async function realtimeWriteEvidenceArtifacts(reportDir, network, debugSnapshots) {
|
||||
const requestsFile = path.join(reportDir, "request-ledger.jsonl");
|
||||
const eventsFile = path.join(reportDir, "events.jsonl");
|
||||
await writeFile(requestsFile, network.map((item) => JSON.stringify(item)).join("\n") + "\n", { mode: 0o600 });
|
||||
const eventRows = [];
|
||||
for (const snapshot of debugSnapshots) {
|
||||
for (const [stream, rows] of Object.entries(snapshot.records || {})) {
|
||||
for (const row of rows) eventRows.push({ ...row, stream, valuesRedacted: true });
|
||||
}
|
||||
}
|
||||
await writeFile(eventsFile, eventRows.map((item) => JSON.stringify(item)).join("\n") + "\n", { mode: 0o600 });
|
||||
const requestMeta = await fileMeta(requestsFile);
|
||||
const eventMeta = await fileMeta(eventsFile);
|
||||
const artifacts = {
|
||||
requestLedger: { path: requestsFile, ...requestMeta, valuesRedacted: true },
|
||||
events: { path: eventsFile, ...eventMeta, count: eventRows.length, valuesRedacted: true },
|
||||
valuesRedacted: true,
|
||||
};
|
||||
await appendJsonl(files.artifacts, { ts: new Date().toISOString(), kind: "realtime-fanout-request-ledger", commandId: activeCommandId, ...artifacts.requestLedger });
|
||||
await appendJsonl(files.artifacts, { ts: new Date().toISOString(), kind: "realtime-fanout-events", commandId: activeCommandId, ...artifacts.events });
|
||||
return artifacts;
|
||||
}
|
||||
|
||||
async function realtimeWriteReport(reportDir, report) {
|
||||
const reportFile = path.join(reportDir, "report.json");
|
||||
await writeFile(reportFile, JSON.stringify(sanitize(report), null, 2) + "\n", { mode: 0o600 });
|
||||
const meta = await fileMeta(reportFile);
|
||||
const artifact = { kind: "realtime-fanout-report", path: reportFile, byteCount: meta.byteCount, sha256: meta.sha256, commandId: activeCommandId, valuesRedacted: true };
|
||||
await appendJsonl(files.artifacts, { ts: new Date().toISOString(), ...artifact });
|
||||
return artifact;
|
||||
}
|
||||
|
||||
async function realtimePoll(read, timeoutMs, label) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
const value = await read();
|
||||
if (value) return value;
|
||||
await sleep(250);
|
||||
}
|
||||
throw new Error(label + " timed out after " + timeoutMs + "ms");
|
||||
}
|
||||
|
||||
function realtimeFirstText(...values) {
|
||||
for (const value of values) if (typeof value === "string" && value.trim()) return value.trim();
|
||||
return "";
|
||||
}
|
||||
`;
|
||||
}
|
||||
@@ -8,6 +8,7 @@ import { nodeWebObserveRunnerControlSource } from "./hwlab-node-web-observe-runn
|
||||
import { nodeWebObserveRunnerWorkbenchSource } from "./hwlab-node-web-observe-runner-workbench-source";
|
||||
import { nodeWebObserveRunnerSamplingSource } from "./hwlab-node-web-observe-runner-sampling-source";
|
||||
import { nodeWebObserveRunnerUtilitySource } from "./hwlab-node-web-observe-runner-utility-source";
|
||||
import { nodeWebObserveRunnerRealtimeSource } from "./hwlab-node-web-observe-runner-realtime-source";
|
||||
|
||||
export function nodeWebObserveRunnerSource(): string {
|
||||
return String.raw`#!/usr/bin/env node
|
||||
@@ -42,6 +43,7 @@ const navigationMaxAttempts = boundedInteger(process.env.UNIDESK_WEB_OBSERVE_NAV
|
||||
const alertThresholds = parseAlertThresholds(process.env.UNIDESK_WEB_OBSERVE_ALERT_THRESHOLDS_JSON);
|
||||
const browserFreezePolicy = parseBrowserFreezePolicy(process.env.UNIDESK_WEB_OBSERVE_BROWSER_FREEZE_POLICY_JSON);
|
||||
const projectManagement = parseProjectManagementConfig(process.env.UNIDESK_WEB_OBSERVE_PROJECT_MANAGEMENT_JSON);
|
||||
const realtimeFanoutProfiles = parseRealtimeFanoutProfiles(process.env.UNIDESK_WEB_OBSERVE_REALTIME_FANOUT_PROFILES_JSON);
|
||||
const playwrightProxy = proxyConfigFromEnv(baseUrl);
|
||||
const chromiumLaunchOptions = chromiumLaunchOptionsForProxy(playwrightProxy);
|
||||
const pageId = "control-" + randomBytes(4).toString("hex");
|
||||
@@ -185,6 +187,8 @@ ${nodeWebObserveRunnerCommandActionsSource()}
|
||||
|
||||
${nodeWebObserveRunnerWorkbenchSource()}
|
||||
|
||||
${nodeWebObserveRunnerRealtimeSource()}
|
||||
|
||||
${nodeWebObserveRunnerSamplingSource()}
|
||||
|
||||
${nodeWebObserveRunnerUtilitySource()}
|
||||
|
||||
@@ -129,6 +129,7 @@ export type NodeWebProbeObserveCommandType =
|
||||
| "gotoProjectMdtodo"
|
||||
| "newSession"
|
||||
| "sendPrompt"
|
||||
| "validateRealtimeFanout"
|
||||
| "steer"
|
||||
| "cancel"
|
||||
| "selectProvider"
|
||||
@@ -206,10 +207,12 @@ export interface NodeWebProbeObserveOptions {
|
||||
force: boolean;
|
||||
commandType: NodeWebProbeObserveCommandType | null;
|
||||
commandText: string | null;
|
||||
commandSecondText: string | null;
|
||||
commandPath: string | null;
|
||||
commandLabel: string | null;
|
||||
commandSessionId: string | null;
|
||||
commandProvider: string | null;
|
||||
commandProfile: string | null;
|
||||
commandAfterRound: number | null;
|
||||
commandSeverity: string | null;
|
||||
commandAlternateSessionStrategy: string | null;
|
||||
|
||||
@@ -1858,6 +1858,7 @@ export function nodeRuntimePipelinePostprocessScript(): string[] {
|
||||
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_ENABLED', String(kafka.enabled)) || codeAgentRuntimeChanged;",
|
||||
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED', String(kafka.enabled && (kafka.features.directPublish || kafka.features.transactionalProjector))) || codeAgentRuntimeChanged;",
|
||||
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_BOOTSTRAP_SERVERS', String(kafka.bootstrapServers)) || codeAgentRuntimeChanged;",
|
||||
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_STDIO_TOPIC', String(kafka.stdioTopic)) || codeAgentRuntimeChanged;",
|
||||
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC', String(kafka.agentRunEventTopic)) || codeAgentRuntimeChanged;",
|
||||
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_EVENT_TOPIC', String(kafka.hwlabEventTopic)) || codeAgentRuntimeChanged;",
|
||||
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_CLIENT_ID', String(kafka.clientId)) || codeAgentRuntimeChanged;",
|
||||
@@ -2269,6 +2270,7 @@ export function nodeRuntimePipelinePostprocessScript(): string[] {
|
||||
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_ENABLED', String(kafka.enabled));",
|
||||
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED', String(kafka.enabled && (kafka.features.directPublish || kafka.features.transactionalProjector)));",
|
||||
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_BOOTSTRAP_SERVERS', String(kafka.bootstrapServers));",
|
||||
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_STDIO_TOPIC', String(kafka.stdioTopic));",
|
||||
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC', String(kafka.agentRunEventTopic));",
|
||||
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_EVENT_TOPIC', String(kafka.hwlabEventTopic));",
|
||||
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_CLIENT_ID', String(kafka.clientId));",
|
||||
|
||||
@@ -3,6 +3,7 @@ import { test } from "bun:test";
|
||||
|
||||
import { withWebObserveCommandRendered } from "../hwlab-node-web-observe-render";
|
||||
import { renderWebProbeScriptResult } from "./web-observe-render";
|
||||
import { webProbeScriptCommandGovernance } from "./web-observe-scripts";
|
||||
|
||||
test("observe command renderer separates control completion from async turn terminal state", () => {
|
||||
const rendered = withWebObserveCommandRendered({
|
||||
@@ -24,7 +25,21 @@ test("observe command renderer separates control completion from async turn term
|
||||
});
|
||||
|
||||
test("web-probe script render warns to promote repeated scripts into commands", () => {
|
||||
const commandPromotionHint = {
|
||||
schemaVersion: "unidesk.web-probe.command-promotion.v1",
|
||||
code: "prefer_repo_owned_typed_command",
|
||||
promotionRequiredBeforeRepeat: true,
|
||||
repoOwnedTypedCommand: false,
|
||||
sourceKind: "stdin",
|
||||
currentCommand: "web-probe script --node D601 --lane v03",
|
||||
preferredArtifact: "repo-owned-typed-command",
|
||||
preferredSurface: "web-probe observe command or a dedicated web-probe subcommand",
|
||||
action: "promote-before-repeat",
|
||||
message: "Before repeating this one-off script, promote the workflow to a repo-owned typed web-probe command.",
|
||||
valuesRedacted: true,
|
||||
};
|
||||
const rendered = renderWebProbeScriptResult({
|
||||
commandPromotionHint,
|
||||
ok: true,
|
||||
status: "pass",
|
||||
command: "web-probe script --node D601 --lane v03",
|
||||
@@ -34,12 +49,15 @@ test("web-probe script render warns to promote repeated scripts into commands",
|
||||
warnings: [{
|
||||
code: "web_probe_script_ad_hoc_only",
|
||||
severity: "warning",
|
||||
message: "web-probe script is for one-off exploration; repeated or high-frequency checks must be promoted to a typed command instead of rerunning temporary scripts.",
|
||||
requiredAction: "promote-before-repeat",
|
||||
preferredArtifact: "repo-owned-typed-command",
|
||||
message: "web-probe script is a one-off exploration escape hatch; before repeating it, promote the workflow to a repo-owned typed command instead of rerunning temporary scripts.",
|
||||
}],
|
||||
hints: [
|
||||
"Prefer `web-probe observe start` plus `web-probe observe command` for interactive flows.",
|
||||
"Before rerunning this script, promote the workflow to a repo-owned typed `web-probe` command.",
|
||||
],
|
||||
preferredCommands: {
|
||||
typedCommandCatalog: "bun scripts/cli.ts web-probe --help",
|
||||
startObserver: "bun scripts/cli.ts web-probe observe start --node D601 --lane v03 --target-path /projects/mdtodo",
|
||||
mdtodoSummary: "bun scripts/cli.ts web-probe observe collect <observerId> --view project-mdtodo-summary",
|
||||
},
|
||||
@@ -51,10 +69,64 @@ test("web-probe script render warns to promote repeated scripts into commands",
|
||||
});
|
||||
|
||||
const text = String(rendered.renderedText ?? "");
|
||||
const marker = "UNIDESK_WEB_PROBE_COMMAND_PROMOTION_HINT ";
|
||||
const firstLine = text.split("\n", 1)[0] ?? "";
|
||||
assert.ok(firstLine.startsWith(marker));
|
||||
assert.deepEqual(JSON.parse(firstLine.slice(marker.length)), commandPromotionHint);
|
||||
assert.equal(Object.keys(rendered)[0], "commandPromotionHint");
|
||||
assert.match(text, /WARNINGS/u);
|
||||
assert.match(text, /web_probe_script_ad_hoc_only/u);
|
||||
assert.match(text, /HINTS/u);
|
||||
assert.match(text, /web-probe observe command/u);
|
||||
assert.match(text, /repo-owned typed `web-probe` command/u);
|
||||
assert.match(text, /typedCommandCatalog: bun scripts\/cli\.ts web-probe --help/u);
|
||||
assert.match(text, /startObserver: bun scripts\/cli\.ts web-probe observe start/u);
|
||||
assert.match(text, /mdtodoSummary: bun scripts\/cli\.ts web-probe observe collect/u);
|
||||
});
|
||||
|
||||
test("web-probe command governance distinguishes ad-hoc scripts from repo-owned generated commands", () => {
|
||||
const base = {
|
||||
action: "script" as const,
|
||||
node: "D601",
|
||||
lane: "v03",
|
||||
url: "https://hwlab.example.test",
|
||||
timeoutMs: 30000,
|
||||
viewport: "1920x1080",
|
||||
browserProxyMode: "auto" as const,
|
||||
commandTimeoutSeconds: 60,
|
||||
scriptText: "return { ok: true };",
|
||||
scriptSource: {
|
||||
kind: "stdin" as const,
|
||||
path: null,
|
||||
byteCount: 20,
|
||||
sha256: "sha256:fixture",
|
||||
},
|
||||
};
|
||||
|
||||
const adHoc = webProbeScriptCommandGovernance(base);
|
||||
assert.equal(adHoc.commandPromotionHint.promotionRequiredBeforeRepeat, true);
|
||||
assert.equal(adHoc.commandPromotionHint.action, "promote-before-repeat");
|
||||
assert.equal(adHoc.warnings.length, 1);
|
||||
assert.equal(adHoc.warnings[0]?.requiredAction, "promote-before-repeat");
|
||||
assert.match(adHoc.hints[0] ?? "", /repo-owned typed `web-probe` command/u);
|
||||
assert.deepEqual(Object.keys(adHoc.preferredCommands)[0], "typedCommandCatalog");
|
||||
assert.ok(Object.values(adHoc.preferredCommands).every((command) => command.startsWith("bun scripts/cli.ts web-probe")));
|
||||
|
||||
const repoOwned = webProbeScriptCommandGovernance({
|
||||
...base,
|
||||
commandLabel: "web-probe opencode-smoke --node D601 --lane v03",
|
||||
suppressAdHocWarning: true,
|
||||
generatedHints: ["OpenCode smoke is a repo-owned typed web-probe command."],
|
||||
generatedPreferredCommands: {
|
||||
withExpectedText: "web-probe opencode-smoke --node D601 --lane v03 --expect-text <assistant-substring>",
|
||||
},
|
||||
scriptSource: {
|
||||
...base.scriptSource,
|
||||
kind: "generated",
|
||||
path: "builtin:opencode-smoke",
|
||||
},
|
||||
});
|
||||
assert.equal(repoOwned.commandPromotionHint.promotionRequiredBeforeRepeat, false);
|
||||
assert.equal(repoOwned.commandPromotionHint.action, "reuse-repo-owned-typed-command");
|
||||
assert.deepEqual(repoOwned.warnings, []);
|
||||
assert.doesNotMatch(repoOwned.hints.join("\n"), /temporary script|promote-before-repeat/iu);
|
||||
});
|
||||
|
||||
@@ -1353,6 +1353,9 @@ export function renderWebObserveStatusResult(result: Record<string, unknown>): R
|
||||
const commandResult = record(result.result);
|
||||
const manifest = record(observer.manifest);
|
||||
const heartbeat = record(observer.heartbeat);
|
||||
const exactCommand = record(observer.exactCommand);
|
||||
const exactResult = record(exactCommand.result);
|
||||
const exactError = record(exactCommand.error);
|
||||
const tails = record(observer.tails);
|
||||
const samples = webObserveArray(tails.samples);
|
||||
const network = webObserveArray(tails.network);
|
||||
@@ -1375,6 +1378,21 @@ export function renderWebObserveStatusResult(result: Record<string, unknown>): R
|
||||
],
|
||||
),
|
||||
];
|
||||
const exactCommandSection = Object.keys(exactCommand).length === 0 ? [] : [
|
||||
"",
|
||||
webObserveTable(
|
||||
["COMMAND_ID", "TYPE", "PHASE", "OK", "STATUS", "REPORT_SHA", "DETAIL"],
|
||||
[[
|
||||
exactCommand.commandId ?? "-",
|
||||
exactCommand.type ?? "-",
|
||||
exactCommand.phase ?? "-",
|
||||
exactCommand.ok ?? "-",
|
||||
exactResult.status ?? "-",
|
||||
exactResult.reportSha256 ?? "-",
|
||||
exactError.message ?? exactResult.reportPath ?? "-",
|
||||
]],
|
||||
),
|
||||
];
|
||||
const renderedText = [
|
||||
webObserveTable(
|
||||
["OBSERVER", "NODE", "LANE", "STATUS", "ALIVE", "PID", "SAMPLE", "COMMAND", "UPDATED"],
|
||||
@@ -1395,6 +1413,7 @@ export function renderWebObserveStatusResult(result: Record<string, unknown>): R
|
||||
["artifacts", artifacts.length, record(artifacts[artifacts.length - 1]).sha256 ?? "-"],
|
||||
],
|
||||
),
|
||||
...exactCommandSection,
|
||||
...resultSection,
|
||||
"",
|
||||
"NEXT",
|
||||
@@ -1421,13 +1440,16 @@ export function renderWebObserveCommandResult(result: Record<string, unknown>):
|
||||
["label", observerCommand.label ?? "-"],
|
||||
["path", observerCommand.path ?? "-"],
|
||||
["provider", observerCommand.provider ?? "-"],
|
||||
["profile", observerCommand.profile ?? "-"],
|
||||
["sessionId", observerCommand.sessionId ?? "-"],
|
||||
["text", observerCommand.textHash === null || observerCommand.textHash === undefined ? "-" : `${observerCommand.textBytes ?? "-"}B ${observerCommand.textHash}`],
|
||||
["secondText", observerCommand.secondTextHash === null || observerCommand.secondTextHash === undefined ? "-" : `${observerCommand.secondTextBytes ?? "-"}B ${observerCommand.secondTextHash}`],
|
||||
],
|
||||
),
|
||||
"",
|
||||
"NEXT",
|
||||
` status: bun scripts/cli.ts web-probe observe status ${id}`,
|
||||
` status: bun scripts/cli.ts web-probe observe status ${id} --command-id ${result.commandId}`,
|
||||
` result: bun scripts/cli.ts web-probe observe collect ${id} --file commands/done/${result.commandId}.json`,
|
||||
` analyze: bun scripts/cli.ts web-probe observe analyze ${id}`,
|
||||
].join("\n");
|
||||
return withWebObserveRendered(result, renderedText);
|
||||
@@ -1626,6 +1648,7 @@ export function renderWebObserveAnalyzeResult(result: Record<string, unknown>):
|
||||
}
|
||||
|
||||
export function renderWebProbeScriptResult(result: Record<string, unknown>): Record<string, unknown> {
|
||||
const commandPromotionHint = record(result.commandPromotionHint);
|
||||
const summary = record(result.summary);
|
||||
const issueEvidence = record(result.issueEvidence);
|
||||
const probe = record(result.probe);
|
||||
@@ -1656,6 +1679,8 @@ export function renderWebProbeScriptResult(result: Record<string, unknown>): Rec
|
||||
["exitCode", commandResult.exitCode ?? "-"],
|
||||
];
|
||||
const renderedText = [
|
||||
`UNIDESK_WEB_PROBE_COMMAND_PROMOTION_HINT ${JSON.stringify(commandPromotionHint)}`,
|
||||
"",
|
||||
webObserveTable(
|
||||
["WEB_PROBE_SCRIPT", "NODE", "LANE", "STATUS", "OK", "URL"],
|
||||
[[result.command ?? "web-probe script", result.node, result.lane, result.status, result.ok, result.url]],
|
||||
@@ -1711,7 +1736,7 @@ export function renderWebProbeScriptResult(result: Record<string, unknown>): Rec
|
||||
` rerun: ${result.command ?? `web-probe script --node ${result.node ?? "-"} --lane ${result.lane ?? "-"}`}`,
|
||||
...Object.entries(preferredCommands).map(([name, command]) => ` ${name}: ${String(command)}`),
|
||||
].join("\n");
|
||||
return withWebObserveRendered(result, renderedText);
|
||||
return withWebObserveRendered(Object.assign({ commandPromotionHint }, result), renderedText);
|
||||
}
|
||||
|
||||
export function webProbeScriptRecordRows(value: Record<string, unknown>, limit: number): unknown[][] {
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { mkdir, mkdtemp, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { test } from "bun:test";
|
||||
|
||||
import { nodeWebObserveStatusNodeScript } from "./web-observe-scripts";
|
||||
|
||||
async function runStatusScript(stateDir: string, commandId: string): Promise<Record<string, any>> {
|
||||
const child = Bun.spawn(["bash", "-lc", nodeWebObserveStatusNodeScript(1, "NC01", "v03", commandId)], {
|
||||
env: { ...process.env, state_dir: stateDir },
|
||||
stdout: "pipe",
|
||||
stderr: "pipe",
|
||||
});
|
||||
const [stdout, stderr, exitCode] = await Promise.all([
|
||||
new Response(child.stdout).text(),
|
||||
new Response(child.stderr).text(),
|
||||
child.exited,
|
||||
]);
|
||||
assert.equal(exitCode, 0, stderr);
|
||||
return JSON.parse(stdout) as Record<string, any>;
|
||||
}
|
||||
|
||||
test("observe status returns one exact completed command without prompt payloads", async () => {
|
||||
const stateDir = await mkdtemp(join(tmpdir(), "unidesk-web-observe-exact-status-"));
|
||||
const commandId = "cmd-fixture";
|
||||
await mkdir(join(stateDir, "commands", "done"), { recursive: true });
|
||||
await mkdir(join(stateDir, "commands", "processing"), { recursive: true });
|
||||
await writeFile(join(stateDir, "commands", "done", `${commandId}.json`), JSON.stringify({
|
||||
ok: true,
|
||||
commandId,
|
||||
type: "validateRealtimeFanout",
|
||||
completedAt: "2026-07-10T12:00:00.000Z",
|
||||
result: {
|
||||
ok: true,
|
||||
status: "passed",
|
||||
profile: "pure-kafka-live",
|
||||
sessionId: "ses_fixture",
|
||||
traceIds: ["trc_first", "trc_second"],
|
||||
runIds: ["run_fixture"],
|
||||
commandIds: ["cmd_runner_first", "cmd_runner_second"],
|
||||
reportPath: "/tmp/report.json",
|
||||
reportSha256: "sha256:fixture",
|
||||
prompt: "must-not-leak",
|
||||
},
|
||||
}) + "\n");
|
||||
await writeFile(join(stateDir, "commands", "processing", `${commandId}.json`), JSON.stringify({
|
||||
ok: null,
|
||||
commandId,
|
||||
type: "validateRealtimeFanout",
|
||||
}) + "\n");
|
||||
|
||||
const status = await runStatusScript(stateDir, commandId);
|
||||
assert.equal(status.exactCommand.phase, "done");
|
||||
assert.equal(status.exactCommand.type, "validateRealtimeFanout");
|
||||
assert.equal(status.exactCommand.result.reportSha256, "sha256:fixture");
|
||||
assert.equal(status.exactCommand.result.prompt, undefined);
|
||||
assert.equal(status.exactCommand.valuesRedacted, true);
|
||||
});
|
||||
|
||||
test("observe status keeps exact command not-found structured", async () => {
|
||||
const stateDir = await mkdtemp(join(tmpdir(), "unidesk-web-observe-exact-missing-"));
|
||||
const status = await runStatusScript(stateDir, "cmd-missing");
|
||||
assert.deepEqual(status.exactCommand, {
|
||||
commandId: "cmd-missing",
|
||||
phase: "not-found",
|
||||
ok: false,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
});
|
||||
@@ -36,12 +36,13 @@ import { compactCommandResultRedacted, nullableRecord, redactKnownSecrets, shell
|
||||
import { renderWebProbeScriptResult } from "./web-observe-render";
|
||||
import { nodeWebProbeHostProxyEnv } from "./web-probe-observe";
|
||||
|
||||
export function nodeWebObserveStatusNodeScript(tailLines: number, node: string, lane: string): string {
|
||||
export function nodeWebObserveStatusNodeScript(tailLines: number, node: string, lane: string, commandId: string | null = null): string {
|
||||
return `node -e ${shellQuote(`
|
||||
const fs=require('fs'),path=require('path');
|
||||
const dir=process.env.state_dir||process.argv[1];
|
||||
const node=process.argv[2];
|
||||
const lane=process.argv[3];
|
||||
const exactCommandId=process.argv[4]||'';
|
||||
const tailN=Math.min(${tailLines},3);
|
||||
const readJson=(name)=>{try{return JSON.parse(fs.readFileSync(path.join(dir,name),'utf8'))}catch{return null}};
|
||||
const readJsonPath=(file)=>{try{return JSON.parse(fs.readFileSync(file,'utf8'))}catch{return null}};
|
||||
@@ -65,6 +66,17 @@ export function nodeWebObserveStatusNodeScript(tailLines: number, node: string,
|
||||
const ages=items.map((item)=>item.ageSeconds).filter((value)=>Number.isFinite(value));
|
||||
return {count:entries.length,oldestAgeSeconds:ages.length?Math.max(...ages):null,items};
|
||||
};
|
||||
const exactCommand=()=>{
|
||||
if(!exactCommandId)return null;
|
||||
for(const bucket of ['done','failed','abandoned','processing','pending']){
|
||||
const parsed=readJsonPath(path.join(commandDir(bucket),exactCommandId+'.json'));
|
||||
if(!parsed)continue;
|
||||
const result=parsed.result&&typeof parsed.result==='object'?parsed.result:null;
|
||||
const error=parsed.error&&typeof parsed.error==='object'?parsed.error:null;
|
||||
return {commandId:exactCommandId,phase:bucket,ok:parsed.ok??null,type:parsed.type||null,createdAt:parsed.createdAt||null,completedAt:parsed.completedAt||null,failedAt:parsed.failedAt||null,result:result?{ok:result.ok??null,status:result.status||null,profile:result.profile||null,sessionId:result.sessionId||null,traceIds:Array.isArray(result.traceIds)?result.traceIds.slice(0,4):[],runIds:Array.isArray(result.runIds)?result.runIds.slice(0,4):[],commandIds:Array.isArray(result.commandIds)?result.commandIds.slice(0,4):[],warmRunnerReused:result.warmRunnerReused??null,forbiddenRequestCount:result.forbiddenRequestCount??null,replayedEventCount:result.replayedEventCount??null,reportPath:result.reportPath||null,reportSha256:result.reportSha256||null,valuesRedacted:true}:null,error:error?{name:error.name||null,message:short(error.message,200),details:error.details&&typeof error.details==='object'?{profile:error.details.profile||null,reportPath:error.details.reportPath||null,reportSha256:error.details.reportSha256||null,valuesRedacted:true}:null}:null,valuesRedacted:true};
|
||||
}
|
||||
return {commandId:exactCommandId,phase:'not-found',ok:false,valuesRedacted:true};
|
||||
};
|
||||
const pidText=fs.existsSync(path.join(dir,'pid'))?fs.readFileSync(path.join(dir,'pid'),'utf8').trim():null;
|
||||
let alive=false; if(pidText){try{process.kill(Number(pidText),0); alive=true}catch{}}
|
||||
const manifest=readJson('manifest.json');
|
||||
@@ -82,8 +94,8 @@ export function nodeWebObserveStatusNodeScript(tailLines: number, node: string,
|
||||
const commandsFailed=commandSummary('failed');
|
||||
const diagnostics={heartbeatAgeSeconds,heartbeatStale,heartbeatStaleAfterSeconds:Math.round(staleAfterMs/1000),heartbeatUpdatedAt:updatedRaw,terminal,processAlive:alive,effectiveLiveness:heartbeatStale?'stale':alive?'alive':'not-running',commandBacklog:commandsPending.count+commandsProcessing.count,oldestPendingAgeSeconds:commandsPending.oldestAgeSeconds,oldestProcessingAgeSeconds:commandsProcessing.oldestAgeSeconds,valuesRedacted:true};
|
||||
const commands={pendingCount:commandsPending.count,processingCount:commandsProcessing.count,abandonedCount:commandsAbandoned.count,failedCount:commandsFailed.count,pending:commandsPending.items,processing:commandsProcessing.items,abandoned:commandsAbandoned.items.slice(0,6),failed:commandsFailed.items.slice(0,6),valuesRedacted:true};
|
||||
console.log(JSON.stringify({ok:true,command:'web-probe-observe status',stateDir:dir,pid:pidText?Number(pidText):null,processAlive:alive,manifest:compactManifest(manifest),heartbeat:compactHeartbeat(heartbeat,diagnostics),diagnostics,commands,tails:{control:tailJsonl('control.jsonl').map(compactControl),samples:tailJsonl('samples.jsonl').map(compactSample),network:tailJsonl('network.jsonl').map(compactNetwork)},next:{command:'bun scripts/cli.ts web-probe observe command --node '+node+' --lane '+lane+' --state-dir '+dir+' --type mark --label checkpoint',stop:'bun scripts/cli.ts web-probe observe stop --node '+node+' --lane '+lane+' --state-dir '+dir,forceStop:'bun scripts/cli.ts web-probe observe stop --node '+node+' --lane '+lane+' --state-dir '+dir+' --force',analyze:'bun scripts/cli.ts web-probe observe analyze --node '+node+' --lane '+lane+' --state-dir '+dir},valuesRedacted:true}));
|
||||
`)} "$state_dir" ${shellQuote(node)} ${shellQuote(lane)}`;
|
||||
console.log(JSON.stringify({ok:true,command:'web-probe-observe status',stateDir:dir,pid:pidText?Number(pidText):null,processAlive:alive,manifest:compactManifest(manifest),heartbeat:compactHeartbeat(heartbeat,diagnostics),diagnostics,commands,exactCommand:exactCommand(),tails:{control:tailJsonl('control.jsonl').map(compactControl),samples:tailJsonl('samples.jsonl').map(compactSample),network:tailJsonl('network.jsonl').map(compactNetwork)},next:{command:'bun scripts/cli.ts web-probe observe command --node '+node+' --lane '+lane+' --state-dir '+dir+' --type mark --label checkpoint',stop:'bun scripts/cli.ts web-probe observe stop --node '+node+' --lane '+lane+' --state-dir '+dir,forceStop:'bun scripts/cli.ts web-probe observe stop --node '+node+' --lane '+lane+' --state-dir '+dir+' --force',analyze:'bun scripts/cli.ts web-probe observe analyze --node '+node+' --lane '+lane+' --state-dir '+dir},valuesRedacted:true}));
|
||||
`)} "$state_dir" ${shellQuote(node)} ${shellQuote(lane)} ${shellQuote(commandId ?? "")}`;
|
||||
}
|
||||
|
||||
export function nodeWebObserveForceStopNodeScript(reason: string, forcedByCommandId: string): string {
|
||||
@@ -385,6 +397,7 @@ export function nodeWebObserveWaitCommandShell(commandId: string, waitMs: number
|
||||
|
||||
export function commandSummaryForOutput(payload: Record<string, unknown>): Record<string, unknown> {
|
||||
const text = typeof payload.text === "string" ? payload.text : null;
|
||||
const secondText = typeof payload.secondText === "string" ? payload.secondText : null;
|
||||
const title = typeof payload.title === "string" ? payload.title : null;
|
||||
const body = typeof payload.body === "string" ? payload.body : null;
|
||||
const opaque = (value: unknown) => typeof value === "string" && value.length > 0
|
||||
@@ -401,6 +414,7 @@ export function commandSummaryForOutput(payload: Record<string, unknown>): Recor
|
||||
label: payload.label ?? null,
|
||||
sessionId: payload.sessionId ?? null,
|
||||
provider: payload.provider ?? null,
|
||||
profile: payload.profile ?? null,
|
||||
durationMs: payload.durationMs ?? null,
|
||||
sourceId: opaque(payload.sourceId),
|
||||
fileRef: opaque(payload.fileRef),
|
||||
@@ -416,6 +430,8 @@ export function commandSummaryForOutput(payload: Record<string, unknown>): Recor
|
||||
root: opaque(payload.root),
|
||||
textHash: text === null ? null : `sha256:${createHash("sha256").update(text).digest("hex")}`,
|
||||
textBytes: text === null ? null : Buffer.byteLength(text),
|
||||
secondTextHash: secondText === null ? null : `sha256:${createHash("sha256").update(secondText).digest("hex")}`,
|
||||
secondTextBytes: secondText === null ? null : Buffer.byteLength(secondText),
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
@@ -478,6 +494,7 @@ export function runNodeWebProbeScript(
|
||||
credential: Record<string, unknown>,
|
||||
): Record<string, unknown> {
|
||||
const commandLabel = options.commandLabel ?? `web-probe script --node ${options.node} --lane ${options.lane}`;
|
||||
const commandGovernance = webProbeScriptCommandGovernance(options);
|
||||
const webProbeProxy = nodeWebProbeHostProxyEnv(spec, options.browserProxyMode);
|
||||
const script = nodeWebProbeScriptRemoteShell(options, secretSpec, material.username ?? secretSpec.bootstrapAdminUsername, material.password ?? "", webProbeProxy, spec.webProbe?.playwrightBrowsersPath);
|
||||
const result = runTransWorkspaceStdinScript(options.node, spec.workspace, script, options.commandTimeoutSeconds);
|
||||
@@ -570,6 +587,7 @@ export function runNodeWebProbeScript(
|
||||
compactResult.stderrTail = redactKnownSecrets(result.stderr.slice(-2000), [material.password ?? ""]);
|
||||
}
|
||||
return renderWebProbeScriptResult({
|
||||
commandPromotionHint: commandGovernance.commandPromotionHint,
|
||||
ok: passed,
|
||||
status: passed ? "pass" : "blocked",
|
||||
command: commandLabel,
|
||||
@@ -593,9 +611,9 @@ export function runNodeWebProbeScript(
|
||||
stdoutRecovery: recoveredReport.stdoutRecovery ?? null,
|
||||
},
|
||||
stdoutRecovery: stdoutResolution.diagnostics,
|
||||
warnings: webProbeScriptGovernanceWarnings(options),
|
||||
hints: webProbeScriptGovernanceHints(options),
|
||||
preferredCommands: webProbeScriptPreferredCommands(options),
|
||||
warnings: commandGovernance.warnings,
|
||||
hints: commandGovernance.hints,
|
||||
preferredCommands: commandGovernance.preferredCommands,
|
||||
recoveredArtifacts: recoveredArtifacts === null ? null : {
|
||||
source: recoveredArtifacts.source,
|
||||
degradedReason: recoveredArtifacts.degradedReason,
|
||||
@@ -607,34 +625,72 @@ export function runNodeWebProbeScript(
|
||||
});
|
||||
}
|
||||
|
||||
function webProbeScriptGovernanceWarnings(options: NodeWebProbeScriptOptions): Record<string, unknown>[] {
|
||||
export function webProbeScriptCommandGovernance(options: NodeWebProbeScriptOptions): {
|
||||
commandPromotionHint: Record<string, unknown>;
|
||||
warnings: Record<string, unknown>[];
|
||||
hints: string[];
|
||||
preferredCommands: Record<string, string>;
|
||||
} {
|
||||
return {
|
||||
commandPromotionHint: webProbeScriptCommandPromotionHint(options),
|
||||
warnings: webProbeScriptGovernanceWarnings(options),
|
||||
hints: webProbeScriptGovernanceHints(options),
|
||||
preferredCommands: webProbeScriptPreferredCommands(options),
|
||||
};
|
||||
}
|
||||
|
||||
export function webProbeScriptCommandPromotionHint(options: NodeWebProbeScriptOptions): Record<string, unknown> {
|
||||
const repoOwnedTypedCommand = options.suppressAdHocWarning === true;
|
||||
const currentCommand = options.commandLabel ?? `web-probe script --node ${options.node} --lane ${options.lane}`;
|
||||
return {
|
||||
schemaVersion: "unidesk.web-probe.command-promotion.v1",
|
||||
code: "prefer_repo_owned_typed_command",
|
||||
promotionRequiredBeforeRepeat: !repoOwnedTypedCommand,
|
||||
repoOwnedTypedCommand,
|
||||
sourceKind: options.scriptSource.kind,
|
||||
currentCommand,
|
||||
preferredArtifact: "repo-owned-typed-command",
|
||||
preferredSurface: "web-probe observe command or a dedicated web-probe subcommand",
|
||||
action: repoOwnedTypedCommand ? "reuse-repo-owned-typed-command" : "promote-before-repeat",
|
||||
message: repoOwnedTypedCommand
|
||||
? "This probe is already a repo-owned typed command; reuse its public command instead of copying the generated script."
|
||||
: "Before repeating this one-off script, promote the workflow to a repo-owned typed web-probe command.",
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
|
||||
export function webProbeScriptGovernanceWarnings(options: NodeWebProbeScriptOptions): Record<string, unknown>[] {
|
||||
if (options.suppressAdHocWarning === true) return [];
|
||||
return [{
|
||||
code: "web_probe_script_ad_hoc_only",
|
||||
severity: "warning",
|
||||
message: "web-probe script is for one-off exploration; repeated or high-frequency checks must be promoted to a typed command instead of rerunning temporary scripts.",
|
||||
requiredAction: "promote-before-repeat",
|
||||
preferredArtifact: "repo-owned-typed-command",
|
||||
preferredSurface: "web-probe observe command or a dedicated web-probe subcommand",
|
||||
temporaryScriptScope: "one-off-exploration-only",
|
||||
message: "web-probe script is a one-off exploration escape hatch; before repeating it, promote the workflow to a repo-owned typed command instead of rerunning temporary scripts.",
|
||||
node: options.node,
|
||||
lane: options.lane,
|
||||
valuesRedacted: true,
|
||||
}];
|
||||
}
|
||||
|
||||
function webProbeScriptGovernanceHints(options: NodeWebProbeScriptOptions): string[] {
|
||||
export function webProbeScriptGovernanceHints(options: NodeWebProbeScriptOptions): string[] {
|
||||
if (options.generatedHints !== undefined) return options.generatedHints;
|
||||
return [
|
||||
"Before rerunning this script, promote the workflow to a repo-owned typed `web-probe` command so auth, artifacts, output bounds and evidence contracts remain reusable.",
|
||||
"Prefer `web-probe observe start` plus `web-probe observe command` for interactive flows; use `observe collect/analyze` for repeated evidence reads.",
|
||||
"If the same script is needed more than once, add or extend a reusable command type in the web-probe observe command surface.",
|
||||
`For this target, start with: bun scripts/cli.ts web-probe observe start --node ${options.node} --lane ${options.lane} --target-path /projects/mdtodo`,
|
||||
];
|
||||
}
|
||||
|
||||
function webProbeScriptPreferredCommands(options: NodeWebProbeScriptOptions): Record<string, string> {
|
||||
export function webProbeScriptPreferredCommands(options: NodeWebProbeScriptOptions): Record<string, string> {
|
||||
if (options.generatedPreferredCommands !== undefined) return options.generatedPreferredCommands;
|
||||
return {
|
||||
typedCommandCatalog: "bun scripts/cli.ts web-probe --help",
|
||||
startObserver: `bun scripts/cli.ts web-probe observe start --node ${options.node} --lane ${options.lane} --target-path /projects/mdtodo`,
|
||||
mdtodoSummary: "bun scripts/cli.ts web-probe observe collect <observerId> --view project-mdtodo-summary",
|
||||
analyze: "bun scripts/cli.ts web-probe observe analyze <observerId>",
|
||||
addCommandType: "Add a repo-owned web-probe observe command type before repeating this script.",
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -4,7 +4,30 @@ import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { test } from "bun:test";
|
||||
|
||||
import { buildWebObserveCommandVisibility, resolveWebObserveActionJson } from "./web-probe-observe-actions";
|
||||
import { hwlabRuntimeLaneSpecForNode } from "../hwlab-node-lanes";
|
||||
import { buildWebObserveCommandVisibility, nodeWebProbeRealtimeFanoutProfiles, resolveWebObserveActionJson } from "./web-probe-observe-actions";
|
||||
|
||||
test("realtime fanout runner contract derives topics, groups, and independent capabilities from owning YAML", () => {
|
||||
const profiles = nodeWebProbeRealtimeFanoutProfiles(hwlabRuntimeLaneSpecForNode("v03", "NC01"));
|
||||
const expectedKafka = profiles["pure-kafka-live"]?.expectedKafka as Record<string, any>;
|
||||
assert.deepEqual(expectedKafka.topics, {
|
||||
stdio: "codex-stdio.raw.v1",
|
||||
agentrun: "agentrun.event.v1",
|
||||
hwlab: "hwlab.event.v1",
|
||||
});
|
||||
assert.deepEqual(expectedKafka.groups, {
|
||||
directPublish: "hwlab-v03-agentrun-event-direct-publish",
|
||||
transactionalProjector: "hwlab-v03-agentrun-event-projector",
|
||||
liveSse: "hwlab-v03-workbench-live-sse",
|
||||
});
|
||||
assert.deepEqual(expectedKafka.capabilities, {
|
||||
directPublish: true,
|
||||
liveKafkaSse: true,
|
||||
transactionalProjector: false,
|
||||
projectionOutboxRelay: false,
|
||||
projectionRealtime: false,
|
||||
});
|
||||
});
|
||||
|
||||
test("web observe command visibility does not equate control completion with async turn completion", () => {
|
||||
const visibility = buildWebObserveCommandVisibility({
|
||||
|
||||
@@ -17,6 +17,37 @@ import { commandSummaryForOutput, nodeWebObserveForceStopNodeScript, nodeWebObse
|
||||
|
||||
type WebObserveActionJsonContract = "gc" | "start" | "status" | "command" | "force-stop";
|
||||
|
||||
export function nodeWebProbeRealtimeFanoutProfiles(spec: HwlabRuntimeLaneSpec): Record<string, Record<string, unknown>> {
|
||||
const profiles = spec.webProbe?.realtimeFanoutProfiles ?? {};
|
||||
if (Object.keys(profiles).length === 0) return {};
|
||||
const kafka = spec.codeAgentRuntime?.kafkaEventBridge;
|
||||
if (kafka === undefined) throw new Error("realtimeFanoutProfiles requires codeAgentRuntime.kafkaEventBridge in the owning YAML");
|
||||
const capabilities = Object.fromEntries(Object.entries(kafka.features).map(([name, enabled]) => [name, kafka.enabled && enabled]));
|
||||
return Object.fromEntries(Object.entries(profiles).map(([id, profile]) => {
|
||||
if (profile.businessEvent !== kafka.hwlabEventTopic) {
|
||||
throw new Error(`realtimeFanoutProfiles.${id}.businessEvent must match codeAgentRuntime.kafkaEventBridge.hwlabEventTopic`);
|
||||
}
|
||||
return [id, {
|
||||
...profile,
|
||||
expectedKafka: {
|
||||
enabled: kafka.enabled,
|
||||
topics: {
|
||||
stdio: kafka.stdioTopic,
|
||||
agentrun: kafka.agentRunEventTopic,
|
||||
hwlab: kafka.hwlabEventTopic,
|
||||
},
|
||||
groups: {
|
||||
directPublish: kafka.directPublishConsumerGroupId,
|
||||
transactionalProjector: kafka.transactionalProjectorConsumerGroupId,
|
||||
liveSse: kafka.hwlabEventConsumerGroupId,
|
||||
},
|
||||
capabilities,
|
||||
valuesRedacted: true,
|
||||
},
|
||||
}];
|
||||
}));
|
||||
}
|
||||
|
||||
export function resolveWebObserveActionJson(
|
||||
result: { stdout: string; stderr?: string; exitCode?: number | null; timedOut?: boolean },
|
||||
contract: WebObserveActionJsonContract,
|
||||
@@ -439,6 +470,7 @@ export function runNodeWebProbeObserveStart(
|
||||
`UNIDESK_WEB_OBSERVE_ALERT_THRESHOLDS_JSON=${shellQuote(JSON.stringify(alertThresholds))}`,
|
||||
`UNIDESK_WEB_OBSERVE_BROWSER_FREEZE_POLICY_JSON=${shellQuote(JSON.stringify(browserFreezePolicy))}`,
|
||||
`UNIDESK_WEB_OBSERVE_PROJECT_MANAGEMENT_JSON=${shellQuote(JSON.stringify(projectManagement))}`,
|
||||
`UNIDESK_WEB_OBSERVE_REALTIME_FANOUT_PROFILES_JSON=${shellQuote(JSON.stringify(nodeWebProbeRealtimeFanoutProfiles(spec)))}`,
|
||||
...(authLogin === null
|
||||
? []
|
||||
: [
|
||||
@@ -583,7 +615,7 @@ export function readNodeWebProbeObserveRemoteStatus(
|
||||
const script = [
|
||||
"set -eu",
|
||||
nodeWebObserveResolveStateDirShell(options),
|
||||
nodeWebObserveStatusNodeScript(tailLines, options.node, options.lane),
|
||||
nodeWebObserveStatusNodeScript(tailLines, options.node, options.lane, options.collectCommandId),
|
||||
].join("\n");
|
||||
const result = runTransWorkspaceStdinScript(options.node, spec.workspace, script, timeoutSeconds);
|
||||
const statusResolution = resolveWebObserveActionJson(result, "status");
|
||||
@@ -668,6 +700,15 @@ export function buildWebObserveCommandVisibility(input: {
|
||||
export function runNodeWebProbeObserveCommand(options: NodeWebProbeObserveOptions, spec: HwlabRuntimeLaneSpec, stopCommand: boolean): Record<string, unknown> | RenderedCliResult {
|
||||
const type = options.commandType ?? (stopCommand ? "stop" : null);
|
||||
if (type === null) throw new Error("web-probe observe command requires --type");
|
||||
if (type === "validateRealtimeFanout") {
|
||||
if (!options.commandProfile?.trim()) throw new Error("validateRealtimeFanout requires --profile");
|
||||
if (spec.webProbe?.realtimeFanoutProfiles?.[options.commandProfile.trim()] === undefined) {
|
||||
throw new Error(`validateRealtimeFanout profile is not declared by the owning YAML: ${options.commandProfile.trim()}`);
|
||||
}
|
||||
if (!options.commandProvider?.trim()) throw new Error("validateRealtimeFanout requires --provider");
|
||||
if (!options.commandText?.trim()) throw new Error("validateRealtimeFanout requires --text or --text-stdin for the first turn");
|
||||
if (!options.commandSecondText?.trim()) throw new Error("validateRealtimeFanout requires --second-text for the follow-up turn");
|
||||
}
|
||||
const commandId = `cmd-${Date.now().toString(36)}-${randomBytes(3).toString("hex")}`;
|
||||
const payload = {
|
||||
id: commandId,
|
||||
@@ -676,9 +717,11 @@ export function runNodeWebProbeObserveCommand(options: NodeWebProbeObserveOption
|
||||
source: "cli",
|
||||
path: options.commandPath,
|
||||
text: options.commandText,
|
||||
secondText: options.commandSecondText,
|
||||
label: options.commandLabel,
|
||||
sessionId: options.commandSessionId,
|
||||
provider: options.commandProvider,
|
||||
profile: options.commandProfile,
|
||||
afterRound: options.commandAfterRound,
|
||||
severity: options.commandSeverity,
|
||||
alternateSessionStrategy: options.commandAlternateSessionStrategy,
|
||||
|
||||
@@ -124,10 +124,12 @@ function collectOptions(): NodeWebProbeObserveOptions {
|
||||
force: false,
|
||||
commandType: null,
|
||||
commandText: null,
|
||||
commandSecondText: null,
|
||||
commandPath: null,
|
||||
commandLabel: null,
|
||||
commandSessionId: null,
|
||||
commandProvider: null,
|
||||
commandProfile: null,
|
||||
commandAfterRound: null,
|
||||
commandSeverity: null,
|
||||
commandAlternateSessionStrategy: null,
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "bun:test";
|
||||
|
||||
import { hwlabRuntimeLaneSpecForNode } from "../hwlab-node-lanes";
|
||||
import { parseNodeWebProbeObserveOptions } from "./web-probe-observe";
|
||||
|
||||
const spec = hwlabRuntimeLaneSpecForNode("v03", "NC01");
|
||||
|
||||
function parseRealtimeCommand(extra: string[]) {
|
||||
return parseNodeWebProbeObserveOptions(
|
||||
["command", "--type", "validateRealtimeFanout", ...extra],
|
||||
"NC01",
|
||||
"v03",
|
||||
spec,
|
||||
"webobs-fixture",
|
||||
null,
|
||||
);
|
||||
}
|
||||
|
||||
test("validateRealtimeFanout parses the YAML profile and two independent prompts", () => {
|
||||
const options = parseRealtimeCommand([
|
||||
"--profile", "pure-kafka-live",
|
||||
"--provider", "gpt.pika",
|
||||
"--text", "first turn",
|
||||
"--second-text", "second turn",
|
||||
]);
|
||||
|
||||
assert.equal(options.commandType, "validateRealtimeFanout");
|
||||
assert.equal(options.commandProfile, "pure-kafka-live");
|
||||
assert.equal(options.commandProvider, "gpt.pika");
|
||||
assert.equal(options.commandText, "first turn");
|
||||
assert.equal(options.commandSecondText, "second turn");
|
||||
assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.debugStreams, ["stdio", "agentrun", "hwlab"]);
|
||||
assert.equal(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.fromBeginning, false);
|
||||
});
|
||||
|
||||
test("validateRealtimeFanout refuses body as a substitute for the second prompt", () => {
|
||||
assert.throws(
|
||||
() => parseRealtimeCommand([
|
||||
"--profile", "pure-kafka-live",
|
||||
"--provider", "gpt.pika",
|
||||
"--text", "first turn",
|
||||
"--body", "not a second prompt",
|
||||
]),
|
||||
/requires --second-text/u,
|
||||
);
|
||||
});
|
||||
|
||||
test("validateRealtimeFanout rejects profiles absent from the owning YAML", () => {
|
||||
assert.throws(
|
||||
() => parseRealtimeCommand([
|
||||
"--profile", "projection-mode",
|
||||
"--provider", "gpt.pika",
|
||||
"--text", "first turn",
|
||||
"--second-text", "second turn",
|
||||
]),
|
||||
/profile is not declared by the owning YAML/u,
|
||||
);
|
||||
});
|
||||
@@ -419,10 +419,12 @@ export function parseNodeWebProbeObserveOptions(
|
||||
"--job-id",
|
||||
"--type",
|
||||
"--text",
|
||||
"--second-text",
|
||||
"--path",
|
||||
"--label",
|
||||
"--session-id",
|
||||
"--provider",
|
||||
"--profile",
|
||||
"--account-id",
|
||||
"--account",
|
||||
"--from-account-id",
|
||||
@@ -501,7 +503,10 @@ export function parseNodeWebProbeObserveOptions(
|
||||
throw new Error("web-probe observe command accepts either --text or --text-stdin, not both");
|
||||
}
|
||||
const commandText = commandTextFromStdin ? readFileSync(0, "utf8") : commandTextOption;
|
||||
const commandSecondText = optionValue(args, "--second-text") ?? null;
|
||||
const commandSourceId = optionValue(args, "--source-id") ?? null;
|
||||
const commandProfile = optionValue(args, "--profile") ?? null;
|
||||
const commandProvider = optionValue(args, "--provider") ?? null;
|
||||
const commandAccountId = optionValue(args, "--account-id") ?? optionValue(args, "--account") ?? null;
|
||||
const commandFromAccountId = optionValue(args, "--from-account-id") ?? null;
|
||||
const commandToAccountId = optionValue(args, "--to-account-id") ?? null;
|
||||
@@ -540,6 +545,8 @@ export function parseNodeWebProbeObserveOptions(
|
||||
const commandBlocking = args.includes("--blocking") ? true : args.includes("--non-blocking") ? false : null;
|
||||
for (const [label, value] of [
|
||||
["--severity", commandSeverity],
|
||||
["--profile", commandProfile],
|
||||
["--second-text", commandSecondText],
|
||||
["--alternate-session-strategy", commandAlternateSessionStrategy],
|
||||
["--expected-sentinel-range", commandExpectedSentinelRange],
|
||||
["--finding-id", commandFindingId],
|
||||
@@ -563,6 +570,15 @@ export function parseNodeWebProbeObserveOptions(
|
||||
] as const) {
|
||||
if (value !== null && (value.includes("\0") || value.length > 500)) throw new Error(`unsafe web-probe observe ${label}: expected 1-500 non-NUL chars`);
|
||||
}
|
||||
if (observeActionRaw === "command" && commandType === "validateRealtimeFanout") {
|
||||
if (!commandProfile?.trim()) throw new Error("validateRealtimeFanout requires --profile");
|
||||
if (spec.webProbe?.realtimeFanoutProfiles?.[commandProfile.trim()] === undefined) {
|
||||
throw new Error(`validateRealtimeFanout profile is not declared by the owning YAML: ${commandProfile.trim()}`);
|
||||
}
|
||||
if (!commandProvider?.trim()) throw new Error("validateRealtimeFanout requires --provider");
|
||||
if (!commandText?.trim()) throw new Error("validateRealtimeFanout requires --text or --text-stdin for the first turn");
|
||||
if (!commandSecondText?.trim()) throw new Error("validateRealtimeFanout requires --second-text for the follow-up turn");
|
||||
}
|
||||
return {
|
||||
action: "observe",
|
||||
observeAction: observeActionRaw,
|
||||
@@ -606,10 +622,12 @@ export function parseNodeWebProbeObserveOptions(
|
||||
force: args.includes("--force"),
|
||||
commandType,
|
||||
commandText,
|
||||
commandSecondText,
|
||||
commandPath: optionValue(args, "--path") ?? null,
|
||||
commandLabel: optionValue(args, "--label") ?? null,
|
||||
commandSessionId: optionValue(args, "--session-id") ?? null,
|
||||
commandProvider: optionValue(args, "--provider") ?? null,
|
||||
commandProvider,
|
||||
commandProfile,
|
||||
commandAfterRound,
|
||||
commandSeverity,
|
||||
commandAlternateSessionStrategy,
|
||||
@@ -651,6 +669,7 @@ export function parseNodeWebProbeObserveCommandType(value: string): NodeWebProbe
|
||||
|| value === "goto"
|
||||
|| value === "newSession"
|
||||
|| value === "sendPrompt"
|
||||
|| value === "validateRealtimeFanout"
|
||||
|| value === "steer"
|
||||
|| value === "cancel"
|
||||
|| value === "selectProvider"
|
||||
@@ -686,7 +705,7 @@ export function parseNodeWebProbeObserveCommandType(value: string): NodeWebProbe
|
||||
|| value === "mark"
|
||||
|| value === "stop"
|
||||
) return value;
|
||||
throw new Error(`web-probe observe command --type must be login, loginAccount, logout, listSessions, switchSessions, preflight, goto, gotoProjectMdtodo, newSession, sendPrompt, steer, cancel, selectProvider, clickSession, refreshCurrentSession, switchAwayAndBack, assertSessionInvariant, selectProjectSource, selectMdtodoSource, selectMdtodoFile, selectMdtodoTask, expandMdtodoTask, openMdtodoReportPreview, toggleMdtodoReportFullscreen, openMdtodoSourceConfig, closeMdtodoSourceConfig, configureMdtodoHwpodSource, probeMdtodoSource, reindexMdtodoSource, editMdtodoTaskInline, editMdtodoTaskTitle, editMdtodoTaskBody, toggleMdtodoTaskStatus, addMdtodoRootTask, addMdtodoSubTask, continueMdtodoTask, deleteMdtodoTask, launchWorkbenchFromTask, launchWorkbenchFromMdtodo, performanceCapture, screenshot, mark, or stop; got ${value}`);
|
||||
throw new Error(`web-probe observe command --type must be login, loginAccount, logout, listSessions, switchSessions, preflight, goto, gotoProjectMdtodo, newSession, sendPrompt, validateRealtimeFanout, steer, cancel, selectProvider, clickSession, refreshCurrentSession, switchAwayAndBack, assertSessionInvariant, selectProjectSource, selectMdtodoSource, selectMdtodoFile, selectMdtodoTask, expandMdtodoTask, openMdtodoReportPreview, toggleMdtodoReportFullscreen, openMdtodoSourceConfig, closeMdtodoSourceConfig, configureMdtodoHwpodSource, probeMdtodoSource, reindexMdtodoSource, editMdtodoTaskInline, editMdtodoTaskTitle, editMdtodoTaskBody, toggleMdtodoTaskStatus, addMdtodoRootTask, addMdtodoSubTask, continueMdtodoTask, deleteMdtodoTask, launchWorkbenchFromTask, launchWorkbenchFromMdtodo, performanceCapture, screenshot, mark, or stop; got ${value}`);
|
||||
}
|
||||
|
||||
export function parseWebProbeBrowserProxyMode(value: string | undefined): WebProbeBrowserProxyMode {
|
||||
|
||||
@@ -1627,6 +1627,7 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti
|
||||
expectValue("HWLAB_KAFKA_ENABLED", String(kafkaEventBridge.enabled));
|
||||
expectValue("HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED", String(kafkaEventBridge.enabled && (kafkaEventBridge.features.directPublish || kafkaEventBridge.features.transactionalProjector)));
|
||||
expectValue("HWLAB_KAFKA_BOOTSTRAP_SERVERS", kafkaEventBridge.bootstrapServers);
|
||||
expectValue("HWLAB_KAFKA_STDIO_TOPIC", kafkaEventBridge.stdioTopic);
|
||||
expectValue("HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC", kafkaEventBridge.agentRunEventTopic);
|
||||
expectValue("HWLAB_KAFKA_EVENT_TOPIC", kafkaEventBridge.hwlabEventTopic);
|
||||
expectValue("HWLAB_KAFKA_CLIENT_ID", kafkaEventBridge.clientId);
|
||||
|
||||
Reference in New Issue
Block a user