tech-log-frontend/src/adapters/browser-rpc/browser-rpc-runtime.ts

import {
  createReadOnlyRegistry,
  type ReadOnlyRegistry,
} from "../../contracts/read-only-registry.ts";
import type {
  BrowserRpcGenerationFence,
  BrowserRpcServerStreamPort,
  BrowserRpcUnaryPort,
} from "../../application/ports/browser-rpc/index.ts";
import type { ClockPort } from "../../application/ports/clock-port.ts";
import type { Result } from "../../application/result.ts";
import {
  installBrowserRpcContractBindings,
  type InstalledBrowserRpcContractBindings,
  BROWSER_RPC_HARD_LIMITS,
  validateBrowserRpcContractBindings,
  type BrowserRpcOperationV3,
  type BrowserRpcProviderProfile,
  type BrowserRpcRequestEncoder,
  type BrowserRpcTransportFailureCode,
} from "../../contracts/browser-rpc.ts";
import type { InstalledBoundaryMapper } from "../../contracts/boundary-mapper.ts";
import {
  createFailure,
  type AppFailure,
  type FailureKind,
} from "../../contracts/errors.ts";
import type { RuntimeSchemaCodec } from "../../contracts/schema-registry.ts";
import { systemClock } from "../platform/system-clock.ts";
import { decodeServerStreamLease } from "./transport.ts";
import type {
  BrowserRpcServerStreamLease,
  BrowserRpcStreamFrame,
  BrowserRpcTransport,
  BrowserRpcTransportFailure,
  BrowserRpcUnaryTransportResult,
} from "./transport.ts";

export type BrowserRpcObservationOutcome =
  | "SUCCESS"
  | "FAILED"
  | "ABORTED"
  | "TIMEOUT"
  | "CONTRACT_REJECTED";

export type BrowserRpcObservation = Readonly<{
  operationId: string;
  protocol: "CONNECT_HTTP" | "GRPC_WEB";
  runtimeProfileId: string;
  rpcKind: "UNARY" | "SERVER_STREAM";
  outcome: BrowserRpcObservationOutcome;
  attemptCount: number;
  messageCount: number;
}>;

export type BrowserRpcObservationSink = Readonly<{
  observe(observation: BrowserRpcObservation): void;
}>;

export type BrowserRpcRuntimeDependencies = Readonly<{
  operations: Readonly<Record<string, BrowserRpcOperationV3>>;
  profiles: Readonly<Record<string, BrowserRpcProviderProfile>>;
  schemaCodecs: Readonly<Record<string, RuntimeSchemaCodec>>;
  mappers: Readonly<Record<string, InstalledBoundaryMapper>>;
  requestEncoders: Readonly<Record<string, BrowserRpcRequestEncoder>>;
  transports: Readonly<Record<string, BrowserRpcTransport>>;
  generationFence?: BrowserRpcGenerationFence;
  clock?: ClockPort;
  observations?: BrowserRpcObservationSink;
}>;

export type BrowserRpcRuntime = Readonly<{
  bindUnary<Input, Output>(
    operationId: string,
    isOutput: (value: unknown) => value is Output,
  ): BrowserRpcUnaryPort<Input, Output>;
  bindServerStream<Input, Event>(
    operationId: string,
    isEvent: (value: unknown) => value is Event,
  ): BrowserRpcServerStreamPort<Input, Event>;
}>;

type BoundOperation = Readonly<{
  operation: BrowserRpcOperationV3;
  profile: BrowserRpcProviderProfile;
  encoder: BrowserRpcRequestEncoder;
  transport: BrowserRpcTransport;
}>;

type PreparedRequest =
  | Readonly<{
      ok: true;
      value: unknown;
      encodedBytes: number;
    }>
  | Readonly<{ ok: false; error: AppFailure }>;

type TimedResult<Value> =
  | Readonly<{ kind: "VALUE"; value: Value }>
  | Readonly<{ kind: "TIMEOUT" }>
  | Readonly<{ kind: "ABORTED" }>
  | Readonly<{ kind: "THREW" }>;

const stableGenerationFence: BrowserRpcGenerationFence<string> =
  Object.freeze({
    capture: () => "stable",
    isCurrent: (token) => token === "stable",
  });

export function createBrowserRpcRuntime(
  dependencies: BrowserRpcRuntimeDependencies,
): BrowserRpcRuntime {
  const clock = dependencies.clock ?? systemClock;
  const generationFence =
    dependencies.generationFence ?? stableGenerationFence;
  // RPC-RR-03 / RPC-03. Snapshot before validating — every registry, not just
  // the transports. Handing the caller's own objects to the join validation
  // first ran their accessors, so a hostile getter could observe validation and
  // then answer the runtime differently.
  const installedTransports = snapshotTransports(dependencies.transports);
  // R-04. Install exact immutable snapshots once. Every later `bind()` reads
  // the snapshot, never the caller's registry objects, so a post-composition
  // mutation cannot change replay policy, deadlines, byte ceilings or
  // transport selection away from what was validated.
  const installed = installBrowserRpcContractBindings({
    operations: dependencies.operations,
    profiles: dependencies.profiles,
    schemaCodecs: dependencies.schemaCodecs,
    mappers: dependencies.mappers,
    requestEncoders: dependencies.requestEncoders,
  });
  validateRuntimeDependencies(installed, installedTransports);

  // RPC-RR-01. One entry per physical stream that has been opened and not yet
  // confirmed closed. A stream stays here through DRAINING — after cancel,
  // before `waitClosed()` settles — so a second stream for the same operation
  // cannot be admitted while the first is still running against the server.
  const activeStreams = new Map<string, ActiveStreamLease>();

  function bind(
    operationId: string,
    expectedKind: "UNARY" | "SERVER_STREAM",
  ): BoundOperation {
    const operation = installed.operations.get(operationId);
    if (!operation || operation.rpcKind !== expectedKind) {
      throw new TypeError(
        `Browser RPC operation cannot be bound as ${expectedKind}: ${operationId}`,
      );
    }
    const profile = installed.profiles.get(operation.runtimeProfileId);
    const encoder = installed.requestEncoders.get(operation.requestEncoderId);
    const transport = installedTransports.get(operation.runtimeProfileId);
    if (!profile || !encoder || !transport) {
      throw new TypeError(
        `Browser RPC runtime binding is incomplete: ${operationId}`,
      );
    }
    return Object.freeze({ operation, profile, encoder, transport });
  }

  return Object.freeze({
    bindUnary<Input, Output>(
      operationId: string,
      isOutput: (value: unknown) => value is Output,
    ): BrowserRpcUnaryPort<Input, Output> {
      if (typeof isOutput !== "function") {
        throw new TypeError("Browser RPC unary result guard is required.");
      }
      const bound = bind(operationId, "UNARY");
      return Object.freeze({
        execute: (input, context = {}) =>
          executeUnary(
            installed,
            dependencies,
            bound,
            input,
            context,
            isOutput,
            generationFence,
            clock,
          ),
      });
    },

    bindServerStream<Input, Event>(
      operationId: string,
      isEvent: (value: unknown) => value is Event,
    ): BrowserRpcServerStreamPort<Input, Event> {
      if (typeof isEvent !== "function") {
        throw new TypeError("Browser RPC stream result guard is required.");
      }
      const bound = bind(operationId, "SERVER_STREAM");
      return Object.freeze({
        open: (input, context = {}) =>
          executeServerStream(
            installed,
            dependencies,
            bound,
            input,
            context,
            isEvent,
            generationFence,
            clock,
            activeStreams,
          ),
      });
    },
  });
}

