diff --git a/scripts/src/hwlab-node-web-observe-render.ts b/scripts/src/hwlab-node-web-observe-render.ts index 31f8b7fc..8b825f8e 100644 --- a/scripts/src/hwlab-node-web-observe-render.ts +++ b/scripts/src/hwlab-node-web-observe-render.ts @@ -176,6 +176,14 @@ function renderWebObserveStatusTable(value: Record): 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 { return lines.join("\n"); } +function renderExistingSessionRefreshEvidence( + evidence: Record, + 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, reportSha: unknown, @@ -391,6 +462,14 @@ function renderWebObserveCommandTable(value: Record): 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( diff --git a/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts b/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts index 85e3ce9d..f1082eff 100644 --- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts +++ b/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts @@ -187,15 +187,73 @@ test("existing session refresh rejects duplicate or reordered Workbench identiti 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;`) as () => ( - phase: string, error: Error, - ) => Record; - const failure = build()("retention-barrier", new Error("connected evidence timed out")); + const build = new Function(`function errorSummary(error) { return { message: error.message }; }\n${source}\nreturn { realtimeExistingRefreshFailure, realtimeExistingRefreshStreamError };`) as () => { + realtimeExistingRefreshFailure: (phase: string, error: Error & { details?: Record }) => Record; + realtimeExistingRefreshStreamError: (value: Record) => Error & { details?: Record }; + }; + 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).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: (value: T) => T) => { + realtimeExistingRefreshConnectedEvidence: (connected: Record, sessionId: string, profile: Record) => Record; + realtimeAssertExistingRefreshDelivery: (browser: Record, refreshReplay: Record) => 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 }) => { + 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) => void; diff --git a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts b/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts index fb635630..ffd82a98 100644 --- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts +++ b/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts @@ -390,9 +390,26 @@ async function validateExistingSessionRefresh(command) { 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 connected = await realtimePoll(async () => page.evaluate(() => window.__unideskExistingSessionRefresh?.connected?.at(-1) || null), effective.barrierTimeoutMs, "existing session refresh connected evidence"); - realtimeAssertConnectedContract(connected, sessionId, "existing-session-refresh", effective, { requireRetainedEvents: true }); - refreshReplay = realtimeConnectedSummary(connected).refreshReplay; + 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); @@ -402,7 +419,9 @@ async function validateExistingSessionRefresh(command) { 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"); @@ -420,6 +439,7 @@ async function validateExistingSessionRefresh(command) { openCount: browserEvidence.sources[0]?.opened || 0, connectedEventCount: browserEvidence.connected.length, businessEventCount: browserEvidence.businessEventCount, + typedFailureCount: browserEvidence.failures.length, livePhase: refreshReplay?.phase === "live", valuesRedacted: true, }; @@ -430,7 +450,7 @@ async function validateExistingSessionRefresh(command) { code: "refresh_live_handoff_validated", completedAt: new Date().toISOString(), after, - connected: realtimeConnectedSummary(connected), + connected: connectedSummary, refreshReplay, productSse, forbiddenRequests: { count: 0, rows: [], valuesRedacted: true }, @@ -500,6 +520,7 @@ async function validateExistingSessionRefresh(command) { 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, @@ -523,14 +544,67 @@ function realtimeExistingRefreshFailure(phase, error) { "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"; @@ -540,7 +614,7 @@ async function realtimeArmExistingSessionRefreshProbe(targetPage, profile) { let config; try { config = JSON.parse(raw); } catch { return; } const NativeEventSource = window.EventSource; - const state = { sources: [], connected: [], businessEventCount: 0, errors: [] }; + const state = { sources: [], connected: [], businessEventCount: 0, failures: [], errors: [] }; window.__unideskExistingSessionRefresh = state; if (typeof NativeEventSource !== "function") { state.errors.push({ kind: "eventsource-unavailable" }); @@ -559,6 +633,9 @@ async function realtimeArmExistingSessionRefreshProbe(targetPage, profile) { 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" }); }); } } diff --git a/scripts/src/hwlab-node/web-observe-render.test.ts b/scripts/src/hwlab-node/web-observe-render.test.ts index b69e2077..ed3d35ad 100644 --- a/scripts/src/hwlab-node/web-observe-render.test.ts +++ b/scripts/src/hwlab-node/web-observe-render.test.ts @@ -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, diff --git a/scripts/src/hwlab-node/web-observe-scripts.test.ts b/scripts/src/hwlab-node/web-observe-scripts.test.ts index 9c501864..78e69a86 100644 --- a/scripts/src/hwlab-node/web-observe-scripts.test.ts +++ b/scripts/src/hwlab-node/web-observe-scripts.test.ts @@ -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"); diff --git a/scripts/src/hwlab-node/web-observe-scripts.ts b/scripts/src/hwlab-node/web-observe-scripts.ts index 6569d2e9..961b6129 100644 --- a/scripts/src/hwlab-node/web-observe-scripts.ts +++ b/scripts/src/hwlab-node/web-observe-scripts.ts @@ -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 ?? "")}`; } diff --git a/scripts/src/hwlab-node/web-probe-observe-collect.test.ts b/scripts/src/hwlab-node/web-probe-observe-collect.test.ts index 17bdce2d..d5d85de3 100644 --- a/scripts/src/hwlab-node/web-probe-observe-collect.test.ts +++ b/scripts/src/hwlab-node/web-probe-observe-collect.test.ts @@ -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", () => { @@ -174,14 +175,23 @@ test("accepts the standard file collect envelope without requiring a summary vie 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: JSON.stringify({ ok: true, command: "web-probe-observe collect", stateDir: "/remote/state", mode: "list", files: [], valuesRedacted: true }), - stderr: "", signal: null, timedOut: false, + 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).mode).toBe("list"); + expect((payload.collect as Record).view).toBe("files"); }); function collectOptions(): NodeWebProbeObserveOptions {