feat(temporal): Let an operator bound what a turn and a session spend.

packages/core/src/session/runner/index.ts

@@ -33,6 +33,9 @@ export interface StepResult {
   readonly continue: boolean
   readonly step: number
   readonly promotion: SessionInput.Delivery | undefined
+  /** What this step's provider attempt was billed for, for a caller keeping a budget. Absent from a
+   * step that made no attempt, and from one whose attempt reported no counts. */
+  readonly spent?: { readonly tokens: number }
 }
 
 /** A tool call the provider asked for, recorded but not run, handed to the caller to dispatch. Every

packages/core/src/session/runner/llm.ts

@@ -45,7 +45,7 @@ import {
   Service,
 } from "./index"
 import { SessionRunnerModel } from "./model"
-import { createLLMEventPublisher, emitToolResult, record } from "./publish-llm-event"
+import { billed, createLLMEventPublisher, emitToolResult, record } from "./publish-llm-event"
 import { toLLMMessages } from "./to-llm-message"
 import { MAX_STEPS_PROMPT } from "./max-steps"
 import { DEFAULT_MAX_STEPS, REPEAT_LIMIT, REPEATED_CALLS_PROMPT, trailingIdenticalToolSteps } from "./loop-guard"
@@ -859,7 +859,12 @@ const layer = Layer.effect(
       const prologue = yield* stepPrologue(input)
       if (prologue.kind === "settled") return prologue.result
       const result = yield* runTurn(input.sessionID, prologue.promotion, input.step)
-      return yield* stepContinuation(input.sessionID, result.needsContinuation, result.step)
+      const closed = yield* stepContinuation(input.sessionID, result.needsContinuation, result.step)
+      // The attempt's own counts, carried out for a caller keeping a budget. The split path gets
+      // them from the settlement it hands back; this path publishes its own step ending, so
+      // without this a whole-step deployment has no bound on what a turn spends.
+      const spent = billed(result.settlement)
+      return spent ? { ...closed, spent } : closed
     })
 
     const runModelCall = Effect.fn("SessionRunner.runModelCall")(function* (input: StepInput) {

packages/core/src/session/runner/publish-llm-event.ts

@@ -38,6 +38,26 @@ export interface StepSettlement {
   readonly tokens: ReturnType<typeof tokens>
 }
 
+/**
+ * What the attempt was billed for, as one number, for a budget to add up. Cache reads and writes
+ * are billed too, and at different rates, so they are counted rather than dropped: a bound that
+ * ignores them is one a long context walks straight through.
+ *
+ * Undefined when the attempt said nothing, which a budget must read as "unknown" rather than as
+ * nothing spent.
+ *
+ * The split path in `packages/temporal/src/l2-step.ts` writes this arithmetic out again, because
+ * that file is bundled into the workflow sandbox and may not import this one. Change one, change
+ * the other.
+ */
+export const billed = (settlement: StepSettlement | undefined) => {
+  const t = settlement?.tokens
+  if (!t) return undefined
+  return {
+    tokens: (t.input ?? 0) + (t.output ?? 0) + (t.reasoning ?? 0) + (t.cache?.read ?? 0) + (t.cache?.write ?? 0),
+  }
+}
+
 /** The shape a tool call's input is recorded in. Exported so a dispatcher publishing Tool.Called
  * for a deferred call records it exactly as the streaming path would. */
 export const record = (value: unknown): Record<string, unknown> =>

packages/core/test/session-runner-model-call.test.ts

@@ -44,6 +44,25 @@ const callsTool: LLMClientShape["stream"] = () =>
     LLMEvent.toolCall({ id: "call_probe", name: "probe_write", input: {} }),
     LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
   ])
