| null;
+ return {
+ ...last,
+ 'serve.usage.prior': facts['serve.usage.prior'],
+ 'response.usage.billable': move(observed),
+ ...(pending === null ? {} : {
+ [pendingUsage!]: move(defer(pending.then(outcome => ({ ...outcome, billable: [...prior, ...outcome.billable] })))),
+ }),
+ };
+ }
+ await use.log.info('candidate failed, trying the next', { upstream: candidate.upstreamId });
+ }
+ if (last === undefined) throw new Error('failover: assembly handed it an empty candidate list');
+ // Every candidate failed, and the last failure is the base — so the client sees real
+ // upstream telemetry rather than a synthesized gateway envelope.
+ return { ...last, 'serve.usage.prior': facts['serve.usage.prior'], 'response.usage.billable': move(observed) };
+ },
+});
diff --git a/packages/gateway/src/data-plane/pipeline/provider-entry.ts b/packages/gateway/src/data-plane/pipeline/provider-entry.ts
new file mode 100644
index 0000000000..4ea7645f5c
--- /dev/null
+++ b/packages/gateway/src/data-plane/pipeline/provider-entry.ts
@@ -0,0 +1,20 @@
+import type { AttemptSelector } from './facts.ts';
+import { filterInboundHeadersForProvider } from '../shared/inbound-headers.ts';
+import { move } from '@floway-dev/pipeline';
+import { providerModelFacts, providerModelOf, type ModelCandidate, type ProviderRequest } from '@floway-dev/provider';
+
+export const providerEntry = (
+ facts: { 'route.attempt': AttemptSelector; 'ingress.http.headers': readonly (readonly [string, string])[] },
+ candidate: ModelCandidate,
+ payload: P,
+ forwardedHeaders: readonly (readonly [string, string])[] = facts['ingress.http.headers'],
+): ProviderRequest
& Record => move({
+ ...facts,
+ 'request.provider.model': providerModelFacts(providerModelOf(candidate)),
+ 'request.provider.payload': payload,
+ 'request.http.callId': facts['route.attempt'].candidateId,
+ 'request.http.headers': [...filterInboundHeadersForProvider(
+ new Headers(forwardedHeaders.map(([name, value]): [string, string] => [name, value])),
+ candidate.provider,
+ )],
+});
diff --git a/packages/gateway/src/data-plane/pipeline/provider-usage.ts b/packages/gateway/src/data-plane/pipeline/provider-usage.ts
new file mode 100644
index 0000000000..e96ef0556f
--- /dev/null
+++ b/packages/gateway/src/data-plane/pipeline/provider-usage.ts
@@ -0,0 +1,25 @@
+import type { BillableEntity } from './facts.ts';
+import { telemetryModelIdentity } from '../shared/telemetry/attribution.ts';
+import type { Facts } from '@floway-dev/pipeline';
+import type { ModelCandidate, ProviderResponse } from '@floway-dev/provider';
+
+export const providerUsage = (
+ candidate: ModelCandidate,
+ response: Pick,
+ finalCall: readonly BillableEntity[],
+): readonly BillableEntity[] => {
+ const previous = response['response.provider.previousCalls'];
+ if (previous.length === 0) return finalCall;
+ return [
+ ...previous.map(call => ({ identity: telemetryModelIdentity(candidate, call.modelKey), quantities: {} })),
+ ...finalCall,
+ ];
+};
+
+export const providerCalls = (candidate: ModelCandidate, facts: Facts): readonly BillableEntity[] => {
+ if (!('response.provider.called' in facts)) return [];
+ const response = facts as unknown as ProviderResponse;
+ const current: readonly BillableEntity[] = response['response.provider.called']
+ ? [{ identity: telemetryModelIdentity(candidate, response['response.provider.modelKey']), quantities: {} }] : [];
+ return providerUsage(candidate, response, current);
+};
diff --git a/packages/gateway/src/data-plane/pipeline/resolve-candidates.ts b/packages/gateway/src/data-plane/pipeline/resolve-candidates.ts
new file mode 100644
index 0000000000..953af1f3dc
--- /dev/null
+++ b/packages/gateway/src/data-plane/pipeline/resolve-candidates.ts
@@ -0,0 +1,98 @@
+import type { GatewayServices } from './services.ts';
+import type { Slice } from './stage-contracts.ts';
+import { enumerateModelCandidates } from '../providers/resolution.ts';
+import { appendFailedUpstreams } from '../shared/failed-upstreams.ts';
+import { defineStage, move } from '@floway-dev/pipeline';
+import type { ModelKind } from '@floway-dev/protocols/common';
+import type { ModelCandidate } from '@floway-dev/provider';
+
+/** What a family narrows its candidates by, and what it says when nothing is left. A
+ * candidate that resolves but cannot serve this request — no endpoint for the kind, or a
+ * payload the target protocol cannot express — is not a candidate, and saying why is what
+ * turns an empty list into a usable 400 rather than a bare 404. */
+export interface Narrowing {
+ readonly kind: ModelKind;
+ /** Keeps a candidate, or says in one phrase why it cannot serve this request. */
+ readonly reject: (candidate: ModelCandidate) => string | null;
+ /** What the client is told when the model resolved but nothing can serve the request.
+ * `reasons` holds what `reject` said, and is empty when no candidate of this kind was
+ * found at all — which is the difference between "this model is not an embeddings model"
+ * and "this reranker cannot do what this request asks". Families differ in how much of
+ * that they spell out, so the sentence is the family's rather than this stage's. */
+ readonly unsupported: (model: string, reasons: readonly string[]) => string;
+ /** What this family answers with when there is no candidate to try. The family decides
+ * which of its own keys carries the failure, which is why the refusal slice is a type
+ * parameter rather than a shared key: a reusable stage whose short-circuit provides
+ * family-specific keys cannot be written against the shared space alone. */
+ readonly refuse: (status: number, message: string) => Refusal;
+ /** The same keys as strings, so assembly can check that a short-circuit here covers what
+ * the stages above it need. */
+ readonly refuses: readonly (keyof Refusal)[];
+}
+
+/**
+ * Provides `serve.candidates`, or answers with the failure that says why there are none —
+ * both traits, because "no upstream serves this model" is an answer this stage already
+ * holds and nothing below it could produce one.
+ */
+export const resolveCandidates = (narrowing: Narrowing) => defineStage<
+ Slice<'serve.model'>, // what arrives
+ Slice<'serve.model' | 'serve.candidates'>, // what it hands down
+ Slice<'response.usage.billable' | 'response.http.headers' | 'response.http.status'>, // what comes back
+ Slice<'response.usage.billable' | 'response.http.headers' | 'response.http.status'>, // what it hands up, having descended
+ Slice<'response.usage.billable' | 'response.http.headers' | 'response.http.status'> & Refusal, // and what it answers with instead
+ GatewayServices
+>({
+ name: 'resolveCandidates',
+ through: {
+ request: { needs: ['serve.model'], consumes: [], provides: ['serve.candidates'] },
+ response: { needs: ['response.usage.billable', 'response.http.headers'], consumes: [], provides: [] },
+ },
+ return: { provides: ['response.usage.billable', 'response.http.headers', 'response.http.status', ...narrowing.refuses] },
+ execute: async (facts, next, use) => {
+ const model = facts['serve.model'];
+ const { candidates, sawModel, failedUpstreams } = await enumerateModelCandidates({
+ subject: use.gateway.user,
+ upstreamIds: use.gateway.upstreamIds,
+ randomizeUpstreams: use.gateway.randomizeUpstreams,
+ model,
+ kind: narrowing.kind,
+ scheduler: use.gateway.backgroundScheduler,
+ runtimeLocation: use.gateway.runtimeLocation,
+ });
+
+ // An empty billed set is what "we did not call an upstream" looks like, and an empty
+ // header list is the same statement on the other key. The settlement stages still run
+ // and still write; the row simply names no billed entity.
+ const refuse = (status: number, message: string) =>
+ move({
+ ...facts,
+ 'response.usage.billable': [],
+ 'response.http.headers': [],
+ 'response.http.status': status,
+ ...narrowing.refuse(status, message),
+ });
+
+ if (candidates.length === 0) {
+ const missing = sawModel
+ ? narrowing.unsupported(model, [])
+ : `Model ${model} is not available on any configured upstream.`;
+ return refuse(sawModel ? 400 : 404, appendFailedUpstreams(missing, failedUpstreams));
+ }
+
+ const refused = new Set();
+ const viable = candidates.filter(candidate => {
+ const why = narrowing.reject(candidate);
+ if (why !== null) refused.add(why);
+ return why === null;
+ });
+ // The live half stays with the resolver; only selectors travel.
+ if (viable.length === 0) {
+ await use.log.debug('no viable candidate', { model, refused: [...refused] });
+ return refuse(400, appendFailedUpstreams(narrowing.unsupported(model, [...refused]), failedUpstreams));
+ }
+
+ await use.log.debug('resolved candidates', { model, viable: viable.length, resolved: candidates.length });
+ return await next({ ...facts, 'serve.candidates': move(use.rememberCandidates(viable)) });
+ },
+});
diff --git a/packages/gateway/src/data-plane/pipeline/serialize-client-json.ts b/packages/gateway/src/data-plane/pipeline/serialize-client-json.ts
new file mode 100644
index 0000000000..299ac9bb0e
--- /dev/null
+++ b/packages/gateway/src/data-plane/pipeline/serialize-client-json.ts
@@ -0,0 +1,26 @@
+import type { GatewayFacts } from './facts.ts';
+import { defineStage, move } from '@floway-dev/pipeline';
+
+export const serializeClientJson = (renderedKey: K) => defineStage<
+ Record,
+ Record,
+ Record,
+ Pick
+>({
+ name: 'serializeClientJson',
+ through: {
+ request: { needs: [], consumes: [], provides: [] },
+ response: { needs: [renderedKey], consumes: [], provides: ['response.http.jsonBody'] },
+ },
+ execute: async (facts, next) => {
+ const back = await next(facts);
+ const rendered = back[renderedKey];
+ if (typeof rendered === 'object' && rendered !== null
+ && (rendered instanceof Uint8Array || Symbol.asyncIterator in rendered)) {
+ return move({ ...back, 'response.http.jsonBody': null });
+ }
+ const json = JSON.stringify(rendered);
+ if (json === undefined) throw new TypeError('Client JSON body has no JSON representation');
+ return move({ ...back, 'response.http.jsonBody': new TextEncoder().encode(json) });
+ },
+});
diff --git a/packages/gateway/src/data-plane/pipeline/serve.ts b/packages/gateway/src/data-plane/pipeline/serve.ts
new file mode 100644
index 0000000000..002118c6e3
--- /dev/null
+++ b/packages/gateway/src/data-plane/pipeline/serve.ts
@@ -0,0 +1,253 @@
+// The seam. Every family's route handler does the same things around its pipeline — read
+// the body, build what the run is given, run it, turn the exit facts into a response — so
+// it is written once here and each family adds only its own keys.
+//
+// This is also where "entering the pipeline system requires no capability" is concrete: a
+// handler calls `run` like any other caller. What makes a run *recorded* is the prologue
+// resolving a dump sink, not a flag anywhere below it.
+
+import type { Context } from 'hono';
+import { streamSSE } from 'hono/streaming';
+import type { ContentfulStatusCode } from 'hono/utils/http-status';
+
+import { createCandidateRegistry } from './candidates.ts';
+import type { AttemptSelector, BillableEntity, GatewayFacts } from './facts.ts';
+import type { GatewayServices, RunGatewayCtx } from './services.ts';
+import { openRunDump, type RunDump } from '../../dump/run-sink.ts';
+import { apiKeyFromContext, type AuthedContext } from '../../middleware/auth.ts';
+import { internalErrorResponse } from '../../middleware/internal-error-response.ts';
+import { backgroundSchedulerFromContext } from '../../runtime/background.ts';
+import { consoleLogSink } from '../../runtime/log.ts';
+import { stampUpstreamCallStart } from '../shared/attempt-timing.ts';
+import { createGatewayCtxFromHono, finalizeGatewayResponse, type AttemptState } from '../shared/gateway-ctx.ts';
+import { readRequestBody, takeRequestBody, type RequestBody } from '../shared/request-body.ts';
+import { writeSSEFrames } from '../shared/sse.ts';
+import { recordStream } from '@floway-dev/dump';
+import { run, getFailureFacts, type Pipeline } from '@floway-dev/pipeline';
+import { sseCommentFrame, type SseFrame } from '@floway-dev/protocols/common';
+
+type Slice = { [P in K]: GatewayFacts[P] };
+
+/** What the client sent, read once, before anything has decided what it means. */
+export interface Ingress {
+ readonly body: RequestBody;
+ readonly headers: readonly (readonly [string, string])[];
+}
+
+/**
+ * Reads the request.
+ *
+ * This is separate from opening the run because whether a request streams is written *in*
+ * the request — `stream: true` in a JSON body, a form field in a multipart upload — and the
+ * body can only be read once. A run opened before the read would have to guess, and guessing
+ * `false` is not harmless: the abort controller a streaming run cancels its upstream with is
+ * minted from that flag, so a client that disconnected would stop cancelling anything.
+ */
+export const readIngress = async (c: Context): Promise => ({
+ body: await readRequestBody(c),
+ headers: [...c.req.raw.headers],
+});
+
+export interface Prologue {
+ readonly runDump: RunDump | null;
+ readonly services: GatewayServices;
+ readonly gateway: RunGatewayCtx;
+ readonly headers: readonly (readonly [string, string])[];
+}
+
+/**
+ * Opens a run: the request context the telemetry stages read, and the services the stages
+ * are given. It takes the bytes the handler has already read and hands them to the dump,
+ * which is what leaves the handler's own copy free to be released.
+ *
+ * The candidate store is the resolver the ruling names — "the resolver is the service and
+ * the selector is a fact". A `ModelCandidate` carries the provider's instance, its fetcher
+ * and its models cache; those never enter the record, because `move()` would freeze them
+ * and the provider's own cache refresh would break. So the stage that enumerates hands the
+ * live ones here, and the stage that dials asks for one back by selector.
+ */
+export const openPrologue = (
+ c: AuthedContext,
+ ingress: Ingress,
+ options: { readonly wantsStream: boolean; readonly model?: string },
+): Prologue => {
+ const backgroundScheduler = backgroundSchedulerFromContext(c);
+ const attempt: AttemptState = { timing: { firstOutputTokenAt: null, upstreamCallStartedAt: null }, telemetry: undefined };
+ // The shape follows the endpoint. A pipelined turn is recorded as its whole run — every
+ // stage, both directions — so it opens that recording here instead of the edge one, and
+ // no turn is ever written twice.
+ const runDump = openRunDump(
+ apiKeyFromContext(c),
+ { method: c.req.method, path: new URL(c.req.raw.url).pathname, body: ingress.body },
+ backgroundScheduler,
+ options.wantsStream,
+ attempt.timing,
+ );
+ const gateway: RunGatewayCtx = {
+ ...createGatewayCtxFromHono(c, {
+ wantsStream: options.wantsStream,
+ attempt,
+ ...(options.model === undefined ? {} : { model: options.model }),
+ requestBody: takeRequestBody(ingress.body),
+ backgroundScheduler,
+ dump: runDump,
+ }),
+ dump: runDump,
+ };
+ const candidates = createCandidateRegistry();
+
+ return {
+ runDump,
+ gateway,
+ headers: ingress.headers,
+ services: {
+ gateway,
+ log: consoleLogSink,
+ background: work => { gateway.backgroundScheduler(work); },
+ ...candidates,
+ recordProtocolFrames: frames => recordStream(frames, runDump),
+ httpCall: callId => {
+ const candidate = candidates.resolveCandidate(callId);
+ return {
+ fetcher: candidate.fetcher,
+ signal: candidate.model.kind === 'image' ? undefined : gateway.abortSignal,
+ waitUntil: gateway.backgroundScheduler,
+ wrapUpstreamCall: stampUpstreamCallStart(gateway.attempt.timing),
+ };
+ },
+ // Absent when this key has no retention configured, which is what keeps recording
+ // conditional: the runner does none of it rather than doing it and discarding.
+ ...(runDump === null ? {} : { dump: runDump.sink }),
+
+ },
+ };
+};
+
+/** What a family writes for the client. The status and the upstream's own headers are facts
+ * the pipeline already carries, so what is left here is the answer itself. */
+export type Rendered =
+ | {
+ readonly body: BodyInit;
+ /** Owned by whichever stage serialized the body, never by the upstream — a media type
+ * describing bytes the gateway wrote itself is the one thing an upstream cannot say.
+ * `null` where a family carries an upstream's own body and the upstream declared none:
+ * inventing one would describe bytes nobody described. */
+ readonly contentType: string | null;
+ }
+ | {
+ /** An answer that *is* a stream. The frames go out as they arrive, which is why nothing
+ * above waits for them and why what they billed is settled afterwards. */
+ readonly frames: AsyncIterable;
+ };
+
+/** Which of the two an answer turned out to be. A family's rendered fact carries whichever
+ * shape its run produced, and only the family knows the key it is under. */
+export const isFrames = (rendered: unknown): rendered is AsyncIterable =>
+ typeof rendered === 'object' && rendered !== null && Symbol.asyncIterator in rendered;
+
+/** How a streaming turn ended, once its frames ran out. Both halves arrive together because
+ * both are known at the same moment — the last chunk states the usage, and running out
+ * without a terminal event is what "it did not finish" means. */
+export interface StreamOutcome {
+ readonly billable: readonly BillableEntity[];
+ readonly failed: boolean;
+}
+
+/**
+ * Runs a family's pipeline and turns what it answered with into a response.
+ *
+ * The drain is scheduled rather than awaited. A streaming family's answer *is* the stream,
+ * so draining before returning would consume the frames the client is waiting for. Release
+ * is not cancel — an aborted connection cannot be reused and leaves the upstream's own
+ * billing unsettled — so it still happens, just after the answer is on its way.
+ *
+ * A run that threw is answered here rather than by the app's own handler, because the record
+ * is closed here. Letting the throw out would lose the whole record: nothing above this knows
+ * a dump is open, so the turn that most needs explaining would be the one that left nothing
+ * behind. What goes to the client is the same envelope the app's handler writes, from the same
+ * function — the stack included, which is what a gateway owes an operator for its own fault.
+ */
+export const serveThrough = async <
+ Entry extends object,
+ Exit extends Slice<'response.http.status' | 'response.http.headers'>,
+>(
+ c: Context,
+ prologue: Prologue,
+ pipeline: Pipeline,
+ entry: Entry,
+ render: (facts: Exit) => Rendered,
+): Promise => {
+ try {
+ return await serveRun(c, prologue, pipeline, entry, render);
+ } catch (error) {
+ // `run` drains what it opened before it rethrows, so there is nothing left outstanding to
+ // release here — only the record to close, and the reason to put on it.
+ const failureFacts = getFailureFacts(error);
+ if (failureFacts !== undefined && 'route.attempt' in failureFacts) {
+ const attempt = failureFacts['route.attempt'] as AttemptSelector;
+ prologue.gateway.dump?.error('gateway', attempt.upstreamId);
+ }
+ prologue.gateway.dump?.failed(error);
+ return finalizeGatewayResponse(prologue.gateway, internalErrorResponse(asError(error), c));
+ }
+};
+
+/** Anything can be thrown; only an `Error` carries a stack. A throw that was not one is
+ * reported as what it was rather than being dressed up as something with a call site. */
+const asError = (thrown: unknown): Error =>
+ thrown instanceof Error ? thrown : new Error(String(thrown));
+
+const serveRun = async <
+ Entry extends object,
+ Exit extends Slice<'response.http.status' | 'response.http.headers'>,
+>(
+ c: Context,
+ prologue: Prologue,
+ pipeline: Pipeline,
+ entry: Entry,
+ render: (facts: Exit) => Rendered,
+): Promise => {
+ const { facts, drain } = await run(pipeline, entry, prologue.services as never);
+ prologue.runDump?.afterRun(drain);
+ let answer: Rendered;
+ try { answer = render(facts); } catch (error) { await drain(); throw error; }
+
+ const status = facts['response.http.status'] as ContentfulStatusCode;
+ if ('frames' in answer) {
+ // Hono's streamSSE builds the response itself, so what the client is to see has to be
+ // staged on the context before it is called rather than passed to a constructor.
+ for (const [name, value] of facts['response.http.headers']) c.header(name, value);
+ c.status(status);
+ // The dump is closed the same way on both paths. Hono builds the streaming response
+ // itself, so what is finalized is what it returned rather than one constructed here.
+ return finalizeGatewayResponse(prologue.gateway, streamSSE(c, async stream => {
+ try {
+ await writeSSEFrames(stream, answer.frames, {
+ keepAlive: { frame: sseCommentFrame('keepalive') },
+ ...(prologue.gateway.downstreamAbortController === undefined
+ ? {}
+ : { downstreamAbortController: prologue.gateway.downstreamAbortController }),
+ });
+ } catch (error) {
+ prologue.gateway.dump?.failed(error);
+ throw error;
+ } finally {
+ // Reading the frames to the client *is* releasing the body they came from, so the
+ // drain waits for that to finish. Draining alongside it would take frames out of the
+ // client's own stream — one connection has one reader. A client that stopped reading
+ // still gets here, which is what leaves nothing open behind it.
+ await drain();
+ }
+ }));
+ }
+
+ // Nothing is left to read either: what the client is sent was serialized from facts the run
+ // already held, so releasing can start at once.
+ prologue.services.background(drain());
+ const headers = new Headers(facts['response.http.headers'].map(([name, value]): [string, string] => [name, value]));
+ // An upstream that declared no media type is answered without one, rather than having one
+ // invented for bytes nobody described.
+ if (answer.contentType !== null) headers.set('content-type', answer.contentType);
+ else headers.delete('content-type');
+ return finalizeGatewayResponse(prologue.gateway, new Response(answer.body, { status, headers }));
+};
diff --git a/packages/gateway/src/data-plane/pipeline/services.ts b/packages/gateway/src/data-plane/pipeline/services.ts
new file mode 100644
index 0000000000..55eab63a43
--- /dev/null
+++ b/packages/gateway/src/data-plane/pipeline/services.ts
@@ -0,0 +1,41 @@
+// What a run is given beside its facts. Services are wiring: they are fixed for the run at
+// the prologue and never change on a handoff, because what a stage hands to `next` is the
+// next segment's facts — if it also supplied capabilities, the same pipeline value would
+// run with different capabilities depending on who called it.
+//
+// Facts carry content and explicitly tagged run resources. Service handles bind application
+// behavior and stay outside those values.
+
+import type { AttemptSelector } from './facts.ts';
+import type { RunDump } from '../../dump/run-sink.ts';
+import type { GatewayCtx } from '../shared/gateway-ctx.ts';
+import type { HttpServices } from '@floway-dev/http/pipeline';
+import type { Event, Logger } from '@floway-dev/pipeline';
+import type { ProtocolFrame } from '@floway-dev/protocols/common';
+import type { ModelCandidate } from '@floway-dev/provider';
+
+export type RunGatewayCtx = Omit & { readonly dump: RunDump | null };
+
+export interface GatewayServices extends HttpServices {
+ readonly recordProtocolFrames: >(frames: AsyncIterable) => AsyncIterable;
+ /** The global sink. Every stage's lines reach it, tagged with the stage's name. */
+ readonly log?: Logger;
+ /** Present only when this request is being dumped, which is what keeps recording
+ * conditional: with no sink resolved here, the runner does none of it. */
+ readonly dump?: (event: Event) => void | Promise;
+
+ /** The request-scoped context the settlement and telemetry stages read. It is a service
+ * and not a fact because it holds live handles — the scheduler, the abort signal. */
+ readonly gateway: RunGatewayCtx;
+ /** Turns a selector back into the thing that dials. The resolver is the service and the
+ * selector is the fact: a per-upstream transport is not a fact and is not pinned at the
+ * prologue either, so what is injected is the thing that resolves one and what travels
+ * is the identifier it resolves from. */
+ readonly resolveAttempt: (selector: AttemptSelector) => ModelCandidate;
+ /** The other half of the resolver: the stage that enumerates hands the live candidates
+ * here, so the ones that travel can be selectors. Per run, because a candidate list is. */
+ readonly rememberCandidates: (candidates: readonly ModelCandidate[]) => readonly AttemptSelector[];
+ /** Binds a promise to the request's lifetime: `waitUntil` on Workers, the event loop on
+ * Node. What the drain is handed to, so an answer is not held up by it. */
+ readonly background: (work: Promise) => void;
+}
diff --git a/packages/gateway/src/data-plane/pipeline/settlement.ts b/packages/gateway/src/data-plane/pipeline/settlement.ts
new file mode 100644
index 0000000000..5578d198b9
--- /dev/null
+++ b/packages/gateway/src/data-plane/pipeline/settlement.ts
@@ -0,0 +1,112 @@
+import type { BillableEntity, GatewayFacts, AttemptSelector } from './facts.ts';
+import { providerCalls } from './provider-usage.ts';
+import type { StreamOutcome } from './serve.ts';
+import type { GatewayServices } from './services.ts';
+import type { TokenUsage, UsageQuantities } from '../../repo/types.ts';
+import { recordPerformance } from '../shared/telemetry/performance.ts';
+import { recordUsage } from '../shared/telemetry/usage.ts';
+import { defineStage, getFailureFacts, move, type Logger, type Deferred, type Facts } from '@floway-dev/pipeline';
+import type { BillingMetric } from '@floway-dev/protocols/common';
+
+type Slice = { [P in K]: GatewayFacts[P] };
+
+const dumpUsage = (quantities: UsageQuantities): TokenUsage => {
+ const count = (metric: BillingMetric): number | undefined =>
+ quantities[metric] === undefined ? undefined : Number(quantities[metric]);
+ return {
+ ...(count('input_tokens') === undefined ? {} : { input: count('input_tokens') }),
+ ...(count('input_cache_read_tokens') === undefined ? {} : { input_cache_read: count('input_cache_read_tokens') }),
+ ...(count('input_cache_write_tokens') === undefined ? {} : { input_cache_write: count('input_cache_write_tokens') }),
+ ...(count('input_cache_write_1h_tokens') === undefined ? {} : { input_cache_write_1h: count('input_cache_write_1h_tokens') }),
+ ...(count('input_image_tokens') === undefined ? {} : { input_image: count('input_image_tokens') }),
+ ...(count('output_tokens') === undefined ? {} : { output: count('output_tokens') }),
+ ...(count('output_image_tokens') === undefined ? {} : { output_image: count('output_image_tokens') }),
+ };
+};
+
+/** Usage writes are bound to the request lifetime without delaying an upstream response. */
+export const settleBillable = (
+ services: Pick & { readonly log: Logger },
+ billable: readonly BillableEntity[],
+ failed: boolean,
+ finishedAt: number = performance.now(),
+): void => {
+ for (const entity of billable) {
+ services.gateway.dump?.success(entity.identity, dumpUsage(entity.quantities));
+ services.background(recordUsage(
+ services.gateway.apiKeyId,
+ entity.identity,
+ entity.quantities,
+ entity.pricingFacts ?? {},
+ ));
+ }
+ if (failed) services.gateway.dump?.failed('The upstream response failed.', { fallback: true });
+ const outputTokens = billable.reduce((total, entity) => total + Number(entity.quantities.output_tokens ?? 0), 0);
+ recordPerformance(services.gateway, services.gateway.attempt.telemetry, failed, outputTokens, finishedAt);
+};
+
+/** Settles a run once at its exit, including calls completed before a later stage throws. */
+export const writeSettlement = (
+ failed: (handedUp: Record) => boolean,
+ pendingUsage?: string,
+ recoverCalls = true,
+) => defineStage<
+ Record,
+ Slice<'serve.usage.prior'>,
+ Slice<'response.usage.billable' | 'serve.usage.prior'>,
+ Slice<'response.usage.billable'>,
+ GatewayServices
+>({
+ name: 'writeSettlement',
+ through: {
+ request: { needs: [], consumes: [], provides: ['serve.usage.prior'] },
+ response: { needs: ['response.usage.billable'], consumes: ['serve.usage.prior'], provides: [] },
+ },
+ execute: async (facts, next, use) => {
+ const entry = { ...facts, 'serve.usage.prior': move([]) };
+ let back: Slice<'response.usage.billable' | 'serve.usage.prior'>;
+ try {
+ back = await next(entry);
+ } catch (error) {
+ const atFailure = getFailureFacts(error) ?? entry;
+ settleExit(use, atFailure, true, pendingUsage, true, recoverCalls);
+ throw error;
+ }
+ settleExit(use, back, failed(back as Facts), pendingUsage, false, recoverCalls);
+ const { 'serve.usage.prior': _prior, ...rest } = back as Slice<'response.usage.billable' | 'serve.usage.prior'>;
+ return rest;
+ },
+});
+
+const settleExit = (
+ use: GatewayServices & { readonly log: Logger },
+ facts: Facts,
+ failed: boolean,
+ pendingUsage: string | undefined,
+ exceptional: boolean,
+ recoverCalls: boolean,
+): void => {
+ const prior = exceptional ? facts['serve.usage.prior'] as readonly BillableEntity[] : [];
+ const pending = pendingUsage === undefined || !(pendingUsage in facts)
+ ? null : facts[pendingUsage] as Deferred | null;
+ if (pending !== null) {
+ use.background(pending.then(outcome => {
+ settleBillable(use, [...prior, ...outcome.billable], failed || outcome.failed);
+ }, error => {
+ use.gateway.dump?.failed(error);
+ settleBillable(use, [...prior, ...facts['response.usage.billable'] as readonly BillableEntity[]], true);
+ throw error;
+ }));
+ return;
+ }
+ if ('response.usage.billable' in facts) {
+ settleBillable(use, [...prior, ...facts['response.usage.billable'] as readonly BillableEntity[]], failed);
+ return;
+ }
+ if (!recoverCalls || !('response.provider.called' in facts)) {
+ settleBillable(use, prior, failed);
+ return;
+ }
+ const candidate = use.resolveAttempt(facts['route.attempt'] as AttemptSelector);
+ settleBillable(use, [...prior, ...providerCalls(candidate, facts)], failed);
+};
diff --git a/packages/gateway/src/data-plane/pipeline/stage-contracts.ts b/packages/gateway/src/data-plane/pipeline/stage-contracts.ts
new file mode 100644
index 0000000000..74e6a1923c
--- /dev/null
+++ b/packages/gateway/src/data-plane/pipeline/stage-contracts.ts
@@ -0,0 +1,3 @@
+import type { GatewayFacts } from './facts.ts';
+
+export type Slice = { [P in K]: GatewayFacts[P] };
diff --git a/packages/gateway/src/data-plane/pipeline/upstream-body.ts b/packages/gateway/src/data-plane/pipeline/upstream-body.ts
new file mode 100644
index 0000000000..2e12edb8df
--- /dev/null
+++ b/packages/gateway/src/data-plane/pipeline/upstream-body.ts
@@ -0,0 +1,66 @@
+// Reading what an upstream answered with, for the families whose protocol is JSON.
+//
+// Every protocol the gateway carries is one it fully understands, so a body it cannot read
+// is not handed on unread: the family synthesizes an error from what it did see. That is
+// also why the text is kept alongside the parse — a dump reader is owed what actually came
+// back, and a message that quotes the body is the only thing a client can act on when the
+// upstream answered with something the protocol does not admit.
+
+import type { Failure } from './facts.ts';
+import { setRelease, type Owned } from '@floway-dev/pipeline';
+
+export interface UpstreamBody {
+ readonly text: string;
+ /** Absent when the body was not JSON at all, which is itself the answer to a protocol that
+ * requires JSON. */
+ readonly json?: unknown;
+}
+
+/** `parse` is the protocol's own JSON reader, for a protocol whose numbers must keep their
+ * source tokens through a round trip. */
+export const readUpstreamBody = async (response: Response, parse: (text: string) => unknown = JSON.parse): Promise => {
+ const text = await response.text();
+ try {
+ return { text, json: parse(text) };
+ } catch {
+ return { text };
+ }
+};
+
+/**
+ * What an upstream that answered in something other than this protocol is worth to a client.
+ *
+ * A gateway that cannot read the answer has not served the request, whatever status came with
+ * it, so this is 502 rather than the upstream's own — and it is a value, because the fork
+ * above has another candidate to try and an answer nobody can read is exactly the outcome it
+ * exists to move past.
+ */
+export const unreadableBody = (response: Response, body: UpstreamBody, protocolName: string): Failure => ({
+ status: 502,
+ message: `The upstream answered ${response.status} with a body ${protocolName} cannot read: ${body.text.slice(0, 200)}`,
+ body: 'json' in body ? body.json : body.text,
+});
+
+/**
+ * What the platform raised while dialling, as a value.
+ *
+ * Nothing in the domain throws. A connection that was refused, timed out or was reset is an
+ * outcome failover has to be able to see — a run that ends there has more candidates to try —
+ * so the ending catches whatever the platform raised and hands it up like any other failure.
+ * The status says what happened rather than what the error was: the gateway reached no
+ * upstream at all, which is what 502 states.
+ */
+export const dialFailure = (error: unknown): Failure => ({
+ status: 502,
+ message: error instanceof Error ? error.message : String(error),
+});
+
+export const spentBody = (body: (ReadableStream & Owned) | null): typeof body => {
+ if (body !== null) setRelease(body, async () => {});
+ return body;
+};
+
+/** Consumer return closes its projection; the owned body still finishes this reader at drain. */
+export const retainReader = (reader: AsyncGenerator): AsyncIterable => ({
+ [Symbol.asyncIterator]: () => ({ next: () => reader.next() }),
+});
diff --git a/packages/gateway/src/data-plane/rerank/attempt.ts b/packages/gateway/src/data-plane/rerank/attempt.ts
deleted file mode 100644
index 1842a852dc..0000000000
--- a/packages/gateway/src/data-plane/rerank/attempt.ts
+++ /dev/null
@@ -1,42 +0,0 @@
-import type { Context } from 'hono';
-
-import type { GatewayCtx } from '../shared/gateway-ctx.ts';
-import { inboundHeaders } from '../shared/inbound-headers.ts';
-import { telemetryModelIdentity, upstreamPerformanceContext } from '../shared/telemetry/attribution.ts';
-import { buildUpstreamCallOptions } from '../shared/upstream-call-options.ts';
-import type { RerankTarget } from '@floway-dev/protocols/common';
-import type { CanonicalRerankRequest } from '@floway-dev/protocols/rerank';
-import { providerModelOf } from '@floway-dev/provider';
-import type { ModelCandidate, PerformanceTelemetryContext, ProviderRerankCallResult, TelemetryModelIdentity } from '@floway-dev/provider';
-
-export interface RerankAttemptResult {
- readonly type: 'plain';
- readonly status: number;
- readonly response: Response;
- readonly target: RerankTarget;
- readonly performance: PerformanceTelemetryContext;
- readonly identity: TelemetryModelIdentity;
-}
-
-export const rerankAttempt = async (
- c: Context,
- ctx: GatewayCtx,
- candidate: ModelCandidate,
- request: CanonicalRerankRequest,
-): Promise => {
- const model = providerModelOf(candidate);
- const result: ProviderRerankCallResult = await candidate.provider.instance.callRerank(
- model,
- request,
- ctx.abortSignal,
- buildUpstreamCallOptions(candidate, ctx, inboundHeaders(c)),
- );
- return {
- type: 'plain',
- status: result.response.status,
- response: result.response,
- target: result.target,
- performance: upstreamPerformanceContext(ctx, candidate, 'rerank'),
- identity: telemetryModelIdentity(candidate, result.modelKey),
- };
-};
diff --git a/packages/gateway/src/data-plane/rerank/call-upstream.ts b/packages/gateway/src/data-plane/rerank/call-upstream.ts
new file mode 100644
index 0000000000..de9a1c3933
--- /dev/null
+++ b/packages/gateway/src/data-plane/rerank/call-upstream.ts
@@ -0,0 +1,158 @@
+import type { Fields } from './facts.ts';
+import type { UsageQuantities } from '../../repo/types.ts';
+import type { BillableEntity } from '../pipeline/facts.ts';
+import { providerEntry } from '../pipeline/provider-entry.ts';
+import { providerUsage } from '../pipeline/provider-usage.ts';
+import type { GatewayServices } from '../pipeline/services.ts';
+import { dialFailure, readUpstreamBody, unreadableBody, spentBody } from '../pipeline/upstream-body.ts';
+import { upstreamPerformanceContext, telemetryModelIdentity } from '../shared/telemetry/attribution.ts';
+import { exchangeResponse } from '@floway-dev/http/pipeline';
+import { defineStage, move } from '@floway-dev/pipeline';
+import { upstreamErrorMessage, parseDecimalString } from '@floway-dev/protocols/common';
+import { parseRerankUsage, parseRerankResponse, type CanonicalRerankResponse } from '@floway-dev/protocols/rerank';
+import { providerModelOf } from '@floway-dev/provider';
+import type { ProviderRequest, ProviderOperationPayloads, ProviderRerankResponse } from '@floway-dev/provider';
+
+/**
+ * The ending. It dials, reads the upstream's body, and provides the canonical answer and
+ * what the call is billable for. A failure is a value: a 429 here is what an earlier stage
+ * fails over, and even a 400 can be, because the next candidate's path and flags may differ.
+ */
+export const callRerankUpstream = defineStage<
+ Fields<'request.rerank.canonical' | 'route.attempt' | 'ingress.http.headers' | 'ingress.rerank.sourceProtocol'>,
+ ProviderRequest,
+ ProviderRerankResponse,
+ Fields<'response.rerank.canonical' | 'response.rerank.targetProtocol' | 'response.http.status' | 'response.http.headers' | 'response.http.body' | 'response.usage.billable'>,
+ GatewayServices
+>({
+ name: 'callRerankUpstream',
+ into: {
+ request: { needs: ['request.rerank.canonical', 'route.attempt', 'ingress.http.headers', 'ingress.rerank.sourceProtocol'], consumes: [], provides: ['request.provider.model', 'request.provider.payload', 'request.http.callId', 'request.http.headers'] },
+ response: { needs: ['response.http.exchange', 'response.provider.modelKey', 'response.provider.called', 'response.provider.previousCalls', 'response.http.body', 'response.provider.rerankTarget'], consumes: ['response.http.exchange', 'response.provider.modelKey', 'response.provider.called', 'response.provider.previousCalls', 'response.provider.previousCalls', 'response.provider.rerankTarget'], provides: ['response.rerank.canonical', 'response.rerank.targetProtocol', 'response.http.status', 'response.http.headers', 'response.http.body', 'response.usage.billable'] },
+ },
+ execute: async (facts, next, use) => {
+ const candidate = use.resolveAttempt(facts['route.attempt']);
+ const request = facts['request.rerank.canonical'];
+ const model = providerModelOf(candidate);
+ // Configuration refuses a rerank model without a target, so this holds for every candidate
+ // the narrowing kept; saying so here is what lets a dial that never answered still name
+ // the protocol it spoke.
+ const configuredTarget = model.rerankTarget;
+ if (configuredTarget === undefined) {
+ throw new Error(`${candidate.provider.upstreamId} serves rerank for ${candidate.model.id} without a target protocol`);
+ }
+ // Attribution is set before the dial, so an attempt that never completes still names the
+ // candidate it was made against rather than the one tried before it.
+ use.gateway.attempt.telemetry = upstreamPerformanceContext(use.gateway, candidate, 'rerank');
+
+ const pipeline = candidate.provider.pipelines.rerank;
+ if (pipeline === undefined) throw new Error(`Provider ${candidate.provider.kind} has no rerank pipeline`);
+ const back = await next(providerEntry(facts, candidate, request), pipeline);
+ const { 'response.http.exchange': exchange, 'response.provider.modelKey': modelKey, 'response.provider.called': wasCalled, 'response.provider.previousCalls': _previousCalls, 'response.provider.rerankTarget': target, ...rest } = back;
+ if (exchange.type === 'transportFailure') {
+ return move({
+ ...rest,
+ 'response.rerank.canonical': dialFailure(exchange.error),
+ 'response.rerank.targetProtocol': configuredTarget.protocol,
+ 'response.http.status': 502,
+ 'response.http.headers': [],
+ 'response.usage.billable': providerUsage(candidate, back, []),
+ });
+ }
+ const result = { response: exchangeResponse(exchange), modelKey, target };
+
+ const identity = telemetryModelIdentity(candidate, result.modelKey);
+ // What came back, unfiltered: the edge is where a client's view of it is decided.
+ const headers = [...result.response.headers];
+ spentBody(exchange.body);
+ const body = await readUpstreamBody(result.response);
+
+ if (!result.response.ok) {
+ await use.log.warn('upstream refused', { status: result.response.status });
+ return move({
+ ...rest,
+ 'response.rerank.canonical': {
+ status: result.response.status,
+ message: upstreamErrorMessage(body.json) ?? body.text,
+ ...('json' in body ? { body: body.json } : {}),
+ },
+ 'response.rerank.targetProtocol': result.target.protocol,
+ 'response.http.status': result.response.status,
+ 'response.http.headers': headers,
+ // The upstream was called and reported nothing, which is a different situation
+ // from reporting zero — so the entity is present with no quantities.
+ 'response.usage.billable': providerUsage(candidate, back, wasCalled ? [{ identity, quantities: {} }] : []),
+ });
+ }
+
+ if (!('json' in body)) {
+ return move({
+ ...rest,
+ 'response.rerank.canonical': unreadableBody(result.response, body, 'the rerank protocol'),
+ 'response.rerank.targetProtocol': result.target.protocol,
+ 'response.http.status': result.response.status,
+ 'response.http.headers': headers,
+ 'response.usage.billable': providerUsage(candidate, back, wasCalled ? [{ identity, quantities: {} }] : []),
+ });
+ }
+
+ // A usage block the reader cannot make sense of is a report we cannot parse, which from
+ // here is no report. It is read before the results, so an answer this gateway could not
+ // model still bills for what the upstream did meter — the two readings are independent
+ // and one of them failing is not a reason to discard the other.
+ let usage: Pick;
+ try {
+ usage = parseRerankUsage(result.target.protocol, body.json);
+ } catch (error) {
+ await use.log.warn('upstream reported usage the rerank protocol cannot read', { error: String(error) });
+ usage = {};
+ }
+ const metered: readonly BillableEntity[] = [{
+ identity,
+ quantities: billed(usage),
+ // A rerank rate can depend on how large the input was and not only on how much of it
+ // there was, so the token total is a pricing input as well as a quantity.
+ ...(usage.totalTokens === undefined ? {} : { pricingFacts: { inputTokens: usage.totalTokens } }),
+ }];
+
+ // An answer the client's own protocol produced is what that client already reads, so the
+ // edge renders it back out unchanged and nothing here has to model it. Reading the results
+ // is what a *translation* needs, and only a cross-protocol run does one — so a result item
+ // this gateway cannot model is a failure there and a field it simply carries here.
+ const translating = facts['ingress.rerank.sourceProtocol'] !== result.target.protocol;
+ let canonical: CanonicalRerankResponse;
+ try {
+ canonical = parseRerankResponse(result.target.protocol, body.json);
+ } catch (error) {
+ if (translating) {
+ await use.log.warn('upstream answered with results the rerank protocol cannot read', { error: String(error) });
+ return move({
+ ...rest,
+ 'response.rerank.canonical': unreadableBody(result.response, body, 'the rerank protocol'),
+ 'response.rerank.targetProtocol': result.target.protocol,
+ 'response.http.status': result.response.status,
+ 'response.http.headers': headers,
+ 'response.usage.billable': providerUsage(candidate, back, wasCalled ? metered : []),
+ });
+ }
+ await use.log.debug('same-protocol answer carries results this gateway does not model', { error: String(error) });
+ canonical = { raw: body.json as Record, results: [] };
+ }
+
+ return move({
+ ...rest,
+ 'response.rerank.canonical': canonical,
+ 'response.rerank.targetProtocol': result.target.protocol,
+ 'response.http.status': result.response.status,
+ 'response.http.headers': headers,
+ 'response.usage.billable': providerUsage(candidate, back, wasCalled ? metered : []),
+ });
+ },
+});
+
+const billed = (usage: Pick | undefined): UsageQuantities => {
+ const quantities: UsageQuantities = {};
+ if (usage?.searchUnits !== undefined) quantities.rerank_searches = parseDecimalString(String(usage.searchUnits));
+ if (usage?.totalTokens !== undefined) quantities.input_tokens = parseDecimalString(String(usage.totalTokens));
+ return quantities;
+};
diff --git a/packages/gateway/src/data-plane/rerank/emit.ts b/packages/gateway/src/data-plane/rerank/emit.ts
new file mode 100644
index 0000000000..0a1a60029b
--- /dev/null
+++ b/packages/gateway/src/data-plane/rerank/emit.ts
@@ -0,0 +1,78 @@
+import type { Fields } from './facts.ts';
+import { isFailure } from '../pipeline/facts.ts';
+import { isForwardableUpstreamHeader } from '../shared/upstream-response.ts';
+import { defineStage, move } from '@floway-dev/pipeline';
+import { renderErrorEnvelope } from '@floway-dev/protocols/common';
+import { renderRerankResponse } from '@floway-dev/protocols/rerank';
+
+/**
+ * The outermost edge. Renders the canonical answer into the protocol the client spoke —
+ * which is why `ingress.rerank.sourceProtocol` is an ingress fact and not a request one:
+ * it survives the switch to whatever protocol the upstream turned out to speak.
+ */
+export const emitRerank = defineStage<
+ Fields<'ingress.rerank.sourceProtocol' | 'request.rerank.canonical'>,
+ Fields<'ingress.rerank.sourceProtocol' | 'request.rerank.canonical'>,
+ Fields<'ingress.rerank.sourceProtocol' | 'request.rerank.canonical' | 'response.rerank.canonical' | 'response.rerank.targetProtocol' | 'response.http.status' | 'response.http.headers'>,
+ Fields<'response.rerank.rendered' | 'response.http.status' | 'response.http.headers'>
+>({
+ name: 'emitRerank',
+ through: {
+ request: {
+ needs: ['ingress.rerank.sourceProtocol', 'request.rerank.canonical'],
+ consumes: [],
+ provides: [],
+ },
+ response: {
+ needs: ['response.rerank.canonical', 'response.http.headers', 'response.http.status'],
+ consumes: ['response.rerank.canonical', 'response.rerank.targetProtocol', 'response.http.headers'],
+ provides: ['response.rerank.rendered', 'response.http.status', 'response.http.headers'],
+ },
+ },
+ execute: async (facts, next) => {
+ const back = await next(facts);
+ const { 'response.rerank.canonical': answer, 'response.rerank.targetProtocol': target, 'response.http.headers': headers, ...rest } = back;
+ // Vendor traces and quota state stay visible; what an intermediary must strip, and what
+ // would misdescribe a body this gateway serialized itself, does not. A filter that removed
+ // nothing hands the same array on, so the record shows no change where none happened.
+ const forwardable = headers.filter(([name]) => isForwardableUpstreamHeader(name));
+ const forClient = forwardable.length === headers.length ? headers : move(forwardable);
+ if (isFailure(answer)) {
+ return {
+ ...rest,
+ 'response.http.headers': forClient,
+ 'response.rerank.rendered': move(renderErrorEnvelope(answer.message, answer.body)),
+ // The upstream's own status, or the gateway's own when it refused before dialling.
+ // A client is not owed the upstream's exact bytes; it is owed the truth about what
+ // happened, and a 429 arriving as a 200 is not that.
+ 'response.http.status': answer.status,
+ };
+ }
+ // Translating can fail on an answer that parsed: a result may index a document the
+ // request never sent. The upstream answered and the gateway cannot put that answer in the
+ // protocol the client speaks, which is the gateway failing to serve rather than anything
+ // the client can fix — so it is 502, and a value like every other failure here.
+ let rendered: Record