diff --git a/CHANGELOG.md b/CHANGELOG.md index ff0c4d9d..1a080fba 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/mendcode/packages/opencode/script/session-runtime-smoke.ts b/src/mendcode/packages/opencode/script/session-runtime-smoke.ts index cb3f816a..364c24e9 100644 --- a/src/mendcode/packages/opencode/script/session-runtime-smoke.ts +++ b/src/mendcode/packages/opencode/script/session-runtime-smoke.ts @@ -318,7 +318,7 @@ function openAIStream(chunks: Record[], done = true) { return encoder.encode(payload) } -async function startFakeProvider(sandbox: Sandbox): Promise { +async function startFakeProvider(sandbox: Sandbox, longStream = false): Promise { const provider = new URL(sandbox.providerURL) const calls: FakeProviderCall[] = [] const active = new Set>() @@ -382,7 +382,9 @@ async function startFakeProvider(sandbox: Sandbox): Promise { 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 { diff --git a/src/mendcode/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx b/src/mendcode/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx index b72afbff..f1305fd5 100644 --- a/src/mendcode/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx +++ b/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({ diff --git a/src/mendcode/packages/opencode/src/cli/cmd/tui/util/session-layout.ts b/src/mendcode/packages/opencode/src/cli/cmd/tui/util/session-layout.ts index 9dd27b51..823cc541 100644 --- a/src/mendcode/packages/opencode/src/cli/cmd/tui/util/session-layout.ts +++ b/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>>, +) { + 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 diff --git a/src/mendcode/packages/opencode/src/session/loop-runner.ts b/src/mendcode/packages/opencode/src/session/loop-runner.ts index 5c53b28c..e621bb6a 100644 --- a/src/mendcode/packages/opencode/src/session/loop-runner.ts +++ b/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()("@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), diff --git a/src/mendcode/packages/opencode/src/session/prompt.ts b/src/mendcode/packages/opencode/src/session/prompt.ts index cd635d05..7d172785 100644 --- a/src/mendcode/packages/opencode/src/session/prompt.ts +++ b/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 }) const findOrphanedAssistantOnCancel = Effect.fn("SessionPrompt.findOrphanedAssistantOnCancel")(function* ( diff --git a/src/mendcode/packages/opencode/src/tool/loop.ts b/src/mendcode/packages/opencode/src/tool/loop.ts index 99711f01..69c6b9c4 100644 --- a/src/mendcode/packages/opencode/src/tool/loop.ts +++ b/src/mendcode/packages/opencode/src/tool/loop.ts @@ -1344,9 +1344,9 @@ export const LoopTool = Tool.define | undefined + const promptOps = ctx.extra?.promptOps as Pick | 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 { + 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) diff --git a/src/mendcode/packages/opencode/test/session/loop.test.ts b/src/mendcode/packages/opencode/test/session/loop.test.ts index 1e1071da..21270d6c 100644 --- a/src/mendcode/packages/opencode/test/session/loop.test.ts +++ b/src/mendcode/packages/opencode/test/session/loop.test.ts @@ -107,10 +107,11 @@ function runWithWorktree(fx: Effect.Effect( fx: Effect.Effect, 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( 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( : 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( 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('') 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]) }, }) }) diff --git a/src/mendcode/packages/opencode/test/tool/loop.test.ts b/src/mendcode/packages/opencode/test/tool/loop.test.ts index 24bd0d1b..8217a797 100644 --- a/src/mendcode/packages/opencode/test/tool/loop.test.ts +++ b/src/mendcode/packages/opencode/test/tool/loop.test.ts @@ -29,6 +29,87 @@ afterEach(async () => { const it = testEffect(Layer.mergeAll(ToolRegistry.defaultLayer, Session.defaultLayer, LoopWorkflow.defaultLayer, CrossSpawnSpawner.defaultLayer)) describe("tool.loop", () => { + it.live("run_once returns its terminal result to the calling owner without queueing another prompt", () => + provideTmpdirInstance( + () => + Effect.gen(function* () { + const registry = yield* ToolRegistry.Service + const sessions = yield* Session.Service + const parent = yield* sessions.create({ title: "Calling loop owner" }) + const tool = (yield* registry.tools({ + providerID: ProviderID.opencode, + modelID: ModelID.make("gpt-5"), + agent: { name: "build", mode: "primary" as const, permission: [], options: {} }, + })).find((item) => item.id === LoopTool.id) + if (!tool) throw new Error("Loop tool not found") + const ctx = { ...baseCtx, sessionID: parent.id, ask: () => Effect.void } + const activated = yield* tool.execute( + { + action: "activate", + objective: "Observe the device without restarting it.", + triggerMode: "manual", + maxTurns: 3, + reportOnly: true, + ensureService: false, + }, + ctx, + ) + let notifications = 0 + const result = yield* tool + .execute( + { + action: "run_once", + workflowID: activated.metadata.workflowID, + ensureService: false, + }, + { + ...ctx, + extra: { + promptOps: { + prompt: (input: { sessionID: SessionID }) => + Effect.succeed({ + info: { + id: MessageID.make("msg_loop_worker"), + sessionID: input.sessionID, + role: "assistant", + parentID: baseCtx.messageID, + agent: "build", + mode: "build", + modelID: ModelID.make("gpt-5"), + providerID: ProviderID.opencode, + time: { created: 1, completed: 2 }, + finish: "stop", + path: { cwd: "/tmp", root: "/tmp" }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + }, + parts: [ + { + id: PartID.make("prt_loop_stop"), + sessionID: input.sessionID, + messageID: MessageID.make("msg_loop_worker"), + type: "text", + text: "LOOP_CHECKPOINT:\nstatus: stop\nsummary: Device unexpectedly rebooted.\nevidence:\n- boot identity changed", + }, + ], + } satisfies MessageV2.WithParts), + promptAsync: () => + Effect.sync(() => { + notifications++ + }).pipe(Effect.andThen(Effect.never)), + }, + }, + }, + ) + .pipe(Effect.timeout("2 seconds")) + expect(result.metadata.state).toBe("stopped") + expect(result.output).toContain("Device unexpectedly rebooted.") + expect(notifications).toBe(0) + }), + { git: true }, + ), + ) + it.live("registry exposes loop tool and activation creates a durable workflow", () => provideTmpdirInstance( () =>