Fix loop completion deadlock and queued prompt controls

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## 0.1.44-beta.6 - 2026-09-09
+
+### Fixed
+
+- Complete terminal loop runs without waiting on the owner session that is
+  executing `run_once`, preventing a circular wait and duplicate notification.
+- Deliver background loop notifications asynchronously while returning direct
+  `run_once` results to the calling session.
+- Keep `Send now` attached to the first visible queued prompt when synthetic
+  runtime notifications are present.
+- Add native PTY coverage for queued notifications, transcript paging and
+  double-`Esc` cancellation.
+
 ## 0.1.44-beta.5 - 2026-09-09
 
 ### Fixed

src/mendcode/packages/opencode/script/session-runtime-smoke.ts

@@ -318,7 +318,7 @@ function openAIStream(chunks: Record<string, unknown>[], done = true) {
   return encoder.encode(payload)
 }
 
-async function startFakeProvider(sandbox: Sandbox): Promise<RunningFakeProvider> {
+async function startFakeProvider(sandbox: Sandbox, longStream = false): Promise<RunningFakeProvider> {
   const provider = new URL(sandbox.providerURL)
   const calls: FakeProviderCall[] = []
   const active = new Set<ReadableStreamDefaultController<Uint8Array>>()
@@ -382,7 +382,9 @@ async function startFakeProvider(sandbox: Sandbox): Promise<RunningFakeProvider>
                   {
                     id: `chatcmpl-${prompt === UNAFFECTED_PROMPT ? "unaffected" : "primary"}`,
                     object: "chat.completion.chunk",
-                    choices: [{ index: 0, delta: { content: `${prompt} active` } }],
+                    choices: [{ index: 0, delta: { content: longStream
+                      ? `${prompt} active\n\n${Array.from({ length: 80 }, (_, index) => `SCROLL-ROW-${index}`).join("\n\n")}\n\nSCROLL-END`
+                      : `${prompt} active` } }],
                   },
                 ],
                 false,
@@ -818,6 +820,77 @@ async function abortSession(server: RunningServer, sandbox: Sandbox, sessionID:
   expectStatus(response, 200, "session abort")
 }
 
+async function runQueuedLoopTuiScenario(sandbox: Sandbox) {
+  const provider = await startFakeProvider(sandbox, true)
+  let backend: RunningServer | undefined
+  let proxy: RunningTransportProxy | undefined
+  let tui: RunningTui | undefined
+  try {
+    backend = await startServer({ entrypoint: SOURCE_ENTRYPOINT, installed: false, sandbox })
+    await waitForHealth(backend, sandbox)
+    const primary = await createSession(backend, sandbox)
+    proxy = await startTransportProxy(backend.url)
+    tui = await startTui(sandbox, proxy.url, primary.id)
+    await waitUntil("queued loop TUI event subscription", () => proxy!.activeStreams() > 0)
+    await submitAsyncPrompt(backend, sandbox, primary.id, STREAM_PROMPT)
+    const call = await provider.waitForCall(STREAM_PROMPT)
+    await waitUntil("long stream prompt rendered", () => terminalContains(tui!.output(), STREAM_PROMPT))
+    for (let page = 0; page < 10; page++) {
+      tui.write("\x1b[6~")
+      await sleep(50)
+    }
+    await waitUntil("long stream end rendered", () => terminalContains(tui!.output(), "SCROLL-END"))
+
+    const notification = await requestJSON(backend, sandbox, `/session/${primary.id}/prompt_async`, {
+      method: "POST",
+      body: JSON.stringify({
+        agent: "build",
+        model: { providerID: FAKE_PROVIDER_ID, modelID: FAKE_MODEL_ID },
+        parts: [
+          {
+            type: "text",
+            text: "Internal loop stopped notification",
+            synthetic: true,
+            metadata: { kind: "loop_owner_notification" },
+          },
+        ],
+      }),
+    })
+    expectStatus(notification, 204, "queued internal loop notification")
+    const queueOffset = tui.output().length
+    await submitAsyncPrompt(backend, sandbox, primary.id, "VISIBLE-QUEUED-QUESTION")
+    await waitUntil("visible queued question keeps Send now", () => {
+      const output = tui!.output().slice(queueOffset)
+      return terminalContains(output, "VISIBLE-QUEUED-QUESTION") && terminalContains(output, "SEND")
+    })
+
+    const upOffset = tui.output().length
+    tui.write("\x1b[5~")
+    await waitUntil("PageUp reveals earlier long stream content", () =>
+      terminalContains(tui!.output().slice(upOffset), "SCROLL-ROW-"),
+    )
+    const downOffset = tui.output().length
+    tui.write("\x1b[6~")
+    await waitUntil("PageDown reaches the queued question", () =>
+      terminalContains(tui!.output().slice(downOffset), "VISIBLE-QUEUED-QUESTION"),
+    )
+
+    await interruptThroughTui(tui, "queued loop notification")
+    await withTimeout("queued loop active provider cancellation", call.abort)
+    await waitForSessionStatus(backend, sandbox, primary.id, "idle")
+    assert(provider.calls.length === 1, "double Esc started a hidden or human queued prompt")
+  } catch (error) {
+    throw new Error(
+      `${error instanceof Error ? error.message : String(error)}\nPTY:\n${stripAnsi(tui?.output() ?? "").slice(-8000)}`,
+    )
+  } finally {
+    await tui?.stop()
+    await proxy?.stop()
+    await backend?.stop()
+    await provider.stop()
+  }
+}
+
 async function runTuiPtyScenario(sandbox: Sandbox, tuiEntrypoint = SOURCE_ENTRYPOINT) {
   let provider: RunningFakeProvider | undefined
   let backend: RunningServer | undefined
@@ -1352,6 +1425,16 @@ async function main() {
   if (mode === "--managed-recovery") return managedRecovery(process.argv[3], process.argv[4], process.argv[5])
 
   const requireInstalled = process.argv.includes("--require-installed")
+  if (process.argv.includes("--loop-queue-only")) {
+    const sandbox = await createSandbox("loop-queue")
+    try {
+      await runQueuedLoopTuiScenario(sandbox)
+      console.log("PASS real TUI/PTY hidden loop notification, visible Send now, PageUp/PageDown, and double Esc")
+      return
+    } finally {
+      await rm(sandbox.root, { recursive: true, force: true })
+    }
+  }
   if (process.argv.includes("--pty-only")) {
     const sandbox = await createSandbox("pty")
     try {

src/mendcode/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx

@@ -174,6 +174,7 @@ import {
   sessionPendingInputSessionIDs,
   sessionPendingInputStatus,
   sessionPromptVisible,
+  sessionNextVisibleQueuedMessageID,
   sessionTranscriptBottomSpacer,
   sessionLoopReceipt,
   shouldRenderSessionLoopCard,
@@ -970,7 +971,7 @@ export function Session() {
       .filter((message): message is UserMessage => message.role === "user" && queuedIDs.has(message.id))
       .toSorted(compareSessionMessages)
   })
-  const nextQueuedMessageID = createMemo(() => queuedMessages()[0]?.id)
+  const nextQueuedMessageID = createMemo(() => sessionNextVisibleQueuedMessageID(queuedMessages(), sync.data.part))
   const messageByID = createMemo(() => new Map(messages().map((message) => [message.id, message] as const)))
   const pinnedTurnUserMessageID = createMemo(() =>
     sessionPinnedUserMessageID({

src/mendcode/packages/opencode/src/cli/cmd/tui/util/session-layout.ts

@@ -1,3 +1,12 @@
+export function sessionNextVisibleQueuedMessageID(
+  messages: ReadonlyArray<{ id: string }>,
+  parts: Readonly<Record<string, ReadonlyArray<{ type: string; synthetic?: boolean; text?: string }>>>,
+) {
+  return messages.find((message) =>
+    parts[message.id]?.some((part) => part.type === "text" && !part.synthetic && Boolean(part.text)),
+  )?.id
+}
+
 export function sessionPromptVisible(input: {
   isChildSession: boolean
   permissionCount: number

src/mendcode/packages/opencode/src/session/loop-runner.ts

@@ -4,6 +4,7 @@ import { SessionPrompt } from "@/session/prompt"
 import { LoopWorkflow } from "@/session/loop"
 import * as MessageV2 from "@/session/message-v2"
 import { Session } from "@/session/session"
+import type { SessionID } from "@/session/schema"
 import { ModelID, ProviderID } from "@/provider/schema"
 import { InstanceState } from "@/effect/instance-state"
 import { InstanceStore } from "@/project/instance-store"
@@ -41,6 +42,7 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/Lo
 
 export type RunOneInput = {
   id: LoopWorkflow.LoopID
+  callerSessionID?: SessionID
   execute?: boolean
   reportOnly?: boolean
   reason?: string
@@ -1139,12 +1141,18 @@ export const layer = Layer.effect(
       }
       const notifyOwner = (workflow: LoopWorkflow.Info, run: LoopWorkflow.RunInfo, summary: string) => {
         if (!shouldNotifyOwner(workflow) || !workflow.ownerSessionID) return Effect.void
+        // An owner executing run_once receives the result through its tool.
+        // Queueing a second prompt here both duplicates that result and puts an
+        // internal notification ahead of the user's already waiting messages.
+        if (workflow.ownerSessionID === input.callerSessionID) return Effect.void
         const text =
           workflow.state === "completed"
             ? parentCompletionPrompt(workflow, { summary }, run)
             : parentStatusPrompt(workflow, run, summary)
+        // Background runs must persist the notification without waiting for
+        // the owner's current turn, which may itself be waiting for this run.
         return prompt
-          .prompt({
+          .promptAsync({
             sessionID: workflow.ownerSessionID,
             agent: workflow.spec.agent,
             model: promptModel(workflow),

src/mendcode/packages/opencode/src/session/prompt.ts

@@ -576,7 +576,8 @@ export const layer = Layer.effect(
         cancel: (sessionID: SessionID) => cancel(sessionID),
         resolvePromptParts: (template: string) => resolvePromptParts(template),
         prompt: (input: PromptInput) => prompt(input),
-      } satisfies TaskPromptOps
+        promptAsync: (input: PromptInput) => promptAsync(input),
+      } satisfies TaskPromptOps & Pick<Interface, "promptAsync">
     })
 
     const findOrphanedAssistantOnCancel = Effect.fn("SessionPrompt.findOrphanedAssistantOnCancel")(function* (

src/mendcode/packages/opencode/src/tool/loop.ts

@@ -1344,9 +1344,9 @@ export const LoopTool = Tool.define<typeof Parameters, Metadata, LoopWorkflow.Se
           if (action === "resume") workflow = yield* workflows.resume({ id: workflow.id, reason: params.reason })
           if (action === "stop") workflow = yield* workflows.stop({ id: workflow.id, reason: params.reason })
           if (action === "run_once") {
-            const promptOps = ctx.extra?.promptOps as Pick<SessionPrompt.Interface, "prompt"> | undefined
+            const promptOps = ctx.extra?.promptOps as Pick<SessionPrompt.Interface, "prompt" | "promptAsync"> | undefined
             if (!promptOps) throw new Error("Loop run_once requires prompt operations from the active session.")
-            let result = yield* runner.runOne({ id: workflow.id, execute: true, reason: params.reason, trigger: "run-once" }).pipe(
+            let result = yield* runner.runOne({ id: workflow.id, callerSessionID: ctx.sessionID, execute: true, reason: params.reason, trigger: "run-once" }).pipe(
               Effect.provideService(LoopWorkflow.Service, workflows),
               Effect.provideService(SessionPrompt.Service, promptOps as SessionPrompt.Interface),
               Effect.provideService(Session.Service, sessions),
@@ -1360,7 +1360,7 @@ export const LoopTool = Tool.define<typeof Parameters, Metadata, LoopWorkflow.Se
               auditRuns < maxAuditRuns
             ) {
               auditRuns++
-              result = yield* runner.runOne({ id: workflow.id, execute: true, reason: "Fresh completion audit after run_once candidate.", trigger: "run-once" }).pipe(
+              result = yield* runner.runOne({ id: workflow.id, callerSessionID: ctx.sessionID, execute: true, reason: "Fresh completion audit after run_once candidate.", trigger: "run-once" }).pipe(
                 Effect.provideService(LoopWorkflow.Service, workflows),
                 Effect.provideService(SessionPrompt.Service, promptOps as SessionPrompt.Interface),
                 Effect.provideService(Session.Service, sessions),

src/mendcode/packages/opencode/test/cli/tui/session-layout.test.ts

@@ -18,10 +18,24 @@ import {
   sessionTopbarNavLayout,
   sessionUsageBarDisplayWidth,
   sessionPromptVisible,
+  sessionNextVisibleQueuedMessageID,
   truncateEndDisplay,
 } from "../../../src/cli/cmd/tui/util/session-layout"
 
 describe("session layout", () => {
+  test("hidden loop notifications do not take the visible queued prompt send control", () => {
+    const messages = [{ id: "notification" }, { id: "question" }, { id: "followup" }]
+    const parts = {
+      notification: [{ type: "text", text: "Loop stopped", synthetic: true }],
+      question: [{ type: "text", text: "What happened?" }],
+      followup: [{ type: "text", text: "Hello?" }],
+    }
+    expect(sessionNextVisibleQueuedMessageID(messages, parts)).toBe("question")
+    expect(sessionNextVisibleQueuedMessageID(messages.slice(2), parts)).toBe("followup")
+    expect(sessionNextVisibleQueuedMessageID(messages.slice(0, 1), parts)).toBeUndefined()
+    expect(sessionNextVisibleQueuedMessageID(messages, {})).toBeUndefined()
+  })
+
   test("renders workflow cards for normalized and persisted full presentation profiles", () => {
     expect(shouldRenderSessionWorkflowCard("mendcode")).toBe(true)
     expect(shouldRenderSessionWorkflowCard("full")).toBe(true)

src/mendcode/packages/opencode/test/session/loop.test.ts

@@ -107,10 +107,11 @@ function runWithWorktree<A, E>(fx: Effect.Effect<A, E, LoopWorkflowService | Wor
 function runRunner<A, E>(
   fx: Effect.Effect<A, E, LoopRunner.Service | LoopWorkflowService | SessionPrompt.Service | Session.Service>,
   promptText?: string | MessageV2.WithParts | ((call: number) => string | MessageV2.WithParts),
-  options?: { isolatedWorkspaceContext?: boolean },
+  options?: { isolatedWorkspaceContext?: boolean; busySessionID?: string },
 ) {
   let prompts = 0
   const promptCalls: PromptInput[] = []
+  const asyncPromptCalls: PromptInput[] = []
   const promptDirectories: string[] = []
   const promptLayer = Layer.succeed(
     SessionPrompt.Service,
@@ -121,6 +122,7 @@ function runRunner<A, E>(
       interrupt: () => Effect.succeed(false),
       prompt: (input: PromptInput) =>
         Effect.gen(function* () {
+          if (input.sessionID === options?.busySessionID) return yield* Effect.never
           prompts++
           promptCalls.push(input)
           promptDirectories.push((yield* InstanceState.context).directory)
@@ -132,9 +134,11 @@ function runRunner<A, E>(
               : response
         }),
       promptAsync: (input: PromptInput) =>
-        Effect.sync(() => {
+        Effect.gen(function* () {
           prompts++
           promptCalls.push(input)
+          asyncPromptCalls.push(input)
+          promptDirectories.push((yield* InstanceState.context).directory)
           const response = typeof promptText === "function" ? promptText(prompts) : promptText
           return typeof response === "string"
             ? promptMessage(response)
@@ -192,7 +196,7 @@ function runRunner<A, E>(
         options?.isolatedWorkspaceContext ? runnerLayer.pipe(Layer.provideMerge(instanceStoreLayer)) : runnerLayer,
       ),
     ),
-  ).then((value) => ({ value, prompts, promptCalls, promptDirectories }))
+  ).then((value) => ({ value, prompts, promptCalls, asyncPromptCalls, promptDirectories }))
 }
 
 const svc = {
@@ -2605,6 +2609,46 @@ describe("loop workflow service", () => {
     })
   })
 
+  for (const directOwner of [false, true]) {
+    test(`stopped run_once returns while its owner is busy (direct owner: ${directOwner})`, async () => {
+      await using tmp = await tmpdir({ git: true })
+      await WithInstance.provide({
+        directory: tmp.path,
+        fn: async () => {
+          const owner = await Effect.runPromise(
+            Session.Service.use((session) => session.create({ title: "Busy loop owner" })).pipe(
+              Effect.provide(Session.defaultLayer),
+            ),
+          )
+          const draft = await svc.createDraft({
+            name: "Stopped device observation",
+            objective: "Observe the same device boot without restarting it.",
+            ownerSessionID: owner.id,
+            trigger: { mode: "manual" },
+            policy: { maxTurns: 5 },
+          })
+          await svc.activate(draft.id)
+          const result = await runRunner(
+            LoopRunner.Service.use((runner) =>
+              runner.runOne({
+                id: draft.id,
+                callerSessionID: directOwner ? owner.id : undefined,
+                execute: true,
+                trigger: "run-once",
+              }),
+            ).pipe(Effect.timeout("2 seconds")),
+            "LOOP_CHECKPOINT:\nstatus: stop\nsummary: Device unexpectedly rebooted.\nevidence:\n- boot identity changed",
+            { busySessionID: owner.id },
+          )
+          expect(result.value.state).toBe("stopped")
+          expect(result.asyncPromptCalls.map((input) => input.sessionID)).toEqual(directOwner ? [] : [owner.id])
+          expect(result.prompts).toBe(directOwner ? 1 : 2)
+          expect((await svc.snapshot(draft.id)).workflow.state).toBe("stopped")
+        },
+      })
+    })
+  }
+
   test("terminal loop failures notify the owner with authoritative persisted state", async () => {
     await using tmp = await tmpdir({ git: true })
     await WithInstance.provide({
@@ -2648,6 +2692,7 @@ describe("loop workflow service", () => {
         expect(terminal.value.state).toBe("failed")
         expect(terminal.value.summary).toContain("will not retry automatically")
         expect(terminal.prompts).toBe(2)
+        expect(terminal.asyncPromptCalls.map((input) => input.sessionID)).toEqual([owner.id])
         expect(terminal.promptCalls[1]?.sessionID).toBe(owner.id)
         expect(promptInputText(terminal.promptCalls[1])).toContain('<mendcode_runtime_event type="loop_status">')
         expect(promptInputText(terminal.promptCalls[1])).toContain("State: failed")
@@ -4736,6 +4781,7 @@ describe("loop workflow service", () => {
         expect(executed.promptCalls[0]?.sessionID).toBe(rootSessionID)
         expect(executed.promptCalls[1]?.sessionID).toBe(owner.id)
         expect(promptInputText(executed.promptCalls[1])).toContain("Loop workflow completed: Notify owner")
+        expect(executed.asyncPromptCalls.map((input) => input.sessionID)).toEqual([owner.id])
       },
     })
   })