From 661aef55a9027f245f7eb74215c04f776ade18b5 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sat, 24 Oct 2026 03:01:10 +0800 Subject: [PATCH] fix: retain completed historical replies through finalization --- docs/web-network-captures.md | 6 + src/artifacts/readStableArtifact.ts | 18 +++ src/browser/WebBundleAnalysisWorker.ts | 22 +--- src/browser/history/CaptureFailures.ts | 25 ++-- .../history/HistoricalCaptureDecoder.ts | 113 ++++++++---------- src/browser/history/HistoricalCaptureReply.ts | 93 ++++++++++++++ .../browser/historicalCaptureDecoder.test.ts | 113 ++++++++++++++++++ 7 files changed, 294 insertions(+), 96 deletions(-) create mode 100644 src/browser/history/HistoricalCaptureReply.ts diff --git a/docs/web-network-captures.md b/docs/web-network-captures.md index e5625237aa..a1a70c7930 100644 --- a/docs/web-network-captures.md +++ b/docs/web-network-captures.md @@ -744,6 +744,12 @@ budget. Those budgets have no runtime override. `capture_timeout_ms` does not raise them. Exceeding a budget returns no partial success. Oversized selected captures return an input error identifying `capture_path` and the byte budget; owned reply/output failures remain output or provider errors. +An explicit complete-reply refusal names the 100663296-byte allowance and its +lack of a runtime override. Record selection, paging, delivery mode and worker +memory do not increase that allowance; a smaller producer export is a narrower +observation, and the refusal leaves no completed capture Evidence to retrieve. +Cancellation during reply-file finalization retains validated completed capture +facts as partial observations. Malformed or unvalidated replies are not promoted. HAR, SAZ and qlog run in an owned Node old-generation heap with one V8 worker. `REA_CAPTURE_DECODER_OLD_GENERATION_MIB` is a decimal integer of mebibytes from 8 to 16384. Unset uses 192, the previous constant. Invalid values fail diff --git a/src/artifacts/readStableArtifact.ts b/src/artifacts/readStableArtifact.ts index 0a9a5a2c73..3fcd79bba4 100644 --- a/src/artifacts/readStableArtifact.ts +++ b/src/artifacts/readStableArtifact.ts @@ -4,6 +4,7 @@ import { lstat, open, type FileHandle } from "node:fs/promises"; import { ArtifactReaderFailure } from "./ArtifactReader.js"; import type { ArtifactResourceScope } from "./ArtifactResourceScope.js"; import { + artifactFileCompletion, withArtifactFileHandle, assertArtifactReadActive, } from "./ArtifactFileOperation.js"; @@ -14,6 +15,23 @@ export interface StableArtifactSnapshot { readonly sha256: string; } +/** Recover only the exact stable-read outcome retained by descriptor finalization. */ +export const completedStableArtifact = ( + cause: unknown, +): StableArtifactSnapshot | undefined => { + const value = artifactFileCompletion(cause)?.value; + if ( + value === null || + typeof value !== "object" || + !("bytes" in value) || + !Buffer.isBuffer(value.bytes) || + !("sha256" in value) || + typeof value.sha256 !== "string" + ) + return undefined; + return { bytes: value.bytes, sha256: value.sha256 }; +}; + /** Filesystem boundary for stable reads, including deterministic replacement probes. */ export interface StableArtifactFileSystem { lstat(path: string): Promise; diff --git a/src/browser/WebBundleAnalysisWorker.ts b/src/browser/WebBundleAnalysisWorker.ts index 38efaac391..1413d9eef6 100644 --- a/src/browser/WebBundleAnalysisWorker.ts +++ b/src/browser/WebBundleAnalysisWorker.ts @@ -16,7 +16,10 @@ import { artifactFileCancellationCause, withArtifactFileHandle, } from "../artifacts/ArtifactFileOperation.js"; -import { readStableArtifact } from "../artifacts/readStableArtifact.js"; +import { + completedStableArtifact, + readStableArtifact, +} from "../artifacts/readStableArtifact.js"; import type { StableArtifactSnapshot } from "../artifacts/readStableArtifact.js"; import { AnalysisError } from "../domain/analysisErrorBase.js"; import { @@ -107,21 +110,6 @@ type WebBundleActiveJob = { readonly resultBudgetBytes: number; }; -const stableSnapshotValue = ( - value: unknown, -): StableArtifactSnapshot | undefined => { - if ( - value === null || - typeof value !== "object" || - !("bytes" in value) || - !Buffer.isBuffer(value.bytes) || - !("sha256" in value) || - typeof value.sha256 !== "string" - ) - return undefined; - return { bytes: value.bytes, sha256: value.sha256 }; -}; - export interface WebBundleScriptAnalysisFailure { readonly error: AnalysisError; readonly completed: WebBundleScriptAnalysis | null; @@ -394,7 +382,7 @@ export class WebBundleAnalysisWorker { result = ok(facts); } catch (cause) { const completion = artifactFileCompletion(cause); - const snapshot = stableSnapshotValue(completion?.value); + const snapshot = completedStableArtifact(cause); if (snapshot !== undefined && resultDescriptor !== undefined) { try { completed = this.#validateCompletedResult( diff --git a/src/browser/history/CaptureFailures.ts b/src/browser/history/CaptureFailures.ts index 34600a7a69..d1db464bc0 100644 --- a/src/browser/history/CaptureFailures.ts +++ b/src/browser/history/CaptureFailures.ts @@ -134,11 +134,7 @@ export const historicalCaptureFailure = ( cause.message, { cause }, ); - if ( - phase !== "decoder" && - cause instanceof ArtifactReaderFailure && - cause.reason === "cancelled" - ) + if (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled") return new AnalysisCancelledError(OPERATION, { cause: artifactFileCancellationCause(cause), }); @@ -290,16 +286,6 @@ export const withCaptureReaderCleanup = ( ): AnalysisError => { if (!(cause instanceof ArtifactReaderFailure) || cause.cleanup === undefined) return primary; - if (cause.reason === "unavailable") { - const cleanup = new ProviderCleanupError( - "web-network-capture", - cause.cleanup.resources, - { reason: cause.cleanup.reason }, - { operation: OPERATION, cause }, - ); - bindArtifactFileCleanup(cause, cleanup); - return cleanup; - } const cleanup = new ProviderCleanupError( "web-network-capture", cause.cleanup.resources, @@ -307,6 +293,15 @@ export const withCaptureReaderCleanup = ( { operation: OPERATION, cause }, ); bindArtifactFileCleanup(cause, cleanup); + if ( + cause.reason === "unavailable" && + primary instanceof ProviderAdapterError && + primary.cause === cause + ) { + if (primary.partialObservation !== undefined) + cleanup.retainPartialObservation(primary.partialObservation); + return cleanup; + } const composed = analysisErrorWithCleanupFailure( primary, cleanup, diff --git a/src/browser/history/HistoricalCaptureDecoder.ts b/src/browser/history/HistoricalCaptureDecoder.ts index bb0c61cb62..f6fbb894c5 100644 --- a/src/browser/history/HistoricalCaptureDecoder.ts +++ b/src/browser/history/HistoricalCaptureDecoder.ts @@ -18,15 +18,13 @@ import { NODE_CAPTURE_HEAP_ENFORCED, captureDecoderProcessMemoryLimitBytes, } from "./CaptureRelease.js"; -import { projectAnalysisError } from "../../domain/analysisErrorProjection.js"; import type { HistoricalCaptureFormatAdapter } from "./HistoricalCaptureFormatAdapter.js"; import { randomUUID } from "node:crypto"; import { writeFile } from "node:fs/promises"; import { join } from "node:path"; -import { z } from "zod"; import type { ExecutionOptions } from "../../application/AnalysisProvider.js"; import { readStableArtifact } from "../../artifacts/readStableArtifact.js"; -import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js"; +import type { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js"; import { ArtifactResourceScope, type ArtifactCleanupAttempt, @@ -47,7 +45,6 @@ import { err, ok, type Result } from "../../domain/result.js"; import { WEB_NETWORK_CAPTURE_LIMITS, selectedWebNetworkCaptureTimeoutMs, - webNetworkCapturePayloadSchema, type InspectWebNetworkCaptureInput, type WebNetworkCapture, } from "../../domain/webNetworkCapture.js"; @@ -55,27 +52,10 @@ import { runOwnedCommand } from "../../process/OwnedCommand.js"; import { PrivateRuntimeRoot } from "../../process/PrivateRuntimeRoot.js"; import { redactExplicitFailure } from "../../domain/explicitSensitiveFailure.js"; +import { projectAnalysisError } from "../../domain/analysisErrorProjection.js"; +import { readHistoricalCaptureReply } from "./HistoricalCaptureReply.js"; + const OPERATION = "inspect_web_network_capture"; -const replySchema = z.discriminatedUnion("ok", [ - z.strictObject({ - ok: z.literal(true), - value: webNetworkCapturePayloadSchema, - }), - z.strictObject({ - ok: z.literal(false), - reason: z.enum([ - "format", - "unsupported", - "input-limit", - "expanded-limit", - "resource-limit", - "decoder", - "limit", - ]), - message: z.string(), - pointer: z.string(), - }), -]); /** One owned snapshot/process/root lifecycle for replaceable historical format decoders. */ export class HistoricalCaptureDecoder { @@ -145,6 +125,7 @@ export class HistoricalCaptureDecoder { let phase: "capture-read" | "decoder" = "capture-read"; let encodedReplyBytes = 0; let reportedHeap = this.#heap; + let replyFinalization: ArtifactReaderFailure | undefined; try { const adapter = this.adapters.find( (candidate) => candidate.format === input.format, @@ -212,34 +193,15 @@ export class HistoricalCaptureDecoder { undefined, reportedHeap, ); - let reply: z.output; - try { - const replyFile = await readStableArtifact( - replyPath, - WEB_NETWORK_CAPTURE_LIMITS.outputBytes, - options?.signal, - undefined, - this.#resources, - ); - encodedReplyBytes = replyFile.bytes.length; - reply = replySchema.parse(JSON.parse(replyFile.bytes.toString("utf8"))); - } catch (cause: unknown) { - if (cause instanceof ArtifactReaderFailure) throw cause; - if ( - options?.signal?.aborted === true && - (cause === options.signal.reason || - (cause instanceof Error && - cause.name === "AbortError" && - "code" in cause && - cause.code === "ABORT_ERR")) - ) - throw cause; - throw new AnalysisOutputError( - OPERATION, - `Owned capture decoder reply failed for ${input.capture_path}: ${replyFailureReason(cause)}`, - { cause }, - ); - } + const decodedReply = await readHistoricalCaptureReply( + replyPath, + input.capture_path, + this.#resources, + options?.signal, + ); + const reply = decodedReply.value; + encodedReplyBytes = decodedReply.bytes; + replyFinalization = decodedReply.finalization; if (!reply.ok) { if (reply.reason === "format") throw new AnalysisInputError(OPERATION, undefined, [ @@ -294,7 +256,14 @@ export class HistoricalCaptureDecoder { OPERATION, reply.message, ); - throw new AnalysisOutputError(OPERATION, reply.message); + throw new AnalysisOutputError(OPERATION, reply.message, { + userMessage: `The capture decoder refused a complete reply exceeding ${String(WEB_NETWORK_CAPTURE_LIMITS.outputBytes)} bytes: ${reply.message}`, + remediationAction: `The ${String(WEB_NETWORK_CAPTURE_LIMITS.outputBytes)}-byte complete-reply allowance has no runtime override. Select a smaller capture exported by its producer when a narrower observation is acceptable. Record selection, delivery mode and worker memory do not raise it; no completed capture Evidence is available from this refusal.`, + diagnostics: { + maximum_reply_bytes: WEB_NETWORK_CAPTURE_LIMITS.outputBytes, + boundary: "historical-capture-reply", + }, + }); } if ( reply.value.decoder.id !== adapter.identity.id || @@ -388,6 +357,17 @@ export class HistoricalCaptureDecoder { if (safePartialSource) error.retainPartialObservation(result.value); throw error; } + if (replyFinalization !== undefined) { + const error = historicalCaptureFailure( + input, + replyFinalization, + phase, + options, + reportedHeap, + ); + error.retainPartialObservation(result.value); + throw error; + } } } catch (cause: unknown) { const error = historicalCaptureFailure( @@ -397,9 +377,24 @@ export class HistoricalCaptureDecoder { options, reportedHeap, ); + if ( + replyFinalization?.reason === "cancelled" && + !(error instanceof AnalysisCancelledError) + ) { + const cancellation = historicalCaptureFailure( + input, + replyFinalization, + phase, + options, + reportedHeap, + ); + error.retainSecondaryFailure(cancellation, { + reply_finalization_error: projectAnalysisError(cancellation), + }); + } result = err( redactExplicitFailure( - withCaptureReaderCleanup(error, cause), + withCaptureReaderCleanup(error, replyFinalization ?? cause), input.sensitive_values, ), ); @@ -491,13 +486,3 @@ const nodeMemoryLimitation = (heap: CaptureDecoderHeap): string => { ? `HAR/SAZ/qlog run on Bun, which does not enforce REA's V8 heap flags. ${recovery}` : `HAR/SAZ/qlog run on Bun, which does not enforce REA's V8 heap flags, under a ${String(commitLimit)}-byte Windows Job Object commit limit. ${recovery}`; }; - -const replyFailureReason = (cause: unknown): string => { - if (cause instanceof ArtifactReaderFailure) return cause.message; - if (cause instanceof z.ZodError) - return `Reply violates its schema at ${cause.issues[0]?.path.map(String).join(".") ?? "root"}.`; - if (cause instanceof SyntaxError) return "Reply is malformed JSON."; - if (cause instanceof Error && "code" in cause) - return `Reply file could not be read (${String(cause.code)}).`; - return "Reply could not be validated."; -}; diff --git a/src/browser/history/HistoricalCaptureReply.ts b/src/browser/history/HistoricalCaptureReply.ts new file mode 100644 index 0000000000..c2607a95dd --- /dev/null +++ b/src/browser/history/HistoricalCaptureReply.ts @@ -0,0 +1,93 @@ +import { z } from "zod"; +import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js"; +import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; +import { + completedStableArtifact, + readStableArtifact, +} from "../../artifacts/readStableArtifact.js"; +import { AnalysisOutputError } from "../../domain/analysisErrorCore.js"; +import { + WEB_NETWORK_CAPTURE_LIMITS, + webNetworkCapturePayloadSchema, +} from "../../domain/webNetworkCapture.js"; + +const replySchema = z.discriminatedUnion("ok", [ + z.strictObject({ + ok: z.literal(true), + value: webNetworkCapturePayloadSchema, + }), + z.strictObject({ + ok: z.literal(false), + reason: z.enum([ + "format", + "unsupported", + "input-limit", + "expanded-limit", + "resource-limit", + "decoder", + "limit", + ]), + message: z.string(), + pointer: z.string(), + }), +]); + +/** Decode a bounded reply while keeping its completed read and descriptor finalization distinct. */ +export const readHistoricalCaptureReply = async ( + path: string, + capturePath: string, + resources: ArtifactResourceScope, + signal?: AbortSignal, +) => { + let finalization: ArtifactReaderFailure | undefined; + try { + let file; + try { + file = await readStableArtifact( + path, + WEB_NETWORK_CAPTURE_LIMITS.outputBytes, + signal, + undefined, + resources, + ); + } catch (cause: unknown) { + const completed = completedStableArtifact(cause); + if (!(cause instanceof ArtifactReaderFailure) || completed === undefined) + throw cause; + finalization = cause; + file = completed; + } + return { + value: replySchema.parse(JSON.parse(file.bytes.toString("utf8"))), + bytes: file.bytes.length, + finalization, + }; + } catch (cause: unknown) { + if (finalization !== undefined) throw finalization; + if (cause instanceof ArtifactReaderFailure) throw cause; + if ( + signal?.aborted === true && + (cause === signal.reason || + (cause instanceof Error && + cause.name === "AbortError" && + "code" in cause && + cause.code === "ABORT_ERR")) + ) + throw cause; + throw new AnalysisOutputError( + "inspect_web_network_capture", + `Owned capture decoder reply failed for ${capturePath}: ${replyFailureReason(cause)}`, + { cause }, + ); + } +}; + +const replyFailureReason = (cause: unknown): string => { + if (cause instanceof ArtifactReaderFailure) return cause.message; + if (cause instanceof z.ZodError) + return `Reply violates its schema at ${cause.issues[0]?.path.map(String).join(".") ?? "root"}.`; + if (cause instanceof SyntaxError) return "Reply is malformed JSON."; + if (cause instanceof Error && "code" in cause) + return `Reply file could not be read (${String(cause.code)}).`; + return "Reply could not be validated."; +}; diff --git a/tests/boundary/browser/historicalCaptureDecoder.test.ts b/tests/boundary/browser/historicalCaptureDecoder.test.ts index c5f001fcb2..73a7d3d93d 100644 --- a/tests/boundary/browser/historicalCaptureDecoder.test.ts +++ b/tests/boundary/browser/historicalCaptureDecoder.test.ts @@ -681,6 +681,25 @@ it.each([ ], }, }); + } else if (reason === "limit") { + expect(projected).toMatchObject({ + code: "unreadable_output", + retryable: false, + message: expect.stringContaining( + String(WEB_NETWORK_CAPTURE_LIMITS.outputBytes), + ), + details: { + diagnostics: { + maximum_reply_bytes: WEB_NETWORK_CAPTURE_LIMITS.outputBytes, + }, + }, + }); + expect(projected.remediation.action).toContain("no runtime override"); + expect(projected.remediation.action).toContain( + "no completed capture Evidence", + ); + expect(projected.remediation.action).not.toContain("rea doctor"); + expect(result.error.partialObservation).toBeUndefined(); } else if (tag === "ProviderAdapterError") { expect(projected).toMatchObject({ details: { @@ -837,3 +856,97 @@ it.each(["format", "process"])( expect(JSON.stringify(projected)).not.toContain("capture-private"); }, ); + +it.each([ + { closeFails: false, reply: "capture" }, + { closeFails: true, reply: "capture" }, + { closeFails: false, reply: "malformed" }, + { closeFails: true, reply: "refusal" }, +])( + "retains validated reply completion through descriptor cancellation %j", + async ({ closeFails, reply }) => { + const root = await mkdtemp(join(tmpdir(), "rea-reply-finalization-")); + onTestFinished(() => rm(root, { recursive: true, force: true })); + const path = join(root, "input.har"); + const text = JSON.stringify(historicalHar()); + await writeFile(path, text); + const controller = new AbortController(); + const reason = { caller: "reply descriptor close" }; + const originalClose = OwnedFileHandle.prototype.close; + let closes = 0; + const close = vi + .spyOn(OwnedFileHandle.prototype, "close") + .mockImplementation(async function (this: OwnedFileHandle) { + closes += 1; + if (closes === 2 && closeFails) { + controller.abort(reason); + throw new Error("controlled reply close denial"); + } + await originalClose.call(this); + if (closes === 2) controller.abort(reason); + }); + const decoder = new HistoricalCaptureDecoder( + [ + { + format: "har", + identity: HAR_CAPTURE_PROVIDER_IDENTITY, + command: async (_, runtimePath) => { + await writeFile( + join(runtimePath, "reply.json"), + reply === "malformed" + ? "{" + : JSON.stringify( + reply === "refusal" + ? { + ok: false, + reason: "limit", + message: "Complete reply allowance exceeded", + pointer: "", + } + : { ok: true, value: decodeHarCapture(text, []) }, + ), + ); + return { command: process.execPath, arguments: ["-e", ""] }; + }, + }, + ], + process.env, + ); + try { + const result = await decoder.inspect( + inspectWebNetworkCaptureInputSchema.parse({ + capture_path: path, + format: "har", + }), + { signal: controller.signal }, + ); + if (result.ok) + throw new Error("Expected reply finalization cancellation"); + expect(projectAnalysisError(result.error).code).toBe( + closeFails ? "cleanup_incomplete" : "cancelled", + ); + if (reply !== "capture") + expect(result.error.partialObservation).toBeUndefined(); + else + expect(result.error.partialObservation).toMatchObject({ + artifact: { path }, + records: [{ ordinal: 0 }], + }); + await decoder.close(); + expect(result.error.cleanupIncomplete).toBe(false); + expect(projectAnalysisError(result.error).code).toBe( + reply === "refusal" ? "unreadable_output" : "cancelled", + ); + const primary = result.error.cleanupPrimary ?? result.error; + if (reply === "refusal") + expect(primary.diagnostics).toMatchObject({ + reply_finalization_error: { code: "cancelled" }, + }); + else expect(primary.cause).toBe(reason); + expect(closes).toBe(closeFails ? 3 : 2); + } finally { + close.mockRestore(); + await decoder.close(); + } + }, +);