+// The same, with what the provider says it billed for. A budget outside the agent adds these up,
+// so what the whole-step path carries out of a turn is as load-bearing as what it recorded.
+const callsToolAndBills: LLMClientShape["stream"] = () =>
+  Stream.fromIterable([
+    LLMEvent.stepStart({ index: 0 }),
+    LLMEvent.toolCall({ id: "call_probe", name: "probe_write", input: {} }),
+    LLMEvent.stepFinish({
+      index: 0,
+      reason: "tool-calls",
+      usage: {
+        inputTokens: 10,
+        nonCachedInputTokens: 8,
+        outputTokens: 4,
+        reasoningTokens: 1,
+        cacheReadInputTokens: 2,
+        cacheWriteInputTokens: 3,
+      },
+    }),
+  ])
 // Calls the tool that declines.
 const callsDecliningTool: LLMClientShape["stream"] = () =>
   Stream.fromIterable([
@@ -310,6 +329,27 @@ describe("SessionRunner model-only attempt", () => {
     }),
   )
 
+  runnerHarness(callsToolAndBills).effect("carries what the attempt was billed for out of a whole step", () =>
+    Effect.gen(function* () {
+      yield* seedSession(sessionID)
+      yield* registerProbes(counters())
+      const runner = yield* SessionRunner.Service
+
+      const result = yield* runner.runStep({
+        sessionID,
+        step: 2,
+        promotion: undefined,
+        first: false,
+        force: false,
+      })
+
+      // 8 not-cached in, 3 visible out (4 less the 1 of reasoning), 1 reasoning, 2 cache read and
+      // 3 cache write. A deployment that never splits a step has no other way to know what a turn
+      // cost, and every one of those five is billed.
+      expect(result.spent).toEqual({ tokens: 17 })
+    }),
+  )
+
   // A step ending is not a turn ending: a steer or a queued prompt continues the same turn through
   // another step, so the one place that decides publishes the turn's own ending.
   runnerHarness(textOnly).effect("says the turn ended, once, when nothing follows it", () =>

packages/opencode/src/cli/cmd/detached.ts

@@ -182,7 +182,11 @@ export const SessionScheduleCommand = cmd({
               {
                 sessionID,
                 text: args.prompt,
-                session: { idleTimeout: config.idleTimeout, stepped: config.stepped === true },
+                session: {
+                  idleTimeout: config.idleTimeout,
+                  stepped: config.stepped === true,
+                  ...(config.budget ? { budget: config.budget } : {}),
+                },
               },
             ],
           },

packages/temporal/README.md

@@ -65,6 +65,9 @@ session runs as the workflow `session-exec-<sessionID>`.
 | `OPENCODE_TEMPORAL_PROFILE` | `local` | `fleet` sets the defaults a multi-host deployment needs and refuses combinations that cannot work. See below. |
 | `OPENCODE_TEMPORAL_API_KEY`, `OPENCODE_TEMPORAL_API_KEY_FILE` | | Temporal Cloud credentials, the second read from a file. |
 | `OPENCODE_TEMPORAL_TLS_CERT`, `OPENCODE_TEMPORAL_TLS_KEY`, `OPENCODE_TEMPORAL_TLS_CA` | | A certificate pair for a cluster with mTLS. `OPENCODE_TEMPORAL_TLS=1` for TLS without a client certificate. |
+| `OPENCODE_TEMPORAL_BUDGET_TOKENS`, `OPENCODE_TEMPORAL_BUDGET_SECONDS` | | What one turn may spend before the supervisor stops driving it, checked between steps. |
+| `OPENCODE_TEMPORAL_BUDGET_HARD_SECONDS` | | A deadline for one turn. It stops the step that is running, where the other bounds let it finish. |
+| `OPENCODE_TEMPORAL_BUDGET_SESSION_TOKENS`, `OPENCODE_TEMPORAL_BUDGET_SESSION_SECONDS` | | The same two bounds for the whole session, across every turn it runs. |
 | `OPENCODE_EVENT_POLL_MS` | | How often a live subscriber re-reads the log for events another process appended. `0` turns the tick off. |
 | `OPENCODE_WATCH_POLL_MS` | 30000 | How often `session watch` asks the running set whether the turn ended in a gap it could not see. |
 | `OPENCODE_DB` | | One absolute path shared by every process on a host. |
@@ -345,6 +348,7 @@ TEMPORAL_ADDRESS=temporal.internal:7233 \
 | A fresh worker receives a project | Needs the directory | Packs rebuild the tracked files at the recorded path; ignored files do not travel |
 | A later turn reuses a directory an abandoned tool may still write | Not reachable | The directory is refused until the call can be shown to be over, then moved aside and rebuilt |
 | A turn never stops stepping | The coordinator's own loop | The supervisor stops after 200 steps and says so |
+| A turn spends more than intended | Nothing bounds it | An operator's budget on the supervisor stops the turn on tokens or wall clock |
 
 An unknown tool outcome is a loss of evidence, not proof that the tool stopped or failed. The model
 can ask for a new call after reading it. A non-idempotent external effect needs an idempotency key,

packages/temporal/src/config.ts

@@ -56,6 +56,18 @@ export interface Interface {
   /** Whether the store is shared across hosts (`OPENCODE_DB_URL`). A fleet without it is a set of
    * workers that cannot see each other's sessions. */
   readonly sharedStore: boolean
+  /** What one turn, and the whole session, may spend before the supervisor stops driving it. Off
+   * unless set. Tokens are what the provider reported billed, added up as each attempt reports;
+   * seconds are wall clock on the workflow's own clock. `hardSeconds` is a deadline that stops the
+   * step that is running, where the other bounds wait for it to finish. Rides the workflow input,
+   * like the mode does, because the sandbox cannot read env and a rollover has to carry it. */
+  readonly budget?: {
+    readonly tokens?: number
+    readonly seconds?: number
+    readonly hardSeconds?: number
+    readonly sessionTokens?: number
+    readonly sessionSeconds?: number
+  }
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/temporal/Config") {}
@@ -64,6 +76,30 @@ const read = (path: string | undefined) => (path ? readFileSync(path, "utf8") :
 const given = (name: string) => process.env[name] !== undefined && process.env[name] !== ""
 const onOff = (name: string, fallback: boolean) => (given(name) ? process.env[name] === "1" : fallback)
 
+const BUDGET_VARS = {
+  tokens: "OPENCODE_TEMPORAL_BUDGET_TOKENS",
+  seconds: "OPENCODE_TEMPORAL_BUDGET_SECONDS",
+  hardSeconds: "OPENCODE_TEMPORAL_BUDGET_HARD_SECONDS",
+  sessionTokens: "OPENCODE_TEMPORAL_BUDGET_SESSION_TOKENS",
+  sessionSeconds: "OPENCODE_TEMPORAL_BUDGET_SESSION_SECONDS",
+} as const
+
+// A number a bound can be read as, or nothing. What is set but unreadable is left for `preflight`
+// to refuse: a bound that silently reads as none is worse than a process that will not start.
+const bound = (name: string) => {
+  if (!given(name)) return undefined
+  const value = Number(process.env[name])
+  return Number.isFinite(value) && value > 0 ? value : undefined
+}
+
+const budgetFromEnv = (): Interface["budget"] => {
+  const set = Object.entries(BUDGET_VARS).flatMap(([key, name]) => {
+    const value = bound(name)
+    return value === undefined ? [] : [[key, value] as const]
+  })
+  return set.length === 0 ? undefined : Object.fromEntries(set)
+}
+
 export const fromEnv = (): Interface => {
   const stepAffinity = process.env.OPENCODE_TEMPORAL_STEP_AFFINITY !== "0"
   const worktreeAffinity = process.env.OPENCODE_TEMPORAL_WORKTREE_AFFINITY === "1"
@@ -96,6 +132,7 @@ export const fromEnv = (): Interface => {
     // neither kind of affinity keeping the step's tools on one worker.
     serialTools: onOff("OPENCODE_TEMPORAL_SERIAL_TOOLS", !stepAffinity && !worktreeAffinity && sharedStore),
     sharedStore,
+    budget: budgetFromEnv(),
   }
 }
 
@@ -146,6 +183,9 @@ export const preflight = (config: Interface): string[] => {
     problems.push("an API key is set but TEMPORAL_NAMESPACE is `default`, which is not a Cloud namespace")
   if (!!process.env.OPENCODE_TEMPORAL_TLS_CERT !== !!process.env.OPENCODE_TEMPORAL_TLS_KEY)
     problems.push("OPENCODE_TEMPORAL_TLS_CERT and OPENCODE_TEMPORAL_TLS_KEY come as a pair")
+  for (const name of Object.values(BUDGET_VARS))
+    if (given(name) && bound(name) === undefined)
+      problems.push(`${name} is \`${process.env[name]}\`, which is not a positive number`)
   return problems
 }
 
@@ -176,4 +216,9 @@ export const describe = (config: Interface): Record<string, string> => ({
   stepAffinity: String(config.stepAffinity !== false),
   serialTools: String(config.serialTools === true),
   credentials: config.apiKey ? "api key" : config.tls ? "certificate pair" : "none (plaintext)",
+  budget: config.budget
+    ? Object.entries(config.budget)
+        .map(([name, value]) => `${name}=${value}`)
+        .join(" ")
+    : "none",
 })

packages/temporal/src/drain.ts

@@ -32,14 +32,51 @@ export interface StepDrainResult {
   continue: boolean
   step: number
   promotion: string | null
+  /** What this step's provider attempt was billed for, as the attempt itself reported it. The
+   * supervisor adds these up for the turn's budget. Absent from a step that made no attempt. */
+  spent?: { readonly tokens: number }
+  /** What the session has been billed in total, as its own row holds it. Read off the record rather
+   * than added up by the supervisor, so it survives everything the supervisor does not: a run that
+   * rolled over, a session that went idle and was woken again, a turn some other client ran. */
+  session?: { readonly tokens: number }
+}
+
+/**
+ * What the session has been billed in total, off its own row. Every count the record keeps, because
+ * a provider bills for all of them and a bound that drops the cached ones is one a long context
+ * walks straight through.
+ *
+ * Read before the step runs, so it lags by the step now running. A session's bound overshoots by
+ * the same step a turn's does, and for the same reason: what a step cost is not known until it is
+ * taken.
+ */
+export const sessionSpend = (session: SessionSchema.Info | undefined) => {
+  const t = session?.tokens
+  // Nothing rather than zero when the row has no counts: a bound that reads a missing number as
+  // nothing spent is one that never stops a session.
+  if (!t) return undefined
+  return {
+    tokens: (t.input ?? 0) + (t.output ?? 0) + (t.reasoning ?? 0) + (t.cache?.read ?? 0) + (t.cache?.write ?? 0),
+  }
 }
 
 /** The step result as it crosses the activity boundary. A missing session reads as a step with
  * nothing left to do, which is what a resume waiting on it should see. */
-export const toStepResult = (step: number, result?: SessionRunner.StepResult): StepDrainResult =>
-  result === undefined
-    ? { continue: false, step, promotion: null }
-    : { continue: result.continue, step: result.step, promotion: result.promotion ?? null }
+export const toStepResult = (
+  step: number,
+  result?: SessionRunner.StepResult,
+  session?: SessionSchema.Info,
+): StepDrainResult => {
+  if (result === undefined) return { continue: false, step, promotion: null }
+  const total = sessionSpend(session)
+  return {
+    continue: result.continue,
+    step: result.step,
+    promotion: result.promotion ?? null,
+    ...(result.spent ? { spent: result.spent } : {}),
+    ...(total ? { session: total } : {}),
+  }
+}
 
 export interface DrainDeps {
   readonly store: SessionStore.Interface
@@ -109,7 +146,7 @@ export const makeDrains = (deps: DrainDeps) => {
             first: input.first,
             force: input.force,
           })
-          .pipe(Effect.map((result) => toStepResult(input.step, result))),
+          .pipe(Effect.map((result) => toStepResult(input.step, result, session))),
       ).pipe(Effect.map((result) => result ?? toStepResult(input.step))),
       // A whole step turns a declined permission into an interrupt to halt its own loop, so an
       // interrupt with nothing cancelling it is that refusal and nothing else.

packages/temporal/src/executor.ts

@@ -70,6 +70,9 @@ const layer = Layer.effect(
     // Only the client can read whether the store is shared, so whether a step's tools may overlap
     // is decided here and rides the workflow input.
     const SERIAL_TOOLS = config.serialTools === true
+    // The operator's bound on what a turn may spend. It rides the workflow input like the mode does,
+    // because the sandbox cannot read env and a rollover has to carry it.
+    const BUDGET = config.budget
     const AFFINITY = config.worktreeAffinity === true
     // The tree this process serves when affinity is on. A serve process with an embedded worker is
     // already sitting in it, so the process directory is the right default.
@@ -226,6 +229,7 @@ const layer = Layer.effect(
                 idleTimeout: IDLE_TIMEOUT,
                 stepped: STEPPED,
                 serialTools: SERIAL_TOOLS,
+                ...(BUDGET ? { budget: BUDGET } : {}),
               } satisfies WF.SessionTurnOptions,
             ],
             signal: WF.wake,
@@ -292,6 +296,7 @@ const layer = Layer.effect(
                       idleTimeout: IDLE_TIMEOUT,
                       stepped: STEPPED,
                       serialTools: SERIAL_TOOLS,
+                      ...(BUDGET ? { budget: BUDGET } : {}),
                     } satisfies WF.SessionTurnOptions,
                   ],
                   workflowIdConflictPolicy: "USE_EXISTING",

packages/temporal/src/stepped-drain.ts

@@ -13,7 +13,7 @@ import type { StepSettlement } from "@opencode-ai/core/session/runner/publish-ll
 import type { SessionInput } from "@opencode-ai/core/session/input"
 import { WorktreeMaterializer } from "@opencode-ai/core/session/execution/worktree"
 import { runAtBoundary } from "./boundary"
-import { type InSession, type StepDrainInput, type StepDrainResult, toStepResult } from "./drain"
+import { type InSession, type StepDrainInput, type StepDrainResult, sessionSpend, toStepResult } from "./drain"
 
 /** The provider attempt of one step. Same shape as a whole-step drain: the difference is what it
  * does with the tool calls, not what it needs to start. */
@@ -36,6 +36,8 @@ export type ModelCallDrainResult =
        * worker was not given a queue of its own, and never required: the step falls back to the
        * shared queue and the tree is rebuilt there. */
       readonly queue?: string
+      /** What the session has been billed in total, off its own row. See `sessionSpend`. */
+      readonly session?: { readonly tokens: number }
     }
 
 export interface ToolCallDrainInput {
@@ -90,19 +92,28 @@ export const makeSteppedDrains = ({ inSession, worktrees, stepQueue }: SteppedDr
       input.sessionID,
       signal,
       inSession(input.sessionID, input.owner, { claim: true }, (runner, session) =>
-        runner.runModelCall({
-          sessionID: session.id,
-          step: input.step,
-          promotion: (input.promotion ?? undefined) as SessionInput.Delivery | undefined,
-          first: input.first,
-          force: input.force,
-        }),
+        runner
+          .runModelCall({
+            sessionID: session.id,
+            step: input.step,
+            promotion: (input.promotion ?? undefined) as SessionInput.Delivery | undefined,
+            first: input.first,
+            force: input.force,
+          })
+          // The session's own totals, off the row this drain already loaded. The attempt says what
+          // it cost; the row says what the session has cost, which is what outlives this run.
+          .pipe(Effect.map((called) => ({ ...called, session: sessionSpend(session) }))),
       ).pipe(
         Effect.map(
           (result): ModelCallDrainResult =>
             result === undefined || result.kind === "settled"
               ? { kind: "settled", result: toStepResult(input.step, result?.result) }
-              : { ...result, owner: input.owner, ...(stepQueue === undefined ? {} : { queue: stepQueue }) },
+              : {
+                  ...result,
+                  owner: input.owner,
+                  ...(stepQueue === undefined ? {} : { queue: stepQueue }),
+                  ...(result.session ? { session: result.session } : {}),
+                },
         ),
       ),
     )
@@ -156,7 +167,14 @@ export const makeSteppedDrains = ({ inSession, worktrees, stepQueue }: SteppedDr
             needsContinuation: input.needsContinuation,
             withoutTheTree: input.withoutTheTree,
           })
-          .pipe(Effect.map((result) => toStepResult(input.step, result))),
+          // Only the loop decision: what the step spent was reported by its model call.
+          .pipe(
+            Effect.map((result) => ({
+              continue: result.continue,
+              step: result.step,
+              promotion: result.promotion ?? null,
+            })),
+          ),
       ).pipe(Effect.map((result) => result ?? toStepResult(input.step))),
     )