From 2420e09ccb015915de3671852f593688ccc4be6a Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Mon, 14 Sep 2026 17:30:33 -0700 Subject: [PATCH] Carried the turn past a worker that lost a step it was running. A pinned attempt that started and then failed leaves the tool's outcome unknown, and nothing can say whether the process behind it stopped. The step is closed on the shared queue without touching the tree, so the turn goes on, and the pack store refuses files from an attempt the session has moved past, under the same token the transcript is fenced by. --- packages/core/src/session/runner/index.ts | 4 + packages/core/src/session/runner/llm.ts | 14 ++- packages/core/src/snapshot-sync.ts | 45 +++++++- .../core/test/worktree-materialize.test.ts | 56 ++++++++++ packages/temporal/src/drain.ts | 6 +- packages/temporal/src/stepped-drain.ts | 7 +- packages/temporal/src/stepped-turn.ts | 31 +++++- packages/temporal/test/pinned-retry.test.ts | 105 ++++++++++++++++++ packages/temporal/test/replay.test.ts | 70 ++++++++++++ .../test/stepped-drain-writers.test.ts | 95 ++++++++++++++++ packages/temporal/test/stepped-turn.test.ts | 61 ++++++++++ 11 files changed, 482 insertions(+), 12 deletions(-) create mode 100644 packages/temporal/test/pinned-retry.test.ts create mode 100644 packages/temporal/test/replay.test.ts create mode 100644 packages/temporal/test/stepped-drain-writers.test.ts diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index 9be39946695a..f7347d60fe39 100644 --- a/packages/core/src/session/runner/index.ts +++ b/packages/core/src/session/runner/index.ts @@ -59,6 +59,10 @@ export interface SealStepInput { * provider error, so a seal that re-derives this from the log would keep calling a provider that * just failed. Absent on a re-drive, where the log is all there is. */ readonly needsContinuation?: boolean + /** This step is being closed away from the host that ran it, so the files are not this seal's to + * touch: it is standing in a directory that never saw the tools, and the host that did may still + * be inside one of them. Writing the step down is the whole job here. */ + readonly withoutTheTree?: boolean } /** One recorded tool call, to run on its own. */ diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 03501c828882..199c46812fe8 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -280,7 +280,7 @@ const layer = Layer.effect( return yield* Effect.die(continueAfterCompaction(currentStep)) const startSnapshot = yield* snapshots.capture() // Ship the pre-step tree so another host can rebuild the worktree; best-effort inside push. - if (startSnapshot) yield* snapshotSync.push(startSnapshot) + if (startSnapshot) yield* snapshotSync.push(startSnapshot, session.id) const publisher = createLLMEventPublisher(events, { sessionID: session.id, agent: agent.id, @@ -393,7 +393,7 @@ const layer = Layer.effect( if (stepSettlement && !publisher.hasProviderError() && !deferTools) { const endSnapshot = yield* snapshots.capture() // Ship the post-step tree: this is the state a resumed step on another host needs. - if (endSnapshot) yield* snapshotSync.push(endSnapshot) + if (endSnapshot) yield* snapshotSync.push(endSnapshot, session.id) const files = startSnapshot && endSnapshot ? yield* snapshots @@ -572,7 +572,7 @@ const layer = Layer.effect( yield* failInterruptedTools(input.sessionID) const startSnapshot = inFlight.snapshot?.start const endSnapshot = yield* snapshots.capture() - if (endSnapshot) yield* snapshotSync.push(endSnapshot) + if (endSnapshot) yield* snapshotSync.push(endSnapshot, input.sessionID) const files = startSnapshot && endSnapshot ? yield* snapshots @@ -654,9 +654,11 @@ const layer = Layer.effect( // sends a request carrying a tool_use with no tool_result and the provider rejects it. yield* failInterruptedTools(input.sessionID, context) const startSnapshot = target.snapshot?.start - const endSnapshot = yield* snapshots.capture() + // A seal closing a step away from its host captures a directory that never ran the tools, so + // what it would ship is the state before them. The host that has them is the one that ships. + const endSnapshot = input.withoutTheTree ? undefined : yield* snapshots.capture() // Ship the post-step tree: this is the state a later step on another host needs. - if (endSnapshot) yield* snapshotSync.push(endSnapshot) + if (endSnapshot) yield* snapshotSync.push(endSnapshot, input.sessionID) const files = startSnapshot && endSnapshot ? yield* snapshots @@ -823,7 +825,7 @@ const layer = Layer.effect( yield* shipping.withLock(location.directory)( Effect.gen(function* () { const afterTool = yield* snapshots.capture().pipe(Effect.catch(() => Effect.succeed(undefined))) - if (afterTool) yield* snapshotSync.push(afterTool) + if (afterTool) yield* snapshotSync.push(afterTool, input.sessionID) }), ) return { outcome: "settled" } as ToolCallResult diff --git a/packages/core/src/snapshot-sync.ts b/packages/core/src/snapshot-sync.ts index be29569430c7..e2f08bbb8e72 100644 --- a/packages/core/src/snapshot-sync.ts +++ b/packages/core/src/snapshot-sync.ts @@ -13,6 +13,8 @@ import { ChildProcess } from "effect/unstable/process" import { eq } from "drizzle-orm" import { Database } from "./database/database" import { makeLocationNode } from "./effect/app-node" +import { EventV2 } from "./event" +import { EventSequenceTable } from "./event/sql" import { FSUtil } from "./fs-util" import { Git } from "./git" import { Global } from "./global" @@ -26,8 +28,23 @@ import { readWorktreeTip, writeWorktreeTip } from "./snapshot/tip" import { Hash } from "./util/hash" export interface Interface { - /** A stale tip fails the caller; packing and insertion errors are logged. */ - readonly push: (tree: Snapshot.ID) => Effect.Effect + /** + * A stale tip fails the caller; packing and insertion errors are logged. + * + * `sessionID` is what the files are being shipped for. Given it, a publisher a newer attempt has + * superseded is refused, which is the one case the tip check cannot answer: a tool whose dispatch + * was abandoned is still standing on the tree it read, so its pack is clean and reverts whatever + * ran in its place. Omitted by callers with no session behind them, and then nothing is fenced. + * + * Two clauses a lost host is held to, and a change here has to keep both: + * + * 1. The seal that closes a step away from its host publishes nothing. Between the dispatch + * failing and the next attempt claiming the log there is a window where nothing fences the + * old host, and the only thing that makes it harmless is that nobody else publishes then. + * 2. Once the session has moved past that host, what it publishes for the closed step is + * refused. + */ + readonly push: (tree: Snapshot.ID, sessionID?: string) => Effect.Effect } export class Service extends Context.Service()("@opencode/v2/SnapshotSync") {} @@ -73,7 +90,29 @@ const layer = Layer.effect( .all() .pipe(Effect.orDie, Effect.map(chainHead)) - const push = Effect.fn("SnapshotSync.push")(function* (tree: Snapshot.ID) { + // Whether a newer attempt holds this session's log. The token is the one the drain claimed and + // provided, so this asks the same question the log's own fence asks of every append: is what is + // writing still the attempt the session is on? + const superseded = Effect.fn("SnapshotSync.superseded")(function* (sessionID: string) { + const owner = yield* EventV2.EventOwner + // Outside a drain nothing claimed anything, so there is nothing to be superseded by. + if (owner === undefined) return false + const row = yield* db + .select({ ownerID: EventSequenceTable.owner_id }) + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, sessionID)) + .get() + .pipe(Effect.orDie) + return row?.ownerID != null && row.ownerID !== owner + }) + + const push = Effect.fn("SnapshotSync.push")(function* (tree: Snapshot.ID, sessionID?: string) { + // The files travel under the token the transcript does. A dispatch the session stopped + // waiting for keeps running, and what it publishes afterwards would otherwise be the newest + // state the store holds, because it is standing exactly where it was told to stand. + if (source && sessionID && (yield* superseded(sessionID))) { + return yield* Effect.die(new Error(`refusing to ship ${worktree}: a newer attempt holds ${sessionID}`)) + } // A host that has not caught up must not publish its older files as the next tree. // This reading is not an atomic head claim and does not fence concurrent publishers. if (source) { diff --git a/packages/core/test/worktree-materialize.test.ts b/packages/core/test/worktree-materialize.test.ts index 3e004606a86f..7befc9a2c946 100644 --- a/packages/core/test/worktree-materialize.test.ts +++ b/packages/core/test/worktree-materialize.test.ts @@ -12,6 +12,8 @@ import { Effect, Fiber, Layer } from "effect" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Database } from "@opencode-ai/core/database/database" +import { EventV2 } from "@opencode-ai/core/event" +import { EventSequenceTable } from "@opencode-ai/core/event/sql" import { Global } from "@opencode-ai/core/global" import { Location } from "@opencode-ai/core/location" import { AbsolutePath } from "@opencode-ai/core/schema" @@ -448,3 +450,57 @@ describe("WorktreeMaterializer", () => { ) }) +// A dispatch the session stopped waiting for keeps running, and the files it ships afterwards would +// be the newest state the store holds: the tip check cannot refuse them, because the host is still +// standing exactly where it was told to stand. The event log already fences a superseded attempt +// out of the transcript, and the packs travel under the same token. +describe("SnapshotSync owner fence", () => { + it.live("refuses a pack from an attempt the session has moved past", () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + const root = realpathSync(tmp.path) + const worktree = path.join(root, "project") + const file = path.join(root, "shared.db") + yield* Effect.promise(async () => { + await mkdir(worktree, { recursive: true }) + await $`git init -q ${worktree}`.quiet() + await $`git -C ${worktree} config user.email t@t`.quiet() + await $`git -C ${worktree} config user.name t`.quiet() + await writeFile(path.join(worktree, "tracked.txt"), "v1\n") + await $`git -C ${worktree} add .`.quiet() + await $`git -C ${worktree} commit -qm seed`.quiet() + }) + + const A = yield* Layer.build(captureStack(file, worktree, path.join(root, "host-a-data"))) + const onDatabase = (use: (db: Database.Interface["db"]) => Effect.Effect) => + Database.Service.use(({ db }) => use(db)).pipe( + Effect.orDie, + Effect.provide(Database.layerFromPath(file)), + Effect.scoped, + ) + // The session is on a later attempt than the one that is about to publish. + yield* onDatabase((db) => + db.insert(EventSequenceTable).values({ aggregate_id: "ses_fenced", seq: 1, owner_id: "run:1:2" }).run(), + ) + + const captured = yield* Snapshot.Service.use((s) => s.capture()).pipe(Effect.provide(A)) + if (!captured) throw new Error("expected a capture") + const shipped = (owner: string) => + Effect.exit( + SnapshotSync.Service.use((s) => s.push(captured, "ses_fenced")).pipe( + Effect.provideService(EventV2.EventOwner, owner), + Effect.provide(A), + ), + ) + + const stale = yield* shipped("run:1:1") + expect(stale._tag).toBe("Failure") + expect(yield* onDatabase((db) => db.select().from(SnapshotPackTable).all())).toHaveLength(0) + + // The attempt the session is actually on ships as usual. + const current = yield* shipped("run:1:2") + expect(current._tag).toBe("Success") + expect(yield* onDatabase((db) => db.select().from(SnapshotPackTable).all())).toHaveLength(1) + }), + ) +}) diff --git a/packages/temporal/src/drain.ts b/packages/temporal/src/drain.ts index ed96d6d806d2..60cb438e55b2 100644 --- a/packages/temporal/src/drain.ts +++ b/packages/temporal/src/drain.ts @@ -57,6 +57,10 @@ export interface InSessionOptions { /** Take the event log for this attempt. True for the writer that supersedes the attempt before it; * the other units of a stepped turn ride that writer's token. */ readonly claim?: boolean + /** Leave the project tree alone. For a seal closing a step away from its host: rebuilding here + * would put this host on the newest state while the one that ran the tools may still be writing, + * and nothing the seal does needs the files. */ + readonly withoutTheTree?: boolean } /** Run `use` against a session's runner. Undefined when the session is gone, which is not a run @@ -78,7 +82,7 @@ export const makeInSession = // Take the event log before running so a superseded attempt's later appends are fenced. if (options.claim && owner) yield* events.claim(session.id, owner) // A worker taking this step on a host without the project tree rebuilds it from snapshot packs. - yield* worktrees.ensure(session.location.directory) + if (!options.withoutTheTree) yield* worktrees.ensure(session.location.directory) return yield* SessionRunner.Service.use((runner) => use(runner, session)).pipe( Effect.provide(locations.get(session.location)), ) diff --git a/packages/temporal/src/stepped-drain.ts b/packages/temporal/src/stepped-drain.ts index ffe111687c9e..6faacc08e53d 100644 --- a/packages/temporal/src/stepped-drain.ts +++ b/packages/temporal/src/stepped-drain.ts @@ -54,6 +54,10 @@ export interface SealDrainInput { readonly assistantMessageID?: string readonly needsContinuation?: boolean readonly owner: string + /** This step is being closed away from the host that was running it. The tree is not this seal's + * to rebuild or to ship: the host that ran the tools is the only one holding what they did, and + * it may still be inside one of them. Writing the step down is the whole job. */ + readonly withoutTheTree?: boolean } export interface SteppedDrainDeps { @@ -120,7 +124,7 @@ export const makeSteppedDrains = ({ inSession, stepQueue }: SteppedDrainDeps) => runAtBoundary( input.sessionID, signal, - inSession(input.sessionID, input.owner, {}, (runner, session) => + inSession(input.sessionID, input.owner, { withoutTheTree: input.withoutTheTree }, (runner, session) => runner .sealStep({ sessionID: session.id, @@ -128,6 +132,7 @@ export const makeSteppedDrains = ({ inSession, stepQueue }: SteppedDrainDeps) => settlement: input.settlement, assistantMessageID: input.assistantMessageID, needsContinuation: input.needsContinuation, + withoutTheTree: input.withoutTheTree, }) .pipe(Effect.map((result) => toStepResult(input.step, result))), ).pipe(Effect.map((result) => result ?? toStepResult(input.step))), diff --git a/packages/temporal/src/stepped-turn.ts b/packages/temporal/src/stepped-turn.ts index c264294b97cb..1ff26740928b 100644 --- a/packages/temporal/src/stepped-turn.ts +++ b/packages/temporal/src/stepped-turn.ts @@ -220,5 +220,34 @@ export const makeSteppedTurn = if (unsettled.length > 0) log?.("step did not settle every call it dispatched", { step: model.step, calls: unsettled }) - return seal(false) + // A pinned attempt that started and failed does not take the turn with it, even though nothing + // can say whether the tool over there has stopped: the log fences a superseded attempt out of + // the transcript, and the pack store refuses its files under the same token, so what that host + // is still doing cannot reach the session. The step is closed on the shared queue, where a + // worker that is answering can take it, and the turn goes on with whatever the seal says + // follows. What does not move is the rest of this step: its calls stay where their host has them. + const closeElsewhere = () => { + log?.("the step lost its host; closing it elsewhere and carrying the turn on", { + sessionID: input.sessionID, + step: model.step, + }) + // Without the tree. This seal is standing in a directory that never ran the step's tools, so + // what it would ship is the state before them, and the host that did run them may still be + // inside one. Between the dispatch failing and the next step claiming the log there is a + // window where nothing fences that host, and the only thing that makes the window harmless + // is that nobody else publishes during it. + return activities.sealStep({ ...sealing(false), withoutTheTree: true }) + } + const resumes = () => uncertain !== undefined + if (resumes()) return closeElsewhere() + try { + return await seal(false) + } catch (error) { + // The seal is the other way a step loses its host, and it is the half that has to be written + // down: a step with no ending recorded is one no follower ever hears about. Re-sealing where + // a worker is answering is what an ordinary retry of this activity does; the pinned rule is + // the only reason it did not already happen. + if (stopped || !resumes()) throw error + return await closeElsewhere() + } } diff --git a/packages/temporal/test/pinned-retry.test.ts b/packages/temporal/test/pinned-retry.test.ts new file mode 100644 index 000000000000..0416047ef40f --- /dev/null +++ b/packages/temporal/test/pinned-retry.test.ts @@ -0,0 +1,105 @@ +import { expect, it } from "bun:test" +import { fileURLToPath } from "node:url" +import { ApplicationFailure } from "@temporalio/common" +import { TestWorkflowEnvironment } from "@temporalio/testing" +import { Worker } from "@temporalio/worker" + +// The server must not retry a pinned body or move it while its physical state is unknown. What the +// step does instead is close itself where a worker is answering: the call stays with the host that +// has it, and the turn goes on rather than ending with that host. +it("gives a pinned dispatch one attempt, keeps its call, and closes the step elsewhere", async () => { + const env = await TestWorkflowEnvironment.createLocal() + let phase: "tool" | "seal" = "tool" + const attempts = { tool: 0, seal: 0 } + const shared = { tool: 0, seal: 0 } + const sealed: Array = [] + try { + const worker = await Worker.create({ + connection: env.nativeConnection, + namespace: env.namespace, + taskQueue: "pin-retry-main", + workflowsPath: fileURLToPath(new URL("../src/workflow.ts", import.meta.url)), + activities: { + runModelCall: async () => ({ + kind: "called", + step: 1, + calls: phase === "tool" ? [{ id: "call_write", name: "write", assistantMessageID: "msg_write" }] : [], + owner: "run:1:1", + queue: "pin-retry-tools", + }), + runToolCall: async () => { + shared.tool++ + return { outcome: "settled" } + }, + sealStep: async (input: { withoutTheTree?: boolean }) => { + shared.seal++ + sealed.push(input.withoutTheTree === true) + return { continue: false, step: 1, promotion: null } + }, + }, + }) + const pinned = await Worker.create({ + connection: env.nativeConnection, + namespace: env.namespace, + taskQueue: "pin-retry-tools", + activities: { + runToolCall: async () => { + attempts.tool++ + throw ApplicationFailure.create({ message: "tool failed after dispatch", type: "ToolUnavailable" }) + }, + sealStep: async () => { + attempts.seal++ + throw ApplicationFailure.create({ message: "seal failed after dispatch", type: "SealUnavailable" }) + }, + }, + }) + await worker.runUntil(() => + pinned.runUntil(async () => { + for (const kind of ["tool", "seal"] as const) { + phase = kind + shared.tool = 0 + shared.seal = 0 + sealed.length = 0 + const handle = await env.client.workflow.start("sessionTurn", { + workflowId: `pin-retry-session-${kind}`, + taskQueue: "pin-retry-main", + args: ["ses_pin_retry", { stepped: true, startWithWake: false }], + }) + const resumed = handle.executeUpdate("resume").then( + () => "completed", + () => "failed", + ) + let timer: ReturnType | undefined + try { + const outcome = await Promise.race([ + resumed, + new Promise((resolve) => { + timer = setTimeout(() => resolve("waiting for retries"), 2_000) + }), + ]) + // The turn is handed back, not ended: the step was closed on the shared queue and the + // supervisor gets a result to carry on from. + expect(outcome).toBe("completed") + // One attempt on the pinned queue, and never a second one anywhere: a retry's queue + // timeout cannot rule out the first attempt still running. + expect(attempts[kind]).toBe(1) + // The call itself does not move. Only the seal does, and that is the step saying what + // it had rather than the tool being run somewhere else. + expect(shared.tool).toBe(0) + expect(shared.seal).toBe(1) + // And it seals without the tree: this worker never ran the step, and the one that did + // may still be inside a tool, so rebuilding here would put it on the newest state and + // shipping from here would publish the state before the step. + expect(sealed).toEqual([true]) + } finally { + clearTimeout(timer) + await handle.terminate() + await resumed + } + } + }), + ) + } finally { + await env.teardown() + } +}, 120_000) diff --git a/packages/temporal/test/replay.test.ts b/packages/temporal/test/replay.test.ts new file mode 100644 index 000000000000..498221517aa9 --- /dev/null +++ b/packages/temporal/test/replay.test.ts @@ -0,0 +1,70 @@ +// A stepped turn's decisions are workflow code, so they are written into the history of every run +// that hits them. The one check that must always hold is that this code replays a history it +// wrote itself; a run that lost a pinned dispatch is the branch with the most decisions in it. + +import { expect, it } from "bun:test" +import { fileURLToPath } from "node:url" +import { ApplicationFailure } from "@temporalio/common" +import { TestWorkflowEnvironment } from "@temporalio/testing" +import { Worker } from "@temporalio/worker" + +const workflowsPath = fileURLToPath(new URL("../src/workflow.ts", import.meta.url)) + +it("replays a stepped turn that lost its host", async () => { + const env = await TestWorkflowEnvironment.createLocal() + try { + let called = false + const worker = await Worker.create({ + connection: env.nativeConnection, + namespace: env.namespace, + taskQueue: "replay-main", + workflowsPath, + activities: { + runModelCall: async () => { + // One step with a tool, then nothing left to do: the turn has to be able to end after the + // step that lost its host, or the replay would only ever see the failure. + if (called) return { kind: "settled", result: { continue: false, step: 2, promotion: null } } + called = true + return { + kind: "called", + step: 1, + calls: [{ id: "call_lost", name: "write", assistantMessageID: "msg_lost" }], + owner: "run:1:1", + queue: "replay-tools", + } + }, + runToolCall: async () => ({ outcome: "settled" }), + sealStep: async () => ({ continue: true, step: 2, promotion: null }), + }, + }) + const pinned = await Worker.create({ + connection: env.nativeConnection, + namespace: env.namespace, + taskQueue: "replay-tools", + activities: { + runToolCall: async () => { + throw ApplicationFailure.create({ message: "the tool failed after starting", type: "ToolUnavailable" }) + }, + sealStep: async () => ({ continue: false, step: 2, promotion: null }), + }, + }) + + const history = await worker.runUntil(() => + pinned.runUntil(async () => { + const handle = await env.client.workflow.start("sessionTurn", { + workflowId: "replay-session", + taskQueue: "replay-main", + args: ["ses_replay", { stepped: true, startWithWake: false }], + }) + await handle.executeUpdate("resume") + await handle.terminate().catch(() => undefined) + return handle.fetchHistory() + }), + ) + + await Worker.runReplayHistory({ workflowsPath }, history) + expect(history.events?.length ?? 0).toBeGreaterThan(0) + } finally { + await env.teardown() + } +}, 120_000) diff --git a/packages/temporal/test/stepped-drain-writers.test.ts b/packages/temporal/test/stepped-drain-writers.test.ts new file mode 100644 index 000000000000..6359d7bdddfb --- /dev/null +++ b/packages/temporal/test/stepped-drain-writers.test.ts @@ -0,0 +1,95 @@ +// The seal is the one unit of a step that can run away from the host that ran the step, and when it +// does it has to leave the tree alone: rebuilding here would put this host on the newest state while +// the one that ran the tools may still be writing, and what it captured would be the state before +// them. This drives the real drain, because the option is threaded through it and nothing else +// checks that it reaches the materializer. + +import { expect } from "bun:test" +import { mkdtemp } from "node:fs/promises" +import { tmpdir } from "node:os" +import path from "path" +import { Context, Effect, Layer } from "effect" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { Global } from "@opencode-ai/core/global" +import { WorktreeMaterializer } from "@opencode-ai/core/session/execution/worktree" +import { SessionRunner } from "@opencode-ai/core/session/runner" +import { testEffect } from "../../core/test/lib/effect" +import { type DrainDeps, makeInSession } from "../src/drain" +import { makeSteppedDrains } from "../src/stepped-drain" + +const it = testEffect(Layer.empty) + +const materializerStack = (data: string) => + AppNodeBuilder.build(WorktreeMaterializer.node, [ + [Database.node, Database.layerFromPath(":memory:")], + [Global.node, Layer.succeed(Global.Service, Global.make({ data }))], + ]) + +/** A drain over a runner that reports what it was asked to do. */ +const drainsOver = ( + worktrees: WorktreeMaterializer.Interface, + directory: string, + runner: Partial, +) => + makeSteppedDrains({ + inSession: makeInSession({ + store: { + get: () => Effect.succeed({ id: "ses_writers", location: { directory } }), + } as unknown as DrainDeps["store"], + // The location resolves to a runner and nothing else: what is under test is what surrounds + // the runner call, not what the runner does inside it. + locations: { + get: () => Layer.succeed(SessionRunner.Service, runner as SessionRunner.Interface), + } as unknown as DrainDeps["locations"], + ctx: Context.empty() as DrainDeps["ctx"], + events: { claim: () => Effect.void } as unknown as DrainDeps["events"], + worktrees, + }), + }) + +it.live("seals a step away from its host without touching the tree", () => + Effect.gen(function* () { + const root = yield* Effect.promise(() => mkdtemp(path.join(tmpdir(), "opencode-l2-writers-"))) + const data = path.join(root, "host-data") + const directory = path.join(root, "project") + const real = yield* WorktreeMaterializer.Service.pipe(Effect.provide(yield* Layer.build(materializerStack(data)))) + // The rebuild is the part that must not happen, and with no packs stored it would return + // without doing anything, so what is counted is the call rather than its effect. + let rebuilds = 0 + const worktrees: WorktreeMaterializer.Interface = { + ...real, + ensure: (dir, options) => { + rebuilds++ + return real.ensure(dir, options) + }, + } + + const sealed: unknown[] = [] + const { sealDrain } = drainsOver(worktrees, directory, { + sealStep: ((input: unknown) => { + sealed.push(input) + return Effect.succeed({ continue: true, step: 2, promotion: undefined }) + }) as unknown as SessionRunner.Interface["sealStep"], + }) + const seal = (withoutTheTree?: boolean) => + Effect.promise(() => + sealDrain( + { sessionID: "ses_writers", step: 1, owner: "run:1:1", withoutTheTree }, + new AbortController().signal, + ), + ) + + // An ordinary seal is on the host that ran the step, and it ships what the step produced. + yield* seal() + expect(rebuilds).toBe(1) + expect((sealed[0] as { withoutTheTree?: boolean }).withoutTheTree).toBeUndefined() + + // One closing a step away from that host is not. Rebuilding here would put this host on the + // newest state while the one that ran the tools may still be writing, and what it captured + // would be the state before them. + yield* seal(true) + expect(rebuilds).toBe(1) + expect((sealed[1] as { withoutTheTree?: boolean }).withoutTheTree).toBe(true) + }), +) diff --git a/packages/temporal/test/stepped-turn.test.ts b/packages/temporal/test/stepped-turn.test.ts index 9f2b9703dfb8..36fee9799c4c 100644 --- a/packages/temporal/test/stepped-turn.test.ts +++ b/packages/temporal/test/stepped-turn.test.ts @@ -430,6 +430,67 @@ describe("stepped turn, pinned to a worker", () => { } }) + it("closes the step elsewhere when a started pinned attempt times out", async () => { + // Heartbeat expiry also covers a process that can still write its directory, so its calls stay + // where they are and only the seal moves, without the tree. + const hostGone = () => + new ActivityFailure( + "activity Heartbeat timeout", + "runToolCall", + "1", + undefined, + undefined, + new TimeoutFailure("heartbeat timed out", undefined, "HEARTBEAT" as never), + ) + const shared = fakes(withQueue) + const failed = hostGone() + const run = makeSteppedTurn({ + activities: shared.activities, + isCancellation, + isHalt, + pinnedTo: () => ({ + runToolCall: async () => { + throw failed + }, + sealStep: async () => SEALED, + }), + })(INPUT) + + expect(await run).toEqual(SEALED) + expect(shared.tools).toHaveLength(0) + expect(shared.seals.map((s) => s.withoutTheTree)).toEqual([true]) + }) + + it("keeps an uncertain pinned sibling's calls off the shared queue and seals without the tree", async () => { + const release = Promise.withResolvers() + const refused = Promise.withResolvers() + let sharedBeforeTheSiblingEnded = 0 + const shared = fakes(withQueue) + const run = makeSteppedTurn({ + activities: shared.activities, + isCancellation, + isHalt, + pinnedTo: () => ({ + runToolCall: async (input) => { + if (input.call.id === "call_a") return release.promise + refused.resolve() + throw unclaimed() + }, + sealStep: async () => SEALED, + }), + })(INPUT) + await refused.promise + await new Promise((resolve) => setTimeout(resolve, 0)) + sharedBeforeTheSiblingEnded = shared.tools.length + const failed = new Error("the started activity timed out") + release.reject(failed) + expect(await run).toEqual(SEALED) + + expect(sharedBeforeTheSiblingEnded).toBe(0) + expect(shared.tools).toHaveLength(0) + expect(shared.seals.map((s) => s.withoutTheTree)).toEqual([true]) + }) + it("uses the shared queue when the model call reported no queue of its own", async () => { const { run, tools, pinnedTools } = called({ ...withQueue, queue: undefined })