fix: skip the billing drain wait when the gateway has no receipts

docs/collective-runtime.md

@@ -365,6 +365,12 @@ cursor. Generic OpenAI-compatible endpoints, Claude Code, Codex, OpenCode and
 Pi never become billing authorities implicitly. Missing or late receipts stay
 unknown instead of becoming zero-cost observations.
 
+Final reconciliation waits (bounded, 5 s by default) for outstanding receipts
+only when the feed response carries `x-baro-billing-receipts: authoritative`,
+which a Gateway sends once it finalizes receipts. Without that header Baro
+still ingests whatever one final feed pass returns, then stops waiting and does
+not report the unreceipted invocations as a reconciliation failure.
+
 Selection is currently deterministic and greedy after policy constraints;
 there is no uncertainty-driven route exploration yet, so configured priors
 still matter. Learning is advisory and run-local; durable, cross-run

packages/baro-orchestrator/scripts/run-architect.ts

@@ -519,7 +519,11 @@ async function main(): Promise<void> {
     } finally {
         try {
             const billingResult = await reconcileAndCloseGatewayBilling(billing)
-            if (billingResult && !billingResult.complete) {
+            if (
+                billingResult &&
+                !billingResult.complete &&
+                !billingResult.receiptsUnavailable
+            ) {
                 process.stderr.write(
                     `[run-architect] billing reconciliation incomplete (${billingResult.unresolvedInvocationIds.length} invocation(s))\n`,
                 )

packages/baro-orchestrator/scripts/run-conversation.ts

@@ -423,7 +423,7 @@ async function main(): Promise<void> {
         intake?.close()
         try {
             const result = await reconcileAndCloseGatewayBilling(billing)
-            if (result && !result.complete) {
+            if (result && !result.complete && !result.receiptsUnavailable) {
                 process.stderr.write(
                     `[run-conversation] billing reconciliation incomplete ` +
                         `(${result.unresolvedInvocationIds.length} invocation(s))\n`,

packages/baro-orchestrator/scripts/run-intake.ts

@@ -129,7 +129,7 @@ async function main(): Promise<void> {
                 })
             } finally {
                 const result = await reconcileAndCloseGatewayBilling(billing)
-                if (result && !result.complete) {
+                if (result && !result.complete && !result.receiptsUnavailable) {
                     process.stderr.write(
                         `[run-intake] billing reconciliation incomplete (${result.unresolvedInvocationIds.length} invocation(s))\n`,
                     )

packages/baro-orchestrator/scripts/run-planner.ts

@@ -352,7 +352,7 @@ async function main(): Promise<void> {
                 })
             } finally {
                 const result = await reconcileAndCloseGatewayBilling(billing)
-                if (result && !result.complete) {
+                if (result && !result.complete && !result.receiptsUnavailable) {
                     process.stderr.write(
                         `[run-planner] billing reconciliation incomplete (${result.unresolvedInvocationIds.length} invocation(s))\n`,
                     )

packages/baro-orchestrator/src/orchestrate.ts

@@ -789,7 +789,7 @@ export async function orchestrate(
         gatewayBillingReconciled = true
         try {
             const billing = await gatewayBillingCoordinator.drain()
-            if (!billing.complete) {
+            if (!billing.complete && !billing.receiptsUnavailable) {
                 process.stderr.write(
                     `[billing] ${billing.unresolvedInvocationIds.length} invocation(s) ` +
                         `remain without acknowledged authoritative telemetry` +

packages/baro-orchestrator/src/telemetry/billing/gateway-billing-coordinator.ts

@@ -67,6 +67,11 @@ export interface GatewayBillingDrainResult {
     readonly unresolvedInvocationIds: readonly string[]
     readonly complete: boolean
     readonly feedError: string | null
+    /**
+     * The gateway answered without advertising authoritative receipts, so
+     * unresolved invocations were never going to settle; not a failure.
+     */
+    readonly receiptsUnavailable: boolean
 }
 
 /**
@@ -100,6 +105,8 @@ export class GatewayBillingCoordinator {
     private readonly activePublications = new Map<string, Promise<void>>()
     private activePull: Promise<void> | null = null
     private pullRequested = false
+    private pullsStarted = 0
+    private lastSuccessfulPull = 0
     private lastFeedError: Error | null = null
     private closed = false
 
@@ -242,6 +249,11 @@ export class GatewayBillingCoordinator {
      */
     async drain(): Promise<GatewayBillingDrainResult> {
         const deadline = Date.now() + this.drainTimeoutMs
+        // Only a pass started inside drain() can vouch that nothing more is coming.
+        const firstDrainPull = this.pullsStarted + 1
+        const receiptsUnavailable = () =>
+            this.lastSuccessfulPull >= firstDrainPull &&
+            this.feed.authoritativeReceipts === false
         do {
             if (Date.now() >= deadline) {
                 this.feed.abort("billing reconciliation deadline reached")
@@ -263,8 +275,8 @@ export class GatewayBillingCoordinator {
                 break
             }
             if (
-                this.pendingInvocations.size === 0 &&
-                this.measurementOutbox.size === 0
+                this.measurementOutbox.size === 0 &&
+                (this.pendingInvocations.size === 0 || receiptsUnavailable())
             ) {
                 break
             }
@@ -286,6 +298,7 @@ export class GatewayBillingCoordinator {
                 this.pendingInvocations.size === 0 &&
                 this.measurementOutbox.size === 0,
             feedError: this.lastFeedError?.message ?? null,
+            receiptsUnavailable: receiptsUnavailable(),
         })
     }
 
@@ -298,6 +311,7 @@ export class GatewayBillingCoordinator {
     private async runRequestedPulls(): Promise<void> {
         while (this.pullRequested && !this.closed) {
             this.pullRequested = false
+            const pull = ++this.pullsStarted
             try {
                 await this.flushDetachedOutbox()
                 await this.feed.drain(async (receipt) => {
@@ -326,6 +340,7 @@ export class GatewayBillingCoordinator {
                     // Await its acknowledgement before the feed commits cursor.
                     await this.publishOutboxEntry(measurementId)
                 })
+                this.lastSuccessfulPull = pull
                 this.lastFeedError = null
             } catch (error) {
                 this.lastFeedError = safeError(error)

packages/baro-orchestrator/src/telemetry/billing/index.ts

@@ -34,6 +34,8 @@ export {
 } from "./receipt-ledger.js"
 
 export {
+    BILLING_RECEIPTS_AUTHORITATIVE,
+    BILLING_RECEIPTS_HEADER,
     BillingReceiptFeedClient,
     BillingFeedProtocolError,
     BillingFeedHttpError,

packages/baro-orchestrator/src/telemetry/billing/receipt-feed-client.ts

@@ -15,6 +15,9 @@ const DEFAULT_PAGE_SIZE = 100
 const DEFAULT_MAX_RESPONSE_BYTES = 1_048_576
 const DEFAULT_RETRY_BASE_DELAY_MS = 50
 
+export const BILLING_RECEIPTS_HEADER = "x-baro-billing-receipts"
+export const BILLING_RECEIPTS_AUTHORITATIVE = "authoritative"
+
 export interface CloudBillingFeedPage {
     readonly schemaVersion: 1
     readonly billingSessionId: string
@@ -73,6 +76,7 @@ export class BillingReceiptFeedClient {
     private readonly commitCursor?: (cursor: string) => void | Promise<void>
     private readonly fetchImpl: typeof fetch
     private currentCursor: string | null
+    private receiptsAdvertised: boolean | null = null
     private activeController: AbortController | null = null
     private closed = false
 
@@ -132,6 +136,14 @@ export class BillingReceiptFeedClient {
         return this.currentCursor
     }
 
+    /**
+     * Whether the last valid page came from a gateway that finalizes receipts;
+     * null until one page has been read.
+     */
+    get authoritativeReceipts(): boolean | null {
+        return this.receiptsAdvertised
+    }
+
     /** Read and commit exactly one page. */
     async pull(sink: BillingReceiptSink): Promise<BillingFeedPullResult> {
         return this.withOperation(async (controller) => {
@@ -302,7 +314,11 @@ export class BillingReceiptFeedClient {
                 } catch {
                     throw new BillingFeedProtocolError("billing feed returned malformed JSON")
                 }
-                return parseCloudBillingFeedPage(parsed, this.pageSize)
+                const page = parseCloudBillingFeedPage(parsed, this.pageSize)
+                this.receiptsAdvertised =
+                    response.headers.get(BILLING_RECEIPTS_HEADER)?.toLowerCase() ===
+                    BILLING_RECEIPTS_AUTHORITATIVE
+                return page
             } catch (error) {
                 if (operationSignal.aborted) throw abortReason(operationSignal)
                 if (

packages/baro-orchestrator/test/telemetry/billing/gateway-billing-coordinator.test.ts

@@ -10,6 +10,8 @@ import {
 
 import {
     BILLING_MONEY_SCALE,
+    BILLING_RECEIPTS_AUTHORITATIVE,
+    BILLING_RECEIPTS_HEADER,
     GatewayBillingCoordinator,
     parseCloudBillingReceipt,
     resolveGatewayBillingEnvironment,
@@ -189,6 +191,143 @@ describe("GatewayBillingCoordinator", () => {
         gateway.close()
     })
 
+    it("skips the drain wait when the gateway does not advertise authoritative receipts", async () => {
+        let feedCalls = 0
+        const gateway = new GatewayBillingCoordinator({
+            runId: "run-no-receipts",
+            gatewayBaseUrl: "https://gateway.example/v1",
+            apiKey: "tenant-a",
+            drainTimeoutMs: 5_000,
+            feedMaxRetries: 0,
+            publishMeasurement: () => undefined,
+            fetchImpl: async () => {
+                feedCalls += 1
+                return jsonResponse(page(gateway.billingSessionId, [], "cursor-1"))
+            },
+        })
+        const dispatch = gateway.prepareDispatch(
+            "https://gateway.example/v1",
+            "tenant-a",
+            invocationContext,
+        )
+        assert.ok(dispatch)
+        await gateway.observeRunner(dispatch, { status: "succeeded", durationMs: 1 })
+
+        const started = Date.now()
+        const result = await gateway.drain()
+        const elapsed = Date.now() - started
+
+        assert.equal(result.complete, false)
+        assert.equal(result.receiptsUnavailable, true)
+        assert.equal(result.feedError, null)
+        assert.deepEqual(result.unresolvedInvocationIds, [dispatch.record.invocationId])
+        assert.equal(gateway.feed.authoritativeReceipts, false)
+        assert.ok(feedCalls >= 1)
+        assert.ok(elapsed < 1_000, `drain took ${elapsed}ms`)
+        gateway.close()
+    })
+
+    it("still ingests receipts from a gateway that does not advertise them", async () => {
+        let receipt: CloudBillingReceipt | null = null
+        const gateway = new GatewayBillingCoordinator({
+            runId: "run-unadvertised-receipt",
+            gatewayBaseUrl: "https://gateway.example/v1",
+            apiKey: "tenant-a",
+            drainTimeoutMs: 5_000,
+            feedMaxRetries: 0,
+            publishMeasurement: () => undefined,
+            fetchImpl: async () =>
+                jsonResponse(
+                    page(gateway.billingSessionId, receipt ? [receipt] : [], "cursor-1"),
+                ),
+        })
+        const dispatch = gateway.prepareDispatch(
+            "https://gateway.example/v1",
+            "tenant-a",
+            invocationContext,
+        )
+        assert.ok(dispatch)
+        receipt = completeReceipt(dispatch.record.billingSessionId, dispatch.record.invocationId)
+
+        const result = await gateway.drain()
+
+        assert.equal(result.complete, true)
+        assert.equal(result.settledInvocations, 1)
+        gateway.close()
+    })
+
+    it("keeps waiting for receipts a gateway advertises as authoritative", async () => {
+        const gateway = new GatewayBillingCoordinator({
+            runId: "run-authoritative",
+            gatewayBaseUrl: "https://gateway.example/v1",
+            apiKey: "tenant-a",
+            drainTimeoutMs: 200,
+            feedMaxRetries: 0,
+            publishMeasurement: () => undefined,
+            fetchImpl: async () =>
+                jsonResponse(page(gateway.billingSessionId, [], "cursor-1"), {
+                    [BILLING_RECEIPTS_HEADER]: BILLING_RECEIPTS_AUTHORITATIVE,
+                }),
+        })
+        assert.ok(
+            gateway.prepareDispatch(
+                "https://gateway.example/v1",
+                "tenant-a",
+                invocationContext,
+            ),
+        )
+
+        const started = Date.now()
+        const result = await gateway.drain()
+        const elapsed = Date.now() - started
+
+        assert.equal(result.complete, false)
+        assert.equal(result.receiptsUnavailable, false)
+        assert.equal(gateway.feed.authoritativeReceipts, true)
+        assert.ok(elapsed >= 150, `drain returned after ${elapsed}ms`)
+        gateway.close()
+    })
+
+    it("does not conclude receipts are unavailable from a pull that predates drain", async () => {
+        let releaseFirst!: () => void
+        const firstReleased = new Promise<void>((resolve) => {
+            releaseFirst = resolve
+        })
+        let feedCalls = 0
+        const gateway = new GatewayBillingCoordinator({
+            runId: "run-stale-pull",
+            gatewayBaseUrl: "https://gateway.example/v1",
+            apiKey: "tenant-a",
+            drainTimeoutMs: 150,
+            feedMaxRetries: 0,
+            publishMeasurement: () => undefined,
+            fetchImpl: async () => {
+                feedCalls += 1
+                if (feedCalls > 1) {
+                    return new Response("unavailable", { status: 503 })
+                }
+                await firstReleased
+                return jsonResponse(page(gateway.billingSessionId, [], "cursor-1"))
+            },
+        })
+        const dispatch = gateway.prepareDispatch(
+            "https://gateway.example/v1",
+            "tenant-a",
+            invocationContext,
+        )
+        assert.ok(dispatch)
+        await gateway.observeRunner(dispatch, { status: "succeeded", durationMs: 1 })
+
+        const draining = gateway.drain()
+        releaseFirst()
+        const result = await draining
+
+        assert.equal(gateway.feed.authoritativeReceipts, false)
+        assert.equal(result.receiptsUnavailable, false)
+        assert.match(result.feedError ?? "", /503/)
+        gateway.close()
+    })
+
     it("uses one upstream HTTP attempt and sends correlation only to the trusted gateway", async () => {
         let providerCalls = 0
         let requestBody: Record<string, unknown> | null = null
@@ -800,10 +939,13 @@ function page(
     }
 }
 
-function jsonResponse(value: unknown): Response {
+function jsonResponse(
+    value: unknown,
+    headers: Record<string, string> = {},
+): Response {
     return new Response(JSON.stringify(value), {
         status: 200,
-        headers: { "content-type": "application/json" },
+        headers: { "content-type": "application/json", ...headers },
     })
 }