feat: Grok broker usage accounting, per-wake limits and public projection

docs/engines.md

@@ -88,6 +88,21 @@ skills, workflows, plan mode, subagents, memory or web search, and a declared
 model and reasoning effort from a closed list (default `grok-4.6` at `low`).
 The broker proxy refuses any request outside that shape before it spends.
 
+Each broker registration (`service.json` v2) declares its model and effort,
+its usage ledger, and turn limits `{maxRequests, maxTokens, timeoutMs}`. A wake
+may only lower them (`DAIMON_ENGINE_WAKE_TIMEOUT_MS`,
+`DAIMON_ENGINE_WAKE_TOKEN_CEILING`; the `DAIMON_CODEX_WAKE_*` names are
+aliases). The proxy refuses request `maxRequests + 1` and any request after the
+deadline with HTTP 429 before upstream, and stops admitting requests once the
+upstream-reported running total (cached input included) reaches `maxTokens`, so
+a turn overshoots its token ceiling by at most one request. A tripped limit
+kills the worker. The broker seals every terminal turn with its usage, request
+count, declared model and limit reason, and writes one usage row (keyed by
+`turn`) plus per-request rows for completed and failed turns alike; a replayed
+turn is never metered twice. `resolveOrganizationGrokBrokerProjection` exposes
+a slot's full declared shape, and `noopolis.daimon.grok-slot-preflight.v1`
+receipts bind a slot's denied-path canaries to that projection's digest.
+
 AGY uses OS-native secure storage through one private D-Bus and Secret Service
 realm. Enroll it once with:
 

package.json

@@ -66,7 +66,8 @@
     "@earendil-works/pi-coding-agent": "^0.79.10",
     "@modelcontextprotocol/sdk": "^1.29.0",
     "@noopolis/mneme": "^0.1.1",
-    "ajv": "^8.17.1"
+    "ajv": "^8.17.1",
+    "zod": "^4.4.3"
   },
   "devDependencies": {
     "@types/node": "^24.12.4",

scripts/liveGrokBrokerSession.ts

@@ -8,7 +8,9 @@ import { readChild } from "../src/pi/cliChildOutput.ts";
 import { terminateChild, trackCliChild } from "../src/pi/cliProcess.ts";
 import { decodeGrokHeadlessTurn } from "../src/pi/grokHeadlessResult.ts";
 import { readGrokBrokerCredential } from "../src/runtime/grokBrokerCredentialReader.ts";
+import { DEFAULT_GROK_BROKER_TURN_LIMITS } from "../src/runtime/engineBrokerTurnAccounting.ts";
 import { startGrokBrokerProxy } from "../src/runtime/grokBrokerProxy.ts";
+import { GrokBrokerTurnMeter } from "../src/runtime/grokBrokerTurnMeter.ts";
 import { DEFAULT_GROK_BROKER_MODEL_POLICY } from "../src/runtime/grokBrokerModelPolicy.ts";
 import { GROK_BROKER_PROVIDER_CAPABILITY_ENV, renderGrokBrokerWorkerArgs, renderGrokBrokerWorkerConfigWith } from "../src/runtime/grokBrokerWorkerConfig.ts";
 
@@ -38,6 +40,8 @@ try {
       const capability = proxy.capabilities.issue("local-auth-probe", turnId);
       // This local transport probe deliberately does not attest a native worker.
       proxy.registerIsolationGuard(turnId, async () => undefined);
+      // The proxy forwards nothing unmetered; the probe runs under the default v1 limits.
+      proxy.registerTurn(turnId, { policy: DEFAULT_GROK_BROKER_MODEL_POLICY, meter: new GrokBrokerTurnMeter(DEFAULT_GROK_BROKER_TURN_LIMITS) });
       // No MCP tools are needed for this exact-reply authentication probe.
       await writeFile(path.join(home, "config.toml"), renderGrokBrokerWorkerConfigWith(DEFAULT_GROK_BROKER_MODEL_POLICY, { proxyPort: proxy.port, mcpUrl: "http://127.0.0.1:43124/mcp" }).split("[mcp_servers.daimon]")[0]);
       const prompt = path.join(home, "prompt.txt");

src/contracts/organizationRuntimeContract.ts

@@ -1,3 +1,5 @@
+import { GROK_BROKER_MODELS, GROK_BROKER_REASONING_EFFORTS } from "./grokWorkerContract.js";
+
 /** The data-only organization-runtime constants shared by product code and artifacts. */
 export const ORGANIZATION_RUNTIME_VERSION = "noopolis.daimon.organization-runtime.v1" as const;
 export const ORGANIZATION_RUNTIME_V2_VERSION = "noopolis.daimon.organization-runtime.v2" as const;
@@ -74,7 +76,11 @@ export const ORGANIZATION_RUNTIME_CONFIG_SCHEMA = {
           codexSandbox: { type: "object", additionalProperties: false, required: ["mode", "networkAccess", "webSearch"], properties: {
             mode: { const: "workspace-write" }, networkAccess: { const: false }, webSearch: { const: "disabled" }
           } }
-        } },
+        }, allOf: [
+          // grok: a declared model is the closed broker pair, both or neither; never a Codex sandbox.
+          { if: { properties: { kind: { const: "grok" } } }, then: { properties: { model: { enum: GROK_BROKER_MODELS }, reasoningEffort: { enum: GROK_BROKER_REASONING_EFFORTS }, codexSandbox: false }, dependentRequired: { model: ["reasoningEffort"], reasoningEffort: ["model"] } } },
+          { if: { properties: { kind: { const: "agy" } } }, then: { properties: { model: false, reasoningEffort: false, codexSandbox: false } } }
+        ] },
         ...PRODUCTION_TOOL_PROPERTIES
       }
     } }

