diff --git a/extensions/reef/protocol/pipeline.test.ts b/extensions/reef/protocol/pipeline.test.ts index 06ed20d623d0..7bce9a45a419 100644 --- a/extensions/reef/protocol/pipeline.test.ts +++ b/extensions/reef/protocol/pipeline.test.ts @@ -55,6 +55,88 @@ function identities() { return { alice: generateIdentity(), bob: generateIdentity() }; } +type Identity = ReturnType; +type OutboundOptions = Parameters[0]; +type InboundOptions = Parameters[0]; + +function outboundOptions( + alice: Identity, + bob: Identity, + overrides: Partial = {}, +): 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[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 { + 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): Promise { + 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" }, diff --git a/extensions/reef/src/flow-receipts.test.ts b/extensions/reef/src/flow-receipts.test.ts index 565db2cb92c4..aeac636f478e 100644 --- a/extensions/reef/src/flow-receipts.test.ts +++ b/extensions/reef/src/flow-receipts.test.ts @@ -33,25 +33,59 @@ import type { InboxEntry } from "./types.js"; beforeEach(resetFlowStoresForTests); afterEach(resetFlowStoresForTests); +type FlowOptions = ConstructorParameters[0]; +type TrustedReef = ReturnType; + +function createFlow(params: { + alice: ReturnType; + bob: ReturnType; + audit: MemoryAuditStore; + trusted?: TrustedReef; + relay?: ReturnType; + 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[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"); diff --git a/extensions/reef/src/transport.test.ts b/extensions/reef/src/transport.test.ts index a7e09b1d1ae7..ee15ce82b095 100644 --- a/extensions/reef/src/transport.test.ts +++ b/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((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,