/**
 * RPC-RR-01. A physical stream the runtime has opened and not yet confirmed
 * closed. `draining` is true from the moment cancellation is requested until
 * `waitClosed()` settles; while it is true the operation admits no new stream.
 */
type ActiveStreamLease = Readonly<{
  streamId: string;
  cancel(reason: string): void;
  /**
   * RPC-01. Fulfils with `true` only when the transport positively confirmed
   * the physical stream closed, and with `false` for every negative receipt —
   * a rejection, a synchronous throw, or a `waitClosed()` that did not even
   * return a promise. A negative receipt keeps the operation DRAINING.
   */
  closed: Promise<boolean>;
}>;

/**
 * RPC-01. Turns a lease's close receipt into evidence the registry can trust.
 * Only a fulfilled promise from a contract-shaped `waitClosed()` counts as
 * confirmation that the physical stream ended.
 */
function confirmPhysicalClosure(
  lease: Readonly<{ waitClosed(): Promise<void> }>,
): Promise<boolean> {
  return Promise.resolve()
    .then(() => {
      const receipt: unknown = lease.waitClosed();
      if (
        receipt === null ||
        (typeof receipt !== "object" && typeof receipt !== "function") ||
        typeof (receipt as PromiseLike<void>).then !== "function"
      ) {
        return false;
      }
      return Promise.resolve(receipt as PromiseLike<void>).then(
        () => true,
        () => false,
      );
    })
    .catch(() => false);
}

/**
 * RPC-02. Reads an iterator's optional `return` once, through its own data
 * descriptor. Testing `iterator.return` directly ran a foreign accessor outside
 * any boundary, so a throwing getter replaced the already selected stream
 * outcome with a native rejection and skipped the rest of the cleanup.
 */
function safeIteratorReturn(
  iterator: AsyncIterator<unknown> | null | undefined,
): (() => unknown) | null {
  if (!iterator) return null;
  try {
    let current: object | null = iterator;
    while (current !== null) {
      const descriptor = Object.getOwnPropertyDescriptor(current, "return");
      if (descriptor) {
        if (!("value" in descriptor)) return null;
        return typeof descriptor.value === "function"
          ? (descriptor.value as () => unknown).bind(iterator)
          : null;
      }
      current = Reflect.getPrototypeOf(current);
    }
    return null;
  } catch {
    return null;
  }
}

async function executeUnary<Output>(
  installed: InstalledBrowserRpcContractBindings,
  dependencies: BrowserRpcRuntimeDependencies,
  bound: BoundOperation,
  input: unknown,
  context: Readonly<{ signal?: AbortSignal; idempotencyKey?: string }>,
  isOutput: (value: unknown) => value is Output,
  generationFence: BrowserRpcGenerationFence,
  clock: ClockPort,
): Promise<Result<Output, AppFailure>> {
  const { operation, profile, encoder, transport } = bound;
  const linked = linkedAbortController(context.signal);
  let attemptCount = 0;

  const finish = (
    result: Result<Output, AppFailure>,
    outcome: BrowserRpcObservationOutcome,
  ): Result<Output, AppFailure> => {
    // R-06. Listener and timer cleanup happens exactly once, on every exit.
    linked.cleanup();
    observe(dependencies, operation, outcome, attemptCount, result.ok ? 1 : 0);
    return result;
  };

  const generation = safeCapture(generationFence);
  if (generation === FENCE_FAILURE) {
    return finish(
      failureResult(
        callFailure(
          operation,
          0,
          "SCOPE_GENERATION_CHANGED",
          "RPC_SCOPE_GENERATION_UNAVAILABLE",
        ),
      ),
      "FAILED",
    );
  }
  const startedAt = safeNow(clock);
  if (startedAt === CLOCK_FAILURE) {
    return finish(
      failureResult(
        callFailure(
          operation,
          0,
          "SERVER_FAILURE",
          "RPC_RUNTIME_DEPENDENCY_FAILED",
        ),
      ),
      "FAILED",
    );
  }
  const deadlineAt = startedAt + operation.totalDeadlineMs;

  const contextFailure = validateCallContext(operation, context, 0);
  if (contextFailure) {
    return finish(failureResult(contextFailure), "CONTRACT_REJECTED");
  }
  if (linked.controller.signal.aborted) {
    return finish(
      failureResult(callFailure(operation, 0, "REQUEST_ABORTED", "RPC_ABORTED")),
      "ABORTED",
    );
  }

  const prepared = prepareRequest(installed, operation, encoder, input, 0);
  if (!prepared.ok) {
    return finish(failureResult(prepared.error), "CONTRACT_REJECTED");
  }

  for (let attempt = 0; attempt < profile.maxAttempts; attempt += 1) {
    attemptCount = attempt + 1;
    const attemptNow = safeNow(clock);
    if (attemptNow === CLOCK_FAILURE) {
      linked.controller.abort("clock");
      return finish(
        failureResult(
          callFailure(
            operation,
            attemptCount,
            "SERVER_FAILURE",
            "RPC_RUNTIME_DEPENDENCY_FAILED",
          ),
        ),
        "FAILED",
      );
    }
    const remainingMs = deadlineAt - attemptNow;
    if (remainingMs <= 0) {
      linked.controller.abort("deadline");
      return finish(
        failureResult(
          callFailure(
            operation,
            attempt,
            "REQUEST_TIMEOUT",
            "RPC_TOTAL_DEADLINE_EXCEEDED",
          ),
        ),
        "TIMEOUT",
      );
    }
    const call = Object.freeze({
      operation,
      profile,
      request: prepared.value,
      encodedRequestBytes: prepared.encodedBytes,
      attempt: attempt + 1,
      timeoutMs: Math.max(1, Math.min(remainingMs, operation.totalDeadlineMs)),
      signal: linked.controller.signal,
      ...(context.idempotencyKey
        ? { idempotencyKey: context.idempotencyKey }
        : {}),
    });
    const timed = await raceWithin(
      Promise.resolve().then(() => transport.invokeUnary!(call)),
      remainingMs,
      linked.controller.signal,
      clock,
    );
    if (timed.kind === "TIMEOUT") {
      linked.controller.abort("deadline");
      return finish(
        failureResult(
          callFailure(
            operation,
            attempt,
            "REQUEST_TIMEOUT",
            "RPC_TOTAL_DEADLINE_EXCEEDED",
          ),
        ),
        "TIMEOUT",
      );
    }
    if (timed.kind === "ABORTED" || linked.controller.signal.aborted) {
      return finish(
        failureResult(
          callFailure(operation, attempt, "REQUEST_ABORTED", "RPC_ABORTED"),
        ),
        "ABORTED",
      );
    }
    if (timed.kind === "THREW") {
      return finish(
        failureResult(
          callFailure(
            operation,
            attempt,
            "SERVER_FAILURE",
            "RPC_TRANSPORT_EXECUTION_FAILED",
          ),
        ),
        "FAILED",
      );
    }

    const transportResult = validateUnaryTransportResult(timed.value);
    if (!transportResult) {
      return finish(
        failureResult(protocolFailure(operation, attempt)),
        "CONTRACT_REJECTED",
      );
    }
    if (!transportResult.ok) {
      if (
        shouldRetry(
          operation,
          profile,
          transportResult.failure,
          attempt,
          context.idempotencyKey,
        )
      ) {
        const delay = retryDelay(profile, transportResult.failure, attempt);
        if (delay >= deadlineAt - safeNowOrDeadline(clock, deadlineAt)) {
          linked.controller.abort("deadline");
          return finish(
            failureResult(
              callFailure(
                operation,
                attempt,
                "REQUEST_TIMEOUT",
                "RPC_RETRY_BUDGET_EXHAUSTED",
              ),
            ),
            "TIMEOUT",
          );
        }
        try {
          await clock.sleep(delay, linked.controller.signal);
        } catch {
          return finish(
            failureResult(
              callFailure(
                operation,
                attempt,
                "REQUEST_ABORTED",
                "RPC_ABORTED",
              ),
            ),
            "ABORTED",
          );
        }
        continue;
      }
      const mapped = mapTransportFailure(
        operation,
        attempt,
        transportResult.failure,
      );
      return finish(
        failureResult(mapped),
        mapped.kind === "REQUEST_ABORTED"
          ? "ABORTED"
          : mapped.kind === "REQUEST_TIMEOUT"
            ? "TIMEOUT"
            : "FAILED",
      );
    }

    const mapped = mapResponse(
      installed,
      operation,
      attempt,
      transportResult.message,
      transportResult.encodedBytes,
      generation,
      generationFence,
      isOutput,
      deadlineAt,
      clock,
    );
    return finish(
      mapped,
      mapped.ok
        ? "SUCCESS"
        : mapped.error.kind === "REQUEST_TIMEOUT"
          ? "TIMEOUT"
          : "CONTRACT_REJECTED",
    );
  }

  return finish(
    failureResult(
      callFailure(
        operation,
        Math.max(0, attemptCount - 1),
        "SERVER_FAILURE",
        "RPC_RETRY_EXHAUSTED",
      ),
    ),
    "FAILED",
  );
}

