test(reef): consolidate transport protocol fixtures (#118281)

extensions/reef/protocol/pipeline.test.ts

@@ -55,6 +55,88 @@ function identities() {
   return { alice: generateIdentity(), bob: generateIdentity() };
 }
 
+type Identity = ReturnType<typeof generateIdentity>;
+type OutboundOptions = Parameters<typeof composeOutbound>[0];
+type InboundOptions = Parameters<typeof composeInbound>[0];
+
+function outboundOptions(
+  alice: Identity,
+  bob: Identity,
+  overrides: Partial<OutboundOptions> = {},
+): OutboundOptions {
+  return {
+    id: "01JZ0000000000000000000000",
+    from: "alice#1",
+    to: "bob#1",
+    body: { text: "ordinary text" },
+    senderSigningSecretKey: alice.signing.secretKey,
+    recipientEncryptionPublicKey: bob.encryption.publicKey,
+    guard: mockGuard(allow),
+    audit: audit(),
+    policyVersion: "v1",
+    ...overrides,
+  };
+}
+
+function sealedEnvelope(
+  alice: Identity,
+  bob: Identity,
+  id: string,
+  body: Parameters<typeof seal>[0]["body"],
+): Envelope {
+  return seal({
+    id,
+    from: "alice#1",
+    to: "bob#1",
+    body,
+    senderSigningSecretKey: alice.signing.secretKey,
+    recipientEncryptionPublicKey: bob.encryption.publicKey,
+    ts: now,
+  });
+}
+
+function inboundOptions(
+  envelope: Envelope,
+  alice: Identity,
+  bob: Identity,
+  overrides: Partial<InboundOptions> = {},
+): InboundOptions {
+  return {
+    envelope,
+    self: "bob#1",
+    recipientEncryptionSecretKey: bob.encryption.secretKey,
+    recipientSigningSecretKey: bob.signing.secretKey,
+    senderSigningPublicKey: alice.signing.publicKey,
+    replayStore: new MemoryReplayStore(),
+    now,
+    guard: mockGuard(allow),
+    audit: audit(),
+    policyVersion: "v1",
+    ...overrides,
+  };
+}
+
+function reviewVerdict(): Verdict {
+  return {
+    ...allow,
+    decision: "review",
+    category: "ambiguous",
+    reason: "Review.",
+  };
+}
+
+async function capturePipelineError(promise: Promise<unknown>): Promise<PipelineError> {
+  try {
+    await promise;
+  } catch (error) {
+    if (error instanceof PipelineError) {
+      return error;
+    }
+    throw error;
+  }
+  throw new Error("expected pipeline error");
+}
+
 class FailOnceAuditStore implements AuditStore {
   readonly inner = audit();
   #fail = true;
@@ -76,34 +158,18 @@ describe("pipeline", () => {
   it("runs an allowed outbound and inbound exchange end to end", async () => {
     const { alice, bob } = identities();
     const outboundAudit = audit();
-    const outbound = await composeOutbound({
-      id: "01JZ0000000000000000000000",
-      from: "alice#1",
-      to: "bob#1",
-      body: { text: "hello" },
-      senderSigningSecretKey: alice.signing.secretKey,
-      recipientEncryptionPublicKey: bob.encryption.publicKey,
-      ts: now,
-      guard: mockGuard(allow),
-      audit: outboundAudit,
-      policyVersion: "v1",
-    });
+    const outbound = await composeOutbound(
+      outboundOptions(alice, bob, { body: { text: "hello" }, ts: now, audit: outboundAudit }),
+    );
     const inboundAudit = audit();
     const inboundGuard = mockGuard(allow);
     const replayStore = new MemoryReplayStore();
-    const inboundOptions = {
-      envelope: outbound.envelope,
-      self: "bob#1",
-      recipientEncryptionSecretKey: bob.encryption.secretKey,
-      recipientSigningSecretKey: bob.signing.secretKey,
-      senderSigningPublicKey: alice.signing.publicKey,
+    const options = inboundOptions(outbound.envelope, alice, bob, {
       replayStore,
-      now,
       guard: inboundGuard,
       audit: inboundAudit,
-      policyVersion: "v1",
-    };
-    const inbound = await composeInbound(inboundOptions);
+    });
+    const inbound = await composeInbound(options);
     expect(inbound.disposition).toBe("accepted");
     if (inbound.disposition !== "accepted") {
       throw new Error("expected accepted result");
@@ -136,7 +202,7 @@ describe("pipeline", () => {
         ),
       ).size,
     ).toBe(1);
-    const duplicate = await composeInbound({ ...inboundOptions, now: now + 10 * 60 });
+    const duplicate = await composeInbound({ ...options, now: now + 10 * 60 });
     expect(duplicate).toEqual({
       disposition: "duplicate",
       body: inbound.body,
@@ -151,18 +217,7 @@ describe("pipeline", () => {
     const guard = mockGuard(allow);
     const envelope = craftEnvelope({ text: "hello", thread: "free-form thread" }, alice, bob);
     await expect(
-      composeInbound({
-        envelope,
-        self: "bob#1",
-        recipientEncryptionSecretKey: bob.encryption.secretKey,
-        recipientSigningSecretKey: bob.signing.secretKey,
-        senderSigningPublicKey: alice.signing.publicKey,
-        replayStore: new MemoryReplayStore(),
-        now,
-        guard,
-        audit: audit(),
-        policyVersion: "v1",
-      }),
+      composeInbound(inboundOptions(envelope, alice, bob, { guard })),
     ).rejects.toMatchObject({ code: "malformed" });
     expect(guard.calls).toBe(0);
   });
@@ -172,21 +227,16 @@ describe("pipeline", () => {
     const guard = mockGuard(allow);
     let rngCalls = 0;
     await expect(
-      composeOutbound({
-        id: "01JZ0000000000000000000000",
-        from: "alice#1",
-        to: "bob#1",
-        body: { text: ["sk-", "abcdefghijklmnopqrstuvwxyz123456"].join("") },
-        senderSigningSecretKey: alice.signing.secretKey,
-        recipientEncryptionPublicKey: bob.encryption.publicKey,
-        guard,
-        audit: audit(),
-        policyVersion: "v1",
-        rng(length) {
-          rngCalls++;
-          return new Uint8Array(length);
-        },
-      }),
+      composeOutbound(
+        outboundOptions(alice, bob, {
+          body: { text: ["sk-", "abcdefghijklmnopqrstuvwxyz123456"].join("") },
+          guard,
+          rng(length) {
+            rngCalls++;
+            return new Uint8Array(length);
+          },
+        }),
+      ),
     ).rejects.toMatchObject({ stage: "deterministic" });
     expect(guard.calls).toBe(0);
     expect(rngCalls).toBe(0);
@@ -202,21 +252,15 @@ describe("pipeline", () => {
     };
     let rngCalls = 0;
     await expect(
-      composeOutbound({
-        id: "01JZ0000000000000000000000",
-        from: "alice#1",
-        to: "bob#1",
-        body: { text: "ordinary text" },
-        senderSigningSecretKey: alice.signing.secretKey,
-        recipientEncryptionPublicKey: bob.encryption.publicKey,
-        guard: mockGuard(deny),
-        audit: audit(),
-        policyVersion: "v1",
-        rng(length) {
-          rngCalls++;
-          return new Uint8Array(length);
-        },
-      }),
+      composeOutbound(
+        outboundOptions(alice, bob, {
+          guard: mockGuard(deny),
+          rng(length) {
+            rngCalls++;
+            return new Uint8Array(length);
+          },
+        }),
+      ),
     ).rejects.toBeInstanceOf(PipelineError);
     expect(rngCalls).toBe(0);
   });
@@ -232,21 +276,17 @@ describe("pipeline", () => {
     for (const [name, pinnedModel, rawVerdict] of cases) {
       let rngCalls = 0;
       await expect(
-        composeOutbound({
-          id: "01JZ0000000000000000000008",
-          from: "alice#1",
-          to: "bob#1",
-          body: { text: `case ${name}` },
-          senderSigningSecretKey: alice.signing.secretKey,
-          recipientEncryptionPublicKey: bob.encryption.publicKey,
-          guard: structuralGuard(pinnedModel, rawVerdict),
-          audit: audit(),
-          policyVersion: "v1",
-          rng(length) {
-            rngCalls++;
-            return new Uint8Array(length);
-          },
-        }),
+        composeOutbound(
+          outboundOptions(alice, bob, {
+            id: "01JZ0000000000000000000008",
+            body: { text: `case ${name}` },
+            guard: structuralGuard(pinnedModel, rawVerdict),
+            rng(length) {
+              rngCalls++;
+              return new Uint8Array(length);
+            },
+          }),
+        ),
       ).rejects.toMatchObject({
         stage: "guard",
         verdict: { decision: "deny", category: "guard_failure" },
@@ -257,25 +297,14 @@ describe("pipeline", () => {
 
   it("accepts a valid structural adapter and admits the post-review verdict again", async () => {
     const { alice, bob } = identities();
-    const common = {
+    const common = outboundOptions(alice, bob, {
       id: "01JZ0000000000000000000009",
-      from: "alice#1",
-      to: "bob#1",
       body: { text: "structural adapter" },
-      senderSigningSecretKey: alice.signing.secretKey,
-      recipientEncryptionPublicKey: bob.encryption.publicKey,
-      audit: audit(),
-      policyVersion: "v1",
-    };
+    });
     await expect(
       composeOutbound({ ...common, guard: structuralGuard(allow.model, allow) }),
     ).resolves.toMatchObject({ verdict: allow });
-    const review: Verdict = {
-      ...allow,
-      decision: "review",
-      category: "ambiguous",
-      reason: "Review.",
-    };
+    const review = reviewVerdict();
     const invalidAfterApproval = { ...allow, model: "wrong-2026-07-12" };
     await expect(
       composeOutbound({
@@ -292,28 +321,15 @@ describe("pipeline", () => {
 
   it("rejects an invalid structural inbound verdict instead of accepting it", async () => {
     const { alice, bob } = identities();
-    const envelope = seal({
-      id: "01JZ0000000000000000000010",
-      from: "alice#1",
-      to: "bob#1",
-      body: { text: "inbound structural adapter" },
-      senderSigningSecretKey: alice.signing.secretKey,
-      recipientEncryptionPublicKey: bob.encryption.publicKey,
-      ts: now,
+    const envelope = sealedEnvelope(alice, bob, "01JZ0000000000000000000010", {
+      text: "inbound structural adapter",
     });
     await expect(
-      composeInbound({
-        envelope,
-        self: "bob#1",
-        recipientEncryptionSecretKey: bob.encryption.secretKey,
-        recipientSigningSecretKey: bob.signing.secretKey,
-        senderSigningPublicKey: alice.signing.publicKey,
-        replayStore: new MemoryReplayStore(),
-        now,
-        guard: structuralGuard(allow.model, { ...allow, policyVersion: "wrong" }),
-        audit: audit(),
-        policyVersion: "v1",
-      }),
+      composeInbound(
+        inboundOptions(envelope, alice, bob, {
+          guard: structuralGuard(allow.model, { ...allow, policyVersion: "wrong" }),
+        }),
+      ),
     ).rejects.toMatchObject({
       stage: "guard",
       verdict: { decision: "deny", category: "guard_failure" },
@@ -323,21 +339,8 @@ describe("pipeline", () => {
 
   it("requires exact full-proposal approval and fresh classification for review", async () => {
     const { alice, bob } = identities();
-    const review: Verdict = {
-      ...allow,
-      decision: "review",
-      category: "ambiguous",
-      reason: "Review.",
-    };
-    const common = {
-      id: "01JZ0000000000000000000000",
-      from: "alice#1",
-      to: "bob#1",
-      body: { text: "ordinary text" },
-      senderSigningSecretKey: alice.signing.secretKey,
-      recipientEncryptionPublicKey: bob.encryption.publicKey,
-      policyVersion: "v1",
-    };
+    const review = reviewVerdict();
+    const common = outboundOptions(alice, bob);
     await expect(
       composeOutbound({ ...common, audit: audit(), guard: mockGuard(review) }),
     ).rejects.toMatchObject({ stage: "review" });
@@ -370,20 +373,12 @@ describe("pipeline", () => {
 
   it("does not reuse an approval across recipients", async () => {
     const { alice, bob } = identities();
-    const review: Verdict = {
-      ...allow,
-      decision: "review",
-      category: "ambiguous",
-      reason: "Review.",
-    };
-    const common = {
+    const review = reviewVerdict();
+    const common = outboundOptions(alice, bob, {
       id: "01JZ0000000000000000000007",
       from: "sender#1",
       body: { text: "identical body" },
-      senderSigningSecretKey: alice.signing.secretKey,
-      recipientEncryptionPublicKey: bob.encryption.publicKey,
-      policyVersion: "v1",
-    };
+    });
     let vincentDigest = "";
     await expect(
       composeOutbound({
@@ -417,29 +412,15 @@ describe("pipeline", () => {
 
   it("releases a replay claim after transient audit failure and retries successfully", async () => {
     const { alice, bob } = identities();
-    const envelope = seal({
-      id: "01JZ0000000000000000000001",
-      from: "alice#1",
-      to: "bob#1",
-      body: { text: "retry me" },
-      senderSigningSecretKey: alice.signing.secretKey,
-      recipientEncryptionPublicKey: bob.encryption.publicKey,
-      ts: now,
+    const envelope = sealedEnvelope(alice, bob, "01JZ0000000000000000000001", {
+      text: "retry me",
     });
     const replayStore = new MemoryReplayStore();
     const inboundAudit = new FailOnceAuditStore();
-    const options = {
-      envelope,
-      self: "bob#1",
-      recipientEncryptionSecretKey: bob.encryption.secretKey,
-      recipientSigningSecretKey: bob.signing.secretKey,
-      senderSigningPublicKey: alice.signing.publicKey,
+    const options = inboundOptions(envelope, alice, bob, {
       replayStore,
-      now,
-      guard: mockGuard(allow),
       audit: inboundAudit,
-      policyVersion: "v1",
-    };
+    });
     await expect(composeInbound(options)).rejects.toThrow("transient audit failure");
     const retried = await composeInbound(options);
     expect(retried.disposition).toBe("accepted");
@@ -448,45 +429,23 @@ describe("pipeline", () => {
 
   it("returns an identical cached rejection receipt on guard-deny redelivery", async () => {
     const { alice, bob } = identities();
-    const envelope = seal({
-      id: "01JZ0000000000000000000002",
-      from: "alice#1",
-      to: "bob#1",
-      body: { text: "classify me" },
-      senderSigningSecretKey: alice.signing.secretKey,
-      recipientEncryptionPublicKey: bob.encryption.publicKey,
-      ts: now,
+    const envelope = sealedEnvelope(alice, bob, "01JZ0000000000000000000002", {
+      text: "classify me",
     });
     const replayStore = new MemoryReplayStore();
     const inboundAudit = audit();
     const deny: Verdict = { ...allow, decision: "deny", category: "injection", reason: "Denied." };
     const guard = mockGuard(deny);
-    const options = {
-      envelope,
-      self: "bob#1",
-      recipientEncryptionSecretKey: bob.encryption.secretKey,
-      recipientSigningSecretKey: bob.signing.secretKey,
-      senderSigningPublicKey: alice.signing.publicKey,
+    const options = inboundOptions(envelope, alice, bob, {
       replayStore,
-      now,
       guard,
       audit: inboundAudit,
-      policyVersion: "v1",
-    };
-    let rejection: PipelineError | undefined;
-    try {
-      await composeInbound(options);
-    } catch (error) {
-      if (error instanceof PipelineError) {
-        rejection = error;
-      } else {
-        throw error;
-      }
-    }
-    expect(rejection?.receipt).toMatchObject({ status: "rejected", category: "guard_deny" });
+    });
+    const rejection = await capturePipelineError(composeInbound(options));
+    expect(rejection.receipt).toMatchObject({ status: "rejected", category: "guard_deny" });
     const entryCount = (await inboundAudit.entries()).length;
     const duplicate = await composeInbound({ ...options, now: now + 1_000 });
-    expect(duplicate).toEqual({ disposition: "duplicate", receipt: rejection!.receipt });
+    expect(duplicate).toEqual({ disposition: "duplicate", receipt: rejection.receipt });
     expect(duplicate).not.toHaveProperty("body");
     expect((await inboundAudit.entries()).length).toBe(entryCount);
     expect(guard.calls).toBe(1);
@@ -494,50 +453,23 @@ describe("pipeline", () => {
 
   it("completes explicit inbound review denial and caches its receipt", async () => {
     const { alice, bob } = identities();
-    const envelope = seal({
-      id: "01JZ0000000000000000000005",
-      from: "alice#1",
-      to: "bob#1",
-      body: { text: "review me" },
-      senderSigningSecretKey: alice.signing.secretKey,
-      recipientEncryptionPublicKey: bob.encryption.publicKey,
-      ts: now,
+    const envelope = sealedEnvelope(alice, bob, "01JZ0000000000000000000005", {
+      text: "review me",
     });
-    const review: Verdict = {
-      ...allow,
-      decision: "review",
-      category: "ambiguous",
-      reason: "Review.",
-    };
+    const review = reviewVerdict();
     const guard = mockGuard(review);
     const replayStore = new MemoryReplayStore();
     const inboundAudit = audit();
-    const options = {
-      envelope,
-      self: "bob#1",
-      recipientEncryptionSecretKey: bob.encryption.secretKey,
-      recipientSigningSecretKey: bob.signing.secretKey,
-      senderSigningPublicKey: alice.signing.publicKey,
+    const options = inboundOptions(envelope, alice, bob, {
       replayStore,
-      now,
       guard,
       audit: inboundAudit,
-      policyVersion: "v1",
       reviewGate: async ({ approvalDigest }: { approvalDigest: string }) => ({
         approved: false,
         approvalDigest,
       }),
-    };
-    let rejection: PipelineError | undefined;
-    try {
-      await composeInbound(options);
-    } catch (error) {
-      if (error instanceof PipelineError) {
-        rejection = error;
-      } else {
-        throw error;
-      }
-    }
+    });
+    const rejection = await capturePipelineError(composeInbound(options));
     expect(rejection).toMatchObject({
       stage: "review",
       reviewOutcome: "denied",
@@ -546,7 +478,7 @@ describe("pipeline", () => {
     const entryCount = (await inboundAudit.entries()).length;
     await expect(composeInbound(options)).resolves.toEqual({
       disposition: "duplicate",
-      receipt: rejection!.receipt,
+      receipt: rejection.receipt,
     });
     expect((await inboundAudit.entries()).length).toBe(entryCount);
     expect(guard.calls).toBe(1);
@@ -554,37 +486,17 @@ describe("pipeline", () => {
 
   it("releases pending inbound review and accepts a later approved retry", async () => {
     const { alice, bob } = identities();
-    const envelope = seal({
-      id: "01JZ0000000000000000000006",
-      from: "alice#1",
-      to: "bob#1",
-      body: { text: "decide later" },
-      senderSigningSecretKey: alice.signing.secretKey,
-      recipientEncryptionPublicKey: bob.encryption.publicKey,
-      ts: now,
+    const envelope = sealedEnvelope(alice, bob, "01JZ0000000000000000000006", {
+      text: "decide later",
     });
-    const review: Verdict = {
-      ...allow,
-      decision: "review",
-      category: "ambiguous",
-      reason: "Review.",
-    };
+    const review = reviewVerdict();
     const guard = mockGuard(review, review, allow);
     let decided = false;
-    const options = {
-      envelope,
-      self: "bob#1",
-      recipientEncryptionSecretKey: bob.encryption.secretKey,
-      recipientSigningSecretKey: bob.signing.secretKey,
-      senderSigningPublicKey: alice.signing.publicKey,
-      replayStore: new MemoryReplayStore(),
-      now,
+    const options = inboundOptions(envelope, alice, bob, {
       guard,
-      audit: audit(),
-      policyVersion: "v1",
       reviewGate: async ({ approvalDigest }: { approvalDigest: string }) =>
         decided ? { approved: true, approvalDigest } : undefined,
-    };
+    });
     await expect(composeInbound(options)).rejects.toMatchObject({
       stage: "review",
       reviewOutcome: "pending",
@@ -600,29 +512,12 @@ describe("pipeline", () => {
 
   it("completes deterministic inbound denial with a signed rejection", async () => {
     const { alice, bob } = identities();
-    const envelope = seal({
-      id: "01JZ0000000000000000000003",
-      from: "alice#1",
-      to: "bob#1",
-      body: { text: ["sk-", "abcdefghijklmnopqrstuvwxyz123456"].join("") },
-      senderSigningSecretKey: alice.signing.secretKey,
-      recipientEncryptionPublicKey: bob.encryption.publicKey,
-      ts: now,
+    const envelope = sealedEnvelope(alice, bob, "01JZ0000000000000000000003", {
+      text: ["sk-", "abcdefghijklmnopqrstuvwxyz123456"].join(""),
     });
     const guard = mockGuard(allow);
     await expect(
-      composeInbound({
-        envelope,
-        self: "bob#1",
-        recipientEncryptionSecretKey: bob.encryption.secretKey,
-        recipientSigningSecretKey: bob.signing.secretKey,
-        senderSigningPublicKey: alice.signing.publicKey,
-        replayStore: new MemoryReplayStore(),
-        now,
-        guard,
-        audit: audit(),
-        policyVersion: "v1",
-      }),
+      composeInbound(inboundOptions(envelope, alice, bob, { guard })),
     ).rejects.toMatchObject({
       stage: "deterministic",
       receipt: { status: "rejected", category: "deterministic_deny" },

extensions/reef/src/flow-receipts.test.ts

@@ -33,25 +33,59 @@ import type { InboxEntry } from "./types.js";
 beforeEach(resetFlowStoresForTests);
 afterEach(resetFlowStoresForTests);
 
+type FlowOptions = ConstructorParameters<typeof ReefMessageFlow>[0];
+type TrustedReef = ReturnType<typeof trust>;
+
+function createFlow(params: {
+  alice: ReturnType<typeof generateIdentity>;
+  bob: ReturnType<typeof reefKeys>;
+  audit: MemoryAuditStore;
+  trusted?: TrustedReef;
+  relay?: ReturnType<typeof transport>;
+  onOwnerNotice?: FlowOptions["onOwnerNotice"];
+}): ReefMessageFlow {
+  return new ReefMessageFlow({
+    config: config(),
+    trust: (params.trusted ?? trust({ alice: peerTrust(params.alice) })).store,
+    keys: params.bob,
+    transport: (params.relay ?? transport()) as unknown as ReefTransportClient,
+    guard: guard(allow),
+    audit: params.audit,
+    replay: new MemoryReplayStore(),
+    ...flowStores(),
+    onIngress: async () => {},
+    onOwnerNotice: params.onOwnerNotice ?? (async () => {}),
+  });
+}
+
+function createReceiptNotifier(
+  trusted: TrustedReef,
+  notify: ConstructorParameters<typeof ReefReceiptNotifier>[0],
+): ReefReceiptNotifier {
+  return new ReefReceiptNotifier(notify, {
+    loadState: (peer) => trusted.store.rejectionNoticeState(peer),
+    reserve: (rejection, noticeState) =>
+      trusted.store.reserveOutboundRejectionNotice(
+        rejection.peer,
+        rejection.id,
+        rejection.recipient,
+        noticeState,
+      ),
+    complete: (rejection, noticeState) => {
+      if (!trusted.store.completeOutboundRejection(rejection.peer, rejection.id, noticeState)) {
+        throw new Error(`missing rejection ${rejection.id}`);
+      }
+    },
+  });
+}
+
 describe("ReefMessageFlow delivery receipts", () => {
   it("quarantines an unmatched forged receipt without scanning audit history", async () => {
     const alice = generateIdentity();
     const bob = reefKeys();
     const audit = new MemoryAuditStore(new Uint8Array(32).fill(17));
     const entries = vi.spyOn(audit, "entries");
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trust({ alice: peerTrust(alice) }).store,
-      keys: bob,
-
-      transport: transport() as unknown as ReefTransportClient,
-      guard: guard(allow),
-      audit,
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
-      onOwnerNotice: async () => {},
-    });
+    const flow = createFlow({ alice, bob, audit });
     const id = "01JZ0000000000000000000130";
     const receipt = signReceipt(
       {
@@ -88,19 +122,7 @@ describe("ReefMessageFlow delivery receipts", () => {
       audit,
       policyVersion: "v1",
     });
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trusted.store,
-      keys: bob,
-
-      transport: transport() as unknown as ReefTransportClient,
-      guard: guard(allow),
-      audit,
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
-      onOwnerNotice: async () => {},
-    });
+    const flow = createFlow({ alice, bob, audit, trusted });
     const receipt = signReceipt(
       {
         id,
@@ -166,19 +188,7 @@ describe("ReefMessageFlow delivery receipts", () => {
       },
     ];
     vi.spyOn(audit, "entries").mockResolvedValueOnce(entries);
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trust({ alice: peerTrust(alice) }).store,
-      keys: bob,
-
-      transport: transport() as unknown as ReefTransportClient,
-      guard: guard(allow),
-      audit,
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
-      onOwnerNotice: async () => {},
-    });
+    const flow = createFlow({ alice, bob, audit });
     const receipt = signReceipt(
       {
         id,
@@ -223,19 +233,7 @@ describe("ReefMessageFlow delivery receipts", () => {
       },
     ];
     vi.spyOn(audit, "entries").mockResolvedValueOnce(entries);
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trust({ alice: peerTrust(alice) }).store,
-      keys: bob,
-
-      transport: transport() as unknown as ReefTransportClient,
-      guard: guard(allow),
-      audit,
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
-      onOwnerNotice: async () => {},
-    });
+    const flow = createFlow({ alice, bob, audit });
     const receipt = signReceipt(
       {
         id,
@@ -279,19 +277,7 @@ describe("ReefMessageFlow delivery receipts", () => {
     ];
     const auditEntries = vi.spyOn(audit, "entries").mockResolvedValueOnce(entries);
     const nowSpy = vi.spyOn(Date, "now").mockReturnValue(now);
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trusted.store,
-      keys: bob,
-
-      transport: transport() as unknown as ReefTransportClient,
-      guard: guard(allow),
-      audit,
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
-      onOwnerNotice: async () => {},
-    });
+    const flow = createFlow({ alice, bob, audit, trusted });
     const miss = signReceipt(
       {
         id: missId,
@@ -346,19 +332,7 @@ describe("ReefMessageFlow delivery receipts", () => {
       audit,
       policyVersion: "v1",
     });
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trusted.store,
-      keys: bob,
-
-      transport: transport() as unknown as ReefTransportClient,
-      guard: guard(allow),
-      audit,
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
-      onOwnerNotice: async () => {},
-    });
+    const flow = createFlow({ alice, bob, audit, trusted });
     const receipt = signReceipt(
       {
         id,
@@ -373,21 +347,7 @@ describe("ReefMessageFlow delivery receipts", () => {
       { seq: 1, peer: "alice", id, kind: "receipt", receipt, ts: 1 },
     ]);
     const notify = vi.fn(async () => {});
-    const receiptNotifier = new ReefReceiptNotifier(notify, {
-      loadState: (peer) => trusted.store.rejectionNoticeState(peer),
-      reserve: (rejection, noticeState) =>
-        trusted.store.reserveOutboundRejectionNotice(
-          rejection.peer,
-          rejection.id,
-          rejection.recipient,
-          noticeState,
-        ),
-      complete: (rejection, noticeState) => {
-        if (!trusted.store.completeOutboundRejection(rejection.peer, rejection.id, noticeState)) {
-          throw new Error(`missing rejection ${rejection.id}`);
-        }
-      },
-    });
+    const receiptNotifier = createReceiptNotifier(trusted, notify);
 
     await receiptNotifier.notifyRejections(rejections);
 
@@ -418,34 +378,8 @@ describe("ReefMessageFlow delivery receipts", () => {
     const relay = transport();
     const trusted = trust({ alice: peerTrust(alice) });
     const audit = new MemoryAuditStore(new Uint8Array(32).fill(11));
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trusted.store,
-      keys: bob,
-
-      transport: relay as unknown as ReefTransportClient,
-      guard: guard(allow),
-      audit,
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
-      onOwnerNotice: async () => {},
-    });
-    const receiptNotifier = new ReefReceiptNotifier(onOwnerNotice, {
-      loadState: (peer) => trusted.store.rejectionNoticeState(peer),
-      reserve: (rejection, noticeState) =>
-        trusted.store.reserveOutboundRejectionNotice(
-          rejection.peer,
-          rejection.id,
-          rejection.recipient,
-          noticeState,
-        ),
-      complete: (rejection, noticeState) => {
-        if (!trusted.store.completeOutboundRejection(rejection.peer, rejection.id, noticeState)) {
-          throw new Error(`missing rejection ${rejection.id}`);
-        }
-      },
-    });
+    const flow = createFlow({ alice, bob, audit, trusted, relay });
+    const receiptNotifier = createReceiptNotifier(trusted, onOwnerNotice);
     const id = await flow.send("alice", "ordinary coordination");
     const receipt = signReceipt(
       {
@@ -544,19 +478,7 @@ describe("ReefMessageFlow delivery receipts", () => {
     const trusted = trust({ alice: peerTrust(alice) });
     const audit = new MemoryAuditStore(new Uint8Array(32).fill(12));
     const auditEntries = vi.spyOn(audit, "entries");
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trusted.store,
-      keys: bob,
-
-      transport: transport() as unknown as ReefTransportClient,
-      guard: guard(allow),
-      audit,
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
-      onOwnerNotice: async () => {},
-    });
+    const flow = createFlow({ alice, bob, audit, trusted });
     const id = "01JZ0000000000000000000113";
     const receipt = signReceipt(
       {
@@ -608,19 +530,7 @@ describe("ReefMessageFlow delivery receipts", () => {
     const originalRecipient = reefPeerIdentity(originalTrust);
     const trusted = trust({ alice: originalTrust });
     const audit = new MemoryAuditStore(new Uint8Array(32).fill(14));
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trusted.store,
-      keys: bob,
-
-      transport: transport() as unknown as ReefTransportClient,
-      guard: guard(allow),
-      audit,
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
-      onOwnerNotice: async () => {},
-    });
+    const flow = createFlow({ alice, bob, audit, trusted });
     const id = await flow.send("alice", "expected body");
     const bodyHash = sha256Hex(canonicalBytes({ text: "expected body" }));
     trusted.values.set("alice", peerTrust(rotatedAlice, { keyEpoch: 2 }));
@@ -678,19 +588,7 @@ describe("ReefMessageFlow delivery receipts", () => {
     const bob = reefKeys();
     const trusted = trust({ alice: peerTrust(alice) });
     const audit = new MemoryAuditStore(new Uint8Array(32).fill(13));
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trusted.store,
-      keys: bob,
-
-      transport: transport() as unknown as ReefTransportClient,
-      guard: guard(allow),
-      audit,
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
-      onOwnerNotice: async () => {},
-    });
+    const flow = createFlow({ alice, bob, audit, trusted });
     const id = await flow.send("alice", "expected body");
     const receipt = signReceipt(
       {
@@ -779,16 +677,12 @@ describe("ReefMessageFlow overdue delivery follow-up", () => {
     const trusted = trust({ alice: peerTrust(alice) });
     const relay = transport();
     const onOwnerNotice = vi.fn(async (_text: string) => {});
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trusted.store,
-      keys: bob,
-      transport: relay as unknown as ReefTransportClient,
-      guard: guard(allow),
+    const flow = createFlow({
+      alice,
+      bob,
       audit: new MemoryAuditStore(new Uint8Array(32).fill(23)),
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
+      trusted,
+      relay,
       onOwnerNotice,
     });
     const id = await flow.send("alice", "are you there?");
@@ -814,16 +708,12 @@ describe("ReefMessageFlow overdue delivery follow-up", () => {
     const trusted = trust({ alice: peerTrust(alice) });
     const relay = transport();
     const onOwnerNotice = vi.fn(async (_text: string) => {});
-    const flow = new ReefMessageFlow({
-      config: config(),
-      trust: trusted.store,
-      keys: bob,
-      transport: relay as unknown as ReefTransportClient,
-      guard: guard(allow),
+    const flow = createFlow({
+      alice,
+      bob,
       audit: new MemoryAuditStore(new Uint8Array(32).fill(24)),
-      replay: new MemoryReplayStore(),
-      ...flowStores(),
-      onIngress: async () => {},
+      trusted,
+      relay,
       onOwnerNotice,
     });
     const id = await flow.send("alice", "quick ping");

extensions/reef/src/transport.test.ts

@@ -31,6 +31,14 @@ const keys: ReefKeys = {
   keyEpoch: 1,
 };
 
+function createClient(
+  fetcher: typeof fetch,
+  clock: () => number = () => ts,
+  baseUrl = "https://relay.example",
+): ReefTransportClient {
+  return new ReefTransportClient(baseUrl, "alice", keys, fetcher, clock);
+}
+
 afterEach(() => {
   vi.useRealTimers();
 });
@@ -54,7 +62,7 @@ describe("isRetryableReefRelayFailure", () => {
 describe("ReefTransportClient network failures", () => {
   it("normalizes fetch failures without swallowing the cause", async () => {
     const cause = new TypeError("fetch failed");
-    const client = new ReefTransportClient("https://relay.example", "alice", keys, async () => {
+    const client = createClient(async () => {
       throw cause;
     });
 
@@ -76,12 +84,7 @@ describe("ReefTransportClient network failures", () => {
         },
       }),
     );
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => response,
-    );
+    const client = createClient(async () => response);
 
     await expect(client.listFriends()).rejects.toMatchObject({
       name: "ReefRelayUnavailableError",
@@ -116,13 +119,7 @@ describe("ReefTransportClient device authentication", () => {
       calls.push([input, init]);
       return Response.json({ entries: [], cursor: 5 });
     };
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      fetcher,
-      () => ts,
-    );
+    const client = createClient(fetcher);
 
     await expect(client.pull(5)).resolves.toEqual({ entries: [], cursor: 5 });
 
@@ -151,13 +148,7 @@ describe("ReefTransportClient device authentication", () => {
   });
 
   it("puts WebSocket auth in the query but signs the bare relay path", () => {
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      vi.fn() as typeof fetch,
-      () => ts,
-    );
+    const client = createClient(vi.fn() as typeof fetch);
     const url = new URL(client.websocketUrl());
 
     expect(url.protocol).toBe("wss:");
@@ -183,13 +174,7 @@ describe("ReefTransportClient device authentication", () => {
       calls.push(init ?? {});
       return Response.json({ peer: "bob", status: "active" });
     };
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      fetcher,
-      () => ts,
-    );
+    const client = createClient(fetcher);
     const friend: RelayFriend = {
       peer: "bob",
       status: "pending",
@@ -219,13 +204,7 @@ describe("ReefTransportClient device authentication", () => {
       seenTs.push(new Headers(init?.headers).get("x-reef-ts")!);
       return Response.json({ friendships: [] });
     };
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      fetcher,
-      () => ts,
-    );
+    const client = createClient(fetcher);
 
     await client.listFriends();
     await client.listFriends();
@@ -290,13 +269,7 @@ describe("ReefTransportClient response body bounds", () => {
       }),
       { status: 200, headers: { "content-type": "application/json" } },
     );
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => response,
-      () => ts,
-    );
+    const client = createClient(async () => response);
 
     const result = await client.pull(0);
     const pad = (result as unknown as { pad: string }).pad;
@@ -317,13 +290,7 @@ describe("ReefTransportClient response body bounds", () => {
         new Uint8Array(1024).fill(0x78),
       ],
     });
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => offered.response,
-      () => ts,
-    );
+    const client = createClient(async () => offered.response);
 
     await expect(client.pull(0)).rejects.toThrow(
       /reef\.relay: JSON response exceeds 16777216 bytes/,
@@ -335,13 +302,7 @@ describe("ReefTransportClient response body bounds", () => {
 
   it("surfaces relay error JSON exactly at the error byte limit", async () => {
     const body = jsonObjectBodyAtSize(ERROR_RESPONSE_MAX_BYTES, "error");
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => new Response(body, { status: 400 }),
-      () => ts,
-    );
+    const client = createClient(async () => new Response(body, { status: 400 }));
 
     const error = await client.requestFriend("bob", "code").catch((cause: unknown) => cause);
     expect(error).toBeInstanceOf(ReefRelayError);
@@ -355,13 +316,7 @@ describe("ReefTransportClient response body bounds", () => {
       status: 503,
       chunks: Array.from({ length: 16 }, () => new Uint8Array(8 * 1024).fill(0x78)),
     });
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => offered.response,
-      () => ts,
-    );
+    const client = createClient(async () => offered.response);
 
     await expect(client.listFriends()).rejects.toMatchObject({
       name: "ReefRelayError",
@@ -374,13 +329,7 @@ describe("ReefTransportClient response body bounds", () => {
   });
 
   it("keeps the typed status fallback for malformed error JSON", async () => {
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => new Response("{", { status: 502 }),
-      () => ts,
-    );
+    const client = createClient(async () => new Response("{", { status: 502 }));
 
     await expect(client.requestFriend("bob", "code")).rejects.toMatchObject({
       name: "ReefRelayError",
@@ -440,19 +389,13 @@ describe("ReefInboxConnection recovery", () => {
     const requestedAfter: number[] = [];
     const persisted: number[] = [];
     const processed: number[] = [];
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async (input) => {
-        const after = Number(parseRequestUrl(input).searchParams.get("after"));
-        requestedAfter.push(after);
-        return after === 7
-          ? Response.json({ entries: [receiptEntry(8)], cursor: 8 })
-          : Response.json({ entries: [], cursor: after });
-      },
-      () => ts,
-    );
+    const client = createClient(async (input) => {
+      const after = Number(parseRequestUrl(input).searchParams.get("after"));
+      requestedAfter.push(after);
+      return after === 7
+        ? Response.json({ entries: [receiptEntry(8)], cursor: 8 })
+        : Response.json({ entries: [], cursor: after });
+    });
     const inbox = new ReefInboxConnection(
       client,
       async (entries) => {
@@ -473,12 +416,8 @@ describe("ReefInboxConnection recovery", () => {
 
   it("does not advance past an entry that failed processing", async () => {
     const persisted: number[] = [];
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => Response.json({ entries: [receiptEntry(8), receiptEntry(9)], cursor: 9 }),
-      () => ts,
+    const client = createClient(async () =>
+      Response.json({ entries: [receiptEntry(8), receiptEntry(9)], cursor: 9 }),
     );
     const inbox = new ReefInboxConnection(
       client,
@@ -500,12 +439,8 @@ describe("ReefInboxConnection recovery", () => {
   it("rejects an inconsistent REST page before dispatch or persistence", async () => {
     const processed: number[] = [];
     const persisted: number[] = [];
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => Response.json({ entries: [receiptEntry(9)], cursor: 8 }),
-      () => ts,
+    const client = createClient(async () =>
+      Response.json({ entries: [receiptEntry(9)], cursor: 8 }),
     );
     const inbox = new ReefInboxConnection(
       client,
@@ -526,16 +461,10 @@ describe("ReefInboxConnection recovery", () => {
   it("persists cursor-only progress when retained entries have expired", async () => {
     const requestedAfter: number[] = [];
     const persisted: number[] = [];
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async (input) => {
-        requestedAfter.push(Number(parseRequestUrl(input).searchParams.get("after")));
-        return Response.json({ entries: [], cursor: 12 });
-      },
-      () => ts,
-    );
+    const client = createClient(async (input) => {
+      requestedAfter.push(Number(parseRequestUrl(input).searchParams.get("after")));
+      return Response.json({ entries: [], cursor: 12 });
+    });
     const inbox = new ReefInboxConnection(
       client,
       async () => {},
@@ -560,17 +489,11 @@ describe("ReefInboxConnection recovery", () => {
       releasePull = resolve;
     });
     let pullStarted = false;
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => {
-        pullStarted = true;
-        await pullGate;
-        return Response.json({ entries: [], cursor: 0 });
-      },
-      () => ts,
-    );
+    const client = createClient(async () => {
+      pullStarted = true;
+      await pullGate;
+      return Response.json({ entries: [], cursor: 0 });
+    });
     const abort = new AbortController();
     const inbox = new ReefInboxConnection(
       client,
@@ -598,20 +521,14 @@ describe("ReefInboxConnection recovery", () => {
     const firstPullGate = new Promise<void>((resolve) => {
       releaseFirstPull = resolve;
     });
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async (input) => {
-        const after = Number(parseRequestUrl(input).searchParams.get("after"));
-        if (after === 0) {
-          await firstPullGate;
-          return Response.json({ entries: [receiptEntry(1), receiptEntry(2)], cursor: 2 });
-        }
-        return Response.json({ entries: [], cursor: after });
-      },
-      () => ts,
-    );
+    const client = createClient(async (input) => {
+      const after = Number(parseRequestUrl(input).searchParams.get("after"));
+      if (after === 0) {
+        await firstPullGate;
+        return Response.json({ entries: [receiptEntry(1), receiptEntry(2)], cursor: 2 });
+      }
+      return Response.json({ entries: [], cursor: after });
+    });
     const abort = new AbortController();
     const inbox = new ReefInboxConnection(
       client,
@@ -642,17 +559,11 @@ describe("ReefInboxConnection recovery", () => {
       releasePull = resolve;
     });
     let pullStarted = false;
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => {
-        pullStarted = true;
-        await pullGate;
-        return Response.json({ entries: [], cursor: 0 });
-      },
-      () => ts,
-    );
+    const client = createClient(async () => {
+      pullStarted = true;
+      await pullGate;
+      return Response.json({ entries: [], cursor: 0 });
+    });
     const abort = new AbortController();
     const inbox = new ReefInboxConnection(
       client,
@@ -676,13 +587,7 @@ describe("ReefInboxConnection recovery", () => {
     const socket = new ControlledSocket();
     const states: string[] = [];
     const errors: string[] = [];
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => Response.json({ entries: [], cursor: 0 }),
-      () => ts,
-    );
+    const client = createClient(async () => Response.json({ entries: [], cursor: 0 }));
     const abort = new AbortController();
     const inbox = new ReefInboxConnection(
       client,
@@ -710,13 +615,7 @@ describe("ReefInboxConnection recovery", () => {
     vi.useFakeTimers();
     const sockets: ControlledSocket[] = [];
     const persisted: number[] = [];
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => Response.json({ entries: [], cursor: 1 }),
-      () => ts,
-    );
+    const client = createClient(async () => Response.json({ entries: [], cursor: 1 }));
     const abort = new AbortController();
     const inbox = new ReefInboxConnection(
       client,
@@ -756,12 +655,8 @@ describe("ReefInboxConnection recovery", () => {
       releaseHandler = resolve;
     });
     let handlerStarted = false;
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => Response.json({ entries: [receiptEntry(1)], cursor: 1 }),
-      () => ts,
+    const client = createClient(async () =>
+      Response.json({ entries: [receiptEntry(1)], cursor: 1 }),
     );
     const abort = new AbortController();
     const inbox = new ReefInboxConnection(
@@ -797,17 +692,11 @@ describe("ReefInboxConnection recovery", () => {
       releasePull = resolve;
     });
     let pullStarted = false;
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => {
-        pullStarted = true;
-        await pullGate;
-        return Response.json({ entries: [], cursor: 0 });
-      },
-      () => ts,
-    );
+    const client = createClient(async () => {
+      pullStarted = true;
+      await pullGate;
+      return Response.json({ entries: [], cursor: 0 });
+    });
     const abort = new AbortController();
     const inbox = new ReefInboxConnection(
       client,
@@ -840,15 +729,9 @@ describe("ReefInboxConnection recovery", () => {
     const states: string[] = [];
     const errors: string[] = [];
     const abort = new AbortController();
-    const client = new ReefTransportClient(
-      "https://relay.example",
-      "alice",
-      keys,
-      async () => {
-        throw new Error("relay catch-up failed");
-      },
-      () => ts,
-    );
+    const client = createClient(async () => {
+      throw new Error("relay catch-up failed");
+    });
     const inbox = new ReefInboxConnection(
       client,
       async () => {},
@@ -893,12 +776,10 @@ async function deliverInboxFrame(frame: string): Promise<{
   const abort = new AbortController();
   const timeout = setTimeout(() => abort.abort(), 2_000);
   server.once("connection", (socket) => socket.send(frame));
-  const client = new ReefTransportClient(
-    `http://127.0.0.1:${address.port}`,
-    "alice",
-    keys,
+  const client = createClient(
     async () => Response.json({ entries: [], cursor: 0 }),
     () => ts,
+    `http://127.0.0.1:${address.port}`,
   );
   const inbox = new ReefInboxConnection(
     client,