diff --git a/bun.lock b/bun.lock index fd7bd2a28917..ab2ce122b2aa 100644 --- a/bun.lock +++ b/bun.lock @@ -673,6 +673,7 @@ "@opencode-ai/core": "workspace:*", "@opencode-ai/http-recorder": "workspace:*", "@opencode-ai/script": "workspace:*", + "@opencode-ai/temporal": "workspace:*", "@standard-schema/spec": "1.0.0", "@tsconfig/bun": "catalog:", "@types/babel__core": "7.20.5", diff --git a/packages/opencode/package.json b/packages/opencode/package.json index 5e628b3c7055..da2fff200461 100644 --- a/packages/opencode/package.json +++ b/packages/opencode/package.json @@ -32,6 +32,7 @@ "@babel/core": "7.28.4", "@octokit/webhooks-types": "7.6.1", "@opencode-ai/core": "workspace:*", + "@opencode-ai/temporal": "workspace:*", "@opencode-ai/http-recorder": "workspace:*", "@opencode-ai/script": "workspace:*", "@standard-schema/spec": "1.0.0", diff --git a/packages/opencode/src/cli/cmd/detached.ts b/packages/opencode/src/cli/cmd/detached.ts new file mode 100644 index 000000000000..8bc7b10633f4 --- /dev/null +++ b/packages/opencode/src/cli/cmd/detached.ts @@ -0,0 +1,31 @@ +// Commands about the deployment rather than about a session in front of you. `doctor` loads the +// Temporal driver only when it runs, so the rest of the `session` command never pays for it. + +import type { Argv } from "yargs" +import { cmd } from "./cmd" +import { UI } from "../ui" + +// What this process resolved, and what is wrong with it. Deploying was a handful of variables that +// have to agree, with no way to ask whether they did: every mistake in them fails as something +// else, hours later, on whoever prompted the session rather than on whoever deployed it. +export const SessionDoctorCommand = cmd({ + command: "doctor", + describe: "what this deployment resolved, and what is wrong with it", + builder: (yargs: Argv) => yargs, + handler: async () => { + const { TemporalConfig } = await import("@opencode-ai/temporal/config") + const config = TemporalConfig.fromEnv() + UI.println("opencode, temporal execution") + for (const [name, value] of Object.entries(TemporalConfig.describe(config))) { + UI.println(` ${name}: ${value}`) + } + for (const note of TemporalConfig.notes(config)) UI.println(`note: ${note}`) + const problems = TemporalConfig.preflight(config) + for (const problem of problems) UI.println(`problem: ${problem}`) + if (problems.length > 0) { + process.exitCode = 1 + return + } + UI.println("this deployment looks consistent") + }, +}) diff --git a/packages/opencode/src/cli/cmd/session.ts b/packages/opencode/src/cli/cmd/session.ts index 9e6ddda9d2d8..13befb6e9eed 100644 --- a/packages/opencode/src/cli/cmd/session.ts +++ b/packages/opencode/src/cli/cmd/session.ts @@ -1,6 +1,7 @@ import type { Argv } from "yargs" import { Effect } from "effect" import { cmd } from "./cmd" +import { SessionDoctorCommand } from "./detached" import { effectCmd, fail } from "../effect-cmd" import { Session } from "@/session/session" import { SessionID } from "../../session/schema" @@ -44,7 +45,8 @@ function pagerCmd(): string[] { export const SessionCommand = cmd({ command: "session", describe: "manage sessions", - builder: (yargs: Argv) => yargs.command(SessionListCommand).command(SessionDeleteCommand).demandCommand(), + builder: (yargs: Argv) => + yargs.command(SessionListCommand).command(SessionDeleteCommand).command(SessionDoctorCommand).demandCommand(), async handler() {}, }) diff --git a/packages/temporal/README.md b/packages/temporal/README.md index 80f9aa3d06e0..4c699e0199db 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -62,6 +62,9 @@ session runs as the workflow `session-exec-`. | `OPENCODE_TEMPORAL_WORKTREE_AFFINITY`, `OPENCODE_TEMPORAL_WORKTREE` | off | `1` routes a session to the workers serving its project tree, named by the second variable. | | `OPENCODE_TEMPORAL_STEP_AFFINITY` | on | Send a step's tool calls and seal back to the worker that made its model call. `0` turns it off. | | `OPENCODE_TEMPORAL_SERIAL_TOOLS` | derived | Run a step's tool calls one at a time. On by default only where two hosts could write one step's tree: a shared store with neither kind of affinity. | +| `OPENCODE_TEMPORAL_PROFILE` | `local` | `fleet` sets the defaults a multi-host deployment needs and refuses combinations that cannot work. See below. | +| `OPENCODE_TEMPORAL_API_KEY`, `OPENCODE_TEMPORAL_API_KEY_FILE` | | Temporal Cloud credentials, the second read from a file. | +| `OPENCODE_TEMPORAL_TLS_CERT`, `OPENCODE_TEMPORAL_TLS_KEY`, `OPENCODE_TEMPORAL_TLS_CA` | | A certificate pair for a cluster with mTLS. `OPENCODE_TEMPORAL_TLS=1` for TLS without a client certificate. | | `OPENCODE_EVENT_POLL_MS` | | How often a live subscriber re-reads the log for events another process appended. `0` turns the tick off. | | `OPENCODE_DB` | | One absolute path shared by every process on a host. | | `OPENCODE_DB_URL`, `OPENCODE_DB_AUTH_TOKEN` | | A libSQL URL for a store shared across hosts. Takes precedence over `OPENCODE_DB`. | @@ -260,3 +263,29 @@ One rule bounds what the rebuild may touch, because checking a stored tree out o destroys work: a tree is moved only when a host-local note (`snapshot/tip.ts`) says this host is behind the store, so a host holding a capture that never shipped is left alone. Packs are ordered by their chain, not by `time_created`, because hosts do not agree on the time. + +## Picking a deployment rather than assembling one + +The settings are not independent, and getting them wrong fails as something else later: a store +only one process can see reads as a worker that never picks anything up. `OPENCODE_TEMPORAL_PROFILE` +picks one deployment and the rest follow. + +| | `local` (default) | `fleet` | +|---|---|---| +| what it is | one serve, worker inside it | serve processes and workers, separate | +| store | this process only | you set `OPENCODE_DB_URL` | +| role | `both` | `client` for serve, `worker` for workers | +| unit of work | a whole step | the model call, each tool call, the seal | + +Individual settings override the profile's defaults. `opencode session doctor` prints what this +process resolved and refuses what cannot work: a fleet without a shared store, a serve that also +polls, an API key pointed at a dev server, half a certificate pair. It checks one process; it does +not compare processes or probe another host's store. + +```bash +TEMPORAL_ADDRESS=your-ns.a1b2c.tmprl.cloud:7233 TEMPORAL_NAMESPACE=your-ns.a1b2c \ + OPENCODE_TEMPORAL_API_KEY_FILE=/run/secrets/temporal-key # Temporal Cloud +TEMPORAL_ADDRESS=temporal.internal:7233 \ + OPENCODE_TEMPORAL_TLS_CERT=/run/secrets/tls.crt \ + OPENCODE_TEMPORAL_TLS_KEY=/run/secrets/tls.key # a cluster with mTLS +``` diff --git a/packages/temporal/src/config.ts b/packages/temporal/src/config.ts index 6f5a28b01798..04e72dac9143 100644 --- a/packages/temporal/src/config.ts +++ b/packages/temporal/src/config.ts @@ -3,6 +3,7 @@ export * as TemporalConfig from "./config" // Connection and behavior settings for the Temporal executor. The executor reads them at layer // build: an embedder or a test provides the service to override, and absent that the values come // from env. Nothing reads env at module load, so import order carries no configuration. +import { readFileSync } from "node:fs" import { Context } from "effect" import { DEFAULTS } from "./protocol" @@ -11,11 +12,22 @@ import { DEFAULTS } from "./protocol" // worker's bundler); `worker` runs a standalone activity worker with no HTTP surface. export type Role = "both" | "client" | "worker" +// Which deployment this is. The settings below are not independent: a fleet whose store is not +// shared is a set of workers that cannot see each other's sessions, and finding that out takes a +// session that answers with the wrong files. `fleet` sets what has to agree, and `preflight` +// refuses what cannot. +export type Profile = "local" | "fleet" + export interface Interface { + readonly profile: Profile readonly address: string readonly namespace: string readonly taskQueue: string readonly role: Role + /** How a server that is not the dev server is reached: an API key for Cloud, a certificate pair + * for a cluster with mTLS. Read from files, never from argv, and never logged. */ + readonly apiKey?: string + readonly tls?: { readonly cert: string; readonly key: string; readonly ca?: string } | true /** Override for the supervisor's idle self-termination; local mode honors the same variable. */ readonly idleTimeout?: string /** Drive each step as a provider attempt, one activity per tool call, and a seal. Off by default: @@ -41,10 +53,14 @@ export interface Interface { * default: a pin nobody answers falls back to the shared queue after a schedule-to-start bound, * so the worst it costs is that wait. `OPENCODE_TEMPORAL_STEP_AFFINITY=0` turns it off. */ readonly stepAffinity?: boolean + /** Whether the store is shared across hosts (`OPENCODE_DB_URL`). A fleet without it is a set of + * workers that cannot see each other's sessions. */ + readonly sharedStore: boolean } export class Service extends Context.Service()("@opencode/temporal/Config") {} +const read = (path: string | undefined) => (path ? readFileSync(path, "utf8") : undefined) const given = (name: string) => process.env[name] !== undefined && process.env[name] !== "" const onOff = (name: string, fallback: boolean) => (given(name) ? process.env[name] === "1" : fallback) @@ -53,17 +69,111 @@ export const fromEnv = (): Interface => { const worktreeAffinity = process.env.OPENCODE_TEMPORAL_WORKTREE_AFFINITY === "1" const sharedStore = !!process.env.OPENCODE_DB_URL return { + profile: process.env.OPENCODE_TEMPORAL_PROFILE === "fleet" ? "fleet" : "local", address: process.env.TEMPORAL_ADDRESS ?? DEFAULTS.address, namespace: process.env.TEMPORAL_NAMESPACE ?? DEFAULTS.namespace, taskQueue: process.env.OPENCODE_TEMPORAL_TASK_QUEUE ?? DEFAULTS.taskQueue, role: (process.env.OPENCODE_TEMPORAL_ROLE as Role | undefined) ?? "both", + apiKey: process.env.OPENCODE_TEMPORAL_API_KEY ?? read(process.env.OPENCODE_TEMPORAL_API_KEY_FILE), + tls: + process.env.OPENCODE_TEMPORAL_TLS_CERT && process.env.OPENCODE_TEMPORAL_TLS_KEY + ? { + cert: readFileSync(process.env.OPENCODE_TEMPORAL_TLS_CERT, "utf8"), + key: readFileSync(process.env.OPENCODE_TEMPORAL_TLS_KEY, "utf8"), + ca: read(process.env.OPENCODE_TEMPORAL_TLS_CA), + } + : process.env.OPENCODE_TEMPORAL_TLS === "1" + ? true + : undefined, idleTimeout: process.env.OPENCODE_SESSION_IDLE_TIMEOUT, - stepped: onOff("OPENCODE_TEMPORAL_STEPPED", false), + // A fleet's unit of work is the smaller one: a worker dying takes one tool call with it rather + // than a whole step, and a tool call is where the retry policy and the approval belong. + stepped: onOff("OPENCODE_TEMPORAL_STEPPED", process.env.OPENCODE_TEMPORAL_PROFILE === "fleet"), worktreeAffinity, worktree: process.env.OPENCODE_TEMPORAL_WORKTREE, stepAffinity, // Serial by default only where two hosts can end up writing one step's tree: a shared store with // neither kind of affinity keeping the step's tools on one worker. serialTools: onOff("OPENCODE_TEMPORAL_SERIAL_TOOLS", !stepAffinity && !worktreeAffinity && sharedStore), + sharedStore, + } +} + +/** What `Connection.connect` and `NativeConnection.connect` both take, built once so a client and a + * worker in different processes cannot disagree about how the cluster is reached. */ +export const connectionOptions = (config: Interface) => { + const tls = + config.tls === true || (config.apiKey && config.tls === undefined) + ? true + : config.tls + ? { + clientCertPair: { crt: Buffer.from(config.tls.cert), key: Buffer.from(config.tls.key) }, + ...(config.tls.ca ? { serverRootCACertificate: Buffer.from(config.tls.ca) } : {}), + } + : undefined + return { + address: config.address, + ...(tls ? { tls } : {}), + ...(config.apiKey ? { apiKey: config.apiKey } : {}), + } +} + +const LOOPBACK = /^(127\.0\.0\.1|localhost|\[::1\]|0\.0\.0\.0)(:|$)/ + +/** + * What is wrong with this deployment, said before it takes work rather than after. Each of these + * fails as something else: a store only one process can see reads as a worker that never picks + * anything up, and a client with no worker anywhere reads as a session that accepts a prompt and + * never answers it. + */ +export const preflight = (config: Interface): string[] => { + const problems: string[] = [] + if (config.profile === "fleet") { + if (!config.sharedStore) + problems.push( + "the fleet profile needs OPENCODE_DB_URL: the store is the record, and workers that do " + + "not share it cannot serve each other's sessions", + ) + if (config.role === "both") + problems.push( + "OPENCODE_TEMPORAL_ROLE is `both` in a fleet: a serve that also polls is a laptop " + + "deployment. Run `client` next to standalone `worker` processes", + ) } + if (config.apiKey && LOOPBACK.test(config.address)) + problems.push(`an API key is set but TEMPORAL_ADDRESS is ${config.address}, which is a dev server`) + if (config.apiKey && config.namespace === "default") + problems.push("an API key is set but TEMPORAL_NAMESPACE is `default`, which is not a Cloud namespace") + if (!!process.env.OPENCODE_TEMPORAL_TLS_CERT !== !!process.env.OPENCODE_TEMPORAL_TLS_KEY) + problems.push("OPENCODE_TEMPORAL_TLS_CERT and OPENCODE_TEMPORAL_TLS_KEY come as a pair") + return problems } + +/** + * Worth saying, not worth refusing. Only what cannot work belongs in `preflight`, because a process + * that exits takes a deployment with it, and plaintext to an address that is not loopback is a + * private network in most deployments and a mistake in some. Nothing here can tell which. + */ +export const notes = (config: Interface): string[] => { + const said: string[] = [] + if (!LOOPBACK.test(config.address) && !config.apiKey && !config.tls) + said.push( + `reaching ${config.address} in plaintext. For Temporal Cloud set OPENCODE_TEMPORAL_API_KEY; ` + + "for a cluster with mTLS set OPENCODE_TEMPORAL_TLS_CERT and OPENCODE_TEMPORAL_TLS_KEY", + ) + return said +} + +/** Every setting that decides how this process behaves, and nothing that is a credential. */ +export const describe = (config: Interface): Record => ({ + profile: config.profile, + address: config.address, + namespace: config.namespace, + taskQueue: config.taskQueue, + role: config.role, + store: config.sharedStore ? "shared (OPENCODE_DB_URL)" : "this process only", + stepped: String(config.stepped === true), + stepAffinity: String(config.stepAffinity !== false), + serialTools: String(config.serialTools === true), + credentials: config.apiKey ? "api key" : config.tls ? "certificate pair" : "none (plaintext)", +}) diff --git a/packages/temporal/src/executor.ts b/packages/temporal/src/executor.ts index b7050a51c3e9..052163b71d1e 100644 --- a/packages/temporal/src/executor.ts +++ b/packages/temporal/src/executor.ts @@ -99,6 +99,13 @@ const layer = Layer.effect( return project ? queueForWorktree(TASK_QUEUE, project.worktree) : TASK_QUEUE }) : Effect.succeed(TASK_QUEUE) + // Before anything is accepted, not after: every one of these fails as something else later, and + // the failure lands on whoever prompted the session rather than on whoever deployed it. + for (const note of TemporalConfig.notes(config)) yield* Effect.logInfo(`configuration: ${note}`) + const problems = TemporalConfig.preflight(config) + for (const problem of problems) yield* Effect.logError(`configuration: ${problem}`) + if (problems.length > 0) yield* Effect.die(`this deployment cannot serve sessions: ${problems[0]}`) + const events = yield* EventV2.Service const worktrees = yield* WorktreeMaterializer.Service @@ -121,7 +128,7 @@ const layer = Layer.effect( ), ) const nativeConn = yield* Effect.acquireRelease( - Effect.promise(() => NativeConnection.connect({ address: ADDRESS })), + Effect.promise(() => NativeConnection.connect(TemporalConfig.connectionOptions(config))), (conn) => Effect.promise(() => conn.close().catch(() => {})), ) const worker = yield* Effect.promise(() => @@ -189,7 +196,7 @@ const layer = Layer.effect( // Client connection drives the per-session workflows. const clientConn = yield* Effect.acquireRelease( - Effect.promise(() => Connection.connect({ address: ADDRESS })), + Effect.promise(() => Connection.connect(TemporalConfig.connectionOptions(config))), (conn) => Effect.promise(() => conn.close().catch(() => {})), ) const client = new Client({ connection: clientConn, namespace: NAMESPACE })