async function* executeServerStream<Event>(
  installed: InstalledBrowserRpcContractBindings,
  dependencies: BrowserRpcRuntimeDependencies,
  bound: BoundOperation,
  input: unknown,
  context: Readonly<{ signal?: AbortSignal; idempotencyKey?: string }>,
  isEvent: (value: unknown) => value is Event,
  generationFence: BrowserRpcGenerationFence,
  clock: ClockPort,
  activeStreams: Map<string, ActiveStreamLease>,
): AsyncIterable<Result<Event, AppFailure>> {
  const { operation, profile, encoder, transport } = bound;
  const deadlineAt = safeNowOrDeadline(clock, 0) + operation.totalDeadlineMs;
  const linked = linkedAbortController(context.signal);
  let iterator: AsyncIterator<BrowserRpcStreamFrame> | null = null;
  let lease: BrowserRpcServerStreamLease | null = null;
  let observed = false;
  let messageCount = 0;

  const finish = (outcome: BrowserRpcObservationOutcome) => {
    if (observed) return;
    observed = true;
    observe(dependencies, operation, outcome, 1, messageCount);
  };

  // RPC-RR-02. The fence is a caller collaborator, so a synchronous throw from
  // it is an ordinary failure of this call. Capturing outside the protected
  // boundary let that throw escape the stream's Result contract entirely.
  const generation = safeCapture(generationFence);
  if (generation === FENCE_FAILURE) {
    finish("FAILED");
    yield failureResult(
      callFailure(
        operation,
        0,
        "SCOPE_GENERATION_CHANGED",
        "RPC_SCOPE_GENERATION_UNAVAILABLE",
      ),
    );
    return;
  }

  try {
    const contextFailure = validateCallContext(operation, context, 0);
    if (contextFailure) {
      finish("CONTRACT_REJECTED");
      yield failureResult(contextFailure);
      return;
    }
    if (linked.controller.signal.aborted) {
      finish("ABORTED");
      yield failureResult(
        callFailure(operation, 0, "REQUEST_ABORTED", "RPC_ABORTED"),
      );
      return;
    }
    const prepared = prepareRequest(
      installed,
      operation,
      encoder,
      input,
      0,
    );
    if (!prepared.ok) {
      finish("CONTRACT_REJECTED");
      yield failureResult(prepared.error);
      return;
    }

    const remainingMs = deadlineAt - safeNowOrDeadline(clock, deadlineAt);
    if (remainingMs <= 0) {
      finish("TIMEOUT");
      yield failureResult(
        callFailure(
          operation,
          0,
          "REQUEST_TIMEOUT",
          "RPC_TOTAL_DEADLINE_EXCEEDED",
        ),
      );
      return;
    }
    // RPC-RR-01. Admission fence. A previous physical stream for this operation
    // that has not confirmed closure still owns the server-side resource, so a
    // second one is refused rather than opened alongside it.
    if (activeStreams.has(operation.operationId)) {
      finish("CONTRACT_REJECTED");
      yield failureResult(
        callFailure(operation, 0, "CONFLICT", "RPC_STREAM_DRAINING"),
      );
      return;
    }

    let decodedLease: BrowserRpcServerStreamLease | null;
    try {
      decodedLease = decodeServerStreamLease(
        transport.openServerStream!(
          Object.freeze({
            operation,
            profile,
            request: prepared.value,
            encodedRequestBytes: prepared.encodedBytes,
            attempt: 1,
            timeoutMs: Math.max(1, remainingMs),
            signal: linked.controller.signal,
            ...(context.idempotencyKey
              ? { idempotencyKey: context.idempotencyKey }
              : {}),
          }),
        ),
      );
    } catch {
      finish("FAILED");
      yield failureResult(
        callFailure(
          operation,
          0,
          "SERVER_FAILURE",
          "RPC_TRANSPORT_EXECUTION_FAILED",
        ),
      );
      return;
    }
    if (!decodedLease) {
      finish("CONTRACT_REJECTED");
      yield failureResult(protocolFailure(operation, 0));
      return;
    }
    lease = decodedLease;
    // Registered at the moment the physical stream exists, not after a timeout.
    activeStreams.set(
      operation.operationId,
      Object.freeze({
        streamId: lease.streamId,
        cancel: lease.cancel,
        // RPC-01. Only a fulfilled, contract-shaped receipt is evidence that the
        // physical stream closed. Absorbing a rejection, a synchronous throw or
        // a non-promise into `undefined` forged that evidence, and the registry
        // then admitted a second stream for the same operation while the first
        // was still running against the server.
        closed: confirmPhysicalClosure(lease),
      }),
    );
    try {
      iterator = lease.frames[Symbol.asyncIterator]();
    } catch {
      finish("FAILED");
      yield failureResult(
        callFailure(
          operation,
          0,
          "SERVER_FAILURE",
          "RPC_TRANSPORT_EXECUTION_FAILED",
        ),
      );
      return;
    }

    let totalBytes = 0;
    let terminal:
      | Readonly<{ ok: true }>
      | Readonly<{ ok: false; failure: BrowserRpcTransportFailure }>
      | null = null;

    while (true) {
      const totalRemaining = deadlineAt - safeNowOrDeadline(clock, deadlineAt);
      if (totalRemaining <= 0) {
        linked.controller.abort("deadline");
        finish("TIMEOUT");
        yield failureResult(
          callFailure(
            operation,
            0,
            "REQUEST_TIMEOUT",
            "RPC_TOTAL_DEADLINE_EXCEEDED",
          ),
        );
        return;
      }
      const waitMs = Math.min(
        totalRemaining,
        operation.idleDeadlineMs ?? totalRemaining,
      );
      const next = await raceWithin(
        Promise.resolve().then(() => iterator!.next()),
        waitMs,
        linked.controller.signal,
        clock,
      );
      if (next.kind === "TIMEOUT") {
        linked.controller.abort("idle-or-deadline");
        const totalExpired = safeNowOrDeadline(clock, deadlineAt) >= deadlineAt;
        finish("TIMEOUT");
        yield failureResult(
          callFailure(
            operation,
            0,
            "REQUEST_TIMEOUT",
            totalExpired
              ? "RPC_TOTAL_DEADLINE_EXCEEDED"
              : "RPC_STREAM_IDLE_TIMEOUT",
          ),
        );
        return;
      }
      if (next.kind === "ABORTED" || linked.controller.signal.aborted) {
        finish("ABORTED");
        yield failureResult(
          callFailure(operation, 0, "REQUEST_ABORTED", "RPC_ABORTED"),
        );
        return;
      }
      if (next.kind === "THREW") {
        finish("FAILED");
        yield failureResult(
          callFailure(
            operation,
            0,
            "SERVER_FAILURE",
            "RPC_STREAM_EXECUTION_FAILED",
          ),
        );
        return;
      }
      if (next.value.done) {
        if (!terminal) {
          finish("CONTRACT_REJECTED");
          yield failureResult(protocolFailure(operation, 0));
          return;
        }
        if (!terminal.ok) {
          const mapped = mapTransportFailure(
            operation,
            0,
            terminal.failure,
          );
          finish(
            mapped.kind === "REQUEST_ABORTED"
              ? "ABORTED"
              : mapped.kind === "REQUEST_TIMEOUT"
                ? "TIMEOUT"
                : "FAILED",
          );
          yield failureResult(mapped);
          return;
        }
        finish("SUCCESS");
        return;
      }

      const frame = validateStreamFrame(next.value.value);
      if (!frame || terminal) {
        finish("CONTRACT_REJECTED");
        yield failureResult(protocolFailure(operation, 0));
        return;
      }
      if (frame.kind === "TERMINAL") {
        terminal = frame.ok
          ? Object.freeze({ ok: true })
          : Object.freeze({ ok: false, failure: frame.failure });
        continue;
      }

      messageCount += 1;
      totalBytes += frame.encodedBytes;
      if (
        messageCount > operation.maxResponseMessages ||
        frame.encodedBytes > operation.maxResponseMessageBytes ||
        totalBytes > operation.maxTotalResponseBytes
      ) {
        linked.controller.abort("message-limit");
        finish("CONTRACT_REJECTED");
        yield failureResult(
          callFailure(
            operation,
            0,
            "RESPONSE_BODY_LIMIT",
            "RPC_STREAM_MESSAGE_LIMIT",
          ),
        );
        return;
      }
      const mapped = mapResponse(
        installed,
        operation,
        0,
        frame.message,
        frame.encodedBytes,
        generation,
        generationFence,
        isEvent,
        deadlineAt,
        clock,
      );
      if (!mapped.ok) {
        linked.controller.abort("mapping-failure");
        finish(
          mapped.error.kind === "REQUEST_TIMEOUT"
            ? "TIMEOUT"
            : "CONTRACT_REJECTED",
        );
        yield mapped;
        return;
      }
      yield mapped;
    }
  } finally {
    // R-01 / RPC-RR-01. The commit/admission generation is fenced immediately
    // and listeners are released without waiting for the transport.
    // `iterator.return()` is a cleanup request, not a lifecycle authority, so
    // the lease's own `cancel` is what actually stops the physical stream.
    linked.controller.abort("stream-closed");
    linked.cleanup();
    const registered = lease
      ? activeStreams.get(operation.operationId)
      : undefined;
    if (registered && registered.streamId === lease?.streamId) {
      try {
        registered.cancel("stream-closed");
      } catch {
        // A transport that refuses to cancel stays DRAINING below.
      }
    }
    if (registered && registered.streamId === lease?.streamId) {
      // RPC-02. The positive-close subscription is installed before any
      // fallible cleanup. Running cleanup first meant a throwing `return`
      // accessor could skip it entirely and strand the registry entry.
      // The entry is removed only once the transport confirms the physical
      // stream closed. Until then the operation stays DRAINING and admits
      // nothing new — a bounded wait here would re-open the very hole this
      // registry exists to close.
      void registered.closed.then((confirmed) => {
        if (
          confirmed &&
          activeStreams.get(operation.operationId) === registered
        ) {
          activeStreams.delete(operation.operationId);
        }
      });
    }
    const returnIterator = safeIteratorReturn(iterator);
    if (returnIterator) {
      const cleanup = Promise.resolve()
        .then(async () => await returnIterator())
        // Cleanup cannot replace the already selected stream outcome.
        .catch(() => undefined);
      await boundedStreamCleanup(cleanup, clock, STREAM_CLEANUP_BOUND_MS);
    }
    if (registered && registered.streamId === lease?.streamId) {
      await boundedStreamCleanup(
        registered.closed.then(() => undefined),
        clock,
        STREAM_CLEANUP_BOUND_MS,
      );
    }
    finish("ABORTED");
  }
}

