Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | 3x 15x 3x 48x 8x 42x 42x 17x 1x 1x 16x 7x 7x 9x 3x 22x | /**
* Forward policy for the channel-gateway downstream link (issue #483, Phase 2).
*
* Pure control flow, kept out of the MCP adapter so it is unit-testable:
*
* - A call that arrives while the link is down is BUFFERED (not failed): it was
* never sent, so replaying it after reconnect cannot double-execute. The
* buffer is bounded — once `maxQueued` calls are already waiting, further
* calls are rejected (backpressure) instead of growing without limit.
* - A call that was SENT and then failed is retried once after reconnect only
* if it is a read (idempotent). Writes are never blind-retried: the request
* may have executed before the link dropped, and a retry would double it.
*/
/** The captured downstream client lost its transport while this request was in flight. */
export class DownstreamDisconnectedError extends Error {
constructor(cause: unknown) {
super('Downstream connection closed', { cause });
}
}
/** Injected hooks; the gateway supplies real transport, tests supply fakes. */
export interface ForwardPolicy {
/** True if `name` is an idempotent read (safe to retry). */
isRead: (name: string) => boolean;
/** Current downstream link health. */
isConnected: () => boolean;
/**
* Wait for the link to come back, bounded. Resolves once connected; rejects on
* timeout or when the bounded buffer is full. The gateway counts in-flight
* waiters here to enforce `maxQueued`.
*/
waitForConnection: () => Promise<void>;
/** Send one call over the (assumed live) downstream link. */
call: (name: string, args: unknown) => Promise<unknown>;
}
/**
* Forward one call under the read-safe / write-no-retry policy with bounded
* buffering across a reconnect. Throws if the buffer is full (via
* `waitForConnection`) or a non-retryable call fails.
*/
export async function forwardWithPolicy(
policy: ForwardPolicy,
name: string,
args: unknown,
): Promise<unknown> {
// Not yet sent: safe to buffer until the link returns, for reads and writes.
if (!policy.isConnected()) {
await policy.waitForConnection();
}
try {
return await policy.call(name, args);
} catch (err) {
if (name === 'manage_duo_session' && err instanceof DownstreamDisconnectedError) {
err.message =
'Duo mutation outcome is unknown. Inspect browse_duo_sessions list or get before another command. Do not automatically replay the write.';
throw err;
}
// Replay only after a lost link, never a valid protocol/schema rejection.
if (policy.isRead(name) && err instanceof DownstreamDisconnectedError) {
await policy.waitForConnection();
return await policy.call(name, args);
}
throw err;
}
}
/** A tool call is an idempotent read by name prefix. */
export function isReadCall(name: string): boolean {
return name.startsWith('browse_') || name.startsWith('get_') || name.startsWith('list_');
}
|