src/contracts/runtimeContractManifest.ts

@@ -68,6 +68,27 @@ export const GROK_ENGINE_BROKER = {
     }
   },
   bounds: { promptBytes: 65_536, capabilityBytes: 4_096, capabilityBundleBytes: 8_196, outputBytes: 65_536 },
+  // Accounting and limits (P2). The broker is the single sealed usage writer.
+  controlProtocolVersion: "noopolis.daimon.engine-broker.v2",
+  turnRecordVersions: ["noopolis.daimon.engine-broker-turn.v1", "noopolis.daimon.engine-broker-turn.v2"],
+  serviceConfigVersions: ["noopolis.daimon.engine-broker-service.v1", "noopolis.daimon.engine-broker-service.v2"],
+  turnLimits: {
+    keys: ["maxRequests", "maxTokens", "timeoutMs"],
+    v1Defaults: { maxRequests: 32, maxTokens: 300_000, timeoutMs: 240_000 },
+    bounds: { maxRequests: [1, GROK_WORKER_MAX_TURNS], maxTokens: [1, 10_000_000], timeoutMs: [1_000, 3_600_000] },
+    limitReasons: ["tokens", "requests", "timeout", "none"],
+    wakeMayOnlyLower: true,
+    tokenCeilingOvershoot: "at-most-one-request",
+    maxInFlightRequests: 1,
+    // A per-request usage block above this is implausible (beyond the model
+    // context window) and treated as invalid rather than added to any total.
+    requestUsageMaxTokens: 500_000,
+    // A request whose response carries no valid usage is charged this estimate.
+    missingUsageEstimate: { inputBytesPerToken: 2, outputTokens: 4_096 }
+  },
+  wakeLimitEnvironment: { timeoutMs: "DAIMON_ENGINE_WAKE_TIMEOUT_MS", maxTokens: "DAIMON_ENGINE_WAKE_TOKEN_CEILING" },
+  projectionVersion: "noopolis.daimon.grok-broker-projection.v1",
+  slotPreflightVersion: "noopolis.daimon.grok-slot-preflight.v1",
   artifacts: {
     sourceSha256: "36f60689f0a8af0e3108f5f53d78ed52b7d4b6f934c75b6184606dfa82bc741e",
     x64Sha256: "36dc76b134eb59cf5a6720b6f94228eb279108e20ea3343fa6efd9ffcb60a4d3",

src/pi/cliSession.ts

@@ -59,36 +59,10 @@ export type CliEngineKind = "agy" | "codex" | "grok";
  */
 export const AGY_MAX_TOOL_TURNS = 16;
 
-/**
- * Codex's per-wake bounds, and the one place they are decided.
- *
- * `maxToolTurns` mediates only daimon-MCP tool calls; Codex's own shell
- * (`exec_command`) is never routed through that gate, so a single Codex turn
- * previously had no ceiling at all — one production wake ran 23:32→23:42
- * (unbounded wall clock) making 51 shell calls. Codex's `--json` stream
- * reports token usage exactly once, on `turn.completed` — there is no
- * incremental total to watch mid-turn (verified against a live multi-tool-call
- * turn: `item.completed` fires once per tool call, but usage is reported only
- * on the single terminal `turn.completed`) — so the token ceiling is the best
- * bound obtainable from that wire shape: it converts an over-budget turn into
- * an explicit, killed, named failure instead of a silent success, and the
- * wall-clock timeout is what actually interrupts a runaway turn in progress.
- */
-export const DEFAULT_CODEX_WAKE_TIMEOUT_MS = 240_000;
-export const DEFAULT_CODEX_WAKE_TOKEN_CEILING = 300_000;
-export const DAIMON_CODEX_WAKE_TIMEOUT_MS_ENV = "DAIMON_CODEX_WAKE_TIMEOUT_MS";
-export const DAIMON_CODEX_WAKE_TOKEN_CEILING_ENV = "DAIMON_CODEX_WAKE_TOKEN_CEILING";
-
-const positiveInteger = (value: string | undefined, fallback: number, name: string): number => {
-  if (value === undefined) return fallback;
-  const parsed = Number(value);
-  if (!Number.isSafeInteger(parsed) || parsed <= 0) throw new Error(`${name} must be a positive integer`);
-  return parsed;
-};
-export const resolveCodexWakeTimeoutMs = (environment: NodeJS.ProcessEnv = process.env): number =>
-  positiveInteger(environment[DAIMON_CODEX_WAKE_TIMEOUT_MS_ENV], DEFAULT_CODEX_WAKE_TIMEOUT_MS, DAIMON_CODEX_WAKE_TIMEOUT_MS_ENV);
-export const resolveCodexWakeTokenCeiling = (environment: NodeJS.ProcessEnv = process.env): number =>
-  positiveInteger(environment[DAIMON_CODEX_WAKE_TOKEN_CEILING_ENV], DEFAULT_CODEX_WAKE_TOKEN_CEILING, DAIMON_CODEX_WAKE_TOKEN_CEILING_ENV);
+export {
+  DAIMON_CODEX_WAKE_TIMEOUT_MS_ENV, DAIMON_CODEX_WAKE_TOKEN_CEILING_ENV, DAIMON_ENGINE_WAKE_TIMEOUT_MS_ENV, DAIMON_ENGINE_WAKE_TOKEN_CEILING_ENV,
+  DEFAULT_CODEX_WAKE_TIMEOUT_MS, DEFAULT_CODEX_WAKE_TOKEN_CEILING, resolveCodexWakeTimeoutMs, resolveCodexWakeTokenCeiling, resolveEngineWakeLimitOverrides
+} from "./engineWakeLimits.js";
 
 export type CliEngineOptions = {
   readonly commandArgs?: readonly string[];

src/pi/codexRolloutUsage.test.ts

@@ -46,6 +46,14 @@ test("a real multi-request rollout yields one row per model request, cached and
     [288, 228, 50_292]
   ]);
   assert.deepEqual(requests.map((request) => request.cacheWrite), [0, 0, 0, 0]);
+  // End is the usage frame; start is the first non-usage frame after the previous
+  // request's usage frame, else the previous request's end.
+  assert.deepEqual(requests.map((request) => [request.startedAt, request.endedAt]), [
+    ["2026-09-05T01:32:00.678Z", "2026-09-05T01:32:19.183Z"],
+    ["2026-09-05T01:34:00.679Z", "2026-09-05T01:52:22.056Z"],
+    ["2026-09-05T01:52:22.056Z", "2026-09-05T01:52:37.604Z"],
+    ["2026-09-05T01:52:37.604Z", "2026-09-05T01:52:51.112Z"]
+  ]);
 });
 
 test("reasoning tokens, which the per-wake ledger drops entirely, survive per request", async () => {

src/pi/codexRolloutUsage.ts

@@ -44,6 +44,17 @@ export type CodexRequestUsage = Readonly<{
   output: number;
   reasoning: number;
   total: number;
+  /**
+   * When the request began and ended, from the rollout's own frame
+   * `timestamp`s. `endedAt` is the usage frame's timestamp (Codex appends it at
+   * `response.completed`); `startedAt` is the first non-usage frame after the
+   * previous request's usage frame — the tool output or turn context that
+   * triggers the next request — falling back to the previous request's end.
+   * Either is absent when the frames carry no valid timestamp: a wake-end stamp
+   * substituted here would be indistinguishable from a measured one.
+   */
+  startedAt?: string;
+  endedAt?: string;
 }>;
 
 /**
@@ -177,17 +188,20 @@ export const parseCodexRolloutRequests = (text: string, threadId: string): reado
   const requests: CodexRequestUsage[] = [];
   const fallback: CodexRequestUsage[] = [];
   let previousFallback = "";
+  const clocks = { record: requestClock(), fallback: requestClock() };
   for (const line of text.split("\n")) {
     if (line.trim().length === 0) continue;
     let frame: unknown;
     try { frame = JSON.parse(line); } catch { continue; }
     if (!isRecord(frame)) continue;
     const block = usageBlock(frame, threadId);
+    const usageFrame = frame.type === "token_usage_record" || (frame.type === "event_msg" && isRecord(frame.payload) && frame.payload.type === "token_count");
+    if (!usageFrame) { if (frame.type !== "session_meta") { clocks.record.observe(frame.timestamp); clocks.fallback.observe(frame.timestamp); } continue; }
     if (block === undefined) continue;
     if (frame.type === "token_usage_record") {
       const decoded = decodeRequestUsage(block.usage, requests.length);
       if (decoded === undefined) return [];
-      requests.push(decoded);
+      requests.push({ ...decoded, ...clocks.record.close(frame.timestamp) });
       continue;
     }
     // `token_count` is NOT one frame per request: the captured fixture carries
@@ -201,14 +215,30 @@ export const parseCodexRolloutRequests = (text: string, threadId: string): reado
     previousFallback = serialized;
     const decoded = decodeRequestUsage(block.usage, fallback.length);
     if (decoded === undefined) return [];
-    fallback.push(decoded);
+    fallback.push({ ...decoded, ...clocks.fallback.close(frame.timestamp) });
   }
   // A Codex version that emits both shapes emits `token_usage_record` once per
   // request, so the richer one wins outright rather than being merged into a
   // double count. The fallback exists only for a version that has neither.
   return requests.length > 0 ? requests : fallback;
 };
 
+const timestampOf = (value: unknown): string | undefined =>
+  typeof value === "string" && /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{1,9})?Z$/u.test(value) && !Number.isNaN(Date.parse(value)) ? value : undefined;
+
+/** Tracks one request stream's start/end stamps; see {@link CodexRequestUsage.startedAt}. */
+const requestClock = () => {
+  let start: string | undefined, previousEnd: string | undefined;
+  return {
+    observe(value: unknown): void { start ??= timestampOf(value); },
+    close(value: unknown): { startedAt?: string; endedAt?: string } {
+      const endedAt = timestampOf(value), startedAt = start ?? previousEnd;
+      start = undefined; previousEnd = endedAt;
+      return { ...(startedAt === undefined ? {} : { startedAt }), ...(endedAt === undefined ? {} : { endedAt }) };
+    }
+  };
+};
+
 /**
  * Read one turn's per-request usage. Never throws.
  *

src/pi/engineWakeLimits.test.ts

@@ -0,0 +1,20 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { resolveCodexWakeTimeoutMs, resolveCodexWakeTokenCeiling, resolveEngineWakeLimitOverrides } from "./engineWakeLimits.js";
+
+test("engine-neutral wake bounds drive Codex, with the Codex names kept as aliases", () => {
+  assert.equal(resolveCodexWakeTimeoutMs({}), 240_000);
+  assert.equal(resolveCodexWakeTokenCeiling({}), 300_000);
+  assert.equal(resolveCodexWakeTimeoutMs({ DAIMON_ENGINE_WAKE_TIMEOUT_MS: "5000" }), 5_000);
+  assert.equal(resolveCodexWakeTokenCeiling({ DAIMON_CODEX_WAKE_TOKEN_CEILING: "7000" }), 7_000);
+  assert.equal(resolveCodexWakeTokenCeiling({ DAIMON_ENGINE_WAKE_TOKEN_CEILING: "7000", DAIMON_CODEX_WAKE_TOKEN_CEILING: "7000" }), 7_000);
+  assert.throws(() => resolveCodexWakeTokenCeiling({ DAIMON_ENGINE_WAKE_TOKEN_CEILING: "7000", DAIMON_CODEX_WAKE_TOKEN_CEILING: "8000" }), /disagree/u);
+  assert.throws(() => resolveCodexWakeTimeoutMs({ DAIMON_ENGINE_WAKE_TIMEOUT_MS: "0" }), /positive integer/u);
+});
+
+test("the broker receives only the bounds an operator actually set, as lowering limits", () => {
+  assert.equal(resolveEngineWakeLimitOverrides({}), undefined);
+  assert.deepEqual(resolveEngineWakeLimitOverrides({ DAIMON_ENGINE_WAKE_TOKEN_CEILING: "400000" }), { maxTokens: 400_000 });
+  assert.deepEqual(resolveEngineWakeLimitOverrides({ DAIMON_CODEX_WAKE_TIMEOUT_MS: "480000", DAIMON_ENGINE_WAKE_TOKEN_CEILING: "1" }), { timeoutMs: 480_000, maxTokens: 1 });
+});

src/pi/engineWakeLimits.ts

@@ -0,0 +1,58 @@
+/**
+ * Per-wake engine bounds, and the one place they are decided.
+ *
+ * `maxToolTurns` mediates only daimon-MCP tool calls; Codex's own shell
+ * (`exec_command`) is never routed through that gate, so a single Codex turn
+ * previously had no ceiling at all — one production wake ran 23:32→23:42
+ * (unbounded wall clock) making 51 shell calls. Codex's `--json` stream
+ * reports token usage exactly once, on `turn.completed` — there is no
+ * incremental total to watch mid-turn (verified against a live multi-tool-call
+ * turn: `item.completed` fires once per tool call, but usage is reported only
+ * on the single terminal `turn.completed`) — so the token ceiling is the best
+ * bound obtainable from that wire shape: it converts an over-budget turn into
+ * an explicit, killed, named failure instead of a silent success, and the
+ * wall-clock timeout is what actually interrupts a runaway turn in progress.
+ *
+ * The names are engine-neutral: `DAIMON_ENGINE_WAKE_TIMEOUT_MS` and
+ * `DAIMON_ENGINE_WAKE_TOKEN_CEILING` bound Codex locally and are passed to the
+ * Grok broker as the wake's *lowering* limits (the broker refuses a value above
+ * its registration). The `DAIMON_CODEX_*` names remain aliases; setting both
+ * names of one bound to different values is refused rather than guessed.
+ */
+export const DEFAULT_CODEX_WAKE_TIMEOUT_MS = 240_000;
+export const DEFAULT_CODEX_WAKE_TOKEN_CEILING = 300_000;
+export const DAIMON_ENGINE_WAKE_TIMEOUT_MS_ENV = "DAIMON_ENGINE_WAKE_TIMEOUT_MS";
+export const DAIMON_ENGINE_WAKE_TOKEN_CEILING_ENV = "DAIMON_ENGINE_WAKE_TOKEN_CEILING";
+export const DAIMON_CODEX_WAKE_TIMEOUT_MS_ENV = "DAIMON_CODEX_WAKE_TIMEOUT_MS";
+export const DAIMON_CODEX_WAKE_TOKEN_CEILING_ENV = "DAIMON_CODEX_WAKE_TOKEN_CEILING";
+
+const positiveInteger = (value: string, name: string): number => {
+  const parsed = Number(value);
+  if (!Number.isSafeInteger(parsed) || parsed <= 0) throw new Error(`${name} must be a positive integer`);
+  return parsed;
+};
+
+const declared = (environment: NodeJS.ProcessEnv, neutral: string, alias: string): number | undefined => {
+  const primary = environment[neutral], legacy = environment[alias];
+  const value = primary === undefined ? undefined : positiveInteger(primary, neutral);
+  const aliased = legacy === undefined ? undefined : positiveInteger(legacy, alias);
+  if (value !== undefined && aliased !== undefined && value !== aliased) throw new Error(`${neutral} and ${alias} disagree; set one`);
+  return value ?? aliased;
+};
+
+export const resolveCodexWakeTimeoutMs = (environment: NodeJS.ProcessEnv = process.env): number =>
+  declared(environment, DAIMON_ENGINE_WAKE_TIMEOUT_MS_ENV, DAIMON_CODEX_WAKE_TIMEOUT_MS_ENV) ?? DEFAULT_CODEX_WAKE_TIMEOUT_MS;
+export const resolveCodexWakeTokenCeiling = (environment: NodeJS.ProcessEnv = process.env): number =>
+  declared(environment, DAIMON_ENGINE_WAKE_TOKEN_CEILING_ENV, DAIMON_CODEX_WAKE_TOKEN_CEILING_ENV) ?? DEFAULT_CODEX_WAKE_TOKEN_CEILING;
+
+/**
+ * The limits a wake asks a broker to lower to: only the bounds the operator
+ * actually set, never the Codex defaults (a broker registration's declared
+ * limits already are the defaults there).
+ */
+export const resolveEngineWakeLimitOverrides = (environment: NodeJS.ProcessEnv = process.env): Readonly<{ timeoutMs?: number; maxTokens?: number }> | undefined => {
+  const timeoutMs = declared(environment, DAIMON_ENGINE_WAKE_TIMEOUT_MS_ENV, DAIMON_CODEX_WAKE_TIMEOUT_MS_ENV);
+  const maxTokens = declared(environment, DAIMON_ENGINE_WAKE_TOKEN_CEILING_ENV, DAIMON_CODEX_WAKE_TOKEN_CEILING_ENV);
+  if (timeoutMs === undefined && maxTokens === undefined) return undefined;
+  return { ...(timeoutMs === undefined ? {} : { timeoutMs }), ...(maxTokens === undefined ? {} : { maxTokens }) };
+};

src/pi/fixtures/README.md

@@ -117,3 +117,23 @@ the previous block and why `token_usage_record` wins outright when both exist.
 The four requests also show exactly the shape the study predicted, in one wake:
 fresh input 15,742 → 248 → 4,276 → 10,068 against a context that only grows from
 34,686 to 50,004 — most of every request after the first is cache-read replay.
+
+
+# Grok 1.0.34 per-request stream fixture
+
+`grok-1.0.34-streaming-two-requests.jsonl` is a real two-request turn captured on
+2026-09-17 from `grok 1.0.34` (macOS arm64) with the lean worker flags and
+`--output-format streaming-messages-json` (P0 host matrix cell c14: one MCP
+`use_tool` call, then the answer). Sanitization before commit: the capturing
+scratchpad `cwd` was replaced with `/workspace`; every frame is otherwise
+verbatim.
+
+It pins what `grokStreamUsage.ts` reads and the broker meters per request:
+
+    assistant.message.id                       one request per distinct id
+    assistant.message.usage                    that request's own four buckets
+    result.modelUsage keys                     "grok-4.6-build" for grok-4.6
+
+The two per-request totals (2,775 + 2,810) sum exactly to the terminal
+`result.usage` (5,585), which is why a failed turn's frames are trusted as its
+partial usage.