/**
 * R-04. Transports are collaborators, not data rows, so only their identity is
 * snapshotted; the callable itself is captured once and rebound.
 */
const TRANSPORT_KEYS: ReadonlySet<string> = new Set([
  "runtimeProfileId",
  "providerId",
  "protocol",
  "rpcKind",
  "invokeUnary",
  "openServerStream",
]);

/**
 * RPC-RR-03. Decodes the transport registry from own data descriptors only, so
 * no accessor on a caller's object is ever invoked, and returns a read facade
 * whose backing store cannot be reached by `set`, `delete` or `clear`.
 */
function snapshotTransports(
  source: Readonly<Record<string, BrowserRpcTransport>>,
): ReadOnlyRegistry<string, BrowserRpcTransport> {
  const installed = new Map<string, BrowserRpcTransport>();
  if (Object.getOwnPropertySymbols(source).length > 0) {
    throw new TypeError("Browser RPC transport registry has symbol keys.");
  }
  for (const key of Object.keys(source)) {
    const descriptor = Object.getOwnPropertyDescriptor(source, key);
    if (!descriptor || !("value" in descriptor)) {
      throw new TypeError(
        `Browser RPC transport registry entry is not a data property: ${key}`,
      );
    }
    const transport = descriptor.value as unknown;
    if (!transport || typeof transport !== "object") {
      throw new TypeError(
        `Browser RPC transport registry entry is not an object: ${key}`,
      );
    }
    if (Object.getOwnPropertySymbols(transport).length > 0) {
      throw new TypeError(
        `Browser RPC transport row has symbol keys: ${key}`,
      );
    }
    const row: Record<string, unknown> = Object.create(null);
    for (const field of Object.getOwnPropertyNames(transport)) {
      if (!TRANSPORT_KEYS.has(field)) {
        throw new TypeError(
          `Browser RPC transport row has an unexpected key: ${key}.${field}`,
        );
      }
      const fieldDescriptor = Object.getOwnPropertyDescriptor(
        transport,
        field,
      );
      if (!fieldDescriptor || !("value" in fieldDescriptor)) {
        throw new TypeError(
          `Browser RPC transport row key is not a data property: ${key}.${field}`,
        );
      }
      row[field] = fieldDescriptor.value;
    }
    const invokeUnary = row.invokeUnary;
    const openServerStream = row.openServerStream;
    installed.set(
      key,
      Object.freeze({
        runtimeProfileId: row.runtimeProfileId,
        providerId: row.providerId,
        protocol: row.protocol,
        rpcKind: row.rpcKind,
        ...(typeof invokeUnary === "function"
          ? { invokeUnary: invokeUnary.bind(transport) }
          : {}),
        ...(typeof openServerStream === "function"
          ? { openServerStream: openServerStream.bind(transport) }
          : {}),
      }) as BrowserRpcTransport,
    );
  }
  return createReadOnlyRegistry(installed);
}

