test: execute the HTTP scenario catalog

scripts/lib/bounded-pnpm-script.ts

@@ -0,0 +1,52 @@
+import type { SpawnSyncOptionsWithStringEncoding } from "node:child_process";
+
+export const PNPM_SCRIPT_TIMEOUT_MS = 60_000;
+export const PNPM_SCRIPT_MAX_OUTPUT_BYTES = 16 * 1024 * 1024;
+
+type BoundedPnpmScriptInvocation = Readonly<{
+  command: string;
+  arguments: readonly string[];
+  options: SpawnSyncOptionsWithStringEncoding;
+}>;
+
+type PnpmScriptResult = Readonly<{
+  status: number | null;
+  signal: NodeJS.Signals | null;
+  error?: Error;
+}>;
+
+export function createBoundedPnpmScriptInvocation(input: Readonly<{
+  nodePath: string;
+  pnpmCli: string;
+  script: string;
+  environment: NodeJS.ProcessEnv;
+}>): BoundedPnpmScriptInvocation {
+  return Object.freeze({
+    command: input.nodePath,
+    arguments: Object.freeze([input.pnpmCli, "run", input.script]),
+    options: Object.freeze({
+      encoding: "utf8",
+      env: input.environment,
+      killSignal: "SIGTERM",
+      maxBuffer: PNPM_SCRIPT_MAX_OUTPUT_BYTES,
+      timeout: PNPM_SCRIPT_TIMEOUT_MS,
+    }),
+  });
+}
+
+export function formatPnpmScriptFailure(
+  script: string,
+  result: PnpmScriptResult,
+): string {
+  const code =
+    result.error && "code" in result.error &&
+    typeof result.error.code === "string"
+      ? result.error.code
+      : null;
+  const message = result.error?.message.trim().replace(/\s+/g, " ") ?? null;
+  const error =
+    result.error === undefined
+      ? "none"
+      : `${code ?? result.error.name}: ${message || "no message"}`;
+  return `${script} failed: exit=${String(result.status)}, signal=${result.signal ?? "none"}, error=${error}`;
+}

scripts/lib/http-scenario-evidence.ts

