From 33c31b972cfa30a9959fd79ad1504d8724435704 Mon Sep 17 00:00:00 2001 From: Khushvendra Date: Thu, 15 Oct 2026 00:42:38 +0530 Subject: [PATCH 1/2] fix(payments): harden client payment attempts and retries --- src/__mocks__/client-payments-harness.ts | 143 ++ src/payments/client-payment-attempts.ts | 89 ++ src/payments/client-payment-lifecycle.ts | 218 +++ src/payments/client-payment-options.ts | 137 ++ src/payments/client-payment-progress.ts | 56 + src/payments/client-payment-validation.ts | 144 ++ src/payments/client-payments-limits.test.ts | 334 +++++ .../client-payments-reliability.test.ts | 597 ++++++++ src/payments/client-payments.test.ts | 20 +- src/payments/client-payments.ts | 1266 +++++++---------- src/payments/types.ts | 2 + src/transport/nostr-client-transport.ts | 11 +- .../nostr-client/correlation-store.test.ts | 21 + .../nostr-client/correlation-store.ts | 8 + .../payment-client-reliability.e2e.test.ts | 149 ++ src/transport/payments-flow.test.ts | 53 +- 16 files changed, 2438 insertions(+), 810 deletions(-) create mode 100644 src/__mocks__/client-payments-harness.ts create mode 100644 src/payments/client-payment-attempts.ts create mode 100644 src/payments/client-payment-lifecycle.ts create mode 100644 src/payments/client-payment-options.ts create mode 100644 src/payments/client-payment-progress.ts create mode 100644 src/payments/client-payment-validation.ts create mode 100644 src/payments/client-payments-limits.test.ts create mode 100644 src/payments/client-payments-reliability.test.ts create mode 100644 src/transport/payment-client-reliability.e2e.test.ts diff --git a/src/__mocks__/client-payments-harness.ts b/src/__mocks__/client-payments-harness.ts new file mode 100644 index 0000000..c3a68f2 --- /dev/null +++ b/src/__mocks__/client-payments-harness.ts @@ -0,0 +1,143 @@ +import type { Transport } from '@contextvm/mcp-sdk/shared/transport'; +import type { + JSONRPCMessage, + JSONRPCRequest, +} from '@contextvm/mcp-sdk/types.js'; +import { + withClientPayments, + type ClientPaymentsOptions, +} from '../payments/client-payments.js'; +import type { PaymentOption } from '../payments/types.js'; +import { waitFor as waitForValue } from '../core/utils/test.utils.js'; + +/** Coordinates isolated test work without changing the global clock. */ +export async function waitFor( + condition: () => boolean, + options: { timeoutMs?: number } = {}, +): Promise { + await waitForValue({ + produce: () => (condition() ? true : undefined), + timeoutMs: options.timeoutMs ?? 1000, + intervalMs: 2, + }); +} + +/** Builds or advances isolated payment test state. */ +export const tick = (): Promise => + new Promise((resolve) => setTimeout(resolve, 0)); +/** Builds or advances isolated payment test state. */ +export const wait = (ms: number): Promise => + new Promise((resolve) => setTimeout(resolve, ms)); +/** Builds or advances isolated payment test state. */ +export const request = (id: string | number = 1): JSONRPCRequest => ({ + jsonrpc: '2.0', + id, + method: 'tools/call', + params: { name: 'paid', arguments: { value: id } }, +}); +/** Builds or advances isolated payment test state. */ +export const option = ( + pay_req = 'invoice', + amount = 5, + pmi = 'fake', +): PaymentOption => ({ amount, pmi, pay_req }); +/** Builds or advances isolated payment test state. */ +export const offer = (input: unknown = option()): JSONRPCMessage => + ({ + jsonrpc: '2.0', + method: 'notifications/payment_required', + params: input, + }) as JSONRPCMessage; +/** Builds or advances isolated payment test state. */ +export const required = ( + id: string | number = 1, + options: unknown = [option()], +): JSONRPCMessage => ({ + jsonrpc: '2.0', + id, + error: { + code: -32042, + message: 'Payment Required', + data: { payment_options: options }, + }, +}); +/** Builds or advances isolated payment test state. */ +export const pending = ( + id: string | number = 1, + retry_after?: unknown, +): JSONRPCMessage => ({ + jsonrpc: '2.0', + id, + error: { + code: -32043, + message: 'Payment Pending', + data: retry_after === undefined ? {} : { retry_after }, + }, +}); +/** Lets a test decide when an external payment action completes. */ +export function deferred(): { + promise: Promise; + resolve: (value: T) => void; + reject: (error: Error) => void; +} { + let resolve!: (value: T) => void; + let reject!: (error: Error) => void; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + return { promise, resolve, reject }; +} +/** One transport, its wrapper, and all observable test outcomes. */ +export interface Harness { + paid: Transport; + base: Transport; + sent: JSONRPCMessage[]; + received: JSONRPCMessage[]; + errors: Error[]; + deliver: (message: JSONRPCMessage) => void; +} +/** Coordinates isolated test work without changing the global clock. */ +export async function run( + options: ClientPaymentsOptions, + body: (harness: Harness) => Promise, +): Promise { + const sent: JSONRPCMessage[] = []; + const received: JSONRPCMessage[] = []; + const errors: Error[] = []; + const base: Transport = { + async start(): Promise {}, + async send(message): Promise { + sent.push(message); + }, + async close(): Promise {}, // Deliberately does not fire onclose. + }; + const paid = withClientPayments(base, { minRetryDelayMs: 2, ...options }); + paid.onmessage = (message): void => { + received.push(message); + }; + paid.onerror = (error): void => { + errors.push(error); + }; + await paid.start(); + try { + await body({ + paid, + base, + sent, + received, + errors, + deliver: (message): void => base.onmessage?.(message), + }); + } finally { + await paid.close(); + } +} +/** Collects terminal errors observed by the test caller. */ +export function failures( + harness: Harness, +): Array<{ error: { data?: unknown } }> { + return harness.received.filter((message) => 'error' in message) as Array<{ + error: { data?: unknown }; + }>; +} diff --git a/src/payments/client-payment-attempts.ts b/src/payments/client-payment-attempts.ts new file mode 100644 index 0000000..a8ac981 --- /dev/null +++ b/src/payments/client-payment-attempts.ts @@ -0,0 +1,89 @@ +import type { PaymentOption } from './types.js'; +import { ClientPaymentError } from './client-payment-validation.js'; + +type AttemptOutcome = 'in_flight' | 'handled' | 'declined' | 'unknown'; + +/** A local payment action; handled means wallet callback returned, not settlement. */ +export interface ClientPaymentAttempt { + outcome: AttemptOutcome; + completion: Promise; + complete: (outcome: Exclude) => void; +} + +/** + * Wrapper-scoped attempt history. A Nostr wrapper is bound to one authenticated + * server; generic wrappers use their transport instance as the peer boundary. + * No live/completed/unknown entry is evicted and silently made payable again. + */ +export class ClientPaymentAttempts { + private readonly entries = new Map< + string, + { amount: number; attempt: ClientPaymentAttempt } + >(); + + public constructor(private readonly maxEntries: number) {} + + private key(option: PaymentOption): string { + return JSON.stringify([option.pmi, option.pay_req]); + } + + /** Finds a prior action without reserving or paying any new option. */ + public find( + options: readonly PaymentOption[], + ): ClientPaymentAttempt | undefined { + let found: ClientPaymentAttempt | undefined; + for (const option of options) { + const prior = this.entries.get(this.key(option)); + if (prior) { + if (prior.amount !== option.amount) { + throw new ClientPaymentError( + 'payment_offer_changed', + 'Payment amount changed for an existing payment identity', + ); + } + if ( + !found || + (found.outcome !== 'unknown' && + (prior.attempt.outcome === 'unknown' || + prior.attempt.outcome === 'in_flight')) + ) + found = prior.attempt; + } + } + return found; + } + + /** Atomically reserves all selectable options before external approval work. */ + public reserve( + options: readonly PaymentOption[], + signal: AbortSignal, + ): ClientPaymentAttempt { + const unique = new Map(options.map((option) => [this.key(option), option])); + if (this.entries.size + unique.size > this.maxEntries) { + throw new ClientPaymentError( + 'payment_attempt_capacity', + 'Client payment attempt capacity reached', + ); + } + let resolve!: () => void; + const attempt: ClientPaymentAttempt = { + outcome: 'in_flight', + completion: new Promise((done) => { + resolve = done; + }), + complete: (outcome): void => { + if (attempt.outcome !== 'in_flight') return; + attempt.outcome = outcome; + signal.removeEventListener('abort', onAbort); + resolve(); + }, + }; + const onAbort = (): void => attempt.complete('unknown'); + for (const [key, option] of unique) { + this.entries.set(key, { amount: option.amount, attempt }); + } + signal.addEventListener('abort', onAbort, { once: true }); + if (signal.aborted) onAbort(); + return attempt; + } +} diff --git a/src/payments/client-payment-lifecycle.ts b/src/payments/client-payment-lifecycle.ts new file mode 100644 index 0000000..ccc4cd2 --- /dev/null +++ b/src/payments/client-payment-lifecycle.ts @@ -0,0 +1,218 @@ +import type { JSONRPCRequest } from '@contextvm/mcp-sdk/types.js'; +import { LruCache } from '../core/utils/lru-cache.js'; +import type { ClientPaymentLimits } from './client-payment-options.js'; +import type { PaymentOption } from './types.js'; +import { ClientPaymentError } from './client-payment-validation.js'; + +/** One caller request generation, including its automatic payment retries. */ +export interface ClientPaymentRequest { + key: string; + id?: string | number; + request?: JSONRPCRequest; + controller: AbortController; + active: boolean; + retries: number; + attempts: number; + paymentInFlight: boolean; + amounts: Map; + deadline?: number; + deadlineTimer?: ReturnType; + retryTimer?: ReturnType; +} + +/** Owns request generations, deadlines, limits, and retry timer cancellation. */ +export class ClientPaymentLifecycle { + private readonly requests: LruCache; + private closed = false; + + public constructor( + private readonly limits: ClientPaymentLimits | undefined, + private readonly onFailure: ( + request: ClientPaymentRequest, + error: ClientPaymentError, + ) => void, + private readonly onFinish: (request: ClientPaymentRequest) => void, + ) { + this.requests = new LruCache(1000, (_key, request) => { + if (!request.active) return; + this.finish(request); + this.onFailure( + request, + new ClientPaymentError( + 'payment_request_capacity', + 'Client request capacity reached', + ), + ); + }); + } + + /** Makes future generations usable; closed generations remain inactive. */ + public open(): void { + this.closed = false; + } + + /** Whether the wrapper has begun closing. */ + public get isClosed(): boolean { + return this.closed; + } + + /** Finds the current generation using a type-aware JSON-RPC identifier. */ + public get(id: string | number): ClientPaymentRequest | undefined { + return this.requests.get(JSON.stringify(['request', id])); + } + + private create( + key: string, + id?: string | number, + request?: JSONRPCRequest, + ): ClientPaymentRequest { + const prior = this.requests.get(key); + if (prior) this.finish(prior); + const state: ClientPaymentRequest = { + key, + id, + request, + active: true, + controller: new AbortController(), + retries: 0, + attempts: 0, + paymentInFlight: false, + amounts: new Map(), + }; + this.requests.set(key, state); + return state; + } + + /** Records the wire-shaped request before publishing and isolates it from callbacks. */ + public register(request: JSONRPCRequest): ClientPaymentRequest { + if (this.closed) throw new Error('Client payments transport is closed'); + const copy = JSON.parse(JSON.stringify(request)) as JSONRPCRequest; + return this.create( + JSON.stringify(['request', request.id]), + request.id, + copy, + ); + } + + /** Adopts correlation supplied by a transport when no original request was cached. */ + public correlated(id: string | number): ClientPaymentRequest { + return this.get(id) ?? this.create(JSON.stringify(['request', id]), id); + } + + /** Returns the sole live invocation for an uncorrelated generic transport. */ + public onlyRequest(): ClientPaymentRequest | undefined { + const active = [...this.requests.entries()] + .map(([, state]) => state) + .filter((state) => state.active && state.request); + return active.length === 1 ? active[0] : undefined; + } + + /** Whether a generic notification is ambiguous between multiple invocations. */ + public hasRequests(): boolean { + return [...this.requests.entries()].some( + ([, state]) => state.active && state.request !== undefined, + ); + } + + /** Tracks legacy standalone notifications in the generic transport's scope. */ + public detached(eventId: string): ClientPaymentRequest { + const key = JSON.stringify(['event', eventId]); + return this.requests.get(key) ?? this.create(key); + } + + /** Checks generation ownership and expires an elapsed payment deadline. */ + public isActive(state: ClientPaymentRequest): boolean { + if (this.closed || !state.active || this.requests.get(state.key) !== state) + return false; + if (state.deadline !== undefined && Date.now() >= state.deadline) { + this.expire(state); + return false; + } + return true; + } + + private expire(state: ClientPaymentRequest): void { + if (!state.active) return; + this.finish(state); + this.onFailure( + state, + new ClientPaymentError( + 'payment_deadline_exceeded', + 'Client payment deadline exceeded', + ), + ); + } + + /** Starts one deadline spanning every offer and pending response for this invocation. */ + public startPayment(state: ClientPaymentRequest): void { + if (this.limits?.timeoutMs === undefined || state.deadline !== undefined) + return; + state.deadline = Date.now() + this.limits.timeoutMs; + state.deadlineTimer = setTimeout( + () => this.expire(state), + this.limits.timeoutMs, + ); + } + + /** Reserves a conservative per-PMI approval budget before external code runs. */ + public approve( + state: ClientPaymentRequest, + options: readonly PaymentOption[], + ): PaymentOption[] { + const limits = this.limits; + if ( + limits?.maxAttempts !== undefined && + state.attempts >= limits.maxAttempts + ) { + throw new ClientPaymentError( + 'payment_limit_exceeded', + 'Client payment attempt limit exceeded', + ); + } + const allowed = options.filter((option) => { + if (limits?.maxAmountByPmi === undefined) return true; + const bound = limits.maxAmountByPmi[option.pmi]; + return ( + bound !== undefined && + option.amount <= bound - (state.amounts.get(option.pmi) ?? 0) + ); + }); + if (allowed.length === 0) { + throw new ClientPaymentError( + 'payment_limit_exceeded', + 'Client payment amount limit exceeded', + ); + } + const reserved = new Map(); + for (const option of allowed) + reserved.set( + option.pmi, + Math.max(reserved.get(option.pmi) ?? 0, option.amount), + ); + for (const [pmi, amount] of reserved) + state.amounts.set(pmi, (state.amounts.get(pmi) ?? 0) + amount); + state.attempts++; + return allowed; + } + + /** Reaps request-owned timers and signals before any terminal callback is delivered. */ + public finish(state: ClientPaymentRequest): void { + if (!state.active) return; + state.active = false; + clearTimeout(state.retryTimer); + clearTimeout(state.deadlineTimer); + state.retryTimer = undefined; + state.controller.abort(); + // Retain an inactive generation so late offers cannot revive a terminal id. + state.request = undefined; + state.amounts.clear(); + this.onFinish(state); + } + + /** Stops every current generation synchronously, even when transport close awaits I/O. */ + public close(): void { + this.closed = true; + for (const [, state] of [...this.requests.entries()]) this.finish(state); + this.requests.clear(); + } +} diff --git a/src/payments/client-payment-options.ts b/src/payments/client-payment-options.ts new file mode 100644 index 0000000..05b5d41 --- /dev/null +++ b/src/payments/client-payment-options.ts @@ -0,0 +1,137 @@ +import type { JSONRPCRequest } from '@contextvm/mcp-sdk/types.js'; +import type { OriginalRequestContext } from '../transport/nostr-client/correlation-store.js'; +import type { + PaymentHandler, + PaymentHandlerRequest, + PaymentInteractionMode, + PaymentOption, +} from './types.js'; + +/** Limits for one request lifecycle, including replacement invoices. */ +export interface ClientPaymentLimits { + /** + * Cumulative advertised amount approved per PMI, in that PMI's amount unit. + * When set, unlisted PMIs cannot be paid automatically. Multi-option explicit + * callbacks reserve the largest offered amount for each PMI conservatively. + * Wallet handlers must verify the settlement payload amount; this generic + * policy cannot decode or constrain arbitrary PMI payloads or external wallets. + */ + maxAmountByPmi?: Readonly>; + /** Maximum distinct offers passed to policy/approval work. Zero refuses payment. */ + maxAttempts?: number; + /** Payment deadline from the first offer/pending response, in milliseconds. */ + timeoutMs?: number; +} + +/** Client-side CEP-8 payment handlers, approval policy, and lifecycle limits. */ +export interface ClientPaymentsOptions { + /** + * Payment handlers for in-band (programmatic) payment, indexed by their + * {@link PaymentHandler.pmi}. Each handler's `pmi` is advertised to the + * server so it can pick a matching rail (the wallet-client fast path). + * + * **Omit entirely** for a PMI-agnostic client that pays out-of-band: no PMIs + * are advertised, so per CEP-8 the server sends `payment_required` for any of + * its processors. The notification is forwarded to the application via + * `onmessage`, synthetic progress keeps the original MCP request alive while + * the payment settles out-of-band, and the server's TTL is the timeout. + */ + handlers?: readonly PaymentHandler[]; + /** Optional limits across all automatic payment offers for one invocation. */ + limits?: ClientPaymentLimits; + /** + * Maximum distinct payment identities retained by this wrapper. Completed and + * uncertain attempts are never evicted to make room for a new payment. + * At capacity, new attempts fail without invoking a wallet. @default 1000 + * Explicit callbacks reserve every offered identity because option selection + * is opaque; they must pay only one of the supplied options. + */ + maxTrackedPaymentAttempts?: number; + /** + * Interval for periodic synthetic progress heartbeats (milliseconds). + * + * One heartbeat is also emitted immediately when `payment_required` is + * received so the MCP timeout is always reset as soon as the payment flow + * begins. + * @default DEFAULT_SYNTHETIC_PROGRESS_INTERVAL_MS (30_000 ms) + */ + syntheticProgressIntervalMs?: number; + /** + * Duration in ms for synthetic progress when `payment_required` carries no `ttl`. + * + * Mirrors the server-side `paymentTtlMs` default so the client keeps the MCP + * request alive for at least as long as the server will wait. + * @default DEFAULT_PAYMENT_TTL_MS (300_000 ms) + */ + defaultPaymentTtlMs?: number; + /** + * Minimum delay before an explicit-payment retry is re-sent + * (milliseconds), applied after the server-provided `retry_after` backoff. + * + * Guards against `retry_after: 0`: an immediate retry within the same second + * can produce a byte-identical Nostr event (same content, same tags, same + * `created_at` at second resolution), which relays and servers swallow as a + * duplicate — silently losing the retry. + * Nostr transports always enforce at least 1000 ms, including paid retries. + * Shorter delays are supported only by other transports. + * @default 1000 + */ + minRetryDelayMs?: number; + + /** + * Optional policy hook invoked when a `payment_required` notification is received. + * + * This is evaluated before checking rail-specific `canHandle` methods. + * + * When the underlying transport supports request correlation (e.g. {@link NostrClientTransport}), + * `originalRequestContext` provides minimal details about the original JSON-RPC request that + * triggered the payment. + */ + paymentPolicy?: ( + req: PaymentHandlerRequest, + originalRequestContext?: OriginalRequestContext, + ) => boolean | Promise; + + /** Requested payment interaction mode. @default 'transparent' */ + paymentInteraction?: PaymentInteractionMode; + + /** + * Maximum number of -32043 (Payment Pending) retries before giving up. + * + * With retry_after=2 and 1.5× exponential backoff capped at 10s, the default + * of 10 retries gives ~76s of cumulative wait — enough for typical verification + * flows. Increase for slow payment processors (e.g. on-chain confirmation). + * @default 10 + */ + maxPendingRetries?: number; + + /** + * Handler for explicit-gating -32042 errors. + * Called when a priced invocation returns Payment Required. + * The handler should pay one option and signal completion. + * + * **Error handling contract**: + * - If the promise resolves with `{ paid: true }`, the wrapper auto-retries the + * original request with the same `method` and `params`. + * - If the promise resolves with `{ paid: false, reason }`, the wrapper synthesizes + * a JSON-RPC error to the caller with code `-32042` and `data: { reason }`. + * Use `reason: 'user_cancelled'` for user-initiated cancellations. + * - If the promise **rejects**, the wrapper MUST NOT silently fall back. + * It synthesizes a JSON-RPC error with code `-32042` and + * `data: { reason: error.message, type: 'payment_handler_error' }`. + * - Transient payment-provider failures should reject with an Error whose + * `message` contains the provider error details. + * + * **Verify-timeout window**: if the server's verification times out or fails + * after the client paid, its pending state is cleared and the client's retry + * can receive a fresh `-32042` with a new invoice. Distinct invoices still + * pass through the invocation limits; an uncertain wallet outcome is terminal. + */ + onPaymentRequired?: (params: { + options: PaymentOption[]; + instructions?: string; + originalRequest: JSONRPCRequest; + /** Aborted on completion, cancellation, deadline, or close. Check before paying. */ + abortSignal: AbortSignal; + }) => Promise<{ paid: boolean; reason?: string }>; +} diff --git a/src/payments/client-payment-progress.ts b/src/payments/client-payment-progress.ts new file mode 100644 index 0000000..71bc2ab --- /dev/null +++ b/src/payments/client-payment-progress.ts @@ -0,0 +1,56 @@ +import type { JSONRPCMessage } from '@contextvm/mcp-sdk/types.js'; + +/** Maintains bounded synthetic MCP progress while transparent settlement is pending. */ +export class ClientPaymentProgress { + private readonly entries = new Map< + string, + { token: string | number; stopAt: number } + >(); + private timer?: ReturnType; + + public constructor( + private readonly intervalMs: number, + private readonly emit: (message: JSONRPCMessage) => void, + ) {} + + private tick(): void { + for (const [key, entry] of this.entries) { + if (Date.now() >= entry.stopAt) this.stop(key); + else this.heartbeat(entry.token); + } + } + + private heartbeat(token: string | number): void { + this.emit({ + jsonrpc: '2.0', + method: 'notifications/progress', + params: { progressToken: token, progress: 0 }, + }); + } + + /** Starts an immediate heartbeat and periodic resets for the request's wire token. */ + public start(key: string, token: string, ttlMs: number): void { + if (this.entries.has(key)) return; + const wireToken = Number.isFinite(Number(token)) ? Number(token) : token; + this.entries.set(key, { token: wireToken, stopAt: Date.now() + ttlMs }); + if (!this.timer) + this.timer = setInterval(() => this.tick(), this.intervalMs); + this.heartbeat(wireToken); + } + + /** Stops only this generation's progress and reaps an idle scheduler. */ + public stop(key: string): void { + this.entries.delete(key); + if (this.entries.size === 0 && this.timer) { + clearInterval(this.timer); + this.timer = undefined; + } + } + + /** Reaps every heartbeat during wrapper teardown. */ + public clear(): void { + this.entries.clear(); + clearInterval(this.timer); + this.timer = undefined; + } +} diff --git a/src/payments/client-payment-validation.ts b/src/payments/client-payment-validation.ts new file mode 100644 index 0000000..e86c555 --- /dev/null +++ b/src/payments/client-payment-validation.ts @@ -0,0 +1,144 @@ +import { z } from 'zod'; +import type { ClientPaymentsOptions } from './client-payment-options.js'; +import type { + PaymentOption, + PaymentPendingErrorData, + PaymentRequiredErrorData, +} from './types.js'; + +const MAX_TIMER_MS = 2_147_483_647; +const amountSchema = z + .number() + .finite() + .nonnegative() + .max(Number.MAX_SAFE_INTEGER); +const pmiSchema = z.string().regex(/^[a-z0-9-]+$/); +const timerSchema = z.number().finite().positive().max(MAX_TIMER_MS); + +/** Internal payment failure with a stable application-visible reason. */ +export class ClientPaymentError extends Error { + public constructor( + public readonly reason: string, + message: string, + ) { + super(message); + this.name = 'ClientPaymentError'; + } +} + +/** Runtime offer schema; unknown extension metadata is preserved. */ +export const clientPaymentOptionSchema = z.object({ + amount: amountSchema, + pmi: pmiSchema, + pay_req: z + .string() + .min(1) + .refine((value) => value.trim().length > 0), + description: z.string().optional(), + ttl: z + .number() + .finite() + .positive() + .max(Number.MAX_SAFE_INTEGER / 1000) + .optional(), + _meta: z.record(z.string(), z.unknown()).optional(), +}); + +const requiredSchema = z.object({ + instructions: z.string().optional(), + payment_options: z + .array(clientPaymentOptionSchema) + .min(1) + .max(100) + .refine((options) => { + const amounts = new Map(); + for (const option of options) { + const key = JSON.stringify([option.pmi, option.pay_req]); + if (amounts.has(key) && amounts.get(key) !== option.amount) + return false; + amounts.set(key, option.amount); + } + return true; + }), +}); +const pendingSchema = z.object({ + instructions: z.string().optional(), + retry_after: z + .number() + .finite() + .nonnegative() + .max(Number.MAX_SAFE_INTEGER / 1000) + .optional(), +}); + +function parse(schema: z.ZodType, input: unknown): T { + const result = schema.safeParse(input); + if (!result.success) { + throw new ClientPaymentError('invalid_payment', 'Invalid payment message'); + } + return result.data; +} + +/** Validates a transparent payment offer before invoking policy or wallet code. */ +export function parseClientPaymentOption(input: unknown): PaymentOption { + return parse(clientPaymentOptionSchema, input); +} + +/** Validates all explicit options, including options that might not be selected. */ +export function parseClientPaymentRequired( + input: unknown, +): PaymentRequiredErrorData { + return parse(requiredSchema, input); +} + +/** Validates pending backoff; omitted hints use the client's default. */ +export function parseClientPaymentPending( + input: unknown, +): PaymentPendingErrorData { + return parse(pendingSchema, input === undefined ? {} : input); +} + +/** Rejects invalid client configuration synchronously, before transport mutation. */ +export function validateClientPaymentOptions( + options: ClientPaymentsOptions, +): void { + const schema = z.object({ + syntheticProgressIntervalMs: timerSchema.optional(), + defaultPaymentTtlMs: timerSchema.optional(), + minRetryDelayMs: z + .number() + .finite() + .nonnegative() + .max(MAX_TIMER_MS) + .optional(), + maxPendingRetries: z + .number() + .int() + .nonnegative() + .max(Number.MAX_SAFE_INTEGER) + .optional(), + maxTrackedPaymentAttempts: z + .number() + .int() + .nonnegative() + .max(Number.MAX_SAFE_INTEGER) + .optional(), + paymentInteraction: z.enum(['transparent', 'explicit_gating']).optional(), + handlers: z.array(z.object({ pmi: pmiSchema })).optional(), + limits: z + .object({ + maxAmountByPmi: z.record(pmiSchema, amountSchema).optional(), + maxAttempts: z + .number() + .int() + .nonnegative() + .max(Number.MAX_SAFE_INTEGER) + .optional(), + timeoutMs: timerSchema.optional(), + }) + .optional(), + }); + if (!schema.safeParse(options).success) { + throw new TypeError('Invalid client payment options'); + } +} diff --git a/src/payments/client-payments-limits.test.ts b/src/payments/client-payments-limits.test.ts new file mode 100644 index 0000000..a53c91b --- /dev/null +++ b/src/payments/client-payments-limits.test.ts @@ -0,0 +1,334 @@ +import { describe, expect, test } from 'bun:test'; +import type { Transport } from '@contextvm/mcp-sdk/shared/transport'; +import { withClientPayments } from './client-payments.js'; +import type { PaymentOption } from './types.js'; +import { + run, + waitFor, + tick, + wait, + request, + option, + offer, + required, + pending, + deferred, + failures, +} from '../__mocks__/client-payments-harness.js'; + +describe('client payment limits and validation', () => { + test('replacement invoices share an attempt allowance', async () => { + let calls = 0; + await run( + { + limits: { maxAttempts: 1 }, + onPaymentRequired: async () => { + calls++; + return { paid: true }; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + await waitFor(() => h.sent.length === 2); + h.deliver(required(1, [option('replacement')])); + await waitFor(() => failures(h).length === 1); + expect(calls).toBe(1); + expect(failures(h)[0]?.error.data).toMatchObject({ + reason: 'payment_limit_exceeded', + }); + }, + ); + }); + + test('replacement invoices share a cumulative per-PMI amount limit', async () => { + let calls = 0; + await run( + { + limits: { maxAmountByPmi: { fake: 8 } }, + onPaymentRequired: async () => { + calls++; + return { paid: true }; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + await waitFor(() => h.sent.length === 2); + h.deliver(required(1, [option('replacement', 4)])); + await waitFor(() => failures(h).length === 1); + expect(calls).toBe(1); + }, + ); + }); + + test('multi-option offers reserve a conservative maximum per PMI and exclude unlisted rails', async () => { + const approved: PaymentOption[][] = []; + await run( + { + limits: { maxAmountByPmi: { fake: 10 } }, + onPaymentRequired: async ({ options }) => { + approved.push(options); + return { paid: true }; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver( + required(1, [ + option('low', 2), + option('high', 8), + option('other', 1, 'other'), + ]), + ); + await waitFor(() => h.sent.length === 2); + expect(approved[0]).toEqual([option('low', 2), option('high', 8)]); + h.deliver(required(1, [option('replacement', 3)])); + await waitFor(() => failures(h).length === 1); + expect(approved).toHaveLength(1); + }, + ); + }); + + test('limits are independent for genuinely new invocations', async () => { + let calls = 0; + await run( + { + limits: { maxAttempts: 1, maxAmountByPmi: { fake: 5 } }, + onPaymentRequired: async () => { + calls++; + return { paid: true }; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + await waitFor(() => h.sent.length === 2); + h.deliver({ jsonrpc: '2.0', id: 1, result: {} }); + await h.paid.send(request(2)); + h.deliver(required(2, [option('independent')])); + await waitFor(() => calls === 2); + expect(failures(h)).toHaveLength(0); + }, + ); + }); + + test('zero limits refuse wallet work before policy approval', async () => { + let policies = 0; + let payments = 0; + await run( + { + limits: { maxAttempts: 0 }, + paymentPolicy: () => { + policies++; + return true; + }, + handlers: [ + { + pmi: 'fake', + async handle(): Promise { + payments++; + }, + }, + ], + }, + async (h) => { + await h.paid.send(request()); + h.deliver(offer()); + await waitFor(() => failures(h).length === 1); + expect(policies).toBe(0); + expect(payments).toBe(0); + }, + ); + }); + + test('deadline aborts awaiting approval and prevents a later retry', async () => { + const decision = deferred<{ paid: boolean }>(); + let signal: AbortSignal | undefined; + await run( + { + limits: { timeoutMs: 10 }, + onPaymentRequired: async ({ abortSignal }) => { + signal = abortSignal; + return decision.promise; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + await waitFor(() => failures(h).length === 1); + expect(signal?.aborted).toBe(true); + decision.resolve({ paid: true }); + await wait(10); + expect(h.sent).toHaveLength(1); + expect(failures(h)).toHaveLength(1); + expect(failures(h)[0]?.error.data).toMatchObject({ + reason: 'payment_deadline_exceeded', + }); + }, + ); + }); + + for (const invalid of [ + { ...option(), amount: -1 }, + { ...option(), amount: NaN }, + { ...option(), amount: Infinity }, + { ...option(), amount: Number.MAX_SAFE_INTEGER + 1 }, + { ...option(), amount: '5' }, + { ...option(), pmi: 'INVALID:PMI' }, + { ...option(), pay_req: '' }, + { ...option(), pay_req: ' ' }, + { ...option(), ttl: 0 }, + { ...option(), ttl: -1 }, + { ...option(), ttl: Infinity }, + { ...option(), _meta: [] }, + null, + ]) { + test(`rejects malformed transparent offer ${JSON.stringify(invalid)}`, async () => { + let calls = 0; + await run( + { + handlers: [ + { + pmi: 'fake', + async handle(): Promise { + calls++; + }, + }, + ], + }, + async (h) => { + await h.paid.send(request()); + h.deliver(offer(invalid)); + await waitFor(() => failures(h).length === 1); + expect(calls).toBe(0); + expect(failures(h)[0]?.error.data).toMatchObject({ + reason: 'invalid_payment', + }); + }, + ); + }); + } + + for (const invalid of [ + [], + [option(), { ...option('invalid'), amount: -1 }], + [option(), option('invoice', 6)], + ]) { + test(`validates every explicit option ${JSON.stringify(invalid)}`, async () => { + let calls = 0; + await run( + { + onPaymentRequired: async () => { + calls++; + return { paid: true }; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required(1, invalid)); + await waitFor(() => failures(h).length === 1); + expect(calls).toBe(0); + }, + ); + }); + } + + for (const invalid of [-1, NaN, Infinity, '1', null]) { + test(`invalid retry hint ${String(invalid)} cannot schedule a send`, async () => { + await run({}, async (h) => { + await h.paid.send(request()); + h.deliver(pending(1, invalid)); + await waitFor(() => failures(h).length === 1); + await wait(5); + expect(h.sent).toHaveLength(1); + }); + }); + } + + test('omitted retry hint uses a finite default delay', async () => { + await run({}, async (h) => { + await h.paid.send(request()); + h.deliver(pending()); + await waitFor(() => h.sent.length === 2, { timeoutMs: 1500 }); + expect(failures(h)).toHaveLength(0); + }); + }); + + test('unknown metadata is preserved without influencing validation', async () => { + let metadata: unknown; + await run( + { + handlers: [ + { + pmi: 'fake', + async handle(req): Promise { + metadata = req._meta; + }, + }, + ], + }, + async (h) => { + await h.paid.send(request()); + h.deliver( + offer({ + ...option(), + _meta: { custom: { future: true } }, + extra: 'ignored', + }), + ); + await waitFor(() => metadata !== undefined); + expect(metadata).toEqual({ custom: { future: true } }); + expect(failures(h)).toHaveLength(0); + }, + ); + }); + + test('ambiguous uncorrelated notifications never choose between concurrent requests', async () => { + let calls = 0; + await run( + { + handlers: [ + { + pmi: 'fake', + async handle(): Promise { + calls++; + }, + }, + ], + }, + async (h) => { + await h.paid.send(request()); + await h.paid.send(request(2)); + h.deliver(offer()); + await tick(); + expect(calls).toBe(0); + expect(h.errors).toHaveLength(1); + }, + ); + }); + + for (const options of [ + { minRetryDelayMs: -1 }, + { minRetryDelayMs: NaN }, + { maxPendingRetries: 0.5 }, + { maxTrackedPaymentAttempts: -1 }, + { syntheticProgressIntervalMs: 0 }, + { limits: { timeoutMs: 0 } }, + { limits: { timeoutMs: 2_147_483_648 } }, + { limits: { maxAttempts: -1 } }, + { limits: { maxAmountByPmi: { fake: Infinity } } }, + ]) { + test(`configuration validation rejects ${JSON.stringify(options)} before wrapping`, () => { + const base: Transport = { + async start(): Promise {}, + async send(): Promise {}, + async close(): Promise {}, + }; + expect(() => withClientPayments(base, options)).toThrow( + 'Invalid client payment options', + ); + expect(() => withClientPayments(base, {})).not.toThrow(); + }); + } +}); diff --git a/src/payments/client-payments-reliability.test.ts b/src/payments/client-payments-reliability.test.ts new file mode 100644 index 0000000..27aa664 --- /dev/null +++ b/src/payments/client-payments-reliability.test.ts @@ -0,0 +1,597 @@ +import { describe, expect, test } from 'bun:test'; +import type { JSONRPCMessage } from '@contextvm/mcp-sdk/types.js'; +import { + run, + waitFor, + tick, + wait, + request, + option, + offer, + required, + pending, + deferred, + failures, +} from '../__mocks__/client-payments-harness.js'; + +describe('client payment attempt reliability', () => { + test('retains transparent attempt history after handler return and invoice TTL', async () => { + let calls = 0; + await run( + { + handlers: [ + { + pmi: 'fake', + async handle(): Promise { + calls++; + }, + }, + ], + }, + async (h) => { + await h.paid.send(request()); + h.deliver(offer({ ...option(), ttl: 1 })); + await waitFor(() => calls === 1); + await wait(1100); + h.deliver(offer(option())); + await tick(); + expect(calls).toBe(1); + }, + ); + }); + + test('deduplicates concurrent and sequential explicit approvals', async () => { + let calls = 0; + const decision = deferred<{ paid: boolean }>(); + await run( + { + onPaymentRequired: async () => { + calls++; + return decision.promise; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + h.deliver(required()); + expect(calls).toBe(1); + decision.resolve({ paid: true }); + await waitFor(() => h.sent.length === 2); + h.deliver(required()); + await waitFor(() => h.sent.length === 3); + expect(calls).toBe(1); + expect(failures(h)).toHaveLength(0); + }, + ); + }); + + test('an uncertain wallet result cannot be paid again on another request id', async () => { + let calls = 0; + await run( + { + onPaymentRequired: async () => { + calls++; + throw new Error('wallet response lost'); + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + await waitFor(() => failures(h).length === 1); + await h.paid.send(request(2)); + h.deliver(required(2)); + await waitFor(() => failures(h).length === 2); + expect(calls).toBe(1); + expect(failures(h)[1]?.error.data).toMatchObject({ + reason: 'payment_outcome_unknown', + }); + }, + ); + }); + + test('different PMIs with the same opaque request remain independent', async () => { + let calls = 0; + await run( + { + handlers: ['fake', 'other'].map((pmi) => ({ + pmi, + async handle(): Promise { + calls++; + }, + })), + }, + async (h) => { + await h.paid.send(request()); + h.deliver(offer()); + await waitFor(() => calls === 1); + h.deliver({ jsonrpc: '2.0', id: 1, result: {} }); + await h.paid.send(request(2)); + h.deliver(offer(option('invoice', 5, 'other'))); + await waitFor(() => calls === 2); + expect(failures(h)).toHaveLength(0); + }, + ); + }); + + test('attempt history is isolated between stable transport peers', async () => { + let calls = 0; + const options = { + handlers: [ + { + pmi: 'fake', + async handle(): Promise { + calls++; + }, + }, + ], + }; + await run(options, async (one) => { + await run(options, async (two) => { + await one.paid.send(request()); + await two.paid.send(request()); + one.deliver(offer()); + two.deliver(offer()); + await waitFor(() => calls === 2); + expect(failures(one)).toHaveLength(0); + expect(failures(two)).toHaveLength(0); + }); + }); + }); + + test('attempt capacity refuses a new invoice without forgetting a completed payment', async () => { + let calls = 0; + await run( + { + maxTrackedPaymentAttempts: 1, + onPaymentRequired: async () => { + calls++; + return { paid: true }; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + await waitFor(() => h.sent.length === 2); + h.deliver({ jsonrpc: '2.0', id: 1, result: {} }); + await h.paid.send(request(2)); + h.deliver(required(2, [option('new-invoice')])); + await waitFor(() => failures(h).length === 1); + expect(calls).toBe(1); + expect(failures(h)[0]?.error.data).toMatchObject({ + reason: 'payment_attempt_capacity', + }); + await h.paid.send(request(3)); + h.deliver(required(3)); + await waitFor(() => h.sent.length === 5); + expect(calls).toBe(1); + }, + ); + }); + + test('changed amount for a known invoice never triggers another approval', async () => { + let calls = 0; + await run( + { + onPaymentRequired: async () => { + calls++; + return { paid: true }; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + await waitFor(() => h.sent.length === 2); + h.deliver(required(1, [option('invoice', 6)])); + await waitFor(() => failures(h).length === 1); + expect(calls).toBe(1); + expect(failures(h)[0]?.error.data).toMatchObject({ + reason: 'payment_offer_changed', + }); + }, + ); + }); + + test('different offers cannot run approval callbacks concurrently for one invocation', async () => { + let calls = 0; + const decision = deferred<{ paid: boolean }>(); + await run( + { + onPaymentRequired: async () => { + calls++; + return decision.promise; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + h.deliver(required(1, [option('replacement')])); + await waitFor(() => failures(h).length === 1); + decision.resolve({ paid: true }); + await tick(); + expect(calls).toBe(1); + expect(h.sent).toHaveLength(1); + expect(failures(h)[0]?.error.data).toMatchObject({ + reason: 'payment_outcome_unknown', + }); + }, + ); + }); +}); + +test('an unknown option takes precedence over another pending payment attempt', async () => { + const pendingDecision = deferred<{ paid: boolean }>(); + let approvals = 0; + await run( + { + onPaymentRequired: async ({ options }) => { + approvals++; + if (options[0]?.pay_req === 'unknown') + throw new Error('lost wallet result'); + return pendingDecision.promise; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required(1, [option('unknown')])); + await waitFor(() => failures(h).length === 1); + await h.paid.send(request(2)); + h.deliver(required(2, [option('pending')])); + await h.paid.send(request(3)); + h.deliver(required(3, [option('unknown'), option('pending')])); + await waitFor(() => failures(h).length === 2); + expect(failures(h)[1]?.error.data).toMatchObject({ + reason: 'payment_outcome_unknown', + }); + expect(approvals).toBe(2); + pendingDecision.resolve({ paid: false }); + await tick(); + }, + ); +}); + +describe('wallet eligibility and attempt retention', () => { + test('policy mutations cannot change the validated invoice or amount sent to a handler', async () => { + let handled: unknown; + await run( + { + paymentPolicy: (req) => { + req.amount = 100; + req.pay_req = 'changed'; + return true; + }, + handlers: [ + { + pmi: 'fake', + async handle(req): Promise { + handled = req; + }, + }, + ], + }, + async (h) => { + await h.paid.send(request()); + h.deliver(offer()); + await waitFor(() => handled !== undefined); + expect(handled).toMatchObject(option()); + expect(failures(h)).toHaveLength(0); + }, + ); + }); + + test('expiry during an eligibility check prevents a wallet action', async () => { + const eligibility = deferred(); + let calls = 0; + await run( + { + handlers: [ + { + pmi: 'fake', + canHandle: () => eligibility.promise, + async handle(): Promise { + calls++; + }, + }, + ], + }, + async (h) => { + await h.paid.send(request()); + h.deliver(offer({ ...option(), ttl: 0.001 })); + await wait(10); + eligibility.resolve(true); + await waitFor(() => failures(h).length === 1); + expect(calls).toBe(0); + expect(failures(h)[0]?.error.data).toMatchObject({ + reason: 'payment_offer_expired', + }); + }, + ); + }); + + test('a failed initial send releases request state before late payment offers', async () => { + let calls = 0; + await run( + { + onPaymentRequired: async () => { + calls++; + return { paid: true }; + }, + }, + async (h) => { + h.base.send = async (): Promise => { + throw new Error('publish failed'); + }; + await expect(h.paid.send(request())).rejects.toThrow('publish failed'); + h.deliver(required()); + await tick(); + expect(calls).toBe(0); + expect(h.received).toHaveLength(0); + }, + ); + }); + + test('closed/restarted wrappers retain completed offer history', async () => { + let calls = 0; + await run( + { + onPaymentRequired: async () => { + calls++; + return { paid: true }; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + await waitFor(() => h.sent.length === 2); + await h.paid.close(); + await h.paid.start(); + await h.paid.send(request(2)); + h.deliver(required(2)); + await waitFor(() => h.sent.length === 4); + expect(calls).toBe(1); + }, + ); + }); + + test('standalone generic notifications cannot bypass per-invocation limits', async () => { + let calls = 0; + await run( + { + limits: { maxAttempts: 1 }, + handlers: [ + { + pmi: 'fake', + async handle(): Promise { + calls++; + }, + }, + ], + }, + async (h) => { + h.deliver(offer()); + await waitFor(() => h.errors.length === 1); + expect(calls).toBe(0); + expect(h.errors[0]).toMatchObject({ + reason: 'payment_request_uncorrelated', + }); + }, + ); + }); + + test('zero attempt-history capacity refuses a fresh offer before approval', async () => { + let calls = 0; + await run( + { + maxTrackedPaymentAttempts: 0, + onPaymentRequired: async () => { + calls++; + return { paid: true }; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + await waitFor(() => failures(h).length === 1); + expect(calls).toBe(0); + expect(failures(h)[0]?.error.data).toMatchObject({ + reason: 'payment_attempt_capacity', + }); + }, + ); + }); +}); + +describe('client payment request lifecycle', () => { + for (const terminal of [ + { jsonrpc: '2.0', id: 1, result: {} }, + { jsonrpc: '2.0', id: 1, error: { code: -32000, message: 'done' } }, + ] as JSONRPCMessage[]) { + test(`terminal ${'result' in terminal ? 'result' : 'error'} cancels a queued retry`, async () => { + await run({ minRetryDelayMs: 20 }, async (h) => { + await h.paid.send(request()); + h.deliver(pending(1, 0)); + h.deliver(terminal); + await wait(40); + expect(h.sent).toHaveLength(1); + expect(h.received).toEqual([terminal]); + }); + }); + } + + test('outbound MCP cancellation stops retries for numeric id zero only', async () => { + await run({ minRetryDelayMs: 20 }, async (h) => { + await h.paid.send(request(0)); + await h.paid.send(request('0')); + h.deliver(pending(0, 0)); + h.deliver(pending('0', 0)); + await h.paid.send({ + jsonrpc: '2.0', + method: 'notifications/cancelled', + params: { requestId: 0 }, + }); + await wait(40); + const retries = h.sent.slice(3); + expect(retries).toHaveLength(1); + expect(retries[0]).toMatchObject({ id: '0' }); + }); + }); + + test('inbound cancellation prevents a delayed policy from starting the wallet', async () => { + let calls = 0; + const policy = deferred(); + await run( + { + paymentPolicy: () => policy.promise, + handlers: [ + { + pmi: 'fake', + async handle(): Promise { + calls++; + }, + }, + ], + }, + async (h) => { + await h.paid.send(request()); + h.deliver(offer()); + h.deliver({ + jsonrpc: '2.0', + method: 'notifications/cancelled', + params: { requestId: 1 }, + }); + policy.resolve(true); + await tick(); + expect(calls).toBe(0); + expect(failures(h)).toHaveLength(0); + }, + ); + }); + + test('close fences approval even when the underlying transport never fires onclose', async () => { + const decision = deferred<{ paid: boolean }>(); + let signal: AbortSignal | undefined; + await run( + { + onPaymentRequired: async ({ abortSignal }) => { + signal = abortSignal; + return decision.promise; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver(required()); + await h.paid.close(); + expect(signal?.aborted).toBe(true); + decision.resolve({ paid: true }); + await wait(10); + expect(h.sent).toHaveLength(1); + expect(failures(h)).toHaveLength(0); + }, + ); + }); + + test('terminal result fences an awaiting handler eligibility check', async () => { + let calls = 0; + const eligibility = deferred(); + await run( + { + handlers: [ + { + pmi: 'fake', + canHandle: () => eligibility.promise, + async handle(): Promise { + calls++; + }, + }, + ], + }, + async (h) => { + await h.paid.send(request()); + h.deliver(offer()); + h.deliver({ jsonrpc: '2.0', id: 1, result: {} }); + eligibility.resolve(true); + await tick(); + expect(calls).toBe(0); + expect(failures(h)).toHaveLength(0); + }, + ); + }); + + test('request-id reuse does not allow the previous generation to pay or fail the replacement', async () => { + const oldPolicy = deferred(); + let policies = 0; + let payments = 0; + await run( + { + paymentPolicy: () => (++policies === 1 ? oldPolicy.promise : true), + handlers: [ + { + pmi: 'fake', + async handle(): Promise { + payments++; + }, + }, + ], + }, + async (h) => { + await h.paid.send(request()); + h.deliver(offer()); + await h.paid.send({ ...request(), params: { name: 'replacement' } }); + h.deliver(offer(option('new-invoice'))); + oldPolicy.resolve(false); + await waitFor(() => payments === 1); + expect(failures(h)).toHaveLength(0); + }, + ); + }); + + test('approval cannot mutate the cached retry invocation', async () => { + await run( + { + onPaymentRequired: async ({ originalRequest }) => { + originalRequest.id = 'changed'; + originalRequest.params = { name: 'changed' }; + return { paid: true }; + }, + }, + async (h) => { + const original = request(); + await h.paid.send(original); + original.params = { name: 'also-changed' }; + h.deliver(required()); + await waitFor(() => h.sent.length === 2); + expect(h.sent[1]).toEqual(request()); + }, + ); + }); + + test('duplicate pending responses queue one retry and consume one retry allowance', async () => { + await run({ maxPendingRetries: 1, minRetryDelayMs: 20 }, async (h) => { + await h.paid.send(request()); + h.deliver(pending(1, 0)); + h.deliver(pending(1, 0)); + await waitFor(() => h.sent.length === 2); + expect(failures(h)).toHaveLength(0); + h.deliver(pending(1, 0)); + expect(failures(h)).toHaveLength(1); + await wait(30); + expect(h.sent).toHaveLength(2); + }); + }); + + test('a retry publish failure terminates the request once', async () => { + await run({}, async (h) => { + await h.paid.send(request()); + h.base.send = async (): Promise => { + throw new Error('relay failed'); + }; + h.deliver(pending(1, 0)); + await waitFor(() => failures(h).length === 1); + h.deliver(required()); + await tick(); + expect(failures(h)).toHaveLength(1); + expect(failures(h)[0]?.error.data).toMatchObject({ + reason: 'payment_retry_failed', + }); + }); + }); +}); diff --git a/src/payments/client-payments.test.ts b/src/payments/client-payments.test.ts index fb039c8..b95bd01 100644 --- a/src/payments/client-payments.test.ts +++ b/src/payments/client-payments.test.ts @@ -66,6 +66,7 @@ describe('withClientPayments()', () => { pmi: 'fake', description: undefined, requestEventId: 'req-id', + abortSignal: expect.any(AbortSignal), }); }); @@ -396,6 +397,7 @@ describe('withClientPayments()', () => { amount: 1, method: 'tools/call', capability: 'tool:add', + reason: 'payment_declined', }); }); @@ -457,6 +459,7 @@ describe('withClientPayments()', () => { amount: 2, method: 'prompts/get', capability: 'prompt:hi', + reason: 'payment_declined', }); }); @@ -525,6 +528,7 @@ describe('withClientPayments()', () => { amount: 1, method: 'tools/call', capability: 'tool:paid', + reason: 'payment_negotiation_failed', }); }); @@ -883,8 +887,8 @@ describe('withClientPayments()', () => { { eventId: 'evt4', correlatedEventId: 'req-event-id-3' }, ); - // Wait for async processing - await new Promise((r) => setTimeout(r, 0)); + // Wait for the Nostr retry floor. + await new Promise((r) => setTimeout(r, 1100)); // Error should not be delivered to caller expect(observed).toHaveLength(0); @@ -1019,8 +1023,8 @@ describe('withClientPayments()', () => { expect(observed).toHaveLength(0); expect(sentMessage).toBeUndefined(); - // Wait for retry_after timer to fire - await new Promise((r) => setTimeout(r, 60)); + // Nostr transport enforces a one-second floor despite a shorter hint. + await new Promise((r) => setTimeout(r, 1100)); // Error should not be delivered to caller expect(observed).toHaveLength(0); @@ -1148,11 +1152,11 @@ describe('withClientPayments()', () => { // First two: intercepted and retried (not observed by caller). deliverPending(); - await new Promise((r) => setTimeout(r, 20)); + await new Promise((r) => setTimeout(r, 1100)); expect(observed).toHaveLength(0); deliverPending(); - await new Promise((r) => setTimeout(r, 25)); + await new Promise((r) => setTimeout(r, 1100)); expect(observed).toHaveLength(0); // Third: retry budget exhausted → -32043 reaches the caller. @@ -1230,8 +1234,8 @@ describe('withClientPayments()', () => { await new Promise((r) => setTimeout(r, 10)); expect(sentMessage).toBeUndefined(); - // After the floor: the original request is retried. - await new Promise((r) => setTimeout(r, 120)); + // Nostr retries always wait at least one second. + await new Promise((r) => setTimeout(r, 1100)); expect(sentMessage as unknown).toEqual({ jsonrpc: '2.0', id: 77, diff --git a/src/payments/client-payments.ts b/src/payments/client-payments.ts index 916a305..058b7d0 100644 --- a/src/payments/client-payments.ts +++ b/src/payments/client-payments.ts @@ -1,32 +1,41 @@ import type { Transport } from '@contextvm/mcp-sdk/shared/transport'; import { isJSONRPCNotification, + isJSONRPCRequest, isJSONRPCResultResponse, isJSONRPCErrorResponse, - type JSONRPCNotification, type JSONRPCMessage, - type JSONRPCRequest, type JSONRPCErrorResponse, } from '@contextvm/mcp-sdk/types.js'; -import { NostrClientTransport } from '../transport/nostr-client-transport.js'; -import type { TransportWithContext } from '../transport/nostr-client-transport.js'; +import { + NostrClientTransport, + type TransportWithContext, +} from '../transport/nostr-client-transport.js'; +import type { PendingRequest } from '../transport/nostr-client/correlation-store.js'; +import { createLogger } from '../core/utils/logger.js'; import type { PaymentHandler, - PaymentRejectedNotification, - PaymentRequiredNotification, PaymentHandlerRequest, - PaymentInteractionMode, PaymentOption, - PaymentRequiredErrorData, - PaymentPendingErrorData, + PaymentRequiredNotification, } from './types.js'; -import { LruCache } from '../core/utils/lru-cache.js'; -import { createLogger } from '../core/utils/logger.js'; -import type { - OriginalRequestContext, - PendingRequest, -} from '../transport/nostr-client/correlation-store.js'; - +import type { ClientPaymentsOptions } from './client-payment-options.js'; +import { + ClientPaymentAttempts, + type ClientPaymentAttempt, +} from './client-payment-attempts.js'; +import { + ClientPaymentLifecycle, + type ClientPaymentRequest, +} from './client-payment-lifecycle.js'; +import { ClientPaymentProgress } from './client-payment-progress.js'; +import { + ClientPaymentError, + parseClientPaymentOption, + parseClientPaymentRequired, + parseClientPaymentPending, + validateClientPaymentOptions, +} from './client-payment-validation.js'; import { DEFAULT_SYNTHETIC_PROGRESS_INTERVAL_MS, DEFAULT_PAYMENT_TTL_MS, @@ -37,749 +46,529 @@ import { PAYMENT_PENDING_ERROR_CODE, } from './constants.js'; -export interface ClientPaymentsOptions { - /** - * Payment handlers for in-band (programmatic) payment, indexed by their - * {@link PaymentHandler.pmi}. Each handler's `pmi` is advertised to the - * server so it can pick a matching rail (the wallet-client fast path). - * - * **Omit entirely** for a PMI-agnostic client that pays out-of-band: no PMIs - * are advertised, so per CEP-8 the server sends `payment_required` for any of - * its processors. The notification is forwarded to the application via - * `onmessage`, synthetic progress keeps the original MCP request alive while - * the payment settles out-of-band, and the server's TTL is the timeout. - */ - handlers?: readonly PaymentHandler[]; - /** - * Interval for periodic synthetic progress heartbeats (milliseconds). - * - * One heartbeat is also emitted immediately when `payment_required` is - * received so the MCP timeout is always reset as soon as the payment flow - * begins. - * @default DEFAULT_SYNTHETIC_PROGRESS_INTERVAL_MS (30_000 ms) - */ - syntheticProgressIntervalMs?: number; - /** - * Duration in ms for synthetic progress when `payment_required` carries no `ttl`. - * - * Mirrors the server-side `paymentTtlMs` default so the client keeps the MCP - * request alive for at least as long as the server will wait. - * @default DEFAULT_PAYMENT_TTL_MS (300_000 ms) - */ - defaultPaymentTtlMs?: number; - /** - * Minimum delay before a `-32043` Payment Pending retry is re-sent - * (milliseconds), applied after the server-provided `retry_after` backoff. - * - * Guards against `retry_after: 0`: an immediate retry within the same second - * can produce a byte-identical Nostr event (same content, same tags, same - * `created_at` at second resolution), which relays and servers swallow as a - * duplicate — silently losing the retry. - * @default 1000 - */ - minRetryDelayMs?: number; - - /** - * Optional policy hook invoked when a `payment_required` notification is received. - * - * This is evaluated before checking rail-specific `canHandle` methods. - * - * When the underlying transport supports request correlation (e.g. {@link NostrClientTransport}), - * `originalRequestContext` provides minimal details about the original JSON-RPC request that - * triggered the payment. - */ - paymentPolicy?: ( - req: PaymentHandlerRequest, - originalRequestContext?: OriginalRequestContext, - ) => boolean | Promise; - - /** Requested payment interaction mode. @default 'transparent' */ - paymentInteraction?: PaymentInteractionMode; - - /** - * Maximum number of -32043 (Payment Pending) retries before giving up. - * - * With retry_after=2 and 1.5× exponential backoff capped at 10s, the default - * of 10 retries gives ~45s of cumulative wait — enough for typical verification - * flows. Increase for slow payment processors (e.g. on-chain confirmation). - * @default 10 - */ - maxPendingRetries?: number; - - /** - * Handler for explicit-gating -32042 errors. - * Called when a priced invocation returns Payment Required. - * The handler should pay one option and signal completion. - * - * **Error handling contract**: - * - If the promise resolves with `{ paid: true }`, the wrapper auto-retries the - * original request with the same `method` and `params`. - * - If the promise resolves with `{ paid: false, reason }`, the wrapper synthesizes - * a JSON-RPC error to the caller with code `-32042` and `data: { reason }`. - * Use `reason: 'user_cancelled'` for user-initiated cancellations. - * - If the promise **rejects**, the wrapper MUST NOT silently fall back. - * It synthesizes a JSON-RPC error with code `-32042` and - * `data: { reason: error.message, type: 'payment_handler_error' }`. - * - Transient payment-provider failures should reject with an Error whose - * `message` contains the provider error details. - * - * **Verify-timeout window**: if the server's verification times out or fails - * after the client paid, its pending state is cleared and the client's retry - * receives a fresh `-32042` with a new invoice (CEP-8-compliant). The wrapper - * does not dedup across distinct `pay_req` values. - */ - onPaymentRequired?: (params: { - options: PaymentOption[]; - instructions?: string; - originalRequest: JSONRPCRequest; - }) => Promise<{ paid: boolean; reason?: string }>; -} - -type ProgressToken = string; +export type { + ClientPaymentsOptions, + ClientPaymentLimits, +} from './client-payment-options.js'; -type SyntheticProgressEntry = { - stopAtMs: number; - wireProgressToken: string | number; -}; - -function supportsOnmessageWithContext( - transport: Transport, -): transport is TransportWithContext { - // Avoid using the `in` operator here; prefer an ES3-compatible check. - return Object.prototype.hasOwnProperty.call( - transport, - 'onmessageWithContext', - ); -} +const clientPaymentsWrapped = new WeakSet(); -function isExplicitPaymentRequiredError( - msg: JSONRPCMessage, -): msg is JSONRPCErrorResponse { +function isPaymentOffer( + message: JSONRPCMessage, +): message is PaymentRequiredNotification { + // Classify by envelope before schema helpers can reject malformed params. return ( - isJSONRPCErrorResponse(msg) && - msg.error.code === PAYMENT_REQUIRED_ERROR_CODE && - typeof msg.error.data === 'object' && - msg.error.data !== null && - Array.isArray( - (msg.error.data as { payment_options?: unknown }).payment_options, - ) && - (msg.error.data as { payment_options: unknown[] }).payment_options.length > - 0 + 'method' in message && + message.method === PAYMENT_REQUIRED_METHOD && + !('id' in message) ); } -function isExplicitPaymentPendingError( - msg: JSONRPCMessage, -): msg is JSONRPCErrorResponse { +function isExplicitPaymentError( + message: JSONRPCMessage, +): message is JSONRPCErrorResponse { return ( - isJSONRPCErrorResponse(msg) && - msg.error.code === PAYMENT_PENDING_ERROR_CODE && - typeof msg.error.data === 'object' && - msg.error.data !== null && - typeof (msg.error.data as { retry_after?: unknown }).retry_after === - 'number' + isJSONRPCErrorResponse(message) && + (message.error.code === PAYMENT_REQUIRED_ERROR_CODE || + message.error.code === PAYMENT_PENDING_ERROR_CODE) ); } -function isPaymentRequiredNotification( - msg: JSONRPCMessage, -): msg is PaymentRequiredNotification { - return isJSONRPCNotification(msg) && msg.method === PAYMENT_REQUIRED_METHOD; +function cancelledRequestId( + message: JSONRPCMessage, +): string | number | undefined { + if ( + !isJSONRPCNotification(message) || + message.method !== 'notifications/cancelled' + ) + return undefined; + const id = message.params?.requestId; + return typeof id === 'string' || typeof id === 'number' ? id : undefined; } -/** Transports already wrapped by {@link withClientPayments}. */ -const clientPaymentsWrapped = new WeakSet(); - /** - * Wraps a transport to automatically handle CEP-8 payment requests. + * Wraps a transport with validated CEP-8 offers, bounded payment attempts, and + * request-owned retries. Nostr peers are authenticated by their transport; + * generic transports isolate attempt history per wrapper and must keep their + * peer identity stable. History is retained for the lifetime of the wrapper. * - * When `transport` is a {@link NostrClientTransport}, PMI advertisement and - * synthetic progress injection are enabled automatically. - * - * Note: `payment_rejected` → JSON-RPC error synthesis requires a - * {@link NostrClientTransport} because correlation (requestEventId → MCP request id) - * depends on Nostr event tagging. On plain transports a rejected payment surfaces - * as an unhandled notification and the MCP request times out naturally. + * Unmatched handlers leave transparent payment to the application. Generic + * notifications without request correlation support legacy handler execution, + * but per-invocation limits require an identifiable original request. */ export function withClientPayments( transport: Transport, options: ClientPaymentsOptions, ): Transport { - // A second wrap silently double-pays every offer (each wrapper runs its own - // pipeline with its own dedup on separate closures) or, for sibling wraps, - // silently kills the first wrapper's trampolines. Refuse outright. - if (clientPaymentsWrapped.has(transport)) { + if (clientPaymentsWrapped.has(transport)) throw new Error('withClientPayments already called on this transport'); - } + validateClientPaymentOptions(options); const logger = createLogger('client-payments'); - - const syntheticProgressIntervalMs = - options.syntheticProgressIntervalMs ?? - DEFAULT_SYNTHETIC_PROGRESS_INTERVAL_MS; - - const defaultPaymentTtlMs = - options.defaultPaymentTtlMs ?? DEFAULT_PAYMENT_TTL_MS; - - const minRetryDelayMs = options.minRetryDelayMs ?? 1000; - - const syntheticProgress = new Map(); - - let syntheticProgressScheduler: ReturnType | undefined; - - const maybeStopScheduler = (): void => { - if (syntheticProgress.size > 0) return; - if (!syntheticProgressScheduler) return; - clearInterval(syntheticProgressScheduler); - syntheticProgressScheduler = undefined; + const nostr = + transport instanceof NostrClientTransport ? transport : undefined; + const minRetryDelayMs = Math.max( + options.minRetryDelayMs ?? 1000, + nostr ? 1000 : 0, + ); + const maxRetries = options.maxPendingRetries ?? 10; + const defaultTtlMs = options.defaultPaymentTtlMs ?? DEFAULT_PAYMENT_TTL_MS; + const limits = options.limits && { + ...options.limits, + maxAmountByPmi: options.limits.maxAmountByPmi && { + ...options.limits.maxAmountByPmi, + }, }; + let onmessage: Transport['onmessage']; + let onerror: Transport['onerror']; + let onclose: Transport['onclose']; + const progress = new ClientPaymentProgress( + options.syntheticProgressIntervalMs ?? + DEFAULT_SYNTHETIC_PROGRESS_INTERVAL_MS, + (message) => onmessage?.(message), + ); - const tickSyntheticProgress = (): void => { - if (syntheticProgress.size === 0) { - maybeStopScheduler(); + const emitFailure = ( + state: ClientPaymentRequest, + error: ClientPaymentError, + code = -32000, + data: Record = {}, + ): void => { + if (state.id === undefined) { + onerror?.(error); return; } - - const now = Date.now(); - for (const [token, entry] of syntheticProgress) { - if (now >= entry.stopAtMs) { - syntheticProgress.delete(token); - continue; - } - - onmessage?.({ - jsonrpc: '2.0', - method: 'notifications/progress', - params: { - progressToken: entry.wireProgressToken, - progress: 0, - }, - } as JSONRPCNotification); - } - - maybeStopScheduler(); - }; - - const ensureSchedulerStarted = (): void => { - if (syntheticProgressScheduler) return; - syntheticProgressScheduler = setInterval( - tickSyntheticProgress, - syntheticProgressIntervalMs, - ); - }; - - const stopSyntheticProgress = (token: ProgressToken): void => { - if (!syntheticProgress.delete(token)) return; - maybeStopScheduler(); - }; - - const pendingTimers = new Set>(); - // Bounded together with rawRequestCache: an abandoned request's counter must - // not accumulate forever (terminal responses and close() clear these, but a - // request that never terminates otherwise would leak). - const retryCounts = new LruCache(1000); - const rawRequestCache = new LruCache(1000); - const MAX_RETRIES = options.maxPendingRetries ?? 10; - - /** - * Disposes all client-side payment state: synthetic progress, pending retry - * timers, retry counters, and the raw-request cache. Called on transport close. - */ - const disposeClientState = (): void => { - syntheticProgress.clear(); - if (syntheticProgressScheduler) { - clearInterval(syntheticProgressScheduler); - syntheticProgressScheduler = undefined; - } - for (const timer of pendingTimers) { - clearTimeout(timer); - } - pendingTimers.clear(); - retryCounts.clear(); - rawRequestCache.clear(); - }; - - // Ensure CEP-8 discovery/negotiation: when using Nostr transports, always - // advertise the handler PMIs in preference order. Omitting handlers entirely - // advertises no PMI, so the server sends payment_required for any processor - // (CEP-8: client that specified no PMI). - const advertisedPmis = (options.handlers ?? []).map((h) => h.pmi); - if (transport instanceof NostrClientTransport) { - transport.setClientPmis(advertisedPmis); - logger.debug('advertised client PMIs', { pmis: advertisedPmis }); - if (options.paymentInteraction) { - transport.setPaymentInteraction(options.paymentInteraction); - logger.debug('advertised requested payment interaction mode', { - mode: options.paymentInteraction, - }); - } - } - - // Index handlers by PMI. Warn on duplicates — Map construction silently keeps - // only the last. - const handlersByPmi = new Map(); - for (const h of options.handlers ?? []) { - if (handlersByPmi.has(h.pmi)) { - logger.warn('duplicate PMI handler registered, last one wins', { - pmi: h.pmi, - }); - } - handlersByPmi.set(h.pmi, h); - } - - // Prevent double-paying if relays or servers deliver duplicate payment_required notifications. - const inFlightPayReqs = new Set(); - - let onmessage: ((message: JSONRPCMessage) => void) | undefined; - let onerror: ((error: Error) => void) | undefined; - let onclose: (() => void) | undefined; - - /** Emits a synthesized JSON-RPC error to the upstream consumer via `onmessage`. */ - const synthesizePaymentError = (params: { - id: string | number | undefined; - code: number; - message: string; - data: Record; - }): void => { onmessage?.({ jsonrpc: '2.0', - id: params.id, + id: state.id, error: { - code: params.code, - message: params.message, - data: params.data, + code, + message: error.message, + data: { reason: error.reason, ...data }, }, - } as JSONRPCMessage); + }); + }; + const lifecycle = new ClientPaymentLifecycle(limits, emitFailure, (state) => { + progress.stop(state.key); + if (state.id !== undefined) + nostr?.forgetPendingRequestsForRequestId(state.id); + }); + const attempts = new ClientPaymentAttempts( + options.maxTrackedPaymentAttempts ?? 1000, + ); + const handlers = new Map(); + for (const handler of options.handlers ?? []) { + if (handlers.has(handler.pmi)) + logger.warn('duplicate PMI handler registered, last one wins', { + pmi: handler.pmi, + }); + handlers.set(handler.pmi, handler); + } + nostr?.setClientPmis([...handlers.keys()]); + if (options.paymentInteraction) + nostr?.setPaymentInteraction(options.paymentInteraction); + + const fail = ( + state: ClientPaymentRequest, + error: ClientPaymentError, + code = -32000, + data: Record = {}, + ): void => { + if (!lifecycle.isActive(state)) return; + lifecycle.finish(state); + emitFailure(state, error, code, data); }; - /** - * Synthesize a generic `-32000` decline for the original request, carrying the - * PMI/amount and the original request's method/capability. No-op when there - * is no correlated pending request to fail. Shared by the no-handler, - * explicit-gating-rejected, and payment_rejected paths. - */ - const synthesizePaymentDecline = ( - pending: PendingRequest | undefined, - message: string, - pmi: string, - amount: number, + const scheduleRetry = ( + state: ClientPaymentRequest, + delayMs: number, + code: number, ): void => { - if (pending?.originalRequestId == null) { + if ( + !lifecycle.isActive(state) || + !state.request || + state.retryTimer !== undefined + ) return; - } - synthesizePaymentError({ - id: pending.originalRequestId, - code: -32000, - message, - data: { - pmi, - amount, - method: pending.originalRequestContext?.method, - capability: pending.originalRequestContext?.capability, + state.retryTimer = setTimeout( + () => { + state.retryTimer = undefined; + if (!lifecycle.isActive(state) || !state.request) return; + void transport.send(state.request).catch((error: unknown) => { + fail( + state, + new ClientPaymentError( + 'payment_retry_failed', + 'Failed to retry payment request', + ), + code, + { detail: error instanceof Error ? error.message : String(error) }, + ); + }); }, - }); + Math.max(delayMs, minRetryDelayMs), + ); }; - /** - * Handles explicit-gating -32042 (invoke `onPaymentRequired`, then retry) and - * -32043 (backoff, then retry). Both are intercepted here, never forwarded. - */ - async function handleExplicitPaymentError( - message: JSONRPCMessage, - requestEventId: string, - ): Promise { - if (isExplicitPaymentRequiredError(message)) { - const errorMsg = message; - const data = errorMsg.error.data as PaymentRequiredErrorData; + const priorOutcome = async ( + state: ClientPaymentRequest, + attempt: ClientPaymentAttempt, + explicit: boolean, + ): Promise => { + await attempt.completion; + if (!lifecycle.isActive(state)) return; + if (attempt.outcome === 'handled') { + if (explicit) + scheduleRetry(state, minRetryDelayMs, PAYMENT_PENDING_ERROR_CODE); + } else { + fail( + state, + new ClientPaymentError( + attempt.outcome === 'unknown' + ? 'payment_outcome_unknown' + : 'payment_previously_declined', + attempt.outcome === 'unknown' + ? 'Payment outcome is uncertain; reconcile before paying again' + : 'Payment was previously declined', + ), + explicit ? PAYMENT_REQUIRED_ERROR_CODE : -32000, + { outcome: attempt.outcome }, + ); + } + }; - if (!options.onPaymentRequired) { + async function explicitPayment(message: JSONRPCErrorResponse): Promise { + const state = message.id != null ? lifecycle.get(message.id) : undefined; + if (!state) { + onmessage?.(message); + return; + } + if (!lifecycle.isActive(state)) return; + lifecycle.startPayment(state); + if (message.error.code === PAYMENT_PENDING_ERROR_CODE) { + const data = parseClientPaymentPending(message.error.data); + if (!state.request) { + lifecycle.finish(state); onmessage?.(message); return; } - - const requestId = errorMsg.id; - const rawRequest = - requestId != null - ? rawRequestCache.get(JSON.stringify(requestId)) - : undefined; - if (!rawRequest) { - logger.warn( - 'missing raw original request, cannot retry explicit payment', - { requestEventId }, - ); + if (state.retryTimer !== undefined) return; + if (state.retries >= maxRetries) { + lifecycle.finish(state); onmessage?.(message); return; } - - logger.info('invoking onPaymentRequired for explicit gating', { - requestEventId, - optionsCount: data.payment_options.length, - }); - - try { - const result = await options.onPaymentRequired({ - options: data.payment_options, - instructions: data.instructions, - originalRequest: rawRequest, - }); - - if (result.paid) { - logger.info('explicit payment satisfied, retrying original request', { - requestEventId, - method: rawRequest.method, - }); - await transport.send(rawRequest); - return; - } - - logger.debug('onPaymentRequired returned paid=false', { - requestEventId, - reason: result.reason, - }); - synthesizePaymentError({ - id: errorMsg.id, - code: PAYMENT_REQUIRED_ERROR_CODE, - message: 'Payment Required', - data: { reason: result.reason || 'user_cancelled' }, - }); - } catch (err) { - logger.error('onPaymentRequired callback failed', { - requestEventId, - error: err instanceof Error ? err.message : String(err), - }); - synthesizePaymentError({ - id: errorMsg.id, - code: PAYMENT_REQUIRED_ERROR_CODE, - message: 'Payment Required', - data: { - reason: err instanceof Error ? err.message : String(err), - type: 'payment_handler_error', - }, - }); - } + const multiplier = Math.pow(1.5, state.retries++); + const baseDelay = (data.retry_after ?? 1) * 1000; + const delay = + baseDelay === 0 ? 0 : Math.min(baseDelay * multiplier, 10000); + scheduleRetry(state, delay, PAYMENT_PENDING_ERROR_CODE); return; } - - // -32043 Payment Pending - if (isExplicitPaymentPendingError(message)) { - const errorMsg = message; - const data = errorMsg.error.data as PaymentPendingErrorData; - const retryAfterSeconds = data.retry_after; - - const requestId = errorMsg.id; - const rawRequest = - requestId != null - ? rawRequestCache.get(JSON.stringify(requestId)) - : undefined; - if (!rawRequest) { - logger.warn( - 'missing raw original request, cannot retry explicit payment pending', - { requestEventId }, + const data = parseClientPaymentRequired(message.error.data); + if (!options.onPaymentRequired || !state.request) { + lifecycle.finish(state); + onmessage?.(message); + return; + } + const prior = attempts.find(data.payment_options); + if (prior) { + await priorOutcome(state, prior, true); + return; + } + if (state.paymentInFlight) { + throw new ClientPaymentError( + 'payment_outcome_unknown', + 'Another payment decision is still in progress for this invocation', + ); + } + const allowed = lifecycle.approve(state, data.payment_options); + const attempt = attempts.reserve(allowed, state.controller.signal); + state.paymentInFlight = true; + try { + const result = await options.onPaymentRequired({ + options: allowed, + instructions: data.instructions, + originalRequest: JSON.parse( + JSON.stringify(state.request), + ) as NonNullable, + abortSignal: state.controller.signal, + }); + if (!lifecycle.isActive(state)) return; + if (result.paid) { + attempt.complete('handled'); + scheduleRetry(state, minRetryDelayMs, PAYMENT_REQUIRED_ERROR_CODE); + } else { + attempt.complete('declined'); + fail( + state, + new ClientPaymentError( + result.reason || 'user_cancelled', + 'Payment Required', + ), + PAYMENT_REQUIRED_ERROR_CODE, ); - onmessage?.(message); - return; } + } catch (error: unknown) { + attempt.complete('unknown'); + if (lifecycle.isActive(state)) + fail( + state, + new ClientPaymentError( + error instanceof Error ? error.message : String(error), + 'Payment Required', + ), + PAYMENT_REQUIRED_ERROR_CODE, + { type: 'payment_handler_error', outcome: 'unknown' }, + ); + } finally { + state.paymentInFlight = false; + } + } - // Type-aware key: JSON.stringify keeps numeric 5 and string "5" - // distinct, and must be used consistently for both maps below. - const requestIdKey = JSON.stringify(errorMsg.id); - const retries = retryCounts.get(requestIdKey) ?? 0; - if (retries >= MAX_RETRIES) { - logger.error('max explicit payment retries exceeded', { - requestEventId, - id: requestIdKey, - maxRetries: MAX_RETRIES, - }); - onmessage?.(message); - return; - } - - retryCounts.set(requestIdKey, retries + 1); - - logger.info('payment pending, retrying after backoff', { - requestEventId, - retryAfterSeconds, - retryCount: retries + 1, - }); - - const baseDelayMs = (retryAfterSeconds ?? 1) * 1000; - const exponentialMultiplier = Math.pow(1.5, retries); - // Floor at minRetryDelayMs: a sub-second retry can produce a - // byte-identical Nostr event (created_at has second resolution), which - // relays and servers swallow as a duplicate. - const delayMs = Math.min( - Math.max(baseDelayMs * exponentialMultiplier, minRetryDelayMs), - 10000, + function transparentState( + eventId: string, + pending?: PendingRequest, + ): ClientPaymentRequest | undefined { + if (nostr && !pending) return undefined; + if (pending?.originalRequestId != null) + return lifecycle.correlated(pending.originalRequestId); + const state = lifecycle.onlyRequest(); + if (state) return state; + if (lifecycle.hasRequests()) { + onerror?.( + new ClientPaymentError( + 'payment_request_uncorrelated', + 'Payment notification is ambiguous between active requests', + ), ); - - const timer = setTimeout(() => { - pendingTimers.delete(timer); - transport.send(rawRequest).catch((err) => { - logger.error('failed to retry pending request', { - requestEventId, - error: err instanceof Error ? err.message : String(err), - }); - synthesizePaymentError({ - id: rawRequest.id, - code: PAYMENT_PENDING_ERROR_CODE, - message: 'Failed to retry pending request', - data: { - reason: err instanceof Error ? err.message : String(err), - }, - }); - }); - }, delayMs); - pendingTimers.add(timer); + return undefined; } + return lifecycle.detached(eventId); } - /** - * Handles transparent `notifications/payment_required`: satisfies the request - * in-band via configured handlers, gated by `paymentPolicy`, `canHandle`, and - * the effective-mode guard. - */ - async function handleTransparentPaymentRequired( - message: PaymentRequiredNotification, - requestEventId: string, + async function transparentPayment( + option: PaymentOption, + eventId: string, + state: ClientPaymentRequest, + pending?: PendingRequest, ): Promise { - const isNostrTransport = transport instanceof NostrClientTransport; - - const pending = isNostrTransport - ? transport.getPendingRequestForEventId(requestEventId) - : undefined; - - if (isNostrTransport && !pending) { - logger.warn('dropping uncorrelated payment_required notification', { - requestEventId, - pmi: message.params.pmi, - amount: message.params.amount, - }); - return; - } - - // CEP-8: a client that required explicit_gating SHOULD NOT auto-satisfy a - // transparent payment_required when the server did not accept it. + if (!lifecycle.isActive(state)) return; if ( - isNostrTransport && + nostr && options.paymentInteraction === 'explicit_gating' && - transport.getEffectivePaymentInteraction() !== 'explicit_gating' + nostr.getEffectivePaymentInteraction() !== 'explicit_gating' ) { - logger.warn( - 'declining transparent payment_required: explicit_gating was not accepted by the server', - { requestEventId, pmi: message.params.pmi }, - ); - synthesizePaymentDecline( - pending, - 'Payment declined: explicit_gating was not accepted by the server', - message.params.pmi, - message.params.amount, - ); - return; - } - - // If the transport can provide the original request's progressToken, emit synthetic - // progress notifications locally to keep the upstream MCP request alive while the - // payment settles (CEP-8 TTL can exceed the default MCP timeout). - if (isNostrTransport) { - const token = pending?.progressToken; - // Fall back to defaultPaymentTtlMs when the server omits ttl so the client - // keeps the MCP request alive for the same duration the server will wait. - const ttlSeconds = - message.params.ttl !== undefined - ? message.params.ttl - : defaultPaymentTtlMs / 1000; - if (token && Number.isFinite(ttlSeconds)) { - // Guard against nonsensical TTLs. - const ttlMs = Math.floor(ttlSeconds * 1000); - if (ttlMs > 0 && !syntheticProgress.has(token)) { - const stopAtMs = Date.now() + ttlMs; - - const wireProgressToken = Number.isFinite(Number(token)) - ? Number(token) - : token; - syntheticProgress.set(token, { - stopAtMs, - wireProgressToken, - }); - ensureSchedulerStarted(); - - // Reset the MCP timeout immediately — don't wait for the interval tick - onmessage?.({ - jsonrpc: '2.0', - method: 'notifications/progress', - params: { - progressToken: wireProgressToken, - progress: 0, - }, - } as JSONRPCNotification); - - logger.debug('started synthetic progress', { - requestEventId, - progressToken: token, - ttlSeconds, - intervalMs: syntheticProgressIntervalMs, - }); - } - } - } - - // Resolve an in-band handler for this PMI. If none matches, the payment is - // left to the application: the notification was already forwarded via - // onmessage, synthetic progress above keeps the request alive while it - // settles out-of-band, and the server's TTL is the ultimate timeout. - const handler = handlersByPmi.get(message.params.pmi); - if (!handler) { - logger.debug( - 'no in-band handler for PMI; leaving payment to the application', + fail( + state, + new ClientPaymentError( + 'payment_negotiation_failed', + 'Payment declined: explicit_gating was not accepted by the server', + ), + -32000, { - pmi: message.params.pmi, - requestEventId, + pmi: option.pmi, + amount: option.amount, + method: pending?.originalRequestContext?.method, + capability: pending?.originalRequestContext?.capability, }, ); return; } - - // Best-effort client-side dedupe keyed by pay_req. - // IMPORTANT: claim synchronously before any await to avoid double-pay races. - if (inFlightPayReqs.has(message.params.pay_req)) { - logger.debug('duplicate pay_req detected, skipping', { - payReq: message.params.pay_req.substring(0, 20) + '...', - requestEventId, - }); + lifecycle.startPayment(state); + if (nostr && pending?.progressToken) + progress.start( + state.key, + pending.progressToken, + option.ttl !== undefined ? option.ttl * 1000 : defaultTtlMs, + ); + const handler = handlers.get(option.pmi); + if (!handler || !lifecycle.isActive(state)) return; + if (limits && state.id === undefined) + throw new ClientPaymentError( + 'payment_request_uncorrelated', + 'Payment limits require a correlated original request', + ); + const prior = attempts.find([option]); + if (prior) { + await priorOutcome(state, prior, false); return; } - - inFlightPayReqs.add(message.params.pay_req); - + if (state.paymentInFlight) + throw new ClientPaymentError( + 'payment_outcome_unknown', + 'Another payment decision is still in progress for this invocation', + ); + lifecycle.approve(state, [option]); + const attempt = attempts.reserve([option], state.controller.signal); + state.paymentInFlight = true; const req: PaymentHandlerRequest = { - amount: message.params.amount, - pay_req: message.params.pay_req, - pmi: message.params.pmi, - description: message.params.description, - ttl: message.params.ttl, - _meta: message.params._meta, - requestEventId, + amount: option.amount, + pmi: option.pmi, + pay_req: option.pay_req, + description: option.description, + ttl: option.ttl, + _meta: option._meta, + requestEventId: eventId, + abortSignal: state.controller.signal, }; - - const synthesizeClientDeclineError = (params: { - message: string; - }): void => { - if (pending?.progressToken) { - stopSyntheticProgress(pending.progressToken); - } - synthesizePaymentDecline(pending, params.message, req.pmi, req.amount); + const decline = (message: string): void => { + attempt.complete('declined'); + fail(state, new ClientPaymentError('payment_declined', message), -32000, { + pmi: option.pmi, + amount: option.amount, + method: pending?.originalRequestContext?.method, + capability: pending?.originalRequestContext?.capability, + }); }; - + let walletStarted = false; + const expiresAt = + option.ttl !== undefined ? Date.now() + option.ttl * 1000 : undefined; try { - logger.info('processing payment_required', { - requestEventId, - pmi: message.params.pmi, - amount: message.params.amount, - }); - if (options.paymentPolicy) { - const isApproved = await options.paymentPolicy( - req, + const approved = await options.paymentPolicy( + { ...req }, pending?.originalRequestContext, ); - if (!isApproved) { - logger.debug('paymentPolicy declined the payment', { - requestEventId, - pmi: message.params.pmi, - amount: message.params.amount, - }); - synthesizeClientDeclineError({ - message: 'Payment declined by client policy', - }); + if (!lifecycle.isActive(state)) return; + if (!approved) { + decline('Payment declined by client policy'); return; } } - - const canHandle = handler.canHandle ? await handler.canHandle(req) : true; + const canHandle = handler.canHandle + ? await handler.canHandle({ ...req }) + : await Promise.resolve(true); + if (!lifecycle.isActive(state)) return; if (!canHandle) { - logger.debug('handler declined to handle', { - requestEventId, - pmi: message.params.pmi, - }); - synthesizeClientDeclineError({ - message: 'Payment declined by client handler', - }); + decline('Payment declined by client handler'); return; } - - logger.debug('invoking payment handler', { - requestEventId, - handler: handler.constructor.name, - pmi: message.params.pmi, - }); - + if (expiresAt !== undefined && Date.now() >= expiresAt) { + attempt.complete('declined'); + throw new ClientPaymentError( + 'payment_offer_expired', + 'Payment offer expired before wallet action', + ); + } + walletStarted = true; await handler.handle(req); - - logger.info('payment handler completed successfully', { - requestEventId, - pmi: message.params.pmi, - }); - } catch (error) { - logger.error('payment handler failed', { - requestEventId, - pmi: message.params.pmi, - error: error instanceof Error ? error.message : String(error), - }); - // Same-severity outcomes must behave the same: policy and canHandle - // declines resolve the pending request with a synthesized error. A - // handler crash must too, or the MCP request hangs until the server TTL. - synthesizeClientDeclineError({ - message: `Payment handler failed: ${ - error instanceof Error ? error.message : String(error) - }`, - }); - throw error; + if (lifecycle.isActive(state)) attempt.complete('handled'); + } catch (error: unknown) { + attempt.complete(walletStarted ? 'unknown' : 'declined'); + if (!lifecycle.isActive(state)) return; + const failure = error instanceof Error ? error : new Error(String(error)); + fail( + state, + error instanceof ClientPaymentError + ? error + : new ClientPaymentError( + 'payment_handler_error', + `Payment handler failed: ${failure.message}`, + ), + -32000, + { + pmi: option.pmi, + amount: option.amount, + outcome: walletStarted ? 'unknown' : 'declined', + }, + ); + if (state.id !== undefined && !(error instanceof ClientPaymentError)) + onerror?.(failure); } finally { - inFlightPayReqs.delete(message.params.pay_req); + state.paymentInFlight = false; } } - /** Classifies an inbound payment message and delegates to the relevant handler. */ - async function maybeHandlePaymentRequired( - message: JSONRPCMessage, - requestEventId: string, - ): Promise { - if ( - isExplicitPaymentRequiredError(message) || - isExplicitPaymentPendingError(message) - ) { - await handleExplicitPaymentError(message, requestEventId); + function dispatch(message: JSONRPCMessage, eventId: string): void { + if (lifecycle.isClosed) return; + const cancelled = cancelledRequestId(message); + if (cancelled !== undefined) { + const state = lifecycle.get(cancelled); + if (state) lifecycle.finish(state); + onmessage?.(message); return; } - if (isPaymentRequiredNotification(message)) { - await handleTransparentPaymentRequired(message, requestEventId); + if (isExplicitPaymentError(message)) { + const state = message.id != null ? lifecycle.get(message.id) : undefined; + void explicitPayment(message).catch((error: unknown) => { + const failure = + error instanceof ClientPaymentError + ? error + : new ClientPaymentError( + 'invalid_payment', + 'Invalid payment message', + ); + if (state) fail(state, failure, message.error.code); + else onerror?.(failure); + }); + return; + } + if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) { + const state = message.id != null ? lifecycle.get(message.id) : undefined; + if (state && !state.active) return; + if (state) lifecycle.finish(state); + onmessage?.(message); + return; + } + if (isPaymentOffer(message)) { + const pending = nostr?.getPendingRequestForEventId(eventId); + const state = transparentState(eventId, pending); + if (!state || !lifecycle.isActive(state)) { + onmessage?.(message); + return; + } + let option: PaymentOption; + try { + option = parseClientPaymentOption(message.params); + } catch (error: unknown) { + fail(state, error as ClientPaymentError); + return; + } + void transparentPayment(option, eventId, state, pending).catch( + (error: unknown) => { + fail( + state, + error instanceof ClientPaymentError + ? error + : new ClientPaymentError('payment_handler_error', String(error)), + ); + }, + ); + onmessage?.(message); return; } - } - - /** - * Runs the payment handler, then forwards to the upstream consumer unless the - * message is an explicit-gating error (those are re-emitted/retried internally). - */ - const dispatchAndForward = ( - message: JSONRPCMessage, - requestEventId: string, - ): void => { - void maybeHandlePaymentRequired(message, requestEventId).catch( - (err: unknown) => { - const error = err instanceof Error ? err : new Error(String(err)); - onerror?.(error); - }, - ); if ( - isExplicitPaymentRequiredError(message) || - isExplicitPaymentPendingError(message) + isJSONRPCNotification(message) && + (message.method === PAYMENT_ACCEPTED_METHOD || + message.method === PAYMENT_REJECTED_METHOD) ) { - return; + const pending = nostr?.getPendingRequestForEventId(eventId); + const state = + pending?.originalRequestId != null + ? lifecycle.correlated(pending.originalRequestId) + : lifecycle.onlyRequest(); + if (state) { + progress.stop(state.key); + if (message.method === PAYMENT_REJECTED_METHOD) { + const detail = + typeof message.params?.message === 'string' + ? message.params.message + : undefined; + fail( + state, + new ClientPaymentError( + 'payment_rejected', + detail ? `Payment rejected: ${detail}` : 'Payment rejected', + ), + ); + return; + } + } } onmessage?.(message); - }; + } - const wrapped = { + const wrapped: Transport = { get onmessage() { return onmessage; }, @@ -798,117 +587,60 @@ export function withClientPayments( set onclose(fn) { onclose = fn; }, - async start(): Promise { - // Only suppress notifications in `onmessage` when the transport delivers - // them through a separate context path (NostrClientTransport). - const hasContextPath = supportsOnmessageWithContext(transport); - - transport.onmessage = (message: JSONRPCMessage) => { - // `NostrClientTransport` delivers notifications through BOTH `onmessage` - // and `onmessageWithContext`. Forward only from the context path to avoid - // duplicate delivery to the upstream MCP Protocol. - if (hasContextPath && isJSONRPCNotification(message)) { - return; - } - - // Stop synthetic progress when the server responds (success or error). - // message.id was restored to originalRequestId by resolveResponse(). - // When onprogress was set, originalRequestId = progressToken, making this stop exact. - // Otherwise syntheticProgress has no entry and the call is a no-op. - if ( - isJSONRPCResultResponse(message) || - isJSONRPCErrorResponse(message) - ) { - stopSyntheticProgress(String(message.id)); - if ( - !isExplicitPaymentRequiredError(message) && - !isExplicitPaymentPendingError(message) - ) { - const reqId = message.id as string | number; - rawRequestCache.delete(JSON.stringify(reqId)); - retryCounts.delete(JSON.stringify(reqId)); - } - } - - if (hasContextPath) { - return; - } - - // Best-effort: execute handler asynchronously, but never block delivery. - dispatchAndForward(message, 'unknown'); + lifecycle.open(); + const hasContext = Object.prototype.hasOwnProperty.call( + transport, + 'onmessageWithContext', + ); + transport.onmessage = (message): void => { + if (!hasContext) dispatch(message, 'unknown'); }; - - if (hasContextPath) { - transport.onmessageWithContext = ( - message: JSONRPCMessage, - ctx: { eventId: string; correlatedEventId?: string }, - ) => { - const requestEventId = ctx.correlatedEventId ?? 'unknown'; - - // Stop synthetic progress on terminal outcomes (context path). - if (isJSONRPCNotification(message)) { - const isAccepted = message.method === PAYMENT_ACCEPTED_METHOD; - const isRejected = message.method === PAYMENT_REJECTED_METHOD; - if (isAccepted || isRejected) { - const pending = - transport instanceof NostrClientTransport - ? transport.getPendingRequestForEventId(requestEventId) - : undefined; - if (pending?.progressToken) { - stopSyntheticProgress(pending.progressToken); - } - // On rejection, synthesize a JSON-RPC error response so the caller's - // pending promise is rejected immediately rather than timing out after 60 s. - // The server never calls forward() for a rejected payment, so no real - // response will arrive. - if (isRejected && pending?.originalRequestId != null) { - const rejMsg = (message as PaymentRejectedNotification).params - ?.message; - onmessage?.({ - jsonrpc: '2.0', - id: pending.originalRequestId, - error: { - code: -32000, - message: rejMsg - ? `Payment rejected: ${rejMsg}` - : 'Payment rejected', - }, - } as JSONRPCMessage); - return; - } - } - } - - // Forward exactly once (see duplicate-delivery guard in `transport.onmessage`). - dispatchAndForward(message, requestEventId); + if (hasContext) + (transport as TransportWithContext).onmessageWithContext = ( + message, + ctx, + ): void => { + dispatch(message, ctx.correlatedEventId ?? 'unknown'); }; - } - - transport.onerror = (err: Error) => onerror?.(err); - transport.onclose = () => { - disposeClientState(); + transport.onerror = (error): void => onerror?.(error); + transport.onclose = (): void => { + lifecycle.close(); + progress.clear(); onclose?.(); }; - await transport.start(); + try { + await transport.start(); + } catch (error) { + lifecycle.close(); + progress.clear(); + throw error; + } }, - async send(message: JSONRPCMessage): Promise { - if ('method' in message && 'id' in message && message.id != null) { - rawRequestCache.set( - JSON.stringify(message.id), - message as JSONRPCRequest, - ); + if (lifecycle.isClosed) + throw new Error('Client payments transport is closed'); + const cancelled = cancelledRequestId(message); + if (cancelled !== undefined) { + const state = lifecycle.get(cancelled); + if (state) lifecycle.finish(state); + } + const state = isJSONRPCRequest(message) + ? lifecycle.register(message) + : undefined; + try { + await transport.send(message); + } catch (error) { + if (state) lifecycle.finish(state); + throw error; } - await transport.send(message); }, - async close(): Promise { - // disposeClientState is called via transport.onclose, no need to call it here + lifecycle.close(); + progress.clear(); await transport.close(); }, }; - clientPaymentsWrapped.add(transport); clientPaymentsWrapped.add(wrapped); return wrapped; diff --git a/src/payments/types.ts b/src/payments/types.ts index f9fe83d..20c132c 100644 --- a/src/payments/types.ts +++ b/src/payments/types.ts @@ -158,6 +158,8 @@ export interface PaymentHandlerRequest { /** Transparency metadata from the server's `payment_required._meta`. */ _meta?: Record; requestEventId: string; + /** Aborted when the invocation ends; check before starting an awaited wallet action. */ + abortSignal?: AbortSignal; } /** diff --git a/src/transport/nostr-client-transport.ts b/src/transport/nostr-client-transport.ts index 39492a0..226d0ae 100644 --- a/src/transport/nostr-client-transport.ts +++ b/src/transport/nostr-client-transport.ts @@ -42,8 +42,10 @@ import { ServerMetadataStore } from './nostr-client/server-metadata-store.js'; import { ClientOutboundSender } from './nostr-client/outbound-sender.js'; import { ClientInboundNotificationDispatcher } from './nostr-client/inbound-notification-dispatcher.js'; import { ClientEventPipeline } from './nostr-client/event-pipeline.js'; -import { ClientOpenStreamFactory, - type ClientOpenStreamHandle,} from './nostr-client/open-stream-factory.js'; +import { + ClientOpenStreamFactory, + type ClientOpenStreamHandle, +} from './nostr-client/open-stream-factory.js'; import { DEFAULT_CHUNK_SIZE, DEFAULT_OVERSIZED_THRESHOLD, @@ -458,6 +460,11 @@ export class NostrClientTransport return this.correlationStore.getPendingRequest(eventId); } + /** Releases correlation for every wire retry of a terminal caller request. */ + public forgetPendingRequestsForRequestId(requestId: string | number): void { + this.correlationStore.removeRequestsForRequestId(requestId); + } + /** * Returns the CEP-41 stream session for a progress token, creating it lazily if needed. */ diff --git a/src/transport/nostr-client/correlation-store.test.ts b/src/transport/nostr-client/correlation-store.test.ts index dea7948..d97c9f7 100644 --- a/src/transport/nostr-client/correlation-store.test.ts +++ b/src/transport/nostr-client/correlation-store.test.ts @@ -74,6 +74,27 @@ describe('ClientCorrelationStore', () => { }); }); + it('releases all retries of a terminal request while preserving typed id isolation', () => { + const store = new ClientCorrelationStore(); + store.registerRequest('numeric-first', { + originalRequestId: 0, + isInitialize: false, + }); + store.registerRequest('numeric-retry', { + originalRequestId: 0, + isInitialize: false, + }); + store.registerRequest('string', { + originalRequestId: '0', + isInitialize: false, + }); + store.removeRequestsForRequestId(0); + expect(store.hasPendingRequest('numeric-first')).toBe(false); + expect(store.hasPendingRequest('numeric-retry')).toBe(false); + expect(store.hasPendingRequest('string')).toBe(true); + expect(store.pendingRequestCount).toBe(1); + }); + describe('clear', () => { it('removes all pending requests', () => { const store = new ClientCorrelationStore(); diff --git a/src/transport/nostr-client/correlation-store.ts b/src/transport/nostr-client/correlation-store.ts index f422bac..4a89c58 100644 --- a/src/transport/nostr-client/correlation-store.ts +++ b/src/transport/nostr-client/correlation-store.ts @@ -109,6 +109,14 @@ export class ClientCorrelationStore { return this.pendingRequests.delete(eventId); } + /** Removes all wire retries of a terminal or cancelled caller request. */ + removeRequestsForRequestId(requestId: string | number): void { + for (const [eventId, request] of this.pendingRequests.entries()) { + if (request.originalRequestId === requestId) + this.pendingRequests.delete(eventId); + } + } + /** Gets the current number of pending requests */ get pendingRequestCount(): number { return this.pendingRequests.size; diff --git a/src/transport/payment-client-reliability.e2e.test.ts b/src/transport/payment-client-reliability.e2e.test.ts new file mode 100644 index 0000000..169980e --- /dev/null +++ b/src/transport/payment-client-reliability.e2e.test.ts @@ -0,0 +1,149 @@ +import { describe, expect, test } from 'bun:test'; +import { Client } from '@contextvm/mcp-sdk/client'; +import { McpServer } from '@contextvm/mcp-sdk/server/mcp'; +import { z } from 'zod'; +import { generateSecretKey, getPublicKey } from 'nostr-tools/pure'; +import { bytesToHex } from 'nostr-tools/utils'; +import { MockRelayHub } from '../__mocks__/mock-relay-handler.js'; +import { EncryptionMode, GiftWrapMode } from '../core/interfaces.js'; +import { PrivateKeySigner } from '../signer/private-key-signer.js'; +import { NostrClientTransport } from './nostr-client-transport.js'; +import { NostrServerTransport } from './nostr-server-transport.js'; +import { withClientPayments, withServerPayments } from '../payments/index.js'; +import type { PaymentProcessor } from '../payments/types.js'; + +const cases = [ + { + name: 'cleartext', + encryptionMode: EncryptionMode.DISABLED, + giftWrapMode: GiftWrapMode.OPTIONAL, + kind: 25910, + }, + { + name: 'persistent encryption', + encryptionMode: EncryptionMode.REQUIRED, + giftWrapMode: GiftWrapMode.PERSISTENT, + kind: 1059, + }, + { + name: 'ephemeral encryption', + encryptionMode: EncryptionMode.REQUIRED, + giftWrapMode: GiftWrapMode.EPHEMERAL, + kind: 21059, + }, +]; + +describe('instant explicit payment retry delivery', () => { + for (const mode of cases) { + test(`instant payment executes once over ${mode.name} without callback sleeps`, async () => { + const hub = new MockRelayHub(); + const serverKey = generateSecretKey(); + const clientSigner = new PrivateKeySigner( + bytesToHex(generateSecretKey()), + ); + const requests: Array<{ id: string; createdAt: number }> = []; + const sign = clientSigner.signEvent.bind(clientSigner); + clientSigner.signEvent = async (event) => { + const signed = await sign(event); + if ( + event.kind === 25910 && + JSON.parse(event.content).method === 'tools/call' + ) { + requests.push({ id: signed.id, createdAt: event.created_at }); + } + return signed; + }; + let executions = 0; + let invoices = 0; + let approvals = 0; + const processor: PaymentProcessor = { + pmi: 'fake', + async createPaymentRequired(params) { + invoices++; + return { + amount: params.amount, + pmi: 'fake', + pay_req: `invoice-${invoices}`, + ttl: 60, + }; + }, + async verifyPayment() { + return {}; + }, + }; + const server = new McpServer({ + name: 'payment-retry-server', + version: '1.0.0', + }); + server.registerTool( + 'paid', + { inputSchema: { value: z.number() } }, + async ({ value }) => { + executions++; + return { content: [{ type: 'text', text: String(value) }] }; + }, + ); + const serverTransport = withServerPayments( + new NostrServerTransport({ + signer: new PrivateKeySigner(bytesToHex(serverKey)), + relayHandler: hub.createRelayHandler(), + encryptionMode: mode.encryptionMode, + giftWrapMode: mode.giftWrapMode, + }), + { + processors: [processor], + pricedCapabilities: [ + { + method: 'tools/call', + name: 'paid', + amount: 5, + currencyUnit: 'sats', + }, + ], + }, + ); + const base = new NostrClientTransport({ + signer: clientSigner, + relayHandler: hub.createRelayHandler(), + serverPubkey: getPublicKey(serverKey), + encryptionMode: mode.encryptionMode, + giftWrapMode: mode.giftWrapMode, + }); + const paid = withClientPayments(base, { + paymentInteraction: 'explicit_gating', + minRetryDelayMs: 0, + async onPaymentRequired() { + approvals++; + return { paid: true }; + }, + }); + const client = new Client({ + name: 'payment-retry-client', + version: '1.0.0', + }); + try { + await server.connect(serverTransport); + await client.connect(paid); + // Initialization finishes negotiation, so routine invocations have stable tags. + const result = await client.callTool( + { name: 'paid', arguments: { value: 7 } }, + undefined, + { timeout: 5000 }, + ); + expect(result.content).toEqual([{ type: 'text', text: '7' }]); + expect(approvals).toBe(1); + expect(invoices).toBe(1); + expect(executions).toBe(1); + expect(requests).toHaveLength(2); + expect(requests[1]!.id).not.toBe(requests[0]!.id); + expect(requests[1]!.createdAt).toBeGreaterThan(requests[0]!.createdAt); + expect(hub.getEvents().some((event) => event.kind === mode.kind)).toBe( + true, + ); + } finally { + await client.close(); + await server.close(); + } + }); + } +}); diff --git a/src/transport/payments-flow.test.ts b/src/transport/payments-flow.test.ts index 3bbd471..0a52d24 100644 --- a/src/transport/payments-flow.test.ts +++ b/src/transport/payments-flow.test.ts @@ -90,13 +90,8 @@ async function captureNextCtxvmEvent(params: { } /** - * Nostr event ids are derived from (content, author, created_at, tags) at - * second granularity. An explicit-gating retry republishes the same JSON-RPC - * request; if it lands in the same second as the original, the identical event - * id is deduplicated by relay subscriptions and the server never sees the - * retry. Sleep past the current second boundary so the retry's event id - * differs. Real-world payment flows naturally take >1s (wallet interaction, - * settlement); this only affects instant-pay tests. + * Separates intentional test invocations at Nostr second granularity. + * Automatic payment retries are delayed by the client payment lifecycle. */ async function sleepPastSecondBoundary(): Promise { const waitMs = 1000 - (Date.now() % 1000) + 10; @@ -217,11 +212,11 @@ describe.serial('payments fake flow (transport-level)', () => { ); const processorB = new FakePaymentProcessor({ - pmi: 'pmi:B', + pmi: 'pmi-b', verifyDelayMs: 1, }); const processorC = new FakePaymentProcessor({ - pmi: 'pmi:C', + pmi: 'pmi-c', verifyDelayMs: 1, }); const pricedCapabilities = [ @@ -252,8 +247,8 @@ describe.serial('payments fake flow (transport-level)', () => { const clientPrivateKey = bytesToHex(clientSK); const clientPubkey = getPublicKey(clientSK); const handlers = [ - new FakePaymentHandler({ pmi: 'pmi:A', delayMs: 1 }), - new FakePaymentHandler({ pmi: 'pmi:B', delayMs: 1 }), + new FakePaymentHandler({ pmi: 'pmi-a', delayMs: 1 }), + new FakePaymentHandler({ pmi: 'pmi-b', delayMs: 1 }), ]; const clientTransport = new NostrClientTransport({ @@ -271,7 +266,7 @@ describe.serial('payments fake flow (transport-level)', () => { if (!event.content.includes('"method":"initialize"')) { return false; } - return event.tags.some((t) => t[0] === 'pmi' && t[1] === 'pmi:A'); + return event.tags.some((t) => t[0] === 'pmi' && t[1] === 'pmi-a'); }, }); @@ -285,8 +280,8 @@ describe.serial('payments fake flow (transport-level)', () => { const initializeEvent = await initializeEventPromise; const pmiTags = initializeEvent.tags.filter((t) => t[0] === 'pmi'); expect(pmiTags).toEqual([ - ['pmi', 'pmi:A'], - ['pmi', 'pmi:B'], + ['pmi', 'pmi-a'], + ['pmi', 'pmi-b'], ]); const originalOnMessage = paidClientTransport.onmessage; @@ -300,7 +295,7 @@ describe.serial('payments fake flow (transport-level)', () => { arguments: { a: 1, b: 2 }, }); - expect(observedPaymentPmi).toBe('pmi:B'); + expect(observedPaymentPmi).toBe('pmi-b'); await client.close(); await mcpServer.close(); @@ -352,7 +347,7 @@ describe.serial('payments fake flow (transport-level)', () => { const clientSK = generateSecretKey(); const clientPrivateKey = bytesToHex(clientSK); const clientPubkey = getPublicKey(clientSK); - const handlers = [new FakePaymentHandler({ pmi: 'pmi:A', delayMs: 1 })]; + const handlers = [new FakePaymentHandler({ pmi: 'pmi-a', delayMs: 1 })]; const clientTransport = new NostrClientTransport({ signer: new PrivateKeySigner(clientPrivateKey), @@ -374,7 +369,7 @@ describe.serial('payments fake flow (transport-level)', () => { if (!event.content.includes('"method":"tools/call"')) { return false; } - return event.tags.some((t) => t[0] === 'pmi' && t[1] === 'pmi:A'); + return event.tags.some((t) => t[0] === 'pmi' && t[1] === 'pmi-a'); }, timeoutMs: 4000, }); @@ -388,7 +383,7 @@ describe.serial('payments fake flow (transport-level)', () => { const callEvent = await callEventPromise; const pmiTags = callEvent.tags.filter((t) => t[0] === 'pmi'); - expect(pmiTags).toEqual([['pmi', 'pmi:A']]); + expect(pmiTags).toEqual([['pmi', 'pmi-a']]); await client.close(); await mcpServer.close(); @@ -413,11 +408,11 @@ describe.serial('payments fake flow (transport-level)', () => { ); const processorB = new FakePaymentProcessor({ - pmi: 'pmi:B', + pmi: 'pmi-b', verifyDelayMs: 1, }); const processorC = new FakePaymentProcessor({ - pmi: 'pmi:C', + pmi: 'pmi-c', verifyDelayMs: 1, }); const pricedCapabilities = [ @@ -445,7 +440,7 @@ describe.serial('payments fake flow (transport-level)', () => { await mcpServer.connect(serverTransport); const clientPrivateKey = bytesToHex(generateSecretKey()); - const handlers = [new FakePaymentHandler({ pmi: 'pmi:A', delayMs: 1 })]; + const handlers = [new FakePaymentHandler({ pmi: 'pmi-a', delayMs: 1 })]; const clientTransport = new NostrClientTransport({ signer: new PrivateKeySigner(clientPrivateKey), @@ -472,7 +467,7 @@ describe.serial('payments fake flow (transport-level)', () => { arguments: { a: 1, b: 2 }, }); - expect(observedPaymentPmi).toBe('pmi:B'); + expect(observedPaymentPmi).toBe('pmi-b'); await client.close(); await mcpServer.close(); @@ -497,11 +492,11 @@ describe.serial('payments fake flow (transport-level)', () => { ); const processorB = new FakePaymentProcessor({ - pmi: 'pmi:B', + pmi: 'pmi-b', verifyDelayMs: 1, }); const processorC = new FakePaymentProcessor({ - pmi: 'pmi:C', + pmi: 'pmi-c', verifyDelayMs: 1, }); const pricedCapabilities = [ @@ -553,7 +548,7 @@ describe.serial('payments fake flow (transport-level)', () => { arguments: { a: 1, b: 2 }, }); - expect(observedPaymentPmi).toBe('pmi:B'); + expect(observedPaymentPmi).toBe('pmi-b'); await client.close(); await mcpServer.close(); @@ -1091,7 +1086,6 @@ describe.serial('payments fake flow (transport-level)', () => { handlers: [], paymentInteraction: 'explicit_gating', onPaymentRequired: async () => { - await sleepPastSecondBoundary(); explicitPaymentHandled = true; return { paid: true }; }, @@ -1268,7 +1262,6 @@ describe.serial('payments fake flow (transport-level)', () => { handlers: [], paymentInteraction: 'explicit_gating', onPaymentRequired: async () => { - await sleepPastSecondBoundary(); return { paid: true }; }, }); @@ -1548,7 +1541,6 @@ describe.serial('payments fake flow (transport-level)', () => { handlers: [], paymentInteraction: 'explicit_gating', onPaymentRequired: async () => { - await sleepPastSecondBoundary(); return { paid: true }; }, }); @@ -1646,7 +1638,6 @@ describe.serial('payments fake flow (transport-level)', () => { handlers: [], paymentInteraction: 'explicit_gating', onPaymentRequired: async () => { - await sleepPastSecondBoundary(); return { paid: true }; }, }); @@ -1976,7 +1967,6 @@ describe.serial('payments fake flow (transport-level)', () => { handlers: [], paymentInteraction: 'explicit_gating', onPaymentRequired: async () => { - await sleepPastSecondBoundary(); explicitHandled = true; return { paid: true }; }, @@ -2082,7 +2072,6 @@ describe.serial('payments fake flow (transport-level)', () => { handlers: [], paymentInteraction: 'explicit_gating', onPaymentRequired: async () => { - await sleepPastSecondBoundary(); explicitHandled = true; return { paid: true }; }, @@ -2273,7 +2262,6 @@ describe.serial('payments fake flow (transport-level)', () => { handlers: [], paymentInteraction: 'explicit_gating', onPaymentRequired: async () => { - await sleepPastSecondBoundary(); return { paid: true }; }, }); @@ -2413,7 +2401,6 @@ describe.serial('payments fake flow (transport-level)', () => { handlers: [], paymentInteraction: 'explicit_gating', onPaymentRequired: async () => { - await sleepPastSecondBoundary(); explicitHandled = true; return { paid: true }; }, From 9d2990c40514bff924bb5a70ab4a3c1674cc7f0c Mon Sep 17 00:00:00 2001 From: Khushvendra Date: Tue, 27 Oct 2026 02:26:05 +0530 Subject: [PATCH 2/2] fix(payments): pin unresolved attempts and restore client compatibility --- src/payments/client-payment-attempts.test.ts | 95 +++++++++++++++++++ src/payments/client-payment-attempts.ts | 28 ++++-- src/payments/client-payment-options.ts | 7 +- src/payments/client-payment-validation.ts | 10 +- src/payments/client-payments-limits.test.ts | 61 +++++++++++- .../client-payments-reliability.test.ts | 33 +++---- src/payments/client-payments.ts | 31 +++++- .../payment-client-reliability.e2e.test.ts | 8 +- 8 files changed, 237 insertions(+), 36 deletions(-) create mode 100644 src/payments/client-payment-attempts.test.ts diff --git a/src/payments/client-payment-attempts.test.ts b/src/payments/client-payment-attempts.test.ts new file mode 100644 index 0000000..8d72853 --- /dev/null +++ b/src/payments/client-payment-attempts.test.ts @@ -0,0 +1,95 @@ +import { describe, expect, test } from 'bun:test'; +import { ClientPaymentAttempts } from './client-payment-attempts.js'; +import { option } from '../__mocks__/client-payments-harness.js'; + +describe('client payment attempt retention under capacity pressure', () => { + for (const outcome of ['handled', 'declined'] as const) { + test(`evicts ${outcome} history least recently used first`, () => { + const attempts = new ClientPaymentAttempts(2); + const signal = new AbortController().signal; + const first = attempts.reserve([option('first')], signal); + first.complete(outcome); + const second = attempts.reserve([option('second')], signal); + second.complete(outcome); + + expect(attempts.find([option('first')])).toBe(first); + const third = attempts.reserve([option('third')], signal); + expect(attempts.find([option('first')])).toBe(first); + expect(attempts.find([option('second')])).toBeUndefined(); + expect(attempts.find([option('third')])).toBe(third); + third.complete(outcome); + }); + } + + test('skips in-flight and unknown pins when evicting older history', () => { + const attempts = new ClientPaymentAttempts(3); + const signal = new AbortController().signal; + const inFlight = attempts.reserve([option('pending')], signal); + const unknown = attempts.reserve([option('unknown')], signal); + unknown.complete('unknown'); + const handled = attempts.reserve([option('handled')], signal); + handled.complete('handled'); + + const replacement = attempts.reserve([option('replacement')], signal); + expect(attempts.find([option('handled')])).toBeUndefined(); + expect(attempts.find([option('pending')])).toBe(inFlight); + expect(attempts.find([option('unknown')])).toBe(unknown); + expect(attempts.find([option('replacement')])).toBe(replacement); + inFlight.complete('handled'); + replacement.complete('handled'); + }); + + test('refuses atomically when a multi-option reservation cannot fit beside pins', () => { + const attempts = new ClientPaymentAttempts(2); + const signal = new AbortController().signal; + const unknown = attempts.reserve([option('unknown')], signal); + unknown.complete('unknown'); + const handled = attempts.reserve([option('handled')], signal); + handled.complete('handled'); + + expect(() => + attempts.reserve([option('new-one'), option('new-two')], signal), + ).toThrow(expect.objectContaining({ reason: 'payment_attempt_capacity' })); + expect(attempts.find([option('unknown')])).toBe(unknown); + expect(attempts.find([option('handled')])).toBe(handled); + expect( + attempts.find([option('new-one'), option('new-two')]), + ).toBeUndefined(); + }); + + test('pins every distinct option of an in-flight or unknown explicit attempt', async () => { + const attempts = new ClientPaymentAttempts(2); + const controller = new AbortController(); + const attempt = attempts.reserve( + [option('one'), option('one'), option('two')], + controller.signal, + ); + expect(attempts.find([option('one')])).toBe(attempt); + expect(attempts.find([option('two')])).toBe(attempt); + expect(() => + attempts.reserve([option('three')], controller.signal), + ).toThrow(expect.objectContaining({ reason: 'payment_attempt_capacity' })); + + controller.abort(); + await attempt.completion; + attempt.complete('handled'); + expect(attempt.outcome).toBe('unknown'); + expect(() => + attempts.reserve([option('three')], new AbortController().signal), + ).toThrow(expect.objectContaining({ reason: 'payment_attempt_capacity' })); + expect(attempts.find([option('one')])).toBe(attempt); + expect(attempts.find([option('two')])).toBe(attempt); + }); + + test('reclaims enough terminal identities for a multi-option reservation', () => { + const attempts = new ClientPaymentAttempts(2); + const signal = new AbortController().signal; + const prior = attempts.reserve([option('one'), option('two')], signal); + prior.complete('declined'); + const next = attempts.reserve([option('three'), option('four')], signal); + expect(attempts.find([option('one'), option('two')])).toBeUndefined(); + expect(attempts.find([option('three')])).toBe(next); + expect(attempts.find([option('four')])).toBe(next); + next.complete('handled'); + }); +}); diff --git a/src/payments/client-payment-attempts.ts b/src/payments/client-payment-attempts.ts index a8ac981..7bef17b 100644 --- a/src/payments/client-payment-attempts.ts +++ b/src/payments/client-payment-attempts.ts @@ -1,4 +1,5 @@ import type { PaymentOption } from './types.js'; +import { LruCache } from '../core/utils/lru-cache.js'; import { ClientPaymentError } from './client-payment-validation.js'; type AttemptOutcome = 'in_flight' | 'handled' | 'declined' | 'unknown'; @@ -13,15 +14,18 @@ export interface ClientPaymentAttempt { /** * Wrapper-scoped attempt history. A Nostr wrapper is bound to one authenticated * server; generic wrappers use their transport instance as the peer boundary. - * No live/completed/unknown entry is evicted and silently made payable again. + * In-flight and unknown attempts stay pinned; handled and declined identities + * may be evicted least recently used first under capacity pressure. */ export class ClientPaymentAttempts { - private readonly entries = new Map< - string, - { amount: number; attempt: ClientPaymentAttempt } - >(); + private readonly entries: LruCache<{ + amount: number; + attempt: ClientPaymentAttempt; + }>; - public constructor(private readonly maxEntries: number) {} + public constructor(private readonly maxEntries: number) { + this.entries = new LruCache(maxEntries); + } private key(option: PaymentOption): string { return JSON.stringify([option.pmi, option.pay_req]); @@ -59,12 +63,20 @@ export class ClientPaymentAttempts { signal: AbortSignal, ): ClientPaymentAttempt { const unique = new Map(options.map((option) => [this.key(option), option])); - if (this.entries.size + unique.size > this.maxEntries) { + const needed = this.entries.size + unique.size - this.maxEntries; + const evictable: string[] = []; + for (const [key, { attempt }] of this.entries.entries()) { + if (evictable.length >= needed) break; + if (attempt.outcome === 'handled' || attempt.outcome === 'declined') + evictable.push(key); + } + if (evictable.length < needed) { throw new ClientPaymentError( 'payment_attempt_capacity', - 'Client payment attempt capacity reached', + 'Client payment attempt capacity exceeded by unresolved payments or offered options', ); } + for (const key of evictable) this.entries.delete(key); let resolve!: () => void; const attempt: ClientPaymentAttempt = { outcome: 'in_flight', diff --git a/src/payments/client-payment-options.ts b/src/payments/client-payment-options.ts index 95f33c9..f16d6b0 100644 --- a/src/payments/client-payment-options.ts +++ b/src/payments/client-payment-options.ts @@ -40,9 +40,10 @@ export interface ClientPaymentsOptions { /** Optional limits across all automatic payment offers for one invocation. */ limits?: ClientPaymentLimits; /** - * Maximum distinct payment identities retained by this wrapper. Completed and - * uncertain attempts are never evicted to make room for a new payment. - * At capacity, new attempts fail without invoking a wallet. @default 1000 + * Maximum distinct payment identities retained by this wrapper. In-flight and + * unknown attempts stay pinned; handled and declined identities are evicted + * least recently used first. New attempts fail before wallet work if pinned + * identities and offered options cannot fit. Zero refuses payment. @default 1000 * Explicit callbacks reserve every offered identity because option selection * is opaque; they must pay only one of the supplied options. */ diff --git a/src/payments/client-payment-validation.ts b/src/payments/client-payment-validation.ts index e86c555..ea23e03 100644 --- a/src/payments/client-payment-validation.ts +++ b/src/payments/client-payment-validation.ts @@ -12,7 +12,7 @@ const amountSchema = z .finite() .nonnegative() .max(Number.MAX_SAFE_INTEGER); -const pmiSchema = z.string().regex(/^[a-z0-9-]+$/); +const pmiSchema = z.string().min(1); const timerSchema = z.number().finite().positive().max(MAX_TIMER_MS); /** Internal payment failure with a stable application-visible reason. */ @@ -125,6 +125,14 @@ export function validateClientPaymentOptions( .optional(), paymentInteraction: z.enum(['transparent', 'explicit_gating']).optional(), handlers: z.array(z.object({ pmi: pmiSchema })).optional(), + onPaymentRequired: z + .unknown() + .refine((value) => typeof value === 'function') + .optional(), + paymentPolicy: z + .unknown() + .refine((value) => typeof value === 'function') + .optional(), limits: z .object({ maxAmountByPmi: z.record(pmiSchema, amountSchema).optional(), diff --git a/src/payments/client-payments-limits.test.ts b/src/payments/client-payments-limits.test.ts index a53c91b..62488a7 100644 --- a/src/payments/client-payments-limits.test.ts +++ b/src/payments/client-payments-limits.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from 'bun:test'; import type { Transport } from '@contextvm/mcp-sdk/shared/transport'; import { withClientPayments } from './client-payments.js'; +import type { ClientPaymentsOptions } from './client-payment-options.js'; import type { PaymentOption } from './types.js'; import { run, @@ -175,7 +176,8 @@ describe('client payment limits and validation', () => { { ...option(), amount: Infinity }, { ...option(), amount: Number.MAX_SAFE_INTEGER + 1 }, { ...option(), amount: '5' }, - { ...option(), pmi: 'INVALID:PMI' }, + { ...option(), pmi: '' }, + { ...option(), pmi: 1 }, { ...option(), pay_req: '' }, { ...option(), pay_req: ' ' }, { ...option(), ttl: 0 }, @@ -210,9 +212,55 @@ describe('client payment limits and validation', () => { }); } + for (const mode of ['transparent', 'explicit'] as const) { + test(`accepts a non-kebab PMI in ${mode} offers, handlers, and amount limits`, async () => { + const pmi = 'pmi:A'; + const paid: PaymentOption[] = []; + await run( + { + limits: { maxAmountByPmi: { [pmi]: 5 } }, + handlers: [ + { + pmi, + async handle(req): Promise { + paid.push(req); + }, + }, + ], + onPaymentRequired: async ({ options }) => { + paid.push(...options); + return { paid: true }; + }, + }, + async (h) => { + await h.paid.send(request()); + h.deliver( + mode === 'transparent' + ? offer(option('invoice', 5, pmi)) + : required(1, [option('invoice', 5, pmi)]), + ); + await waitFor(() => paid.length === 1); + expect(paid[0]).toMatchObject(option('invoice', 5, pmi)); + expect(failures(h)).toHaveLength(0); + h.deliver( + mode === 'transparent' + ? offer(option('replacement', 1, pmi)) + : required(1, [option('replacement', 1, pmi)]), + ); + await waitFor(() => failures(h).length === 1); + expect(paid).toHaveLength(1); + expect(failures(h)[0]?.error.data).toMatchObject({ + reason: 'payment_limit_exceeded', + }); + }, + ); + }); + } + for (const invalid of [ [], [option(), { ...option('invalid'), amount: -1 }], + [{ ...option(), pmi: '' }], [option(), option('invoice', 6)], ]) { test(`validates every explicit option ${JSON.stringify(invalid)}`, async () => { @@ -318,6 +366,11 @@ describe('client payment limits and validation', () => { { limits: { timeoutMs: 2_147_483_648 } }, { limits: { maxAttempts: -1 } }, { limits: { maxAmountByPmi: { fake: Infinity } } }, + { handlers: [{ pmi: '', handle: async () => {} }] }, + { limits: { maxAmountByPmi: { '': 1 } } }, + ...['onPaymentRequired', 'paymentPolicy'].flatMap((key) => + [true, false, null, 1, 'typo', {}].map((value) => ({ [key]: value })), + ), ]) { test(`configuration validation rejects ${JSON.stringify(options)} before wrapping`, () => { const base: Transport = { @@ -325,9 +378,9 @@ describe('client payment limits and validation', () => { async send(): Promise {}, async close(): Promise {}, }; - expect(() => withClientPayments(base, options)).toThrow( - 'Invalid client payment options', - ); + expect(() => + withClientPayments(base, options as ClientPaymentsOptions), + ).toThrow('Invalid client payment options'); expect(() => withClientPayments(base, {})).not.toThrow(); }); } diff --git a/src/payments/client-payments-reliability.test.ts b/src/payments/client-payments-reliability.test.ts index 27aa664..84b0d2b 100644 --- a/src/payments/client-payments-reliability.test.ts +++ b/src/payments/client-payments-reliability.test.ts @@ -69,6 +69,7 @@ describe('client payment attempt reliability', () => { let calls = 0; await run( { + maxTrackedPaymentAttempts: 1, onPaymentRequired: async () => { calls++; throw new Error('wallet response lost'); @@ -85,6 +86,13 @@ describe('client payment attempt reliability', () => { expect(failures(h)[1]?.error.data).toMatchObject({ reason: 'payment_outcome_unknown', }); + await h.paid.send(request(3)); + h.deliver(required(3, [option('replacement')])); + await waitFor(() => failures(h).length === 3); + expect(calls).toBe(1); + expect(failures(h)[2]?.error.data).toMatchObject({ + reason: 'payment_attempt_capacity', + }); }, ); }); @@ -138,7 +146,7 @@ describe('client payment attempt reliability', () => { }); }); - test('attempt capacity refuses a new invoice without forgetting a completed payment', async () => { + test('completed history does not exhaust a long-lived client payment capacity', async () => { let calls = 0; await run( { @@ -149,21 +157,14 @@ describe('client payment attempt reliability', () => { }, }, async (h) => { - await h.paid.send(request()); - h.deliver(required()); - await waitFor(() => h.sent.length === 2); - h.deliver({ jsonrpc: '2.0', id: 1, result: {} }); - await h.paid.send(request(2)); - h.deliver(required(2, [option('new-invoice')])); - await waitFor(() => failures(h).length === 1); - expect(calls).toBe(1); - expect(failures(h)[0]?.error.data).toMatchObject({ - reason: 'payment_attempt_capacity', - }); - await h.paid.send(request(3)); - h.deliver(required(3)); - await waitFor(() => h.sent.length === 5); - expect(calls).toBe(1); + for (let id = 1; id <= 3; id++) { + await h.paid.send(request(id)); + h.deliver(required(id, [option(`invoice-${id}`)])); + await waitFor(() => h.sent.length === id * 2); + h.deliver({ jsonrpc: '2.0', id, result: {} }); + } + expect(calls).toBe(3); + expect(failures(h)).toHaveLength(0); }, ); }); diff --git a/src/payments/client-payments.ts b/src/payments/client-payments.ts index 058b7d0..a13ca64 100644 --- a/src/payments/client-payments.ts +++ b/src/payments/client-payments.ts @@ -90,7 +90,8 @@ function cancelledRequestId( * Wraps a transport with validated CEP-8 offers, bounded payment attempts, and * request-owned retries. Nostr peers are authenticated by their transport; * generic transports isolate attempt history per wrapper and must keep their - * peer identity stable. History is retained for the lifetime of the wrapper. + * peer identity stable. In-flight and unknown attempts stay pinned; terminal + * history is bounded and may be evicted under capacity pressure. * * Unmatched handlers leave transparent payment to the application. Generic * notifications without request correlation support legacy handler execution, @@ -133,6 +134,11 @@ export function withClientPayments( code = -32000, data: Record = {}, ): void => { + logger.info('Client payment failed', { + requestId: state.id, + reason: data.type === 'payment_handler_error' ? data.type : error.reason, + outcome: data.outcome, + }); if (state.id === undefined) { onerror?.(error); return; @@ -207,6 +213,11 @@ export function withClientPayments( }, Math.max(delayMs, minRetryDelayMs), ); + logger.debug('Payment retry scheduled', { + requestId: state.id, + delayMs: Math.max(delayMs, minRetryDelayMs), + code, + }); }; const priorOutcome = async ( @@ -214,6 +225,10 @@ export function withClientPayments( attempt: ClientPaymentAttempt, explicit: boolean, ): Promise => { + logger.debug('Payment offer deduplicated', { + requestId: state.id, + outcome: attempt.outcome, + }); await attempt.completion; if (!lifecycle.isActive(state)) return; if (attempt.outcome === 'handled') { @@ -284,6 +299,10 @@ export function withClientPayments( const allowed = lifecycle.approve(state, data.payment_options); const attempt = attempts.reserve(allowed, state.controller.signal); state.paymentInFlight = true; + logger.debug('Explicit payment offer accepted; invoking approval handler', { + requestId: state.id, + optionCount: allowed.length, + }); try { const result = await options.onPaymentRequired({ options: allowed, @@ -401,6 +420,11 @@ export function withClientPayments( lifecycle.approve(state, [option]); const attempt = attempts.reserve([option], state.controller.signal); state.paymentInFlight = true; + logger.debug('Transparent payment offer accepted', { + requestId: state.id, + pmi: option.pmi, + amount: option.amount, + }); const req: PaymentHandlerRequest = { amount: option.amount, pmi: option.pmi, @@ -451,6 +475,11 @@ export function withClientPayments( ); } walletStarted = true; + logger.debug('Invoking payment handler', { + requestId: state.id, + pmi: option.pmi, + amount: option.amount, + }); await handler.handle(req); if (lifecycle.isActive(state)) attempt.complete('handled'); } catch (error: unknown) { diff --git a/src/transport/payment-client-reliability.e2e.test.ts b/src/transport/payment-client-reliability.e2e.test.ts index 169980e..fd253be 100644 --- a/src/transport/payment-client-reliability.e2e.test.ts +++ b/src/transport/payment-client-reliability.e2e.test.ts @@ -57,12 +57,12 @@ describe('instant explicit payment retry delivery', () => { let invoices = 0; let approvals = 0; const processor: PaymentProcessor = { - pmi: 'fake', + pmi: 'pmi:A', async createPaymentRequired(params) { invoices++; return { amount: params.amount, - pmi: 'fake', + pmi: 'pmi:A', pay_req: `invoice-${invoices}`, ttl: 60, }; @@ -112,7 +112,9 @@ describe('instant explicit payment retry delivery', () => { const paid = withClientPayments(base, { paymentInteraction: 'explicit_gating', minRetryDelayMs: 0, - async onPaymentRequired() { + limits: { maxAmountByPmi: { 'pmi:A': 5 } }, + async onPaymentRequired({ options }) { + expect(options[0]?.pmi).toBe('pmi:A'); approvals++; return { paid: true }; },