function validateRuntimeDependencies(
  installed: InstalledBrowserRpcContractBindings,
  transports: ReadOnlyRegistry<string, BrowserRpcTransport>,
): void {
  const runtimeBindings: Record<
    string,
    Pick<
      BrowserRpcTransport,
      "runtimeProfileId" | "providerId" | "protocol" | "rpcKind"
    >
  > = Object.create(null);
  for (const [profileId, transport] of transports.entries()) {
    if (
      profileId !== transport.runtimeProfileId ||
      Object.hasOwn(runtimeBindings, profileId)
    ) {
      throw new TypeError(
        `Browser RPC transport registry is invalid: ${profileId}`,
      );
    }
    runtimeBindings[profileId] = Object.freeze({
      runtimeProfileId: transport.runtimeProfileId,
      providerId: transport.providerId,
      protocol: transport.protocol,
      rpcKind: transport.rpcKind,
    });
  }
  // Only the installed snapshot reaches the join validation.
  validateBrowserRpcContractBindings({
    operations: Object.fromEntries(installed.operations.entries()),
    profiles: Object.fromEntries(installed.profiles.entries()),
    schemaCodecs: Object.fromEntries(installed.schemaCodecs.entries()),
    mappers: Object.fromEntries(installed.mappers.entries()),
    requestEncoders: Object.fromEntries(installed.requestEncoders.entries()),
    runtimeBindings: Object.freeze(runtimeBindings),
  });
  for (const operation of installed.operations.values()) {
    if (!transports.has(operation.runtimeProfileId)) {
      throw new TypeError(
        `Browser RPC transport is missing: ${operation.operationId}`,
      );
    }
  }
}

function prepareRequest(
  installed: InstalledBrowserRpcContractBindings,
  operation: BrowserRpcOperationV3,
  encoder: BrowserRpcRequestEncoder,
  input: unknown,
  attempt: number,
): PreparedRequest {
  const schema = installed.schemaCodecs.get(operation.requestSchemaId);
  let validated;
  try {
    validated = schema?.parse(input);
  } catch {
    validated = undefined;
  }
  if (!validated?.success) {
    return Object.freeze({
      ok: false,
      error: callFailure(
        operation,
        attempt,
        "VALIDATION_REJECTED",
        "RPC_REQUEST_SCHEMA_INVALID",
      ),
    });
  }
  try {
    const encoded = encoder.encode(validated.data);
    if (
      !encoded.ok ||
      !validEncodedByteCount(
        encoded.encodedBytes,
        operation.maxRequestMessageBytes,
      )
    ) {
      return Object.freeze({
        ok: false,
        error: callFailure(
          operation,
          attempt,
          "MAPPING_CONTRACT_VIOLATION",
          encoded.ok ? "RPC_REQUEST_MESSAGE_LIMIT" : safeCode(encoded.code),
        ),
      });
    }
    return Object.freeze({
      ok: true,
      value: encoded.value,
      encodedBytes: encoded.encodedBytes,
    });
  } catch {
    return Object.freeze({
      ok: false,
      error: callFailure(
        operation,
        attempt,
        "MAPPING_CONTRACT_VIOLATION",
        "RPC_REQUEST_ENCODING_FAILED",
      ),
    });
  }
}