@@ -0,0 +1,261 @@
+import { createHash } from "node:crypto";
+
+import { z } from "zod";
+
+export type AttemptStatus = number | "NETWORK_REJECTION" | "PENDING_ABORT";
+export type RetryReason = "NETWORK_FAILURE" | "HTTP_429" | "HTTP_503";
+export type BodyDisposition =
+  | "FULLY_READ_WITHIN_BOUND"
+  | "CANCELLED_WITHOUT_READ"
+  | "REJECTED_LIMIT"
+  | "NO_RESPONSE";
+
+const attemptStatusSchema = z.union([
+  z.number().int().min(100).max(599),
+  z.literal("NETWORK_REJECTION"),
+  z.literal("PENDING_ABORT"),
+]);
+
+export const httpScenarioAssertionGroupsSchema = z
+  .object({
+    status: z
+      .object({
+        attempts: z.array(attemptStatusSchema).min(1),
+        final: attemptStatusSchema,
+      })
+      .strict(),
+    outcome: z.object({ kind: z.string().min(1), detail: z.string().nullable() }).strict(),
+    effect: z.object({ outcome: z.string().min(1), observer: z.string().min(1) }).strict(),
+    retry: z
+      .object({
+        count: z.number().int().nonnegative(),
+        reasons: z.array(z.enum(["NETWORK_FAILURE", "HTTP_429", "HTTP_503"])),
+      })
+      .strict(),
+    fetch: z
+      .object({
+        count: z.number().int().positive(),
+        observerAttempts: z.number().int().positive(),
+        agrees: z.boolean(),
+      })
+      .strict(),
+    media: z.object({ attempts: z.array(z.string().nullable()).min(1), final: z.string().nullable() }).strict(),
+    body: z
+      .object({
+        attempts: z
+          .array(
+            z
+              .object({
+                disposition: z.enum([
+                  "FULLY_READ_WITHIN_BOUND",
+                  "CANCELLED_WITHOUT_READ",
+                  "REJECTED_LIMIT",
+                  "NO_RESPONSE",
+                ]),
+                pulledBytes: z.number().int().nonnegative(),
+                ceiling: z.number().int().nonnegative(),
+              })
+              .strict(),
+          )
+          .min(1),
+      })
+      .strict(),
+    scope: z
+      .object({
+        start: z.literal("CURRENT"),
+        end: z.enum(["CURRENT", "STALE"]),
+        signal: z.enum(["ACTIVE", "ABORTED"]),
+        cancellationOwner: z.enum(["NONE", "CALLER", "SCOPE_FENCE", "DEADLINE"]),
+      })
+      .strict(),
+  })
+  .strict()
+  .superRefine((groups, context) => {
+    const fail = (path: (string | number)[], message: string) => {
+      context.addIssue({ code: "custom", path, message });
+    };
+    if (groups.status.final !== groups.status.attempts.at(-1)) {
+      fail(["status", "final"], "must equal the final attempt status");
+    }
+    if (groups.retry.count !== groups.retry.reasons.length) {
+      fail(["retry", "count"], "must equal retry reasons length");
+    }
+    if (groups.fetch.count !== groups.status.attempts.length) {
+      fail(["fetch", "count"], "must equal status attempts length");
+    }
+    if (groups.fetch.count !== groups.media.attempts.length) {
+      fail(["media", "attempts"], "must equal fetch count");
+    }
+    if (groups.fetch.count !== groups.body.attempts.length) {
+      fail(["body", "attempts"], "must equal fetch count");
+    }
+    if (groups.fetch.observerAttempts !== groups.fetch.count) {
+      fail(["fetch", "observerAttempts"], "must equal physical fetch count");
+    }
+    if (!groups.fetch.agrees) {
+      fail(["fetch", "agrees"], "must prove physical/observer agreement");
+    }
+    if (groups.media.final !== groups.media.attempts.at(-1)) {
+      fail(["media", "final"], "must equal the final attempt media essence");
+    }
+  });
+
+export type HttpScenarioAssertionGroups = Readonly<{
+  status: Readonly<{ attempts: readonly AttemptStatus[]; final: AttemptStatus }>;
+  outcome: Readonly<{ kind: string; detail: string | null }>;
+  effect: Readonly<{ outcome: string; observer: string }>;
+  retry: Readonly<{ count: number; reasons: readonly RetryReason[] }>;
+  fetch: Readonly<{
+    count: number;
+    observerAttempts: number;
+    agrees: boolean;
+  }>;
+  media: Readonly<{
+    attempts: readonly (string | null)[];
+    final: string | null;
+  }>;
+  body: Readonly<{
+    attempts: readonly Readonly<{
+      disposition: BodyDisposition;
+      pulledBytes: number;
+      ceiling: number;
+    }>[];
+  }>;
+  scope: Readonly<{
+    start: "CURRENT";
+    end: "CURRENT" | "STALE";
+    signal: "ACTIVE" | "ABORTED";
+    cancellationOwner: "NONE" | "CALLER" | "SCOPE_FENCE" | "DEADLINE";
+  }>;
+}>;
+
+export const httpScenarioExpectationSchema = z
+  .object({
+    executionId: z.string().min(1),
+    operationId: z.string().min(1),
+    scenarioId: z.string().min(1),
+    expected: httpScenarioAssertionGroupsSchema,
+    testDeadlineOverrideMs: z.number().int().positive().nullable(),
+  })
+  .strict()
+  .superRefine((entry, context) => {
+    if (entry.executionId !== `${entry.operationId}::${entry.scenarioId}`) {
+      context.addIssue({
+        code: "custom",
+        path: ["executionId"],
+        message: "must equal operationId::scenarioId",
+      });
+    }
+  });
+
+export type HttpScenarioExpectation = Readonly<{
+  executionId: string;
+  operationId: string;
+  scenarioId: string;
+  expected: HttpScenarioAssertionGroups;
+  testDeadlineOverrideMs: number | null;
+}>;
+
+export const httpScenarioReceiptRowSchema = z
+  .object({
+    executionId: z.string().min(1),
+    expected: httpScenarioAssertionGroupsSchema,
+    observed: httpScenarioAssertionGroupsSchema,
+    testDeadlineOverrideMs: z.number().int().positive().nullable(),
+  })
+  .strict();
+
+export const httpScenarioReceiptSchema = z
+  .object({
+    schemaVersion: z.number().int().positive(),
+    catalogDigest: z.string().regex(/^sha256:[0-9a-f]{64}$/),
+    catalogTotal: z.number().int().nonnegative(),
+    executedIds: z.array(z.string().min(1)),
+    rows: z.array(httpScenarioReceiptRowSchema),
+  })
+  .strict()
+  .superRefine((receipt, context) => {
+    const rowIds = receipt.rows.map((row) => row.executionId);
+    const sortedIds = [...receipt.executedIds].sort((left, right) =>
+      left.localeCompare(right),
+    );
+    const sortedRowIds = [...rowIds].sort((left, right) =>
+      left.localeCompare(right),
+    );
+    if (!sameScenarioJson(receipt.executedIds, sortedIds)) {
+      context.addIssue({
+        code: "custom",
+        path: ["executedIds"],
+        message: "must be sorted by execution ID",
+      });
+    }
+    if (!sameScenarioJson(rowIds, sortedRowIds)) {
+      context.addIssue({
+        code: "custom",
+        path: ["rows"],
+        message: "must be sorted by execution ID",
+      });
+    }
+    if (!sameScenarioJson(receipt.executedIds, rowIds)) {
+      context.addIssue({
+        code: "custom",
+        path: ["rows"],
+        message: "row IDs must exactly equal executed IDs",
+      });
+    }
+  });
+
+export type HttpScenarioReceipt = Readonly<{
+  schemaVersion: number;
+  catalogDigest: string;
+  catalogTotal: number;
+  executedIds: readonly string[];
+  rows: readonly Readonly<{
+    executionId: string;
+    expected: HttpScenarioAssertionGroups;
+    observed: HttpScenarioAssertionGroups;
+    testDeadlineOverrideMs: number | null;
+  }>[];
+}>;
+
+export function stableScenarioJson(value: unknown): string {
+  const normalize = (candidate: unknown): unknown => {
+    if (Array.isArray(candidate)) return candidate.map(normalize);
+    if (candidate && typeof candidate === "object") {
+      return Object.fromEntries(
+        Object.entries(candidate as Readonly<Record<string, unknown>>)
+          .sort(([left], [right]) => left.localeCompare(right))
+          .map(([key, nested]) => [key, normalize(nested)]),
+      );
+    }
+    return candidate;
+  };
+  return JSON.stringify(normalize(value));
+}
+
+export function sameScenarioJson(left: unknown, right: unknown): boolean {
+  return stableScenarioJson(left) === stableScenarioJson(right);
+}
+
+export function computeHttpScenarioCatalogDigest(
+  schemaVersion: number,
+  expectations: readonly HttpScenarioExpectation[],
+): `sha256:${string}` {
+  const tuples = [...expectations]
+    .sort((left, right) => left.executionId.localeCompare(right.executionId))
+    .map((entry) => [
+      entry.executionId,
+      entry.expected.status,
+      entry.expected.outcome,
+      entry.expected.effect,
+      entry.expected.retry,
+      entry.expected.fetch,
+      entry.expected.media,
+      entry.expected.body,
+      entry.expected.scope,
+      entry.testDeadlineOverrideMs,
+    ]);
+  return `sha256:${createHash("sha256")
+    .update(stableScenarioJson([schemaVersion, ...tuples]))
+    .digest("hex")}`;
+}

