From 80dc5235f591a9a39866caa1aebc8ca6fddb2784 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sun, 25 Oct 2026 08:24:30 -0700 Subject: [PATCH] fix(app): let a later steer move work an earlier move missed (#58984) --- packages/app/src/composer/adapter.ts | 4 +- packages/app/src/composer/submit.test.ts | 71 ++++++++++++++++++- packages/app/src/composer/submit.ts | 6 +- packages/app/src/session/composer/queue.ts | 10 +-- .../session/requests/steer-background.test.ts | 57 +++++++++------ .../src/session/requests/steer-background.ts | 23 +++--- 6 files changed, 120 insertions(+), 51 deletions(-) diff --git a/packages/app/src/composer/adapter.ts b/packages/app/src/composer/adapter.ts index 5f5ea4400f5..e55190d8100 100644 --- a/packages/app/src/composer/adapter.ts +++ b/packages/app/src/composer/adapter.ts @@ -94,8 +94,8 @@ export type ComposerSession = { | undefined > admitted: (messageID: string) => boolean - // Claims the run's blocking work for a steer; undefined when there is none or it was already moved. - background?: () => { move: () => Promise; release: () => void } | undefined + // Moves the run's blocking work to the background once a steer is admitted; undefined for another session. + background?: () => { move: () => Promise } | undefined } type ComposerAdapterBase = { diff --git a/packages/app/src/composer/submit.test.ts b/packages/app/src/composer/submit.test.ts index 73de80caa77..3ff320925c0 100644 --- a/packages/app/src/composer/submit.test.ts +++ b/packages/app/src/composer/submit.test.ts @@ -511,18 +511,41 @@ describe("Composer submission", () => { describe("steering while foreground work runs", () => { // A long shell command blocks the run; moving it resolves only the move, never the admission. - function running(input: { fail?: boolean; prompt?: () => Promise } = {}) { + // `unstarted` moves before the shell's job starts, so the server finds nothing to move and the call still blocks. + function running( + input: { + fail?: boolean + failOnce?: boolean + unstarted?: boolean + blocking?: { partID: string }[] + prompt?: () => Promise + } = {}, + ) { const calls: string[] = [] const failures: string[] = [] const state = createMemoryComposerState({ prompt: "use Beta instead" }).capture() + const blocking = { parts: input.blocking ?? [{ partID: "prt_shell" }] } const claim = createSteerBackground({ sessionID: () => "session-1", - blocking: () => [{ partID: "prt_shell" }], + blocking: () => blocking.parts, move: async () => { calls.push("background") if (input.fail) throw new Error("server unavailable") + + if (input.failOnce) { + input.failOnce = false + throw new Error("server unavailable") + } + + if (input.unstarted) { + input.unstarted = false + + return + } + + blocking.parts = [] }, }) @@ -545,7 +568,7 @@ describe("Composer submission", () => { await new Promise((resolve) => setTimeout(resolve, 0)) } - return { calls, failures, submit } + return { calls, failures, submit, blocking } } test("moves the work to the background once the steer is admitted, and only once", async () => { @@ -558,6 +581,48 @@ describe("Composer submission", () => { expect(run.calls.filter((call) => call === "background")).toHaveLength(1) }) + test("a steer moves work that its previous steer's move missed", async () => { + const run = running({ unstarted: true }) + + await run.submit() + await run.submit() + expect(run.calls.filter((call) => call === "background")).toHaveLength(2) + }) + + test("a steer sent while the previous one is admitted moves the work when the previous move fails", async () => { + const admitted = Promise.withResolvers() + const attempts = { count: 0 } + + const run = running({ + failOnce: true, + prompt: async () => { + attempts.count++ + + if (attempts.count === 1) await admitted.promise + }, + }) + + await run.submit() + await run.submit() + admitted.resolve() + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(run.calls.filter((call) => call === "background")).toHaveLength(2) + expect(run.blocking.parts).toEqual([]) + }) + + test("moves a shell that starts while the steer is admitted", async () => { + const run = running({ + blocking: [], + prompt: async () => { + run.blocking.parts = [{ partID: "prt_shell" }] + }, + }) + + await run.submit() + expect(run.calls.filter((call) => call === "background")).toHaveLength(1) + expect(run.blocking.parts).toEqual([]) + }) + test("leaves the work alone for a queued prompt or an idle session", async () => { const run = running() diff --git a/packages/app/src/composer/submit.ts b/packages/app/src/composer/submit.ts index e6419f91b55..1346a6ee0fc 100644 --- a/packages/app/src/composer/submit.ts +++ b/packages/app/src/composer/submit.ts @@ -196,11 +196,7 @@ export function createComposerSubmit(input: ComposerSubmitInput) { return { ok: true as const } }, - (error) => { - background?.release() - - return { ok: false as const, error } - }, + (error) => ({ ok: false as const, error }), ) await started.cleanupReady diff --git a/packages/app/src/session/composer/queue.ts b/packages/app/src/session/composer/queue.ts index e9e03aadeb8..12d482fdf18 100644 --- a/packages/app/src/session/composer/queue.ts +++ b/packages/app/src/session/composer/queue.ts @@ -141,10 +141,7 @@ export function createSessionQueue(input: { if (!pending.some((item) => item.id === change.original && item.type === "user" && item.delivery === "queue")) throw new Error("Queued prompt was delivered before the edit") const background = input.background() - await data.session.prompt(admission).catch((error) => { - background?.release() - throw error - }) + await data.session.prompt(admission) void background?.move() setState("sent", (edit) => edit.original === change.original, "admitted", true) // Cancelling is a no-op once the original is gone, so one retry is safe. @@ -316,10 +313,7 @@ export function createSessionQueue(input: { void background?.move() hide([id]) }, - () => { - background?.release() - notify() - }, + () => notify(), ) } diff --git a/packages/app/src/session/requests/steer-background.test.ts b/packages/app/src/session/requests/steer-background.test.ts index 1a5c8f5287e..0c3e836b827 100644 --- a/packages/app/src/session/requests/steer-background.test.ts +++ b/packages/app/src/session/requests/steer-background.test.ts @@ -5,7 +5,7 @@ function setup(input: { fail?: boolean } = {}) { const moves: string[] = [] const state = { sessionID: "ses_alpha", blocking: [{ partID: "prt_shell" }] } - const claim = createSteerBackground({ + const steer = createSteerBackground({ sessionID: () => state.sessionID, blocking: () => state.blocking, move: async (sessionID) => { @@ -15,39 +15,52 @@ function setup(input: { fail?: boolean } = {}) { }, }) - return { claim, moves, state } + return { steer, moves, state } } describe("createSteerBackground", () => { - test("claims each blocking work item once", async () => { - const { claim, moves, state } = setup() + test("reads the blocking work when the move runs, after admission", async () => { + const { steer, moves, state } = setup() + state.blocking = [] + const background = steer("ses_alpha") - await claim("ses_alpha")?.move() - expect(claim("ses_alpha")).toBeUndefined() + state.blocking = [{ partID: "prt_shell" }] + await background?.move() + expect(moves).toEqual(["ses_alpha"]) - state.blocking = [{ partID: "prt_shell" }, { partID: "prt_subagent" }] - await claim("ses_alpha")?.move() + state.blocking = [] + await steer("ses_alpha")?.move() + expect(moves).toEqual(["ses_alpha"]) + }) + + // The server moves only work that already blocks the run: a call that starts its shell or subagent just after the + // move still blocks, and the next steer must move it. + test("work still blocking after a move is moved by the next steer", async () => { + const { steer, moves } = setup() + + await steer("ses_alpha")?.move() + await steer("ses_alpha")?.move() expect(moves).toEqual(["ses_alpha", "ses_alpha"]) }) - test("claims nothing without blocking work or for another session", () => { - const { claim, state } = setup() + test("overlapping steers each move, so a failed move leaves nothing behind", async () => { + const { steer, moves } = setup({ fail: true }) + const first = steer("ses_alpha") + const second = steer("ses_alpha") - expect(claim("ses_beta")).toBeUndefined() - state.blocking = [] - expect(claim("ses_alpha")).toBeUndefined() + await Promise.all([first?.move(), second?.move()]) + expect(moves).toEqual(["ses_alpha", "ses_alpha"]) }) - test("a released claim or failed move lets a later steer move the work", async () => { - const released = setup() + test("moves nothing for another session, and keeps a switched-away steer to its own work", async () => { + const { steer, moves, state } = setup() - released.claim("ses_alpha")?.release() - expect(released.claim("ses_alpha")).toBeDefined() + expect(steer("ses_beta")).toBeUndefined() - const failed = setup({ fail: true }) - - await failed.claim("ses_alpha")?.move() - await failed.claim("ses_alpha")?.move() - expect(failed.moves).toEqual(["ses_alpha", "ses_alpha"]) + const background = steer("ses_alpha") + state.sessionID = "ses_beta" + state.blocking = [] + await background?.move() + expect(moves).toEqual(["ses_alpha"]) }) }) diff --git a/packages/app/src/session/requests/steer-background.ts b/packages/app/src/session/requests/steer-background.ts index d8183a2fedd..03b6690e237 100644 --- a/packages/app/src/session/requests/steer-background.ts +++ b/packages/app/src/session/requests/steer-background.ts @@ -1,26 +1,27 @@ /** * A steer moves the run's blocking shells and subagents to the background, so the agent reads it at its next step. - * Each work item is claimed once: work already moved, or being moved, isn't again. A claim the steer never acts on, - * or a move that fails, is released so a later steer or the inline hint can move it. + * The move reads the blocking work once the steer is admitted, so work that started during admission moves too. + * Nothing is claimed: the server moves each job once and ignores repeated moves, so overlapping steers, a failed + * move, or one that ran before a job started never leave work for a later steer to skip. */ export function createSteerBackground(input: { sessionID: () => string | undefined blocking: () => readonly { partID: string }[] move: (sessionID: string) => Promise }) { - const moved = new Set() - return (sessionID: string) => { if (input.sessionID() !== sessionID) return - const parts = input.blocking().flatMap((task) => (moved.has(task.partID) ? [] : [task.partID])) - - if (parts.length === 0) return - parts.forEach((id) => moved.add(id)) - const release = () => parts.forEach((id) => moved.delete(id)) + const initial = input.blocking() return { - release, - move: () => input.move(sessionID).then(() => undefined, release), + move: () => { + // After a switch to another session, the blocking work shown is that session's. + const blocking = input.sessionID() === sessionID ? input.blocking() : initial + + if (blocking.length === 0) return Promise.resolve() + + return input.move(sessionID).catch(() => undefined) + }, } } }