diff --git a/packages/temporal/README.md b/packages/temporal/README.md index 1a2a31ce9e82..475bc10e58f2 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -59,7 +59,8 @@ session runs as the workflow `session-exec-`. | `OPENCODE_TEMPORAL_ROLE` | `both` | `both` hosts the worker and the client in one process. `client` drives workflows without a worker. `worker` runs activities with no HTTP surface. | | `OPENCODE_SESSION_IDLE_TIMEOUT` | | How long an idle session's workflow stays open. Local mode honors the same variable. | | `OPENCODE_TEMPORAL_STEPPED` | off | `1` runs each step as a model call, one activity per tool call, and a seal. | -| `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. | +| `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_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 without affinity. | | `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`. | @@ -143,6 +144,24 @@ activity starts each tool the moment the model asks for it, while here the attem The tools of one step still run concurrently with each other. Against live providers that tail is under a tenth of a second, and no model tested emitted text after asking for its first tool. +### Worker affinity + +Off by default. Without it every worker polls one queue, and a worker drawing a session whose tree it +has never seen rebuilds that tree from snapshot packs. Affinity avoids the rebuild by routing: the +queue name is derived from the project worktree, and only workers serving that tree poll it. + +```bash +# a worker declares the tree it serves; defaults to the process directory +OPENCODE_TEMPORAL_WORKTREE_AFFINITY=1 OPENCODE_TEMPORAL_WORKTREE=/srv/trees/acme \ + OPENCODE_TEMPORAL_ROLE=worker ... bun run packages/server/src/worker.ts +``` + +This trades availability for latency. A session whose tree has no worker polling does not fall back +to another worker; it waits. Two consequences follow. A worker serves one tree, so in the default +`role=both` deployment a session in another project has no poller. And flipping the flag strands +workflows already running: a workflow keeps the task queue it started on for life, so drain before +flipping, in either direction. Both processes log the queue they use. + ## Running workers separately By default the serve process hosts both the worker and the client. To scale workers apart from the diff --git a/packages/temporal/src/config.ts b/packages/temporal/src/config.ts index 6953e4c81f7b..29e16c37da16 100644 --- a/packages/temporal/src/config.ts +++ b/packages/temporal/src/config.ts @@ -21,11 +21,19 @@ export interface Interface { /** Drive each step as a provider attempt, one activity per tool call, and a seal. Off by default: * the whole-step mode is what runs today, and this only changes how new sessions start. */ readonly stepped?: boolean + /** Route a session's work to workers that already hold its project tree, instead of letting any + * worker draw it and rebuild the tree from snapshot packs. Off by default, because it trades the + * reconstruction fallback for latency: a session whose worktree has no worker polling waits + * rather than being served elsewhere. */ + readonly worktreeAffinity?: boolean + /** The worktree this worker serves, when affinity is on. Defaults to the process directory, which + * is what a serve process with an embedded worker is already sitting in. */ + readonly worktree?: string /** Run a step's tool calls one at a time instead of together. Tools of one step write the same * tree and each ships from the host that ran it, so two on two hosts each publish a tree without * the other's work: the second is refused rather than reverting the first, which leaves its work * stranded there. `OPENCODE_TEMPORAL_SERIAL_TOOLS=1` forces it on; it is on by default only where - * the store is shared, which is what lets one step's tools land on different workers. */ + * the store is shared and no affinity keeps a session's tools on one worker. */ readonly serialTools?: boolean } @@ -35,6 +43,7 @@ const given = (name: string) => process.env[name] !== undefined && process.env[n const onOff = (name: string, fallback: boolean) => (given(name) ? process.env[name] === "1" : fallback) export const fromEnv = (): Interface => { + const worktreeAffinity = process.env.OPENCODE_TEMPORAL_WORKTREE_AFFINITY === "1" const sharedStore = !!process.env.OPENCODE_DB_URL return { address: process.env.TEMPORAL_ADDRESS ?? DEFAULTS.address, @@ -43,8 +52,10 @@ export const fromEnv = (): Interface => { role: (process.env.OPENCODE_TEMPORAL_ROLE as Role | undefined) ?? "both", idleTimeout: process.env.OPENCODE_SESSION_IDLE_TIMEOUT, stepped: onOff("OPENCODE_TEMPORAL_STEPPED", false), + worktreeAffinity, + worktree: process.env.OPENCODE_TEMPORAL_WORKTREE, // Serial by default only where two hosts can end up writing one step's tree: a shared store, // with nothing keeping the step's tools on one worker. - serialTools: onOff("OPENCODE_TEMPORAL_SERIAL_TOOLS", sharedStore), + serialTools: onOff("OPENCODE_TEMPORAL_SERIAL_TOOLS", !worktreeAffinity && sharedStore), } } diff --git a/packages/temporal/src/executor.ts b/packages/temporal/src/executor.ts index 416f5b5aaa25..f5863494438d 100644 --- a/packages/temporal/src/executor.ts +++ b/packages/temporal/src/executor.ts @@ -16,6 +16,10 @@ import { SessionExecution } from "@opencode-ai/core/session/execution" import { makeStepActivities, makeSteppedTurnActivities } from "./activities" import { makeDrains } from "./drain" import { makeSteppedDrains } from "./stepped-drain" +import { queueForWorktree } from "./queue" +import { Database } from "@opencode-ai/core/database/database" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { eq } from "drizzle-orm" import { WorktreeMaterializer } from "@opencode-ai/core/session/execution/worktree" import { toRunError } from "./run-error-codec" import * as WF from "./workflow" @@ -48,6 +52,7 @@ const layer = Layer.effect( Effect.gen(function* () { const store = yield* SessionStore.Service const locations = yield* LocationServiceMap.Service + const db = (yield* Database.Service).db // The app context the local drain runs in: providing it, then the per-location layer, supplies // SessionRunner and all of its dependencies. const ctx = yield* Effect.context() @@ -63,6 +68,31 @@ const layer = Layer.effect( // Only the client can read whether the store is shared, so whether a step's tools may overlap // is decided here and rides the workflow input. const SERIAL_TOOLS = config.serialTools === true + const AFFINITY = config.worktreeAffinity === true + // The tree this process serves when affinity is on. A serve process with an embedded worker is + // already sitting in it, so the process directory is the right default. + const SERVED_WORKTREE = config.worktree ?? process.cwd() + // Which queue a worker polls. With affinity off this is the one shared queue and any worker can + // draw any session, rebuilding the tree if it has to. + const POLL_QUEUE = AFFINITY ? queueForWorktree(TASK_QUEUE, SERVED_WORKTREE) : TASK_QUEUE + // Which queue a session's workflow runs on. Keyed on the PROJECT worktree, not the session's + // directory: `worktrees.ensure` rebuilds the project tree, so keying on the directory a session + // happened to start in would split one physical tree across a queue per subdirectory, and a + // session started from a subfolder would wait on a queue nobody polls. + const queueFor = (id: SessionSchema.ID) => + AFFINITY + ? Effect.gen(function* () { + const session = yield* store.get(id) + if (!session) return TASK_QUEUE + const project = yield* db + .select({ worktree: ProjectTable.worktree }) + .from(ProjectTable) + .where(eq(ProjectTable.id, session.projectID)) + .get() + .pipe(Effect.orDie) + return project ? queueForWorktree(TASK_QUEUE, project.worktree) : TASK_QUEUE + }) + : Effect.succeed(TASK_QUEUE) const events = yield* EventV2.Service const worktrees = yield* WorktreeMaterializer.Service @@ -92,7 +122,7 @@ const layer = Layer.effect( Worker.create({ connection: nativeConn, namespace: NAMESPACE, - taskQueue: TASK_QUEUE, + taskQueue: POLL_QUEUE, workflowsPath: fileURLToPath(new URL("./workflow.ts", import.meta.url)), activities: { ...makeStepActivities(drains.stepDrain), ...makeSteppedTurnActivities(stepped) }, }), @@ -111,7 +141,13 @@ const layer = Layer.effect( // unused. Return a service whose driving methods fail loudly if something unexpectedly calls them. if (!HOST_CLIENT) { yield* Effect.logInfo("SessionExecutionTemporal worker ready").pipe( - Effect.annotateLogs({ address: ADDRESS, taskQueue: TASK_QUEUE, workflow: WORKFLOW_TYPE, role: config.role }), + Effect.annotateLogs({ + address: ADDRESS, + taskQueue: POLL_QUEUE, + workflow: WORKFLOW_TYPE, + role: config.role, + ...(AFFINITY ? { worktree: SERVED_WORKTREE } : {}), + }), ) const clientOnly = Effect.die("SessionExecution client is not hosted when OPENCODE_TEMPORAL_ROLE=worker") return SessionExecution.Service.of({ @@ -130,26 +166,33 @@ const layer = Layer.effect( const client = new Client({ connection: clientConn, namespace: NAMESPACE }) const drive = (id: SessionSchema.ID) => - Effect.promise(() => - client.workflow.signalWithStart(WORKFLOW_TYPE, { - taskQueue: TASK_QUEUE, - workflowId: workflowId(id), - args: [ - id, - { - startWithWake: true, - idleTimeout: IDLE_TIMEOUT, - stepped: STEPPED, - serialTools: SERIAL_TOOLS, - } satisfies WF.SessionTurnOptions, - ], - signal: WF.wake, - signalArgs: [], - }), + Effect.flatMap(queueFor(id), (taskQueue) => + Effect.promise(() => + client.workflow.signalWithStart(WORKFLOW_TYPE, { + taskQueue, + workflowId: workflowId(id), + args: [ + id, + { + startWithWake: true, + idleTimeout: IDLE_TIMEOUT, + stepped: STEPPED, + serialTools: SERIAL_TOOLS, + } satisfies WF.SessionTurnOptions, + ], + signal: WF.wake, + signalArgs: [], + }), + ), ) yield* Effect.logInfo("SessionExecutionTemporal ready").pipe( - Effect.annotateLogs({ address: ADDRESS, taskQueue: TASK_QUEUE, workflow: WORKFLOW_TYPE }), + Effect.annotateLogs({ + address: ADDRESS, + taskQueue: TASK_QUEUE, + workflow: WORKFLOW_TYPE, + ...(AFFINITY ? { worktreeAffinity: true, pollQueue: POLL_QUEUE, worktree: SERVED_WORKTREE } : {}), + }), ) return SessionExecution.Service.of({ @@ -184,44 +227,46 @@ const layer = Layer.effect( // resume = coordinator.run: drive a forced run via an Update-with-Start and AWAIT its result, // so a run error is surfaced to the caller (as a RunError) instead of being swallowed. resume: (id) => - Effect.tryPromise({ - try: async () => { - const attempt = () => { - const startOp = new WithStartWorkflowOperation(WORKFLOW_TYPE, { - taskQueue: TASK_QUEUE, - workflowId: workflowId(id), - // startWithWake=false: a fresh resume-with-start must not manufacture a wake drain; - // its forced drain comes from the resume update. Ignored when USE_EXISTING joins a - // running workflow (which keeps its own state). - args: [ - id, - { - startWithWake: false, - idleTimeout: IDLE_TIMEOUT, - stepped: STEPPED, - serialTools: SERIAL_TOOLS, - } satisfies WF.SessionTurnOptions, - ], - workflowIdConflictPolicy: "USE_EXISTING", - }) - return client.workflow.executeUpdateWithStart(WF.resume, { - startWorkflowOperation: startOp, - args: [], - }) - } - try { - await attempt() - } catch (e) { - // The long-lived workflow self-completes after its idle timeout; an update admitted - // in that instant fails against the just-completed run instead of starting a fresh - // one. Retry once so the caller gets a real run, not the race. - const message = String((e as { message?: unknown })?.message ?? "") - if (!/already completed|not found/i.test(message)) throw e - await attempt() - } - }, - catch: (e) => toRunError(id, e), - }), + Effect.flatMap(queueFor(id), (resumeQueue) => + Effect.tryPromise({ + try: async () => { + const attempt = () => { + const startOp = new WithStartWorkflowOperation(WORKFLOW_TYPE, { + taskQueue: resumeQueue, + workflowId: workflowId(id), + // startWithWake=false: a fresh resume-with-start must not manufacture a wake drain; + // its forced drain comes from the resume update. Ignored when USE_EXISTING joins a + // running workflow (which keeps its own state). + args: [ + id, + { + startWithWake: false, + idleTimeout: IDLE_TIMEOUT, + stepped: STEPPED, + serialTools: SERIAL_TOOLS, + } satisfies WF.SessionTurnOptions, + ], + workflowIdConflictPolicy: "USE_EXISTING", + }) + return client.workflow.executeUpdateWithStart(WF.resume, { + startWorkflowOperation: startOp, + args: [], + }) + } + try { + await attempt() + } catch (e) { + // The long-lived workflow self-completes after its idle timeout; an update admitted + // in that instant fails against the just-completed run instead of starting a fresh + // one. Retry once so the caller gets a real run, not the race. + const message = String((e as { message?: unknown })?.message ?? "") + if (!/already completed|not found/i.test(message)) throw e + await attempt() + } + }, + catch: (e) => toRunError(id, e), + }), + ), interrupt: (id) => Effect.tryPromise({ try: () => client.workflow.getHandle(workflowId(id)).signal(WF.interrupt), @@ -246,5 +291,5 @@ const layer = Layer.effect( export const node = makeGlobalNode({ service: SessionExecution.Service, layer, - deps: [SessionStore.node, LocationServiceMap.node, EventV2.node, WorktreeMaterializer.node], + deps: [SessionStore.node, LocationServiceMap.node, EventV2.node, WorktreeMaterializer.node, Database.node], }) diff --git a/packages/temporal/src/queue.ts b/packages/temporal/src/queue.ts new file mode 100644 index 000000000000..f13c052f98e8 --- /dev/null +++ b/packages/temporal/src/queue.ts @@ -0,0 +1,42 @@ +// Routing a session's work to a worker that already has its project tree. +// +// Without affinity every worker polls one queue, and a worker that draws a session whose worktree it +// has never seen rebuilds it from snapshot packs (`session/execution/worktree.ts`). That is the +// portable baseline and it works, but it costs a materialization on the first step a worker takes +// for that session. Affinity avoids the cost by routing instead: the queue name is derived from the +// session's directory, and only workers serving that directory poll it. +// +// The trade is availability for latency, and it is the whole reason this is opt-in. With affinity on, +// a session whose worktree has no worker polling for it does not fall back to another worker; it +// waits. Reconstruction is what makes any worker able to serve any session, and turning affinity on +// is choosing not to use it. +// +// Deliberately NOT imported by workflow code: this reaches for node:crypto, which the Temporal +// sandbox does not have. The queue is chosen by the client that starts the workflow and by the worker +// that polls, both of which are ordinary Node. +import { createHash } from "node:crypto" +import { resolve } from "node:path" + +/** Longer than needed to make a collision implausible, short enough to keep the queue name readable + * in the Temporal UI and in `temporal task-queue describe`. */ +const DIGEST_LENGTH = 12 + +/** + * The queue for one worktree. + * + * Derived from the path alone, never from the filesystem. The two sides that have to agree do not + * see the same disk: the client hashes the session's directory, the worker hashes the tree it was + * told to serve, and a worker may not have materialized that tree yet. Resolving through the + * filesystem would make the key depend on whether the path happens to exist locally, so one side + * would canonicalize and the other would not, and they would sit on different queues while nothing + * reported an error. + * + * The cost is that the operator has to spell the tree the same way on both sides. `/tmp/x` and + * `/private/tmp/x` are one tree on macOS and two keys here. Both processes log the queue they use, + * so a mismatch shows up as two names rather than as a session that never runs. + */ +export const queueForWorktree = (base: string, directory: string): string => { + const canonical = resolve(directory).replace(/[/\\]+$/, "") + const digest = createHash("sha256").update(canonical).digest("hex").slice(0, DIGEST_LENGTH) + return `${base}-wt-${digest}` +} diff --git a/packages/temporal/test/queue.test.ts b/packages/temporal/test/queue.test.ts new file mode 100644 index 000000000000..21ff8784eedb --- /dev/null +++ b/packages/temporal/test/queue.test.ts @@ -0,0 +1,61 @@ +// Worktree affinity is only useful if the client starting a workflow and the worker polling for it +// derive the SAME queue name from the same tree. If they disagree they sit on two queues and the +// session waits forever, which is silent: nothing errors, the work simply never runs. +import { describe, it, expect } from "bun:test" +import { mkdtempSync, mkdirSync, realpathSync, rmSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { queueForWorktree } from "../src/queue" + +describe("worktree queue", () => { + it("gives one tree one name and different trees different names", () => { + const root = realpathSync(mkdtempSync(join(tmpdir(), "queue-"))) + try { + const a = join(root, "a") + const b = join(root, "b") + mkdirSync(a) + mkdirSync(b) + + expect(queueForWorktree("q", a)).toBe(queueForWorktree("q", a)) + expect(queueForWorktree("q", a)).not.toBe(queueForWorktree("q", b)) + // The base is kept as a readable prefix so a queue is identifiable in the Temporal UI. + expect(queueForWorktree("q", a).startsWith("q-wt-")).toBe(true) + } finally { + rmSync(root, { recursive: true, force: true }) + } + }) + + it("agrees whether or not the tree exists locally", () => { + const root = realpathSync(mkdtempSync(join(tmpdir(), "queue-"))) + try { + const present = join(root, "present") + mkdirSync(present) + const absent = join(root, "absent") + + // This is the case that matters. The client hashes the session's directory and the worker + // hashes the tree it was told to serve, and those two processes do not see the same disk. If + // the key depended on the path resolving locally, one side would canonicalize and the other + // would not, and they would sit on different queues with nothing reporting an error. + expect(queueForWorktree("q", present)).toBe(queueForWorktree("q", join(root, "present"))) + expect(queueForWorktree("q", absent)).toBe(queueForWorktree("q", join(root, "absent"))) + } finally { + rmSync(root, { recursive: true, force: true }) + } + }) + + it("normalizes a path without touching the filesystem", () => { + expect(queueForWorktree("q", "/srv/trees/acme/")).toBe(queueForWorktree("q", "/srv/trees/acme")) + expect(queueForWorktree("q", "/srv/trees/./acme")).toBe(queueForWorktree("q", "/srv/trees/acme")) + // Two spellings of one tree are two keys. That is the trade for a key both sides can compute + // without a filesystem, and it is why the derived queue is logged on both sides. + expect(queueForWorktree("q", "/tmp/x")).not.toBe(queueForWorktree("q", "/private/tmp/x")) + }) + + it("still names a tree that does not exist yet", () => { + // A worker can declare the tree it serves before anything has materialized it, so an + // unresolvable path has to stay a stable key rather than throw. + const missing = join(tmpdir(), "queue-missing-tree-that-is-not-there") + expect(queueForWorktree("q", missing)).toBe(queueForWorktree("q", missing)) + expect(queueForWorktree("q", missing).startsWith("q-wt-")).toBe(true) + }) +})