scripts/lib/validated-json-artifact.ts

@@ -1,4 +1,5 @@
 import { randomUUID } from "node:crypto";
+import { constants } from "node:fs";
 import {
   open as openFile,
   rename as renameFile,
@@ -15,8 +16,13 @@ export type ValidatedJsonArtifactInput = Readonly<{
 }>;
 
 export type ValidatedJsonArtifactFileSystem = Readonly<{
-  open: (path: string, flags: "wx") => Promise<{
+  open: (path: string, flags: number, mode: number) => Promise<{
     writeFile(data: string, encoding: "utf8"): Promise<unknown>;
+    sync(): Promise<unknown>;
+    close(): Promise<unknown>;
+  }>;
+  openDirectory: (path: string) => Promise<{
+    sync(): Promise<unknown>;
     close(): Promise<unknown>;
   }>;
   rename: (source: string, destination: string) => Promise<unknown>;
@@ -29,11 +35,21 @@ type ValidatedJsonArtifactWriterDependencies = Readonly<{
 }>;
 
 const defaultFileSystem: ValidatedJsonArtifactFileSystem = Object.freeze({
-  open: async (target, flags) => openFile(target, flags),
+  open: async (target, flags, mode) => openFile(target, flags, mode),
+  openDirectory: async (target) => openFile(target, constants.O_RDONLY),
   rename: async (source, destination) => renameFile(source, destination),
   rm: async (target, options) => removeFile(target, options),
 });
 
+function hasErrorCode(error: unknown, code: string): boolean {
+  return (
+    typeof error === "object" &&
+    error !== null &&
+    "code" in error &&
+    error.code === code
+  );
+}
+
 /**
  * Builds a writer whose only publish operation is an atomic sibling rename.
  * Dependency injection is limited to the file-system boundary so failure
@@ -60,12 +76,20 @@ export function createValidatedJsonArtifactWriter(
     );
     let ownsTemporaryFile = false;
     try {
-      const handle = await fileSystem.open(temporaryPath, "wx");
+      const handle = await fileSystem.open(
+        temporaryPath,
+        constants.O_WRONLY |
+          constants.O_CREAT |
+          constants.O_EXCL |
+          constants.O_NOFOLLOW,
+        0o600,
+      );
       ownsTemporaryFile = true;
       let writeFailed = false;
       let writeFailure: unknown;
       try {
         await handle.writeFile(`${serialized}\n`, "utf8");
+        await handle.sync();
       } catch (error) {
         writeFailed = true;
         writeFailure = error;
@@ -81,6 +105,21 @@ export function createValidatedJsonArtifactWriter(
       if (writeFailed) throw writeFailure;
       if (closeFailed) throw closeFailure;
       await fileSystem.rename(temporaryPath, input.path);
+      ownsTemporaryFile = false;
+      const directoryHandle = await fileSystem.openDirectory(
+        path.dirname(input.path),
+      );
+      try {
+        try {
+          await directoryHandle.sync();
+        } catch (error) {
+          if (!hasErrorCode(error, "EINVAL") && !hasErrorCode(error, "ENOTSUP")) {
+            throw error;
+          }
+        }
+      } finally {
+        await directoryHandle.close();
+      }
     } catch (error) {
       if (ownsTemporaryFile) {
         try {