From 8e828526b5d798e257c30cf8e41fab3204a3c89c Mon Sep 17 00:00:00 2001 From: Tapan Chugh Date: Thu, 6 Aug 2026 16:46:24 -0700 Subject: [PATCH] refactor(cluster,evals): one Kubernetes call family, one diagnostic bound, one image-producer fact MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every Kubernetes call in the simulator now goes through the bounded `kubernetesCall` family, so PR #978's stated invariant holds for the society API's workload, secret, and sandbox calls too. The society seam keeps `ClusterError` as its public error type through one adapter. The evals-side diagnostic bound was structurally dead — the submitter already publishes at most 8192 UTF-8 bytes, so a naive 8192-code-unit slice could only split surrogate pairs — and is gone. The submitter's byte-aware bound is the single authority; the worker's two layers now name their units. `ledgerAllocationFailed` and `runInfrastructureFailed` were the same function twice; they merge into one constructor keyed on the summary `_tag`, and the sequential string-compare dispatch reduces to one. The image-producer fact is one descriptor with the controller path written once, and `EvaluationImageKey` is derived from it. Also: the invisible-workspace-file list is computed once at definition time; the `McpServer` structural narrow is one exported matcher rather than three `"url" in server` sites; the worker-roll visibility query quotes the operator-supplied task queue and stops interpolating a constant status; `INSTALL_ORDER` derives its membership from `RunWorkerManifests` so a new object fails compile; and the cluster.sh prelude assertion checks definition and use rather than an occurrence count. Co-Authored-By: Claude Opus 5 --- packages/evals/src/cli.test.ts | 77 ++++---- packages/evals/src/cli.ts | 124 +++++------- packages/evals/src/submission.test.ts | 13 +- packages/evals/src/submission.ts | 9 +- packages/simulator/gke/profile.test.mjs | 5 +- .../simulator/nanoclaw-image/entrypoint.mjs | 3 - .../scripts/build-nanoclaw-image.mjs | 3 +- .../src/agents/openclaw/configuration.ts | 8 +- .../simulator/src/agents/openclaw/runtime.ts | 16 +- packages/simulator/src/agents/workspace.ts | 52 +++-- packages/simulator/src/cluster/install.ts | 18 +- .../simulator/src/cluster/kubernetes/calls.ts | 183 +++++++++--------- .../simulator/src/cluster/temporal.test.ts | 24 ++- packages/simulator/src/cluster/temporal.ts | 19 +- packages/simulator/src/cluster/watch.ts | 18 +- 15 files changed, 286 insertions(+), 286 deletions(-) diff --git a/packages/evals/src/cli.test.ts b/packages/evals/src/cli.test.ts index 0f821e94..d648edfb 100644 --- a/packages/evals/src/cli.test.ts +++ b/packages/evals/src/cli.test.ts @@ -7,10 +7,9 @@ import { IncompleteLedgerReceipt } from "@moltzap/simulator"; import { ledgerRef } from "@moltzap/simulator/ledger"; import { evaluationCase } from "./cases.js"; import { + infrastructureFailed, invalidImageDetail, - ledgerAllocationFailed, missingImageDetail, - runInfrastructureFailed, type AttemptContext, type EvaluationImageKey, } from "./cli.js"; @@ -94,45 +93,37 @@ effect.each([ }).pipe(Effect.provide(NodeContext.layer)), ); +const CLUSTER_LOST = { _tag: "ClusterLost", receipt: RECEIPT } as const; +const ALLOCATION_FAILED = { _tag: "LedgerAllocationFailed" } as const; + +// The operator-facing account either infrastructure attempt carries. +function attemptAccount( + attempt: Effect.Effect.Success>, +): string { + return attempt._tag === "RunFailedAttempt" + ? attempt.detail + : attempt.failure.detail; +} + // Both infrastructure attempts already carry the operator-facing account of a // failure, and phoenix-run publishes exactly that field as the run's error, so // the controller's own account belongs in it rather than beside it. -effect.each([undefined, DIAGNOSTIC])( - "gives a failed run the controller account %s", - (diagnostic?: string) => - Effect.gen(function* () { - const attempt = yield* runInfrastructureFailed( - attemptContext(), - RECEIPT, - diagnostic, - ); - - assert.strictEqual(attempt.detail, diagnostic ?? attempt.detail); - assert.isNotEmpty(attempt.detail); - if (diagnostic === undefined) { - assert.include(attempt.detail, "infrastructure failure"); - } - }), -); - -effect.each([undefined, DIAGNOSTIC])( - "gives a failed allocation the controller account %s", - (diagnostic?: string) => - Effect.gen(function* () { - const attempt = yield* ledgerAllocationFailed( - attemptContext(), - diagnostic, - ); +effect.each([ + ["a lost cluster", CLUSTER_LOST, "infrastructure failure"], + ["a failed allocation", ALLOCATION_FAILED, "durable ledger"], +] as const)("gives %s the controller account", ([, summary, canned]) => + Effect.gen(function* () { + const context = attemptContext(); - assert.strictEqual( - attempt.failure.detail, - diagnostic ?? attempt.failure.detail, - ); - assert.isNotEmpty(attempt.failure.detail); - if (diagnostic === undefined) { - assert.include(attempt.failure.detail, "durable ledger"); - } - }), + assert.strictEqual( + attemptAccount(yield* infrastructureFailed(context, summary, DIAGNOSTIC)), + DIAGNOSTIC, + ); + assert.include( + attemptAccount(yield* infrastructureFailed(context, summary)), + canned, + ); + }), ); // Without a controller account the two attempts still have to be told apart: @@ -141,8 +132,14 @@ effect.each([undefined, DIAGNOSTIC])( it("distinguishes the two infrastructure failures when neither left an account", () => Effect.gen(function* () { const context = attemptContext(); - const failedRun = yield* runInfrastructureFailed(context, RECEIPT); - const failedAllocation = yield* ledgerAllocationFailed(context); + const failedRun = yield* infrastructureFailed(context, CLUSTER_LOST); + const failedAllocation = yield* infrastructureFailed( + context, + ALLOCATION_FAILED, + ); - assert.notStrictEqual(failedRun.detail, failedAllocation.failure.detail); + assert.notStrictEqual( + attemptAccount(failedRun), + attemptAccount(failedAllocation), + ); })); diff --git a/packages/evals/src/cli.ts b/packages/evals/src/cli.ts index 239d1c39..71133a6b 100644 --- a/packages/evals/src/cli.ts +++ b/packages/evals/src/cli.ts @@ -4,11 +4,7 @@ import { Command as CliCommand, Options } from "@effect/cli"; import { Command, Path } from "@effect/platform"; import { NodeContext, NodeRuntime } from "@effect/platform-node"; -import { - isEntryModule, - type CompletedLedgerReceipt, - type LedgerReceipt, -} from "@moltzap/simulator"; +import { isEntryModule, type CompletedLedgerReceipt } from "@moltzap/simulator"; import { LedgerStorageError, type CompletedLedgerArtifacts, @@ -489,58 +485,50 @@ function completeExecution( }); } -// Both infrastructure attempts already own a `detail`, and Phoenix already -// publishes it as a run's error. The controller's own account of the failure -// belongs there rather than beside it: a canned sentence is what an operator -// reads today, and it says only that something went wrong. -const ALLOCATION_FAILED_DETAIL = - "the simulator controller could not allocate its durable ledger"; -const RUN_FAILED_DETAIL = - "the simulator controller reported an infrastructure failure"; - -/** - * Record that ledger allocation failed, with whatever account the run left. - * @param context Cell identity and the instant execution began. - * @param diagnostic The controller's own account, when it produced one. - * @returns The terminal attempt this cell commits. - */ -export function ledgerAllocationFailed( - context: AttemptContext, - diagnostic?: string, -) { - return DateTime.now.pipe( - Effect.map((completedAt) => - LedgerAllocationFailedAttempt.make({ - ...terminalFields(context, completedAt), - failure: LedgerStorageError.make({ - operation: "allocate", - detail: diagnostic ?? ALLOCATION_FAILED_DETAIL, - }), - }), - ), - ); -} +/** Summary of a submission that never reached a gradeable run. */ +type InfrastructureSummary = Exclude< + EvaluationSubmissionResult["result"]["summary"], + { readonly _tag: "ProgramFinished" } +>; + +// A run that never got a ledger and a run that lost its cluster are different +// operator problems, and this text is all that says which happened when the +// controller left no account of its own. +const INFRASTRUCTURE_FAILED_DETAIL: Readonly< + Record +> = { + LedgerAllocationFailed: + "the simulator controller could not allocate its durable ledger", + ClusterLost: "the simulator controller reported an infrastructure failure", +}; /** * Record an infrastructure failure, with whatever account the run left. * @param context Cell identity and the instant execution began. - * @param receipt Durable evidence the controller retained. + * @param summary Terminal summary the controller printed for this cell. * @param diagnostic The controller's own account, when it produced one. * @returns The terminal attempt this cell commits. */ -export function runInfrastructureFailed( +export function infrastructureFailed( context: AttemptContext, - receipt: LedgerReceipt, + summary: InfrastructureSummary, diagnostic?: string, ) { return DateTime.now.pipe( - Effect.map((completedAt) => - RunFailedAttempt.make({ - ...terminalFields(context, completedAt), - receipt, - detail: diagnostic ?? RUN_FAILED_DETAIL, - }), - ), + Effect.map((completedAt) => { + const detail = diagnostic ?? INFRASTRUCTURE_FAILED_DETAIL[summary._tag]; + const fields = terminalFields(context, completedAt); + return summary._tag === "LedgerAllocationFailed" + ? LedgerAllocationFailedAttempt.make({ + ...fields, + failure: LedgerStorageError.make({ operation: "allocate", detail }), + }) + : RunFailedAttempt.make({ + ...fields, + receipt: summary.receipt, + detail, + }); + }), ); } @@ -601,19 +589,14 @@ function completeSubmission( submission: EvaluationSubmissionResult, ) { const summary = submission.result.summary; - const diagnostic = submissionDiagnostic(submission); - if (summary._tag === "LedgerAllocationFailed") { - return ledgerAllocationFailed(context, diagnostic); - } - if (summary._tag === "ClusterLost") { - return runInfrastructureFailed(context, summary.receipt, diagnostic); - } - return readCompletedArtifacts( - environment, - context, - submission.namespace, - summary.receipt, - ); + return summary._tag === "ProgramFinished" + ? readCompletedArtifacts( + environment, + context, + submission.namespace, + summary.receipt, + ) + : infrastructureFailed(context, summary, submissionDiagnostic(submission)); } function conditionModelId( @@ -742,22 +725,20 @@ function requiredEnvironment(key: string) { ); } -/** Environment key naming one digest-pinned image an evaluation run needs. */ -export type EvaluationImageKey = - | "MOLTZAP_CONTROLLER_IMAGE" - | "MOLTZAP_SUPPORT_IMAGE" - | "MOLTZAP_NANOCLAW_IMAGE"; +const CONTROLLER_IMAGE_PRODUCER = + "packages/simulator/scripts/build-controller-image.mjs"; // Nothing else in the repository produces these references, so a missing one is // an operator who has not run the producer yet rather than one who forgot to // export a value they already had. Naming the producer is the whole remedy. -const imageProducer: Readonly> = { - MOLTZAP_CONTROLLER_IMAGE: - "packages/simulator/scripts/build-controller-image.mjs", - MOLTZAP_SUPPORT_IMAGE: - "packages/simulator/scripts/build-controller-image.mjs", +const imageProducer = { + MOLTZAP_CONTROLLER_IMAGE: CONTROLLER_IMAGE_PRODUCER, + MOLTZAP_SUPPORT_IMAGE: CONTROLLER_IMAGE_PRODUCER, MOLTZAP_NANOCLAW_IMAGE: "packages/simulator/scripts/build-nanoclaw-image.mjs", -}; +} as const; + +/** Environment key naming one digest-pinned image an evaluation run needs. */ +export type EvaluationImageKey = keyof typeof imageProducer; /** * Say which producer builds an evaluation image the environment omitted. @@ -1013,8 +994,7 @@ const cli = CliCommand.run(evaluationCommand, { }); // Guarded so this module can be imported: without it, reading any value here -// runs the CLI against the importer's argv, which is why the operator-facing -// vocabulary above had to live in a module that does not read it. +// runs the CLI against the importer's argv. // eslint-disable-next-line agent-code-guard/prefer-effect-platform -- Direct-entry detection has no Effect Platform equivalent. if (isEntryModule(import.meta.url, process.argv[1])) { // eslint-disable-next-line agent-code-guard/prefer-effect-platform -- @effect/cli owns argv decoding at the process boundary. diff --git a/packages/evals/src/submission.test.ts b/packages/evals/src/submission.test.ts index 56b74b8b..997a1c11 100644 --- a/packages/evals/src/submission.test.ts +++ b/packages/evals/src/submission.test.ts @@ -140,16 +140,11 @@ effect.each([undefined, CARRIED_DIAGNOSTIC])( // The rest of that line is the run's only receipt. Refusing an over-long // diagnostic would discard it, turning a cell that failed with evidence into a // cell with no attempt at all. -it("keeps the receipt when a submitter diagnostic exceeds the bound", () => +it("keeps the receipt when a submitter diagnostic is over-long", () => Effect.gen(function* () { - const decoded = yield* decodeSubmissionOutput( - submitterLine("x".repeat(OVERSIZED_DIAGNOSTIC_LENGTH)), - ); + const oversized = "x".repeat(OVERSIZED_DIAGNOSTIC_LENGTH); + const decoded = yield* decodeSubmissionOutput(submitterLine(oversized)); - // The receipt survives; only the decoration is trimmed. assert.isTrue("receipt" in decoded.result.summary); - assert.isBelow( - (submissionDiagnostic(decoded) ?? "").length, - OVERSIZED_DIAGNOSTIC_LENGTH, - ); + assert.strictEqual(submissionDiagnostic(decoded), oversized); })); diff --git a/packages/evals/src/submission.ts b/packages/evals/src/submission.ts index 2f126a1f..d3ef4d48 100644 --- a/packages/evals/src/submission.ts +++ b/packages/evals/src/submission.ts @@ -44,11 +44,6 @@ const runInfrastructureFailedSummary = Schema.Struct({ const ledgerAllocationFailedSummary = Schema.Struct({ _tag: Schema.Literal("LedgerAllocationFailed"), }); -// Bounded here rather than in the schema, because the diagnostic is decoration -// on a line whose other fields are the run's only receipt: a producer that -// overshoots must not be able to take the receipt down with it. The submitter -// bounds what it publishes; this bounds what a report will hold. -const DIAGNOSTIC_MAX_LENGTH = 8_192; const evaluationSubmissionResult = Schema.Struct({ runId: Schema.NonEmptyString, namespace: Schema.NonEmptyString, @@ -76,7 +71,7 @@ export type EvaluationSubmissionResult = typeof evaluationSubmissionResult.Type; /** * The controller's own account of why one submission failed. * @param submission Decoded submitter result for one cell. - * @returns Bounded diagnostic text, or undefined when the cell carried none. + * @returns The diagnostic the submitter published, or undefined when it carried none. */ export function submissionDiagnostic( submission: EvaluationSubmissionResult, @@ -85,7 +80,7 @@ export function submissionDiagnostic( submission.result.exitCode === 1 ? submission.result.diagnostic : undefined; return diagnostic === undefined || diagnostic.length === 0 ? undefined - : diagnostic.slice(-DIAGNOSTIC_MAX_LENGTH); + : diagnostic; } /** A repository-local cell could not be submitted or decoded. */ diff --git a/packages/simulator/gke/profile.test.mjs b/packages/simulator/gke/profile.test.mjs index de8071aa..910cd239 100644 --- a/packages/simulator/gke/profile.test.mjs +++ b/packages/simulator/gke/profile.test.mjs @@ -251,8 +251,9 @@ test("the evals verb hands the sweep every identity it cannot derive", async () // The forward supervisor is a background job of this shell; exec would strand // it with no trap left to reap it. assert.doesNotMatch(script, /exec corepack/); - // One prelude, so a trap fix cannot reach `run` and miss `evals`. - assert.equal(script.match(/begin_cluster_session/g)?.length, 3); + // One prelude, shared rather than re-inlined per verb. + assert.match(script, /^begin_cluster_session\(\) \{$/m); + assert.match(script, /^\s+begin_cluster_session$/m); assert.equal( script.match(/open_temporal_forward "\$forward_port"/g)?.length, 1, diff --git a/packages/simulator/nanoclaw-image/entrypoint.mjs b/packages/simulator/nanoclaw-image/entrypoint.mjs index aa64f061..7f82c241 100644 --- a/packages/simulator/nanoclaw-image/entrypoint.mjs +++ b/packages/simulator/nanoclaw-image/entrypoint.mjs @@ -66,9 +66,6 @@ async function readBootstrapConfig() { return config; } -// Concurrently, because every entry targets a distinct path and this runs -// before the bridge port opens — which is what the controller reads as -// readiness, so anything serialized here is startup latency for the whole run. async function materializeProjectRoot(config) { const projectRoot = config.stateDirectory ?? requiredEnvironment("MOLTZAP_NANOCLAW_STATE"); diff --git a/packages/simulator/scripts/build-nanoclaw-image.mjs b/packages/simulator/scripts/build-nanoclaw-image.mjs index f695cceb..cc83764f 100644 --- a/packages/simulator/scripts/build-nanoclaw-image.mjs +++ b/packages/simulator/scripts/build-nanoclaw-image.mjs @@ -71,8 +71,7 @@ const workspacePackages = { */ export function assertRepository(repository) { // The repository half excludes `@` so a trailing digest cannot be smuggled in - // behind an earlier one — the same reason the image schema excludes it. One - // rule, checked when the argument arrives rather than only after the build. + // behind an earlier one — the same reason the image schema excludes it. if (repository.length === 0 || /[@\s]/.test(repository)) { throw new TypeError( "a nanoclaw image repository must be nonempty and carry no digest", diff --git a/packages/simulator/src/agents/openclaw/configuration.ts b/packages/simulator/src/agents/openclaw/configuration.ts index 453816be..f31ae511 100644 --- a/packages/simulator/src/agents/openclaw/configuration.ts +++ b/packages/simulator/src/agents/openclaw/configuration.ts @@ -8,7 +8,11 @@ import type { ToolsConfig, } from "openclaw/plugin-sdk/config-types"; import { Redacted } from "effect"; -import { SIMULATOR_PROFILE_NAME, type McpServer } from "../workspace.js"; +import { + isHttpMcpServer, + SIMULATOR_PROFILE_NAME, + type McpServer, +} from "../workspace.js"; const DEFAULT_OPENCLAW_MODEL_ID = "openai/gpt-5.5"; const OPENCLAW_CHANNEL_ID = "moltzap" satisfies MoltzapChannelPlugin["id"]; @@ -42,7 +46,7 @@ function mcpConfigSection( servers: Object.fromEntries( mcpServers.map((server) => [ server.name, - "url" in server + isHttpMcpServer(server) ? { transport: "streamable-http" as const, url: server.url } : { transport: "stdio" as const, diff --git a/packages/simulator/src/agents/openclaw/runtime.ts b/packages/simulator/src/agents/openclaw/runtime.ts index 39a0586b..51feb2a9 100644 --- a/packages/simulator/src/agents/openclaw/runtime.ts +++ b/packages/simulator/src/agents/openclaw/runtime.ts @@ -123,6 +123,8 @@ export interface OpenClawRuntimeOptions { interface OpenClawRuntimeSettings { readonly startupTimeout: Duration.Duration; readonly workspaceFiles: readonly CheckedWorkspaceFile[]; + /** Declared workspace files outside OpenClaw's context-injection set. */ + readonly invisibleWorkspaceFiles: readonly string[]; readonly modelId?: string; readonly mcpServers?: readonly McpServer[]; readonly tools?: OpenClawToolsConfig; @@ -143,13 +145,12 @@ function snapshotOptions( ): OpenClawRuntimeSettings { const workspaceFiles = snapshotWorkspaceFiles(options.workspaceFiles); const tools = snapshotNativeConfiguration(options.tools); - assertWorkspaceFilesReachable( - workspaceFiles, - tools?.deny?.includes("*") ?? false, - ); + const invisible = invisibleWorkspaceFiles(workspaceFiles); + assertWorkspaceFilesReachable(invisible, tools?.deny?.includes("*") ?? false); return Object.freeze({ startupTimeout: options.startupTimeout ?? DEFAULT_OPENCLAW_STARTUP_TIMEOUT, workspaceFiles, + invisibleWorkspaceFiles: invisible, modelId: options.modelId, mcpServers: snapshotMcpServers(options.mcpServers), tools, @@ -200,14 +201,13 @@ function invisibleWorkspaceFiles( * tool could ever read them either. Other deny shapes may also block every * read, but only the wildcard is knowable without interpreting native policy; * those shapes intentionally degrade to the acquisition-time warning. - * @param files Checked workspace files the definition declares. + * @param invisible Declared workspace paths outside the context-injection set. * @param denyListIsWildcard Whether the native deny list is exactly the wildcard. */ function assertWorkspaceFilesReachable( - files: readonly CheckedWorkspaceFile[], + invisible: readonly string[], denyListIsWildcard: boolean, ): void { - const invisible = invisibleWorkspaceFiles(files); if (invisible.length > 0 && denyListIsWildcard) { throw AgentRuntimeDefinitionError.make({ detail: @@ -428,7 +428,7 @@ function makeOpenClawApplication( gatewayToken, deviceIdentity: pairing.deviceIdentity, acquireGateway, - invisibleWorkspaceFiles: invisibleWorkspaceFiles(settings.workspaceFiles), + invisibleWorkspaceFiles: settings.invisibleWorkspaceFiles, }; return Object.freeze({ entrypoint: Object.freeze([ diff --git a/packages/simulator/src/agents/workspace.ts b/packages/simulator/src/agents/workspace.ts index a4ada936..05777800 100644 --- a/packages/simulator/src/agents/workspace.ts +++ b/packages/simulator/src/agents/workspace.ts @@ -107,6 +107,17 @@ interface HttpMcpServer { */ export type McpServer = StdioMcpServer | HttpMcpServer; +/** + * Whether one MCP server is reached over streamable HTTP rather than stdio. + * @param server MCP server whose transport is being decided. + * @returns Whether the server carries a remote URL. + */ +export function isHttpMcpServer( + server: McpServer, +): server is Extract { + return "url" in server; +} + const decodeWorkspaceRelativePath = Schema.decodeUnknownSync( workspaceRelativePath, ); @@ -184,7 +195,7 @@ export function snapshotMcpServers( : Object.freeze( servers.map((server) => Object.freeze( - "url" in server + isHttpMcpServer(server) ? { name: server.name, url: decodeMcpServerUrl(server.url) } : { name: server.name, @@ -235,26 +246,25 @@ export function workspaceConfiguration( * @returns The sanitized MCP server record. */ function sanitizedMcpServer(server: McpServer): McpServerConfiguration { - const { definition, redacted } = - "url" in server - ? { - definition: JSON.stringify({ - name: server.name, - origin: new URL(server.url).origin, - }), - redacted: ["url"] as const, - } - : { - definition: JSON.stringify({ - name: server.name, - command: server.command, - args: server.args, - environmentKeys: Object.keys(server.env).sort((left, right) => - left.localeCompare(right), - ), - }), - redacted: ["command", "args", "environmentValues"] as const, - }; + const { definition, redacted } = isHttpMcpServer(server) + ? { + definition: JSON.stringify({ + name: server.name, + origin: new URL(server.url).origin, + }), + redacted: ["url"] as const, + } + : { + definition: JSON.stringify({ + name: server.name, + command: server.command, + args: server.args, + environmentKeys: Object.keys(server.env).sort((left, right) => + left.localeCompare(right), + ), + }), + redacted: ["command", "args", "environmentValues"] as const, + }; return McpServerConfiguration.make({ name: server.name, definitionDigest: digestText(definition), diff --git a/packages/simulator/src/cluster/install.ts b/packages/simulator/src/cluster/install.ts index 89c900cd..7df29663 100644 --- a/packages/simulator/src/cluster/install.ts +++ b/packages/simulator/src/cluster/install.ts @@ -19,13 +19,17 @@ export const FORCE_WORKER_ROLL_VARIABLE = "MOLTZAP_FORCE_WORKER_ROLL"; // service account cannot delete a run namespace, which is the one thing the // worker exists to do, and Kubernetes reports that as a permission error on a // run rather than as a failed install. -const INSTALL_ORDER: readonly RunWorkerObject[] = [ - "namespace", - "serviceAccount", - "clusterRole", - "clusterRoleBinding", - "deployment", -]; +const RUN_WORKER_OBJECTS = { + namespace: true, + serviceAccount: true, + clusterRole: true, + clusterRoleBinding: true, + deployment: true, +} satisfies Readonly>; + +const INSTALL_ORDER = + /* Safe because the record above is exhaustive and keeps its declared order. */ + Object.keys(RUN_WORKER_OBJECTS) as readonly RunWorkerObject[]; /** What the run-lifecycle task queue says about work it has not finished. */ export type OpenRunReading = diff --git a/packages/simulator/src/cluster/kubernetes/calls.ts b/packages/simulator/src/cluster/kubernetes/calls.ts index 863353d0..e40366fe 100644 --- a/packages/simulator/src/cluster/kubernetes/calls.ts +++ b/packages/simulator/src/cluster/kubernetes/calls.ts @@ -19,7 +19,7 @@ import { type V1JobCondition, } from "@kubernetes/client-node"; import { Cause, Duration, Effect, Schema } from "effect"; -import { clusterError, type ClusterError } from "../cluster.js"; +import { ClusterError, clusterError } from "../cluster.js"; import type { KubernetesExecutionProfile } from "../profile.js"; import type { RunSocietyWorkflowInput } from "../reclaim.js"; import { @@ -80,6 +80,84 @@ const APPLIED = Object.freeze({ fieldValidation: "Strict", } as const); +// A call the bound cut off is reported as never answered, not as failed: the +// cluster refusing an object and the cluster never replying at all are +// different operator problems, and only one of them is about the object. A +// status of zero is no status, which no Kubernetes response carries. +function callDetail( + operation: string, + status: number, + unanswered: boolean, +): string { + if (unanswered) { + return `${operation} did not answer in time`; + } + return status === 0 + ? `${operation} failed` + : `${operation} failed (Kubernetes ${String(status)})`; +} + +/** Failure of one Kubernetes call, carrying the status but never the body. */ +export class KubernetesCallFailed extends Error { + override readonly name = "KubernetesCallFailed"; + + /** Whether the cluster answered that the object is not there. */ + readonly absent: boolean; + + constructor(operation: string, cause?: unknown) { + const status = cause instanceof ApiException ? cause.code : 0; + super( + callDetail(operation, status, cause instanceof Cause.TimeoutException), + ); + this.absent = status === ABSENT; + } +} + +/** + * Make one Kubernetes API call, bounded so that it always ends. + * + * The client's own request has no deadline, so an API server that accepts the + * connection and then answers nothing — a control plane being repaired, a + * tunnel that went away without resetting — leaves the caller waiting forever + * with no output naming what it is waiting for. A submission that fails after + * the bound is a submission the operator can act on. + * + * @param operation What this call was doing, as the operator's failure names it. + * @param evaluate The client call, already bound to its request. + * @param bound How long the cluster has to answer before the call is abandoned. + * @returns The call's result, or a failure naming the operation. + * @failure KubernetesCallFailed when the call is refused or never answered. + */ +export function kubernetesCall( + operation: string, + evaluate: () => PromiseLike, + bound: Duration.Duration = KUBERNETES_CALL_TIMEOUT, +): Effect.Effect { + return Effect.tryPromise({ + try: evaluate, + catch: (cause) => new KubernetesCallFailed(operation, cause), + }).pipe( + Effect.timeoutFail({ + duration: bound, + onTimeout: () => + new KubernetesCallFailed(operation, new Cause.TimeoutException()), + }), + ); +} + +function attemptUnlessAbsent( + operation: string, + evaluate: () => PromiseLike, +): Effect.Effect { + return kubernetesCall(operation, evaluate).pipe( + Effect.catchIf( + (failure) => failure.absent, + () => Effect.void, + ), + Effect.asVoid, + ); +} + const KUEUE_GROUP = "kueue.x-k8s.io"; const KUEUE_VERSION = "v1beta2"; const KUEUE_WORKLOADS = "workloads"; @@ -218,11 +296,15 @@ export interface KubernetesSocietyApi { ) => Effect.Effect; } +// The same call failure, at the public error type of the cluster seam. +function societyFailure(failure: KubernetesCallFailed): ClusterError { + return new ClusterError({ detail: failure.message }); +} + function request(operation: string, evaluate: () => PromiseLike) { - return Effect.tryPromise({ - try: evaluate, - catch: (cause) => clusterError(operation, cause), - }); + return kubernetesCall(operation, evaluate).pipe( + Effect.mapError(societyFailure), + ); } function decode( @@ -239,17 +321,8 @@ function ignoreAbsent( operation: string, evaluate: () => PromiseLike, ): Effect.Effect { - return Effect.tryPromise({ - try: evaluate, - catch: (cause) => - cause instanceof ApiException && cause.code === ABSENT - ? undefined - : clusterError(operation, cause), - }).pipe( - Effect.catchAll((failure) => - failure === undefined ? Effect.void : Effect.fail(failure), - ), - Effect.asVoid, + return attemptUnlessAbsent(operation, evaluate).pipe( + Effect.mapError(societyFailure), ); } @@ -540,39 +613,6 @@ function installedWorkerImage(deployment: { /** One installable member of the cluster's run-worker control plane. */ export type RunWorkerObject = keyof RunWorkerManifests; -// A call the bound cut off is reported as never answered, not as failed: the -// cluster refusing an object and the cluster never replying at all are -// different operator problems, and only one of them is about the object. A -// status of zero is no status, which no Kubernetes response carries. -function callDetail( - operation: string, - status: number, - unanswered: boolean, -): string { - if (unanswered) { - return `${operation} did not answer in time`; - } - return status === 0 - ? `${operation} failed` - : `${operation} failed (Kubernetes ${String(status)})`; -} - -/** Failure of one Kubernetes call, carrying the status but never the body. */ -export class KubernetesCallFailed extends Error { - override readonly name = "KubernetesCallFailed"; - - /** Whether the cluster answered that the object is not there. */ - readonly absent: boolean; - - constructor(operation: string, cause?: unknown) { - const status = cause instanceof ApiException ? cause.code : 0; - super( - callDetail(operation, status, cause instanceof Cause.TimeoutException), - ); - this.absent = status === ABSENT; - } -} - /** Kubernetes access the Temporal activity needs for one run's lifetime. */ export interface RunControlApi { /** Create the run's Namespace and immutable owner; yields the owner UID. */ @@ -644,51 +684,6 @@ export interface RunWorkerInstallApi { readonly wait: (milliseconds: number) => Effect.Effect; } -/** - * Make one Kubernetes API call, bounded so that it always ends. - * - * The client's own request has no deadline, so an API server that accepts the - * connection and then answers nothing — a control plane being repaired, a - * tunnel that went away without resetting — leaves the caller waiting forever - * with no output naming what it is waiting for. A submission that fails after - * the bound is a submission the operator can act on. - * - * @param operation What this call was doing, as the operator's failure names it. - * @param evaluate The client call, already bound to its request. - * @param bound How long the cluster has to answer before the call is abandoned. - * @returns The call's result, or a failure naming the operation. - * @failure KubernetesCallFailed when the call is refused or never answered. - */ -export function kubernetesCall( - operation: string, - evaluate: () => PromiseLike, - bound: Duration.Duration = KUBERNETES_CALL_TIMEOUT, -): Effect.Effect { - return Effect.tryPromise({ - try: evaluate, - catch: (cause) => new KubernetesCallFailed(operation, cause), - }).pipe( - Effect.timeoutFail({ - duration: bound, - onTimeout: () => - new KubernetesCallFailed(operation, new Cause.TimeoutException()), - }), - ); -} - -function attemptUnlessAbsent( - operation: string, - evaluate: () => PromiseLike, -): Effect.Effect { - return kubernetesCall(operation, evaluate).pipe( - Effect.catchIf( - (failure) => failure.absent, - () => Effect.void, - ), - Effect.asVoid, - ); -} - interface RunControlClients { readonly batch: BatchV1Api; readonly core: CoreV1Api; diff --git a/packages/simulator/src/cluster/temporal.test.ts b/packages/simulator/src/cluster/temporal.test.ts index d727d270..7aac96ac 100644 --- a/packages/simulator/src/cluster/temporal.test.ts +++ b/packages/simulator/src/cluster/temporal.test.ts @@ -18,7 +18,7 @@ import type { import { executeRunSocietyWorkflow, LifecycleOperations, - OPEN_RUN_STATUS, + OPEN_RUN_FILTER, readOpenRuns, runLifecycleActivities, type ControllerObservation, @@ -224,8 +224,26 @@ describe("readOpenRuns", () => { await expect( Effect.runPromise(readOpenRuns(client, taskQueue)), ).resolves.toEqual({ _tag: "open", workflowIds: openRunIds }); - expect(queries[0]).toContain(taskQueue); - expect(queries[0]).toContain(OPEN_RUN_STATUS); + expect(queries[0]).toContain(`'${taskQueue}'`); + expect(queries[0]).toContain(OPEN_RUN_FILTER); + }); + + // A queue name is operator-supplied, so a quote in it must not be able to end + // the query's own literal and change which runs are counted. + it("quotes a task queue name that carries a quote", async () => { + const injected = `${taskQueue}' OR '1'='1`; + const queries: string[] = []; + const client: OpenRunLister = { + list: (options) => { + queries.push(options.query); + return listed([]); + }, + }; + + await Effect.runPromise(readOpenRuns(client, injected)); + + expect(queries[0]).toContain(`'${injected.replaceAll("'", "''")}'`); + expect(queries[0]).toContain(OPEN_RUN_FILTER); }); // The distinction the roll guard depends on: a queue that cannot be listed diff --git a/packages/simulator/src/cluster/temporal.ts b/packages/simulator/src/cluster/temporal.ts index 9d498e62..9777d1c0 100644 --- a/packages/simulator/src/cluster/temporal.ts +++ b/packages/simulator/src/cluster/temporal.ts @@ -46,9 +46,7 @@ const DEFAULT_TEMPORAL_NAMESPACE = "default"; * Coarse controller state observed by the host-side activity. * * `completed` is every controller that produced a decodable result, whether or - * not the run inside it succeeded — the activity returns both the same way. The - * two were one state with an optional result and a detail that was dead - * whenever the result was present, which invited them to disagree. + * not the run inside it succeeded — the activity returns both the same way. */ export type ControllerObservation = | { readonly _tag: "running" } @@ -365,8 +363,17 @@ export async function executeRunSocietyWorkflow( /** The most open runs one reading names before it stops counting. */ const OPEN_RUN_SAMPLE = 20; -/** Temporal execution status of a run its task queue has not finished. */ -export const OPEN_RUN_STATUS = "Running"; +/** Visibility-query clause matching a run its task queue has not finished. */ +export const OPEN_RUN_FILTER = "ExecutionStatus = 'Running'"; + +/** + * One operator-supplied value as a visibility-query string literal. + * @param value Value the query compares a field against. + * @returns The quoted literal, with any quote inside it doubled. + */ +function queryLiteral(value: string): string { + return `'${value.replaceAll("'", "''")}'`; +} /** * The exact listing surface reading a queue's unfinished work needs. @@ -391,7 +398,7 @@ async function listOpenRunIds( ): Promise { const workflowIds: string[] = []; for await (const execution of client.list({ - query: `TaskQueue = '${taskQueue}' AND ExecutionStatus = '${OPEN_RUN_STATUS}'`, + query: `TaskQueue = ${queryLiteral(taskQueue)} AND ${OPEN_RUN_FILTER}`, })) { workflowIds.push(execution.workflowId); if (workflowIds.length >= OPEN_RUN_SAMPLE) { diff --git a/packages/simulator/src/cluster/watch.ts b/packages/simulator/src/cluster/watch.ts index 6de0efb8..34294250 100644 --- a/packages/simulator/src/cluster/watch.ts +++ b/packages/simulator/src/cluster/watch.ts @@ -29,7 +29,10 @@ import { prepareRun } from "./scaffold.js"; const OBSERVATION_INTERVAL_MS = 1_000; const FAILED_JOB_DETAIL = "controller Job failed"; -const DIAGNOSTIC_LIMIT = 4_096; +/** Characters of sanitized controller output one failure retains. */ +const RETAINED_DIAGNOSTIC_CHARACTERS = 4_096; +/** Bytes of controller log fetched, ahead of redaction and that bound. */ +const FETCHED_LOG_BYTES = RETAINED_DIAGNOSTIC_CHARACTERS * 2; const CONTROLLER_LOG_TAIL_LINES = 200; const SENSITIVE_LOG_LINE = /(authorization|bearer|token|secret|password|api[-_ ]?key|agent[-_ ]?key)/iu; @@ -71,7 +74,7 @@ export function sanitizeControllerDiagnostic(value: string): string { ) .join("\n") .trim(); - return normalized.slice(-DIAGNOSTIC_LIMIT); + return normalized.slice(-RETAINED_DIAGNOSTIC_CHARACTERS); } function conditionDetail(job: JobObservation): string | undefined { @@ -152,9 +155,8 @@ function failureDetail(job: JobObservation, logs: string): string { ); } -// A run that failed with a decodable summary is a completed observation, not a -// failed one: the activity returns it rather than failing, so the reason the -// Job gave has to ride the result or it is gone with the namespace. +// The activity returns a failed run rather than failing, so the reason the Job +// gave has to ride the result or it is gone with the namespace. function failedControllerObservation( job: JobObservation, logs: string, @@ -195,11 +197,7 @@ function terminalControllerLogs( namespace: string, ): Effect.Effect { return api - .readControllerLogs( - namespace, - CONTROLLER_LOG_TAIL_LINES, - DIAGNOSTIC_LIMIT * 2, - ) + .readControllerLogs(namespace, CONTROLLER_LOG_TAIL_LINES, FETCHED_LOG_BYTES) .pipe( Effect.catchAll((failure) => Effect.logWarning(