fix: 完善既有会话刷新证据合同
This commit is contained in:
Codex
@@ -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(
|
||||
|
||||
@@ -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<string, unknown>;
|
||||
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<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;
|
||||
|
||||
@@ -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" }); });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 ?? "")}`;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<string, unknown>).mode).toBe("list");
|
||||
expect((payload.collect as Record<string, unknown>).view).toBe("files");
|
||||
});
|
||||
|
||||
function collectOptions(): NodeWebProbeObserveOptions {
|
||||
|
||||
Reference in New Issue
Block a user