diff --git a/docs/engines.md b/docs/engines.md index b2517f6..6342210 100644 --- a/docs/engines.md +++ b/docs/engines.md @@ -100,8 +100,18 @@ 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. +a slot's full declared shape, and `noopolis.daimon.grok-slot-preflight.v2` +receipts bind a slot's denied-path canaries to that projection's digest and to +one recycle (the caller's nonce and the slot's increasing generation). + +Evaluators (Paideia judges and the optimizer, organization uid only) borrow the +same credential through inference grants: `request_inference_grant` over the +control socket returns a ten-minute token for one declared model and effort, +which the evaluator's Grok CLI presents to the provider proxy through +`env_key` in a config rendered by `renderGrokInferenceClientConfig`. Grant +requests must carry no tools, are metered like a turn, and are written only to +the broker's separate `inferenceLedgerPath` (`kind: "inference"` rows), never +to a subject usage ledger or the wake fuse. AGY uses OS-native secure storage through one private D-Bus and Secret Service realm. Enroll it once with: diff --git a/src/contracts/runtimeContractManifest.ts b/src/contracts/runtimeContractManifest.ts index 8aebfb1..a9eb1be 100644 --- a/src/contracts/runtimeContractManifest.ts +++ b/src/contracts/runtimeContractManifest.ts @@ -87,8 +87,38 @@ export const GROK_ENGINE_BROKER = { missingUsageEstimate: { inputBytesPerToken: 2, outputTokens: 4_096 } }, wakeLimitEnvironment: { timeoutMs: "DAIMON_ENGINE_WAKE_TIMEOUT_MS", maxTokens: "DAIMON_ENGINE_WAKE_TOKEN_CEILING" }, + // Evaluator inference grants (P2c). Judges and the optimizer (organization uid + // only, over the control socket) borrow the broker's Grok credential through + // the provider proxy; they never hold it, and their spend never reaches the + // subject usage ledger or wake fuse. + inferenceGrants: { + requestKinds: ["request_inference_grant", "release_inference_grant"], + purposes: ["judge", "optimizer"], + tokenPrefix: "inference_", + ttlMs: 600_000, + limits: { maxRequests: 64, maxTokens: 2_000_000 }, + maxLiveGrants: 8, + maxInFlightRequestsPerGrant: 1, + // Top-level request members Grok 1.0.34 sends for a Paideia judge/optimizer call + // (live stub capture); `tools` and `tool_choice` are refused outright. + bodyMembers: ["messages", "model", "reasoning_effort", "response_format", "stream", "stream_options"], + messageRoles: ["system", "user", "assistant"], + failureCodes: ["auth_stale", "grant_limit", "invalid_request", "unavailable"], + ledgerVersion: "noopolis.daimon.inference-usage.v1", + ledgerDedupeKey: ["grant", "request"], + client: { + modelId: "daimon-inference-grok", + envKey: "DAIMON_INFERENCE_GRANT", + // sha256 of `renderGrokInferenceClientConfig` for the production proxy base URL and this env key. + configSha256: { + "grok-4.6": { low: "79314d039f787e4ebfec7dacf57adc969086b948f564dec008f0ed6367e6062f", medium: "6f538de0547c0c4e6a3f04ae08595ceadadabb06b75f6b6ee4c428744bb95cd8", high: "5652656effa82f0c4f09cf8226b16e6140332a5a358b571194bb5563312367ac" }, + "grok-4.5": { low: "a07f7436f1268bb399ec233c65d3b3d8fb99a11a1f175f8da1ca133c9367bc74", medium: "1f4c0d4dad1f3b09419b5739db6423a09e0049abc091594c64123a75dd53dfb9", high: "ffbc33728b821e9854fbc7c93601e599225da421ecfd6ebf10d314afcc28d6f2" }, + "grok-build": { low: "ca15c6a562a008227d39c51d3a3a83715663089b3784e8b46debb1fb67b3c4a1", medium: "01783fb6beadcf6f8486fff0836820ad43fab5662b812a907fe9cfdb83e9804d", high: "98d16f2b7d12f4eb540d625c853e51d227933e204923e43e8b9b4176f10aca2c" } + } + } + }, projectionVersion: "noopolis.daimon.grok-broker-projection.v1", - slotPreflightVersion: "noopolis.daimon.grok-slot-preflight.v1", + slotPreflightVersion: "noopolis.daimon.grok-slot-preflight.v2", artifacts: { sourceSha256: "36f60689f0a8af0e3108f5f53d78ed52b7d4b6f934c75b6184606dfa82bc741e", x64Sha256: "36dc76b134eb59cf5a6720b6f94228eb279108e20ea3343fa6efd9ffcb60a4d3", diff --git a/src/runtime/AGENTS.md b/src/runtime/AGENTS.md index 182a3fa..1628dc5 100644 --- a/src/runtime/AGENTS.md +++ b/src/runtime/AGENTS.md @@ -77,6 +77,46 @@ model (`grok-4.6-build` → `grok-4.6`), otherwise the turn fails as rejected an is still metered. Control protocol v2 is refused-v1 on the wire because both ends ship in this package. +Evaluator inference grants (`grokInferenceGrants.ts`) let Paideia judges and +the DSPy optimizer — uid 2000, the trusted evaluator side — spend the broker's +Grok credential without holding it. `request_inference_grant {model, +reasoningEffort, purpose: judge|optimizer}` is an additive control protocol v2 +verb (`engineBrokerInferenceProtocol.ts`); only the organization uid reaches it, +because the native relay admits only that `SO_PEERCRED` uid on `control.sock` +(the TS backend sees only the relay). The answer is a token +(`inference_` + 32 random bytes), the proxy base URL, an expiry (TTL ten +minutes) and the manifest limits; `release_inference_grant` frees one of the +eight live-grant slots early. Grants are their own kind: their own map keyed by +a random grant id, never the turn capability or turn meter maps, and the proxy +routes a bearer by its prefix to exactly one of the two lookups. A grant has no +worker isolation guard but the same spend gate as a turn (one request in +flight, request ceiling, between-requests token ceiling, estimate on missing +usage), so one grant is one sequential lane — parallel judges each hold one. +`grokInferenceProxyRequest.ts` accepts exactly what Grok 1.0.34 sends for the +Paideia judge argv (live stub capture): `stream: true` with +`stream_options.include_usage`, the declared `model`/`reasoning_effort`, plain +`{role, content}` messages, optional `response_format` json_schema, and **no +`tools` or `tool_choice` member at all** — the CLI's per-call `session_title` +request carries both and is refused locally. Every settled request appends one +`kind: "inference"` row (`purpose`, `grant`, `request`, model, usage, +`usage_source`) to `service.json` v2's optional `inferenceLedgerPath`, which +may never be a subject ledger; readers dedupe on `(grant, request)` +(`dedupeInferenceUsageRows`), and `wakeFuse.ts` skips inference rows. Without +that path every grant request is refused `unavailable`. Grants share the +subject's credential authority, so a stale realm fails both (accepted shared +fate): the grant request is refused `auth_stale`, and a proxied grant request +that meets a stale realm gets HTTP 401 `{"error":"auth_stale"}`, which the CLI +surfaces immediately as `Internal error: "Unauthorized (401) from …: +auth_stale …"`. `grokInferenceClientConfig.ts` renders the evaluator's private +`GROK_HOME` `config.toml` (pinned per model/effort in the manifest): the grant +token through `env_key = "DAIMON_INFERENCE_GRANT"`, the worker's lean settings, +no MCP, and `max_retries = 0` — with the default, Grok retries a refused (503) +request with backoff past 45 s instead of failing in ~0.35 s. Its init frame +reports `apiKeySource: "user"`, `tools: []`, `mcp_servers: []`, and the CLI +must be run with `--model daimon-inference-grok`. The inference ledger +directory must be provisioned setgid to the organization group (e.g. +`2100:2000 2750`) for uid 2000 to read rows the broker creates `0640`. + `grokBrokerProjection.ts` is the public, I/O-free projection of one brokered Grok agent's slot (`noopolis.daimon.grok-broker-projection.v1`): Daimon's own deny collectors plus the caller's evaluator paths, profile/config/prompt @@ -87,9 +127,14 @@ must supply canonical non-symlink paths (its fixed tmpfs and workspace roots) and verify that during provisioning. The projection also carries the seccomp profile digest and the `bubblewrap` sandbox runtime a receipt must match. `grokSlotPreflightReceipt.ts` is the zod schema a root slot supervisor's receipt must satisfy -(`noopolis.daimon.grok-slot-preflight.v1`, fixtures under +(`noopolis.daimon.grok-slot-preflight.v2`, fixtures under `fixtures/grok-slot-preflight/`); `verifyGrokSlotPreflightReceipt` binds it to the projection digest and requires a denied canary for exactly every deny path. +The projection digest does not change across recycles, so the receipt also +carries freshness: a supervisor-owned per-slot `generation` (strictly +increasing) and the caller's recycle `nonce` (32 random bytes, hex). The +verifier requires `{expectedNonce, minGeneration}` and refuses another nonce, a +lower generation, and any v1 receipt. `grokBrokerWorkerConfig.ts` is the only source of worker `config.toml` bytes; the manifest pins the sha256 of every model/effort combination and the broker diff --git a/src/runtime/engineBrokerControlClient.ts b/src/runtime/engineBrokerControlClient.ts index 5054030..6d5c845 100644 --- a/src/runtime/engineBrokerControlClient.ts +++ b/src/runtime/engineBrokerControlClient.ts @@ -1,7 +1,16 @@ import { createHash, randomUUID } from "node:crypto"; import { createConnection } from "node:net"; import { ENGINE_BROKER_VERSION, encodeEngineBrokerFrame,EngineBrokerFrameDecoder,parseEngineBrokerResponse } from "./engineBrokerProtocol.js"; +import type { EngineBrokerInferenceFailureCode, EngineBrokerInferenceRequest, EngineBrokerInferenceResponse } from "./engineBrokerInferenceProtocol.js"; import type { EngineBrokerTurnLimitOverrides } from "./engineBrokerTurnAccounting.js"; +import type { GrokBrokerModel, GrokBrokerReasoningEffort } from "./grokBrokerModelPolicy.js"; +import type { GrokInferencePurpose } from "./inferenceUsageLedger.js"; + +export type EngineBrokerInferenceGrant = Omit, "version" | "kind" | "requestId">; +/** A refused grant request; `code` is closed (`auth_stale` is the stale shared realm, `grant_limit` the live-grant cap). */ +export class EngineBrokerInferenceGrantRefused extends Error { + constructor(readonly code: EngineBrokerInferenceFailureCode) { super(`engine broker inference grant refused (${code})`); } +} /** * What the organization runtime asks of a brokered turn beyond the prompt: @@ -13,9 +22,28 @@ export interface EngineBrokerTurnClient { turn(agentId:string,wakeId:string,prom export class EngineBrokerControlClient implements EngineBrokerTurnClient { constructor(private readonly socketPath="/run/daimon-engine-broker/control.sock"){} async ready():Promise{const requestId=randomUUID(),socket=createConnection({path:this.socketPath}),decoder=new EngineBrokerFrameDecoder();await new Promise((resolve,reject)=>{let settled=false;const fail=()=>{if(settled)return;settled=true;socket.destroy();reject(new Error("engine broker unavailable"));};socket.once("error",fail);socket.once("close",fail);socket.once("connect",()=>socket.write(encodeEngineBrokerFrame({version:ENGINE_BROKER_VERSION,kind:"health",requestId})));socket.on("data",(chunk)=>{try{for(const value of decoder.push(chunk)){const response=parseEngineBrokerResponse(value);if(response.kind!=="ready"||response.requestId!==requestId||settled)throw new Error();settled=true;socket.destroy();resolve();}}catch{fail();}});});} + /** + * Evaluator side (organization uid only; the native relay enforces it): + * borrow the broker credential for one sequential lane of judge or optimizer + * requests. Refusals reject with {@link EngineBrokerInferenceGrantRefused}; + * a broker that cannot answer rejects with `engine broker unavailable`. + */ + async requestInferenceGrant(request:Readonly<{model:GrokBrokerModel;reasoningEffort:GrokBrokerReasoningEffort;purpose:GrokInferencePurpose}>):Promise{ + const response=await this.exchange({version:ENGINE_BROKER_VERSION,kind:"request_inference_grant",requestId:randomUUID(),model:request.model,reasoningEffort:request.reasoningEffort,purpose:request.purpose}); + if(response.kind==="inference_grant_refused")throw new EngineBrokerInferenceGrantRefused(response.code); + if(response.kind!=="inference_grant"||response.model!==request.model||response.reasoningEffort!==request.reasoningEffort||response.purpose!==request.purpose)throw new Error("engine broker unavailable"); + const {version:_version,kind:_kind,requestId:_requestId,...grant}=response;return grant; + } + async releaseInferenceGrant(grantId:string):Promise{ + const response=await this.exchange({version:ENGINE_BROKER_VERSION,kind:"release_inference_grant",requestId:randomUUID(),grantId}); + if(response.kind==="inference_grant_refused")throw new EngineBrokerInferenceGrantRefused(response.code); + if(response.kind!=="inference_grant_released"||response.grantId!==grantId)throw new Error("engine broker unavailable"); + return response.released; + } + private exchange(request:EngineBrokerInferenceRequest):Promise{const socket=createConnection({path:this.socketPath}),decoder=new EngineBrokerFrameDecoder();return new Promise((resolve,reject)=>{let settled=false;const fail=()=>{if(settled)return;settled=true;socket.destroy();reject(new Error("engine broker unavailable"));};socket.once("error",fail);socket.once("close",fail);socket.once("connect",()=>socket.write(encodeEngineBrokerFrame(request)));socket.on("data",(chunk)=>{try{for(const value of decoder.push(chunk)){const response=parseEngineBrokerResponse(value);if(settled||response.requestId!==request.requestId||(response.kind!=="inference_grant"&&response.kind!=="inference_grant_released"&&response.kind!=="inference_grant_refused"))throw new Error();settled=true;socket.destroy();resolve(response);}}catch{fail();}});});} async turn(agentId:string,wakeId:string,prompt:string,mcpEndpoint:string,signal?:AbortSignal,options:EngineBrokerTurnOptions={}):Promise{ const turnId=createHash("sha256").update(`${agentId}\0${wakeId}`).digest("hex"),requestId=randomUUID();const request={version:ENGINE_BROKER_VERSION,kind:"start_turn",requestId,turnId,agentId,wakeId,prompt,mcpEndpoint,...(options.limits===undefined?{}:{limits:options.limits})} as const;const socket=createConnection({path:this.socketPath});const decoder=new EngineBrokerFrameDecoder(); - return new Promise((resolve,reject)=>{let accepted=false,settled=false;const fail=()=>{if(settled)return;settled=true;cleanup();reject(new Error("engine broker unavailable"));};const cleanup=()=>{signal?.removeEventListener("abort",abort);socket.destroy();};const abort=()=>fail();signal?.addEventListener("abort",abort,{once:true});if(signal?.aborted)return abort();socket.once("connect",()=>socket.write(encodeEngineBrokerFrame(request)));socket.on("data",(chunk)=>{try{for(const value of decoder.push(chunk)){const response=parseEngineBrokerResponse(value);if(response.kind==="ready"||response.requestId!==requestId||response.turnId!==turnId)throw new Error();if(response.kind==="accepted"){if(accepted)throw new Error();accepted=true;continue;}if(!accepted||settled)throw new Error();settled=true;cleanup(); + return new Promise((resolve,reject)=>{let accepted=false,settled=false;const fail=()=>{if(settled)return;settled=true;cleanup();reject(new Error("engine broker unavailable"));};const cleanup=()=>{signal?.removeEventListener("abort",abort);socket.destroy();};const abort=()=>fail();signal?.addEventListener("abort",abort,{once:true});if(signal?.aborted)return abort();socket.once("connect",()=>socket.write(encodeEngineBrokerFrame(request)));socket.on("data",(chunk)=>{try{for(const value of decoder.push(chunk)){const response=parseEngineBrokerResponse(value);if((response.kind!=="accepted"&&response.kind!=="completed"&&response.kind!=="failed")||response.requestId!==requestId||response.turnId!==turnId)throw new Error();if(response.kind==="accepted"){if(accepted)throw new Error();accepted=true;continue;}if(!accepted||settled)throw new Error();settled=true;cleanup(); if(options.model!==undefined&&response.model!==options.model){reject(new Error(`engine broker turn used model ${response.model}, not the declared ${options.model}`));return;} if(response.kind==="completed")resolve(response.text);else reject(new Error(`engine broker turn failed (${response.code}${response.limitReason==="none"?"":`; limit=${response.limitReason}`}${response.diagnostic ? `; ${response.diagnostic.stage}/${response.diagnostic.failureClass}; exit=${response.diagnostic.exitCode}; signal=${response.diagnostic.termSignal}` : ""})`));}}catch{fail();}});socket.once("error",fail);socket.once("close",()=>{if(!settled)fail();});}); } diff --git a/src/runtime/engineBrokerInferenceProtocol.ts b/src/runtime/engineBrokerInferenceProtocol.ts new file mode 100644 index 0000000..02310b9 --- /dev/null +++ b/src/runtime/engineBrokerInferenceProtocol.ts @@ -0,0 +1,77 @@ +import { GROK_BROKER_MODELS, GROK_BROKER_REASONING_EFFORTS } from "../contracts/grokWorkerContract.js"; +import { GROK_ENGINE_BROKER } from "../contracts/runtimeContractManifest.js"; +import type { EngineBrokerTurnLimits } from "./engineBrokerTurnAccounting.js"; +import type { GrokBrokerModel, GrokBrokerReasoningEffort } from "./grokBrokerModelPolicy.js"; +import type { GrokInferencePurpose } from "./inferenceUsageLedger.js"; + +/** + * Evaluator inference grant frames, additive kinds of control protocol v2. + * + * Both ends ship in this package (the organization-side client and the broker + * service), so the kinds join v2 without a version bump; a broker that + * predates them refuses the unknown kind and closes the connection, which the + * client reports as `unavailable`. The native relay forwards frames opaquely + * and admits only the organization uid on `control.sock` (`SO_PEERCRED`), so + * that check is what limits grants to the evaluator side. + * + * The grant `token` is the bearer the evaluator's Grok CLI presents to the + * provider proxy (`env_key`). It never carries the broker credential. + */ +const SPEC = GROK_ENGINE_BROKER.inferenceGrants; +export const ENGINE_BROKER_INFERENCE_FAILURE_CODES = SPEC.failureCodes; +export type EngineBrokerInferenceFailureCode = (typeof ENGINE_BROKER_INFERENCE_FAILURE_CODES)[number]; +export const GROK_INFERENCE_PROXY_BASE_URL = `http://${GROK_ENGINE_BROKER.providerProxy.host}:${GROK_ENGINE_BROKER.providerProxy.port}/v1` as const; + +type V = "noopolis.daimon.engine-broker.v2"; +export type EngineBrokerInferenceRequest = + | Readonly<{ version: V; kind: "request_inference_grant"; requestId: string; model: GrokBrokerModel; reasoningEffort: GrokBrokerReasoningEffort; purpose: GrokInferencePurpose }> + | Readonly<{ version: V; kind: "release_inference_grant"; requestId: string; grantId: string }>; +export type EngineBrokerInferenceResponse = + | Readonly<{ version: V; kind: "inference_grant"; requestId: string; grantId: string; token: string; baseUrl: typeof GROK_INFERENCE_PROXY_BASE_URL; model: GrokBrokerModel; reasoningEffort: GrokBrokerReasoningEffort; purpose: GrokInferencePurpose; expiresAt: string; limits: EngineBrokerTurnLimits }> + | Readonly<{ version: V; kind: "inference_grant_released"; requestId: string; grantId: string; released: boolean }> + | Readonly<{ version: V; kind: "inference_grant_refused"; requestId: string; code: EngineBrokerInferenceFailureCode }>; + +type JsonRecord = Record; +const invalid = (): TypeError => new TypeError("invalid broker frame"); +const exact = (value: JsonRecord, fields: readonly string[]): void => { if (Object.keys(value).length !== fields.length || fields.some((field) => !Object.hasOwn(value, field))) throw invalid(); }; +const member = (list: readonly T[], value: unknown): T => { if (!(list as readonly unknown[]).includes(value)) throw invalid(); return value as T; }; +const grantId = (value: unknown): string => { if (typeof value !== "string" || !/^[a-f0-9]{32}$/u.test(value)) throw invalid(); return value; }; +const TOKEN = new RegExp(`^${SPEC.tokenPrefix}[A-Za-z0-9_-]{43}$`, "u"); + +/** `input` has already passed the v2 envelope checks; `requestId` is the parsed id. */ +export function parseEngineBrokerInferenceRequest(input: JsonRecord, requestId: string, version: V): EngineBrokerInferenceRequest { + if (input.kind === "request_inference_grant") { + exact(input, ["version", "kind", "requestId", "model", "reasoningEffort", "purpose"]); + return { version, kind: "request_inference_grant", requestId, model: member(GROK_BROKER_MODELS, input.model), reasoningEffort: member(GROK_BROKER_REASONING_EFFORTS, input.reasoningEffort), purpose: member(SPEC.purposes, input.purpose) }; + } + if (input.kind === "release_inference_grant") { + exact(input, ["version", "kind", "requestId", "grantId"]); + return { version, kind: "release_inference_grant", requestId, grantId: grantId(input.grantId) }; + } + throw invalid(); +} + +export function parseEngineBrokerInferenceResponse(input: JsonRecord, requestId: string, version: V): EngineBrokerInferenceResponse { + if (input.kind === "inference_grant") { + exact(input, ["version", "kind", "requestId", "grantId", "token", "baseUrl", "model", "reasoningEffort", "purpose", "expiresAt", "limits"]); + if (typeof input.token !== "string" || !TOKEN.test(input.token) || input.baseUrl !== GROK_INFERENCE_PROXY_BASE_URL || typeof input.expiresAt !== "string" || Number.isNaN(Date.parse(input.expiresAt)) || new Date(input.expiresAt).toISOString() !== input.expiresAt) throw invalid(); + const limits = input.limits as JsonRecord; + if (limits === null || typeof limits !== "object" || Array.isArray(limits)) throw invalid(); + exact(limits, ["maxRequests", "maxTokens", "timeoutMs"]); + if (!(Number.isSafeInteger(limits.maxRequests) && (limits.maxRequests as number) >= 1 && (limits.maxRequests as number) <= SPEC.limits.maxRequests && Number.isSafeInteger(limits.maxTokens) && (limits.maxTokens as number) >= 1 && (limits.maxTokens as number) <= SPEC.limits.maxTokens && Number.isSafeInteger(limits.timeoutMs) && (limits.timeoutMs as number) >= 1 && (limits.timeoutMs as number) <= SPEC.ttlMs)) throw invalid(); + return { version, kind: "inference_grant", requestId, grantId: grantId(input.grantId), token: input.token, baseUrl: GROK_INFERENCE_PROXY_BASE_URL, model: member(GROK_BROKER_MODELS, input.model), reasoningEffort: member(GROK_BROKER_REASONING_EFFORTS, input.reasoningEffort), purpose: member(SPEC.purposes, input.purpose), expiresAt: input.expiresAt, limits: { maxRequests: limits.maxRequests as number, maxTokens: limits.maxTokens as number, timeoutMs: limits.timeoutMs as number } }; + } + if (input.kind === "inference_grant_released") { + exact(input, ["version", "kind", "requestId", "grantId", "released"]); + if (typeof input.released !== "boolean") throw invalid(); + return { version, kind: "inference_grant_released", requestId, grantId: grantId(input.grantId), released: input.released }; + } + if (input.kind === "inference_grant_refused") { + exact(input, ["version", "kind", "requestId", "code"]); + return { version, kind: "inference_grant_refused", requestId, code: member(ENGINE_BROKER_INFERENCE_FAILURE_CODES, input.code) }; + } + throw invalid(); +} + +export const isEngineBrokerInferenceRequestKind = (kind: unknown): boolean => kind === "request_inference_grant" || kind === "release_inference_grant"; +export const isEngineBrokerInferenceResponseKind = (kind: unknown): boolean => kind === "inference_grant" || kind === "inference_grant_released" || kind === "inference_grant_refused"; diff --git a/src/runtime/engineBrokerInferenceService.test.ts b/src/runtime/engineBrokerInferenceService.test.ts new file mode 100644 index 0000000..27eed5a --- /dev/null +++ b/src/runtime/engineBrokerInferenceService.test.ts @@ -0,0 +1,66 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import test from "node:test"; + +import { EngineBrokerControlClient, EngineBrokerInferenceGrantRefused } from "./engineBrokerControlClient.js"; +import { GROK_INFERENCE_PROXY_BASE_URL } from "./engineBrokerInferenceProtocol.js"; +import { parseEngineBrokerRequest, parseEngineBrokerResponse } from "./engineBrokerProtocol.js"; +import { startEngineBrokerServiceWithIdentity, type EngineBrokerServiceEngine } from "./engineBrokerService.js"; +import { GrokInferenceGrantRefused, GrokInferenceGrants } from "./grokInferenceGrants.js"; + +const V = "noopolis.daimon.engine-broker.v2"; +const baseEngine = (): EngineBrokerServiceEngine => ({ turn: async () => { throw new Error("no turns"); }, readiness: () => ({ providerProxyPort: 43123, mcpFacadePort: 43124, registrations: 1, credentialStale: false, realmLease: true, workerIsolation: true }), close: async () => undefined }); + +async function withService(engine: EngineBrokerServiceEngine, run: (client: EngineBrokerControlClient) => Promise): Promise { + const directory = await mkdtemp(path.join(tmpdir(), "daimon-broker-grants-")), socketPath = path.join(directory, "broker.sock"); + const service = await startEngineBrokerServiceWithIdentity(engine, socketPath, process.getuid!()); + try { await run(new EngineBrokerControlClient(socketPath)); } finally { await service.close(); await rm(directory, { recursive: true, force: true }); } +} +const refusedWith = (code: string) => (error: unknown) => error instanceof EngineBrokerInferenceGrantRefused && error.code === code; + +test("the control socket issues, refuses and releases inference grants", async () => { + const grants = new GrokInferenceGrants({ maxLiveGrants: 1 }); + let stale = false; + const engine: EngineBrokerServiceEngine = { ...baseEngine(), requestInferenceGrant: (request) => { if (stale) throw new GrokInferenceGrantRefused("auth_stale"); return grants.issue(request); }, releaseInferenceGrant: (grantId) => grants.release(grantId) }; + try { + await withService(engine, async (client) => { + const grant = await client.requestInferenceGrant({ model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }); + assert.equal(grant.baseUrl, GROK_INFERENCE_PROXY_BASE_URL); assert.equal(grant.baseUrl, "http://127.0.0.1:43123/v1"); + assert.match(grant.token, /^inference_/u); assert.deepEqual(grant.limits, { maxRequests: 64, maxTokens: 2_000_000, timeoutMs: 600_000 }); + assert.ok(Date.parse(grant.expiresAt) - Date.now() <= 600_000); + assert.ok(grants.authorize(grant.token)); + await assert.rejects(client.requestInferenceGrant({ model: "grok-4.6", reasoningEffort: "low", purpose: "optimizer" }), refusedWith("grant_limit")); + assert.equal(await client.releaseInferenceGrant(grant.grantId), true); + assert.equal(await client.releaseInferenceGrant(grant.grantId), false); + assert.equal(grants.authorize(grant.token), undefined); + stale = true; + await assert.rejects(client.requestInferenceGrant({ model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }), refusedWith("auth_stale")); + }); + } finally { grants.close(); } +}); + +test("an engine without grants refuses them as unavailable, and an off-list model never reaches the engine", async () => { + await withService(baseEngine(), async (client) => { + await assert.rejects(client.requestInferenceGrant({ model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }), refusedWith("unavailable")); + }); + let called = false; + await withService({ ...baseEngine(), requestInferenceGrant: () => { called = true; throw new Error("unreachable"); } }, async (client) => { + await assert.rejects(client.requestInferenceGrant({ model: "grok-3" as "grok-4.6", reasoningEffort: "low", purpose: "judge" }), /unavailable/u); + await assert.rejects(client.requestInferenceGrant({ model: "grok-4.6", reasoningEffort: "xhigh" as "low", purpose: "judge" }), /unavailable/u); + await assert.rejects(client.requestInferenceGrant({ model: "grok-4.6", reasoningEffort: "low", purpose: "subject" as "judge" }), /unavailable/u); + }); + assert.equal(called, false); +}); + +test("grant frames are closed and never carry tools or undeclared members", () => { + const request = { version: V, kind: "request_inference_grant", requestId: "r1", model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }; + assert.deepEqual(parseEngineBrokerRequest(request), request); + for (const bad of [{ ...request, tools: [] }, { ...request, purpose: "subject" }, { ...request, model: "grok-3" }, { ...request, limits: { maxRequests: 1 } }, { ...request, version: "noopolis.daimon.engine-broker.v1" }]) assert.throws(() => parseEngineBrokerRequest(bad), /invalid broker frame/u); + assert.throws(() => parseEngineBrokerRequest({ version: V, kind: "release_inference_grant", requestId: "r1", grantId: "not-hex" }), /invalid broker frame/u); + const grant = { version: V, kind: "inference_grant", requestId: "r1", grantId: "a".repeat(32), token: `inference_${"A".repeat(43)}`, baseUrl: GROK_INFERENCE_PROXY_BASE_URL, model: "grok-4.6", reasoningEffort: "low", purpose: "judge", expiresAt: "2026-09-17T05:00:00.000Z", limits: { maxRequests: 64, maxTokens: 2_000_000, timeoutMs: 600_000 } }; + assert.deepEqual(parseEngineBrokerResponse(grant), grant); + for (const bad of [{ ...grant, baseUrl: "http://evil:43123/v1" }, { ...grant, token: "A".repeat(53) }, { ...grant, limits: { ...grant.limits, timeoutMs: 600_001 } }, { ...grant, limits: { ...grant.limits, maxRequests: 65 } }, { ...grant, credential: "x" }]) assert.throws(() => parseEngineBrokerResponse(bad), /invalid broker frame/u); + assert.throws(() => parseEngineBrokerResponse({ version: V, kind: "inference_grant_refused", requestId: "r1", code: "because" }), /invalid broker frame/u); +}); diff --git a/src/runtime/engineBrokerProtocol.ts b/src/runtime/engineBrokerProtocol.ts index d5de2f0..9b9f644 100644 --- a/src/runtime/engineBrokerProtocol.ts +++ b/src/runtime/engineBrokerProtocol.ts @@ -1,3 +1,4 @@ +import { isEngineBrokerInferenceRequestKind, isEngineBrokerInferenceResponseKind, parseEngineBrokerInferenceRequest, parseEngineBrokerInferenceResponse, type EngineBrokerInferenceRequest, type EngineBrokerInferenceResponse } from "./engineBrokerInferenceProtocol.js"; import { parseEngineBrokerTurnAccounting, parseEngineBrokerTurnLimitOverrides, type EngineBrokerTurnAccounting, type EngineBrokerTurnLimitOverrides } from "./engineBrokerTurnAccounting.js"; /** @@ -16,7 +17,8 @@ const ID = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/u; export type EngineBrokerRequest = | Readonly<{ version: typeof VERSION; kind: "health"; requestId: string }> | Readonly<{ version: typeof VERSION; kind: "start_turn"; requestId: string; turnId: string; agentId: string; wakeId: string; prompt: string; mcpEndpoint: string; limits?: EngineBrokerTurnLimitOverrides }> - | Readonly<{ version: typeof VERSION; kind: "cancel_turn"; requestId: string; turnId: string }>; + | Readonly<{ version: typeof VERSION; kind: "cancel_turn"; requestId: string; turnId: string }> + | EngineBrokerInferenceRequest; export interface EngineBrokerFailureDiagnostic { status:string;stage:string;failureClass:string;profileApplied:boolean;exitCode:number;termSignal:number;workerPid:number;workerUid:number;startTicks:string } @@ -24,7 +26,8 @@ export type EngineBrokerResponse = | Readonly<{ version: typeof VERSION; kind: "ready"; requestId: string; brokerUid: 2100; providerProxyPort: 43123; mcpFacadePort: 43124; registrations: number; credentialStale: false; realmLease: true; workerIsolation: true }> | Readonly<{ version: typeof VERSION; kind: "accepted"; requestId: string; turnId: string }> | (Readonly<{ version: typeof VERSION; kind: "completed"; requestId: string; turnId: string; text: string; workerPid: number; workerUid: number; workerStartTime: string }> & EngineBrokerTurnAccounting) - | (Readonly<{ version: typeof VERSION; kind: "failed"; requestId: string; turnId: string; code: EngineBrokerFailureCode; diagnostic?: EngineBrokerFailureDiagnostic }> & EngineBrokerTurnAccounting); + | (Readonly<{ version: typeof VERSION; kind: "failed"; requestId: string; turnId: string; code: EngineBrokerFailureCode; diagnostic?: EngineBrokerFailureDiagnostic }> & EngineBrokerTurnAccounting) + | EngineBrokerInferenceResponse; export const ENGINE_BROKER_FAILURE_CODES = ["auth_stale", "cancelled", "engine_failed", "invalid_request", "limit_exceeded", "turn_conflict", "unavailable"] as const; export type EngineBrokerFailureCode = (typeof ENGINE_BROKER_FAILURE_CODES)[number]; export type EngineBrokerTerminalResponse = Extract; @@ -63,6 +66,7 @@ export function parseEngineBrokerRequest(value: unknown): EngineBrokerRequest { exact(input, ["version", "kind", "requestId", "turnId"]); return { version: VERSION, kind: "cancel_turn", requestId: id(input.requestId), turnId: id(input.turnId) }; } + if (isEngineBrokerInferenceRequestKind(input.kind)) return parseEngineBrokerInferenceRequest(input, id(input.requestId), VERSION); throw new TypeError("invalid broker frame"); } @@ -74,6 +78,7 @@ export function parseEngineBrokerResponse(value: unknown): EngineBrokerResponse return { version: VERSION, kind: "accepted", requestId: id(input.requestId), turnId: id(input.turnId) }; } if (input.kind === "completed" || input.kind === "failed") return parseTerminal(input, VERSION) as EngineBrokerTerminalResponse; + if (isEngineBrokerInferenceResponseKind(input.kind)) return parseEngineBrokerInferenceResponse(input, id(input.requestId), VERSION); throw new TypeError("invalid broker frame"); } diff --git a/src/runtime/engineBrokerService.ts b/src/runtime/engineBrokerService.ts index 2b3f4c5..aef6e33 100644 --- a/src/runtime/engineBrokerService.ts +++ b/src/runtime/engineBrokerService.ts @@ -3,10 +3,15 @@ import { createServer, type Server, type Socket } from "node:net"; import { encodeEngineBrokerFrame,EngineBrokerFrameDecoder,parseEngineBrokerRequest,type EngineBrokerResponse } from "./engineBrokerProtocol.js"; import type { EngineBrokerTurnAccounting, EngineBrokerTurnLimitOverrides } from "./engineBrokerTurnAccounting.js"; import { EngineBrokerTurnFailure } from "./grokEngineBroker.js"; +import { GROK_INFERENCE_PROXY_BASE_URL, type EngineBrokerInferenceRequest } from "./engineBrokerInferenceProtocol.js"; +import { GrokInferenceGrantRefused, type GrokInferenceGrantIssued, type GrokInferenceGrantRequest } from "./grokInferenceGrants.js"; export interface EngineBrokerServiceEngine { turn(agentId:string,wakeId:string,prompt:string,mcpEndpoint:string,signal?:AbortSignal,limits?:EngineBrokerTurnLimitOverrides):Promise&EngineBrokerTurnAccounting>; readiness():Readonly<{providerProxyPort:number;mcpFacadePort:number;registrations:number;credentialStale:boolean;realmLease:boolean;workerIsolation:boolean}>; close():Promise; + /** Evaluator inference grants; an engine without them refuses every grant request as `unavailable`. Refusals throw {@link GrokInferenceGrantRefused}. */ + requestInferenceGrant?(request:GrokInferenceGrantRequest):GrokInferenceGrantIssued; + releaseInferenceGrant?(grantId:string):boolean; } export function startEngineBrokerService(broker:EngineBrokerServiceEngine,socketPath="/run/daimon-engine-broker/backend.sock"){ @@ -37,7 +42,7 @@ export async function startEngineBrokerServiceWithIdentity(broker:EngineBrokerSe * reader has disconnected. */ function handleSocketError(socket:Socket,owned:()=>Readonly<{turnId:string;controller:AbortController}>|undefined):void{socket.on("error",()=>{owned()?.controller.abort();socket.destroy();});} -function handle(socket:Socket,broker:EngineBrokerServiceEngine,active:Map):void{const decoder=new EngineBrokerFrameDecoder();let started=false,owned:Readonly<{turnId:string;controller:AbortController}>|undefined;handleSocketError(socket,()=>owned);socket.once("close",()=>owned?.controller.abort());socket.on("data",(chunk)=>{try{for(const value of decoder.push(chunk)){const request=parseEngineBrokerRequest(value);if(request.kind==="health"){if(started)throw new Error();started=true;const ready=broker.readiness();if(ready.providerProxyPort!==43123||ready.mcpFacadePort!==43124||ready.registrations<1||ready.credentialStale||!ready.realmLease||!ready.workerIsolation)throw new Error();return send(socket,{version:request.version,kind:"ready",requestId:request.requestId,brokerUid:2100,providerProxyPort:43123,mcpFacadePort:43124,registrations:ready.registrations,credentialStale:false,realmLease:true,workerIsolation:true});}if(request.kind==="cancel_turn"){active.get(request.turnId)?.abort();continue;}if(started||active.has(request.turnId))throw new Error();started=true;const controller=new AbortController();owned={turnId:request.turnId,controller};active.set(request.turnId,controller);socket.write(encodeEngineBrokerFrame({version:request.version,kind:"accepted",requestId:request.requestId,turnId:request.turnId}));void broker.turn(request.agentId,request.wakeId,request.prompt,request.mcpEndpoint,controller.signal,request.limits).then((result)=>send(socket,{version:request.version,kind:"completed",requestId:request.requestId,turnId:request.turnId,text:result.text,workerPid:result.workerPid,workerUid:result.workerUid,workerStartTime:result.workerStartTime,outcome:"completed",usage:result.usage,model:result.model,requests:result.requests,limitReason:"none"}),(error:unknown)=>failed(socket,request,error,controller.signal.aborted)).finally(()=>{if(active.get(request.turnId)===controller)active.delete(request.turnId);owned=undefined;});}}catch{socket.destroy();}});} +function handle(socket:Socket,broker:EngineBrokerServiceEngine,active:Map):void{const decoder=new EngineBrokerFrameDecoder();let started=false,owned:Readonly<{turnId:string;controller:AbortController}>|undefined;handleSocketError(socket,()=>owned);socket.once("close",()=>owned?.controller.abort());socket.on("data",(chunk)=>{try{for(const value of decoder.push(chunk)){const request=parseEngineBrokerRequest(value);if(request.kind==="health"){if(started)throw new Error();started=true;const ready=broker.readiness();if(ready.providerProxyPort!==43123||ready.mcpFacadePort!==43124||ready.registrations<1||ready.credentialStale||!ready.realmLease||!ready.workerIsolation)throw new Error();return send(socket,{version:request.version,kind:"ready",requestId:request.requestId,brokerUid:2100,providerProxyPort:43123,mcpFacadePort:43124,registrations:ready.registrations,credentialStale:false,realmLease:true,workerIsolation:true});}if(request.kind==="cancel_turn"){active.get(request.turnId)?.abort();continue;}if(request.kind==="request_inference_grant"||request.kind==="release_inference_grant"){if(started)throw new Error();started=true;serveInferenceGrant(socket,broker,request);continue;}if(started||active.has(request.turnId))throw new Error();started=true;const controller=new AbortController();owned={turnId:request.turnId,controller};active.set(request.turnId,controller);socket.write(encodeEngineBrokerFrame({version:request.version,kind:"accepted",requestId:request.requestId,turnId:request.turnId}));void broker.turn(request.agentId,request.wakeId,request.prompt,request.mcpEndpoint,controller.signal,request.limits).then((result)=>send(socket,{version:request.version,kind:"completed",requestId:request.requestId,turnId:request.turnId,text:result.text,workerPid:result.workerPid,workerUid:result.workerUid,workerStartTime:result.workerStartTime,outcome:"completed",usage:result.usage,model:result.model,requests:result.requests,limitReason:"none"}),(error:unknown)=>failed(socket,request,error,controller.signal.aborted)).finally(()=>{if(active.get(request.turnId)===controller)active.delete(request.turnId);owned=undefined;});}}catch{socket.destroy();}});} /** * Every failure the broker raises for a known registration carries its * accounting (a wake that tried to raise a limit: `usage: null`, zero @@ -51,6 +56,16 @@ function failed(socket:Socket,request:Extractsend(socket,{version:request.version,kind:"inference_grant_refused",requestId:request.requestId,code}); + try{ + if(request.kind==="release_inference_grant"){if(!broker.releaseInferenceGrant)return refuse("unavailable");return send(socket,{version:request.version,kind:"inference_grant_released",requestId:request.requestId,grantId:request.grantId,released:broker.releaseInferenceGrant(request.grantId)});} + if(!broker.requestInferenceGrant)return refuse("unavailable"); + const grant=broker.requestInferenceGrant({model:request.model,reasoningEffort:request.reasoningEffort,purpose:request.purpose}); + send(socket,{version:request.version,kind:"inference_grant",requestId:request.requestId,grantId:grant.grantId,token:grant.token,baseUrl:GROK_INFERENCE_PROXY_BASE_URL,model:grant.policy.model,reasoningEffort:grant.policy.reasoningEffort,purpose:grant.purpose,expiresAt:new Date(grant.expiresAt).toISOString(),limits:grant.limits}); + }catch(error){refuse(error instanceof GrokInferenceGrantRefused?error.code:"unavailable");} +} function send(socket:Socket,response:EngineBrokerResponse):void{if(!socket.destroyed)socket.end(encodeEngineBrokerFrame(response));} async function removeOwnedSocket(file:string,uid:number):Promise{try{const entry=await lstat(file);if(!entry.isSocket()||Number(entry.uid)!==uid)throw new Error("unsafe broker socket");await unlink(file);}catch(error){if((error as NodeJS.ErrnoException).code!=="ENOENT")throw error;}} async function verifySocket(file:string,uid:number):Promise{const entry=await lstat(file);if(!entry.isSocket()||Number(entry.uid)!==uid||(Number(entry.mode)&0o777)!==0o600)throw new Error("unsafe broker socket");} diff --git a/src/runtime/engineBrokerServiceCli.test.ts b/src/runtime/engineBrokerServiceCli.test.ts index db7771d..20078ae 100644 --- a/src/runtime/engineBrokerServiceCli.test.ts +++ b/src/runtime/engineBrokerServiceCli.test.ts @@ -53,3 +53,13 @@ test("rejects caller-selected commands, duplicate identities, and traversal", () assert.throws(() => parseEngineBrokerServiceConfig({ ...base, registrations: [{ ...reg("agent-a", 0), eventsPath: "/workers/0/.grok/sandbox-events.jsonl" }] })); assert.throws(() => parseEngineBrokerServiceConfig({ ...base, registrations: [{ ...reg("agent-a", 0), eventsPath: "/workers/1/.grok/sessions/sandbox-events.jsonl" }] })); }); + +test("v2 may declare an evaluator inference ledger that is never a subject ledger", () => { + const base = config("v2", [v2("agent-a", 0), v2("agent-b", 1)]); + assert.equal(parseEngineBrokerServiceConfig(base).inferenceLedgerPath, undefined); + assert.equal(parseEngineBrokerServiceConfig({ ...base, inferenceLedgerPath: "/run/paideia-inference/inference.jsonl" }).inferenceLedgerPath, "/run/paideia-inference/inference.jsonl"); + for (const inferenceLedgerPath of ["/run/slots/0/usage/usage.jsonl", "/run/slots/1/usage/requests.jsonl", "/var/lib/spawnfile/daimon/usage/usage.jsonl", "/var/lib/spawnfile/daimon/usage/requests.jsonl", "relative.jsonl", "/run/x/../inference.jsonl", "/run/inference.json", 7]) { + assert.throws(() => parseEngineBrokerServiceConfig({ ...base, inferenceLedgerPath }), /invalid engine broker service config/u, String(inferenceLedgerPath)); + } + assert.throws(() => parseEngineBrokerServiceConfig({ ...config("v1", [reg("agent-a", 0)]), inferenceLedgerPath: "/run/paideia-inference/inference.jsonl" }), /invalid engine broker service config/u); +}); diff --git a/src/runtime/engineBrokerServiceCli.ts b/src/runtime/engineBrokerServiceCli.ts index 2a3967a..8661b02 100644 --- a/src/runtime/engineBrokerServiceCli.ts +++ b/src/runtime/engineBrokerServiceCli.ts @@ -12,7 +12,7 @@ const MAX_CONFIG_BYTES=65_536; export async function runEngineBrokerServiceCli():Promise{ if(process.getuid?.()!==2100)throw new Error("engine broker service requires broker identity"); const config=parseEngineBrokerServiceConfig(await readRootConfig(ENGINE_BROKER_SERVICE_CONFIG)); - const broker=await startGrokEngineBroker({grokCommand:"/usr/local/bin/grok",nativeClient:"/opt/daimon/bin/daimon-engine-broker",credentialHome:config.credentialHome,turnStore:config.turnStore,registrations:config.registrations}); + const broker=await startGrokEngineBroker({grokCommand:"/usr/local/bin/grok",nativeClient:"/opt/daimon/bin/daimon-engine-broker",credentialHome:config.credentialHome,turnStore:config.turnStore,registrations:config.registrations,...(config.inferenceLedgerPath===undefined?{}:{inferenceLedgerPath:config.inferenceLedgerPath})}); const service=await startEngineBrokerService(broker);let stopping:Promise|undefined; const stop=()=>{stopping??=service.close();return stopping;}; const onSignal=()=>{void stop().catch(()=>{process.exitCode=1;});};process.once("SIGINT",onSignal);process.once("SIGTERM",onSignal); diff --git a/src/runtime/engineBrokerServiceConfig.ts b/src/runtime/engineBrokerServiceConfig.ts index 836963c..742e2a5 100644 --- a/src/runtime/engineBrokerServiceConfig.ts +++ b/src/runtime/engineBrokerServiceConfig.ts @@ -3,6 +3,7 @@ import path from "node:path"; import { DEFAULT_GROK_BROKER_TURN_LIMITS, parseEngineBrokerTurnLimits, type EngineBrokerTurnLimits } from "./engineBrokerTurnAccounting.js"; import { DEFAULT_GROK_BROKER_MODEL_POLICY, GROK_BROKER_MODELS, GROK_BROKER_REASONING_EFFORTS, type GrokBrokerModelPolicy } from "./grokBrokerModelPolicy.js"; import { grokWorkerEventsPathFor } from "./grokWorkerSandboxProfile.js"; +import { TURN_REQUEST_LEDGER } from "./turnRequestLedger.js"; import { TURN_USAGE_LEDGER } from "./turnUsageLedger.js"; export const ENGINE_BROKER_SERVICE_V1 = "noopolis.daimon.engine-broker-service.v1" as const; @@ -16,7 +17,14 @@ export type EngineBrokerServiceRegistration = Readonly<{ limits: EngineBrokerTurnLimits; model: GrokBrokerModelPolicy; }>; -export type EngineBrokerServiceConfig = Readonly<{ credentialHome: string; turnStore: string; registrations: readonly EngineBrokerServiceRegistration[] }>; +/** + * `inferenceLedgerPath` (v2, optional) is where evaluator inference grants + * append their rows (`inferenceUsageLedger.ts`). Without it the broker refuses + * every grant request. It can never be a subject ledger: not any + * registration's usage ledger or its `requests.jsonl`, and not the container + * ledger the wake fuse sums. + */ +export type EngineBrokerServiceConfig = Readonly<{ credentialHome: string; turnStore: string; registrations: readonly EngineBrokerServiceRegistration[]; inferenceLedgerPath?: string }>; const V1_REGISTRATION = ["agentId", "slot", "workerUid", "workspace", "profilePath", "eventsPath", "profileSha256"] as const; const V2_REGISTRATION = [...V1_REGISTRATION, "usageLedgerPath", "limits", "model"] as const; @@ -41,7 +49,8 @@ export function parseEngineBrokerServiceConfig(value: unknown): EngineBrokerServ if (!plain(value)) throw invalid(); const v2 = value.version === ENGINE_BROKER_SERVICE_V2; if (!v2 && value.version !== ENGINE_BROKER_SERVICE_V1) throw invalid(); - exact(value, ["version", "credentialHome", "turnStore", "registrations"]); + const top = ["version", "credentialHome", "turnStore", "registrations"]; + exact(value, v2 && Object.hasOwn(value, "inferenceLedgerPath") ? [...top, "inferenceLedgerPath"] : top); if (!absolute(value.credentialHome) || !absolute(value.turnStore) || !Array.isArray(value.registrations) || value.registrations.length === 0) throw invalid(); const seen = new Set(), slots = new Set(); const registrations = value.registrations.map((entry: unknown): EngineBrokerServiceRegistration => { @@ -53,14 +62,22 @@ export function parseEngineBrokerServiceConfig(value: unknown): EngineBrokerServ const base = { agentId, slot: slot as number, workerUid: workerUid as number, workspace, profilePath, eventsPath, profileSha256 }; if (!v2) return { ...base, usageLedgerPath: TURN_USAGE_LEDGER.filePath, limits: DEFAULT_GROK_BROKER_TURN_LIMITS, model: DEFAULT_GROK_BROKER_MODEL_POLICY }; const usageLedgerPath = entry.usageLedgerPath; - if (!absolute(usageLedgerPath) || !usageLedgerPath.endsWith(".jsonl") || usageLedgerPath === engineBrokerRequestLedgerPathFor(usageLedgerPath) || path.posix.normalize(usageLedgerPath) !== usageLedgerPath) throw invalid(); + if (!ledgerPath(usageLedgerPath) || usageLedgerPath === engineBrokerRequestLedgerPathFor(usageLedgerPath)) throw invalid(); let limits: EngineBrokerTurnLimits; try { limits = parseEngineBrokerTurnLimits(entry.limits); } catch { throw invalid(); } return { ...base, usageLedgerPath, limits, model: parseServiceModel(entry.model) }; }); - return { credentialHome: value.credentialHome, turnStore: value.turnStore, registrations }; + const base = { credentialHome: value.credentialHome, turnStore: value.turnStore, registrations }; + if (!Object.hasOwn(value, "inferenceLedgerPath")) return base; + const inferenceLedgerPath = value.inferenceLedgerPath; + if (!ledgerPath(inferenceLedgerPath)) throw invalid(); + const subject = new Set([TURN_USAGE_LEDGER.filePath, TURN_REQUEST_LEDGER.filePath, ...registrations.flatMap((entry) => [entry.usageLedgerPath, engineBrokerRequestLedgerPathFor(entry.usageLedgerPath)])]); + if (subject.has(inferenceLedgerPath)) throw invalid(); + return { ...base, inferenceLedgerPath }; } +const ledgerPath = (item: unknown): item is string => absolute(item) && item.endsWith(".jsonl") && path.posix.normalize(item) === item; + function parseServiceModel(value: unknown): GrokBrokerModelPolicy { if (!plain(value)) throw invalid(); exact(value, ["id", "reasoningEffort"]); diff --git a/src/runtime/fixtures/grok-slot-preflight/receipt.extra-canary.json b/src/runtime/fixtures/grok-slot-preflight/receipt.extra-canary.json index f694edb..a1f00ca 100644 --- a/src/runtime/fixtures/grok-slot-preflight/receipt.extra-canary.json +++ b/src/runtime/fixtures/grok-slot-preflight/receipt.extra-canary.json @@ -1,7 +1,9 @@ { - "version": "noopolis.daimon.grok-slot-preflight.v1", + "version": "noopolis.daimon.grok-slot-preflight.v2", "slot": 0, "worker_uid": 2200, + "generation": 3, + "nonce": "5f0e2a1c9b8d7e6f5a4b3c2d1e0f9a8b7c6d5e4f3a2b1c0d9e8f7a6b5c4d3e2f", "projection_sha256": "9e8bbd4e9337df40b90d5a382ece6c3745cb5e11038d10ea17b5032408bf30e4", "sandbox_profile_sha256": "2a1dcab471092eab77565838b17a5cf138489e3daa1349612ebcedcdb69e23c4", "seccomp_profile_sha256": "7777777777777777777777777777777777777777777777777777777777777777", diff --git a/src/runtime/fixtures/grok-slot-preflight/receipt.valid.v1.json b/src/runtime/fixtures/grok-slot-preflight/receipt.legacy-v1.json similarity index 100% rename from src/runtime/fixtures/grok-slot-preflight/receipt.valid.v1.json rename to src/runtime/fixtures/grok-slot-preflight/receipt.legacy-v1.json diff --git a/src/runtime/fixtures/grok-slot-preflight/receipt.missing-canary.json b/src/runtime/fixtures/grok-slot-preflight/receipt.missing-canary.json index 974b321..0b57369 100644 --- a/src/runtime/fixtures/grok-slot-preflight/receipt.missing-canary.json +++ b/src/runtime/fixtures/grok-slot-preflight/receipt.missing-canary.json @@ -1,7 +1,9 @@ { - "version": "noopolis.daimon.grok-slot-preflight.v1", + "version": "noopolis.daimon.grok-slot-preflight.v2", "slot": 0, "worker_uid": 2200, + "generation": 3, + "nonce": "5f0e2a1c9b8d7e6f5a4b3c2d1e0f9a8b7c6d5e4f3a2b1c0d9e8f7a6b5c4d3e2f", "projection_sha256": "9e8bbd4e9337df40b90d5a382ece6c3745cb5e11038d10ea17b5032408bf30e4", "sandbox_profile_sha256": "2a1dcab471092eab77565838b17a5cf138489e3daa1349612ebcedcdb69e23c4", "seccomp_profile_sha256": "7777777777777777777777777777777777777777777777777777777777777777", diff --git a/src/runtime/fixtures/grok-slot-preflight/receipt.projection-mismatch.json b/src/runtime/fixtures/grok-slot-preflight/receipt.projection-mismatch.json index c8729e3..fd520ed 100644 --- a/src/runtime/fixtures/grok-slot-preflight/receipt.projection-mismatch.json +++ b/src/runtime/fixtures/grok-slot-preflight/receipt.projection-mismatch.json @@ -1,7 +1,9 @@ { - "version": "noopolis.daimon.grok-slot-preflight.v1", + "version": "noopolis.daimon.grok-slot-preflight.v2", "slot": 0, "worker_uid": 2200, + "generation": 3, + "nonce": "5f0e2a1c9b8d7e6f5a4b3c2d1e0f9a8b7c6d5e4f3a2b1c0d9e8f7a6b5c4d3e2f", "projection_sha256": "0000000000000000000000000000000000000000000000000000000000000000", "sandbox_profile_sha256": "2a1dcab471092eab77565838b17a5cf138489e3daa1349612ebcedcdb69e23c4", "seccomp_profile_sha256": "7777777777777777777777777777777777777777777777777777777777777777", diff --git a/src/runtime/fixtures/grok-slot-preflight/receipt.readable-canary.json b/src/runtime/fixtures/grok-slot-preflight/receipt.readable-canary.json index 8f5d407..19cb96f 100644 --- a/src/runtime/fixtures/grok-slot-preflight/receipt.readable-canary.json +++ b/src/runtime/fixtures/grok-slot-preflight/receipt.readable-canary.json @@ -1,7 +1,9 @@ { - "version": "noopolis.daimon.grok-slot-preflight.v1", + "version": "noopolis.daimon.grok-slot-preflight.v2", "slot": 0, "worker_uid": 2200, + "generation": 3, + "nonce": "5f0e2a1c9b8d7e6f5a4b3c2d1e0f9a8b7c6d5e4f3a2b1c0d9e8f7a6b5c4d3e2f", "projection_sha256": "9e8bbd4e9337df40b90d5a382ece6c3745cb5e11038d10ea17b5032408bf30e4", "sandbox_profile_sha256": "2a1dcab471092eab77565838b17a5cf138489e3daa1349612ebcedcdb69e23c4", "seccomp_profile_sha256": "7777777777777777777777777777777777777777777777777777777777777777", diff --git a/src/runtime/fixtures/grok-slot-preflight/receipt.unknown-member.json b/src/runtime/fixtures/grok-slot-preflight/receipt.unknown-member.json index 2e50e6a..91774b2 100644 --- a/src/runtime/fixtures/grok-slot-preflight/receipt.unknown-member.json +++ b/src/runtime/fixtures/grok-slot-preflight/receipt.unknown-member.json @@ -1,7 +1,9 @@ { - "version": "noopolis.daimon.grok-slot-preflight.v1", + "version": "noopolis.daimon.grok-slot-preflight.v2", "slot": 0, "worker_uid": 2200, + "generation": 3, + "nonce": "5f0e2a1c9b8d7e6f5a4b3c2d1e0f9a8b7c6d5e4f3a2b1c0d9e8f7a6b5c4d3e2f", "projection_sha256": "9e8bbd4e9337df40b90d5a382ece6c3745cb5e11038d10ea17b5032408bf30e4", "sandbox_profile_sha256": "2a1dcab471092eab77565838b17a5cf138489e3daa1349612ebcedcdb69e23c4", "seccomp_profile_sha256": "7777777777777777777777777777777777777777777777777777777777777777", diff --git a/src/runtime/fixtures/grok-slot-preflight/receipt.valid.v2.json b/src/runtime/fixtures/grok-slot-preflight/receipt.valid.v2.json new file mode 100644 index 0000000..94542eb --- /dev/null +++ b/src/runtime/fixtures/grok-slot-preflight/receipt.valid.v2.json @@ -0,0 +1,40 @@ +{ + "version": "noopolis.daimon.grok-slot-preflight.v2", + "slot": 0, + "worker_uid": 2200, + "generation": 3, + "nonce": "5f0e2a1c9b8d7e6f5a4b3c2d1e0f9a8b7c6d5e4f3a2b1c0d9e8f7a6b5c4d3e2f", + "projection_sha256": "9e8bbd4e9337df40b90d5a382ece6c3745cb5e11038d10ea17b5032408bf30e4", + "sandbox_profile_sha256": "2a1dcab471092eab77565838b17a5cf138489e3daa1349612ebcedcdb69e23c4", + "seccomp_profile_sha256": "7777777777777777777777777777777777777777777777777777777777777777", + "sandbox_runtime": "bubblewrap", + "grok_executable_sha256": "39ab87666877d64ef3a40aa60fbe0c3b6a6acd7001b78fe60e2c76bb6cfc4a94", + "canaries": [ + { + "path": "/run/paideia", + "method": "sandboxed-read", + "result": "denied" + }, + { + "path": "/run/paideia/control", + "method": "sandboxed-read", + "result": "denied" + }, + { + "path": "/run/training/inputs", + "method": "sandboxed-read", + "result": "denied" + }, + { + "path": "/var/lib/spawnfile/daimon/grok-bootstrap-auth", + "method": "sandboxed-read", + "result": "denied" + }, + { + "path": "/var/lib/spawnfile/daimon/grok-subscription-realm", + "method": "sandboxed-read", + "result": "denied" + } + ], + "created_at": "2026-09-17T12:00:00.000Z" +} diff --git a/src/runtime/grokBrokerProxy.ts b/src/runtime/grokBrokerProxy.ts index 187bf7a..4d1e6d0 100644 --- a/src/runtime/grokBrokerProxy.ts +++ b/src/runtime/grokBrokerProxy.ts @@ -5,11 +5,14 @@ import { EngineBrokerCapabilities } from "./engineBrokerCapabilities.js"; import { DEFAULT_GROK_BROKER_MODEL_POLICY, parseGrokBrokerModelPolicy, type GrokBrokerModelPolicy } from "./grokBrokerModelPolicy.js"; import { authorizeGrokBrokerProxyRequest } from "./grokBrokerProxyRequest.js"; import { parseGrokUpstreamUsage, type GrokBrokerTurnMeter } from "./grokBrokerTurnMeter.js"; +import { GROK_ENGINE_BROKER } from "../contracts/runtimeContractManifest.js"; +import { serveGrokInferenceGrant } from "./grokInferenceProxy.js"; +import type { GrokInferenceGrants } from "./grokInferenceGrants.js"; /** One running turn as the proxy sees it: its declared model/effort and its spend gate. */ export type GrokBrokerProxyTurn = Readonly<{ policy: GrokBrokerModelPolicy; meter: GrokBrokerTurnMeter }>; -export type GrokBrokerCredentialAuthority = Readonly<{ accessToken(forceRefresh: boolean): Promise; refreshAfterRejection?(rejectedTokenDigest:string):Promise; markRejected(rejectedTokenDigest?:string): Promise }>; +export type GrokBrokerCredentialAuthority = Readonly<{ accessToken(forceRefresh: boolean): Promise; refreshAfterRejection?(rejectedTokenDigest:string):Promise; markRejected(rejectedTokenDigest?:string): Promise; isStale?(): boolean }>; export type GrokBrokerUpstream = (request: ReturnType, signal?: AbortSignal) => Promise>; body: Uint8Array }>>; /** @@ -17,20 +20,27 @@ export type GrokBrokerUpstream = (request: ReturnTypePromise):void; revokeIsolationGuard(turnId:string):void; registerTurn(turnId:string,turn:GrokBrokerProxyTurn):void; revokeTurn(turnId:string):void; close(): Promise }>> { +export async function startGrokBrokerProxy(authority: GrokBrokerCredentialAuthority, upstream: GrokBrokerUpstream = defaultUpstream, policy: GrokBrokerModelPolicy = DEFAULT_GROK_BROKER_MODEL_POLICY, listenPort = 43_123, grants?: GrokInferenceGrants): PromisePromise):void; revokeIsolationGuard(turnId:string):void; registerTurn(turnId:string,turn:GrokBrokerProxyTurn):void; revokeTurn(turnId:string):void; close(): Promise }>> { const declared = parseGrokBrokerModelPolicy(policy); const capabilities = new EngineBrokerCapabilities(); - const guards=new MapPromise>();const turns=new Map();const server = createServer((request, response) => { void serve(request, response, authority, upstream, capabilities,guards,turns,declared); }); + const guards=new MapPromise>();const turns=new Map();const server = createServer((request, response) => { void serve(request, response, authority, upstream, capabilities,guards,turns,declared,grants); }); await new Promise((resolve, reject) => { server.once("error", reject); server.listen(listenPort, "127.0.0.1", () => { server.off("error", reject); resolve(); }); }); const address = server.address() as AddressInfo; return { port: address.port, capabilities,registerIsolationGuard(turnId,guard){guards.set(turnId,guard);},revokeIsolationGuard(turnId){guards.delete(turnId);},registerTurn(turnId,turn){turns.set(turnId,{policy:parseGrokBrokerModelPolicy(turn.policy),meter:turn.meter});},revokeTurn(turnId){turns.delete(turnId);}, close: () => new Promise((resolve, reject) => server.close((error) => error === undefined ? resolve() : reject(error))) }; } -async function serve(request: IncomingMessage, response: ServerResponse, authority: GrokBrokerCredentialAuthority, upstream: GrokBrokerUpstream, capabilities: EngineBrokerCapabilities,guards:MapPromise>,turns:Map,fallback:GrokBrokerModelPolicy): Promise { +async function serve(request: IncomingMessage, response: ServerResponse, authority: GrokBrokerCredentialAuthority, upstream: GrokBrokerUpstream, capabilities: EngineBrokerCapabilities,guards:MapPromise>,turns:Map,fallback:GrokBrokerModelPolicy,grants?:GrokInferenceGrants): Promise { let settle:((usage:ReturnType)=>void)|undefined; try { const body = await readBody(request); const headers = Object.fromEntries(Object.entries(request.headers).map(([key, value]) => [key, Array.isArray(value) ? value[0] : value])); - const match=headers.authorization?.match(/^Bearer ([A-Za-z0-9_-]{40,})$/u),scope=match?capabilities.inspectToken(match[1]!):undefined;if(!scope)throw new Error();const guard=guards.get(scope.turnId),turn=turns.get(scope.turnId);if(!guard||!turn)throw new Error();await guard(); + const match=headers.authorization?.match(/^Bearer ([A-Za-z0-9_-]{40,})$/u); + if(match&&match[1]!.startsWith(GROK_ENGINE_BROKER.inferenceGrants.tokenPrefix)){if(!grants)throw new Error();return await serveGrokInferenceGrant({method:request.method??"",pathname:new URL(request.url??"/","http://127.0.0.1").pathname,headers,body,token:match[1]!},response,grants,authority,upstream);} + const scope=match?capabilities.inspectToken(match[1]!):undefined;if(!scope)throw new Error();const guard=guards.get(scope.turnId),turn=turns.get(scope.turnId);if(!guard||!turn)throw new Error();await guard(); let token = await authority.accessToken(false);const rejectedDigest=createHash("sha256").update(token).digest("hex"); let prepared = authorizeGrokBrokerProxyRequest({ method: request.method ?? "", pathname: new URL(request.url ?? "/", "http://127.0.0.1").pathname, headers, body }, capabilities, token, turn.policy ?? fallback); token = ""; // The spend gate runs after the body is proven a real lean worker request // (a refused session-title body never counts) and before any upstream call. diff --git a/src/runtime/grokEngineBroker.ts b/src/runtime/grokEngineBroker.ts index d6d5125..aceb394 100644 --- a/src/runtime/grokEngineBroker.ts +++ b/src/runtime/grokEngineBroker.ts @@ -10,6 +10,7 @@ import { grokBrokerWorkerConfigSha256 } from "./grokBrokerWorkerConfig.js"; import { parseGrokBrokerModelPolicy } from "./grokBrokerModelPolicy.js"; import { runGrokEngineBrokerTurn, type GrokEngineBrokerTurnResult } from "./grokEngineBrokerTurn.js"; import { createGrokWorkerIsolationGuard,prepareGrokWorkerAttestation } from "./grokWorkerAttestation.js"; +import { createLedgeredGrokInferenceGrants, GrokInferenceGrantRefused, type GrokInferenceGrantRequest } from "./grokInferenceGrants.js"; export { EngineBrokerTurnFailure, type GrokEngineBrokerTurnResult } from "./grokEngineBrokerTurn.js"; export { finishBrokerTurnWithUsage } from "./grokEngineBrokerMetering.js"; @@ -21,11 +22,17 @@ export type GrokEngineBroker = Awaited> * facade, and the root-provisioned registrations. Each registration declares * its own model/effort (whose worker config bytes are attested), usage ledger, * and turn limits (`engineBrokerServiceConfig.ts`). + * + * With an `inferenceLedgerPath` the broker also issues evaluator inference + * grants (`grokInferenceGrants.ts`) over the same credential authority and + * proxy; their rows go only to that ledger. A stale realm refuses a grant as + * `auth_stale`, exactly as it fails subject turns. */ -export async function startGrokEngineBroker(options: Readonly<{ grokCommand: string; nativeClient: string; credentialHome: string; turnStore: string; registrations: readonly GrokEngineBrokerRegistration[] }>) { +export async function startGrokEngineBroker(options: Readonly<{ grokCommand: string; nativeClient: string; credentialHome: string; turnStore: string; registrations: readonly GrokEngineBrokerRegistration[]; inferenceLedgerPath?: string }>) { const registrations = new Map(options.registrations.map((entry) => [entry.agentId, { ...entry, model: parseGrokBrokerModelPolicy(entry.model) }])); if (registrations.size !== options.registrations.length) throw new Error("engine broker registration conflict"); const attestationFor = (registration: GrokEngineBrokerRegistration) => ({ ...registration, brokerGid: 2100, configSha256: grokBrokerWorkerConfigSha256(registration.model) }); - const lease=await acquireGrokBrokerRealmLease(options.credentialHome);const authority = new DurableGrokBrokerCredentialAuthority(options.grokCommand, options.credentialHome);try{await authority.initialize();}catch(error){await lease.close();throw error;} let proxy:Awaited>;try{proxy=await startGrokBrokerProxy(authority);}catch(error){await lease.close();throw error;}let mcp:Awaited>|undefined;try{mcp=await startEngineBrokerMcpFacade();for(const registration of registrations.values())await prepareGrokWorkerAttestation(attestationFor(registration));}catch(error){if(mcp)await mcp.close().catch(()=>undefined);await proxy.close();await lease.close();throw error;}if(!mcp)throw new Error("engine broker unavailable");const turns = new EngineBrokerTurnRegistry(options.turnStore); const active = new Map }>(); let closed = false; + const inferenceLedgerPath=options.inferenceLedgerPath;const grants=inferenceLedgerPath===undefined?undefined:createLedgeredGrokInferenceGrants(inferenceLedgerPath); + const lease=await acquireGrokBrokerRealmLease(options.credentialHome);const authority = new DurableGrokBrokerCredentialAuthority(options.grokCommand, options.credentialHome);try{await authority.initialize();}catch(error){await lease.close();throw error;} let proxy:Awaited>;try{proxy=await startGrokBrokerProxy(authority,undefined,undefined,undefined,grants);}catch(error){await lease.close();throw error;}let mcp:Awaited>|undefined;try{mcp=await startEngineBrokerMcpFacade();for(const registration of registrations.values())await prepareGrokWorkerAttestation(attestationFor(registration));}catch(error){if(mcp)await mcp.close().catch(()=>undefined);await proxy.close();await lease.close();throw error;}if(!mcp)throw new Error("engine broker unavailable");const turns = new EngineBrokerTurnRegistry(options.turnStore); const active = new Map }>(); let closed = false; const facade = mcp; const deps = { turns, proxy, mcp: facade, credentialStale: () => authority.isStale(), @@ -40,7 +47,16 @@ export async function startGrokEngineBroker(options: Readonly<{ grokCommand: str active.set(key, { controller, done: done.then(() => undefined, () => undefined) }); try { return await done; } finally { signal?.removeEventListener("abort", onAbort); active.delete(key); } }, - async close(): Promise { if (closed) return; closed = true; const running=[...active.values()];for (const entry of running) entry.controller.abort();await Promise.allSettled(running.map((entry)=>entry.done));const results=await Promise.allSettled([facade.close(),proxy.close(),lease.close()]);const failures=results.flatMap((entry)=>entry.status==="rejected"?[entry.reason]:[]);if(failures.length)throw new AggregateError(failures,"engine broker shutdown failed"); }, + requestInferenceGrant(request: GrokInferenceGrantRequest) { + if (closed || grants === undefined) throw new GrokInferenceGrantRefused("unavailable"); + if (authority.isStale()) throw new GrokInferenceGrantRefused("auth_stale"); + return grants.issue(request); + }, + releaseInferenceGrant(grantId: string): boolean { + if (grants === undefined) throw new GrokInferenceGrantRefused("unavailable"); + return grants.release(grantId); + }, + async close(): Promise { if (closed) return; closed = true; grants?.close(); const running=[...active.values()];for (const entry of running) entry.controller.abort();await Promise.allSettled(running.map((entry)=>entry.done));const results=await Promise.allSettled([facade.close(),proxy.close(),lease.close()]);const failures=results.flatMap((entry)=>entry.status==="rejected"?[entry.reason]:[]);if(failures.length)throw new AggregateError(failures,"engine broker shutdown failed"); }, readiness: () => ({ providerProxyPort: proxy.port, mcpFacadePort:43_124, registrations: registrations.size,credentialStale:authority.isStale(),realmLease:true,workerIsolation:true }) }; } diff --git a/src/runtime/grokInferenceClientConfig.test.ts b/src/runtime/grokInferenceClientConfig.test.ts new file mode 100644 index 0000000..d9ee6df --- /dev/null +++ b/src/runtime/grokInferenceClientConfig.test.ts @@ -0,0 +1,43 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { GROK_ENGINE_BROKER } from "../contracts/runtimeContractManifest.js"; +import { GROK_BROKER_MODELS, GROK_BROKER_REASONING_EFFORTS } from "./grokBrokerModelPolicy.js"; +import { GROK_1_0_34_BUNDLED_SKILLS, GROK_SESSION_TITLE_SINK_KEY } from "./grokBrokerWorkerConfig.js"; +import { GROK_INFERENCE_CLIENT_MODEL_ID, GROK_INFERENCE_GRANT_ENV, grokInferenceClientConfigSha256, renderGrokInferenceClientConfig, renderProductionGrokInferenceClientConfig } from "./grokInferenceClientConfig.js"; + +const production = { baseUrl: "http://127.0.0.1:43123/v1", model: "grok-4.6", reasoningEffort: "low", envKey: "DAIMON_INFERENCE_GRANT" } as const; + +test("the evaluator client config reaches only the grant proxy through env_key, with no MCP and no credential", () => { + const config = renderGrokInferenceClientConfig(production); + assert.match(config, /\[models\]\ndefault = "daimon-inference-grok"\ndefault_reasoning_effort = "low"\nsession_summary = "daimon-session-title-disabled"\n/u); + assert.match(config, /\[model\.daimon-inference-grok\]\nmodel = "grok-4\.6"\nbase_url = "http:\/\/127\.0\.0\.1:43123\/v1"\nenv_key = "DAIMON_INFERENCE_GRANT"\napi_backend = "chat_completions"\ncontext_window = 131072\nsupports_backend_search = false\nmax_retries = 0\n/u); + assert.match(config, /\[\[model\.daimon-inference-grok\.reasoning_efforts\]\]\nvalue = "low"\nlabel = "Low"\ndefault = true\n/u); + assert.equal(config.match(/reasoning_efforts\]\]/gu)?.length, 1); + assert.doesNotMatch(config, /mcp_servers|auth_provider|access_token|refresh_token/u); + assert.equal(GROK_INFERENCE_CLIENT_MODEL_ID, "daimon-inference-grok"); assert.equal(GROK_INFERENCE_GRANT_ENV, "DAIMON_INFERENCE_GRANT"); +}); + +test("the evaluator client config mirrors the worker's lean settings and refuses the session title locally", () => { + const config = renderGrokInferenceClientConfig(production); + for (const skill of GROK_1_0_34_BUNDLED_SKILLS) assert.ok(config.includes(JSON.stringify(skill)), skill); + assert.match(config, /\[workflows\]\nenabled = false\n/u); assert.match(config, /\[managed_mcps\]\nenabled = false\n/u); assert.match(config, /auto_update = false/u); + assert.match(config, new RegExp(`\\[model\\.daimon-session-title-disabled\\]\\nmodel = "disabled"\\nbase_url = "http://127\\.0\\.0\\.1:43123/v1"\\napi_key = "${GROK_SESSION_TITLE_SINK_KEY}"\\nmax_retries = 0\\nhidden = true\\n`, "u")); + assert.ok(GROK_SESSION_TITLE_SINK_KEY.length < 40, "the placeholder can never pass the proxy bearer shape"); +}); + +test("the manifest pins the sha256 of every production evaluator client config", () => { + for (const model of GROK_BROKER_MODELS) for (const reasoningEffort of GROK_BROKER_REASONING_EFFORTS) { + const digest = grokInferenceClientConfigSha256({ ...production, model, reasoningEffort }); + assert.equal(GROK_ENGINE_BROKER.inferenceGrants.client.configSha256[model][reasoningEffort], digest, `${model}/${reasoningEffort}`); + assert.equal(renderProductionGrokInferenceClientConfig({ model, reasoningEffort }), renderGrokInferenceClientConfig({ ...production, model, reasoningEffort })); + } +}); + +test("the evaluator client config refuses non-loopback endpoints, injected keys and undeclared models", () => { + for (const bad of [ + { ...production, baseUrl: "https://cli-chat-proxy.grok.com/v1" }, { ...production, baseUrl: "http://127.0.0.1:43123/v1\"\nx = 1" }, { ...production, baseUrl: "http://127.0.0.1:99999/v1" }, + { ...production, envKey: "X\"\n[mcp_servers.evil]" }, { ...production, envKey: "lower" }, + { ...production, model: "grok-3" }, { ...production, reasoningEffort: "xhigh" } + ]) assert.throws(() => renderGrokInferenceClientConfig(bad as typeof production), /invalid Grok inference client configuration/u); +}); diff --git a/src/runtime/grokInferenceClientConfig.ts b/src/runtime/grokInferenceClientConfig.ts new file mode 100644 index 0000000..d9a043c --- /dev/null +++ b/src/runtime/grokInferenceClientConfig.ts @@ -0,0 +1,63 @@ +import { createHash } from "node:crypto"; + +import { GROK_ENGINE_BROKER } from "../contracts/runtimeContractManifest.js"; +import { GROK_INFERENCE_PROXY_BASE_URL } from "./engineBrokerInferenceProtocol.js"; +import { parseGrokBrokerModelPolicy, type GrokBrokerModelPolicy } from "./grokBrokerModelPolicy.js"; +import { GROK_SESSION_TITLE_SINK_KEY, GROK_SESSION_TITLE_SINK_MODEL_ID, renderGrokLeanBaseConfig } from "./grokBrokerWorkerConfig.js"; + +const SPEC = GROK_ENGINE_BROKER.inferenceGrants.client; +/** The evaluator CLI's only model id: Paideia passes `--model daimon-inference-grok`, never a catalog id. */ +export const GROK_INFERENCE_CLIENT_MODEL_ID = SPEC.modelId; +/** The environment variable the evaluator CLI reads its grant token from. */ +export const GROK_INFERENCE_GRANT_ENV = SPEC.envKey; + +export type GrokInferenceClientConfigInput = Readonly<{ baseUrl: string; model: GrokBrokerModelPolicy["model"]; reasoningEffort: GrokBrokerModelPolicy["reasoningEffort"]; envKey: string }>; + +/** + * `config.toml` bytes for an evaluator Grok CLI (Paideia judge or optimizer, + * uid 2000) that reaches the broker proxy through an inference grant. + * + * Paideia writes it into a private `GROK_HOME` (`0700`, uid 2000) and sets the + * grant token in `envKey`; the CLI never holds the broker credential. Grok + * 1.0.34 ignores `[auth_provider.*]` helpers for a custom model, so the token + * travels through `env_key` exactly as the worker's turn capability does. + * + * It mirrors the worker renderer's lean settings: every bundled skill + * disabled, workflows off, the per-call `session_title` request pointed at a + * hidden model with a placeholder key the proxy refuses before any credential + * read or upstream call, and the declared effort as the model's single + * effort. There is no MCP server. `max_retries = 0`: with the default, a + * refused request (HTTP 503) is retried with backoff past a 45 s bound + * (live stub capture), so a gate refusal would hang the judge until its own + * timeout; with it the CLI fails in ~0.35 s and the caller's routed retry + * policy decides. HTTP 401 is never retried either way. `baseUrl` is the `baseUrl` of the grant (the + * loopback provider proxy); the manifest pins the sha256 for the production + * proxy URL and {@link GROK_INFERENCE_GRANT_ENV}. + * + * Paideia must also accept the init frame this produces: `apiKeySource` is + * `"user"` (not `"oauth"`), with `tools: []` and `mcp_servers: []` (live + * stub capture with the Paideia judge argv). + */ +export function renderGrokInferenceClientConfig(input: GrokInferenceClientConfigInput): string { + let declared: GrokBrokerModelPolicy; + try { declared = parseGrokBrokerModelPolicy({ model: input.model, reasoningEffort: input.reasoningEffort }); } catch { throw new TypeError("invalid Grok inference client configuration"); } + if (declared.model !== input.model || declared.reasoningEffort !== input.reasoningEffort) throw new TypeError("invalid Grok inference client configuration"); + if (typeof input.baseUrl !== "string" || !/^http:\/\/127\.0\.0\.1:([1-9][0-9]{0,4})\/v1$/u.test(input.baseUrl) || Number(input.baseUrl.slice(17, -3)) > 65_535) throw new TypeError("invalid Grok inference client configuration"); + if (typeof input.envKey !== "string" || !/^[A-Z][A-Z0-9_]{2,63}$/u.test(input.envKey)) throw new TypeError("invalid Grok inference client configuration"); + const label = `${declared.reasoningEffort[0]!.toUpperCase()}${declared.reasoningEffort.slice(1)}`; + return [ + renderGrokLeanBaseConfig(), + "[models]", `default = "${GROK_INFERENCE_CLIENT_MODEL_ID}"`, `default_reasoning_effort = "${declared.reasoningEffort}"`, `session_summary = "${GROK_SESSION_TITLE_SINK_MODEL_ID}"`, "", + `[model.${GROK_SESSION_TITLE_SINK_MODEL_ID}]`, 'model = "disabled"', `base_url = "${input.baseUrl}"`, `api_key = "${GROK_SESSION_TITLE_SINK_KEY}"`, "max_retries = 0", "hidden = true", "", + `[model.${GROK_INFERENCE_CLIENT_MODEL_ID}]`, `model = "${declared.model}"`, `base_url = "${input.baseUrl}"`, `env_key = "${input.envKey}"`, + 'api_backend = "chat_completions"', "context_window = 131072", "supports_backend_search = false", "max_retries = 0", "", + `[[model.${GROK_INFERENCE_CLIENT_MODEL_ID}.reasoning_efforts]]`, `value = "${declared.reasoningEffort}"`, `label = "${label}"`, "default = true", "" + ].join("\n"); +} + +export const grokInferenceClientConfigSha256 = (input: GrokInferenceClientConfigInput): string => + createHash("sha256").update(renderGrokInferenceClientConfig(input)).digest("hex"); + +/** The production bytes for a declared model/effort: the grant's proxy URL and the canonical env key. */ +export const renderProductionGrokInferenceClientConfig = (policy: GrokBrokerModelPolicy): string => + renderGrokInferenceClientConfig({ baseUrl: GROK_INFERENCE_PROXY_BASE_URL, model: policy.model, reasoningEffort: policy.reasoningEffort, envKey: GROK_INFERENCE_GRANT_ENV }); diff --git a/src/runtime/grokInferenceGrants.test.ts b/src/runtime/grokInferenceGrants.test.ts new file mode 100644 index 0000000..8fb97e1 --- /dev/null +++ b/src/runtime/grokInferenceGrants.test.ts @@ -0,0 +1,94 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { GROK_ENGINE_BROKER } from "../contracts/runtimeContractManifest.js"; +import { EngineBrokerCapabilities } from "./engineBrokerCapabilities.js"; +import { GrokInferenceGrantRefused, GrokInferenceGrants } from "./grokInferenceGrants.js"; +import type { InferenceUsageEntry } from "./inferenceUsageLedger.js"; + +const judge = { model: "grok-4.6", reasoningEffort: "low", purpose: "judge" } as const; +const refusedWith = (code: string) => (error: unknown) => error instanceof GrokInferenceGrantRefused && error.code === code; + +test("a grant is scoped to its declared model, effort and purpose and carries the manifest limits", () => { + const grants = new GrokInferenceGrants(); + try { + const issued = grants.issue({ model: "grok-4.5", reasoningEffort: "medium", purpose: "optimizer" }); + assert.match(issued.token, /^inference_[A-Za-z0-9_-]{43}$/u); assert.match(issued.grantId, /^[a-f0-9]{32}$/u); + assert.deepEqual(issued.limits, { maxRequests: 64, maxTokens: 2_000_000, timeoutMs: 600_000 }); + const grant = grants.authorize(issued.token); + assert.deepEqual(grant?.policy, { model: "grok-4.5", reasoningEffort: "medium" }); assert.equal(grant?.purpose, "optimizer"); + assert.equal(grants.authorize(issued.token.replace(/.$/u, (last) => last === "A" ? "B" : "A")), undefined); + } finally { grants.close(); } +}); + +test("a grant request naming an undeclared model, effort or purpose, or omitting one, is refused", () => { + const grants = new GrokInferenceGrants(); + for (const request of [{ ...judge, model: "grok-3" }, { ...judge, reasoningEffort: "xhigh" }, { ...judge, purpose: "subject" }, { ...judge, model: undefined }, { ...judge, reasoningEffort: undefined }]) { + assert.throws(() => grants.issue(request), refusedWith("invalid_request")); + } + assert.equal(grants.live(), 0); +}); + +test("the grant TTL can never exceed ten minutes", () => { + assert.equal(GROK_ENGINE_BROKER.inferenceGrants.ttlMs <= 600_000, true); + assert.throws(() => new GrokInferenceGrants({ ttlMs: 600_001 }), /invalid inference grant policy/u); +}); + +test("an expired grant is refused and frees its slot", () => { + let now = 1_000_000; + const grants = new GrokInferenceGrants({ now: () => now, maxLiveGrants: 1 }); + try { + const issued = grants.issue(judge); + now += 599_999; assert.ok(grants.authorize(issued.token)); + now += 1; assert.equal(grants.authorize(issued.token), undefined); + assert.equal(grants.live(), 0); assert.ok(grants.issue(judge)); + } finally { grants.close(); } +}); + +test("live grants are capped and a release frees a slot", () => { + const grants = new GrokInferenceGrants(); + try { + const issued = Array.from({ length: GROK_ENGINE_BROKER.inferenceGrants.maxLiveGrants }, () => grants.issue(judge)); + assert.throws(() => grants.issue(judge), refusedWith("grant_limit")); + assert.equal(grants.release(issued[3]!.grantId), true); assert.equal(grants.authorize(issued[3]!.token), undefined); + assert.ok(grants.issue(judge)); assert.throws(() => grants.issue(judge), refusedWith("grant_limit")); + } finally { grants.close(); } +}); + +test("grants never share a key space with turn capabilities", () => { + const turnId = "0123456789abcdef0123456789abcdef"; + const capabilities = new EngineBrokerCapabilities(); const turnToken = capabilities.issue("agent-a", turnId); + const grants = new GrokInferenceGrants({ grantId: () => turnId }); + try { + const issued = grants.issue(judge); + assert.equal(issued.grantId, turnId); + assert.deepEqual(capabilities.inspectToken(turnToken), { agentId: "agent-a", turnId }); + assert.equal(capabilities.inspectToken(issued.token), undefined); + assert.equal(grants.authorize(turnToken), undefined); + capabilities.revoke(turnId); assert.ok(grants.authorize(issued.token)); + grants.release(turnId); assert.equal(capabilities.inspectToken(turnToken), undefined); + } finally { grants.close(); } +}); + +test("a grant meters one request at a time and emits one row per settled request, estimated when usage is missing", () => { + const rows: InferenceUsageEntry[] = []; + const grants = new GrokInferenceGrants({ onSettled: (row) => rows.push(row) }); + try { + const grant = grants.authorize(grants.issue(judge).token)!; + const first = grant.meter.admit(); assert.ok("index" in first); + assert.deepEqual(grant.meter.admit(), { busy: true }); + grants.settle(grant, first.index, { input: 90, cacheRead: 10, cacheWrite: 0, output: 5, total: 105 }, 400); + grants.settle(grant, first.index, { input: 1, cacheRead: 0, cacheWrite: 0, output: 1, total: 2 }, 400); + const second = grant.meter.admit(); assert.ok("index" in second); + grants.settle(grant, second.index, undefined, 1_000); + assert.deepEqual(rows.map((row) => [row.request, row.usage.total, row.usageSource, row.purpose, row.model]), [[0, 105, "upstream", "judge", "grok-4.6"], [1, 4_596, "estimated", "judge", "grok-4.6"]]); + } finally { grants.close(); } +}); + +test("releasing a grant aborts its in-flight upstream request", () => { + const grants = new GrokInferenceGrants(); + const issued = grants.issue(judge); const grant = grants.authorize(issued.token)!; + const admission = grant.meter.admit(); assert.ok("signal" in admission); + grants.release(issued.grantId); + assert.equal(admission.signal.aborted, true); +}); diff --git a/src/runtime/grokInferenceGrants.ts b/src/runtime/grokInferenceGrants.ts new file mode 100644 index 0000000..a8c1b93 --- /dev/null +++ b/src/runtime/grokInferenceGrants.ts @@ -0,0 +1,119 @@ +import { createHash, randomBytes, timingSafeEqual } from "node:crypto"; + +import { GROK_ENGINE_BROKER } from "../contracts/runtimeContractManifest.js"; +import type { EngineBrokerInferenceFailureCode } from "./engineBrokerInferenceProtocol.js"; +import type { EngineBrokerTurnLimits, EngineBrokerTurnUsage } from "./engineBrokerTurnAccounting.js"; +import { parseGrokBrokerModelPolicy, type GrokBrokerModelPolicy } from "./grokBrokerModelPolicy.js"; +import { GrokBrokerTurnMeter } from "./grokBrokerTurnMeter.js"; +import { GROK_INFERENCE_PURPOSES, recordInferenceUsage, type GrokInferencePurpose, type InferenceUsageEntry } from "./inferenceUsageLedger.js"; + +const SPEC = GROK_ENGINE_BROKER.inferenceGrants; + +/** One live evaluator grant as the proxy sees it. */ +export type GrokInferenceGrant = Readonly<{ grantId: string; purpose: GrokInferencePurpose; policy: GrokBrokerModelPolicy; expiresAt: number; meter: GrokBrokerTurnMeter }>; +export type GrokInferenceGrantIssued = Readonly<{ grantId: string; token: string; purpose: GrokInferencePurpose; policy: GrokBrokerModelPolicy; expiresAt: number; limits: EngineBrokerTurnLimits }>; +export type GrokInferenceGrantRequest = Readonly<{ model: unknown; reasoningEffort: unknown; purpose: unknown }>; + +export class GrokInferenceGrantRefused extends Error { + constructor(readonly code: EngineBrokerInferenceFailureCode) { super(`inference grant refused (${code})`); } +} + +type Entry = { grant: GrokInferenceGrant; digest: Buffer; timer: NodeJS.Timeout }; +export type GrokInferenceGrantsOptions = Readonly<{ + now?: () => number; + /** Test seam: a fixed grant id proves grants never share a key space with turn capabilities. */ + grantId?: () => string; + onSettled?: (entry: InferenceUsageEntry) => void; + ttlMs?: number; + maxLiveGrants?: number; +}>; + +/** + * Evaluator inference grants: a distinct kind beside subject turn capabilities. + * + * Grants live in their own map keyed by a random grant id; turn capabilities + * (`engineBrokerCapabilities.ts`) are keyed by turn id and never consulted + * here, and grant tokens carry `inference_` so the proxy routes a bearer to + * exactly one of the two lookups. A grant has no worker isolation guard (it is + * issued only to the organization uid, the trusted evaluator side), but it + * carries the same spend gate as a subject turn: a {@link GrokBrokerTurnMeter} + * with one request in flight, the request ceiling, the between-requests token + * ceiling and the estimate for a response without usage. Its lifetime is the + * meter's time limit: past `expiresAt` it is gone from the map, and a request + * still in flight at expiry is aborted. + * + * At most `maxLiveGrants` grants exist at once; a caller frees a slot early + * with {@link release}. Nothing here is durable — a broker restart drops every + * grant, and callers request a new one. + */ +export class GrokInferenceGrants { + private readonly grants = new Map(); + private readonly now: () => number; + private readonly ttlMs: number; + private readonly maxLive: number; + constructor(private readonly options: GrokInferenceGrantsOptions = {}) { + this.now = options.now ?? Date.now; + this.ttlMs = options.ttlMs ?? SPEC.ttlMs; + this.maxLive = options.maxLiveGrants ?? SPEC.maxLiveGrants; + if (!Number.isSafeInteger(this.ttlMs) || this.ttlMs < 1 || this.ttlMs > SPEC.ttlMs || !Number.isSafeInteger(this.maxLive) || this.maxLive < 1 || this.maxLive > SPEC.maxLiveGrants) throw new TypeError("invalid inference grant policy"); + } + + issue(request: GrokInferenceGrantRequest): GrokInferenceGrantIssued { + let policy: GrokBrokerModelPolicy; + try { policy = parseGrokBrokerModelPolicy({ model: request.model, reasoningEffort: request.reasoningEffort }); } catch { throw new GrokInferenceGrantRefused("invalid_request"); } + // Nothing is defaulted: the model parser fills an absent member, a grant must name both. + if (request.model !== policy.model || request.reasoningEffort !== policy.reasoningEffort || !(GROK_INFERENCE_PURPOSES as readonly unknown[]).includes(request.purpose)) throw new GrokInferenceGrantRefused("invalid_request"); + this.prune(); + if (this.grants.size >= this.maxLive) throw new GrokInferenceGrantRefused("grant_limit"); + const grantId = this.options.grantId?.() ?? randomBytes(16).toString("hex"); + if (!/^[a-f0-9]{32}$/u.test(grantId) || this.grants.has(grantId)) throw new GrokInferenceGrantRefused("grant_limit"); + const limits: EngineBrokerTurnLimits = Object.freeze({ maxRequests: SPEC.limits.maxRequests, maxTokens: SPEC.limits.maxTokens, timeoutMs: this.ttlMs }); + const token = `${SPEC.tokenPrefix}${randomBytes(32).toString("base64url")}`; + const grant: GrokInferenceGrant = Object.freeze({ grantId, purpose: request.purpose as GrokInferencePurpose, policy, expiresAt: this.now() + this.ttlMs, meter: new GrokBrokerTurnMeter(limits, () => undefined, this.now) }); + const timer = setTimeout(() => this.release(grantId), this.ttlMs); timer.unref?.(); + this.grants.set(grantId, { grant, digest: digest(token), timer }); + return Object.freeze({ grantId, token, purpose: grant.purpose, policy, expiresAt: grant.expiresAt, limits }); + } + + /** The live, unexpired grant a bearer names, or `undefined`. Never counts a request. */ + authorize(token: string): GrokInferenceGrant | undefined { + if (!token.startsWith(SPEC.tokenPrefix)) return undefined; + this.prune(); + const candidate = digest(token); + for (const entry of this.grants.values()) if (timingSafeEqual(entry.digest, candidate)) return entry.grant; + return undefined; + } + + /** Records one admitted request's end on the grant's meter and emits its ledger row. */ + settle(grant: GrokInferenceGrant, index: number, usage: EngineBrokerTurnUsage | undefined, requestBytes: number): void { + const before = grant.meter.snapshot().timings[index]; + if (before === undefined || before.endedAt !== undefined) return; + grant.meter.settle(index, usage, requestBytes); + const timing = grant.meter.snapshot().timings[index]; + if (timing?.usage === undefined || timing.endedAt === undefined) return; + this.options.onSettled?.({ grant: grant.grantId, purpose: grant.purpose, model: grant.policy.model, request: index, usage: timing.usage, usageSource: timing.estimated === true ? "estimated" : "upstream", startedAt: timing.startedAt, endedAt: timing.endedAt }); + } + + /** Revokes a grant and aborts its in-flight request. Returns whether it was live. */ + release(grantId: string): boolean { + const entry = this.grants.get(grantId); + if (entry === undefined) return false; + clearTimeout(entry.timer); entry.grant.meter.trip("timeout"); entry.digest.fill(0); this.grants.delete(grantId); + return true; + } + + live(): number { this.prune(); return this.grants.size; } + + close(): void { for (const grantId of [...this.grants.keys()]) this.release(grantId); } + + private prune(): void { + const now = this.now(); + for (const [grantId, entry] of this.grants) if (entry.grant.expiresAt <= now) this.release(grantId); + } +} + +/** The broker's grants: every settled request is appended to the evaluator inference ledger and nowhere else. */ +export const createLedgeredGrokInferenceGrants = (inferenceLedgerPath: string): GrokInferenceGrants => + new GrokInferenceGrants({ onSettled: (entry) => { void recordInferenceUsage(inferenceLedgerPath, entry); } }); + +const digest = (value: string): Buffer => createHash("sha256").update(value).digest(); diff --git a/src/runtime/grokInferenceLedger.test.ts b/src/runtime/grokInferenceLedger.test.ts new file mode 100644 index 0000000..26fcaab --- /dev/null +++ b/src/runtime/grokInferenceLedger.test.ts @@ -0,0 +1,74 @@ +import assert from "node:assert/strict"; +import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { request as httpRequest } from "node:http"; +import os from "node:os"; +import path from "node:path"; +import test from "node:test"; + +import type { NativeBrokerTurn } from "./engineBrokerNativeClient.js"; +import type { EngineBrokerServiceRegistration } from "./engineBrokerServiceConfig.js"; +import { EngineBrokerTurnRegistry } from "./engineBrokerTurnRegistry.js"; +import { startGrokBrokerProxy } from "./grokBrokerProxy.js"; +import { runGrokEngineBrokerTurn, type GrokEngineBrokerTurnDependencies } from "./grokEngineBrokerTurn.js"; +import { createLedgeredGrokInferenceGrants } from "./grokInferenceGrants.js"; +import { dedupeInferenceUsageRows, INFERENCE_USAGE_LEDGER_VERSION } from "./inferenceUsageLedger.js"; +import { WakeFuse } from "./wakeFuse.js"; + +const lean = ["run_terminal_command", "read_file", "list_dir", "grep", "search_tool", "use_tool"].map((name) => ({ type: "function", function: { name } })); +const leanBody = JSON.stringify({ model: "grok-4.6", reasoning_effort: "low", stream: true, messages: [], tools: lean }); +const judgeBody = JSON.stringify({ messages: [{ role: "system", content: "judge" }, { role: "user", content: "Rate." }], model: "grok-4.5", reasoning_effort: "medium", stream: true, stream_options: { include_usage: true } }); +const post = (port: number, bearer: string, body: string): Promise => new Promise((resolve, reject) => { + const req = httpRequest({ host: "127.0.0.1", port, path: "/v1/chat/completions", method: "POST", agent: false, headers: { authorization: `Bearer ${bearer}`, "x-grok-client-version": "1.0.34", "content-type": "application/json" } }, (response) => { response.resume(); response.on("end", () => resolve(response.statusCode ?? 0)); }); + req.on("error", reject); req.end(body); +}); +const rows = async (file: string): Promise[]> => (await readFile(file, "utf8").catch(() => "")).split("\n").filter(Boolean).map((line) => JSON.parse(line) as Record); +const eventually = async (check: () => Promise): Promise => { for (let attempt = 0; attempt < 100 && !await check(); attempt++) await new Promise((resolve) => setTimeout(resolve, 10)); }; + +test("a judge grant used while a subject turn runs meters only into the inference ledger and never into the subject turn, its ledger or the wake fuse", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "daimon-inference-ledger-")); + const usageLedger = path.join(root, "slot0", "usage.jsonl"), inferenceLedger = path.join(root, "evaluator", "inference.jsonl"); + await mkdir(path.dirname(usageLedger)); await mkdir(path.dirname(inferenceLedger)); + const grants = createLedgeredGrokInferenceGrants(inferenceLedger); + const upstreamUsage = (model: string) => ({ prompt_tokens: model === "grok-4.6" ? 1_000 : 40_000, completion_tokens: 50, total_tokens: model === "grok-4.6" ? 1_050 : 40_050 }); + const proxy = await startGrokBrokerProxy({ accessToken: async () => "provider-token", markRejected: async () => undefined }, async (request) => { + const model = (JSON.parse(Buffer.from(request.body).toString("utf8")) as { model: string }).model; + return { status: 200, headers: { "content-type": "text/event-stream" }, body: Buffer.from(`data: ${JSON.stringify({ choices: [], usage: upstreamUsage(model) })}\n\ndata: [DONE]\n\n`) }; + }, undefined, 0, grants); + try { + const issued = grants.issue({ model: "grok-4.5", reasoningEffort: "medium", purpose: "judge" }); + const registration: EngineBrokerServiceRegistration = { agentId: "foreman", slot: 0, workerUid: 2_200, workspace: "/workspace", profilePath: "/workers/0/.grok/sandbox.toml", eventsPath: "/workers/0/.grok/sessions/sandbox-events.jsonl", profileSha256: "a".repeat(64), usageLedgerPath: usageLedger, limits: { maxRequests: 32, maxTokens: 300_000, timeoutMs: 240_000 }, model: { model: "grok-4.6", reasoningEffort: "low" } }; + let judgeStatus = 0; + const deps: GrokEngineBrokerTurnDependencies = { + turns: new EngineBrokerTurnRegistry(path.join(root, "turns")), proxy, credentialStale: () => false, + mcp: { register: () => "mcp-capability-0123456789abcdef", revoke: () => undefined }, + prepareIsolation: async () => async () => undefined, + runNative: async (input: NativeBrokerTurn) => { + assert.equal(await post(proxy.port, input.providerCapability, leanBody), 200); + judgeStatus = await post(proxy.port, issued.token, judgeBody); + assert.equal(await post(proxy.port, input.providerCapability, leanBody), 200); + return { text: "", workerPid: 4_242, workerUid: 2_200, startTicks: 99n, diagnostic: { status: "ok", stage: "output", failureClass: "none", profileApplied: false, exitCode: 0, termSignal: 0, workerPid: 4_242, workerUid: 2_200, startTicks: "99" } }; + } + }; + await assert.rejects(runGrokEngineBrokerTurn(deps, registration, "wake-1", "prompt", "http://127.0.0.1:43124/mcp")); + assert.equal(judgeStatus, 200); + await eventually(async () => (await rows(inferenceLedger)).length > 0); + + const subject = await rows(usageLedger), subjectRequests = await rows(path.join(path.dirname(usageLedger), "requests.jsonl")), inference = await rows(inferenceLedger); + assert.deepEqual(subject.map((row) => [row.requests ?? row.calls, row.total, row.model]), [[2, 2_100, "grok-4.6"]], "the subject turn counts only its own two requests"); + assert.deepEqual(subjectRequests.map((row) => row.total), [1_050, 1_050]); + assert.deepEqual(inference.map((row) => [row.v, row.kind, row.purpose, row.grant, row.request, row.model, row.total, row.usage_source]), [[INFERENCE_USAGE_LEDGER_VERSION, "inference", "judge", issued.grantId, 0, "grok-4.5", 40_050, "upstream"]]); + + // Even if an inference row reached the subject ledger, the wake fuse would not count it: + // 2,100 subject tokens are under a 2,101 ceiling; counted, the 40,050-token row would trip it. + await writeFile(usageLedger, [...subject, ...inference].map((row) => JSON.stringify(row)).join("\n") + "\n"); + const fuseDirectory = path.join(root, "fuse"); await mkdir(fuseDirectory); + const fuse = await WakeFuse.open({ organizationKey: "org", now: () => new Date(Date.parse(String(subject[0]!.at)) - 1), environment: { DAIMON_WAKE_FUSE_DIRECTORY: fuseDirectory, DAIMON_WAKE_FUSE_EPOCH: "grants", DAIMON_WAKE_FUSE_MAX_WAKES: "10", DAIMON_WAKE_FUSE_MAX_TOKENS: "2101", DAIMON_TURN_USAGE_LEDGER_PATH: usageLedger } }); + assert.deepEqual(await fuse.admit("foreman", "next"), { state: "admitted" }); + } finally { grants.close(); await proxy.close(); await rm(root, { recursive: true, force: true }); } +}); + +test("inference readers count each (grant, request) once", () => { + type Row = Readonly<{ grant?: string; request?: number; total: number }>; + const row = (grant: string, request: number, total: number): Row => ({ grant, request, total }); + assert.deepEqual(dedupeInferenceUsageRows([row("a", 0, 1), row("a", 1, 2), row("a", 0, 1), row("b", 0, 3), { total: 9 }]).map((value) => value.total), [1, 2, 3]); +}); diff --git a/src/runtime/grokInferenceProxy.test.ts b/src/runtime/grokInferenceProxy.test.ts new file mode 100644 index 0000000..8fcb20e --- /dev/null +++ b/src/runtime/grokInferenceProxy.test.ts @@ -0,0 +1,157 @@ +import assert from "node:assert/strict"; +import { request as httpRequest } from "node:http"; +import test from "node:test"; + +import { startGrokBrokerProxy, type GrokBrokerCredentialAuthority, type GrokBrokerUpstream } from "./grokBrokerProxy.js"; +import { GROK_SESSION_TITLE_SINK_KEY } from "./grokBrokerWorkerConfig.js"; +import { GrokBrokerTurnMeter } from "./grokBrokerTurnMeter.js"; +import { GrokInferenceGrants } from "./grokInferenceGrants.js"; +import { GROK_INFERENCE_AUTH_STALE_BODY } from "./grokInferenceProxy.js"; +import type { InferenceUsageEntry } from "./inferenceUsageLedger.js"; + +// The judge main request Grok 1.0.34 sends (live capture, `--json-schema` variant). +const judgeBody = (overrides: Record = {}): string => JSON.stringify({ + messages: [{ role: "system", content: "You are a strict judge." }, { role: "user", content: "..." }, { role: "user", content: "Rate the answer." }], + model: "grok-4.6", reasoning_effort: "low", + response_format: { type: "json_schema", json_schema: { name: "structured_output", schema: { type: "object", properties: { score: { type: "number" } }, required: ["score"], additionalProperties: false }, strict: true } }, + stream: true, stream_options: { include_usage: true }, ...overrides +}); +// The per-call session_title request the same CLI sends first (live capture). +const titleBody = JSON.stringify({ model: "grok-4.6", reasoning_effort: "low", temperature: 1, max_tokens: 100, messages: [{ role: "system", content: "title" }, { role: "user", content: "Rate the answer." }], tools: [{ type: "function", function: { name: "session_title", description: "", parameters: {} } }], tool_choice: { type: "function", function: { name: "session_title" } }, stream: true, stream_options: { include_usage: true } }); +const usageStream = (total: number) => Buffer.from(`data: ${JSON.stringify({ choices: [{ index: 0, delta: { content: "{\"score\":1}" } }] })}\n\ndata: ${JSON.stringify({ choices: [], usage: { prompt_tokens: total - 5, completion_tokens: 5, total_tokens: total } })}\n\ndata: [DONE]\n\n`); + +type Harness = Readonly<{ port: number; grants: GrokInferenceGrants; rows: InferenceUsageEntry[]; bodies: Record[]; proxy: Awaited> }>; +async function withProxy(run: (harness: Harness) => Promise, options: Readonly<{ authority?: GrokBrokerCredentialAuthority; upstream?: GrokBrokerUpstream }> = {}): Promise { + const rows: InferenceUsageEntry[] = [], bodies: Record[] = []; + const grants = new GrokInferenceGrants({ onSettled: (row) => rows.push(row) }); + const upstream: GrokBrokerUpstream = options.upstream ?? (async (request) => { bodies.push(JSON.parse(Buffer.from(request.body).toString("utf8")) as Record); return { status: 200, headers: { "content-type": "text/event-stream" }, body: usageStream(105) }; }); + const proxy = await startGrokBrokerProxy(options.authority ?? { accessToken: async () => "provider-token", markRejected: async () => undefined }, upstream, undefined, 0, grants); + try { await run({ port: proxy.port, grants, rows, bodies, proxy }); } finally { grants.close(); await proxy.close(); } +} + +function post(port: number, bearer: string, body: string, version = "1.0.34"): Promise> { + return new Promise((resolve, reject) => { + const req = httpRequest({ host: "127.0.0.1", port, path: "/v1/chat/completions", method: "POST", agent: false, headers: { authorization: `Bearer ${bearer}`, "x-grok-client-version": version, "content-type": "application/json" } }, (response) => { + const chunks: Buffer[] = []; response.on("data", (chunk: Buffer) => chunks.push(chunk)); response.on("end", () => resolve({ status: response.statusCode ?? 0, text: Buffer.concat(chunks).toString("utf8") })); + }); + req.on("error", reject); req.end(body); + }); +} + +test("a grant forwards the captured judge request re-serialized under the declared model and meters it into the inference rows only", async () => { + await withProxy(async ({ port, grants, rows, bodies }) => { + const issued = grants.issue({ model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }); + const result = await post(port, issued.token, judgeBody().replace('"model":', '"model":"grok-4.5","model":')); + assert.equal(result.status, 200); assert.match(result.text, /score/u); + assert.equal(bodies.length, 1); assert.equal(bodies[0]!.model, "grok-4.6"); + assert.deepEqual(rows.map((row) => [row.grant, row.request, row.usage.total, row.usageSource, row.purpose]), [[issued.grantId, 0, 105, "upstream", "judge"]]); + assert.equal((await post(port, issued.token, judgeBody({ response_format: undefined }))).status, 200); + assert.equal(rows.length, 2); + }); +}); + +test("a grant refuses any tools member, the session_title request, and undeclared model or effort, before any upstream call or row", async () => { + await withProxy(async ({ port, grants, rows, bodies }) => { + const { token } = grants.issue({ model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }); + const refused = [ + judgeBody({ tools: [] }), judgeBody({ tools: [{ type: "function", function: { name: "read_file" } }] }), judgeBody({ tool_choice: "none" }), titleBody, + judgeBody({ model: "grok-4.5" }), judgeBody({ reasoning_effort: "high" }), judgeBody({ reasoning_effort: undefined }), + judgeBody({ stream: false }), judgeBody({ stream_options: undefined }), judgeBody({ temperature: 1 }), + judgeBody({ messages: [{ role: "tool", content: "x" }] }), judgeBody({ messages: [{ role: "assistant", content: null, tool_calls: [] }] }), + judgeBody({ response_format: { type: "json_object" } }) + ]; + for (const body of refused) assert.equal((await post(port, token, body)).status, 503, body.slice(0, 120)); + assert.equal((await post(port, token, judgeBody(), "1.0.30")).status, 503); + assert.equal((await post(port, GROK_SESSION_TITLE_SINK_KEY, titleBody)).status, 503); + assert.equal(bodies.length, 0); assert.equal(rows.length, 0); + }); +}); + +test("an expired, released or unknown grant is refused", async () => { + await withProxy(async ({ port, grants, bodies }) => { + const released = grants.issue({ model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }); grants.release(released.grantId); + assert.equal((await post(port, released.token, judgeBody())).status, 503); + assert.equal((await post(port, `inference_${"A".repeat(43)}`, judgeBody())).status, 503); + assert.equal(bodies.length, 0); + }); + let now = 5_000; + const grants = new GrokInferenceGrants({ now: () => now }); + const proxy = await startGrokBrokerProxy({ accessToken: async () => "provider-token", markRejected: async () => undefined }, async () => ({ status: 200, headers: { "content-type": "text/event-stream" }, body: usageStream(10) }), undefined, 0, grants); + try { + const { token } = grants.issue({ model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }); + assert.equal((await post(proxy.port, token, judgeBody())).status, 200); + now += 600_000; + assert.equal((await post(proxy.port, token, judgeBody())).status, 503); + } finally { grants.close(); await proxy.close(); } +}); + +test("a grant token never authorizes a subject turn and a turn capability never authorizes a grant request", async () => { + await withProxy(async ({ port, grants, proxy, bodies }) => { + const { token: grantToken } = grants.issue({ model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }); + const turnToken = proxy.capabilities.issue("agent-a", "turn-a"); + proxy.registerIsolationGuard("turn-a", async () => undefined); + proxy.registerTurn("turn-a", { policy: { model: "grok-4.6", reasoningEffort: "low" }, meter: new GrokBrokerTurnMeter({ maxRequests: 4, maxTokens: 10_000, timeoutMs: 60_000 }) }); + const lean = ["run_terminal_command", "read_file", "list_dir", "grep", "search_tool", "use_tool"].map((name) => ({ type: "function", function: { name } })); + const leanBody = JSON.stringify({ model: "grok-4.6", reasoning_effort: "low", stream: true, messages: [], tools: lean }); + assert.equal((await post(port, grantToken, leanBody)).status, 503); + assert.equal((await post(port, turnToken, judgeBody())).status, 503); + assert.equal(bodies.length, 0); + assert.equal((await post(port, turnToken, leanBody)).status, 200); + assert.equal((await post(port, grantToken, judgeBody())).status, 200); + }); +}); + +test("a stale realm answers a grant request with the distinct auth_stale failure, and a rejected refresh too", async () => { + let stale = true; + await withProxy(async ({ port, grants, bodies }) => { + const { token } = grants.issue({ model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }); + const result = await post(port, token, judgeBody()); + assert.equal(result.status, 401); assert.equal(result.text, GROK_INFERENCE_AUTH_STALE_BODY); assert.equal(bodies.length, 0); + }, { authority: { accessToken: async () => { if (stale) throw new Error("stale"); return "t"; }, markRejected: async () => undefined, isStale: () => stale } }); + stale = false; let rejected = 0; + await withProxy(async ({ port, grants, rows }) => { + const { token } = grants.issue({ model: "grok-4.6", reasoningEffort: "low", purpose: "optimizer" }); + const result = await post(port, token, judgeBody()); + assert.equal(result.status, 401); assert.equal(result.text, GROK_INFERENCE_AUTH_STALE_BODY); assert.equal(rejected, 1); + assert.deepEqual(rows.map((row) => row.usageSource), ["estimated"]); + }, { authority: { accessToken: async (force) => force ? "second" : "first", markRejected: async () => { rejected++; stale = true; throw new Error("stale"); }, isStale: () => stale }, upstream: async () => ({ status: 401, headers: { "content-type": "application/json" }, body: new Uint8Array() }) }); +}); + +test("a grant's request ceiling and one-in-flight rule hold on the wire", async () => { + let release!: () => void; const gate = new Promise((resolve) => { release = resolve; }); + let calls = 0; + await withProxy(async ({ port, grants, rows }) => { + const { token } = grants.issue({ model: "grok-4.6", reasoningEffort: "low", purpose: "judge" }); + const first = post(port, token, judgeBody()); + while (calls === 0) await new Promise((resolve) => setTimeout(resolve, 5)); + const busy = await post(port, token, judgeBody()); assert.equal(busy.status, 429); assert.match(busy.text, /in flight/u); + release(); assert.equal((await first).status, 200); + const grant = grants.authorize(token)!; + for (let index = 1; index < 64; index++) { const admission = grant.meter.admit(); assert.ok("index" in admission); grants.settle(grant, admission.index, { input: 1, cacheRead: 0, cacheWrite: 0, output: 1, total: 2 }, 10); } + const over = await post(port, token, judgeBody()); assert.equal(over.status, 429); assert.match(over.text, /requests/u); + assert.equal(calls, 1); assert.equal(rows.length, 64); + }, { upstream: async () => { calls++; await gate; return { status: 200, headers: { "content-type": "text/event-stream" }, body: usageStream(50) }; } }); +}); + +test("a grant whose id equals a live turn id leaves that turn's capability, policy and meter untouched", async () => { + const shared = "0123456789abcdef0123456789abcdef"; + const grants = new GrokInferenceGrants({ grantId: () => shared }); + let calls = 0; + const proxy = await startGrokBrokerProxy({ accessToken: async () => "provider-token", markRejected: async () => undefined }, async () => { calls++; return { status: 200, headers: { "content-type": "text/event-stream" }, body: usageStream(20) }; }, undefined, 0, grants); + try { + const turnToken = proxy.capabilities.issue("agent-a", shared); + const turnMeter = new GrokBrokerTurnMeter({ maxRequests: 4, maxTokens: 10_000, timeoutMs: 60_000 }); + proxy.registerIsolationGuard(shared, async () => undefined); + proxy.registerTurn(shared, { policy: { model: "grok-4.6", reasoningEffort: "low" }, meter: turnMeter }); + const issued = grants.issue({ model: "grok-4.5", reasoningEffort: "high", purpose: "judge" }); + assert.equal(issued.grantId, shared); + const lean = ["run_terminal_command", "read_file", "list_dir", "grep", "search_tool", "use_tool"].map((name) => ({ type: "function", function: { name } })); + const leanBody = JSON.stringify({ model: "grok-4.6", reasoningEffort: undefined, reasoning_effort: "low", stream: true, messages: [], tools: lean }); + assert.equal((await post(proxy.port, turnToken, leanBody)).status, 200); + assert.equal((await post(proxy.port, issued.token, judgeBody({ model: "grok-4.5", reasoning_effort: "high" }))).status, 200); + assert.equal(turnMeter.snapshot().requests, 1); assert.equal(grants.authorize(issued.token)!.meter.snapshot().requests, 1); + grants.release(shared); + assert.equal((await post(proxy.port, turnToken, leanBody)).status, 200); + assert.equal(turnMeter.snapshot().requests, 2); assert.equal(calls, 3); + } finally { grants.close(); await proxy.close(); } +}); diff --git a/src/runtime/grokInferenceProxy.ts b/src/runtime/grokInferenceProxy.ts new file mode 100644 index 0000000..95169c1 --- /dev/null +++ b/src/runtime/grokInferenceProxy.ts @@ -0,0 +1,65 @@ +import { createHash } from "node:crypto"; +import type { ServerResponse } from "node:http"; + +import type { GrokBrokerCredentialAuthority, GrokBrokerUpstream } from "./grokBrokerProxy.js"; +import { parseGrokUpstreamUsage } from "./grokBrokerTurnMeter.js"; +import type { GrokInferenceGrants } from "./grokInferenceGrants.js"; +import { authorizeGrokInferenceProxyRequest } from "./grokInferenceProxyRequest.js"; + +export type GrokInferenceProxyInput = Readonly<{ method: string; pathname: string; headers: Readonly>; body: Buffer; token: string }>; + +/** HTTP 401 with a fixed code: the broker realm is stale, distinct from a generic 503 broker failure. */ +export const GROK_INFERENCE_AUTH_STALE_BODY = '{"error":"auth_stale"}'; + +/** + * Serves one evaluator grant request; never throws. + * + * Order mirrors the subject path: the body is proven a grant-shaped request + * (declared model/effort, no tools) before the credential is read, the grant + * meter admits it before any upstream call, and exactly one ledger row is + * written once an admitted request settles. No worker isolation guard applies: + * grants are issued only to the organization uid. + * + * Stale realm: the grant shares the subject's credential authority, so a + * stale realm fails judges and subject turns alike (accepted shared fate). A + * grant request that finds the realm stale — before the upstream call or + * after a rejected refresh — is answered 401 `{"error":"auth_stale"}`, never + * the generic 503, so the evaluator can report it as a credential failure. + */ +export async function serveGrokInferenceGrant(input: GrokInferenceProxyInput, response: ServerResponse, grants: GrokInferenceGrants, authority: GrokBrokerCredentialAuthority, upstream: GrokBrokerUpstream): Promise { + let settle: ((usage: ReturnType) => void) | undefined; + try { + const grant = grants.authorize(input.token); + if (grant === undefined) throw new Error("inference grant unavailable"); + let prepared = authorizeGrokInferenceProxyRequest(input, "pending", grant.policy); + let token = await authority.accessToken(false); const rejectedDigest = createHash("sha256").update(token).digest("hex"); + prepared = withBearer(prepared, token); token = ""; + const admission = grant.meter.admit(); + if ("refused" in admission) return json(response, 429, JSON.stringify({ error: "grant limit reached", limit: admission.refused })); + if ("busy" in admission) return json(response, 429, '{"error":"grant request in flight"}'); + settle = (usage) => { settle = undefined; grants.settle(grant, admission.index, usage, input.body.byteLength); }; + let result = await upstream(prepared, admission.signal); + if (result.status === 401) { + token = authority.refreshAfterRejection ? await authority.refreshAfterRejection(rejectedDigest) : await authority.accessToken(true); + const refreshedDigest = createHash("sha256").update(token).digest("hex"); prepared = withBearer(prepared, token); token = ""; + result = await upstream(prepared, admission.signal); + if (result.status === 401) await authority.markRejected(refreshedDigest); + } + settle?.(parseGrokUpstreamUsage(result.body, result.headers["content-type"])); + json(response, result.status, result.body, result.headers["content-type"]); + } catch { + settle?.(undefined); + if (authority.isStale?.() === true) json(response, 401, GROK_INFERENCE_AUTH_STALE_BODY); + else json(response, 503, '{"error":"broker unavailable"}'); + } +} + +const withBearer = (prepared: ReturnType, token: string): ReturnType => { + if (!token || /[\r\n]/u.test(token)) throw new Error("broker credential authority unavailable"); + return { ...prepared, headers: { ...prepared.headers, authorization: `Bearer ${token}` } }; +}; + +function json(response: ServerResponse, status: number, body: string | Uint8Array, contentType = "application/json"): void { + if (response.headersSent) return; + response.writeHead(status, { "content-type": contentType, "cache-control": "no-store" }); response.end(body); +} diff --git a/src/runtime/grokInferenceProxyRequest.ts b/src/runtime/grokInferenceProxyRequest.ts new file mode 100644 index 0000000..51796cf --- /dev/null +++ b/src/runtime/grokInferenceProxyRequest.ts @@ -0,0 +1,54 @@ +import { GROK_ENGINE_BROKER } from "../contracts/runtimeContractManifest.js"; +import type { GrokBrokerProxyInput, GrokBrokerUpstreamRequest } from "./grokBrokerProxyRequest.js"; +import { parseGrokBrokerModelPolicy, type GrokBrokerModelPolicy } from "./grokBrokerModelPolicy.js"; + +const MAX_BODY = 2 * 1024 * 1024; +const SPEC = GROK_ENGINE_BROKER.inferenceGrants; +const BODY_MEMBERS: ReadonlySet = new Set(SPEC.bodyMembers); +const ROLES: ReadonlySet = new Set(SPEC.messageRoles); +type JsonRecord = Record; +const plain = (value: unknown): value is JsonRecord => value !== null && typeof value === "object" && !Array.isArray(value); +const rejected = (): Error => new Error("inference grant request rejected"); + +/** + * Authorizes one evaluator grant request body and rebuilds it for the provider. + * + * The accepted shape is exactly what Grok CLI 1.0.34 sends for a Paideia + * judge/optimizer call (`--tools read_file --disallowed-tools + * read_file,search_tool,use_tool --max-turns 1`, with and without + * `--system-prompt-override` and `--json-schema`; live stub capture): + * + * - `stream: true` with `stream_options: {include_usage: true}` — the CLI never + * sends a non-streaming request, so none is accepted; + * - `model` and `reasoning_effort` equal to the grant's declaration; + * - `messages` of plain `{role, content}` string turns, roles system/user/assistant; + * - optional `response_format` `{type: "json_schema", json_schema: {name, schema, strict}}`; + * - no `tools` and no `tool_choice` member at all, not even an empty one. The + * CLI's per-call `session_title` request carries both and is refused here. + * + * The client version must be the pinned CLI, and the forwarded body is the + * re-serialized parse, never the caller's bytes. + */ +export function authorizeGrokInferenceProxyRequest(input: Omit, bearer: string, policy: GrokBrokerModelPolicy): GrokBrokerUpstreamRequest { + const declared = parseGrokBrokerModelPolicy(policy); + if (input.method !== "POST" || input.pathname !== "/v1/chat/completions" || input.body.byteLength < 2 || input.body.byteLength > MAX_BODY) throw rejected(); + if (!bearer || /[\r\n]/u.test(bearer)) throw new Error("broker credential authority unavailable"); + const clientVersion = input.headers["x-grok-client-version"]; + if (clientVersion !== GROK_ENGINE_BROKER.grokCliVersion) throw rejected(); + let parsed: unknown; + try { parsed = JSON.parse(Buffer.from(input.body).toString("utf8"), (key, value: unknown) => { if (key === "__proto__") throw new Error(); return value; }); } catch { throw rejected(); } + if (!plain(parsed) || Object.keys(parsed).some((key) => !BODY_MEMBERS.has(key))) throw rejected(); + if (parsed.model !== declared.model || parsed.reasoning_effort !== declared.reasoningEffort || parsed.stream !== true) throw rejected(); + if (!plain(parsed.stream_options) || Object.keys(parsed.stream_options).length !== 1 || parsed.stream_options.include_usage !== true) throw rejected(); + if (!Array.isArray(parsed.messages) || parsed.messages.length === 0 || !parsed.messages.every(plainMessage)) throw rejected(); + if (parsed.response_format !== undefined && !jsonSchemaFormat(parsed.response_format)) throw rejected(); + return { url: "https://cli-chat-proxy.grok.com/v1/chat/completions", headers: { authorization: `Bearer ${bearer}`, "content-type": "application/json", "x-xai-token-auth": "xai-grok-cli", "x-grok-model-override": declared.model, "x-grok-client-version": clientVersion, "x-grok-client-identifier": "grok-shell" }, body: Buffer.from(JSON.stringify(parsed)) }; +} + +const plainMessage = (message: unknown): boolean => plain(message) && Object.keys(message).length === 2 && ROLES.has(message.role as string) && typeof message.content === "string"; + +const jsonSchemaFormat = (value: unknown): boolean => { + if (!plain(value) || Object.keys(value).length !== 2 || value.type !== "json_schema" || !plain(value.json_schema)) return false; + const schema = value.json_schema; + return Object.keys(schema).every((key) => key === "name" || key === "schema" || key === "strict") && typeof schema.name === "string" && plain(schema.schema) && (schema.strict === undefined || typeof schema.strict === "boolean"); +}; diff --git a/src/runtime/grokSlotPreflightReceipt.test.ts b/src/runtime/grokSlotPreflightReceipt.test.ts index b141060..2cc85d2 100644 --- a/src/runtime/grokSlotPreflightReceipt.test.ts +++ b/src/runtime/grokSlotPreflightReceipt.test.ts @@ -5,17 +5,19 @@ import test from "node:test"; import { resolveOrganizationGrokBrokerProjection } from "./grokBrokerProjection.js"; import { parseGrokSlotPreflightReceipt, verifyGrokSlotPreflightReceipt } from "./grokSlotPreflightReceipt.js"; +const fresh = { expectedNonce: "5f0e2a1c9b8d7e6f5a4b3c2d1e0f9a8b7c6d5e4f3a2b1c0d9e8f7a6b5c4d3e2f", minGeneration: 3 } as const; + const fixture = async (name: string): Promise> => JSON.parse(await readFile(new URL(`./fixtures/grok-slot-preflight/${name}`, import.meta.url), "utf8")) as Record; const projection = async () => { const input = await fixture("projection-input.json") as { config: unknown; agentId: string; options: Parameters[2] }; return resolveOrganizationGrokBrokerProjection(input.config, input.agentId, input.options); }; test("the committed valid receipt fixture proves the committed projection input", async () => { - const receipt = verifyGrokSlotPreflightReceipt(await fixture("receipt.valid.v1.json"), await projection()); + const receipt = verifyGrokSlotPreflightReceipt(await fixture("receipt.valid.v2.json"), await projection(), fresh); assert.equal(receipt.canaries.length, (await projection()).denyPaths.length); assert.ok(receipt.canaries.every((canary) => canary.method === "sandboxed-read" && canary.result === "denied")); }); test("the schema refuses a readable canary, an unknown member, duplicates and malformed digests or times", async () => { - const valid = await fixture("receipt.valid.v1.json"); + const valid = await fixture("receipt.valid.v2.json"); await assert.rejects(async () => parseGrokSlotPreflightReceipt(await fixture("receipt.readable-canary.json")), /invalid Grok slot preflight receipt/u); await assert.rejects(async () => parseGrokSlotPreflightReceipt(await fixture("receipt.unknown-member.json")), /invalid Grok slot preflight receipt/u); const canaries = valid.canaries as Record[]; @@ -28,23 +30,40 @@ test("the schema refuses a readable canary, an unknown member, duplicates and ma { ...valid, projection_sha256: "A".repeat(64) }, { ...valid, worker_uid: 2_000 }, { ...valid, created_at: "2026-09-17T12:00:00Z" }, - { ...valid, version: "noopolis.daimon.grok-slot-preflight.v2" } + { ...valid, version: "noopolis.daimon.grok-slot-preflight.v3" } ]) assert.throws(() => parseGrokSlotPreflightReceipt(bad), /invalid Grok slot preflight receipt/u); }); test("a receipt for a different projection, slot, profile or deny set is refused", async () => { const projected = await projection(); - const valid = await fixture("receipt.valid.v1.json"); + const valid = await fixture("receipt.valid.v2.json"); // Mutation guard: dropping the digest comparison accepts this fixture. - await assert.rejects(async () => verifyGrokSlotPreflightReceipt(await fixture("receipt.projection-mismatch.json"), projected), /projection_sha256/u); - await assert.rejects(async () => verifyGrokSlotPreflightReceipt(await fixture("receipt.missing-canary.json"), projected), /canaries/u); + await assert.rejects(async () => verifyGrokSlotPreflightReceipt(await fixture("receipt.projection-mismatch.json"), projected, fresh), /projection_sha256/u); + await assert.rejects(async () => verifyGrokSlotPreflightReceipt(await fixture("receipt.missing-canary.json"), projected, fresh), /canaries/u); // Exact match, both halves: a canary for a path the projection does not deny is as wrong as a missing one. - await assert.rejects(async () => verifyGrokSlotPreflightReceipt(await fixture("receipt.extra-canary.json"), projected), /canaries/u); - assert.throws(() => verifyGrokSlotPreflightReceipt(valid, { ...projected, limits: { ...projected.limits, maxTokens: 1 } }), /projection_sha256/u); - assert.throws(() => verifyGrokSlotPreflightReceipt({ ...valid, sandbox_profile_sha256: "1".repeat(64) }, projected), /sandbox_profile_sha256/u); - assert.throws(() => verifyGrokSlotPreflightReceipt({ ...valid, slot: 1 }, projected), /slot/u); + await assert.rejects(async () => verifyGrokSlotPreflightReceipt(await fixture("receipt.extra-canary.json"), projected, fresh), /canaries/u); + assert.throws(() => verifyGrokSlotPreflightReceipt(valid, { ...projected, limits: { ...projected.limits, maxTokens: 1 } }, fresh), /projection_sha256/u); + assert.throws(() => verifyGrokSlotPreflightReceipt({ ...valid, sandbox_profile_sha256: "1".repeat(64) }, projected, fresh), /sandbox_profile_sha256/u); + assert.throws(() => verifyGrokSlotPreflightReceipt({ ...valid, slot: 1 }, projected, fresh), /slot/u); // Mutation guard: never comparing the seccomp digest accepts a receipt taken under another profile. - assert.throws(() => verifyGrokSlotPreflightReceipt({ ...valid, seccomp_profile_sha256: "8".repeat(64) }, projected), /seccomp_profile_sha256/u); + assert.throws(() => verifyGrokSlotPreflightReceipt({ ...valid, seccomp_profile_sha256: "8".repeat(64) }, projected, fresh), /seccomp_profile_sha256/u); assert.throws(() => parseGrokSlotPreflightReceipt({ ...valid, sandbox_runtime: "none" }), /invalid Grok slot preflight receipt/u); assert.throws(() => parseGrokSlotPreflightReceipt((({ sandbox_runtime: _omit, ...rest }) => rest)(valid)), /invalid Grok slot preflight receipt/u); }); + +test("a receipt from an earlier recycle is refused: another nonce, a lower generation, or a v1 receipt without freshness", async () => { + const projected = await projection(); + const valid = await fixture("receipt.valid.v2.json"); + assert.equal(verifyGrokSlotPreflightReceipt(valid, projected, { ...fresh, minGeneration: 1 }).generation, 3); + // Mutation guard: ignoring the nonce accepts a receipt written for another recycle request. + assert.throws(() => verifyGrokSlotPreflightReceipt(valid, projected, { ...fresh, expectedNonce: "a".repeat(64) }), /stale: nonce/u); + // Mutation guard: ignoring the generation accepts a receipt older than the last one accepted. + assert.throws(() => verifyGrokSlotPreflightReceipt(valid, projected, { ...fresh, minGeneration: 4 }), /stale: generation/u); + await assert.rejects(async () => verifyGrokSlotPreflightReceipt(await fixture("receipt.legacy-v1.json"), projected, fresh), /invalid Grok slot preflight receipt/u); + for (const bad of [{ ...valid, nonce: "A".repeat(64) }, { ...valid, nonce: "ab" }, { ...valid, generation: 0 }, { ...valid, generation: 1.5 }, (({ nonce: _omit, ...rest }) => rest)(valid), (({ generation: _omit, ...rest }) => rest)(valid)]) { + assert.throws(() => parseGrokSlotPreflightReceipt(bad), /invalid Grok slot preflight receipt/u); + } + for (const freshness of [{ expectedNonce: "short", minGeneration: 1 }, { expectedNonce: fresh.expectedNonce, minGeneration: 0 }, { expectedNonce: fresh.expectedNonce.toUpperCase(), minGeneration: 1 }]) { + assert.throws(() => verifyGrokSlotPreflightReceipt(valid, projected, freshness), /invalid Grok slot preflight freshness/u); + } +}); diff --git a/src/runtime/grokSlotPreflightReceipt.ts b/src/runtime/grokSlotPreflightReceipt.ts index 7658581..791b9b3 100644 --- a/src/runtime/grokSlotPreflightReceipt.ts +++ b/src/runtime/grokSlotPreflightReceipt.ts @@ -7,6 +7,7 @@ import { grokBrokerProjectionSha256, type OrganizationGrokBrokerProjection } fro export const GROK_SLOT_PREFLIGHT_VERSION = GROK_ENGINE_BROKER.slotPreflightVersion; const sha256 = z.string().regex(/^[a-f0-9]{64}$/u); +const NONCE = /^[a-f0-9]{64}$/u; const canonicalAbsolute = z.string().max(4_096).refine((value) => path.posix.isAbsolute(value) && path.posix.normalize(value) === value && value !== "/" && !value.endsWith("/") && !value.includes("\0"), "canonical absolute path"); /** @@ -22,17 +23,28 @@ export const grokSlotPreflightCanarySchema = z.strictObject({ }); /** - * `noopolis.daimon.grok-slot-preflight.v1`: what the root slot supervisor (P5) + * `noopolis.daimon.grok-slot-preflight.v2`: what the root slot supervisor (P5) * writes after provisioning or recycling one broker slot, and what an * evaluator (Paideia, P4) must hold before it runs a Grok subject turn in that * slot. It binds the slot to one exact projection by digest, so any change to * the model, limits, deny list, profile, worker config, or pinned executable * invalidates it. + * + * The projection digest is identical across recycles of the same slot, so v1 + * could not tell this recycle's receipt from an earlier one. v2 adds + * freshness: `generation` is the supervisor-owned per-slot counter, strictly + * increasing on every provision/recycle, and `nonce` echoes the 32 random + * bytes (hex) the evaluator passed in its recycle request. A v1 receipt is + * refused. */ export const grokSlotPreflightReceiptSchema = z.strictObject({ version: z.literal(GROK_SLOT_PREFLIGHT_VERSION), slot: z.number().int().min(0).max(Number.MAX_SAFE_INTEGER), worker_uid: z.number().int().min(GROK_ENGINE_BROKER.identities.firstWorkerUid).max(4_294_967_294), + /** Supervisor-owned, strictly increasing per slot across provisions and recycles; starts at 1. */ + generation: z.number().int().min(1).max(Number.MAX_SAFE_INTEGER), + /** The caller's recycle nonce: 32 random bytes, lowercase hex. */ + nonce: z.string().regex(NONCE), projection_sha256: sha256, /** The bubblewrap/Landlock `daimon-strict` profile bytes' digest (the projection's `profileSha256`). */ sandbox_profile_sha256: sha256, @@ -57,15 +69,28 @@ export function parseGrokSlotPreflightReceipt(value: unknown): GrokSlotPreflight return result.data; } +/** + * What the evaluator knows about *this* recycle: the nonce it sent, and the + * lowest generation it will accept — one above the last generation it + * accepted for the slot (1 for a slot it has never seen). + */ +export type GrokSlotPreflightFreshness = Readonly<{ expectedNonce: string; minGeneration: number }>; + /** * Parse a receipt and require that it proves *this* projection's slot: same * digest, slot, worker uid, profile and executable, and a denied canary for * exactly every projected deny path (no more, no fewer), under the projected - * seccomp profile and sandbox runtime. + * seccomp profile and sandbox runtime — and that it is *this* recycle's + * receipt: the caller's nonce, at or above the caller's minimum generation. + * A receipt replayed from an earlier recycle fails on the nonce, and one + * whose generation went backwards fails on the generation. */ -export function verifyGrokSlotPreflightReceipt(value: unknown, projection: OrganizationGrokBrokerProjection): GrokSlotPreflightReceipt { +export function verifyGrokSlotPreflightReceipt(value: unknown, projection: OrganizationGrokBrokerProjection, freshness: GrokSlotPreflightFreshness): GrokSlotPreflightReceipt { + if (freshness === null || typeof freshness !== "object" || typeof freshness.expectedNonce !== "string" || !NONCE.test(freshness.expectedNonce) || !Number.isSafeInteger(freshness.minGeneration) || freshness.minGeneration < 1) throw new TypeError("invalid Grok slot preflight freshness"); const receipt = parseGrokSlotPreflightReceipt(value); const mismatch = (member: string): never => { throw new Error(`Grok slot preflight receipt does not match the projection: ${member}`); }; + if (receipt.nonce !== freshness.expectedNonce) throw new Error("Grok slot preflight receipt is stale: nonce"); + if (receipt.generation < freshness.minGeneration) throw new Error("Grok slot preflight receipt is stale: generation"); if (receipt.projection_sha256 !== grokBrokerProjectionSha256(projection)) mismatch("projection_sha256"); if (receipt.slot !== projection.slot) mismatch("slot"); if (receipt.worker_uid !== projection.workerUid) mismatch("worker_uid"); diff --git a/src/runtime/index.ts b/src/runtime/index.ts index 83d00a4..24195c5 100644 --- a/src/runtime/index.ts +++ b/src/runtime/index.ts @@ -7,9 +7,13 @@ export { CODEX_SANDBOX_PROJECTION_VERSION, resolveOrganizationCodexSandboxProjec type OrganizationCodexSandboxProjection } from "./codexSandboxProjection.js"; export { GROK_BROKER_PROJECTION_VERSION, grokBrokerProjectionSha256, grokBrokerServiceRegistrationFor, resolveOrganizationGrokBrokerProjection, verifyGrokBrokerRegistrationMatchesProjection, type OrganizationGrokBrokerProjection, type OrganizationGrokBrokerProjectionOptions } from "./grokBrokerProjection.js"; -export { GROK_SLOT_PREFLIGHT_VERSION, grokSlotPreflightCanarySchema, grokSlotPreflightReceiptSchema, parseGrokSlotPreflightReceipt, +export { GROK_SLOT_PREFLIGHT_VERSION, grokSlotPreflightCanarySchema, grokSlotPreflightReceiptSchema, parseGrokSlotPreflightReceipt, type GrokSlotPreflightFreshness, verifyGrokSlotPreflightReceipt, type GrokSlotPreflightReceipt } from "./grokSlotPreflightReceipt.js"; export { grokBrokerWorkerConfigSha256, renderGrokBrokerWorkerConfig } from "./grokBrokerWorkerConfig.js"; +export { GROK_INFERENCE_CLIENT_MODEL_ID, GROK_INFERENCE_GRANT_ENV, grokInferenceClientConfigSha256, renderGrokInferenceClientConfig, renderProductionGrokInferenceClientConfig, type GrokInferenceClientConfigInput } from "./grokInferenceClientConfig.js"; +export { GROK_INFERENCE_PROXY_BASE_URL, ENGINE_BROKER_INFERENCE_FAILURE_CODES, type EngineBrokerInferenceFailureCode } from "./engineBrokerInferenceProtocol.js"; +export { dedupeInferenceUsageRows, INFERENCE_USAGE_LEDGER_VERSION, GROK_INFERENCE_PURPOSES, type GrokInferencePurpose } from "./inferenceUsageLedger.js"; +export { GROK_INFERENCE_AUTH_STALE_BODY } from "./grokInferenceProxy.js"; export { grokWorkerSandboxProfileSha256, renderGrokWorkerSandboxProfile } from "./grokWorkerSandboxProfile.js"; export { engineBrokerRequestLedgerPathFor, parseEngineBrokerServiceConfig, type EngineBrokerServiceConfig, type EngineBrokerServiceRegistration } from "./engineBrokerServiceConfig.js"; export { DEFAULT_GROK_BROKER_TURN_LIMITS, ENGINE_BROKER_LIMIT_REASONS, type EngineBrokerLimitReason, type EngineBrokerTurnAccounting, @@ -50,3 +54,4 @@ export { type WakeReceiptState } from "./wakeAcceptanceTypes.js"; export type { OrganizationRuntimeControlHost, OrganizationRuntimeControlOptions } from "./organizationRuntimeControl.js"; +export { EngineBrokerControlClient, EngineBrokerInferenceGrantRefused, type EngineBrokerInferenceGrant } from "./engineBrokerControlClient.js"; diff --git a/src/runtime/inferenceUsageLedger.ts b/src/runtime/inferenceUsageLedger.ts new file mode 100644 index 0000000..7a389c3 --- /dev/null +++ b/src/runtime/inferenceUsageLedger.ts @@ -0,0 +1,63 @@ +import { GROK_BROKER_MODELS } from "../contracts/grokWorkerContract.js"; +import { GROK_ENGINE_BROKER } from "../contracts/runtimeContractManifest.js"; +import type { EngineBrokerTurnUsage } from "./engineBrokerTurnAccounting.js"; +import type { GrokBrokerModel } from "./grokBrokerModelPolicy.js"; +import { recordLedgerLines } from "./turnRequestLedger.js"; + +/** + * Evaluator inference spend, one row per upstream model request of one grant. + * + * This is a separate stream at a separate path (`service.json` + * `inferenceLedgerPath`), never the subject usage ledger: the org wake fuse and + * Spawnfile's subject accounting sum that ledger, and a judge's tokens are not + * a subject wake's. Rows carry `kind: "inference"` so a reader that is pointed + * at the wrong file can still tell them apart (`wakeFuse.ts` skips them). + * + * Grants are not sealed in a durable registry: each request is appended when + * it settles. `(grant, request)` identifies a row; readers dedupe on it + * ({@link dedupeInferenceUsageRows}). + */ +export const INFERENCE_USAGE_LEDGER_VERSION = GROK_ENGINE_BROKER.inferenceGrants.ledgerVersion; +export const GROK_INFERENCE_PURPOSES = GROK_ENGINE_BROKER.inferenceGrants.purposes; +export type GrokInferencePurpose = (typeof GROK_INFERENCE_PURPOSES)[number]; + +export type InferenceUsageEntry = Readonly<{ + grant: string; + purpose: GrokInferencePurpose; + model: GrokBrokerModel; + request: number; + usage: EngineBrokerTurnUsage; + usageSource: "upstream" | "estimated"; + startedAt: string; + endedAt: string; + at?: string; +}>; + +export const renderInferenceUsageLine = (entry: InferenceUsageEntry): string => { + if (!/^[a-f0-9]{32}$/u.test(entry.grant) || !(GROK_INFERENCE_PURPOSES as readonly string[]).includes(entry.purpose) || !(GROK_BROKER_MODELS as readonly string[]).includes(entry.model) || !Number.isSafeInteger(entry.request) || entry.request < 0) throw new TypeError("invalid inference usage entry"); + const { usage } = entry; + return `${JSON.stringify({ + v: INFERENCE_USAGE_LEDGER_VERSION, kind: "inference", purpose: entry.purpose, grant: entry.grant, request: entry.request, + at: entry.at ?? new Date().toISOString(), started_at: entry.startedAt, ended_at: entry.endedAt, model: entry.model, + input: usage.input, cache_read: usage.cacheRead, cache_write: usage.cacheWrite, output: usage.output, total: usage.total, + usage_source: entry.usageSource + })}\n`; +}; + +/** Advisory, never rejects: an evaluator request that already spent tokens must not fail on its ledger. */ +export const recordInferenceUsage = async (file: string, entry: InferenceUsageEntry): Promise => { + let line: string; + try { line = renderInferenceUsageLine(entry); } catch { return false; } + return recordLedgerLines(file, line); +}; + +/** Keeps the first row of each `(grant, request)`; rows without both keys are not inference rows and are dropped. */ +export const dedupeInferenceUsageRows = >(rows: readonly T[]): T[] => { + const seen = new Set(); + return rows.filter((row) => { + if (typeof row.grant !== "string" || typeof row.request !== "number") return false; + const key = `${row.grant}\0${row.request}`; + if (seen.has(key)) return false; + seen.add(key); return true; + }); +}; diff --git a/src/runtime/wakeFuse.ts b/src/runtime/wakeFuse.ts index d5d00f4..04c8038 100644 --- a/src/runtime/wakeFuse.ts +++ b/src/runtime/wakeFuse.ts @@ -206,7 +206,9 @@ async function sumTokens(ledgerPath: string, since: string, agentId?: string): P for (const file of [`${ledgerPath}.1`, ledgerPath]) { for (const line of await lines(file)) { try { - const value = JSON.parse(line) as { at?: unknown; total?: unknown; agent?: unknown; turn?: unknown }; + const value = JSON.parse(line) as { at?: unknown; total?: unknown; agent?: unknown; turn?: unknown; kind?: unknown }; + // Evaluator inference rows belong to their own ledger; one here (a misconfigured path) is never subject spend. + if (value.kind === "inference") continue; if (typeof value.turn === "string") { if (turns.has(value.turn)) continue; turns.add(value.turn); } if ((agentId === undefined || value.agent === agentId) && typeof value.at === "string" && !Number.isNaN(Date.parse(value.at)) && value.at >= since && typeof value.total === "number" && Number.isFinite(value.total) && value.total >= 0) total += value.total; } catch { /* usage accounting is advisory input; malformed lines are skipped */ }