function mapResponse<Output>(
  installed: InstalledBrowserRpcContractBindings,
  operation: BrowserRpcOperationV3,
  attempt: number,
  message: unknown,
  encodedBytes: number,
  generation: unknown,
  generationFence: BrowserRpcGenerationFence,
  isOutput: (value: unknown) => value is Output,
  deadlineAt: number,
  clock: ClockPort,
): Result<Output, AppFailure> {
  if (
    !validEncodedByteCount(
      encodedBytes,
      operation.maxResponseMessageBytes,
    )
  ) {
    return failureResult(
      callFailure(
        operation,
        attempt,
        "RESPONSE_BODY_LIMIT",
        "RPC_RESPONSE_MESSAGE_LIMIT",
      ),
    );
  }
  if (safeNowOrDeadline(clock, deadlineAt) >= deadlineAt) {
    return failureResult(
      callFailure(
        operation,
        attempt,
        "REQUEST_TIMEOUT",
        "RPC_TOTAL_DEADLINE_EXCEEDED",
      ),
    );
  }

  const schema = installed.schemaCodecs.get(operation.responseSchemaId);
  let validated;
  try {
    validated = schema?.parse(message);
  } catch {
    validated = undefined;
  }
  if (!validated?.success) {
    return failureResult(
      callFailure(
        operation,
        attempt,
        "SCHEMA_MISMATCH",
        "RPC_RESPONSE_SCHEMA_INVALID",
      ),
    );
  }
  const mapper = installed.mappers.get(operation.mapperId);
  let mapped;
  try {
    mapped = mapper?.map(validated.data);
  } catch {
    mapped = undefined;
  }
  if (!mapped?.ok) {
    return failureResult(
      callFailure(
        operation,
        attempt,
        "MAPPING_CONTRACT_VIOLATION",
        mapped?.code ?? "RPC_RESPONSE_MAPPING_FAILED",
      ),
    );
  }
  const output = mapped.value;
  const outputMatches = safelyMatches(isOutput, output);
  if (!outputMatches) {
    return failureResult(
      callFailure(
        operation,
        attempt,
        "MAPPING_CONTRACT_VIOLATION",
        "RPC_BOUND_RESULT_TYPE_MISMATCH",
      ),
    );
  }
  if (!safeIsCurrent(generationFence, generation)) {
    return failureResult(
      callFailure(
        operation,
        attempt,
        "SCOPE_GENERATION_CHANGED",
        "RPC_SCOPE_GENERATION_CHANGED",
      ),
    );
  }
  if (safeNowOrDeadline(clock, deadlineAt) >= deadlineAt) {
    return failureResult(
      callFailure(
        operation,
        attempt,
        "REQUEST_TIMEOUT",
        "RPC_TOTAL_DEADLINE_EXCEEDED",
      ),
    );
  }
  return Object.freeze({ ok: true, value: output });
}

/**
 * R-06. Collaborator failures must not escape the closed Result boundary. A
 * broken clock is canonicalised as a runtime dependency failure and a broken
 * fence fails closed as a generation change; neither becomes a native rejection
 * on a port that promises `Result`.
 */
const CLOCK_FAILURE = Symbol("RPC_CLOCK_FAILURE");
const FENCE_FAILURE = Symbol("RPC_FENCE_FAILURE");

function safeNow(clock: ClockPort): number | typeof CLOCK_FAILURE {
  try {
    const value = clock.now();
    return Number.isFinite(value) ? value : CLOCK_FAILURE;
  } catch {
    return CLOCK_FAILURE;
  }
}

/** A broken clock reads as "the deadline has passed", never as a rejection. */
/** Cleanup bound for a non-cooperative transport iterator. */
const STREAM_CLEANUP_BOUND_MS = 1_000;

/**
 * R-01. Waits for transport cleanup only within a bound. A cleanup that has not
 * settled stays observed (so it cannot surface as an unhandled rejection) and
 * the caller-facing generator completes regardless.
 */
async function boundedStreamCleanup(
  cleanup: Promise<unknown>,
  clock: ClockPort,
  boundMs: number,
): Promise<void> {
  const timer = new AbortController();
  const bounded = Promise.resolve()
    .then(async () => {
      await clock.sleep(boundMs, timer.signal);
    })
    .catch(() => undefined);
  await Promise.race([cleanup.then(() => undefined), bounded]);
  timer.abort();
}

function safeNowOrDeadline(clock: ClockPort, deadlineAt: number): number {
  const value = safeNow(clock);
  return value === CLOCK_FAILURE ? deadlineAt : value;
}

function safeCapture(
  fence: BrowserRpcGenerationFence,
): unknown | typeof FENCE_FAILURE {
  try {
    return fence.capture();
  } catch {
    return FENCE_FAILURE;
  }
}

function safeIsCurrent(
  fence: BrowserRpcGenerationFence,
  generation: unknown,
): boolean {
  try {
    return fence.isCurrent(generation);
  } catch {
    // Fail closed: an unreadable fence is treated as a generation change.
    return false;
  }
}

function validateCallContext(
  operation: BrowserRpcOperationV3,
  context: Readonly<{ idempotencyKey?: string }>,
  attempt: number,
): AppFailure | null {
  const key = context.idempotencyKey;
  if (
    (operation.idempotencyKeyPolicy === "REQUIRED" &&
      !validIdempotencyKey(key)) ||
    (operation.idempotencyKeyPolicy === "NONE" && key !== undefined)
  ) {
    return callFailure(
      operation,
      attempt,
      "VALIDATION_REJECTED",
      "RPC_IDEMPOTENCY_KEY_INVALID",
    );
  }
  return null;
}

function validIdempotencyKey(value: unknown): value is string {
  return (
    typeof value === "string" &&
    value.length >= 8 &&
    value.length <= 200 &&
    ![...value].some((character) => {
      const codePoint = character.codePointAt(0) ?? 0;
      return codePoint <= 31 || codePoint === 127 || /\s/u.test(character);
    })
  );
}

function safelyMatches<Output>(
  guard: (value: unknown) => value is Output,
  value: unknown,
): value is Output {
  try {
    return guard(value);
  } catch {
    return false;
  }
}

/**
 * RPC-RR-04. A transport value is decoded, never adopted.
 *
 * `in` and a direct property read run accessors, so a throwing getter escapes
 * the Result contract and an inherited or extra field passes unseen. Each
 * variant is decoded from own data descriptors into a new frozen value, so a
 * transport that mutates its own object after returning it cannot change what
 * the runtime already admitted.
 */
function ownDataValue(source: unknown, key: string): unknown {
  if (source === null || typeof source !== "object") return undefined;
  try {
    const descriptor = Object.getOwnPropertyDescriptor(source, key);
    if (!descriptor || !("value" in descriptor)) return undefined;
    return descriptor.value;
  } catch {
    return undefined;
  }
}

