fix(input): classify JSON failures at their producing boundary

src/application/JsonInputFile.ts

@@ -5,7 +5,14 @@ import type { FileHandle } from "node:fs/promises";
 import { pipeline } from "node:stream/promises";
 import { getHeapStatistics } from "node:v8";
 
-import { asStream, none } from "stream-chain";
+import {
+  asStream,
+  flushable,
+  getFunctionList,
+  isFlushable,
+  isFunctionList,
+  none,
+} from "stream-chain";
 import fun from "stream-chain/fun.js";
 import parser from "stream-json/parser.js";
 import streamValues from "stream-json/streamers/stream-values.js";
@@ -208,6 +215,7 @@ export const readJsonObjectFile = (
                 handle,
                 stats.size,
                 path,
+                operation,
                 signal,
                 (bytes) => {
                   hash.update(bytes);
@@ -216,13 +224,7 @@ export const readJsonObjectFile = (
               ),
               asStream(
                 fun(
-                  parser({
-                    jsonStreaming: false,
-                    streamValues: false,
-                    packKeys: false,
-                    packStrings: false,
-                    packNumbers: false,
-                  }),
+                  jsonTokenizers(operation, stats.size, path),
                   packJsonScalars(operation, stats.size, path),
                   (token: parser.Token): typeof none | Promise<typeof none> => {
                     const opens =
@@ -374,6 +376,7 @@ export const readJsonInputFile = (
               handle,
               stats.size,
               path,
+              operation,
               signal,
               undefined,
               physicalRead,
@@ -382,13 +385,7 @@ export const readJsonInputFile = (
             // stream boundary. Admission still precedes scalar/value assembly.
             asStream(
               fun(
-                parser({
-                  jsonStreaming: false,
-                  streamValues: false,
-                  packKeys: false,
-                  packStrings: false,
-                  packNumbers: false,
-                }),
+                jsonTokenizers(operation, stats.size, path),
                 packJsonScalars(operation, stats.size, path),
                 admitJsonAssembly(
                   operation,
@@ -434,24 +431,52 @@ export const readJsonInputFile = (
     },
   );
 
+/** Keep parser failures inside the tokenizer while retaining synchronous stage fusion. */
+const jsonTokenizers = (
+  operation: string,
+  size: number,
+  path: string,
+): parser.TokenSource => {
+  const tokenizer = parser({
+    jsonStreaming: false,
+    streamValues: false,
+    packKeys: false,
+    packStrings: false,
+    packNumbers: false,
+  });
+  if (!isFunctionList(tokenizer))
+    throw new Error("JSON tokenizer does not expose its processor stages");
+  return fun(
+    ...getFunctionList(tokenizer).map((processor) => {
+      const tokenize = (value: unknown): unknown => {
+        try {
+          return processor(value);
+        } catch (cause: unknown) {
+          const failure = jsonReadFailure(cause, operation, size, path);
+          if (failure !== undefined) throw failure;
+          // The pinned tokenizer reports grammar violations with plain Error;
+          // native runtime failures keep their distinct types and identities.
+          if (cause instanceof Error && cause.constructor === Error)
+            throw invalidJson(operation, cause.message, cause);
+          throw cause;
+        }
+      };
+      return isFlushable(processor) ? flushable(tokenize) : tokenize;
+    }),
+  ) as parser.TokenSource;
+};
+
 const jsonReadFailure = (
   cause: unknown,
   operation: string,
   size: number,
   path: string,
 ): JsonInputFileFailure | undefined => {
-  if (cause instanceof AnalysisResourceConstraintError) return cause;
   if (
-    cause instanceof TypeError &&
-    "code" in cause &&
-    cause.code === "ERR_ENCODING_INVALID_ENCODED_DATA"
+    cause instanceof AnalysisInputError ||
+    cause instanceof AnalysisResourceConstraintError
   )
-    return invalidJson(operation, "JSON input is not valid UTF-8", cause);
-  if (
-    cause instanceof SyntaxError ||
-    (cause instanceof Error && cause.message.startsWith("Parser "))
-  )
-    return invalidJson(operation, cause.message, cause);
+    return cause;
   if (
     cause instanceof Error &&
     ((cause instanceof RangeError &&
@@ -633,6 +658,7 @@ async function* decodedChunks(
   handle: FileHandle,
   admittedBytes: number,
   path: string,
+  operation: string,
   signal: AbortSignal | undefined,
   observe?: (bytes: Uint8Array) => void,
   physicalRead: JsonInputPhysicalReadObservation = {},
@@ -642,10 +668,27 @@ async function* decodedChunks(
   const checked = (text: string): string => {
     if (leading && text.length > 0) {
       leading = false;
-      if (text.startsWith("\uFEFF")) throw new JsonByteOrderMarkError();
+      if (text.startsWith("\uFEFF"))
+        throw invalidJson(operation, JSON_BYTE_ORDER_MARK_MESSAGE);
     }
     return text;
   };
+  const decode = (
+    bytes?: Uint8Array,
+    options?: Parameters<typeof decoder.decode>[1],
+  ): string => {
+    try {
+      return decoder.decode(bytes, options);
+    } catch (cause: unknown) {
+      if (
+        cause instanceof TypeError &&
+        "code" in cause &&
+        cause.code === "ERR_ENCODING_INVALID_ENCODED_DATA"
+      )
+        throw invalidJson(operation, "JSON input is not valid UTF-8", cause);
+      throw cause;
+    }
+  };
   const bytes = Buffer.allocUnsafe(READ_CHUNK_BYTES);
   let observedBytes = 0;
   while (true) {
@@ -664,17 +707,10 @@ async function* decodedChunks(
     if (observedBytes > admittedBytes) throw new RegularFileChangedError(path);
     const chunk = bytes.subarray(0, read.bytesRead);
     observe?.(chunk);
-    yield checked(decoder.decode(chunk, { stream: true }));
+    yield checked(decode(chunk, { stream: true }));
   }
   if (observedBytes !== admittedBytes) throw new RegularFileChangedError(path);
-  yield checked(decoder.decode());
-}
-
-/** A streamed JSON file begins with a byte-order mark; reported as invalid JSON. */
-class JsonByteOrderMarkError extends SyntaxError {
-  constructor() {
-    super(JSON_BYTE_ORDER_MARK_MESSAGE);
-  }
+  yield checked(decode());
 }
 
 const readPrefix = async (

tests/boundary/filesystem/analysisSnapshot.test.ts

@@ -383,3 +383,40 @@ it("rejects copied preparation handles and prevents mutable caller values from c
     /not parsed in this realm/u,
   );
 });
+
+it.each([
+  new Error("Parser callback state is unavailable"),
+  new SyntaxError("selected member consumer failure"),
+  Object.assign(new TypeError("consumer text decoder failure"), {
+    code: "ERR_ENCODING_INVALID_ENCODED_DATA",
+  }),
+])(
+  "preserves a selected-member callback failure instead of blaming JSON: %s",
+  async (failure) => {
+    const directory = await createTestTempDirectory("rea-json-member-failure-");
+    const path = join(directory, "input.json");
+    const members = new Map([
+      [
+        "value",
+        {
+          mode: "value" as const,
+          begin: () => {
+            throw failure;
+          },
+          accept: () => {
+            throw new Error("Unexpected member acceptance");
+          },
+          reject: () => {
+            throw new Error("Unexpected member rejection");
+          },
+        },
+      ],
+    ]);
+    for (const padding of ["", " ".repeat(9 * 1024 * 1024)]) {
+      await writeFile(path, `${padding}{"value":1}`);
+      await expect(
+        readJsonObjectFile(path, "test-input", [members]),
+      ).rejects.toBe(failure);
+    }
+  },
+);