diff --git a/docs/collective-runtime.md b/docs/collective-runtime.md index b5425bd6..e330487a 100644 --- a/docs/collective-runtime.md +++ b/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 diff --git a/packages/baro-orchestrator/scripts/run-architect.ts b/packages/baro-orchestrator/scripts/run-architect.ts index d8975bca..f44ecb82 100644 --- a/packages/baro-orchestrator/scripts/run-architect.ts +++ b/packages/baro-orchestrator/scripts/run-architect.ts @@ -519,7 +519,11 @@ async function main(): Promise { } 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`, ) diff --git a/packages/baro-orchestrator/scripts/run-conversation.ts b/packages/baro-orchestrator/scripts/run-conversation.ts index 2b2f5367..00925655 100644 --- a/packages/baro-orchestrator/scripts/run-conversation.ts +++ b/packages/baro-orchestrator/scripts/run-conversation.ts @@ -423,7 +423,7 @@ async function main(): Promise { 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`, diff --git a/packages/baro-orchestrator/scripts/run-intake.ts b/packages/baro-orchestrator/scripts/run-intake.ts index ce98c918..23843d84 100644 --- a/packages/baro-orchestrator/scripts/run-intake.ts +++ b/packages/baro-orchestrator/scripts/run-intake.ts @@ -129,7 +129,7 @@ async function main(): Promise { }) } 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`, ) diff --git a/packages/baro-orchestrator/scripts/run-planner.ts b/packages/baro-orchestrator/scripts/run-planner.ts index 98c261f6..e03fabd6 100644 --- a/packages/baro-orchestrator/scripts/run-planner.ts +++ b/packages/baro-orchestrator/scripts/run-planner.ts @@ -352,7 +352,7 @@ async function main(): Promise { }) } 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`, ) diff --git a/packages/baro-orchestrator/src/orchestrate.ts b/packages/baro-orchestrator/src/orchestrate.ts index 04b3e29e..32f1808d 100644 --- a/packages/baro-orchestrator/src/orchestrate.ts +++ b/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` + diff --git a/packages/baro-orchestrator/src/telemetry/billing/gateway-billing-coordinator.ts b/packages/baro-orchestrator/src/telemetry/billing/gateway-billing-coordinator.ts index ddb798e8..6002be72 100644 --- a/packages/baro-orchestrator/src/telemetry/billing/gateway-billing-coordinator.ts +++ b/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>() private activePull: Promise | 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 { 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 { 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) diff --git a/packages/baro-orchestrator/src/telemetry/billing/index.ts b/packages/baro-orchestrator/src/telemetry/billing/index.ts index 2bd6deb3..7ead66cc 100644 --- a/packages/baro-orchestrator/src/telemetry/billing/index.ts +++ b/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, diff --git a/packages/baro-orchestrator/src/telemetry/billing/receipt-feed-client.ts b/packages/baro-orchestrator/src/telemetry/billing/receipt-feed-client.ts index 140aeefe..b7a68fb6 100644 --- a/packages/baro-orchestrator/src/telemetry/billing/receipt-feed-client.ts +++ b/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 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 { 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 ( diff --git a/packages/baro-orchestrator/test/telemetry/billing/gateway-billing-coordinator.test.ts b/packages/baro-orchestrator/test/telemetry/billing/gateway-billing-coordinator.test.ts index b26ae32c..d8561337 100644 --- a/packages/baro-orchestrator/test/telemetry/billing/gateway-billing-coordinator.test.ts +++ b/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((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 | null = null @@ -800,10 +939,13 @@ function page( } } -function jsonResponse(value: unknown): Response { +function jsonResponse( + value: unknown, + headers: Record = {}, +): Response { return new Response(JSON.stringify(value), { status: 200, - headers: { "content-type": "application/json" }, + headers: { "content-type": "application/json", ...headers }, }) }