/**
 * RPC-04. An exact union means exactly these own data keys, all of them, on a
 * plain object. Checking only that each own name was *allowed* let a result
 * carry required fields it never declared and let a custom prototype smuggle
 * metadata past the trust boundary while the shape still looked valid.
 */
function exactOwnKeys(
  source: unknown,
  allowed: ReadonlySet<string>,
  required: ReadonlySet<string> = allowed,
): boolean {
  if (source === null || typeof source !== "object") return false;
  try {
    if (Object.getOwnPropertySymbols(source).length > 0) return false;
    const prototype = Reflect.getPrototypeOf(source);
    if (prototype !== Object.prototype && prototype !== null) return false;
    const present = new Set<string>();
    for (const key of Object.getOwnPropertyNames(source)) {
      if (!allowed.has(key)) return false;
      const descriptor = Object.getOwnPropertyDescriptor(source, key);
      if (!descriptor || !("value" in descriptor)) return false;
      if (descriptor.enumerable !== true) return false;
      present.add(key);
    }
    for (const key of required) {
      if (!present.has(key)) return false;
    }
    return true;
  } catch {
    return false;
  }
}

const UNARY_OK_KEYS: ReadonlySet<string> = new Set([
  "ok",
  "message",
  "encodedBytes",
]);
/** `message` is a wire field, so an absent one is a protocol breach even when
 * a permissive schema would happily accept `undefined`. */
const UNARY_OK_REQUIRED: ReadonlySet<string> = UNARY_OK_KEYS;
const UNARY_FAILED_KEYS: ReadonlySet<string> = new Set(["ok", "failure"]);
const FRAME_MESSAGE_KEYS: ReadonlySet<string> = new Set([
  "kind",
  "message",
  "encodedBytes",
]);
const FRAME_TERMINAL_OK_KEYS: ReadonlySet<string> = new Set(["kind", "ok"]);
const FRAME_TERMINAL_FAILED_KEYS: ReadonlySet<string> = new Set([
  "kind",
  "ok",
  "failure",
]);
const TRANSPORT_FAILURE_KEYS: ReadonlySet<string> = new Set([
  "code",
  "retryAfterMs",
]);
/** `retryAfterMs` is genuinely optional; `code` is not. */
const TRANSPORT_FAILURE_REQUIRED: ReadonlySet<string> = new Set(["code"]);

function validateUnaryTransportResult(
  value: unknown,
): BrowserRpcUnaryTransportResult | null {
  const ok = ownDataValue(value, "ok");
  if (typeof ok !== "boolean") return null;
  if (ok) {
    if (!exactOwnKeys(value, UNARY_OK_KEYS, UNARY_OK_REQUIRED)) return null;
    const encodedBytes = ownDataValue(value, "encodedBytes");
    if (!validEncodedByteCount(encodedBytes, Number.MAX_SAFE_INTEGER)) {
      return null;
    }
    return Object.freeze({
      ok: true as const,
      message: ownDataValue(value, "message"),
      encodedBytes,
    });
  }
  if (!exactOwnKeys(value, UNARY_FAILED_KEYS)) return null;
  const failure = decodeTransportFailure(ownDataValue(value, "failure"));
  return failure === null
    ? null
    : Object.freeze({ ok: false as const, failure });
}

function validateStreamFrame(value: unknown): BrowserRpcStreamFrame | null {
  const kind = ownDataValue(value, "kind");
  if (kind === "MESSAGE") {
    if (!exactOwnKeys(value, FRAME_MESSAGE_KEYS)) return null;
    const encodedBytes = ownDataValue(value, "encodedBytes");
    if (!validEncodedByteCount(encodedBytes, Number.MAX_SAFE_INTEGER)) {
      return null;
    }
    return Object.freeze({
      kind: "MESSAGE" as const,
      message: ownDataValue(value, "message"),
      encodedBytes,
    });
  }
  if (kind !== "TERMINAL") return null;
  const ok = ownDataValue(value, "ok");
  if (typeof ok !== "boolean") return null;
  if (ok) {
    if (!exactOwnKeys(value, FRAME_TERMINAL_OK_KEYS)) return null;
    return Object.freeze({ kind: "TERMINAL" as const, ok: true as const });
  }
  if (!exactOwnKeys(value, FRAME_TERMINAL_FAILED_KEYS)) return null;
  const failure = decodeTransportFailure(ownDataValue(value, "failure"));
  return failure === null
    ? null
    : Object.freeze({
        kind: "TERMINAL" as const,
        ok: false as const,
        failure,
      });
}

function decodeTransportFailure(
  value: unknown,
): BrowserRpcTransportFailure | null {
  if (!exactOwnKeys(value, TRANSPORT_FAILURE_KEYS, TRANSPORT_FAILURE_REQUIRED)) {
    return null;
  }
  const code = ownDataValue(value, "code");
  const retryAfterMs = ownDataValue(value, "retryAfterMs");
  if (!TRANSPORT_FAILURE_CODES.has(code as BrowserRpcTransportFailureCode)) {
    return null;
  }
  if (retryAfterMs === undefined) {
    return Object.freeze({ code: code as BrowserRpcTransportFailureCode });
  }
  if (
    !Number.isSafeInteger(retryAfterMs) ||
    (retryAfterMs as number) < 0 ||
    (retryAfterMs as number) > BROWSER_RPC_HARD_LIMITS.maxRetryAfterMs
  ) {
    return null;
  }
  return Object.freeze({
    code: code as BrowserRpcTransportFailureCode,
    retryAfterMs: retryAfterMs as number,
  });
}

function validTransportFailure(
  failure: unknown,
): failure is BrowserRpcTransportFailure {
  if (!failure || typeof failure !== "object" || !("code" in failure)) {
    return false;
  }
  const candidate = failure as BrowserRpcTransportFailure;
  return (
    TRANSPORT_FAILURE_CODES.has(candidate.code) &&
    (candidate.retryAfterMs === undefined ||
      (Number.isSafeInteger(candidate.retryAfterMs) &&
        candidate.retryAfterMs >= 0 &&
        candidate.retryAfterMs <= BROWSER_RPC_HARD_LIMITS.maxRetryAfterMs))
  );
}

const TRANSPORT_FAILURE_CODES =
  new Set<BrowserRpcTransportFailureCode>([
    "NETWORK_UNREACHABLE",
    "CANCELED",
    "DEADLINE_EXCEEDED",
    "UNAUTHENTICATED",
    "PERMISSION_DENIED",
    "NOT_FOUND",
    "ALREADY_EXISTS",
    "ABORTED",
    "FAILED_PRECONDITION",
    "INVALID_ARGUMENT",
    "RESOURCE_EXHAUSTED",
    "UNAVAILABLE",
    "UNIMPLEMENTED",
    "INTERNAL",
    "DATA_LOSS",
    "PROTOCOL_MISMATCH",
    "MESSAGE_LIMIT",
  ]);

