Merge pull request #1785 from pikasTech/fix/1781-webprobe-existing-refresh
fix(web-probe): 增加既有会话 Kafka refresh→live 验收命令
This commit is contained in:
Lyon
@@ -95,6 +95,7 @@ export function hwlabNodeWebProbeHelp(): Record<string, unknown> {
|
||||
"bun scripts/cli.ts web-probe observe start --node <node> --lane <lane> --origin public --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 command webobs-xxxx --type validateExistingSessionRefresh --profile pure-kafka-live --session-id ses_existing",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type validateWorkbenchKafkaDebugReplay",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type validateWorkbenchTraceReadability",
|
||||
"bun scripts/cli.ts web-probe observe status webobs-xxxx --command-id cmd-xxxx",
|
||||
@@ -147,6 +148,7 @@ export function hwlabNodeWebProbeHelp(): Record<string, unknown> {
|
||||
"After observe start, prefer observe status|command|stop|collect|analyze <id> instead of repeating --node/--lane/--state-dir.",
|
||||
"Typed observe commands inherit the semantic origin selected by observe start and must not embed a replacement URL or IP.",
|
||||
"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.",
|
||||
"validateExistingSessionRefresh is a YAML-profiled background command: it submits no prompt and proves an existing session keeps DOM identity while Kafka retention reaches the shared live SSE phase.",
|
||||
"validateWorkbenchKafkaDebugReplay 复用 observer 当前 Workbench session，以 owning YAML 中的 topic、group prefix 和超时验证隔离 Kafka 重放，并通过精确 command id 查询结果。",
|
||||
"validateWorkbenchTraceReadability 只检查当前 Workbench 会话中的产品 Trace 卡片。",
|
||||
"该命令排除隔离调试面板，并分别验证运行中默认展开、source authority 与终态保留。",
|
||||
|
||||
@@ -176,6 +176,14 @@ function renderWebObserveStatusTable(value: Record<string, unknown>): string {
|
||||
]]),
|
||||
"",
|
||||
] : []),
|
||||
...(exactCommand.type === "validateExistingSessionRefresh" && exactReplayEvidence !== null ? [
|
||||
...renderExistingSessionRefreshEvidence(
|
||||
exactReplayEvidence,
|
||||
exactResult?.reportSha256 ?? exactErrorDetails?.reportSha256,
|
||||
record(exactResult?.screenshot ?? exactErrorDetails?.screenshot)?.sha256,
|
||||
),
|
||||
"",
|
||||
] : []),
|
||||
...(exactCommand.type === "validateWorkbenchKafkaDebugReplay" && exactReplayEvidence !== null ? [
|
||||
...renderWorkbenchKafkaDebugReplayEvidence(
|
||||
exactReplayEvidence,
|
||||
@@ -236,6 +244,69 @@ function renderWebObserveStatusTable(value: Record<string, unknown>): string {
|
||||
return lines.join("\n");
|
||||
}
|
||||
|
||||
function renderExistingSessionRefreshEvidence(
|
||||
evidence: Record<string, unknown>,
|
||||
reportSha: unknown,
|
||||
screenshotSha: unknown,
|
||||
): string[] {
|
||||
const before = record(evidence.before);
|
||||
const after = record(evidence.after);
|
||||
const replay = record(evidence.refreshReplay);
|
||||
const counts = record(replay?.counts);
|
||||
const product = record(evidence.productSse);
|
||||
const failure = nullableRecord(evidence.failure);
|
||||
const traceIds = Array.isArray(evidence.traceIds) ? evidence.traceIds.slice(0, 4).join(",") : "";
|
||||
const partitions = Array.isArray(replay?.topicPartitions) ? replay.topicPartitions.slice(0, 8).join(",") : "";
|
||||
const barrier = Array.isArray(replay?.barrier)
|
||||
? replay.barrier.slice(0, 8).map((item) => {
|
||||
const row = record(item);
|
||||
return `${webObserveText(row.partition)}:${webObserveText(row.endOffset)}`;
|
||||
}).join(",")
|
||||
: "";
|
||||
return [
|
||||
"Workbench 既有 Session 刷新：",
|
||||
webObserveTable(["SESSION", "TRACE_IDS", "PHASE", "CODE", "MSG_BEFORE", "MSG_AFTER", "SOURCES", "OPEN", "CONNECTED", "BUSINESS", "FAILURES", "FORBIDDEN", "REPORT_SHA", "SCREENSHOT_SHA"], [[
|
||||
webObserveShort(webObserveText(evidence.sessionId), 32),
|
||||
webObserveShort(webObserveText(traceIds), 40),
|
||||
evidence.phase,
|
||||
evidence.code,
|
||||
before?.messageCount,
|
||||
after?.messageCount,
|
||||
product?.browserSourceCount,
|
||||
product?.openCount,
|
||||
product?.connectedEventCount,
|
||||
product?.businessEventCount,
|
||||
product?.typedFailureCount,
|
||||
evidence.forbiddenRequestCount,
|
||||
webObserveShort(webObserveText(reportSha), 32),
|
||||
webObserveShort(webObserveText(screenshotSha), 32),
|
||||
]]),
|
||||
"Kafka refresh handoff:",
|
||||
webObserveTable(["TOPIC", "PARTITIONS", "BARRIER", "MATCHED", "REPLAYED", "BUFFERED", "BUFFERED_DELIVERED", "LIVE_DELIVERED", "DEDUPLICATED", "LIVE"], [[
|
||||
webObserveShort(webObserveText(replay?.topic), 32),
|
||||
webObserveShort(webObserveText(partitions), 24),
|
||||
webObserveShort(webObserveText(barrier), 56),
|
||||
counts?.matched,
|
||||
counts?.replayed,
|
||||
counts?.buffered,
|
||||
counts?.bufferedDelivered,
|
||||
counts?.liveDelivered,
|
||||
counts?.deduplicated,
|
||||
product?.livePhase,
|
||||
]]),
|
||||
...(failure !== null ? [
|
||||
"Typed failure:",
|
||||
webObserveTable(["PHASE", "CODE", "STREAM_PHASE", "STREAM_CODE", "MESSAGE"], [[
|
||||
failure.phase,
|
||||
failure.code,
|
||||
failure.streamPhase,
|
||||
failure.streamCode,
|
||||
webObserveShort(webObserveText(failure.message), 96),
|
||||
]]),
|
||||
] : []),
|
||||
];
|
||||
}
|
||||
|
||||
function renderWorkbenchKafkaDebugReplayEvidence(
|
||||
evidence: Record<string, unknown>,
|
||||
reportSha: unknown,
|
||||
@@ -391,6 +462,14 @@ function renderWebObserveCommandTable(value: Record<string, unknown>): string {
|
||||
replayScreenshot?.sha256 ?? resultScreenshot?.sha256,
|
||||
),
|
||||
] : []),
|
||||
...(observerCommand?.type === "validateExistingSessionRefresh" && replayEvidence !== null ? [
|
||||
"",
|
||||
...renderExistingSessionRefreshEvidence(
|
||||
replayEvidence,
|
||||
result?.reportSha256 ?? details?.reportSha256,
|
||||
replayScreenshot?.sha256 ?? resultScreenshot?.sha256,
|
||||
),
|
||||
] : []),
|
||||
...(observerCommand?.type === "validateWorkbenchTraceReadability" && replayEvidence !== null ? [
|
||||
"",
|
||||
...renderWorkbenchTraceReadabilityEvidence(
|
||||
|
||||
@@ -102,6 +102,7 @@ async function processCommand(command) {
|
||||
expectedActionWaitMs: command.expectedActionWaitMs,
|
||||
}), "sendPrompt");
|
||||
case "validateRealtimeFanout": return validateRealtimeFanout(command);
|
||||
case "validateExistingSessionRefresh": return validateExistingSessionRefresh(command);
|
||||
case "validateWorkbenchKafkaDebugReplay": return withObserverSync(await validateWorkbenchKafkaDebugReplay(command), "validateWorkbenchKafkaDebugReplay");
|
||||
case "validateWorkbenchTraceReadability": return withObserverSync(await validateWorkbenchTraceReadability(command), "validateWorkbenchTraceReadability");
|
||||
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");
|
||||
|
||||
@@ -172,6 +172,88 @@ test("realtime fanout validates YAML-owned refresh and legacy live-only connecte
|
||||
}, sessionId, "live", liveProfile));
|
||||
});
|
||||
|
||||
test("existing session refresh rejects duplicate or reordered Workbench identities", () => {
|
||||
const source = nodeWebObserveRunnerRealtimeSource();
|
||||
const build = new Function(`${source}\nreturn realtimeAssertSameWorkbenchDom;`) as () => (
|
||||
before: Record<string, unknown>, after: Record<string, unknown>,
|
||||
) => void;
|
||||
const assertSame = build();
|
||||
const before = { messageCount: 2, identities: ["user:m1:", "agent:m2:t1"], missingIdentityRows: [], duplicateIdentities: [] };
|
||||
assert.doesNotThrow(() => assertSame(before, { ...before }));
|
||||
assert.throws(() => assertSame(before, { ...before, identities: [...before.identities].reverse() }), /identity or order changed/u);
|
||||
assert.throws(() => assertSame(before, { ...before, duplicateIdentities: ["agent:m2:t1"] }), /duplicate message identities/u);
|
||||
assert.throws(() => assertSame(before, { ...before, missingIdentityRows: [1] }), /without a stable message identity/u);
|
||||
});
|
||||
|
||||
test("existing session refresh failures preserve typed phase and code", () => {
|
||||
const source = nodeWebObserveRunnerRealtimeSource();
|
||||
const build = new Function(`function errorSummary(error) { return { message: error.message }; }\n${source}\nreturn { realtimeExistingRefreshFailure, realtimeExistingRefreshStreamError };`) as () => {
|
||||
realtimeExistingRefreshFailure: (phase: string, error: Error & { details?: Record<string, unknown> }) => Record<string, unknown>;
|
||||
realtimeExistingRefreshStreamError: (value: Record<string, unknown>) => Error & { details?: Record<string, unknown> };
|
||||
};
|
||||
const helpers = build();
|
||||
const error = helpers.realtimeExistingRefreshStreamError({
|
||||
phase: "flushing",
|
||||
error: { code: "workbench_kafka_refresh_barrier_gap", message: "Kafka retention barrier gap", retryable: true },
|
||||
fallback: false,
|
||||
});
|
||||
const failure = helpers.realtimeExistingRefreshFailure("retention-barrier", error);
|
||||
assert.equal(failure.phase, "retention-barrier");
|
||||
assert.equal(failure.code, "retention_barrier_failed");
|
||||
assert.equal(failure.streamPhase, "flushing");
|
||||
assert.equal(failure.streamCode, "workbench_kafka_refresh_barrier_gap");
|
||||
assert.equal((failure.streamFailure as Record<string, unknown>).fallback, false);
|
||||
assert.deepEqual(failure.valuesRedacted, true);
|
||||
});
|
||||
|
||||
test("existing session refresh preserves rejected connected evidence and rejects missing browser replay delivery", () => {
|
||||
const source = nodeWebObserveRunnerRealtimeSource();
|
||||
const build = new Function(
|
||||
"sanitize",
|
||||
`${source}\nreturn { realtimeExistingRefreshConnectedEvidence, realtimeAssertExistingRefreshDelivery };`,
|
||||
) as (sanitize: <T>(value: T) => T) => {
|
||||
realtimeExistingRefreshConnectedEvidence: (connected: Record<string, any>, sessionId: string, profile: Record<string, any>) => Record<string, unknown>;
|
||||
realtimeAssertExistingRefreshDelivery: (browser: Record<string, unknown>, refreshReplay: Record<string, any>) => void;
|
||||
};
|
||||
const helpers = build((value) => value);
|
||||
const sessionId = "ses_existing";
|
||||
const refreshReplay = {
|
||||
phase: "live",
|
||||
topic: "hwlab.event.v1",
|
||||
topicPartitions: [0],
|
||||
barrier: [{ partition: 0, endOffset: "18" }],
|
||||
counts: { matched: 8, replayed: 8, buffered: 1, bufferedDelivered: 1, liveDelivered: 0, deduplicated: 0 },
|
||||
};
|
||||
const profile = {
|
||||
expectedKafka: { topics: { hwlab: "hwlab.event.v1" }, capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: true } },
|
||||
expectedProductSse: { deliverySemantics: "kafka-retention-then-live", liveOnly: false, replay: true, replaySupported: true, lossPossible: false, refreshHandoff: true },
|
||||
};
|
||||
const connected = {
|
||||
realtimeSource: "hwlab.event.v1",
|
||||
deliverySemantics: "invalid",
|
||||
liveOnly: false,
|
||||
replay: true,
|
||||
replaySupported: true,
|
||||
lossPossible: false,
|
||||
filters: { sessionId, traceId: null },
|
||||
capabilities: profile.expectedKafka.capabilities,
|
||||
refreshReplay,
|
||||
};
|
||||
assert.throws(
|
||||
() => helpers.realtimeExistingRefreshConnectedEvidence(connected, sessionId, profile),
|
||||
(error: Error & { details?: Record<string, any> }) => {
|
||||
assert.deepEqual(error.details?.refreshReplay?.barrier, refreshReplay.barrier);
|
||||
assert.equal(error.details?.refreshReplay?.counts?.replayed, 8);
|
||||
return true;
|
||||
},
|
||||
);
|
||||
assert.doesNotThrow(() => helpers.realtimeAssertExistingRefreshDelivery({ businessEventCount: 9 }, refreshReplay));
|
||||
assert.throws(
|
||||
() => helpers.realtimeAssertExistingRefreshDelivery({ businessEventCount: 8 }, refreshReplay),
|
||||
/fewer than replayed plus buffered delivery 9/u,
|
||||
);
|
||||
});
|
||||
|
||||
test("realtime fanout proves one formal user event through AgentRun, HWLAB Kafka, and product SSE", () => {
|
||||
const source = nodeWebObserveRunnerRealtimeSource();
|
||||
const build = new Function(`${source}\nreturn realtimeValidateTurnEvidence;`) as () => (input: Record<string, unknown>) => void;
|
||||
|
||||
@@ -327,6 +327,391 @@ async function validateRealtimeFanout(command) {
|
||||
}
|
||||
}
|
||||
|
||||
async function validateExistingSessionRefresh(command) {
|
||||
const profileId = String(command.profile || "").trim();
|
||||
const profile = realtimeFanoutProfiles[profileId];
|
||||
if (!profile || typeof profile !== "object") throw new Error("validateExistingSessionRefresh profile is not declared by the owning YAML: " + profileId);
|
||||
const effective = realtimeFanoutEffectiveProfile(profile, command.durationMs);
|
||||
if (effective.expectedProductSse.refreshHandoff !== true) throw new Error("validateExistingSessionRefresh requires a YAML profile with refresh handoff enabled");
|
||||
const reportDir = path.join(stateDir, "artifacts", "existing-session-refresh", 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);
|
||||
};
|
||||
const report = {
|
||||
ok: false,
|
||||
contract: "web-probe-existing-session-refresh-v1",
|
||||
commandId: command.id,
|
||||
profile: profileId,
|
||||
origin,
|
||||
startedAt: new Date().toISOString(),
|
||||
valuesRedacted: true,
|
||||
};
|
||||
let phase = "session-selection";
|
||||
let sessionId = null;
|
||||
let refreshReplay = null;
|
||||
let screenshot = null;
|
||||
let requestPage = null;
|
||||
try {
|
||||
const controlRecovery = await ensureControlPageResponsiveForCommand("validateExistingSessionRefresh");
|
||||
const initial = await workbenchSessionSnapshot();
|
||||
sessionId = String(command.sessionId || initial?.activeSessionId || initial?.routeSessionId || "").trim();
|
||||
if (!sessionId) throw new Error("validateExistingSessionRefresh requires --session-id or an active Workbench session");
|
||||
if (!/^ses_[A-Za-z0-9_-]+$/u.test(sessionId)) throw new Error("validateExistingSessionRefresh session id is invalid");
|
||||
Object.assign(report, { sessionId, controlRecovery });
|
||||
if (initial?.activeSessionId !== sessionId || initial?.routeSessionId !== sessionId) {
|
||||
await gotoTarget("/workbench/sessions/" + encodeURIComponent(sessionId));
|
||||
const selected = await waitForWorkbenchSessionHydrated(page, sessionId, { timeoutMs: effective.barrierTimeoutMs });
|
||||
if (selected.ok !== true) throw new Error("existing session did not hydrate before refresh validation");
|
||||
}
|
||||
const before = await realtimeWorkbenchDomIdentity();
|
||||
if (before.messageCount < 1) throw new Error("existing session has no rendered messages to validate across refresh");
|
||||
report.before = before;
|
||||
|
||||
phase = "refresh-navigation";
|
||||
requestPage = page;
|
||||
requestPage.on("request", requestListener);
|
||||
await realtimeArmExistingSessionRefreshProbe(requestPage, effective);
|
||||
await requestPage.reload({ waitUntil: "domcontentloaded", timeout: effective.barrierTimeoutMs });
|
||||
const settled = await waitForWorkbenchSessionHydrated(page, sessionId, { timeoutMs: effective.barrierTimeoutMs });
|
||||
if (settled.ok !== true) throw new Error("existing session did not hydrate after refresh");
|
||||
phase = "retention-barrier";
|
||||
const refreshState = await realtimePoll(async () => page.evaluate(() => {
|
||||
const state = window.__unideskExistingSessionRefresh;
|
||||
const streamFailure = state?.failures?.at(-1) || null;
|
||||
if (streamFailure) return { streamFailure };
|
||||
const connected = state?.connected?.at(-1) || null;
|
||||
return connected ? { connected } : null;
|
||||
}), effective.barrierTimeoutMs, "existing session refresh connected evidence");
|
||||
if (refreshState.streamFailure) throw realtimeExistingRefreshStreamError(refreshState.streamFailure);
|
||||
const connected = refreshState.connected;
|
||||
let connectedSummary = null;
|
||||
try {
|
||||
const connectedEvidence = realtimeExistingRefreshConnectedEvidence(connected, sessionId, effective);
|
||||
connectedSummary = connectedEvidence.connected;
|
||||
refreshReplay = connectedEvidence.refreshReplay;
|
||||
} catch (error) {
|
||||
connectedSummary = error?.details?.connected || null;
|
||||
refreshReplay = error?.details?.refreshReplay || null;
|
||||
if (connectedSummary) report.connected = connectedSummary;
|
||||
throw error;
|
||||
}
|
||||
phase = "live-handoff";
|
||||
await realtimeWaitForSameWorkbenchDom(before, effective.barrierTimeoutMs);
|
||||
await sleep(effective.reconnectQuietMs);
|
||||
const after = await realtimeWorkbenchDomIdentity();
|
||||
realtimeAssertSameWorkbenchDom(before, after);
|
||||
const browserEvidence = await page.evaluate(() => JSON.parse(JSON.stringify(window.__unideskExistingSessionRefresh || {})));
|
||||
if (browserEvidence.sources.length !== 1) throw new Error("main Workbench product EventSource count is " + browserEvidence.sources.length + ", expected 1");
|
||||
if (browserEvidence.sources[0]?.opened !== 1) throw new Error("main Workbench product EventSource open count is not 1");
|
||||
if (browserEvidence.connected.length !== 1) throw new Error("main Workbench connected event count is " + browserEvidence.connected.length + ", expected 1");
|
||||
if (browserEvidence.failures.length !== 0) throw realtimeExistingRefreshStreamError(browserEvidence.failures.at(-1));
|
||||
if (browserEvidence.errors.length !== 0) throw new Error("main Workbench product EventSource observed an error");
|
||||
realtimeAssertExistingRefreshDelivery(browserEvidence, refreshReplay);
|
||||
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 request count is " + mainStreams.length + ", expected 1");
|
||||
if (mainStreams[0].traceId !== null || mainStreams[0].afterSeq !== null || mainStreams[0].lastEventId !== null) throw new Error("main Workbench product SSE used a trace or cursor scope");
|
||||
if (forbidden.length > 0) throw new Error("forbidden recovery or polling requests observed: " + forbidden.map((item) => item.path).join(","));
|
||||
phase = "evidence-write";
|
||||
const screenshotFile = path.join(reportDir, "existing-session-refresh.png");
|
||||
await page.screenshot({ path: screenshotFile, fullPage: true, animations: "disabled", timeout: screenshotCaptureTimeoutMs });
|
||||
const screenshotMeta = await fileMeta(screenshotFile);
|
||||
screenshot = { kind: "existing-session-refresh-screenshot", path: screenshotFile, ...screenshotMeta, commandId: activeCommandId, valuesRedacted: true };
|
||||
await appendJsonl(files.artifacts, { ts: new Date().toISOString(), ...screenshot });
|
||||
const productSse = {
|
||||
requestCount: mainStreams.length,
|
||||
browserSourceCount: browserEvidence.sources.length,
|
||||
openCount: browserEvidence.sources[0]?.opened || 0,
|
||||
connectedEventCount: browserEvidence.connected.length,
|
||||
businessEventCount: browserEvidence.businessEventCount,
|
||||
typedFailureCount: browserEvidence.failures.length,
|
||||
livePhase: refreshReplay?.phase === "live",
|
||||
valuesRedacted: true,
|
||||
};
|
||||
Object.assign(report, {
|
||||
ok: true,
|
||||
status: "passed",
|
||||
phase: "live",
|
||||
code: "refresh_live_handoff_validated",
|
||||
completedAt: new Date().toISOString(),
|
||||
after,
|
||||
connected: connectedSummary,
|
||||
refreshReplay,
|
||||
productSse,
|
||||
forbiddenRequests: { count: 0, rows: [], valuesRedacted: true },
|
||||
screenshot,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
const artifacts = await realtimeWriteEvidenceArtifacts(reportDir, network, [], "existing-session-refresh");
|
||||
report.artifacts = artifacts;
|
||||
const reportArtifact = await realtimeWriteReport(reportDir, report, "existing-session-refresh-report");
|
||||
const observer = await syncObserverPageToControlSession("validateExistingSessionRefresh", sessionId);
|
||||
await appendJsonl(files.control, eventRecord("existing-session-refresh-validated", {
|
||||
profile: profileId,
|
||||
sessionId,
|
||||
traceIds: after.traceIds,
|
||||
phase: "live",
|
||||
code: "refresh_live_handoff_validated",
|
||||
reportPath: reportArtifact.path,
|
||||
reportSha256: reportArtifact.sha256,
|
||||
screenshotSha256: screenshot.sha256,
|
||||
valuesRedacted: true,
|
||||
}));
|
||||
return {
|
||||
ok: true,
|
||||
status: "passed",
|
||||
profile: profileId,
|
||||
sessionId,
|
||||
traceIds: after.traceIds,
|
||||
phase: "live",
|
||||
code: "refresh_live_handoff_validated",
|
||||
before: realtimeWorkbenchDomSummary(before),
|
||||
after: realtimeWorkbenchDomSummary(after),
|
||||
refreshReplay,
|
||||
productSse,
|
||||
forbiddenRequestCount: 0,
|
||||
reportPath: reportArtifact.path,
|
||||
reportSha256: reportArtifact.sha256,
|
||||
screenshot,
|
||||
observer,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
} catch (error) {
|
||||
const failure = realtimeExistingRefreshFailure(phase, error);
|
||||
if (!screenshot) {
|
||||
screenshot = await captureScreenshot("existing-session-refresh-failed-" + safeId(command.id))
|
||||
.then((artifact) => ({ path: artifact.path, sha256: artifact.sha256, valuesRedacted: true }))
|
||||
.catch((screenshotError) => ({ available: false, error: errorSummary(screenshotError), valuesRedacted: true }));
|
||||
}
|
||||
Object.assign(report, {
|
||||
ok: false,
|
||||
status: "failed",
|
||||
completedAt: new Date().toISOString(),
|
||||
phase: failure.phase,
|
||||
code: failure.code,
|
||||
refreshReplay,
|
||||
failure,
|
||||
screenshot,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
const artifacts = await realtimeWriteEvidenceArtifacts(reportDir, network, [], "existing-session-refresh").catch(() => null);
|
||||
if (artifacts) report.artifacts = artifacts;
|
||||
const reportArtifact = await realtimeWriteReport(reportDir, report, "existing-session-refresh-report").catch(() => null);
|
||||
const wrapped = error instanceof Error ? error : new Error(String(error));
|
||||
wrapped.details = {
|
||||
...(wrapped.details || {}),
|
||||
profile: profileId,
|
||||
sessionId,
|
||||
traceIds: report.after?.traceIds || report.before?.traceIds || [],
|
||||
phase: failure.phase,
|
||||
code: failure.code,
|
||||
failure,
|
||||
before: report.before ? realtimeWorkbenchDomSummary(report.before) : null,
|
||||
after: report.after ? realtimeWorkbenchDomSummary(report.after) : null,
|
||||
refreshReplay,
|
||||
screenshot,
|
||||
reportPath: reportArtifact?.path || null,
|
||||
reportSha256: reportArtifact?.sha256 || null,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
throw wrapped;
|
||||
} finally {
|
||||
if (requestPage) requestPage.off("request", requestListener);
|
||||
await page.evaluate(() => window.sessionStorage.removeItem("__unideskExistingSessionRefreshCapture")).catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
function realtimeExistingRefreshFailure(phase, error) {
|
||||
const codes = {
|
||||
"session-selection": "session_selection_failed",
|
||||
"refresh-navigation": "session_hydration_failed",
|
||||
"retention-barrier": "retention_barrier_failed",
|
||||
"live-handoff": "live_handoff_failed",
|
||||
"evidence-write": "evidence_write_failed",
|
||||
};
|
||||
const streamFailure = realtimeExistingRefreshStreamFailure(error?.details?.streamFailure);
|
||||
return {
|
||||
phase,
|
||||
code: codes[phase] || "existing_session_refresh_failed",
|
||||
streamPhase: streamFailure?.phase || null,
|
||||
streamCode: streamFailure?.code || null,
|
||||
streamFailure,
|
||||
error: errorSummary(error),
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
|
||||
function realtimeExistingRefreshConnectedEvidence(connected, sessionId, profile) {
|
||||
const connectedSummary = realtimeConnectedSummary(connected);
|
||||
const refreshReplay = connectedSummary.refreshReplay;
|
||||
try {
|
||||
realtimeAssertConnectedContract(connected, sessionId, "existing-session-refresh", profile, { requireRetainedEvents: true });
|
||||
} catch (error) {
|
||||
const wrapped = error instanceof Error ? error : new Error(String(error));
|
||||
wrapped.details = { ...(wrapped.details || {}), connected: connectedSummary, refreshReplay, valuesRedacted: true };
|
||||
throw wrapped;
|
||||
}
|
||||
return { connected: connectedSummary, refreshReplay, valuesRedacted: true };
|
||||
}
|
||||
|
||||
function realtimeAssertExistingRefreshDelivery(browserEvidence, refreshReplay) {
|
||||
const businessEventCount = Number(browserEvidence?.businessEventCount);
|
||||
const replayed = Number(refreshReplay?.counts?.replayed);
|
||||
const bufferedDelivered = Number(refreshReplay?.counts?.bufferedDelivered);
|
||||
if (![businessEventCount, replayed, bufferedDelivered].every((value) => Number.isInteger(value) && value >= 0)) {
|
||||
throw new Error("existing session refresh delivery counts are invalid");
|
||||
}
|
||||
const minimumDelivered = replayed + bufferedDelivered;
|
||||
if (businessEventCount < minimumDelivered) {
|
||||
throw new Error("existing session refresh browser observed " + businessEventCount + " business events, fewer than replayed plus buffered delivery " + minimumDelivered);
|
||||
}
|
||||
}
|
||||
|
||||
function realtimeExistingRefreshStreamError(value) {
|
||||
const streamFailure = realtimeExistingRefreshStreamFailure(value);
|
||||
const error = new Error(streamFailure?.message || "existing session refresh stream reported a typed failure");
|
||||
error.details = { streamFailure, valuesRedacted: true };
|
||||
return error;
|
||||
}
|
||||
|
||||
function realtimeExistingRefreshStreamFailure(value) {
|
||||
const nested = value?.error && typeof value.error === "object" ? value.error : value;
|
||||
const phase = typeof value?.phase === "string" ? value.phase : typeof nested?.phase === "string" ? nested.phase : null;
|
||||
const code = typeof nested?.code === "string" ? nested.code : typeof value?.code === "string" ? value.code : null;
|
||||
const message = typeof nested?.message === "string" ? nested.message.slice(0, 240) : typeof value?.message === "string" ? value.message.slice(0, 240) : null;
|
||||
if (!phase && !code && !message) return null;
|
||||
return {
|
||||
phase,
|
||||
code,
|
||||
message,
|
||||
retryable: nested?.retryable === true,
|
||||
fallback: value?.fallback === true,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
|
||||
async function realtimeArmExistingSessionRefreshProbe(targetPage, profile) {
|
||||
await targetPage.addInitScript(() => {
|
||||
const markerKey = "__unideskExistingSessionRefreshCapture";
|
||||
const raw = window.sessionStorage.getItem(markerKey);
|
||||
if (!raw) return;
|
||||
window.sessionStorage.removeItem(markerKey);
|
||||
let config;
|
||||
try { config = JSON.parse(raw); } catch { return; }
|
||||
const NativeEventSource = window.EventSource;
|
||||
const state = { sources: [], connected: [], businessEventCount: 0, failures: [], errors: [] };
|
||||
window.__unideskExistingSessionRefresh = state;
|
||||
if (typeof NativeEventSource !== "function") {
|
||||
state.errors.push({ kind: "eventsource-unavailable" });
|
||||
return;
|
||||
}
|
||||
class ObservedEventSource extends NativeEventSource {
|
||||
constructor(url, options) {
|
||||
super(url, options);
|
||||
let pathname = "";
|
||||
try { pathname = new URL(String(url), window.location.href).pathname; } catch {}
|
||||
if (pathname !== config.productEventsPath) return;
|
||||
state.sources.push({ url: String(url), opened: 0 });
|
||||
const source = state.sources[state.sources.length - 1];
|
||||
this.addEventListener("open", () => { source.opened += 1; });
|
||||
this.addEventListener(config.productReadyEvent, (event) => {
|
||||
try { state.connected.push(JSON.parse(event.data)); } catch { state.errors.push({ kind: "connected-parse" }); }
|
||||
});
|
||||
this.addEventListener(config.businessEvent, () => { state.businessEventCount += 1; });
|
||||
this.addEventListener("workbench.error", (event) => {
|
||||
try { state.failures.push(JSON.parse(event.data)); } catch { state.errors.push({ kind: "workbench-error-parse" }); }
|
||||
});
|
||||
this.addEventListener("error", () => { state.errors.push({ kind: "eventsource-error" }); });
|
||||
}
|
||||
}
|
||||
Object.defineProperty(ObservedEventSource, "CONNECTING", { value: NativeEventSource.CONNECTING });
|
||||
Object.defineProperty(ObservedEventSource, "OPEN", { value: NativeEventSource.OPEN });
|
||||
Object.defineProperty(ObservedEventSource, "CLOSED", { value: NativeEventSource.CLOSED });
|
||||
window.EventSource = ObservedEventSource;
|
||||
});
|
||||
await targetPage.evaluate((config) => {
|
||||
window.sessionStorage.setItem("__unideskExistingSessionRefreshCapture", JSON.stringify(config));
|
||||
}, {
|
||||
productEventsPath: profile.productEventsPath,
|
||||
productReadyEvent: profile.productReadyEvent,
|
||||
businessEvent: profile.businessEvent,
|
||||
});
|
||||
}
|
||||
|
||||
async function realtimeWorkbenchDomIdentity() {
|
||||
return page.evaluate(() => {
|
||||
const cards = Array.from(document.querySelectorAll("#conversation-list > .message-card"));
|
||||
const rows = cards.map((element, index) => ({
|
||||
index,
|
||||
role: element.getAttribute("data-role") || null,
|
||||
messageId: element.getAttribute("data-message-id") || null,
|
||||
traceId: element.getAttribute("data-trace-id") || null,
|
||||
status: element.getAttribute("data-status") || null,
|
||||
}));
|
||||
const identity = (row) => [row.role || "", row.messageId || "", row.traceId || ""].join(":");
|
||||
const identities = rows.map(identity);
|
||||
return {
|
||||
messageCount: rows.length,
|
||||
identities,
|
||||
missingIdentityRows: rows.filter((row) => !row.role || (!row.messageId && !row.traceId)).map((row) => row.index),
|
||||
duplicateIdentities: [...new Set(identities.filter((value, index, all) => all.indexOf(value) !== index))],
|
||||
traceIds: [...new Set(rows.map((row) => row.traceId).filter(Boolean))],
|
||||
rows,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
function realtimeAssertSameWorkbenchDom(before, after) {
|
||||
if (before.missingIdentityRows.length > 0 || after.missingIdentityRows.length > 0) throw new Error("Workbench DOM contains rows without a stable message identity");
|
||||
if (before.duplicateIdentities.length > 0 || after.duplicateIdentities.length > 0) throw new Error("Workbench DOM contains duplicate message identities");
|
||||
if (before.messageCount !== after.messageCount) throw new Error("Workbench message count changed across refresh: " + before.messageCount + " -> " + after.messageCount);
|
||||
if (JSON.stringify(before.identities) !== JSON.stringify(after.identities)) throw new Error("Workbench message identity or order changed across refresh");
|
||||
}
|
||||
|
||||
async function realtimeWaitForSameWorkbenchDom(before, timeoutMs) {
|
||||
let latest = null;
|
||||
return realtimePoll(async () => {
|
||||
latest = await realtimeWorkbenchDomIdentity();
|
||||
try {
|
||||
realtimeAssertSameWorkbenchDom(before, latest);
|
||||
return latest;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}, timeoutMs, "existing session Workbench DOM identity restoration").catch((error) => {
|
||||
const wrapped = error instanceof Error ? error : new Error(String(error));
|
||||
wrapped.details = { ...(wrapped.details || {}), before: realtimeWorkbenchDomSummary(before), after: realtimeWorkbenchDomSummary(latest), valuesRedacted: true };
|
||||
throw wrapped;
|
||||
});
|
||||
}
|
||||
|
||||
function realtimeWorkbenchDomSummary(value) {
|
||||
return value ? {
|
||||
messageCount: value.messageCount,
|
||||
identities: Array.isArray(value.identities) ? value.identities.slice(0, 80) : [],
|
||||
missingIdentityRows: Array.isArray(value.missingIdentityRows) ? value.missingIdentityRows.slice(0, 20) : [],
|
||||
duplicateIdentities: Array.isArray(value.duplicateIdentities) ? value.duplicateIdentities.slice(0, 20) : [],
|
||||
traceIds: Array.isArray(value.traceIds) ? value.traceIds.slice(0, 20) : [],
|
||||
valuesRedacted: true,
|
||||
} : null;
|
||||
}
|
||||
|
||||
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");
|
||||
@@ -1041,7 +1426,7 @@ function realtimeDebugSnapshotSummary(snapshot) {
|
||||
};
|
||||
}
|
||||
|
||||
async function realtimeWriteEvidenceArtifacts(reportDir, network, debugSnapshots) {
|
||||
async function realtimeWriteEvidenceArtifacts(reportDir, network, debugSnapshots, artifactPrefix = "realtime-fanout") {
|
||||
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 });
|
||||
@@ -1059,16 +1444,16 @@ async function realtimeWriteEvidenceArtifacts(reportDir, network, debugSnapshots
|
||||
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 });
|
||||
await appendJsonl(files.artifacts, { ts: new Date().toISOString(), kind: artifactPrefix + "-request-ledger", commandId: activeCommandId, ...artifacts.requestLedger });
|
||||
await appendJsonl(files.artifacts, { ts: new Date().toISOString(), kind: artifactPrefix + "-events", commandId: activeCommandId, ...artifacts.events });
|
||||
return artifacts;
|
||||
}
|
||||
|
||||
async function realtimeWriteReport(reportDir, report) {
|
||||
async function realtimeWriteReport(reportDir, report, artifactKind = "realtime-fanout-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 };
|
||||
const artifact = { kind: artifactKind, path: reportFile, byteCount: meta.byteCount, sha256: meta.sha256, commandId: activeCommandId, valuesRedacted: true };
|
||||
await appendJsonl(files.artifacts, { ts: new Date().toISOString(), ...artifact });
|
||||
return artifact;
|
||||
}
|
||||
|
||||
@@ -141,6 +141,7 @@ export type NodeWebProbeObserveCommandType =
|
||||
| "newSession"
|
||||
| "sendPrompt"
|
||||
| "validateRealtimeFanout"
|
||||
| "validateExistingSessionRefresh"
|
||||
| "validateWorkbenchKafkaDebugReplay"
|
||||
| "validateWorkbenchTraceReadability"
|
||||
| "steer"
|
||||
|
||||
@@ -66,6 +66,98 @@ test("observe status renderer exposes the exact command phase and failed report
|
||||
assert.match(text, /terminal event arrived before subscriber disconnect/u);
|
||||
});
|
||||
|
||||
test("observe status renderer exposes existing-session refresh handoff and typed failure evidence", () => {
|
||||
const base = {
|
||||
ok: true,
|
||||
status: "running",
|
||||
command: "web-probe observe status",
|
||||
id: "webobs-fixture",
|
||||
node: "NC01",
|
||||
lane: "v03",
|
||||
next: {},
|
||||
};
|
||||
const rendered = withWebObserveStatusRendered({
|
||||
...base,
|
||||
observer: {
|
||||
processAlive: true,
|
||||
manifest: {},
|
||||
heartbeat: {},
|
||||
diagnostics: {},
|
||||
commands: {},
|
||||
exactCommand: {
|
||||
commandId: "cmd-existing-refresh",
|
||||
type: "validateExistingSessionRefresh",
|
||||
phase: "done",
|
||||
ok: true,
|
||||
result: {
|
||||
status: "passed",
|
||||
sessionId: "ses_existing",
|
||||
traceIds: ["trc_existing"],
|
||||
phase: "live",
|
||||
code: "refresh_live_handoff_validated",
|
||||
before: { messageCount: 2 },
|
||||
after: { messageCount: 2 },
|
||||
refreshReplay: {
|
||||
phase: "live",
|
||||
topic: "hwlab.event.v1",
|
||||
topicPartitions: [0],
|
||||
barrier: [{ partition: 0, endOffset: "18" }],
|
||||
counts: { matched: 8, replayed: 8, buffered: 1, bufferedDelivered: 1, liveDelivered: 0, deduplicated: 0 },
|
||||
},
|
||||
productSse: { browserSourceCount: 1, openCount: 1, connectedEventCount: 1, businessEventCount: 9, typedFailureCount: 0, livePhase: true },
|
||||
forbiddenRequestCount: 0,
|
||||
reportSha256: "sha256:existing-refresh-report",
|
||||
screenshot: { sha256: "sha256:existing-refresh-image" },
|
||||
},
|
||||
},
|
||||
tails: { samples: [], control: [], network: [] },
|
||||
},
|
||||
});
|
||||
const text = String(rendered.renderedText);
|
||||
assert.match(text, /Workbench 既有 Session 刷新：/u);
|
||||
assert.match(text, /ses_existing.*trc_existing.*live.*refresh_live_handoff_validated.*2.*2.*1.*1.*1.*9.*0.*0/u);
|
||||
assert.match(text, /hwlab\.event\.v1.*0.*0:18.*8.*8.*1.*1.*0.*0.*true/u);
|
||||
assert.match(text, /sha256:existing-refresh-report.*sha256:existing-refresh-image/u);
|
||||
|
||||
const failed = withWebObserveStatusRendered({
|
||||
...base,
|
||||
observer: {
|
||||
processAlive: true,
|
||||
manifest: {},
|
||||
heartbeat: {},
|
||||
diagnostics: {},
|
||||
commands: {},
|
||||
exactCommand: {
|
||||
commandId: "cmd-existing-refresh-failed",
|
||||
type: "validateExistingSessionRefresh",
|
||||
phase: "failed",
|
||||
ok: false,
|
||||
error: {
|
||||
message: "Kafka retention barrier gap",
|
||||
details: {
|
||||
sessionId: "ses_existing",
|
||||
traceIds: ["trc_existing"],
|
||||
phase: "retention-barrier",
|
||||
code: "retention_barrier_failed",
|
||||
failure: {
|
||||
phase: "retention-barrier",
|
||||
code: "retention_barrier_failed",
|
||||
streamPhase: "flushing",
|
||||
streamCode: "workbench_kafka_refresh_barrier_gap",
|
||||
message: "Kafka retention barrier gap",
|
||||
},
|
||||
reportSha256: "sha256:failed-existing-refresh-report",
|
||||
},
|
||||
},
|
||||
},
|
||||
tails: { samples: [], control: [], network: [] },
|
||||
},
|
||||
});
|
||||
const failedText = String(failed.renderedText);
|
||||
assert.match(failedText, /Typed failure:/u);
|
||||
assert.match(failedText, /retention-barrier.*retention_barrier_failed.*flushing.*workbench_kafka_refresh_barrier_gap.*Kafka retention barrier gap/u);
|
||||
});
|
||||
|
||||
test("observe status renderer exposes bounded Workbench Kafka debug replay evidence", () => {
|
||||
const rendered = withWebObserveStatusRendered({
|
||||
ok: true,
|
||||
|
||||
@@ -58,6 +58,91 @@ test("observe status returns one exact completed command without prompt payloads
|
||||
assert.equal(status.exactCommand.valuesRedacted, true);
|
||||
});
|
||||
|
||||
test("observe status preserves bounded existing-session refresh evidence", async () => {
|
||||
const stateDir = await mkdtemp(join(tmpdir(), "unidesk-web-observe-existing-refresh-status-"));
|
||||
const commandId = "cmd-existing-refresh";
|
||||
await mkdir(join(stateDir, "commands", "done"), { recursive: true });
|
||||
await writeFile(join(stateDir, "commands", "done", `${commandId}.json`), JSON.stringify({
|
||||
ok: true,
|
||||
commandId,
|
||||
type: "validateExistingSessionRefresh",
|
||||
completedAt: "2026-07-11T12:00:00.000Z",
|
||||
result: {
|
||||
ok: true,
|
||||
status: "passed",
|
||||
profile: "pure-kafka-live",
|
||||
sessionId: "ses_existing",
|
||||
traceIds: ["trc_existing"],
|
||||
phase: "live",
|
||||
code: "refresh_live_handoff_validated",
|
||||
before: { messageCount: 2, identities: ["must-not-project"], missingIdentityRows: [], duplicateIdentities: [], traceIds: ["trc_existing"] },
|
||||
after: { messageCount: 2, identities: ["must-not-project"], missingIdentityRows: [], duplicateIdentities: [], traceIds: ["trc_existing"] },
|
||||
refreshReplay: {
|
||||
phase: "live",
|
||||
topic: "hwlab.event.v1",
|
||||
topicPartitions: [0],
|
||||
barrier: [{ partition: 0, endOffset: "18" }],
|
||||
counts: { matched: 8, replayed: 8, buffered: 1, bufferedDelivered: 1, liveDelivered: 0, deduplicated: 0 },
|
||||
},
|
||||
productSse: { requestCount: 1, browserSourceCount: 1, openCount: 1, connectedEventCount: 1, businessEventCount: 9, typedFailureCount: 0, livePhase: true },
|
||||
forbiddenRequestCount: 0,
|
||||
reportPath: "/tmp/existing-refresh-report.json",
|
||||
reportSha256: "sha256:existing-refresh-report",
|
||||
screenshot: { path: "/tmp/existing-refresh.png", sha256: "sha256:existing-refresh-image" },
|
||||
},
|
||||
}) + "\n");
|
||||
|
||||
const status = await runStatusScript(stateDir, commandId);
|
||||
const result = status.exactCommand.result;
|
||||
assert.equal(result.phase, "live");
|
||||
assert.equal(result.code, "refresh_live_handoff_validated");
|
||||
assert.equal(result.before.messageCount, 2);
|
||||
assert.equal(result.before.identities, undefined);
|
||||
assert.deepEqual(result.refreshReplay.barrier, [{ partition: 0, endOffset: "18" }]);
|
||||
assert.equal(result.refreshReplay.counts.replayed, 8);
|
||||
assert.equal(result.productSse.typedFailureCount, 0);
|
||||
assert.equal(result.reportSha256, "sha256:existing-refresh-report");
|
||||
assert.equal(result.screenshot.sha256, "sha256:existing-refresh-image");
|
||||
});
|
||||
|
||||
test("observe status preserves typed existing-session refresh stream failures", async () => {
|
||||
const stateDir = await mkdtemp(join(tmpdir(), "unidesk-web-observe-existing-refresh-failure-"));
|
||||
const commandId = "cmd-existing-refresh-failed";
|
||||
await mkdir(join(stateDir, "commands", "failed"), { recursive: true });
|
||||
await writeFile(join(stateDir, "commands", "failed", `${commandId}.json`), JSON.stringify({
|
||||
ok: false,
|
||||
commandId,
|
||||
type: "validateExistingSessionRefresh",
|
||||
failedAt: "2026-07-11T12:00:00.000Z",
|
||||
error: {
|
||||
message: "Kafka retention barrier gap",
|
||||
details: {
|
||||
sessionId: "ses_existing",
|
||||
traceIds: ["trc_existing"],
|
||||
phase: "retention-barrier",
|
||||
code: "retention_barrier_failed",
|
||||
failure: {
|
||||
phase: "retention-barrier",
|
||||
code: "retention_barrier_failed",
|
||||
streamPhase: "flushing",
|
||||
streamCode: "workbench_kafka_refresh_barrier_gap",
|
||||
streamFailure: { phase: "flushing", code: "workbench_kafka_refresh_barrier_gap", message: "Kafka retention barrier gap" },
|
||||
},
|
||||
reportSha256: "sha256:failed-existing-refresh-report",
|
||||
},
|
||||
},
|
||||
}) + "\n");
|
||||
|
||||
const status = await runStatusScript(stateDir, commandId);
|
||||
const details = status.exactCommand.error.details;
|
||||
assert.equal(details.phase, "retention-barrier");
|
||||
assert.equal(details.code, "retention_barrier_failed");
|
||||
assert.equal(details.failure.streamPhase, "flushing");
|
||||
assert.equal(details.failure.streamCode, "workbench_kafka_refresh_barrier_gap");
|
||||
assert.equal(details.failure.message, "Kafka retention barrier gap");
|
||||
assert.deepEqual(details.traceIds, ["trc_existing"]);
|
||||
});
|
||||
|
||||
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");
|
||||
|
||||
@@ -79,6 +79,23 @@ export function nodeWebObserveStatusNodeScript(tailLines: number, node: string,
|
||||
rawHwlabEventWindow:raw?{available:raw.available??null,contract:raw.contract||null,reason:raw.reason||null,samePage:raw.samePage??null,productEventsPath:raw.productEventsPath||null,productEventSourceRequestCount:raw.productEventSourceRequestCount??null,sourceContractPresent:raw.sourceContractPresent??null,unfilteredContractPresent:raw.unfilteredContractPresent??null,textBytes:raw.textBytes??null,textHash:raw.textHash||null,textPresent:raw.textPresent??null,counts:rawCounts?{received:rawCounts.received??null,retained:rawCounts.retained??null,decoded:rawCounts.decoded??null,rejected:rawCounts.rejected??null,evicted:rawCounts.evicted??null,bytes:rawCounts.bytes??null}:null,valuesRedacted:true}:null
|
||||
};
|
||||
};
|
||||
const compactExistingSessionRefreshEvidence=(value)=>{
|
||||
const item=value&&typeof value==='object'?value:null;if(!item)return{};
|
||||
const replay=item.refreshReplay&&typeof item.refreshReplay==='object'?item.refreshReplay:null;
|
||||
const counts=replay&&replay.counts&&typeof replay.counts==='object'?replay.counts:null;
|
||||
const product=item.productSse&&typeof item.productSse==='object'?item.productSse:null;
|
||||
const before=item.before&&typeof item.before==='object'?item.before:null;
|
||||
const after=item.after&&typeof item.after==='object'?item.after:null;
|
||||
const failure=item.failure&&typeof item.failure==='object'?item.failure:null;
|
||||
const streamFailure=failure&&failure.streamFailure&&typeof failure.streamFailure==='object'?failure.streamFailure:item.streamFailure&&typeof item.streamFailure==='object'?item.streamFailure:null;
|
||||
const compactDom=(dom)=>dom?{messageCount:dom.messageCount??null,traceIds:Array.isArray(dom.traceIds)?dom.traceIds.slice(0,4):[],missingIdentityCount:Array.isArray(dom.missingIdentityRows)?dom.missingIdentityRows.length:null,duplicateIdentityCount:Array.isArray(dom.duplicateIdentities)?dom.duplicateIdentities.length:null,valuesRedacted:true}:null;
|
||||
return{
|
||||
before:compactDom(before),after:compactDom(after),
|
||||
refreshReplay:replay?{phase:replay.phase||null,topic:replay.topic||null,topicPartitions:Array.isArray(replay.topicPartitions)?replay.topicPartitions.slice(0,8):[],barrier:Array.isArray(replay.barrier)?replay.barrier.slice(0,8).map((entry)=>({partition:entry?.partition??null,endOffset:entry?.endOffset||null})):[],counts:counts?{matched:counts.matched??null,replayed:counts.replayed??null,buffered:counts.buffered??null,bufferedDelivered:counts.bufferedDelivered??null,liveDelivered:counts.liveDelivered??null,deduplicated:counts.deduplicated??null}:null,valuesRedacted:true}:null,
|
||||
productSse:product?{requestCount:product.requestCount??null,browserSourceCount:product.browserSourceCount??null,openCount:product.openCount??null,connectedEventCount:product.connectedEventCount??null,businessEventCount:product.businessEventCount??null,typedFailureCount:product.typedFailureCount??null,livePhase:product.livePhase??null,valuesRedacted:true}:null,
|
||||
failure:failure||streamFailure?{phase:failure&&failure.phase||item.phase||null,code:failure&&failure.code||item.code||null,streamPhase:failure&&failure.streamPhase||streamFailure&&streamFailure.phase||null,streamCode:failure&&failure.streamCode||streamFailure&&streamFailure.code||null,message:short(streamFailure&&streamFailure.message||failure&&failure.error&&failure.error.message||'',200),valuesRedacted:true}:null
|
||||
};
|
||||
};
|
||||
const compactTraceSnapshot=(value)=>{const item=value&&typeof value==='object'?value:null;if(!item)return null;const disclosure=item.disclosure&&typeof item.disclosure==='object'?item.disclosure:null;return{status:item.status||null,traceId:item.traceId||null,disclosure:disclosure?{available:disclosure.available??null,open:disclosure.open??null,eventCount:disclosure.eventCount??null,readableRowCount:disclosure.readableRowCount??null,renderedRowCount:disclosure.renderedRowCount??null,rowWindowed:disclosure.rowWindowed??null}:null,rowCount:item.rowCount??null,readableRowCount:item.readableRowCount??null,sourceAuthorityRowCount:item.sourceAuthorityRowCount??null,readableSourceRowCount:item.readableSourceRowCount??null,projectedAuthorityRowCount:item.projectedAuthorityRowCount??null,rowSetHash:item.rowSetHash||null,valuesRedacted:true};};
|
||||
const compactTracePhase=(value)=>{const item=value&&typeof value==='object'?value:null;return item?{observed:item.observed??null,stable:item.stable??null,reason:item.reason||null,status:item.status||null,elapsedMs:item.elapsedMs??null,sampleCount:item.sampleCount??null,snapshot:compactTraceSnapshot(item.snapshot),valuesRedacted:true}:null;};
|
||||
const compactTraceReadabilityEvidence=(value)=>{const item=value&&typeof value==='object'?value:null;if(!item)return{};const scope=item.scope&&typeof item.scope==='object'?item.scope:null;const disclosure=item.disclosure&&typeof item.disclosure==='object'?item.disclosure:null;const retention=item.runningRetention&&typeof item.runningRetention==='object'?item.runningRetention:null;return{startedDuringRunning:item.startedDuringRunning??null,forcedExpansion:item.forcedExpansion??null,scope:scope?{selector:short(scope.selector),conversationCount:scope.conversationCount??null,productCardCount:scope.productCardCount??null,debugPanelCountInsideConversation:scope.debugPanelCountInsideConversation??null,isolatedDebugExcluded:scope.isolatedDebugExcluded??null,valuesRedacted:true}:null,disclosure:disclosure?{openBefore:disclosure.openBefore??null,openAfterExpansion:disclosure.openAfterExpansion??null,openAfterTerminal:disclosure.openAfterTerminal??null,autoReadable:disclosure.autoReadable??null,valuesRedacted:true}:null,running:compactTracePhase(item.running),terminalArrival:compactTracePhase(item.terminalArrival),terminal:compactTracePhase(item.terminal),runningRetention:retention?{mode:retention.mode||null,passed:retention.passed??null,retainedRowCount:retention.retainedRowCount??null,runningEventCount:retention.runningEventCount??null,terminalEventCount:retention.terminalEventCount??null,runningSourceSeqMin:retention.runningSourceSeqMin??null,runningSourceSeqMax:retention.runningSourceSeqMax??null,terminalSourceSeqMin:retention.terminalSourceSeqMin??null,terminalSourceSeqMax:retention.terminalSourceSeqMax??null,valuesRedacted:true}:null,retainedRunningRowCount:item.retainedRunningRowCount??null,failures:Array.isArray(item.failures)?item.failures.slice(0,8).map((failure)=>({code:failure&&failure.code||null,message:short(failure&&failure.message),valuesRedacted:true})):null};};
|
||||
@@ -97,7 +114,7 @@ export function nodeWebObserveStatusNodeScript(tailLines: number, node: string,
|
||||
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,traceId:result.traceId||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):[],topic:result.topic||null,groupId:result.groupId||null,groupPrefix:result.groupPrefix||null,receivedCount:result.receivedCount??null,appliedCount:result.appliedCount??null,terminalStatus:result.terminalStatus||null,...compactReplayEvidence(result),...compactTraceReadabilityEvidence(result),warmRunnerReused:result.warmRunnerReused??null,forbiddenRequestCount:result.forbiddenRequestCount??null,replayedEventCount:result.replayedEventCount??null,reportPath:result.reportPath||null,reportSha256:result.reportSha256||null,screenshot:compactScreenshot(result.screenshot),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,...compactReplayEvidence(error.details),...compactTraceReadabilityEvidence(error.details),sessionId:error.details.sessionId||null,traceId:error.details.traceId||null,topic:error.details.topic||null,groupId:error.details.groupId||null,receivedCount:error.details.receivedCount??null,appliedCount:error.details.appliedCount??null,terminalStatus:error.details.terminalStatus||null,reportPath:error.details.reportPath||null,reportSha256:error.details.reportSha256||null,screenshot:compactScreenshot(error.details.screenshot),valuesRedacted:true}:null}:null,valuesRedacted:true};
|
||||
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,traceId:result.traceId||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):[],topic:result.topic||null,groupId:result.groupId||null,groupPrefix:result.groupPrefix||null,receivedCount:result.receivedCount??null,appliedCount:result.appliedCount??null,terminalStatus:result.terminalStatus||null,...compactReplayEvidence(result),...(parsed.type==='validateExistingSessionRefresh'?compactExistingSessionRefreshEvidence(result):{}),...compactTraceReadabilityEvidence(result),warmRunnerReused:result.warmRunnerReused??null,forbiddenRequestCount:result.forbiddenRequestCount??null,replayedEventCount:result.replayedEventCount??null,reportPath:result.reportPath||null,reportSha256:result.reportSha256||null,screenshot:compactScreenshot(result.screenshot),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,...compactReplayEvidence(error.details),...(parsed.type==='validateExistingSessionRefresh'?compactExistingSessionRefreshEvidence(error.details):{}),...compactTraceReadabilityEvidence(error.details),sessionId:error.details.sessionId||null,traceId:error.details.traceId||null,traceIds:Array.isArray(error.details.traceIds)?error.details.traceIds.slice(0,4):[],topic:error.details.topic||null,groupId:error.details.groupId||null,receivedCount:error.details.receivedCount??null,appliedCount:error.details.appliedCount??null,terminalStatus:error.details.terminalStatus||null,forbiddenRequestCount:error.details.forbiddenRequestCount??null,reportPath:error.details.reportPath||null,reportSha256:error.details.reportSha256||null,screenshot:compactScreenshot(error.details.screenshot),valuesRedacted:true}:null}:null,valuesRedacted:true};
|
||||
}
|
||||
return {commandId:exactCommandId,phase:'not-found',ok:false,valuesRedacted:true};
|
||||
};
|
||||
@@ -396,7 +413,7 @@ walk(dir);
|
||||
const selectedFiles=out.slice(0,maxFiles);
|
||||
const files=selectedFiles.slice(0,24).map(file=>{const st=fs.statSync(file); return {relative:path.relative(dir,file),byteCount:st.size,sha256:shaFile(file)}});
|
||||
const totalBytes=selectedFiles.reduce((sum,file)=>sum+fs.statSync(file).size,0);
|
||||
console.log(JSON.stringify({ok:true,command:'web-probe-observe collect',stateDir:dir,fileCount:selectedFiles.length,listedFileCount:files.length,omittedFileCount:Math.max(0,selectedFiles.length-files.length),totalBytes,files,valuesRedacted:true}));
|
||||
console.log(JSON.stringify({ok:true,command:'web-probe-observe collect',stateDir:dir,mode:'list',view:'files',fileCount:selectedFiles.length,listedFileCount:files.length,omittedFileCount:Math.max(0,selectedFiles.length-files.length),totalBytes,files,valuesRedacted:true}));
|
||||
`)} "$state_dir" ${shellQuote(collectFile ?? "")} ${shellQuote(collectFinding ?? "")} ${shellQuote(collectGrep ?? "")}`;
|
||||
}
|
||||
|
||||
|
||||
@@ -753,6 +753,12 @@ export function runNodeWebProbeObserveCommand(options: NodeWebProbeObserveOption
|
||||
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");
|
||||
}
|
||||
if (type === "validateExistingSessionRefresh") {
|
||||
if (!options.commandProfile?.trim()) throw new Error("validateExistingSessionRefresh requires --profile");
|
||||
if (spec.webProbe?.realtimeFanoutProfiles?.[options.commandProfile.trim()] === undefined) {
|
||||
throw new Error(`validateExistingSessionRefresh profile is not declared by the owning YAML: ${options.commandProfile.trim()}`);
|
||||
}
|
||||
}
|
||||
if (type === "validateWorkbenchKafkaDebugReplay" && spec.webProbe?.workbenchKafkaDebugReplay?.enabled !== true) {
|
||||
throw new Error("validateWorkbenchKafkaDebugReplay is not enabled by the owning YAML");
|
||||
}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { mkdirSync, writeFileSync } from "node:fs";
|
||||
import { mkdirSync, mkdtempSync, writeFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { tmpdir } from "node:os";
|
||||
import { describe, expect, test } from "bun:test";
|
||||
@@ -6,6 +6,7 @@ import type { CommandResult } from "../command";
|
||||
import type { NodeWebProbeObserveOptions } from "./entry";
|
||||
import { buildNodeWebProbeObserveCollectPayload } from "./web-probe-observe-collect";
|
||||
import { withWebObserveCollectRendered } from "../hwlab-node-web-observe-render";
|
||||
import { nodeWebObserveCollectNodeScript } from "./web-observe-scripts";
|
||||
|
||||
describe("web-probe observe collect child JSON recovery", () => {
|
||||
test("recovers collect JSON from trans stdout dump", () => {
|
||||
@@ -171,6 +172,28 @@ test("accepts the standard file collect envelope without requiring a summary vie
|
||||
expect(collect.file.byteCount).toBe(21885);
|
||||
});
|
||||
|
||||
test("accepts the structured files listing envelope when --file is absent", () => {
|
||||
const options = collectOptions();
|
||||
options.collectView = "files";
|
||||
const stateDir = mkdtempSync(join(tmpdir(), "unidesk-web-observe-files-list-"));
|
||||
writeFileSync(join(stateDir, "manifest.json"), "{}\n");
|
||||
const child = Bun.spawnSync(["bash", "-lc", nodeWebObserveCollectNodeScript(options.maxFiles, null, null, null)], {
|
||||
env: { ...process.env, state_dir: stateDir },
|
||||
stdout: "pipe",
|
||||
stderr: "pipe",
|
||||
});
|
||||
expect(child.exitCode).toBe(0);
|
||||
const payload = buildNodeWebProbeObserveCollectPayload(options, { workspace: "/workspace" }, {
|
||||
command: ["trans", "NC01:/workspace", "sh"], cwd: "/repo", exitCode: 0,
|
||||
stdout: Buffer.from(child.stdout).toString("utf8"),
|
||||
stderr: Buffer.from(child.stderr).toString("utf8"), signal: null, timedOut: false,
|
||||
} satisfies CommandResult);
|
||||
expect(payload.ok).toBe(true);
|
||||
expect(payload.status).toBe("collected");
|
||||
expect((payload.collect as Record<string, unknown>).mode).toBe("list");
|
||||
expect((payload.collect as Record<string, unknown>).view).toBe("files");
|
||||
});
|
||||
|
||||
function collectOptions(): NodeWebProbeObserveOptions {
|
||||
return {
|
||||
action: "observe",
|
||||
|
||||
@@ -99,7 +99,7 @@ export function isWebObserveCollectJsonContract(value: Record<string, unknown>,
|
||||
|| stringOrNull(value.error) !== null;
|
||||
}
|
||||
return command === "web-probe-observe collect"
|
||||
&& (view !== null || mode === "file")
|
||||
&& (view !== null || mode === "file" || mode === "list")
|
||||
&& stateDir !== null;
|
||||
}
|
||||
|
||||
|
||||
@@ -69,6 +69,17 @@ test("validateRealtimeFanout rejects profiles absent from the owning YAML", () =
|
||||
);
|
||||
});
|
||||
|
||||
test("validateExistingSessionRefresh parses an existing session without prompts", () => {
|
||||
const options = parseNodeWebProbeObserveOptions(
|
||||
["command", "--type", "validateExistingSessionRefresh", "--profile", "pure-kafka-live", "--session-id", "ses_existing"],
|
||||
"NC01", "v03", spec, "webobs-fixture", null,
|
||||
);
|
||||
assert.equal(options.commandType, "validateExistingSessionRefresh");
|
||||
assert.equal(options.commandProfile, "pure-kafka-live");
|
||||
assert.equal(options.commandSessionId, "ses_existing");
|
||||
assert.equal(options.commandText, null);
|
||||
});
|
||||
|
||||
test("validateWorkbenchKafkaDebugReplay is a YAML-enabled argument-free typed command", () => {
|
||||
const options = parseNodeWebProbeObserveOptions(
|
||||
["command", "--type", "validateWorkbenchKafkaDebugReplay"],
|
||||
|
||||
@@ -619,6 +619,12 @@ export function parseNodeWebProbeObserveOptions(
|
||||
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");
|
||||
}
|
||||
if (observeActionRaw === "command" && commandType === "validateExistingSessionRefresh") {
|
||||
if (!commandProfile?.trim()) throw new Error("validateExistingSessionRefresh requires --profile");
|
||||
if (spec.webProbe?.realtimeFanoutProfiles?.[commandProfile.trim()] === undefined) {
|
||||
throw new Error(`validateExistingSessionRefresh profile is not declared by the owning YAML: ${commandProfile.trim()}`);
|
||||
}
|
||||
}
|
||||
if (observeActionRaw === "command" && commandType === "validateWorkbenchKafkaDebugReplay") {
|
||||
if (spec.webProbe?.workbenchKafkaDebugReplay?.enabled !== true) {
|
||||
throw new Error("validateWorkbenchKafkaDebugReplay is not enabled by the owning YAML");
|
||||
@@ -724,6 +730,7 @@ export function parseNodeWebProbeObserveCommandType(value: string): NodeWebProbe
|
||||
|| value === "newSession"
|
||||
|| value === "sendPrompt"
|
||||
|| value === "validateRealtimeFanout"
|
||||
|| value === "validateExistingSessionRefresh"
|
||||
|| value === "validateWorkbenchKafkaDebugReplay"
|
||||
|| value === "validateWorkbenchTraceReadability"
|
||||
|| value === "steer"
|
||||
@@ -761,7 +768,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, validateRealtimeFanout, validateWorkbenchKafkaDebugReplay, validateWorkbenchTraceReadability, 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, validateExistingSessionRefresh, validateWorkbenchKafkaDebugReplay, validateWorkbenchTraceReadability, 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 {
|
||||
|
||||
Reference in New Issue
Block a user