fix(app): let a later steer move work an earlier move missed (#58984)

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<void>; release: () => void } | undefined
+  // Moves the run's blocking work to the background once a steer is admitted; undefined for another session.
+  background?: () => { move: () => Promise<void> } | undefined
 }
 
 type ComposerAdapterBase = {

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<void> } = {}) {
+    // `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<void>
+      } = {},
+    ) {
       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<void>()
+      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()
 

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

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

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

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<void>
 }) {
-  const moved = new Set<string>()
-
   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)
+      },
     }
   }
 }