function shouldRetry(
  operation: BrowserRpcOperationV3,
  profile: BrowserRpcProviderProfile,
  failure: BrowserRpcTransportFailure,
  attempt: number,
  idempotencyKey: string | undefined,
): boolean {
  return (
    profile.retryOwner === "FRONTEND_ADAPTER" &&
    attempt + 1 < profile.maxAttempts &&
    profile.retryableFailures.includes(failure.code) &&
    (["SAFE", "IDEMPOTENT"].includes(operation.replayPolicy) ||
      (operation.replayPolicy === "KEYED_COMMAND" &&
        validIdempotencyKey(idempotencyKey)))
  );
}

function retryDelay(
  profile: BrowserRpcProviderProfile,
  failure: BrowserRpcTransportFailure,
  attempt: number,
): number {
  const backoff = profile.backoffMs[attempt] ?? 0;
  const retryAfter = Math.min(
    failure.retryAfterMs ?? 0,
    profile.maxRetryAfterMs,
  );
  return Math.max(backoff, retryAfter);
}

function mapTransportFailure(
  operation: BrowserRpcOperationV3,
  attempt: number,
  failure: BrowserRpcTransportFailure,
): AppFailure {
  const kind: FailureKind =
    failure.code === "NETWORK_UNREACHABLE"
      ? "NETWORK_UNREACHABLE"
      : failure.code === "CANCELED"
        ? "REQUEST_ABORTED"
        : failure.code === "DEADLINE_EXCEEDED"
          ? "REQUEST_TIMEOUT"
          : failure.code === "UNAUTHENTICATED"
            ? "AUTH_REQUIRED"
            : failure.code === "PERMISSION_DENIED"
              ? "FORBIDDEN"
              : failure.code === "NOT_FOUND"
                ? "NOT_FOUND"
                : ["ALREADY_EXISTS", "ABORTED"].includes(failure.code)
                  ? "CONFLICT"
                  : ["FAILED_PRECONDITION", "INVALID_ARGUMENT"].includes(
                        failure.code,
                      )
                    ? "VALIDATION_REJECTED"
                    : failure.code === "RESOURCE_EXHAUSTED"
                      ? "RATE_LIMITED"
                      : failure.code === "MESSAGE_LIMIT"
                        ? "RESPONSE_BODY_LIMIT"
                        : ["PROTOCOL_MISMATCH", "UNIMPLEMENTED"].includes(
                              failure.code,
                            )
                          ? "API_CONTRACT_MISMATCH"
                          : "SERVER_FAILURE";
  return createFailure(kind, operation.operationId, attempt, {
    code: `RPC_${failure.code}`,
    ...(failure.retryAfterMs !== undefined
      ? { retryAfterMs: failure.retryAfterMs }
      : {}),
  });
}

function protocolFailure(
  operation: BrowserRpcOperationV3,
  attempt: number,
): AppFailure {
  return callFailure(
    operation,
    attempt,
    "API_CONTRACT_MISMATCH",
    "RPC_PROTOCOL_MISMATCH",
  );
}

function callFailure(
  operation: BrowserRpcOperationV3,
  attempt: number,
  kind: FailureKind,
  code: string,
): AppFailure {
  return createFailure(kind, operation.operationId, attempt, {
    code: safeCode(code),
  });
}

function safeCode(value: string): string {
  return /^[A-Z][A-Z0-9_]{2,79}$/.test(value)
    ? value
    : "RPC_ADAPTER_REJECTED";
}

function failureResult(
  error: AppFailure,
): Readonly<{ ok: false; error: AppFailure }> {
  return Object.freeze({ ok: false, error });
}

function validEncodedByteCount(
  value: unknown,
  maximum: number,
): value is number {
  return (
    typeof value === "number" &&
    Number.isSafeInteger(value) &&
    value >= 0 &&
    value <= maximum
  );
}

function linkedAbortController(external?: AbortSignal): Readonly<{
  controller: AbortController;
  cleanup(): void;
}> {
  const controller = new AbortController();
  const onAbort = () => controller.abort(external?.reason);
  external?.addEventListener("abort", onAbort, { once: true });
  if (external?.aborted) onAbort();
  return Object.freeze({
    controller,
    cleanup() {
      external?.removeEventListener("abort", onAbort);
    },
  });
}

async function raceWithin<Value>(
  work: Promise<Value>,
  milliseconds: number,
  signal: AbortSignal,
  clock: ClockPort,
): Promise<TimedResult<Value>> {
  if (signal.aborted) return Object.freeze({ kind: "ABORTED" });
  const timerController = new AbortController();
  const onAbort = () => timerController.abort(signal.reason);
  signal.addEventListener("abort", onAbort, { once: true });
  try {
    const workResult = work.then<TimedResult<Value>, TimedResult<Value>>(
      (value) => Object.freeze({ kind: "VALUE", value }),
      () => Object.freeze({ kind: "THREW" }),
    );
    // RPC-RR-02. A clock is an external collaborator: calling `sleep` inside a
    // promise boundary turns its synchronous throw into a normalized rejection
    // instead of an exception that escapes the Result contract and skips the
    // listener and timer release below.
    const timerResult = Promise.resolve()
      .then(() => clock.sleep(milliseconds, timerController.signal))
      .then<TimedResult<Value>, TimedResult<Value>>(
        () => Object.freeze({ kind: "TIMEOUT" }),
        () => Object.freeze({ kind: "ABORTED" }),
      );
    const selected = await Promise.race([workResult, timerResult]);
    return signal.aborted && selected.kind === "VALUE"
      ? Object.freeze({ kind: "ABORTED" })
      : selected;
  } finally {
    timerController.abort("race-complete");
    signal.removeEventListener("abort", onAbort);
  }
}

function observe(
  dependencies: BrowserRpcRuntimeDependencies,
  operation: BrowserRpcOperationV3,
  outcome: BrowserRpcObservationOutcome,
  attemptCount: number,
  messageCount: number,
): void {
  try {
    dependencies.observations?.observe(
      Object.freeze({
        operationId: operation.operationId,
        protocol: operation.protocol,
        runtimeProfileId: operation.runtimeProfileId,
        rpcKind: operation.rpcKind,
        outcome,
        attemptCount: Math.max(1, attemptCount),
        messageCount: Math.max(0, messageCount),
      }),
    );
  } catch {
    // Observation cannot change the selected application result.
  }
}