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 })