fix(cost): make paid accounting durable

docs/distributed-driver.md

@@ -135,7 +135,7 @@ round-robin, region-affinity from a previous run, scheduling table).
 | **Auth** | Bearer token on `Authorization`; pluggable via `auth: string \| () => string \| Promise<string>` for rotation/refresh. |
 | **Payload size** | Server enforces `maxBodyBytes` (default 10 MB). |
 | **Traces** | Both ends emit OTel — if both point at the same OTLP collector, you get a unified trace per cell. See `docs/adapters-observability.md`. |
-| **Cost** | Worker's `ctx.cost.observe(usd, source)` is local to the worker process. Roll up server-side and attach to your worker-side telemetry; we don't (yet) forward cost back to the coordinator. Tracked as follow-up. |
+| **Cost** | Worker's `ctx.cost.runPaidCall(...)` writes durable receipts in the worker process. Roll up those receipts server-side and attach them to worker telemetry; they are not forwarded to the coordinator automatically. |
 
 ## Running the reference example
 

examples/_shared/extraction-task.ts

@@ -15,7 +15,13 @@
 
 import { createHash } from 'node:crypto'
 import type { DispatchContext, JudgeConfig, JudgeScore, Scenario } from '../../src/campaign'
-import { callLlm, type LlmClientOptions } from '../../src/llm-client'
+import {
+  callLlm,
+  costReceiptFromLlmError,
+  type LlmCallRequest,
+  type LlmClientOptions,
+  maximumChargeForLlmRequest,
+} from '../../src/llm-client'
 import type { RunRecord } from '../../src/run-record'
 
 export interface ExtractScenario extends Scenario {
@@ -242,28 +248,37 @@ export function makeExtractionWorker(opts: ExtractionWorkerOptions) {
     scenario: ExtractScenario,
     ctx: DispatchContext,
   ): Promise<Artifact> {
-    const res = await callLlm(
-      {
-        model: opts.model,
-        messages: [
-          { role: 'system' as const, content: surface },
-          { role: 'user' as const, content: scenario.text },
-        ],
-        jsonMode: true,
-        temperature: 0,
-        maxTokens: 400,
-        timeoutMs,
-      },
-      opts.llm,
-    )
-    const costUsd =
-      res.costUsd ??
-      (res.usage.promptTokens / 1_000_000) * opts.priceInPerMTokens +
-        (res.usage.completionTokens / 1_000_000) * opts.priceOutPerMTokens
-    // Report BOTH so the cell carries real cost AND real tokens — the
-    // backend-integrity guard keys on tokens, the budget gate on cost.
-    ctx.cost.observe(costUsd, 'worker')
-    ctx.cost.observeTokens({ input: res.usage.promptTokens, output: res.usage.completionTokens })
+    const request: LlmCallRequest = {
+      model: opts.model,
+      messages: [
+        { role: 'system', content: surface },
+        { role: 'user', content: scenario.text },
+      ],
+      jsonMode: true,
+      temperature: 0,
+      maxTokens: 400,
+      timeoutMs,
+    }
+    const paid = await ctx.cost.runPaidCall({
+      actor: 'worker',
+      model: opts.model,
+      maximumCharge: maximumChargeForLlmRequest(request, opts.llm),
+      execute: (signal, callId) =>
+        callLlm(request, { ...opts.llm, signal, idempotencyKey: callId }),
+      receipt: (res) => ({
+        model: res.model || opts.model,
+        inputTokens: res.usage.promptTokens,
+        outputTokens: res.usage.completionTokens,
+        actualCostUsd:
+          res.costUsd ??
+          (res.usage.promptTokens / 1_000_000) * opts.priceInPerMTokens +
+            (res.usage.completionTokens / 1_000_000) * opts.priceOutPerMTokens,
+      }),
+      receiptFromError: costReceiptFromLlmError,
+    })
+    if (!paid.succeeded) throw paid.error
+    const res = paid.value
+    const costUsd = paid.receipt.costUsd
     opts.records.push({
       runId: `${scenario.id}-${createHash('sha1').update(surface).digest('hex').slice(0, 8)}-${opts.records.length}`,
       experimentId,

examples/benchmarks/README.md

@@ -31,15 +31,23 @@ assignSplit(itemId: string): 'search' | 'dev' | 'holdout'
 Use `runBenchmarkAdapter` when you want a campaign-backed run with resumability, traces, cost, latency, split reporting, and persisted report artifacts.
 
 ```ts
+import { maximumChargeForLlmRequest } from '@tangle-network/agent-eval'
 import { runBenchmarkAdapter, routing } from '@tangle-network/agent-eval/benchmarks'
 
 const result = await runBenchmarkAdapter({
   adapter: new routing.RoutingAdapter(),
   runDir: 'routing-smoke',
   respond: async ({ item, context }) => {
-    const route = await callRouter(item.payload.prompt)
-    context.cost.observe(0.001, 'router')
-    return route
+    const request = buildRouterRequest(item.payload.prompt, { maxTokens: 256 })
+    const paid = await context.cost.runPaidCall({
+      actor: 'routing-worker',
+      model: request.model,
+      maximumCharge: maximumChargeForLlmRequest(request),
+      execute: (signal) => callRouter(request, signal),
+      receipt: (route) => route.receipt,
+    })
+    if (!paid.succeeded) throw paid.error
+    return paid.value
   },
 })
 

examples/benchmarks/appworld/run-bench.ts

@@ -149,70 +149,74 @@ async function dispatchWithSurface(
   const promptFile = join(dir, 'surface.txt')
   writeFileSync(promptFile, surface)
   const experiment = `bench_${scenario.taskId}_${n}` // unique → no AppWorld output-dir collision
-  const { stdout } = await execFileAsync(
-    PYTHON,
-    [
-      WORKER,
-      '--task-id',
-      scenario.taskId,
-      '--model',
-      MODEL,
-      '--system-prompt-file',
-      promptFile,
-      '--experiment-name',
-      experiment,
-      '--max-steps',
-      String(MAX_STEPS), // 0 = no cap (maxTurns=0): run until complete_task
-      '--max-wall-seconds',
-      String(MAX_WALL),
-      // Genuine independent shots: temperature>0 + a UNIQUE seed per dispatch so
-      // the router can't return a cached completion for an identical prompt
-      // (temp=0 + same seed collapses the 5 reps into 1 via cache).
-      '--temperature',
-      String(TEMPERATURE),
-      '--seed',
-      String(SEED + n),
-      '--call-timeout',
-      String(CALL_TIMEOUT),
-      '--max-tokens',
-      String(MAX_TOKENS),
-      '--rate-limit-budget',
-      '240',
-      '--out-dir',
-      dir,
-    ],
-    {
-      cwd: APPWORLD_DIR,
-      env: { ...process.env, OPENAI_BASE_URL: BASE_URL, OPENAI_API_KEY: API_KEY },
-      maxBuffer: 64 * 1024 * 1024,
-      signal: ctx.signal,
+  const paid = await ctx.cost.runPaidCall({
+    actor: 'appworld-worker',
+    model: MODEL,
+    execute: async (signal) => {
+      await execFileAsync(
+        PYTHON,
+        [
+          WORKER,
+          '--task-id',
+          scenario.taskId,
+          '--model',
+          MODEL,
+          '--system-prompt-file',
+          promptFile,
+          '--experiment-name',
+          experiment,
+          '--max-steps',
+          String(MAX_STEPS),
+          '--max-wall-seconds',
+          String(MAX_WALL),
+          '--temperature',
+          String(TEMPERATURE),
+          '--seed',
+          String(SEED + n),
+          '--call-timeout',
+          String(CALL_TIMEOUT),
+          '--max-tokens',
+          String(MAX_TOKENS),
+          '--rate-limit-budget',
+          '240',
+          '--out-dir',
+          dir,
+        ],
+        {
+          cwd: APPWORLD_DIR,
+          env: { ...process.env, OPENAI_BASE_URL: BASE_URL, OPENAI_API_KEY: API_KEY },
+          maxBuffer: 64 * 1024 * 1024,
+          signal,
+        },
+      )
+      return JSON.parse(readFileSync(join(dir, 'result.json'), 'utf8')) as {
+        tgc: number
+        sgc: number
+        completed: boolean
+        cost_usd: number
+        token_usage?: { input?: number; output?: number }
+        tokenUsage?: { input?: number; output?: number }
+        traces_path?: string
+      }
     },
-  )
-  // The worker prints a compact verdict line; the full record is result.json.
-  const result = JSON.parse(readFileSync(join(dir, 'result.json'), 'utf8')) as {
-    tgc: number
-    sgc: number
-    completed: boolean
-    cost_usd: number
-    token_usage?: { input?: number; output?: number }
-    tokenUsage?: { input?: number; output?: number }
-    traces_path?: string
-  }
+    receipt: (result) => {
+      if (Number.isNaN(result.cost_usd)) {
+        throw new Error(
+          `appworld bench: worker returned NaN cost for model "${MODEL}" — it is unpriced. Add it to PRICE_PER_M in repl_agent.py so the lift comparison has a real cost axis.`,
+        )
+      }
+      return {
+        model: MODEL,
+        inputTokens: result.token_usage?.input ?? result.tokenUsage?.input ?? 0,
+        outputTokens: result.token_usage?.output ?? result.tokenUsage?.output ?? 0,
+        actualCostUsd: result.cost_usd,
+      }
+    },
+  })
+  if (!paid.succeeded) throw paid.error
+  const result = paid.value
   const inTok = result.token_usage?.input ?? result.tokenUsage?.input ?? 0
   const outTok = result.token_usage?.output ?? result.tokenUsage?.output ?? 0
-  // Feed the cost meter so integrity:'assert' is satisfied (no silent stub).
-  ctx.cost.observeTokens({ input: inTok, output: outTok })
-  // The worker emits NaN cost for an UNPRICED model (price() returns NaN by
-  // design — no fabricated zero). Swallowing it here would silently drop the
-  // cell under integrity:'assert' with a misleading "cell errored". Fail loud
-  // with the actionable cause instead.
-  if (Number.isNaN(result.cost_usd)) {
-    throw new Error(
-      `appworld bench: worker returned NaN cost for model "${MODEL}" — it is unpriced. Add it to PRICE_PER_M in repl_agent.py so the lift comparison has a real cost axis.`,
-    )
-  }
-  if (result.cost_usd > 0) ctx.cost.observe(result.cost_usd, 'appworld-worker')
-  void stdout
   return {
     tgc: result.tgc,
     sgc: result.sgc,

examples/benchmarks/gsm8k/compare-proposers.ts

@@ -36,11 +36,18 @@ import {
   type Scenario,
   skillOptEntry,
 } from '../../../src/campaign'
+import { CostLedger } from '../../../src/cost-ledger'
 import {
   assertRealBackend,
   summarizeBackendIntegrity,
 } from '../../../src/integrity/backend-integrity'
-import { callLlm, type LlmClientOptions } from '../../../src/llm-client'
+import {
+  callLlm,
+  costReceiptFromLlmError,
+  type LlmCallRequest,
+  type LlmClientOptions,
+  maximumChargeForLlmRequest,
+} from '../../../src/llm-client'
 import type { RunRecord } from '../../../src/run-record'
 import { evaluate, loadDataset } from './index'
 
@@ -106,25 +113,35 @@ function makeWorker() {
     scenario: GsmScenario,
     ctx: DispatchContext,
   ): Promise<Artifact> {
-    const res = await callLlm(
-      {
-        model: MODEL,
-        messages: [
-          { role: 'system' as const, content: surface },
-          { role: 'user' as const, content: scenario.question },
-        ],
-        temperature: 0,
-        maxTokens: 1024,
-        timeoutMs: CALL_TIMEOUT_MS,
-      },
-      llm,
-    )
-    const costUsd =
-      res.costUsd ??
-      (res.usage.promptTokens / 1_000_000) * PRICE_IN_PER_M +
-        (res.usage.completionTokens / 1_000_000) * PRICE_OUT_PER_M
-    ctx.cost.observe(costUsd, 'worker')
-    ctx.cost.observeTokens({ input: res.usage.promptTokens, output: res.usage.completionTokens })
+    const request: LlmCallRequest = {
+      model: MODEL,
+      messages: [
+        { role: 'system', content: surface },
+        { role: 'user', content: scenario.question },
+      ],
+      temperature: 0,
+      maxTokens: 1024,
+      timeoutMs: CALL_TIMEOUT_MS,
+    }
+    const paid = await ctx.cost.runPaidCall({
+      actor: 'worker',
+      model: MODEL,
+      maximumCharge: maximumChargeForLlmRequest(request, llm),
+      execute: (signal, callId) => callLlm(request, { ...llm, signal, idempotencyKey: callId }),
+      receipt: (res) => ({
+        model: res.model || MODEL,
+        inputTokens: res.usage.promptTokens,
+        outputTokens: res.usage.completionTokens,
+        actualCostUsd:
+          res.costUsd ??
+          (res.usage.promptTokens / 1_000_000) * PRICE_IN_PER_M +
+            (res.usage.completionTokens / 1_000_000) * PRICE_OUT_PER_M,
+      }),
+      receiptFromError: costReceiptFromLlmError,
+    })
+    if (!paid.succeeded) throw paid.error
+    const res = paid.value
+    const costUsd = paid.receipt.costUsd
     records.push({
       runId: `${scenario.id}-${createHash('sha1').update(surface).digest('hex').slice(0, 8)}-${records.length}`,
       experimentId: 'gsm8k-substrate-proof',
@@ -196,10 +213,15 @@ async function main() {
 
   // ── Baseline smoke: confirm the weak baseline leaves headroom (< 0.85). ──
   let baselineSmoke = 0
+  const smokeLedger = new CostLedger()
+  const smokeCost: DispatchContext['cost'] = {
+    runPaidCall: (input) =>
+      smokeLedger.runPaidCall({ ...input, channel: input.channel ?? 'agent', phase: 'smoke' }),
+  }
   for (const sc of holdoutScenarios) {
     const art = await worker(BASELINE_SURFACE, sc, {
       cellId: `smoke-${sc.id}`,
-      cost: { observe: () => {}, observeTokens: () => {} },
+      cost: smokeCost,
     } as unknown as DispatchContext)
     baselineSmoke += (
       await judge.score({ artifact: art, scenario: sc } as Parameters<typeof judge.score>[0])

examples/eval-fixtures-quickstart/index.ts

@@ -27,9 +27,7 @@ const evalsDir = join(here, 'evals')
 const runDir = join(tmpdir(), 'agent-eval-fixtures-quickstart')
 const dispatchRef = 'offline-fixture-agent/v1'
 
-const dispatch: DispatchFn<EvalFixtureScenario, FixtureArtifact> = async (scenario, ctx) => {
-  ctx.cost.observe(0.001, 'offline-agent')
-  ctx.cost.observeTokens({ input: scenario.prompt.length, output: 64 })
+const dispatch: DispatchFn<EvalFixtureScenario, FixtureArtifact> = async (scenario) => {
   return {
     answer: `Implemented ${scenario.fixtureName}: ${scenario.prompt.trim()}`,
     filesTouched: ['src/app.ts', 'README.md'],
@@ -77,7 +75,7 @@ async function main() {
     dispatchRef,
     judges: [judge],
     runDir,
-    expectUsage: 'assert',
+    expectUsage: 'off',
   })
 
   const after = planEvalFixtureRun<FixtureArtifact>({

examples/findings-ablation/index.ts

@@ -45,6 +45,7 @@ import {
   type RunOptimizationOptions,
   runOptimization,
 } from '../../src/campaign'
+import { CostLedger } from '../../src/cost-ledger'
 import { assertRealBackend, summarizeBackendIntegrity } from '../../src/integrity/backend-integrity'
 import type { LlmClientOptions } from '../../src/llm-client'
 import type { RunRecord } from '../../src/run-record'
@@ -99,13 +100,18 @@ const worker = makeExtractionWorker({
 
 const judge = extractionJudge([...SEARCH, ...HOLDOUT])
 const round = (n: number) => Math.round(n * 1000) / 1000
+const scoringLedger = new CostLedger()
+const scoringCost: DispatchContext['cost'] = {
+  runPaidCall: (input) =>
+    scoringLedger.runPaidCall({ ...input, channel: input.channel ?? 'agent', phase: 'holdout' }),
+}
 
 /** A noop dispatch ctx for the standalone holdout scoring pass — the worker still
  *  appends a RunRecord (real tokens) so integrity stays verdictable. */
 function scoringCtx(cellId: string): DispatchContext {
   return {
     cellId,
-    cost: { observe: () => {}, observeTokens: () => {} },
+    cost: scoringCost,
   } as unknown as DispatchContext
 }