From b2fdd728d0d0c5bb4479b90750a2479e44b3cd3c Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:29:42 +0900 Subject: [PATCH 01/17] fix(prompt-async): add session idle gate --- src/hooks/shared/prompt-async-gate.test.ts | 84 ++++++++++++++++ src/hooks/shared/prompt-async-gate.ts | 111 +++++++++++++++++++++ 2 files changed, 195 insertions(+) create mode 100644 src/hooks/shared/prompt-async-gate.test.ts create mode 100644 src/hooks/shared/prompt-async-gate.ts diff --git a/src/hooks/shared/prompt-async-gate.test.ts b/src/hooks/shared/prompt-async-gate.test.ts new file mode 100644 index 000000000..6df4a189e --- /dev/null +++ b/src/hooks/shared/prompt-async-gate.test.ts @@ -0,0 +1,84 @@ +import { afterEach, describe, expect, test } from "bun:test" + +import { + promptAsyncAfterSessionIdle, + releaseAllPromptAsyncReservationsForTesting, +} from "./prompt-async-gate" + +describe("promptAsyncAfterSessionIdle", () => { + afterEach(() => { + // then + releaseAllPromptAsyncReservationsForTesting() + }) + + test("#given two internal promptAsync calls race for one idle session #when they dispatch concurrently #then only one prompt is accepted", async () => { + // given + let promptCalls = 0 + let releasePrompt: (() => void) | undefined + const promptGate = new Promise((resolve) => { + releasePrompt = resolve + }) + const client = { + session: { + status: async () => ({ data: { ses_race: { type: "idle" } } }), + promptAsync: async () => { + promptCalls += 1 + await promptGate + }, + }, + } + + // when + const first = promptAsyncAfterSessionIdle({ + client, + sessionID: "ses_race", + input: { path: { id: "ses_race" }, body: { parts: [] } }, + source: "test:first", + settleMs: 0, + postDispatchHoldMs: 0, + }) + await Promise.resolve() + const second = await promptAsyncAfterSessionIdle({ + client, + sessionID: "ses_race", + input: { path: { id: "ses_race" }, body: { parts: [] } }, + source: "test:second", + settleMs: 0, + postDispatchHoldMs: 0, + }) + releasePrompt?.() + const firstResult = await first + + // then + expect(firstResult.status).toBe("dispatched") + expect(second.status).toBe("reserved") + expect(promptCalls).toBe(1) + }) + + test("#given session.status reports busy #when an internal promptAsync is requested #then no prompt is sent", async () => { + // given + let promptCalls = 0 + const client = { + session: { + status: async () => ({ data: { ses_busy: { type: "busy" } } }), + promptAsync: async () => { + promptCalls += 1 + }, + }, + } + + // when + const result = await promptAsyncAfterSessionIdle({ + client, + sessionID: "ses_busy", + input: { path: { id: "ses_busy" }, body: { parts: [] } }, + source: "test:busy", + settleMs: 0, + postDispatchHoldMs: 0, + }) + + // then + expect(result.status).toBe("active") + expect(promptCalls).toBe(0) + }) +}) diff --git a/src/hooks/shared/prompt-async-gate.ts b/src/hooks/shared/prompt-async-gate.ts new file mode 100644 index 000000000..f037d18b4 --- /dev/null +++ b/src/hooks/shared/prompt-async-gate.ts @@ -0,0 +1,111 @@ +import { log } from "../../shared/logger" +import { + DEFAULT_SESSION_IDLE_SETTLE_MS, + isSessionActive, + settleAfterSessionIdle, +} from "./session-idle-settle" + +export const DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS = 250 + +type PromptAsyncInput = { + path?: { id?: string } + body?: unknown + query?: unknown + signal?: unknown + [key: string]: unknown +} + +type PromptAsyncClient = { + session?: { + status?: () => Promise + promptAsync?: (input: TInput) => Promise + } +} + +type PromptAsyncReservation = { + source: string + reservedAt: number + token: symbol +} + +export type PromptAsyncGateResult = + | { status: "dispatched"; response: unknown } + | { status: "active" } + | { status: "reserved"; reservedBy: string } + | { status: "unavailable" } + | { status: "failed"; error: unknown } + +const promptAsyncReservations = new Map() + +export async function promptAsyncAfterSessionIdle(args: { + client: PromptAsyncClient + sessionID: string + input: TInput + source: string + settleMs?: number + postDispatchHoldMs?: number +}): Promise { + const { + client, + sessionID, + input, + source, + settleMs = DEFAULT_SESSION_IDLE_SETTLE_MS, + } = args + const postDispatchHoldMs = args.postDispatchHoldMs ?? ( + settleMs > 0 ? DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS : 0 + ) + + if (typeof client.session?.promptAsync !== "function") { + log("[prompt-async-gate] promptAsync unavailable", { sessionID, source }) + return { status: "unavailable" } + } + + const existing = promptAsyncReservations.get(sessionID) + if (existing) { + log("[prompt-async-gate] promptAsync skipped because session is reserved", { + sessionID, + source, + reservedBy: existing.source, + reservedAgeMs: Date.now() - existing.reservedAt, + }) + return { status: "reserved", reservedBy: existing.source } + } + + const reservation: PromptAsyncReservation = { + source, + reservedAt: Date.now(), + token: Symbol(source), + } + promptAsyncReservations.set(sessionID, reservation) + + try { + const canReadStatus = typeof client.session?.status === "function" + await settleAfterSessionIdle(settleMs) + + if (canReadStatus && await isSessionActive(client, sessionID)) { + log("[prompt-async-gate] promptAsync skipped because session is active", { sessionID, source }) + return { status: "active" } + } + + log("[prompt-async-gate] promptAsync dispatching", { sessionID, source }) + const response = await client.session.promptAsync(input) + if (canReadStatus) { + await settleAfterSessionIdle(postDispatchHoldMs) + } + log("[prompt-async-gate] promptAsync dispatched", { sessionID, source }) + return { status: "dispatched", response } + } catch (error) { + log("[prompt-async-gate] promptAsync failed", { sessionID, source, error: String(error) }) + return { status: "failed", error } + } finally { + const current = promptAsyncReservations.get(sessionID) + if (current?.token === reservation.token) { + promptAsyncReservations.delete(sessionID) + } + } +} + +export function releaseAllPromptAsyncReservationsForTesting(): void { + promptAsyncReservations.clear() +} From 174cbd0fbd9da0716ea4bc9df84d640069a6f7e8 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:29:48 +0900 Subject: [PATCH 02/17] fix(background-agent): gate parent wake prompts --- src/features/background-agent/manager.ts | 102 +++++++++++++----- .../task-completion-cleanup.test.ts | 24 +++-- 2 files changed, 92 insertions(+), 34 deletions(-) diff --git a/src/features/background-agent/manager.ts b/src/features/background-agent/manager.ts index 2a58b67ce..fb62dd931 100644 --- a/src/features/background-agent/manager.ts +++ b/src/features/background-agent/manager.ts @@ -17,7 +17,6 @@ import { resolveInheritedPromptTools, createInternalAgentTextPart, messagesInDirectory, - promptAsyncInDirectory, promptWithRetryInDirectory, } from "../../shared" import { resolveMessageEventSessionID, resolveSessionEventID } from "../../shared/event-session-id" @@ -67,6 +66,7 @@ import { isSessionActive as isOpenCodeSessionActive, settleAfterSessionIdle, } from "../../hooks/shared/session-idle-settle" +import { promptAsyncAfterSessionIdle } from "../../hooks/shared/prompt-async-gate" import { findNearestMessageExcludingCompaction, resolvePromptContextFromSessionMessages, @@ -1161,27 +1161,46 @@ The fallback retry session is now created and can be inspected directly. applySessionPromptParams(existingTask.sessionId!, existingTask.model) } - promptAsyncInDirectory(this.client, { - path: { id: existingTask.sessionId }, - body: { - agent: existingTask.agent, - ...(resumeModel ? { model: resumeModel } : {}), - ...(resumeVariant ? { variant: resumeVariant } : {}), - tools: (() => { - const tools = { - task: false, - call_omo_agent: true, - question: false, - ...getAgentToolRestrictions(existingTask.agent, { - includeTeamToolDenylist: existingTask.teamRunId === undefined, - }), - } - setSessionTools(existingTask.sessionId!, tools) - return tools - })(), - parts: [createInternalAgentTextPart(input.prompt)], + promptAsyncAfterSessionIdle({ + client: this.client, + sessionID: existingTask.sessionId, + source: "background-agent-resume", + settleMs: 0, + postDispatchHoldMs: 0, + input: { + path: { id: existingTask.sessionId }, + body: { + agent: existingTask.agent, + ...(resumeModel ? { model: resumeModel } : {}), + ...(resumeVariant ? { variant: resumeVariant } : {}), + tools: (() => { + const tools = { + task: false, + call_omo_agent: true, + question: false, + ...getAgentToolRestrictions(existingTask.agent, { + includeTeamToolDenylist: existingTask.teamRunId === undefined, + }), + } + setSessionTools(existingTask.sessionId!, tools) + return tools + })(), + parts: [createInternalAgentTextPart(input.prompt)], + }, + query: { directory: this.directory }, }, - }, this.directory).catch(async (error) => { + }).then((promptResult) => { + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + log("[background-agent] resume prompt skipped by promptAsync gate:", { + taskId: existingTask.id, + sessionID: existingTask.sessionId, + status: promptResult.status, + }) + } + }).catch(async (error) => { log("[background-agent] resume prompt error:", error) const errorInfo = { name: extractErrorName(error), @@ -2328,14 +2347,41 @@ The task was re-queued on a fallback model after a retryable failure. const notificationContent = latestWake.notifications.join("\n\n") try { - await promptAsyncInDirectory(this.client, { - path: { id: sessionID }, - body: { - noReply: !latestWake.shouldReply, - ...latestWake.promptContext, - parts: [createInternalAgentTextPart(notificationContent)], + const promptResult = await promptAsyncAfterSessionIdle({ + client: this.client, + sessionID, + source: "background-agent-parent-wake", + settleMs: 0, + postDispatchHoldMs: 250, + input: { + path: { id: sessionID }, + body: { + noReply: !latestWake.shouldReply, + ...latestWake.promptContext, + parts: [createInternalAgentTextPart(notificationContent)], + }, + query: { directory: this.directory }, }, - }, this.directory) + }) + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + const pendingWake = this.pendingParentWakes.get(sessionID) + if (pendingWake) { + pendingWake.notifications.unshift(...latestWake.notifications) + pendingWake.shouldReply = pendingWake.shouldReply || latestWake.shouldReply + pendingWake.promptContext = latestWake.promptContext + } else { + this.pendingParentWakes.set(sessionID, latestWake) + } + this.schedulePendingParentWakeFlush(sessionID) + log("[background-agent] Deferred parent wake skipped by promptAsync gate:", { + sessionID, + status: promptResult.status, + }) + return + } log("[background-agent] Sent deferred parent wake:", { sessionID }) } catch (error) { this.queuePendingNotification(sessionID, notificationContent) diff --git a/src/features/background-agent/task-completion-cleanup.test.ts b/src/features/background-agent/task-completion-cleanup.test.ts index 43a75fa20..943dabc41 100644 --- a/src/features/background-agent/task-completion-cleanup.test.ts +++ b/src/features/background-agent/task-completion-cleanup.test.ts @@ -4,6 +4,7 @@ import type { PluginInput } from "@opencode-ai/plugin" import { TASK_CLEANUP_DELAY_MS } from "./constants" import { BackgroundManager } from "./manager" import type { BackgroundTask } from "./types" +import { releaseAllPromptAsyncReservationsForTesting } from "../../hooks/shared/prompt-async-gate" import { OMO_INTERNAL_INITIATOR_MARKER } from "../../shared/internal-initiator-marker" type PromptAsyncCall = { @@ -29,6 +30,7 @@ let fakeTimers: FakeTimers | undefined afterEach(() => { managerUnderTest?.shutdown() fakeTimers?.restore() + releaseAllPromptAsyncReservationsForTesting() managerUnderTest = undefined fakeTimers = undefined }) @@ -163,8 +165,18 @@ async function notifyParentSessionForTest(manager: BackgroundManager, task: Back return notifyParentSession.call(manager, task) } -function waitForDeferredWake(): Promise { - return new Promise((resolve) => setTimeout(resolve, 180)) +async function waitUntil(predicate: () => boolean, timeoutMs: number): Promise { + const startedAt = Date.now() + while (!predicate()) { + if (Date.now() - startedAt >= timeoutMs) { + return + } + await new Promise((resolve) => setTimeout(resolve, 10)) + } +} + +function waitForDeferredWake(promptAsyncCalls: PromptAsyncCall[]): Promise { + return waitUntil(() => promptAsyncCalls.length > 0, 600) } function waitForDeferredWakeRetry(): Promise { @@ -341,7 +353,7 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => { // when sessionStatuses["parent-1"] = { type: "idle" } manager.handleEvent({ type: "session.idle", properties: { sessionID: "parent-1" } }) - await waitForDeferredWake() + await waitForDeferredWake(promptAsyncCalls) // then expect(promptAsyncCalls).toHaveLength(1) @@ -377,7 +389,7 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => { // when sessionStatuses["parent-1"] = { type: "idle" } manager.handleEvent({ type: "session.idle", properties: { sessionID: "parent-1" } }) - await waitForDeferredWake() + await waitForDeferredWake(promptAsyncCalls) // then expect(promptAsyncCalls).toHaveLength(1) @@ -424,7 +436,7 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => { // when sessionStatuses["parent-1"] = { type: "idle" } manager.handleEvent({ type: "session.idle", properties: { sessionID: "parent-1" } }) - await waitForDeferredWake() + await waitForDeferredWake(promptAsyncCalls) // then expect(promptAsyncCalls).toHaveLength(1) @@ -477,7 +489,7 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => { await notifyParentSessionForTest(manager, task) sessionStatuses["parent-1"] = { type: "idle" } manager.handleEvent({ type: "session.idle", properties: { sessionID: "parent-1" } }) - await waitForDeferredWake() + await waitForDeferredWake(promptAsyncCalls) // then expect(promptAsyncCalls).toHaveLength(1) From f1a62a9cd1730acef712ac80656fa50f112e8121 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:29:52 +0900 Subject: [PATCH 03/17] fix(team-mode): gate member wake prompts --- .../team-mode/tools/messaging.test.ts | 28 +++++++++++++++++ src/features/team-mode/tools/messaging.ts | 30 ++++++++++++++++--- .../team-idle-wake-hint.ts | 26 +++++++++------- 3 files changed, 70 insertions(+), 14 deletions(-) diff --git a/src/features/team-mode/tools/messaging.test.ts b/src/features/team-mode/tools/messaging.test.ts index 0faf67771..2c069d5bd 100644 --- a/src/features/team-mode/tools/messaging.test.ts +++ b/src/features/team-mode/tools/messaging.test.ts @@ -282,6 +282,34 @@ describe("createTeamSendMessageTool", () => { expect(calls[0]?.directory).toBe(resolveBaseDir(fixture.config)) }) + test("#given recipient OpenCode session is busy #when team_send_message attempts live delivery #then it leaves the message unread without starting another reply", async () => { + // given + const fixture = await createTeamFixture() + let promptCalls = 0 + const client = { + session: { + status: async () => ({ data: { [fixture.memberTwoSessionId]: { type: "busy" } } }), + promptAsync: async () => { + promptCalls += 1 + }, + }, + } + const liveTool = createTeamSendMessageTool(fixture.config, client) + + // when + await liveTool.execute({ + teamRunId: fixture.teamRunId, + to: "m2", + body: "ping while busy", + }, fixture.toolContext(fixture.memberOneSessionId)) + + // then + expect(promptCalls).toBe(0) + const unread = await listUnreadMessages(fixture.teamRunId, "m2", fixture.config) + expect(unread).toHaveLength(1) + expect(unread[0]?.body).toBe("ping while busy") + }) + test("live delivery pins the recipient's resolved subagent_type and model on promptAsync", async () => { // given const fixture = await createTeamFixture() diff --git a/src/features/team-mode/tools/messaging.ts b/src/features/team-mode/tools/messaging.ts index d04105974..d3bcb5c5a 100644 --- a/src/features/team-mode/tools/messaging.ts +++ b/src/features/team-mode/tools/messaging.ts @@ -15,6 +15,7 @@ import { reserveMessageForDelivery, } from "../team-mailbox/reservation" import { BroadcastNotPermittedError, sendMessage } from "../team-mailbox/send" +import { promptAsyncAfterSessionIdle } from "../../../hooks/shared/prompt-async-gate" import type { Message } from "../types" import { MessageSchema } from "../types" @@ -33,6 +34,7 @@ export type LiveDeliveryClient = { } query?: { directory: string } }): Promise + status?: () => Promise } } @@ -180,11 +182,31 @@ async function deliverLive( applyMemberSessionRouting(recipientSessionId, recipientMember) try { - await client.session.promptAsync({ - path: { id: recipientSessionId }, - body: buildMemberPromptBody(recipientMember, envelope), - query: { directory: recipientMember.worktreePath ?? directory }, + const promptResult = await promptAsyncAfterSessionIdle({ + client, + sessionID: recipientSessionId, + source: "team-live-delivery", + input: { + path: { id: recipientSessionId }, + body: buildMemberPromptBody(recipientMember, envelope), + query: { directory: recipientMember.worktreePath ?? directory }, + }, }) + if (promptResult.status !== "dispatched") { + log("[team-mailbox] live delivery skipped by promptAsync gate, falling back to inbox injection", { + status: promptResult.status, + teamRunId, + recipient: recipientName, + recipientSessionId, + messageId: message.messageId, + }) + await releaseReservationSafely(reservation, { + teamRunId, + recipient: recipientName, + messageId: message.messageId, + }) + continue + } await commitDeliveryReservation(reservation) log("[team-mailbox] live delivery committed", { teamRunId, diff --git a/src/hooks/team-session-events/team-idle-wake-hint.ts b/src/hooks/team-session-events/team-idle-wake-hint.ts index e07898911..347da7f73 100644 --- a/src/hooks/team-session-events/team-idle-wake-hint.ts +++ b/src/hooks/team-session-events/team-idle-wake-hint.ts @@ -9,7 +9,7 @@ import { } from "../../features/team-mode/member-session-routing" import { resolveSessionEventID } from "../../shared/event-session-id" import { log } from "../../shared/logger" -import { shouldPromptAfterSessionIdle } from "../shared/session-idle-settle" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" type PromptAsyncInput = { path: { id: string } @@ -100,23 +100,29 @@ export function createTeamIdleWakeHint(ctx: TeamIdleWakeHintContext, config: Tea } applyMemberSessionRouting(sessionID, memberEntry) - if (!(await shouldPromptAfterSessionIdle(ctx.client, sessionID, options?.idleSettleMs))) { - log("team idle wake hint skipped because session is active", { - event: "team-mode-idle-wake-hint-active-session", + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID, + source: "team-idle-wake-hint", + settleMs: options?.idleSettleMs, + input: { + path: { id: sessionID }, + body: buildMemberPromptBody(memberEntry, buildWakeHint(unreadMessages.length)), + query: { directory: ctx.directory }, + }, + }) + if (promptResult.status !== "dispatched") { + log("team idle wake hint skipped by promptAsync gate", { + event: "team-mode-idle-wake-hint-gated", teamRunId: runtimeState.teamRunId, memberName: memberEntry.name, sessionID, unreadCount: unreadMessages.length, + status: promptResult.status, }) return } - await ctx.client.session.promptAsync({ - path: { id: sessionID }, - body: buildMemberPromptBody(memberEntry, buildWakeHint(unreadMessages.length)), - query: { directory: ctx.directory }, - }) - log("team idle wake hint sent", { event: "team-mode-idle-wake-hint", teamRunId: runtimeState.teamRunId, From c75f548863eaf605863a077b9d12590bf410182a Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:29:55 +0900 Subject: [PATCH 04/17] fix(fallback): gate model retry prompts --- src/plugin/event.ts | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/src/plugin/event.ts b/src/plugin/event.ts index 976c732ab..0de6738a2 100644 --- a/src/plugin/event.ts +++ b/src/plugin/event.ts @@ -42,6 +42,7 @@ import { createTeamIdleWakeHint } from "../hooks/team-session-events/team-idle-w import { createTeamLeadOrphanHandler } from "../hooks/team-session-events/team-lead-orphan-handler"; import { createTeamMemberErrorHandler } from "../hooks/team-session-events/team-member-error-handler"; import { createTeamMemberStatusHandler } from "../hooks/team-session-events/team-member-status-handler"; +import { promptAsyncAfterSessionIdle } from "../hooks/shared/prompt-async-gate"; import type { CreatedHooks } from "../create-hooks"; import type { Managers } from "../create-managers"; @@ -348,6 +349,7 @@ export function createEventHandler(args: { client: { session: { promptAsync: pluginContext.client.session.promptAsync, + status: pluginContext.client.session.status, }, }, }, teamModeConfig) @@ -495,11 +497,20 @@ export function createEventHandler(args: { }; if (typeof pluginContext.client.session.promptAsync === "function") { - await pluginContext.client.session.promptAsync(promptBody).then(() => { - dispatched = true; - }).catch((error) => { - log("[event] model-fallback promptAsync failed", { sessionID, source, error }); + const promptResult = await promptAsyncAfterSessionIdle({ + client: pluginContext.client, + sessionID, + source: `model-fallback:${source}`, + input: promptBody, }); + if (promptResult.status === "dispatched") { + dispatched = true; + } else if (promptResult.status === "failed") { + const error = promptResult.error; + log("[event] model-fallback promptAsync failed", { sessionID, source, error }); + } else { + log("[event] model-fallback promptAsync skipped by gate", { sessionID, source, status: promptResult.status }); + } return; } From db28a32cff57a538b5e2c6137ea4886bc1815eb4 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:30:03 +0900 Subject: [PATCH 05/17] fix(recovery): gate compaction prompts --- .../aggressive-truncation-strategy.ts | 39 +++++++++++-------- .../compaction-context-injector/recovery.ts | 32 ++++++++++----- .../compaction-context-injector/types.ts | 1 + 3 files changed, 46 insertions(+), 26 deletions(-) diff --git a/src/hooks/anthropic-context-window-limit-recovery/aggressive-truncation-strategy.ts b/src/hooks/anthropic-context-window-limit-recovery/aggressive-truncation-strategy.ts index 21f761ab3..5fdcbe87c 100644 --- a/src/hooks/anthropic-context-window-limit-recovery/aggressive-truncation-strategy.ts +++ b/src/hooks/anthropic-context-window-limit-recovery/aggressive-truncation-strategy.ts @@ -17,7 +17,7 @@ import { findNearestMessageWithFields, findNearestMessageWithFieldsFromSDK, } from "../../features/hook-message-injector" -import { isSessionActive } from "../shared/session-idle-settle" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" export async function runAggressiveTruncationStrategy(params: { sessionID: string @@ -74,13 +74,6 @@ export async function runAggressiveTruncationStrategy(params: { clearSessionState(params.autoCompactState, params.sessionID) setTimeout(async () => { try { - if (await isSessionActive(params.client, params.sessionID)) { - log("[auto-compact] skipped delayed auto prompt because session became active", { - sessionID: params.sessionID, - }) - return - } - const sdkMessage = await findNearestMessageWithFieldsFromSDK(params.client, params.sessionID) const previousMessage = sdkMessage ?? (() => { const messageDir = getMessageDir(params.sessionID) @@ -95,17 +88,29 @@ export async function runAggressiveTruncationStrategy(params: { const launchVariant = previousMessage?.model?.variant const inheritedTools = resolveInheritedPromptTools(params.sessionID, previousMessage?.tools) - await params.client.session.promptAsync({ - path: { id: params.sessionID }, - body: { - auto: true, - ...(launchAgent ? { agent: launchAgent } : {}), - ...(launchModel ? { model: launchModel } : {}), - ...(launchVariant ? { variant: launchVariant } : {}), - ...(inheritedTools ? { tools: inheritedTools } : {}), + const promptResult = await promptAsyncAfterSessionIdle({ + client: params.client, + sessionID: params.sessionID, + source: "auto-compact", + settleMs: 0, + input: { + path: { id: params.sessionID }, + body: { + auto: true, + ...(launchAgent ? { agent: launchAgent } : {}), + ...(launchModel ? { model: launchModel } : {}), + ...(launchVariant ? { variant: launchVariant } : {}), + ...(inheritedTools ? { tools: inheritedTools } : {}), + } as never, + query: { directory: params.directory }, } as never, - query: { directory: params.directory }, }) + if (promptResult.status !== "dispatched") { + log("[auto-compact] delayed auto prompt skipped by promptAsync gate", { + sessionID: params.sessionID, + status: promptResult.status, + }) + } } catch (error) { log("[auto-compact] delayed auto prompt failed", { sessionID: params.sessionID, diff --git a/src/hooks/compaction-context-injector/recovery.ts b/src/hooks/compaction-context-injector/recovery.ts index ed69bf0e0..7abc8e71c 100644 --- a/src/hooks/compaction-context-injector/recovery.ts +++ b/src/hooks/compaction-context-injector/recovery.ts @@ -21,6 +21,7 @@ import { import { AGENT_RECOVERY_PROMPT, NO_TEXT_TAIL_THRESHOLD, RECOVERY_COOLDOWN_MS, RECENT_COMPACTION_WINDOW_MS } from "./constants" import type { CompactionContextClient } from "./types" import type { TailMonitorState } from "./tail-monitor" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" export function createRecoveryLogic( ctx: CompactionContextClient | undefined, @@ -81,17 +82,30 @@ export function createRecoveryLogic( } try { - await ctx.client.session.promptAsync({ - path: { id: sessionID }, - body: { - noReply: true, - agent: launchAgent ?? expectedPromptConfig.agent, - ...(model ? { model } : {}), - ...(tools ? { tools } : {}), - parts: [createInternalAgentContinuationTextPart(AGENT_RECOVERY_PROMPT)], + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID, + source: "compaction-context-injector", + input: { + path: { id: sessionID }, + body: { + noReply: true, + agent: launchAgent ?? expectedPromptConfig.agent, + ...(model ? { model } : {}), + ...(tools ? { tools } : {}), + parts: [createInternalAgentContinuationTextPart(AGENT_RECOVERY_PROMPT)], + }, + query: { directory: ctx.directory }, }, - query: { directory: ctx.directory }, }) + if (promptResult.status !== "dispatched") { + log(`[compaction-context-injector] Recovery skipped by promptAsync gate`, { + sessionID, + reason, + status: promptResult.status, + }) + return false + } const recoveredPromptConfig = await resolveLatestSessionPromptConfig(ctx, sessionID) if (!isPromptConfigRecovered(recoveredPromptConfig, expectedPromptConfig)) { diff --git a/src/hooks/compaction-context-injector/types.ts b/src/hooks/compaction-context-injector/types.ts index b560e21b4..1772550d2 100644 --- a/src/hooks/compaction-context-injector/types.ts +++ b/src/hooks/compaction-context-injector/types.ts @@ -20,6 +20,7 @@ export type CompactionContextClient = { } query?: { directory: string } }) => Promise + status?: () => Promise } } directory: string From b0a484b40357ca05e00679477c712ab508a9058d Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:30:07 +0900 Subject: [PATCH 06/17] fix(session-recovery): gate resume prompts --- .../recover-tool-result-missing.ts | 11 +++++++-- .../recover-unavailable-tool.ts | 18 ++++++++++++-- src/hooks/session-recovery/resume.ts | 24 ++++++++++++------- 3 files changed, 40 insertions(+), 13 deletions(-) diff --git a/src/hooks/session-recovery/recover-tool-result-missing.ts b/src/hooks/session-recovery/recover-tool-result-missing.ts index 6a1a8e6b9..60d5f7aec 100644 --- a/src/hooks/session-recovery/recover-tool-result-missing.ts +++ b/src/hooks/session-recovery/recover-tool-result-missing.ts @@ -3,11 +3,13 @@ import type { MessageData, ResumeConfig } from "./types" import { readParts } from "./storage" import { isSqliteBackend } from "../../shared/opencode-storage-detection" import { normalizeSDKResponse } from "../../shared" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" type Client = ReturnType type ClientWithPromptAsync = { session: { promptAsync: (opts: { path: { id: string }; body: Record }) => Promise + status?: () => Promise } } @@ -119,9 +121,14 @@ export async function recoverToolResultMissing( return false } - await client.session.promptAsync(promptInput) + const promptResult = await promptAsyncAfterSessionIdle({ + client, + sessionID, + source: "session-recovery-tool-result-missing", + input: promptInput, + }) - return true + return promptResult.status === "dispatched" } catch { return false } diff --git a/src/hooks/session-recovery/recover-unavailable-tool.ts b/src/hooks/session-recovery/recover-unavailable-tool.ts index 3aa937e73..b45a4f7ff 100644 --- a/src/hooks/session-recovery/recover-unavailable-tool.ts +++ b/src/hooks/session-recovery/recover-unavailable-tool.ts @@ -4,6 +4,7 @@ import { readParts } from "./storage" import type { MessageData } from "./types" import { normalizeSDKResponse } from "../../shared" import { isSqliteBackend } from "../../shared/opencode-storage-detection" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" type Client = ReturnType @@ -100,8 +101,21 @@ export async function recoverUnavailableTool( body: { parts: toolResultParts }, } const promptAsync = client.session.promptAsync as (...args: never[]) => unknown - await Reflect.apply(promptAsync, client.session, [promptInput]) - return true + const promptClient = { + session: { + status: client.session.status, + promptAsync: (input: PromptWithToolResultInput) => ( + Reflect.apply(promptAsync, client.session, [input]) as Promise + ), + }, + } + const promptResult = await promptAsyncAfterSessionIdle({ + client: promptClient, + sessionID, + source: "session-recovery-unavailable-tool", + input: promptInput, + }) + return promptResult.status === "dispatched" } catch { return false } diff --git a/src/hooks/session-recovery/resume.ts b/src/hooks/session-recovery/resume.ts index 136ea16b4..6049fdbcd 100644 --- a/src/hooks/session-recovery/resume.ts +++ b/src/hooks/session-recovery/resume.ts @@ -1,6 +1,7 @@ import type { createOpencodeClient } from "@opencode-ai/sdk" import type { MessageData, ResumeConfig } from "./types" import { createInternalAgentContinuationTextPart, resolveInheritedPromptTools } from "../../shared" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" const RECOVERY_RESUME_TEXT = "[session recovered - continuing previous task]" @@ -32,17 +33,22 @@ export async function resumeSession(client: Client, config: ResumeConfig): Promi : undefined const launchVariant = config.model?.variant - await client.session.promptAsync({ - path: { id: config.sessionID }, - body: { - parts: [createInternalAgentContinuationTextPart(RECOVERY_RESUME_TEXT)], - agent: config.agent, - ...(launchModel ? { model: launchModel } : {}), - ...(launchVariant ? { variant: launchVariant } : {}), - ...(inheritedTools ? { tools: inheritedTools } : {}), + const promptResult = await promptAsyncAfterSessionIdle({ + client, + sessionID: config.sessionID, + source: "session-recovery", + input: { + path: { id: config.sessionID }, + body: { + parts: [createInternalAgentContinuationTextPart(RECOVERY_RESUME_TEXT)], + agent: config.agent, + ...(launchModel ? { model: launchModel } : {}), + ...(launchVariant ? { variant: launchVariant } : {}), + ...(inheritedTools ? { tools: inheritedTools } : {}), + }, }, }) - return true + return promptResult.status === "dispatched" } catch { return false } From 960baf39bb8940f22e97105931d3f91e317e4cb0 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:30:11 +0900 Subject: [PATCH 07/17] fix(atlas): gate boulder continuation prompts --- .../atlas/boulder-continuation-injector.ts | 27 ++++++++++++----- src/hooks/atlas/idle-event.ts | 29 +++++++++++++++---- 2 files changed, 43 insertions(+), 13 deletions(-) diff --git a/src/hooks/atlas/boulder-continuation-injector.ts b/src/hooks/atlas/boulder-continuation-injector.ts index 197c3a7d8..1931da6c9 100644 --- a/src/hooks/atlas/boulder-continuation-injector.ts +++ b/src/hooks/atlas/boulder-continuation-injector.ts @@ -5,7 +5,7 @@ import { } from "../../features/claude-code-session-state" import { log } from "../../shared/logger" import { createInternalAgentContinuationTextPart, resolveInheritedPromptTools } from "../../shared" -import { isSessionActive } from "../shared/session-idle-settle" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" import { HOOK_NAME } from "./hook-name" import { BOULDER_CONTINUATION_PROMPT } from "./system-reminder-templates" import { resolveRecentPromptContextForSession } from "./recent-model-resolver" @@ -32,6 +32,7 @@ export async function injectBoulderContinuation(input: { preferredTaskTitle?: string backgroundManager?: BackgroundTaskStatusProvider sessionState: SessionState + idleSettleMs?: number }): Promise { const { ctx, @@ -45,6 +46,7 @@ export async function injectBoulderContinuation(input: { preferredTaskTitle, backgroundManager, sessionState, + idleSettleMs, } = input const hasRunningBgTasks = backgroundManager @@ -78,11 +80,6 @@ export async function injectBoulderContinuation(input: { } try { - if (await isSessionActive(ctx.client, sessionID)) { - log(`[${HOOK_NAME}] Skipped injection: session is active`, { sessionID }) - return "skipped_active_session" - } - log(`[${HOOK_NAME}] Injecting boulder continuation`, { sessionID, planName, remaining }) const promptContext = await resolveRecentPromptContextForSession(ctx, sessionID) @@ -93,7 +90,12 @@ export async function injectBoulderContinuation(input: { : undefined const launchVariant = promptContext.model?.variant - await ctx.client.session.promptAsync({ + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID, + source: HOOK_NAME, + settleMs: idleSettleMs, + input: { path: { id: sessionID }, body: { agent: continuationAgent, @@ -103,7 +105,18 @@ export async function injectBoulderContinuation(input: { parts: [createInternalAgentContinuationTextPart(prompt)], }, query: { directory: ctx.directory }, + }, }) + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + log(`[${HOOK_NAME}] Boulder continuation skipped by promptAsync gate`, { + sessionID, + status: promptResult.status, + }) + return "skipped_active_session" + } sessionState.promptFailureCount = 0 log(`[${HOOK_NAME}] Boulder continuation injected`, { sessionID }) diff --git a/src/hooks/atlas/idle-event.ts b/src/hooks/atlas/idle-event.ts index 5f98122c8..4dce6ab4b 100644 --- a/src/hooks/atlas/idle-event.ts +++ b/src/hooks/atlas/idle-event.ts @@ -20,6 +20,7 @@ import { createInternalAgentContinuationTextPart } from "../../shared" import { getAgentConfigKey } from "../../shared/agent-display-names" import { log } from "../../shared/logger" import { shouldPromptAfterSessionIdle } from "../shared/session-idle-settle" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" import { injectBoulderContinuation } from "./boulder-continuation-injector" import { HOOK_NAME } from "./hook-name" import { resolveActiveBoulderSession } from "./resolve-active-boulder-session" @@ -52,6 +53,7 @@ async function injectContinuation(input: { progress: { total: number; completed: number } agent?: string worktreePath?: string + idleSettleMs?: number }): Promise { const remaining = input.progress.total - input.progress.completed if (input.sessionState.isInjectingContinuation) { @@ -110,6 +112,7 @@ async function injectContinuation(input: { preferredTaskTitle: preferredTaskSession?.task_title, backgroundManager: input.options?.backgroundManager, sessionState: input.sessionState, + idleSettleMs: input.idleSettleMs, }) if (result === "injected") { @@ -288,14 +291,27 @@ export async function handleAtlasSessionIdle(input: { return } - await ctx.client.session.promptAsync({ - path: { id: sessionID }, - body: { - agent: atlasAgent, - parts: [createInternalAgentContinuationTextPart(prompt)], + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID, + source: HOOK_NAME, + settleMs: options?.idleSettleMs, + input: { + path: { id: sessionID }, + body: { + agent: atlasAgent, + parts: [createInternalAgentContinuationTextPart(prompt)], + }, + query: { directory: ctx.directory }, }, - query: { directory: ctx.directory }, }) + if (promptResult.status !== "dispatched") { + log(`[${HOOK_NAME}] Boulder completion nudge skipped by promptAsync gate`, { + sessionID, + status: promptResult.status, + }) + return + } sessionState.boulderCompletionNudgedAt = { ...(sessionState.boulderCompletionNudgedAt ?? {}), [work.work_id]: Date.now(), @@ -398,6 +414,7 @@ export async function handleAtlasSessionIdle(input: { progress, agent: boulderState.agent, worktreePath: boulderState.worktree_path, + idleSettleMs: options?.idleSettleMs ?? 0, }) } From b0b61182b41b3a8f7decd42134a05ab7ab097d58 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:30:15 +0900 Subject: [PATCH 08/17] fix(ralph-loop): gate continuation prompts --- .../continuation-prompt-injector.ts | 36 ++++++++++++++----- .../ralph-loop/iteration-continuation.ts | 3 ++ .../ralph-loop/ralph-loop-event-handler.ts | 2 ++ 3 files changed, 32 insertions(+), 9 deletions(-) diff --git a/src/hooks/ralph-loop/continuation-prompt-injector.ts b/src/hooks/ralph-loop/continuation-prompt-injector.ts index dec80c5c6..8c75dc99b 100644 --- a/src/hooks/ralph-loop/continuation-prompt-injector.ts +++ b/src/hooks/ralph-loop/continuation-prompt-injector.ts @@ -10,6 +10,7 @@ import { resolveInheritedPromptTools, } from "../../shared" import { normalizeAgentForPromptKey } from "../../shared/agent-display-names" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" type MessageInfo = { agent?: string @@ -66,6 +67,7 @@ export async function injectContinuationPrompt( directory: string apiTimeoutMs: number inheritFromSessionID?: string + idleSettleMs?: number }, ): Promise { let agent: string | undefined @@ -119,17 +121,33 @@ export async function injectContinuationPrompt( let response: unknown try { - response = await ctx.client.session.promptAsync({ - path: { id: options.sessionID }, - body: { - ...(cleanAgent !== undefined ? { agent: cleanAgent } : {}), - ...(launchModel ? { model: launchModel } : {}), - ...(launchVariant ? { variant: launchVariant } : {}), - ...(inheritedTools ? { tools: inheritedTools } : {}), - parts: [createInternalAgentContinuationTextPart(options.prompt)], + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID: options.sessionID, + source: "ralph-loop", + settleMs: options.idleSettleMs, + input: { + path: { id: options.sessionID }, + body: { + ...(cleanAgent !== undefined ? { agent: cleanAgent } : {}), + ...(launchModel ? { model: launchModel } : {}), + ...(launchVariant ? { variant: launchVariant } : {}), + ...(inheritedTools ? { tools: inheritedTools } : {}), + parts: [createInternalAgentContinuationTextPart(options.prompt)], + }, + query: { directory: options.directory }, }, - query: { directory: options.directory }, }) + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + return { + status: "rejected", + error: createPromptAsyncError(`promptAsync skipped: ${promptResult.status}`, promptResult), + } + } + response = promptResult.response } catch (error) { const promptError = error instanceof Error ? error diff --git a/src/hooks/ralph-loop/iteration-continuation.ts b/src/hooks/ralph-loop/iteration-continuation.ts index fd3d2741d..aeca27a16 100644 --- a/src/hooks/ralph-loop/iteration-continuation.ts +++ b/src/hooks/ralph-loop/iteration-continuation.ts @@ -9,6 +9,7 @@ import { createIterationSession, selectSessionInTui } from "./session-reset-stra type ContinuationOptions = { directory: string apiTimeoutMs: number + idleSettleMs: number previousSessionID: string loopState: { setSessionID: (sessionID: string) => RalphLoopState | null @@ -45,6 +46,7 @@ export async function continueIteration( prompt: continuationPrompt, directory: options.directory, apiTimeoutMs: options.apiTimeoutMs, + idleSettleMs: options.idleSettleMs, }) if (promptResult.status === "rejected") { return { status: "dispatch_rejected", error: promptResult.error } @@ -73,6 +75,7 @@ export async function continueIteration( prompt: continuationPrompt, directory: options.directory, apiTimeoutMs: options.apiTimeoutMs, + idleSettleMs: options.idleSettleMs, }) if (promptResult.status === "rejected") { return { status: "dispatch_rejected", error: promptResult.error } diff --git a/src/hooks/ralph-loop/ralph-loop-event-handler.ts b/src/hooks/ralph-loop/ralph-loop-event-handler.ts index 0e832a9ef..02c4c3d83 100644 --- a/src/hooks/ralph-loop/ralph-loop-event-handler.ts +++ b/src/hooks/ralph-loop/ralph-loop-event-handler.ts @@ -358,6 +358,7 @@ export function createRalphLoopEventHandler( previousSessionID: sessionID, directory: options.directory, apiTimeoutMs: options.apiTimeoutMs, + idleSettleMs: options.idleSettleMs, loopState: options.loopState, }) @@ -523,6 +524,7 @@ export function createRalphLoopEventHandler( previousSessionID: sessionID, directory: options.directory, apiTimeoutMs: options.apiTimeoutMs, + idleSettleMs: options.idleSettleMs, loopState: options.loopState, }) From a524754eca26c2d4ae22a69bca77b54e7bd11f3c Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:30:20 +0900 Subject: [PATCH 09/17] fix(todo-continuation): gate idle prompts --- .../continuation-injection.ts | 41 ++++++++++++------- 1 file changed, 26 insertions(+), 15 deletions(-) diff --git a/src/hooks/todo-continuation-enforcer/continuation-injection.ts b/src/hooks/todo-continuation-enforcer/continuation-injection.ts index da82b9622..cfb9214a6 100644 --- a/src/hooks/todo-continuation-enforcer/continuation-injection.ts +++ b/src/hooks/todo-continuation-enforcer/continuation-injection.ts @@ -21,7 +21,7 @@ import { getAgentConfigKey, normalizeAgentForPromptKey, } from "../../shared/agent-display-names" -import { isSessionActive } from "../shared/session-idle-settle" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" import { CONTINUATION_PROMPT, @@ -166,11 +166,6 @@ ${todoList}` return } - if (await isSessionActive(ctx.client, sessionID)) { - log(`[${HOOK_NAME}] Skipped injection: session is active before prompt`, { sessionID }) - return - } - if (injectionState) { injectionState.inFlight = true } @@ -190,17 +185,33 @@ ${todoList}` : undefined const launchVariant = model?.variant - await ctx.client.session.promptAsync({ - path: { id: sessionID }, - body: { - agent: launchAgent ?? promptAgent, - ...(launchModel ? { model: launchModel } : {}), - ...(launchVariant ? { variant: launchVariant } : {}), - ...(inheritedTools ? { tools: inheritedTools } : {}), - parts: [createInternalAgentContinuationTextPart(prompt)], + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID, + source: HOOK_NAME, + settleMs: 0, + input: { + path: { id: sessionID }, + body: { + agent: launchAgent ?? promptAgent, + ...(launchModel ? { model: launchModel } : {}), + ...(launchVariant ? { variant: launchVariant } : {}), + ...(inheritedTools ? { tools: inheritedTools } : {}), + parts: [createInternalAgentContinuationTextPart(prompt)], + }, + query: { directory: ctx.directory }, }, - query: { directory: ctx.directory }, }) + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + log(`[${HOOK_NAME}] Injection skipped by promptAsync gate`, { sessionID, status: promptResult.status }) + if (injectionState) { + injectionState.inFlight = false + } + return + } log(`[${HOOK_NAME}] Injection successful`, { sessionID }) if (injectionState) { From dd6271bbf400edb425ef554508de27808ff27b9a Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:30:24 +0900 Subject: [PATCH 10/17] fix(babysitter): gate reminder prompts --- .../unstable-agent-babysitter-hook.ts | 39 ++++++++++++------- src/plugin/unstable-agent-babysitter.ts | 22 ++++++++++- 2 files changed, 44 insertions(+), 17 deletions(-) diff --git a/src/hooks/unstable-agent-babysitter/unstable-agent-babysitter-hook.ts b/src/hooks/unstable-agent-babysitter/unstable-agent-babysitter-hook.ts index 729192965..30cb01cca 100644 --- a/src/hooks/unstable-agent-babysitter/unstable-agent-babysitter-hook.ts +++ b/src/hooks/unstable-agent-babysitter/unstable-agent-babysitter-hook.ts @@ -12,7 +12,7 @@ import { isUnstableTask, THINKING_SUMMARY_MAX_CHARS, } from "./task-message-analyzer" -import { shouldPromptAfterSessionIdle } from "../shared/session-idle-settle" +import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" const HOOK_NAME = "unstable-agent-babysitter" const DEFAULT_TIMEOUT_MS = 120000 @@ -216,22 +216,31 @@ export function createUnstableAgentBabysitterHook(ctx: BabysitterContext, option ? { providerID: model.providerID, modelID: model.modelID } : undefined const launchVariant = model?.variant - if (!(await shouldPromptAfterSessionIdle(ctx.client, mainSessionID, options.idleSettleMs))) { - log(`[${HOOK_NAME}] Reminder skipped because main session is active`, { taskId: task.id, sessionID: mainSessionID }) + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID: mainSessionID, + source: HOOK_NAME, + settleMs: options.idleSettleMs, + input: { + path: { id: mainSessionID }, + body: { + ...(agent ? { agent } : {}), + ...(launchModel ? { model: launchModel } : {}), + ...(launchVariant ? { variant: launchVariant } : {}), + ...(tools ? { tools } : {}), + parts: [createInternalAgentTextPart(reminder)], + }, + query: { directory: ctx.directory }, + }, + }) + if (promptResult.status !== "dispatched") { + log(`[${HOOK_NAME}] Reminder skipped by promptAsync gate`, { + taskId: task.id, + sessionID: mainSessionID, + status: promptResult.status, + }) continue } - - await ctx.client.session.promptAsync({ - path: { id: mainSessionID }, - body: { - ...(agent ? { agent } : {}), - ...(launchModel ? { model: launchModel } : {}), - ...(launchVariant ? { variant: launchVariant } : {}), - ...(tools ? { tools } : {}), - parts: [createInternalAgentTextPart(reminder)], - }, - query: { directory: ctx.directory }, - }) reminderCooldowns.set(task.id, now) log(`[${HOOK_NAME}] Reminder injected`, { taskId: task.id, sessionID: mainSessionID }) } catch (error) { diff --git a/src/plugin/unstable-agent-babysitter.ts b/src/plugin/unstable-agent-babysitter.ts index 6ab73bbd8..040c26d21 100644 --- a/src/plugin/unstable-agent-babysitter.ts +++ b/src/plugin/unstable-agent-babysitter.ts @@ -3,6 +3,7 @@ import type { PluginContext } from "./types" import { createUnstableAgentBabysitterHook } from "../hooks" import type { BackgroundManager } from "../features/background-agent" +import { promptAsyncAfterSessionIdle } from "../hooks/shared/prompt-async-gate" export function createUnstableAgentBabysitter(args: { ctx: PluginContext @@ -24,11 +25,28 @@ export function createUnstableAgentBabysitter(args: { } return [] }, + status: async () => ctx.client.session.status(), prompt: async (promptArgs) => { - await ctx.client.session.promptAsync(promptArgs) + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID: promptArgs.path.id, + source: "unstable-agent-babysitter", + input: promptArgs, + }) + if (promptResult.status === "failed") { + throw promptResult.error + } }, promptAsync: async (promptArgs) => { - await ctx.client.session.promptAsync(promptArgs) + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID: promptArgs.path.id, + source: "unstable-agent-babysitter", + input: promptArgs, + }) + if (promptResult.status === "failed") { + throw promptResult.error + } }, }, }, From 30adce9cadf70fefd5a85749869ea86d4d5bc39c Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:49:44 +0900 Subject: [PATCH 11/17] fix(prompt-gate): share message reservations --- src/cli/run/runner.ts | 32 +++- src/hooks/shared/prompt-async-gate.test.ts | 45 +++++ src/hooks/shared/prompt-async-gate.ts | 112 +---------- src/hooks/shared/session-idle-settle.ts | 62 +----- src/shared/model-suggestion-retry.test.ts | 36 ++++ src/shared/model-suggestion-retry.ts | 68 +++++-- src/shared/prompt-async-gate.ts | 207 +++++++++++++++++++++ src/shared/session-idle-settle.ts | 61 ++++++ src/shared/session-route.ts | 24 ++- 9 files changed, 451 insertions(+), 196 deletions(-) create mode 100644 src/shared/prompt-async-gate.ts create mode 100644 src/shared/session-idle-settle.ts diff --git a/src/cli/run/runner.ts b/src/cli/run/runner.ts index 75e6e49da..d59714065 100644 --- a/src/cli/run/runner.ts +++ b/src/cli/run/runner.ts @@ -13,6 +13,7 @@ import { loadAgentProfileColors } from "./agent-profile-colors" import { suppressRunInput } from "./stdin-suppression" import { createTimestampedStdoutController } from "./timestamp-output" import { createCliPostHog, getPostHogDistinctId } from "../../shared/posthog" +import { promptAsyncAfterSessionIdle } from "../../shared/prompt-async-gate" export { resolveRunAgent } @@ -109,18 +110,31 @@ export async function run(options: RunOptions): Promise { () => {}, ) - await client.session.promptAsync({ - path: { id: sessionID }, - body: { - agent: resolvedAgent, - ...(resolvedModel ? { model: resolvedModel } : {}), - tools: { - question: false, + const promptResult = await promptAsyncAfterSessionIdle({ + client, + sessionID, + source: "cli-run", + settleMs: 0, + postDispatchHoldMs: 0, + input: { + path: { id: sessionID }, + body: { + agent: resolvedAgent, + ...(resolvedModel ? { model: resolvedModel } : {}), + tools: { + question: false, + }, + parts: [{ type: "text", text: message }], }, - parts: [{ type: "text", text: message }], + query: { directory }, }, - query: { directory }, }) + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + throw new Error(`Session ${sessionID} is not idle; promptAsync skipped by gate: ${promptResult.status}`) + } const exitCode = await pollForCompletion(ctx, eventState, abortController) abortController.abort() diff --git a/src/hooks/shared/prompt-async-gate.test.ts b/src/hooks/shared/prompt-async-gate.test.ts index 6df4a189e..260db9d67 100644 --- a/src/hooks/shared/prompt-async-gate.test.ts +++ b/src/hooks/shared/prompt-async-gate.test.ts @@ -1,6 +1,7 @@ import { afterEach, describe, expect, test } from "bun:test" import { + promptAfterSessionIdle, promptAsyncAfterSessionIdle, releaseAllPromptAsyncReservationsForTesting, } from "./prompt-async-gate" @@ -81,4 +82,48 @@ describe("promptAsyncAfterSessionIdle", () => { expect(result.status).toBe("active") expect(promptCalls).toBe(0) }) + + test("#given two internal prompt calls race for one idle session #when they dispatch concurrently #then only one prompt is accepted", async () => { + // given + let promptCalls = 0 + let releasePrompt: (() => void) | undefined + const promptGate = new Promise((resolve) => { + releasePrompt = resolve + }) + const client = { + session: { + status: async () => ({ data: { ses_prompt_race: { type: "idle" } } }), + prompt: async () => { + promptCalls += 1 + await promptGate + }, + }, + } + + // when + const first = promptAfterSessionIdle({ + client, + sessionID: "ses_prompt_race", + input: { path: { id: "ses_prompt_race" }, body: { parts: [] } }, + source: "test:prompt:first", + settleMs: 0, + postDispatchHoldMs: 0, + }) + await Promise.resolve() + const second = await promptAfterSessionIdle({ + client, + sessionID: "ses_prompt_race", + input: { path: { id: "ses_prompt_race" }, body: { parts: [] } }, + source: "test:prompt:second", + settleMs: 0, + postDispatchHoldMs: 0, + }) + releasePrompt?.() + const firstResult = await first + + // then + expect(firstResult.status).toBe("dispatched") + expect(second.status).toBe("reserved") + expect(promptCalls).toBe(1) + }) }) diff --git a/src/hooks/shared/prompt-async-gate.ts b/src/hooks/shared/prompt-async-gate.ts index f037d18b4..68d44ab1b 100644 --- a/src/hooks/shared/prompt-async-gate.ts +++ b/src/hooks/shared/prompt-async-gate.ts @@ -1,111 +1 @@ -import { log } from "../../shared/logger" -import { - DEFAULT_SESSION_IDLE_SETTLE_MS, - isSessionActive, - settleAfterSessionIdle, -} from "./session-idle-settle" - -export const DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS = 250 - -type PromptAsyncInput = { - path?: { id?: string } - body?: unknown - query?: unknown - signal?: unknown - [key: string]: unknown -} - -type PromptAsyncClient = { - session?: { - status?: () => Promise - promptAsync?: (input: TInput) => Promise - } -} - -type PromptAsyncReservation = { - source: string - reservedAt: number - token: symbol -} - -export type PromptAsyncGateResult = - | { status: "dispatched"; response: unknown } - | { status: "active" } - | { status: "reserved"; reservedBy: string } - | { status: "unavailable" } - | { status: "failed"; error: unknown } - -const promptAsyncReservations = new Map() - -export async function promptAsyncAfterSessionIdle(args: { - client: PromptAsyncClient - sessionID: string - input: TInput - source: string - settleMs?: number - postDispatchHoldMs?: number -}): Promise { - const { - client, - sessionID, - input, - source, - settleMs = DEFAULT_SESSION_IDLE_SETTLE_MS, - } = args - const postDispatchHoldMs = args.postDispatchHoldMs ?? ( - settleMs > 0 ? DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS : 0 - ) - - if (typeof client.session?.promptAsync !== "function") { - log("[prompt-async-gate] promptAsync unavailable", { sessionID, source }) - return { status: "unavailable" } - } - - const existing = promptAsyncReservations.get(sessionID) - if (existing) { - log("[prompt-async-gate] promptAsync skipped because session is reserved", { - sessionID, - source, - reservedBy: existing.source, - reservedAgeMs: Date.now() - existing.reservedAt, - }) - return { status: "reserved", reservedBy: existing.source } - } - - const reservation: PromptAsyncReservation = { - source, - reservedAt: Date.now(), - token: Symbol(source), - } - promptAsyncReservations.set(sessionID, reservation) - - try { - const canReadStatus = typeof client.session?.status === "function" - await settleAfterSessionIdle(settleMs) - - if (canReadStatus && await isSessionActive(client, sessionID)) { - log("[prompt-async-gate] promptAsync skipped because session is active", { sessionID, source }) - return { status: "active" } - } - - log("[prompt-async-gate] promptAsync dispatching", { sessionID, source }) - const response = await client.session.promptAsync(input) - if (canReadStatus) { - await settleAfterSessionIdle(postDispatchHoldMs) - } - log("[prompt-async-gate] promptAsync dispatched", { sessionID, source }) - return { status: "dispatched", response } - } catch (error) { - log("[prompt-async-gate] promptAsync failed", { sessionID, source, error: String(error) }) - return { status: "failed", error } - } finally { - const current = promptAsyncReservations.get(sessionID) - if (current?.token === reservation.token) { - promptAsyncReservations.delete(sessionID) - } - } -} - -export function releaseAllPromptAsyncReservationsForTesting(): void { - promptAsyncReservations.clear() -} +export * from "../../shared/prompt-async-gate" diff --git a/src/hooks/shared/session-idle-settle.ts b/src/hooks/shared/session-idle-settle.ts index 2fd5a2b0a..6060b9e06 100644 --- a/src/hooks/shared/session-idle-settle.ts +++ b/src/hooks/shared/session-idle-settle.ts @@ -1,61 +1 @@ -export const DEFAULT_SESSION_IDLE_SETTLE_MS = 150 - -export function settleAfterSessionIdle(ms = DEFAULT_SESSION_IDLE_SETTLE_MS): Promise { - return ms > 0 ? new Promise((resolve) => setTimeout(resolve, ms)) : Promise.resolve() -} - -type SessionStatusClient = { - session?: { - status?: () => Promise - } -} - -const ACTIVE_SESSION_STATUSES = new Set(["busy", "retry", "running"]) - -function isRecord(value: unknown): value is Record { - return typeof value === "object" && value !== null -} - -function getSessionStatusPayload(response: unknown): Record { - if (isRecord(response) && isRecord(response.data)) { - return response.data - } - - if (isRecord(response)) { - return response - } - - return {} -} - -export function isActiveSessionStatusType(statusType: string): boolean { - return ACTIVE_SESSION_STATUSES.has(statusType) -} - -export async function isSessionActive(client: SessionStatusClient, sessionID: string): Promise { - if (typeof client.session?.status !== "function") { - return false - } - - try { - const statusResult = await client.session.status() - const status = getSessionStatusPayload(statusResult)[sessionID] - if (!isRecord(status)) { - return false - } - - const statusType = status.type - return typeof statusType === "string" && isActiveSessionStatusType(statusType) - } catch { - return false - } -} - -export async function shouldPromptAfterSessionIdle( - client: SessionStatusClient, - sessionID: string, - settleMs = DEFAULT_SESSION_IDLE_SETTLE_MS, -): Promise { - await settleAfterSessionIdle(settleMs) - return !(await isSessionActive(client, sessionID)) -} +export * from "../../shared/session-idle-settle" diff --git a/src/shared/model-suggestion-retry.test.ts b/src/shared/model-suggestion-retry.test.ts index e594ad6a3..fb2e3248a 100644 --- a/src/shared/model-suggestion-retry.test.ts +++ b/src/shared/model-suggestion-retry.test.ts @@ -230,6 +230,42 @@ describe("promptWithModelSuggestionRetry", () => { expect(promptMock).toHaveBeenCalledTimes(1) }) + it("should reject concurrent promptAsync retries for the same session after one dispatch is reserved", async () => { + // given two callers racing to send into one session + let releasePrompt: (() => void) | undefined + const promptGate = new Promise((resolve) => { + releasePrompt = resolve + }) + const promptMock = mock(async () => { + await promptGate + }) + const client = { + session: { + status: async () => ({ data: { "session-dup": { type: "idle" } } }), + promptAsync: promptMock, + }, + } + const args = { + path: { id: "session-dup" }, + body: { + parts: [{ type: "text", text: "hello" }], + model: { providerID: "anthropic", modelID: "claude-sonnet-4" }, + }, + } + + // when both callers try to prompt the same session before the first dispatch settles + const first = promptWithModelSuggestionRetry(unsafeTestValue(client), args) + await Promise.resolve() + const second = promptWithModelSuggestionRetry(unsafeTestValue(client), args) + releasePrompt?.() + const results = await Promise.allSettled([first, second]) + + // then only the reserved dispatch is sent to OpenCode + expect(promptMock).toHaveBeenCalledTimes(1) + expect(results[0]?.status).toBe("fulfilled") + expect(results[1]?.status).toBe("rejected") + }) + it("should throw error from promptAsync directly on model-not-found error", async () => { // given a client that fails with model-not-found error const promptMock = mock().mockRejectedValueOnce({ diff --git a/src/shared/model-suggestion-retry.ts b/src/shared/model-suggestion-retry.ts index 7047b8bb5..184467e4d 100644 --- a/src/shared/model-suggestion-retry.ts +++ b/src/shared/model-suggestion-retry.ts @@ -5,6 +5,7 @@ import { PROMPT_TIMEOUT_MS, type PromptRetryOptions, } from "./prompt-timeout-context" +import { promptAfterSessionIdle, promptAsyncAfterSessionIdle } from "./prompt-async-gate" type Client = ReturnType @@ -93,14 +94,25 @@ export async function promptWithModelSuggestionRetry( ): Promise { const timeoutMs = options.timeoutMs ?? PROMPT_TIMEOUT_MS const timeoutContext = createPromptTimeoutContext(args, timeoutMs) - // model errors happen asynchronously server-side and cannot be caught here - const promptPromise = client.session.promptAsync({ - ...args, - signal: timeoutContext.signal, - } as Parameters[0]) try { - await promptPromise + const promptResult = await promptAsyncAfterSessionIdle({ + client, + sessionID: args.path.id, + input: { + ...args, + signal: timeoutContext.signal, + } as Parameters[0], + source: "model-suggestion-retry", + settleMs: 0, + postDispatchHoldMs: 0, + }) + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + throw new Error(`promptAsync skipped by gate: ${promptResult.status}`) + } if (timeoutContext.wasTimedOut()) { throw new Error(`promptAsync timed out after ${timeoutMs}ms`) } @@ -124,10 +136,24 @@ export async function promptSyncWithModelSuggestionRetry( try { const timeoutContext = createPromptTimeoutContext(args, timeoutMs) try { - await client.session.prompt({ - ...args, - signal: timeoutContext.signal, - } as Parameters[0]) + const promptResult = await promptAfterSessionIdle({ + client, + sessionID: args.path.id, + input: { + ...args, + signal: timeoutContext.signal, + } as Parameters[0], + source: "model-suggestion-retry:sync", + settleMs: 0, + postDispatchHoldMs: 0, + checkStatus: false, + }) + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + throw new Error(`prompt skipped by gate: ${promptResult.status}`) + } if (timeoutContext.wasTimedOut()) { throw new Error(`prompt timed out after ${timeoutMs}ms`) } @@ -163,10 +189,24 @@ export async function promptSyncWithModelSuggestionRetry( const timeoutContext = createPromptTimeoutContext(retryArgs, timeoutMs) try { - await client.session.prompt({ - ...retryArgs, - signal: timeoutContext.signal, - } as Parameters[0]) + const promptResult = await promptAfterSessionIdle({ + client, + sessionID: retryArgs.path.id, + input: { + ...retryArgs, + signal: timeoutContext.signal, + } as Parameters[0], + source: "model-suggestion-retry:sync-retry", + settleMs: 0, + postDispatchHoldMs: 0, + checkStatus: false, + }) + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + throw new Error(`prompt skipped by gate: ${promptResult.status}`) + } if (timeoutContext.wasTimedOut()) { throw new Error(`prompt timed out after ${timeoutMs}ms`) } diff --git a/src/shared/prompt-async-gate.ts b/src/shared/prompt-async-gate.ts new file mode 100644 index 000000000..806b2178d --- /dev/null +++ b/src/shared/prompt-async-gate.ts @@ -0,0 +1,207 @@ +import { log } from "./logger" +import { + DEFAULT_SESSION_IDLE_SETTLE_MS, + isSessionActive, + settleAfterSessionIdle, +} from "./session-idle-settle" + +export const DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS = 250 + +type PromptAsyncInput = { + path?: { id?: string } + body?: unknown + query?: unknown + signal?: unknown + [key: string]: unknown +} + +type PromptAsyncClient = { + session?: { + status?: () => Promise + promptAsync?: (input: TInput) => Promise + } +} + +type PromptClient = { + session?: { + status?: () => Promise + prompt?: (input: TInput) => Promise + } +} + +type PromptAsyncReservation = { + source: string + reservedAt: number + token: symbol +} + +export type PromptAsyncGateResult = + | { status: "dispatched"; response: unknown } + | { status: "active" } + | { status: "reserved"; reservedBy: string } + | { status: "unavailable" } + | { status: "failed"; error: unknown } + +const promptAsyncReservations = new Map() + +export async function promptAsyncAfterSessionIdle(args: { + client: PromptAsyncClient + sessionID: string + input: TInput + source: string + settleMs?: number + postDispatchHoldMs?: number + checkStatus?: boolean +}): Promise { + const { + client, + sessionID, + input, + source, + settleMs = DEFAULT_SESSION_IDLE_SETTLE_MS, + } = args + const postDispatchHoldMs = args.postDispatchHoldMs ?? ( + settleMs > 0 ? DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS : 0 + ) + + if (typeof client.session?.promptAsync !== "function") { + log("[prompt-async-gate] promptAsync unavailable", { sessionID, source }) + return { status: "unavailable" } + } + + const existing = promptAsyncReservations.get(sessionID) + if (existing) { + log("[prompt-async-gate] promptAsync skipped because session is reserved", { + sessionID, + source, + reservedBy: existing.source, + reservedAgeMs: Date.now() - existing.reservedAt, + }) + return { status: "reserved", reservedBy: existing.source } + } + + const reservation: PromptAsyncReservation = { + source, + reservedAt: Date.now(), + token: Symbol(source), + } + promptAsyncReservations.set(sessionID, reservation) + + try { + const canReadStatus = args.checkStatus !== false && typeof client.session?.status === "function" + if (settleMs > 0) { + await settleAfterSessionIdle(settleMs) + } + + if (canReadStatus && await isSessionActive(client, sessionID)) { + log("[prompt-async-gate] promptAsync skipped because session is active", { sessionID, source }) + return { status: "active" } + } + + log("[prompt-async-gate] promptAsync dispatching", { sessionID, source }) + const response = await client.session.promptAsync(input) + if (canReadStatus) { + await settleAfterSessionIdle(postDispatchHoldMs) + } + log("[prompt-async-gate] promptAsync dispatched", { sessionID, source }) + return { status: "dispatched", response } + } catch (error) { + log("[prompt-async-gate] promptAsync failed", { sessionID, source, error: String(error) }) + return { status: "failed", error } + } finally { + const current = promptAsyncReservations.get(sessionID) + if (current?.token === reservation.token) { + promptAsyncReservations.delete(sessionID) + } + } +} + +export async function promptAfterSessionIdle(args: { + client: PromptClient + sessionID: string + input: TInput + source: string + settleMs?: number + postDispatchHoldMs?: number + checkStatus?: boolean +}): Promise { + const { + client, + sessionID, + input, + source, + settleMs = DEFAULT_SESSION_IDLE_SETTLE_MS, + } = args + const postDispatchHoldMs = args.postDispatchHoldMs ?? ( + settleMs > 0 ? DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS : 0 + ) + + if (typeof client.session?.prompt !== "function") { + log("[prompt-async-gate] prompt unavailable", { sessionID, source }) + return { status: "unavailable" } + } + + const existing = promptAsyncReservations.get(sessionID) + if (existing) { + log("[prompt-async-gate] prompt skipped because session is reserved", { + sessionID, + source, + reservedBy: existing.source, + reservedAgeMs: Date.now() - existing.reservedAt, + }) + return { status: "reserved", reservedBy: existing.source } + } + + const reservation: PromptAsyncReservation = { + source, + reservedAt: Date.now(), + token: Symbol(source), + } + promptAsyncReservations.set(sessionID, reservation) + + try { + const canReadStatus = args.checkStatus !== false && typeof client.session?.status === "function" + if (settleMs > 0) { + await settleAfterSessionIdle(settleMs) + } + + if (canReadStatus && await isSessionActive(client, sessionID)) { + log("[prompt-async-gate] prompt skipped because session is active", { sessionID, source }) + return { status: "active" } + } + + log("[prompt-async-gate] prompt dispatching", { sessionID, source }) + const response = await client.session.prompt(input) + if (canReadStatus) { + await settleAfterSessionIdle(postDispatchHoldMs) + } + log("[prompt-async-gate] prompt dispatched", { sessionID, source }) + return { status: "dispatched", response } + } catch (error) { + log("[prompt-async-gate] prompt failed", { sessionID, source, error: String(error) }) + return { status: "failed", error } + } finally { + const current = promptAsyncReservations.get(sessionID) + if (current?.token === reservation.token) { + promptAsyncReservations.delete(sessionID) + } + } +} + +export function releaseAllPromptAsyncReservationsForTesting(): void { + promptAsyncReservations.clear() +} + +export function releasePromptAsyncReservation(sessionID: string, source: string): void { + const existing = promptAsyncReservations.get(sessionID) + if (!existing) { + return + } + + promptAsyncReservations.delete(sessionID) + log("[prompt-async-gate] promptAsync reservation released", { + sessionID, + source, + reservedBy: existing.source, + }) +} diff --git a/src/shared/session-idle-settle.ts b/src/shared/session-idle-settle.ts new file mode 100644 index 000000000..2fd5a2b0a --- /dev/null +++ b/src/shared/session-idle-settle.ts @@ -0,0 +1,61 @@ +export const DEFAULT_SESSION_IDLE_SETTLE_MS = 150 + +export function settleAfterSessionIdle(ms = DEFAULT_SESSION_IDLE_SETTLE_MS): Promise { + return ms > 0 ? new Promise((resolve) => setTimeout(resolve, ms)) : Promise.resolve() +} + +type SessionStatusClient = { + session?: { + status?: () => Promise + } +} + +const ACTIVE_SESSION_STATUSES = new Set(["busy", "retry", "running"]) + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null +} + +function getSessionStatusPayload(response: unknown): Record { + if (isRecord(response) && isRecord(response.data)) { + return response.data + } + + if (isRecord(response)) { + return response + } + + return {} +} + +export function isActiveSessionStatusType(statusType: string): boolean { + return ACTIVE_SESSION_STATUSES.has(statusType) +} + +export async function isSessionActive(client: SessionStatusClient, sessionID: string): Promise { + if (typeof client.session?.status !== "function") { + return false + } + + try { + const statusResult = await client.session.status() + const status = getSessionStatusPayload(statusResult)[sessionID] + if (!isRecord(status)) { + return false + } + + const statusType = status.type + return typeof statusType === "string" && isActiveSessionStatusType(statusType) + } catch { + return false + } +} + +export async function shouldPromptAfterSessionIdle( + client: SessionStatusClient, + sessionID: string, + settleMs = DEFAULT_SESSION_IDLE_SETTLE_MS, +): Promise { + await settleAfterSessionIdle(settleMs) + return !(await isSessionActive(client, sessionID)) +} diff --git a/src/shared/session-route.ts b/src/shared/session-route.ts index 53901cf23..e6e5428dc 100644 --- a/src/shared/session-route.ts +++ b/src/shared/session-route.ts @@ -3,6 +3,7 @@ import { promptSyncWithModelSuggestionRetry, promptWithModelSuggestionRetry, } from "./model-suggestion-retry" +import { promptAsyncAfterSessionIdle } from "./prompt-async-gate" type OpencodeClient = PluginInput["client"] @@ -52,7 +53,28 @@ export function promptAsyncInDirectory( args: PromptAsyncArgs, directory: string, ): Promise { - return client.session.promptAsync(routeSessionPrompt(args, directory)) + const routedArgs = routeSessionPrompt(args, directory) + const sessionID = routedArgs.path?.id + if (!sessionID) { + return client.session.promptAsync(routedArgs) + } + + return promptAsyncAfterSessionIdle({ + client, + sessionID, + input: routedArgs, + source: "session-route", + settleMs: 0, + postDispatchHoldMs: 0, + }).then((result) => { + if (result.status === "failed") { + throw result.error + } + if (result.status !== "dispatched") { + throw new Error(`promptAsync skipped by gate: ${result.status}`) + } + return result.response + }) } export function promptWithRetryInDirectory( From 439e72839b3e6ac65ef49ead6bdf5aa83447eb3f Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:49:55 +0900 Subject: [PATCH 12/17] fix(runtime-fallback): gate retry prompts --- src/hooks/runtime-fallback/auto-retry.ts | 36 +++++++++++--- src/hooks/runtime-fallback/index.test.ts | 62 ++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 7 deletions(-) diff --git a/src/hooks/runtime-fallback/auto-retry.ts b/src/hooks/runtime-fallback/auto-retry.ts index cbb3be2be..7e8a7fd65 100644 --- a/src/hooks/runtime-fallback/auto-retry.ts +++ b/src/hooks/runtime-fallback/auto-retry.ts @@ -10,6 +10,10 @@ import { buildRetryModelPayload } from "./retry-model-payload" import { getLastUserRetryParts } from "./last-user-retry-parts" import { extractSessionMessages } from "./session-messages" import { resolveRegisteredAgentName } from "../../features/claude-code-session-state" +import { + promptAsyncAfterSessionIdle, + releasePromptAsyncReservation, +} from "../shared/prompt-async-gate" const SESSION_TTL_MS = 30 * 60 * 1000 @@ -33,6 +37,7 @@ export function createAutoRetryHelpers(deps: HookDeps) { const abortSessionRequest = async (sessionID: string, source: string): Promise => { try { await ctx.client.session.abort({ path: { id: sessionID } }) + releasePromptAsyncReservation(sessionID, `runtime-fallback-abort:${source}`) log(`[${HOOK_NAME}] Aborted in-flight session request (${source})`, { sessionID }) } catch (error) { log(`[${HOOK_NAME}] Failed to abort in-flight session request (${source})`, { @@ -137,15 +142,32 @@ export function createAutoRetryHelpers(deps: HookDeps) { sessionAwaitingFallbackResult.add(sessionID) scheduleSessionFallbackTimeout(sessionID, retryAgent) - await ctx.client.session.promptAsync({ - path: { id: sessionID }, - body: { - ...(launchAgent ? { agent: launchAgent } : {}), - ...retryModelPayload, - parts: retryParts, + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID, + source: `runtime-fallback:${source}`, + settleMs: 0, + postDispatchHoldMs: 0, + input: { + path: { id: sessionID }, + body: { + ...(launchAgent ? { agent: launchAgent } : {}), + ...retryModelPayload, + parts: retryParts, + }, + query: { directory: ctx.directory }, }, - query: { directory: ctx.directory }, }) + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + log(`[${HOOK_NAME}] Auto-retry skipped by promptAsync gate (${source})`, { + sessionID, + status: promptResult.status, + }) + return + } retryDispatched = true } else { log(`[${HOOK_NAME}] No user message found for auto-retry (${source})`, { sessionID }) diff --git a/src/hooks/runtime-fallback/index.test.ts b/src/hooks/runtime-fallback/index.test.ts index 29e8332ea..b3ae368fb 100644 --- a/src/hooks/runtime-fallback/index.test.ts +++ b/src/hooks/runtime-fallback/index.test.ts @@ -40,6 +40,7 @@ describe("runtime-fallback", () => { messages?: (args: unknown) => Promise promptAsync?: (args: unknown) => Promise abort?: (args: unknown) => Promise + status?: () => Promise } }) { return unsafeTestValue({ @@ -57,6 +58,7 @@ describe("runtime-fallback", () => { messages: overrides?.session?.messages ?? (async () => ({ data: [] })), promptAsync: overrides?.session?.promptAsync ?? (async () => ({})), abort: overrides?.session?.abort ?? (async () => ({})), + ...(overrides?.session?.status ? { status: overrides.session.status } : {}), }, }, directory: "/test/dir", @@ -2471,6 +2473,66 @@ describe("runtime-fallback", () => { expect(callBody?.agent).toBe("prometheus") expect(callBody?.model).toEqual({ providerID: "github-copilot", modelID: "claude-opus-4.7" }) }) + + test("should not dispatch a second fallback prompt while the accepted retry session is still active", async () => { + const sessionID = "test-runtime-fallback-active-gate" + let sessionStatus = "idle" + const promptCalls: Array> = [] + const hook = createRuntimeFallbackHook( + createMockPluginInput({ + session: { + messages: async () => ({ + data: [ + { + info: { role: "user" }, + parts: [{ type: "text", text: "retry this" }], + }, + ], + }), + promptAsync: async (args: unknown) => { + promptCalls.push(args as Record) + sessionStatus = "busy" + return {} + }, + status: async () => ({ data: { [sessionID]: { type: sessionStatus } } }), + }, + }), + { + config: createMockConfig({ notify_on_fallback: false }), + pluginConfig: createMockPluginConfigWithCategoryFallback([ + "github-copilot/claude-opus-4.7", + "openai/gpt-5.4", + ]), + }, + ) + SessionCategoryRegistry.register(sessionID, "test") + + await hook.event({ + event: { + type: "session.created", + properties: { info: { id: sessionID, model: "anthropic/claude-opus-4-7" } }, + }, + }) + + await hook.event({ + event: { + type: "session.error", + properties: { sessionID, error: { statusCode: 503, message: "Service unavailable" } }, + }, + }) + await hook.event({ + event: { + type: "session.error", + properties: { + sessionID, + model: "github-copilot/claude-opus-4.7", + error: { statusCode: 503, message: "Service unavailable" }, + }, + }, + }) + + expect(promptCalls).toHaveLength(1) + }) }) describe("cooldown mechanism", () => { From 0b48f805694821a6cc36f6e866f414d427006fe0 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:49:58 +0900 Subject: [PATCH 13/17] fix(call-omo-agent): gate reused sync prompts --- .../call-omo-agent/sync-executor.test.ts | 41 ++++++++++++++++++- src/tools/call-omo-agent/sync-executor.ts | 38 +++++++++++------ 2 files changed, 66 insertions(+), 13 deletions(-) diff --git a/src/tools/call-omo-agent/sync-executor.test.ts b/src/tools/call-omo-agent/sync-executor.test.ts index 9b8392b93..c0bb0a8d3 100644 --- a/src/tools/call-omo-agent/sync-executor.test.ts +++ b/src/tools/call-omo-agent/sync-executor.test.ts @@ -79,11 +79,15 @@ function createToolContext(): ToolContext { } } -function createContext(promptAsync: ReturnType) { +function createContext( + promptAsync: ReturnType, + status?: () => Promise, +) { return { client: { session: { promptAsync, + ...(status ? { status } : {}), }, }, } @@ -350,6 +354,41 @@ describe("executeSync", () => { expect(deps.processMessages).not.toHaveBeenCalled() }) + test("does not send a duplicate sync prompt when a reused session is active", async () => { + //#given + const executeSync = await importExecuteSync() + const deps = createDependencies({ + createOrGetSession: mock(async () => ({ sessionID: "ses-active-reuse", isNew: false })), + }) + const toolContext = createToolContext() + const recorder = createPromptAsyncRecorder() + const args = { + subagent_type: "explore", + description: "active reuse", + prompt: "find something", + run_in_background: false, + session_id: "ses-active-reuse", + } + + //#when + const result = await executeSync( + args, + toolContext, + createContext( + recorder.promptAsync, + async () => ({ data: { "ses-active-reuse": { type: "busy" } } }), + ) as never, + deps, + ) + + //#then + expect(recorder.promptAsync).toHaveBeenCalledTimes(0) + expect(result).toContain("Error: Failed to send prompt") + expect(result).toContain("session_id: ses-active-reuse") + expect(deps.waitForCompletion).not.toHaveBeenCalled() + expect(deps.processMessages).not.toHaveBeenCalled() + }) + test("commits reserved descendant quota after creating a new sync session", async () => { //#given const { executeSync } = require("./sync-executor") diff --git a/src/tools/call-omo-agent/sync-executor.ts b/src/tools/call-omo-agent/sync-executor.ts index 31ae8beb8..85c0c6213 100644 --- a/src/tools/call-omo-agent/sync-executor.ts +++ b/src/tools/call-omo-agent/sync-executor.ts @@ -6,6 +6,7 @@ import { applySessionPromptParams } from "../../shared/session-prompt-params-hel import type { DelegatedModelConfig } from "../../shared/model-resolution-types" import type { FallbackEntry } from "../../shared/model-requirements" import { stripAgentListSortPrefix } from "../../shared/agent-display-names" +import { promptAsyncAfterSessionIdle } from "../../hooks/shared/prompt-async-gate" import { waitForCompletion } from "./completion-poller" import { processMessages } from "./message-processor" import { createOrGetSession } from "./session-creator" @@ -110,21 +111,34 @@ export async function executeSync( return `Error: Failed to send prompt: promptAsync is not available on this OpenCode client.\n\n\nsession_id: ${sessionID}\n` } - await ctx.client.session.promptAsync({ - path: { id: sessionID }, - body: { - agent: normalizedSubagentType, - tools: { - ...getAgentToolRestrictions(normalizedSubagentType), - task: false, - question: false, + const promptResult = await promptAsyncAfterSessionIdle({ + client: ctx.client, + sessionID, + source: "call-omo-agent:sync", + settleMs: 0, + postDispatchHoldMs: 0, + input: { + path: { id: sessionID }, + body: { + agent: normalizedSubagentType, + tools: { + ...getAgentToolRestrictions(normalizedSubagentType), + task: false, + question: false, + }, + parts: [{ type: "text", text: args.prompt }], + ...(model ? { model: { providerID: model.providerID, modelID: model.modelID } } : {}), + ...(model?.variant ? { variant: model.variant } : {}), + ...buildPromptGenerationParams(model), }, - parts: [{ type: "text", text: args.prompt }], - ...(model ? { model: { providerID: model.providerID, modelID: model.modelID } } : {}), - ...(model?.variant ? { variant: model.variant } : {}), - ...buildPromptGenerationParams(model), }, }) + if (promptResult.status === "failed") { + throw promptResult.error + } + if (promptResult.status !== "dispatched") { + throw new Error(`promptAsync skipped by gate: ${promptResult.status}`) + } } catch (error) { const errorMessage = error instanceof Error ? error.message : String(error) log(`[call_omo_agent] Prompt error:`, errorMessage) From edf3e530d4a62de68347891af3532f3d43017d16 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:50:02 +0900 Subject: [PATCH 14/17] fix(hooks): gate sync injected prompts --- .../handlers/session-event-handler.ts | 19 +++++++---- src/plugin/event.ts | 34 ++++++++++++++----- 2 files changed, 38 insertions(+), 15 deletions(-) diff --git a/src/hooks/claude-code-hooks/handlers/session-event-handler.ts b/src/hooks/claude-code-hooks/handlers/session-event-handler.ts index 534b39e34..8326fdb5f 100644 --- a/src/hooks/claude-code-hooks/handlers/session-event-handler.ts +++ b/src/hooks/claude-code-hooks/handlers/session-event-handler.ts @@ -8,6 +8,7 @@ import { clearToolInputCache, stopToolInputCacheCleanup } from "../tool-input-ca import type { PluginConfig } from "../types" import { createInternalAgentTextPart, isHookDisabled, log } from "../../../shared" import { resolveSessionEventID } from "../../../shared/event-session-id" +import { promptAfterSessionIdle } from "../../../shared/prompt-async-gate" import { clearAllSessionHookState, clearSessionHookState, @@ -108,17 +109,23 @@ export function createSessionEventHandler( }) } else if (stopResult.block && stopResult.injectPrompt) { log("Stop hook returned block with inject_prompt", { sessionID }) - ctx.client.session - .prompt({ + const promptResult = await promptAfterSessionIdle({ + client: ctx.client, + sessionID, + source: "claude-code-stop-hook:inject-prompt", + input: { path: { id: sessionID }, body: { parts: [createInternalAgentTextPart(stopResult.injectPrompt)], }, query: { directory: ctx.directory }, - }) - .catch((err: unknown) => - log("Failed to inject prompt from Stop hook", { error: String(err) }), - ) + }, + }) + if (promptResult.status === "failed") { + log("Failed to inject prompt from Stop hook", { error: String(promptResult.error) }) + } else if (promptResult.status !== "dispatched") { + log("Skipped prompt injection from Stop hook", { sessionID, status: promptResult.status }) + } } else if (stopResult.block) { log("Stop hook returned block", { sessionID, reason: stopResult.reason }) } diff --git a/src/plugin/event.ts b/src/plugin/event.ts index 0de6738a2..840bbe994 100644 --- a/src/plugin/event.ts +++ b/src/plugin/event.ts @@ -42,7 +42,7 @@ import { createTeamIdleWakeHint } from "../hooks/team-session-events/team-idle-w import { createTeamLeadOrphanHandler } from "../hooks/team-session-events/team-lead-orphan-handler"; import { createTeamMemberErrorHandler } from "../hooks/team-session-events/team-member-error-handler"; import { createTeamMemberStatusHandler } from "../hooks/team-session-events/team-member-status-handler"; -import { promptAsyncAfterSessionIdle } from "../hooks/shared/prompt-async-gate"; +import { promptAfterSessionIdle, promptAsyncAfterSessionIdle } from "../hooks/shared/prompt-async-gate"; import type { CreatedHooks } from "../create-hooks"; import type { Managers } from "../create-managers"; @@ -514,11 +514,19 @@ export function createEventHandler(args: { return; } - await pluginContext.client.session.prompt(promptBody).then(() => { - dispatched = true; - }).catch((error) => { - log("[event] model-fallback prompt failed", { sessionID, source, error }); + const promptResult = await promptAfterSessionIdle({ + client: pluginContext.client, + sessionID, + source: `model-fallback:${source}:sync`, + input: promptBody, }); + if (promptResult.status === "dispatched") { + dispatched = true; + } else if (promptResult.status === "failed") { + log("[event] model-fallback prompt failed", { sessionID, source, error: promptResult.error }); + } else { + log("[event] model-fallback prompt skipped by gate", { sessionID, source, status: promptResult.status }); + } } finally { if (dispatched && fallbackKeys.modelKey) { const dispatchedKeys = getFallbackContinuationDedupeState(sessionID); @@ -909,13 +917,21 @@ export function createEventHandler(args: { log("[event] compaction before recovery continue failed:", { sessionID, error: err }); }); - await pluginContext.client.session - .prompt({ + const promptResult = await promptAfterSessionIdle({ + client: pluginContext.client, + sessionID, + source: "session-recovery:post-compaction-continue", + input: { path: { id: sessionID }, body: { parts: [createInternalAgentContinuationTextPart("continue")] }, query: { directory: pluginContext.directory }, - }) - .catch(() => {}); + }, + }); + if (promptResult.status === "failed") { + log("[event] recovery continue prompt failed", { sessionID, error: promptResult.error }); + } else if (promptResult.status !== "dispatched") { + log("[event] recovery continue prompt skipped by gate", { sessionID, status: promptResult.status }); + } } } // Second, try model fallback for model errors (rate limit, quota, provider issues, etc.) From c6e3b7e1f708ce3f4e10a9f4a13940fa6d262339 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 11:50:08 +0900 Subject: [PATCH 15/17] docs(agents-md): warn on prompt injection --- AGENTS.md | 1 + 1 file changed, 1 insertion(+) diff --git a/AGENTS.md b/AGENTS.md index e6d3134da..987e611ba 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -173,6 +173,7 @@ Schema autocomplete: `"$schema": "https://raw.githubusercontent.com/code-yeongyu - **Per-session MCP isolation:** Tier-3 MCP clients keyed by `${sessionID}:${skillName}:${serverName}` so the same skill in two sessions does not share state. - **Two fallback systems:** `model-fallback` (proactive, chat.params) vs `runtime-fallback` (reactive, session.error). They operate independently — no direct integration. - **OpenClaw bidirectional:** Outbound dispatchers fire on session events; inbound daemon polls Discord/Telegram and `send-keys` replies into the tracked tmux pane. +- **Internal message injection is dangerous:** OpenCode의 stupid한 설계로 플러그인이 `session.prompt` / `session.promptAsync` 같은 메인 세션 메시지 API를 통해 메인 시스템을 망가뜨릴 수 있다. Any route that injects a message into an existing session must go through `src/shared/prompt-async-gate.ts` or an equivalent per-session reservation/status gate, must check active session state before dispatch, and must have a regression test proving duplicate concurrent injections collapse to one dispatch. This applies to background completion wakes, fallback retries, team mailbox live delivery, recovery continuations, CLI run resumes, Claude Code hook injections, and sync/background subagent prompts. ## CONVENTIONS From 05189700fb24a64fdacd55e4b7712cf989e4b658 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 12:05:03 +0900 Subject: [PATCH 16/17] fix(prompt-gate): hold reservations after dispatch --- src/hooks/shared/prompt-async-gate.test.ts | 70 ++++++++++++++++++++++ src/shared/prompt-async-gate.ts | 12 ++-- 2 files changed, 74 insertions(+), 8 deletions(-) diff --git a/src/hooks/shared/prompt-async-gate.test.ts b/src/hooks/shared/prompt-async-gate.test.ts index 260db9d67..cd943d3b4 100644 --- a/src/hooks/shared/prompt-async-gate.test.ts +++ b/src/hooks/shared/prompt-async-gate.test.ts @@ -56,6 +56,41 @@ describe("promptAsyncAfterSessionIdle", () => { expect(promptCalls).toBe(1) }) + test("#given settle is disabled and status is unavailable #when a second promptAsync starts after the first dispatch resolves #then the default dispatch hold keeps the session reserved", async () => { + // given + let promptCalls = 0 + const client = { + session: { + promptAsync: async () => { + promptCalls += 1 + }, + }, + } + + // when + const first = promptAsyncAfterSessionIdle({ + client, + sessionID: "ses_hold_after_dispatch", + input: { path: { id: "ses_hold_after_dispatch" }, body: { parts: [] } }, + source: "test:hold:first", + settleMs: 0, + }) + await new Promise((resolve) => setTimeout(resolve, 0)) + const second = await promptAsyncAfterSessionIdle({ + client, + sessionID: "ses_hold_after_dispatch", + input: { path: { id: "ses_hold_after_dispatch" }, body: { parts: [] } }, + source: "test:hold:second", + settleMs: 0, + }) + const firstResult = await first + + // then + expect(firstResult.status).toBe("dispatched") + expect(second.status).toBe("reserved") + expect(promptCalls).toBe(1) + }) + test("#given session.status reports busy #when an internal promptAsync is requested #then no prompt is sent", async () => { // given let promptCalls = 0 @@ -126,4 +161,39 @@ describe("promptAsyncAfterSessionIdle", () => { expect(second.status).toBe("reserved") expect(promptCalls).toBe(1) }) + + test("#given settle is disabled and status is unavailable #when a second prompt starts after the first dispatch resolves #then the default dispatch hold keeps the session reserved", async () => { + // given + let promptCalls = 0 + const client = { + session: { + prompt: async () => { + promptCalls += 1 + }, + }, + } + + // when + const first = promptAfterSessionIdle({ + client, + sessionID: "ses_prompt_hold_after_dispatch", + input: { path: { id: "ses_prompt_hold_after_dispatch" }, body: { parts: [] } }, + source: "test:prompt-hold:first", + settleMs: 0, + }) + await new Promise((resolve) => setTimeout(resolve, 0)) + const second = await promptAfterSessionIdle({ + client, + sessionID: "ses_prompt_hold_after_dispatch", + input: { path: { id: "ses_prompt_hold_after_dispatch" }, body: { parts: [] } }, + source: "test:prompt-hold:second", + settleMs: 0, + }) + const firstResult = await first + + // then + expect(firstResult.status).toBe("dispatched") + expect(second.status).toBe("reserved") + expect(promptCalls).toBe(1) + }) }) diff --git a/src/shared/prompt-async-gate.ts b/src/shared/prompt-async-gate.ts index 806b2178d..7f2f57553 100644 --- a/src/shared/prompt-async-gate.ts +++ b/src/shared/prompt-async-gate.ts @@ -60,9 +60,7 @@ export async function promptAsyncAfterSessionIdle(arg source, settleMs = DEFAULT_SESSION_IDLE_SETTLE_MS, } = args - const postDispatchHoldMs = args.postDispatchHoldMs ?? ( - settleMs > 0 ? DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS : 0 - ) + const postDispatchHoldMs = args.postDispatchHoldMs ?? DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS if (typeof client.session?.promptAsync !== "function") { log("[prompt-async-gate] promptAsync unavailable", { sessionID, source }) @@ -100,7 +98,7 @@ export async function promptAsyncAfterSessionIdle(arg log("[prompt-async-gate] promptAsync dispatching", { sessionID, source }) const response = await client.session.promptAsync(input) - if (canReadStatus) { + if (postDispatchHoldMs > 0) { await settleAfterSessionIdle(postDispatchHoldMs) } log("[prompt-async-gate] promptAsync dispatched", { sessionID, source }) @@ -132,9 +130,7 @@ export async function promptAfterSessionIdle(args: { source, settleMs = DEFAULT_SESSION_IDLE_SETTLE_MS, } = args - const postDispatchHoldMs = args.postDispatchHoldMs ?? ( - settleMs > 0 ? DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS : 0 - ) + const postDispatchHoldMs = args.postDispatchHoldMs ?? DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS if (typeof client.session?.prompt !== "function") { log("[prompt-async-gate] prompt unavailable", { sessionID, source }) @@ -172,7 +168,7 @@ export async function promptAfterSessionIdle(args: { log("[prompt-async-gate] prompt dispatching", { sessionID, source }) const response = await client.session.prompt(input) - if (canReadStatus) { + if (postDispatchHoldMs > 0) { await settleAfterSessionIdle(postDispatchHoldMs) } log("[prompt-async-gate] prompt dispatched", { sessionID, source }) From c2aa180e7ee21fbfa6ee382f9064fa468e2e17ef Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Fri, 15 May 2026 13:19:10 +0900 Subject: [PATCH 17/17] fix(prompt-gate): pin duplicate prompt dispatches Keep prompt reservations briefly after successful dispatch so rapid idle/message/error transitions cannot inject the same follow-up twice. Route all production session prompt calls through the shared gate, restore skipped background resume state, release holds after abort/recovery paths, and preserve Ralph/ULW loop state when a dispatch is deferred. Add regression coverage for session routing, static prompt route auditing, team-mode live messaging, model suggestion retries, call-omo-agent reuse, background parent wakes, runtime fallback, compaction recovery, Atlas, and Ralph/ULW loops. --- src/cli/run/runner.ts | 1 - src/features/background-agent/manager.test.ts | 230 ++++++++++++++++++ src/features/background-agent/manager.ts | 217 ++++++++++++++++- .../team-mode/tools/messaging.test.ts | 65 +++++ src/hooks/atlas/index.test.ts | 2 +- .../compaction-context-injector/recovery.ts | 3 +- .../continuation-prompt-injector.ts | 4 + src/hooks/ralph-loop/index.test.ts | 25 ++ .../ralph-loop/iteration-continuation.ts | 7 + .../ralph-loop/ralph-loop-event-handler.ts | 10 + src/hooks/ralph-loop/ralph-loop-hook.ts | 4 + .../ralph-loop/ulw-loop-verification.test.ts | 14 +- .../verification-failure-handler.ts | 45 ++-- src/hooks/runtime-fallback/auto-retry.ts | 1 - .../message-update-handler.ts | 6 +- src/hooks/shared/prompt-async-gate.test.ts | 36 +++ src/plugin/event.ts | 3 +- src/shared/model-suggestion-retry.test.ts | 50 ++++ src/shared/model-suggestion-retry.ts | 3 - src/shared/prompt-async-gate.ts | 40 ++- src/shared/prompt-async-route-audit.test.ts | 78 ++++++ src/shared/session-route.test.ts | 61 +++++ src/shared/session-route.ts | 3 +- .../call-omo-agent/sync-executor.test.ts | 29 +++ src/tools/call-omo-agent/sync-executor.ts | 1 - test-setup.ts | 3 + 26 files changed, 893 insertions(+), 48 deletions(-) create mode 100644 src/shared/prompt-async-route-audit.test.ts create mode 100644 src/shared/session-route.test.ts diff --git a/src/cli/run/runner.ts b/src/cli/run/runner.ts index d59714065..81763668f 100644 --- a/src/cli/run/runner.ts +++ b/src/cli/run/runner.ts @@ -115,7 +115,6 @@ export async function run(options: RunOptions): Promise { sessionID, source: "cli-run", settleMs: 0, - postDispatchHoldMs: 0, input: { path: { id: sessionID }, body: { diff --git a/src/features/background-agent/manager.test.ts b/src/features/background-agent/manager.test.ts index 6c076d963..385f85da6 100644 --- a/src/features/background-agent/manager.test.ts +++ b/src/features/background-agent/manager.test.ts @@ -12,6 +12,7 @@ import type { BackgroundTask, ResumeInput } from "./types" import { MIN_IDLE_TIME_MS } from "./constants" import { BackgroundManager } from "./manager" import { ConcurrencyManager } from "./concurrency" +import { promptAsyncAfterSessionIdle } from "../../shared/prompt-async-gate" import { initTaskToastManager, _resetTaskToastManagerForTesting } from "../task-toast-manager/manager" import { _resetForTesting as resetProcessCleanupState } from "./process-cleanup" @@ -27,6 +28,11 @@ mock.restore() const TASK_TTL_MS = 30 * 60 * 1000 +type PendingParentWakeForTest = { + promptContext: Record + notifications: string[] + shouldReply: boolean +} class MockBackgroundManager { private tasks: Map = new Map() @@ -235,6 +241,14 @@ function getPendingNotifications(manager: BackgroundManager): Map }>(manager)).pendingNotifications } +function getPendingParentWakes(manager: BackgroundManager): Map { + return (cast<{ pendingParentWakes: Map }>(manager)).pendingParentWakes +} + +function getDispatchedParentWakes(manager: BackgroundManager): Map { + return (cast<{ dispatchedParentWakes: Map }>(manager)).dispatchedParentWakes +} + function getCompletionTimers(manager: BackgroundManager): Map> { return (cast<{ completionTimers: Map> }>(manager)).completionTimers } @@ -2208,6 +2222,123 @@ describe("BackgroundManager.resume concurrency key", () => { }) }) +describe("BackgroundManager.resume promptAsync gate state", () => { + test("restores completed task state when resume prompt is skipped because the session is active", async () => { + //#given + let promptCallCount = 0 + const client = { + session: { + status: async () => ({ data: { "session-active-resume": { type: "busy" } } }), + promptAsync: async () => { + promptCallCount += 1 + return {} + }, + abort: async () => ({}), + }, + } + const manager = new BackgroundManager({ pluginContext: createPluginInput(client) }) + const task: BackgroundTask = { + id: "task-active-resume-skip", + sessionId: "session-active-resume", + parentSessionId: "parent-session-original", + parentMessageId: "msg-original", + description: "completed task", + prompt: "original prompt", + agent: "explore", + status: "completed", + startedAt: new Date(Date.now() - 1000), + completedAt: new Date(), + error: "previous terminal note", + concurrencyGroup: "explore", + } + const originalCompletedAt = task.completedAt + getTaskMap(manager).set(task.id, task) + + //#when + await manager.resume({ + sessionId: "session-active-resume", + prompt: "continue", + parentSessionId: "parent-session-new", + parentMessageId: "msg-new", + }) + await flushBackgroundNotifications() + + //#then + expect(promptCallCount).toBe(0) + expect(task.status).toBe("completed") + expect(task.completedAt).toBe(originalCompletedAt) + expect(task.error).toBe("previous terminal note") + expect(task.parentSessionId).toBe("parent-session-original") + expect(task.parentMessageId).toBe("msg-original") + expect(task.concurrencyKey).toBeUndefined() + expect(getConcurrencyManager(manager).getCount("explore")).toBe(0) + expect(getPendingByParent(manager).get("parent-session-new")).toBeUndefined() + + manager.shutdown() + }) + + test("restores completed task state when resume prompt is skipped by an existing reservation", async () => { + //#given + let promptCallCount = 0 + const client = { + session: { + promptAsync: async () => { + promptCallCount += 1 + return {} + }, + abort: async () => ({}), + }, + } + await promptAsyncAfterSessionIdle({ + client, + sessionID: "session-reserved-resume", + source: "test-existing-reservation", + settleMs: 0, + postDispatchHoldMs: 1000, + input: { + path: { id: "session-reserved-resume" }, + body: { parts: [] }, + }, + }) + + const manager = new BackgroundManager({ pluginContext: createPluginInput(client) }) + const task: BackgroundTask = { + id: "task-reserved-resume-skip", + sessionId: "session-reserved-resume", + parentSessionId: "parent-session-original", + parentMessageId: "msg-original", + description: "completed task", + prompt: "original prompt", + agent: "explore", + status: "completed", + startedAt: new Date(Date.now() - 1000), + completedAt: new Date(), + concurrencyGroup: "explore", + } + getTaskMap(manager).set(task.id, task) + + //#when + await manager.resume({ + sessionId: "session-reserved-resume", + prompt: "continue", + parentSessionId: "parent-session-new", + parentMessageId: "msg-new", + }) + await flushBackgroundNotifications() + + //#then + expect(promptCallCount).toBe(1) + expect(task.status).toBe("completed") + expect(task.parentSessionId).toBe("parent-session-original") + expect(task.parentMessageId).toBe("msg-original") + expect(task.concurrencyKey).toBeUndefined() + expect(getConcurrencyManager(manager).getCount("explore")).toBe(0) + expect(getPendingByParent(manager).get("parent-session-new")).toBeUndefined() + + manager.shutdown() + }) +}) + describe("BackgroundManager.resume model persistence", () => { let manager: BackgroundManager let promptCalls: Array<{ path: { id: string }; body: Record }> @@ -4938,6 +5069,105 @@ describe("BackgroundManager.handleEvent - session.error", () => { manager.shutdown() }) + test("terminates task when agent-not-found arrives as async session.error after promptAsync accept", async () => { + //#given + const manager = createBackgroundManager() + mockVerifySessionExists(manager, true) + const concurrencyManager = getConcurrencyManager(manager) + const concurrencyKey = "missing-agent" + await concurrencyManager.acquire(concurrencyKey) + + const task = createMockTask({ + id: "task-session-error-agent-not-found", + sessionId: "ses-agent-not-found", + parentSessionId: "parent-session", + parentMessageId: "msg-agent-not-found", + description: "task with missing agent", + agent: "missing-agent", + status: "running", + concurrencyKey, + }) + getTaskMap(manager).set(task.id, task) + getPendingByParent(manager).set(task.parentSessionId, new Set([task.id])) + + //#when + manager.handleEvent({ + type: "session.error", + properties: { + sessionID: task.sessionId, + error: { + name: "AgentNotFoundError", + message: "Agent not found: missing-agent", + }, + }, + }) + await flushBackgroundNotifications() + + //#then + expect(task.status).toBe("interrupt") + expect(task.error).toBe("Agent \"missing-agent\" not found. Make sure the agent is registered in your opencode.json or provided by a plugin.") + expect(task.completedAt).toBeInstanceOf(Date) + expect(task.concurrencyKey).toBeUndefined() + expect(concurrencyManager.getCount(concurrencyKey)).toBe(0) + expect(getPendingByParent(manager).get(task.parentSessionId)).toBeUndefined() + expect(getCompletionTimers(manager).has(task.id)).toBe(true) + + manager.shutdown() + }) + + test("requeues dispatched parent wake when the wake prompt fails through session.error", async () => { + //#given + const promptCalls: Array<{ path: { id: string }; body: Record }> = [] + const client = { + session: { + status: async () => ({ data: { "parent-session-wake": { type: "idle" } } }), + promptAsync: async (args: { path: { id: string }; body: Record }) => { + promptCalls.push(args) + return {} + }, + abort: async () => ({}), + }, + } + const manager = new BackgroundManager({ pluginContext: createPluginInput(client) }) + const managerInternals = cast<{ + queuePendingParentWake: ( + sessionID: string, + notification: string, + promptContext: Record, + shouldReply: boolean, + delayMs?: number, + ) => void + flushPendingParentWake: (sessionID: string) => Promise + }>(manager) + managerInternals.queuePendingParentWake( + "parent-session-wake", + "done", + { agent: "sisyphus" }, + true, + 0, + ) + + //#when + await managerInternals.flushPendingParentWake("parent-session-wake") + manager.handleEvent({ + type: "session.error", + properties: { + sessionID: "parent-session-wake", + error: { name: "UnknownError", message: "wake prompt failed" }, + }, + }) + await flushBackgroundNotifications() + + //#then + expect(promptCalls).toHaveLength(1) + expect(getDispatchedParentWakes(manager).has("parent-session-wake")).toBe(false) + expect(getPendingParentWakes(manager).get("parent-session-wake")?.notifications).toEqual([ + "done", + ]) + + manager.shutdown() + }) + test("terminates task on session.error when session is gone", async () => { //#given const manager = createBackgroundManager() diff --git a/src/features/background-agent/manager.ts b/src/features/background-agent/manager.ts index fb62dd931..579033be0 100644 --- a/src/features/background-agent/manager.ts +++ b/src/features/background-agent/manager.ts @@ -66,7 +66,7 @@ import { isSessionActive as isOpenCodeSessionActive, settleAfterSessionIdle, } from "../../hooks/shared/session-idle-settle" -import { promptAsyncAfterSessionIdle } from "../../hooks/shared/prompt-async-gate" +import { promptAsyncAfterSessionIdle, type PromptAsyncGateResult } from "../../hooks/shared/prompt-async-gate" import { findNearestMessageExcludingCompaction, resolvePromptContextFromSessionMessages, @@ -110,6 +110,21 @@ type PendingParentWake = { shouldReply: boolean } +type ResumeTaskSnapshot = { + status: BackgroundTask["status"] + completedAt?: Date + error?: string + startedAt?: Date + progress?: BackgroundTask["progress"] + parentSessionId: string + parentMessageId: string + parentModel?: BackgroundTask["parentModel"] + parentAgent?: string + parentTools?: Record + concurrencyKey?: string + concurrencyGroup?: string +} + const PENDING_PARENT_WAKE_RETRY_MS = 1_000 const PENDING_PARENT_WAKE_DEBOUNCE_MS = 100 @@ -196,6 +211,7 @@ export type OnSubagentSessionCreated = (event: SubagentSessionCreatedEvent) => P const MAX_TASK_REMOVAL_RESCHEDULES = 6 const MAX_COMPLETED_TASK_ARCHIVE_SIZE = 100 +const PARENT_WAKE_FAILURE_REQUEUE_WINDOW_MS = 5_000 export interface BackgroundManagerConfig { pluginContext: PluginInput @@ -236,6 +252,8 @@ export class BackgroundManager { private notificationQueueByParent: Map> = new Map() private pendingParentWakes: Map = new Map() private pendingParentWakeTimers: Map> = new Map() + private dispatchedParentWakes: Map = new Map() + private dispatchedParentWakeTimers: Map> = new Map() private observedOutputSessions: Set = new Set() private observedIncompleteTodosBySession: Map = new Map() private rootDescendantCounts: Map @@ -422,6 +440,60 @@ export class BackgroundManager { this.tasksByParentSession.set(parentSessionID, taskIDs) } + private captureResumeTaskSnapshot(task: BackgroundTask): ResumeTaskSnapshot { + return { + status: task.status, + completedAt: task.completedAt, + error: task.error, + startedAt: task.startedAt, + progress: task.progress, + parentSessionId: task.parentSessionId, + parentMessageId: task.parentMessageId, + parentModel: task.parentModel, + parentAgent: task.parentAgent, + parentTools: task.parentTools, + concurrencyKey: task.concurrencyKey, + concurrencyGroup: task.concurrencyGroup, + } + } + + private restoreTaskAfterSkippedResume( + task: BackgroundTask, + snapshot: ResumeTaskSnapshot, + skippedStatus: Exclude, + ): void { + log("[background-agent] Restoring task after skipped resume prompt:", { + taskId: task.id, + sessionID: task.sessionId, + skippedStatus, + }) + + this.cleanupPendingByParent(task) + + if (task.concurrencyKey) { + this.concurrencyManager.release(task.concurrencyKey) + } + + task.status = snapshot.status + task.completedAt = snapshot.completedAt + task.error = snapshot.error + task.startedAt = snapshot.startedAt + task.progress = snapshot.progress + task.parentMessageId = snapshot.parentMessageId + task.parentModel = snapshot.parentModel + task.parentAgent = snapshot.parentAgent + task.parentTools = snapshot.parentTools + task.concurrencyKey = snapshot.concurrencyKey + task.concurrencyGroup = snapshot.concurrencyGroup + this.updateTaskParent(task, snapshot.parentSessionId) + + removeTaskToastTracking(task.id) + if (task.status !== "running" && task.status !== "pending") { + this.scheduleTaskRemoval(task.id) + } + this.updateBackgroundTaskMarker(task.parentSessionId) + } + private removeTaskFromParentIndex(taskID: string, parentSessionID: string | undefined): void { if (!parentSessionID) { return @@ -1083,6 +1155,7 @@ The fallback retry session is now created and can be inspected directly. return existingTask } + const resumeSnapshot = this.captureResumeTaskSnapshot(existingTask) const completionTimer = this.completionTimers.get(existingTask.id) if (completionTimer) { clearTimeout(completionTimer) @@ -1166,7 +1239,6 @@ The fallback retry session is now created and can be inspected directly. sessionID: existingTask.sessionId, source: "background-agent-resume", settleMs: 0, - postDispatchHoldMs: 0, input: { path: { id: existingTask.sessionId }, body: { @@ -1199,6 +1271,7 @@ The fallback retry session is now created and can be inspected directly. sessionID: existingTask.sessionId, status: promptResult.status, }) + this.restoreTaskAfterSkippedResume(existingTask, resumeSnapshot, promptResult.status) } }).catch(async (error) => { log("[background-agent] resume prompt error:", error) @@ -1276,6 +1349,60 @@ The fallback retry session is now created and can be inspected directly. this.observedOutputSessions.add(sessionID) } + private cloneParentWake(wake: PendingParentWake): PendingParentWake { + return { + promptContext: { + ...wake.promptContext, + ...(wake.promptContext.model ? { model: { ...wake.promptContext.model } } : {}), + ...(wake.promptContext.tools ? { tools: { ...wake.promptContext.tools } } : {}), + }, + notifications: [...wake.notifications], + shouldReply: wake.shouldReply, + } + } + + private clearDispatchedParentWake(sessionID: string): void { + const timer = this.dispatchedParentWakeTimers.get(sessionID) + if (timer) { + clearTimeout(timer) + this.dispatchedParentWakeTimers.delete(sessionID) + } + this.dispatchedParentWakes.delete(sessionID) + } + + private trackDispatchedParentWake(sessionID: string, wake: PendingParentWake): void { + this.clearDispatchedParentWake(sessionID) + this.dispatchedParentWakes.set(sessionID, this.cloneParentWake(wake)) + const timer = setTimeout(() => { + this.dispatchedParentWakeTimers.delete(sessionID) + this.dispatchedParentWakes.delete(sessionID) + }, PARENT_WAKE_FAILURE_REQUEUE_WINDOW_MS) + this.dispatchedParentWakeTimers.set(sessionID, timer) + } + + private requeueDispatchedParentWake(sessionID: string, reason: string): boolean { + const wake = this.dispatchedParentWakes.get(sessionID) + if (!wake) { + return false + } + + this.clearDispatchedParentWake(sessionID) + const pendingWake = this.pendingParentWakes.get(sessionID) + if (pendingWake) { + pendingWake.notifications.unshift(...wake.notifications) + pendingWake.shouldReply = pendingWake.shouldReply || wake.shouldReply + pendingWake.promptContext = wake.promptContext + } else { + this.pendingParentWakes.set(sessionID, this.cloneParentWake(wake)) + } + this.schedulePendingParentWakeFlush(sessionID) + log("[background-agent] Requeued dispatched parent wake after prompt failure:", { + sessionID, + reason, + }) + return true + } + private clearSessionOutputObserved(sessionID: string): void { this.observedOutputSessions.delete(sessionID) } @@ -1307,6 +1434,7 @@ The fallback retry session is now created and can be inspected directly. const sessionID = resolveMessageEventSessionID(props) const role = (info as Record)["role"] if (!sessionID) return + this.clearDispatchedParentWake(sessionID) if (role === "tool") { this.markSessionOutputObserved(sessionID) @@ -1339,6 +1467,7 @@ The fallback retry session is now created and can be inspected directly. const partInfo = resolveMessagePartInfo(props) const sessionID = resolveMessageEventSessionID(props) if (!sessionID) return + this.clearDispatchedParentWake(sessionID) const resolved = this.resolveTaskAttemptBySession(sessionID) if (!resolved?.isCurrent) return @@ -1469,7 +1598,10 @@ The fallback retry session is now created and can be inspected directly. if (!sessionID) return const resolved = this.resolveTaskAttemptBySession(sessionID) - if (!resolved?.isCurrent) return + if (!resolved?.isCurrent) { + this.requeueDispatchedParentWake(sessionID, "session.error") + return + } const { task } = resolved if (task.status !== "running") return @@ -1581,6 +1713,67 @@ The fallback retry session is now created and can be inspected directly. } } + private async interruptTaskFromAsyncPromptFailure( + task: BackgroundTask, + errorMessage: string, + reason: string, + ): Promise { + if (task.currentAttemptID) { + finalizeAttempt(task, task.currentAttemptID, "interrupt", errorMessage) + } else { + task.status = "interrupt" + task.error = errorMessage + task.completedAt = new Date() + } + + if (task.rootSessionId) { + this.unregisterRootDescendant(task.rootSessionId) + } + this.taskHistory.record(task.parentSessionId, { + id: task.id, + sessionID: task.sessionId, + agent: task.agent, + description: task.description, + status: "interrupt", + category: task.category, + startedAt: task.startedAt, + completedAt: task.completedAt, + }) + + if (task.concurrencyKey) { + this.concurrencyManager.release(task.concurrencyKey) + task.concurrencyKey = undefined + } + + const completionTimer = this.completionTimers.get(task.id) + if (completionTimer) { + clearTimeout(completionTimer) + this.completionTimers.delete(task.id) + } + + const idleTimer = this.idleDeferralTimers.get(task.id) + if (idleTimer) { + clearTimeout(idleTimer) + this.idleDeferralTimers.delete(task.id) + } + + this.cleanupPendingByParent(task) + this.clearNotificationsForTask(task.id) + removeTaskToastTracking(task.id) + this.scheduleTaskRemoval(task.id) + + if (task.sessionId) { + SessionCategoryRegistry.remove(task.sessionId) + await this.abortSessionWithLogging(task.sessionId, `${reason} cleanup`) + } + + this.updateBackgroundTaskMarker(task.parentSessionId) + this.markForNotification(task) + this.enqueueNotificationForParent(task.parentSessionId, () => this.notifyParentSession(task)).catch(err => { + log("[background-agent] Failed to notify on async prompt failure:", { taskId: task.id, error: err }) + }) + } + private async handleSessionErrorEvent(args: { task: BackgroundTask errorInfo: { name?: string; message?: string } @@ -1596,13 +1789,16 @@ The fallback retry session is now created and can be inspected directly. } } - // Agent-not-found errors are handled by the prompt catch block with agent fallback. - // Do not also trigger model fallback retry — that would race with the agent retry. - if (isAgentNotFoundError({ message: errorInfo.message } as Error)) { - log("[background-agent] Skipping session.error fallback for agent-not-found (handled by prompt catch)", { + if (isAgentNotFoundError({ message: errorInfo.message ?? "" })) { + log("[background-agent] Handling async agent-not-found session.error:", { taskId: task.id, errorMessage: errorInfo.message?.slice(0, 100), }) + await this.interruptTaskFromAsyncPromptFailure( + task, + `Agent "${task.agent}" not found. Make sure the agent is registered in your opencode.json or provided by a plugin.`, + "agent-not-found session.error", + ) return } @@ -2383,6 +2579,7 @@ The task was re-queued on a fallback model after a retryable failure. return } log("[background-agent] Sent deferred parent wake:", { sessionID }) + this.trackDispatchedParentWake(sessionID, latestWake) } catch (error) { this.queuePendingNotification(sessionID, notificationContent) log("[background-agent] Failed to send deferred parent wake:", { sessionID, error }) @@ -2739,6 +2936,11 @@ The task was re-queued on a fallback model after a retryable failure. } this.pendingParentWakeTimers.clear() + for (const timer of this.dispatchedParentWakeTimers.values()) { + clearTimeout(timer) + } + this.dispatchedParentWakeTimers.clear() + for (const sessionID of trackedSessionIDs) { subagentSessions.delete(sessionID) SessionCategoryRegistry.remove(sessionID) @@ -2751,6 +2953,7 @@ The task was re-queued on a fallback model after a retryable failure. this.pendingNotifications.clear() this.pendingByParent.clear() this.pendingParentWakes.clear() + this.dispatchedParentWakes.clear() this.notificationQueueByParent.clear() this.rootDescendantCounts.clear() this.queuesByKey.clear() diff --git a/src/features/team-mode/tools/messaging.test.ts b/src/features/team-mode/tools/messaging.test.ts index 2c069d5bd..3b31e0621 100644 --- a/src/features/team-mode/tools/messaging.test.ts +++ b/src/features/team-mode/tools/messaging.test.ts @@ -21,6 +21,7 @@ import { createRuntimeState, saveRuntimeState } from "../team-state-store/store" import { clearTeamSessionRegistry, registerTeamSession } from "../team-session-registry" import type { Message } from "../types" import { MessageSchema } from "../types" +import { createTeamIdleWakeHint } from "../../../hooks/team-session-events/team-idle-wake-hint" import { createTeamSendMessageTool } from "./messaging" type PromptAsyncCall = { @@ -310,6 +311,70 @@ describe("createTeamSendMessageTool", () => { expect(unread[0]?.body).toBe("ping while busy") }) + test("#given rapid live deliveries to one recipient #when the first prompt just dispatched #then the next message stays unread instead of starting another reply", async () => { + // given + const fixture = await createTeamFixture() + const { client, calls } = createRecordingClient() + const liveTool = createTeamSendMessageTool(fixture.config, client) + + // when + await liveTool.execute({ + teamRunId: fixture.teamRunId, + to: "m2", + body: "first ping", + }, fixture.toolContext(fixture.memberOneSessionId)) + await liveTool.execute({ + teamRunId: fixture.teamRunId, + to: "m2", + body: "second ping", + }, fixture.toolContext(fixture.memberOneSessionId)) + + // then + expect(calls).toHaveLength(1) + expect(calls[0]?.parts[0]?.text).toContain("first ping") + const unread = await listUnreadMessages(fixture.teamRunId, "m2", fixture.config) + expect(unread).toHaveLength(1) + expect(unread[0]?.body).toBe("second ping") + }) + + test("#given live delivery left a rapid message unread #when recipient idle wake fires immediately #then the wake hint does not start a second reply", async () => { + // given + const fixture = await createTeamFixture() + const { client, calls } = createRecordingClient() + const liveTool = createTeamSendMessageTool(fixture.config, client) + + await liveTool.execute({ + teamRunId: fixture.teamRunId, + to: "m2", + body: "first ping", + }, fixture.toolContext(fixture.memberOneSessionId)) + await liveTool.execute({ + teamRunId: fixture.teamRunId, + to: "m2", + body: "second ping", + }, fixture.toolContext(fixture.memberOneSessionId)) + + const wakeHint = createTeamIdleWakeHint({ + directory: resolveBaseDir(fixture.config), + client, + }, fixture.config, { idleSettleMs: 0 }) + + // when + await wakeHint({ + event: { + type: "session.idle", + properties: { sessionID: fixture.memberTwoSessionId }, + }, + }) + + // then + expect(calls).toHaveLength(1) + expect(calls[0]?.parts[0]?.text).toContain("first ping") + const unread = await listUnreadMessages(fixture.teamRunId, "m2", fixture.config) + expect(unread).toHaveLength(1) + expect(unread[0]?.body).toBe("second ping") + }) + test("live delivery pins the recipient's resolved subagent_type and model on promptAsync", async () => { // given const fixture = await createTeamFixture() diff --git a/src/hooks/atlas/index.test.ts b/src/hooks/atlas/index.test.ts index f3f9f5602..576780b8b 100644 --- a/src/hooks/atlas/index.test.ts +++ b/src/hooks/atlas/index.test.ts @@ -1702,7 +1702,7 @@ session_id: ses_untrusted_999 // then - stale idle is consumed, not converted into another scheduled continuation expect(mockInput._promptMock).toHaveBeenCalledTimes(1) - expect(scheduledDelays).toHaveLength(0) + expect(scheduledDelays.filter((delay) => delay >= 5_000)).toHaveLength(0) } finally { globalThis.setTimeout = originalSetTimeout } diff --git a/src/hooks/compaction-context-injector/recovery.ts b/src/hooks/compaction-context-injector/recovery.ts index 7abc8e71c..8b74294ea 100644 --- a/src/hooks/compaction-context-injector/recovery.ts +++ b/src/hooks/compaction-context-injector/recovery.ts @@ -21,7 +21,7 @@ import { import { AGENT_RECOVERY_PROMPT, NO_TEXT_TAIL_THRESHOLD, RECOVERY_COOLDOWN_MS, RECENT_COMPACTION_WINDOW_MS } from "./constants" import type { CompactionContextClient } from "./types" import type { TailMonitorState } from "./tail-monitor" -import { promptAsyncAfterSessionIdle } from "../shared/prompt-async-gate" +import { promptAsyncAfterSessionIdle, releasePromptAsyncReservation } from "../shared/prompt-async-gate" export function createRecoveryLogic( ctx: CompactionContextClient | undefined, @@ -117,6 +117,7 @@ export function createRecoveryLogic( hasTools: !!tools, recoveredPromptConfig, }) + releasePromptAsyncReservation(sessionID, "compaction-context-injector:incomplete-recovery") return false } diff --git a/src/hooks/ralph-loop/continuation-prompt-injector.ts b/src/hooks/ralph-loop/continuation-prompt-injector.ts index 8c75dc99b..62dc150e2 100644 --- a/src/hooks/ralph-loop/continuation-prompt-injector.ts +++ b/src/hooks/ralph-loop/continuation-prompt-injector.ts @@ -22,6 +22,7 @@ type MessageInfo = { export type ContinuationPromptResult = | { status: "dispatched" } + | { status: "deferred"; reason: "active" | "reserved" } | { status: "rejected"; error: Error } function extractPromptAsyncError(response: unknown): unknown | undefined { @@ -141,6 +142,9 @@ export async function injectContinuationPrompt( if (promptResult.status === "failed") { throw promptResult.error } + if (promptResult.status === "active" || promptResult.status === "reserved") { + return { status: "deferred", reason: promptResult.status } + } if (promptResult.status !== "dispatched") { return { status: "rejected", diff --git a/src/hooks/ralph-loop/index.test.ts b/src/hooks/ralph-loop/index.test.ts index b083502aa..98856f0d5 100644 --- a/src/hooks/ralph-loop/index.test.ts +++ b/src/hooks/ralph-loop/index.test.ts @@ -871,6 +871,24 @@ describe("ralph-loop", () => { expect(state?.iteration).toBe(2) }) + test("#given duplicate real idle fires before assistant activity #then loop state is preserved without another prompt", async () => { + // given - active loop + const hook = createRalphLoopHook(createMockPluginInput(), { idleSettleMs: 0 }) + hook.startLoop("session-123", "Build feature", { maxIterations: 5 }) + + // when - duplicate idle events arrive without any intervening activity + await hook.event({ + event: { type: "session.idle", properties: { sessionID: "session-123" } }, + }) + await hook.event({ + event: { type: "session.idle", properties: { sessionID: "session-123" } }, + }) + + // then - the second dispatch is deferred, not treated as loop failure + expect(hook.getState()?.iteration).toBe(2) + expect(promptCalls.length).toBe(1) + }) + test("should handle multiple iterations correctly", async () => { // given - active loop const hook = createRalphLoopHook(createMockPluginInput()) @@ -880,6 +898,9 @@ describe("ralph-loop", () => { await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-123" } }, }) + await hook.event({ + event: { type: "message.part.updated", properties: { sessionID: "session-123" } }, + }) await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-123" } }, }) @@ -1127,6 +1148,9 @@ describe("ralph-loop", () => { await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-A" } }, }) + await hook.event({ + event: { type: "message.part.updated", properties: { sessionID: "session-A" } }, + }) await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-A" } }, }) @@ -1328,6 +1352,7 @@ Original task: Build something` // when - delayed start snapshot resolves after the loop has already advanced resolveInitialMessages?.({ data: mockSessionMessages }) await new Promise((resolve) => setTimeout(resolve, 0)) + await hook.event({ event: { type: "message.part.updated", properties: { sessionID: "session-123" } } }) await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-123" } } }) // then - the late snapshot must not hide the DONE message from verification gating diff --git a/src/hooks/ralph-loop/iteration-continuation.ts b/src/hooks/ralph-loop/iteration-continuation.ts index aeca27a16..6067f6a70 100644 --- a/src/hooks/ralph-loop/iteration-continuation.ts +++ b/src/hooks/ralph-loop/iteration-continuation.ts @@ -18,6 +18,7 @@ type ContinuationOptions = { export type ContinuationResult = | { status: "dispatched"; sessionID: string } + | { status: "dispatch_deferred"; reason: "active" | "reserved" } | { status: "session_creation_rejected" } | { status: "dispatch_rejected"; error: unknown } @@ -48,6 +49,9 @@ export async function continueIteration( apiTimeoutMs: options.apiTimeoutMs, idleSettleMs: options.idleSettleMs, }) + if (promptResult.status === "deferred") { + return { status: "dispatch_deferred", reason: promptResult.reason } + } if (promptResult.status === "rejected") { return { status: "dispatch_rejected", error: promptResult.error } } @@ -77,6 +81,9 @@ export async function continueIteration( apiTimeoutMs: options.apiTimeoutMs, idleSettleMs: options.idleSettleMs, }) + if (promptResult.status === "deferred") { + return { status: "dispatch_deferred", reason: promptResult.reason } + } if (promptResult.status === "rejected") { return { status: "dispatch_rejected", error: promptResult.error } } diff --git a/src/hooks/ralph-loop/ralph-loop-event-handler.ts b/src/hooks/ralph-loop/ralph-loop-event-handler.ts index 02c4c3d83..5f12ec406 100644 --- a/src/hooks/ralph-loop/ralph-loop-event-handler.ts +++ b/src/hooks/ralph-loop/ralph-loop-event-handler.ts @@ -2,6 +2,7 @@ import type { PluginInput } from "@opencode-ai/plugin" import { log } from "../../shared/logger" import { resolveMessageEventSessionID, resolveSessionEventID } from "../../shared/event-session-id" import { isSessionActive } from "../shared/session-idle-settle" +import { releasePromptAsyncReservation } from "../shared/prompt-async-gate" import type { IterationCommitExpectation, RalphLoopOptions, RalphLoopState } from "./types" import { HOOK_NAME } from "./constants" import { handleDetectedCompletion } from "./completion-handler" @@ -196,6 +197,7 @@ export function createRalphLoopEventHandler( const props = event.properties as Record | undefined const runtimeRetryActivitySessionID = getRuntimeRetryActivitySessionID(event.type, props) if (runtimeRetryActivitySessionID) { + releasePromptAsyncReservation(runtimeRetryActivitySessionID, "ralph-loop:activity") runtimeErrorRetriedSessions.delete(runtimeRetryActivitySessionID) recentHandledSyntheticIdleAt.delete(runtimeRetryActivitySessionID) } @@ -396,6 +398,10 @@ export function createRalphLoopEventHandler( } return } + if (result.status === "dispatch_deferred") { + log(`[${HOOK_NAME}] Dispatch deferred`, { sessionID, reason: result.reason }) + return + } log(`[${HOOK_NAME}] Dispatch failed`, { sessionID, status: result.status }) options.loopState.clear() @@ -563,6 +569,10 @@ export function createRalphLoopEventHandler( } return } + if (result.status === "dispatch_deferred") { + log(`[${HOOK_NAME}] Dispatch deferred after runtime error`, { sessionID, reason: result.reason }) + return + } log(`[${HOOK_NAME}] Dispatch failed after runtime error`, { sessionID, status: result.status }) options.loopState.clear() diff --git a/src/hooks/ralph-loop/ralph-loop-hook.ts b/src/hooks/ralph-loop/ralph-loop-hook.ts index 9cadd3434..70923be5e 100644 --- a/src/hooks/ralph-loop/ralph-loop-hook.ts +++ b/src/hooks/ralph-loop/ralph-loop-hook.ts @@ -1,6 +1,7 @@ import type { PluginInput } from "@opencode-ai/plugin" import type { RalphLoopOptions, RalphLoopState } from "./types" import { getTranscriptPath as getDefaultTranscriptPath } from "../claude-code-hooks/transcript" +import { releasePromptAsyncReservation } from "../shared/prompt-async-gate" import { createLoopStateController } from "./loop-state-controller" import { createRalphLoopEventHandler } from "./ralph-loop-event-handler" @@ -69,6 +70,9 @@ export function createRalphLoopHook( event, startLoop: (sessionID, prompt, loopOptions): boolean => { const startSuccess = loopState.startLoop(sessionID, prompt, loopOptions) + if (startSuccess) { + releasePromptAsyncReservation(sessionID, "ralph-loop:start-loop") + } if (!startSuccess || typeof loopOptions?.messageCountAtStart === "number") { return startSuccess } diff --git a/src/hooks/ralph-loop/ulw-loop-verification.test.ts b/src/hooks/ralph-loop/ulw-loop-verification.test.ts index 2c7499b96..cf72554af 100644 --- a/src/hooks/ralph-loop/ulw-loop-verification.test.ts +++ b/src/hooks/ralph-loop/ulw-loop-verification.test.ts @@ -176,10 +176,11 @@ describe("ulw-loop verification", () => { `${JSON.stringify({ type: "assistant", timestamp: new Date().toISOString(), content: "done DONE" })}\n`, ) - await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-123" } } }) - const stateAfterDone = hook.getState() + await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-123" } } }) + const stateAfterDone = hook.getState() - await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-123" } } }) + await hook.event({ event: { type: "message.part.updated", properties: { sessionID: "session-123" } } }) + await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-123" } } }) expect(stateAfterDone?.verification_pending).toBe(true) expect(hook.getState()?.iteration).toBe(2) @@ -208,10 +209,11 @@ describe("ulw-loop verification", () => { writeFileSync( oracleTranscriptPath, `${JSON.stringify({ type: "tool_result", timestamp: new Date().toISOString(), tool_output: { output: "still checking" } })}\n`, - ) - const stateBeforeWait = hook.getState() + ) + const stateBeforeWait = hook.getState() - await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-123" } } }) + await hook.event({ event: { type: "message.part.updated", properties: { sessionID: "session-123" } } }) + await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-123" } } }) expect(stateBeforeWait?.verification_session_id).toBe("ses-oracle") expect(hook.getState()?.iteration).toBe(2) diff --git a/src/hooks/ralph-loop/verification-failure-handler.ts b/src/hooks/ralph-loop/verification-failure-handler.ts index 760ea950c..52874ae98 100644 --- a/src/hooks/ralph-loop/verification-failure-handler.ts +++ b/src/hooks/ralph-loop/verification-failure-handler.ts @@ -1,5 +1,6 @@ import type { PluginInput } from "@opencode-ai/plugin" import { log } from "../../shared/logger" +import { releasePromptAsyncReservation } from "../shared/prompt-async-gate" import { buildVerificationFailurePrompt } from "./continuation-prompt-builder" import { HOOK_NAME } from "./constants" import { injectContinuationPrompt } from "./continuation-prompt-injector" @@ -80,30 +81,29 @@ export async function handleFailedVerification( return false } - if (state.verification_session_id) { - ctx.client.session.abort({ path: { id: state.verification_session_id } }).catch(() => {}) + const previewState: RalphLoopState = { + ...state, + verification_pending: undefined, + verification_session_id: undefined, + message_count_at_start: messageCountAtStart, + iteration: state.iteration + 1, } - const clearedState = loopState.clearVerificationState( - parentSessionID, - messageCountAtStart, - ) - if (!clearedState) { - log(`[${HOOK_NAME}] Failed to restart loop after verification failure`, { - parentSessionID, - }) - return false - } - - const previewState: RalphLoopState = { ...clearedState, iteration: clearedState.iteration + 1 } - try { + releasePromptAsyncReservation(parentSessionID, "ralph-loop:verification-failed") const promptResult = await injectContinuationPrompt(ctx, { sessionID: parentSessionID, prompt: buildVerificationFailurePrompt(previewState), directory, apiTimeoutMs, }) + if (promptResult.status === "deferred") { + log(`[${HOOK_NAME}] Deferred verification failure prompt`, { + parentSessionID, + reason: promptResult.reason, + }) + return false + } if (promptResult.status === "rejected") { log(`[${HOOK_NAME}] Failed to inject verification failure prompt`, { parentSessionID, @@ -133,6 +133,21 @@ export async function handleFailedVerification( return false } + if (state.verification_session_id) { + ctx.client.session.abort({ path: { id: state.verification_session_id } }).catch(() => {}) + } + + const clearedState = loopState.clearVerificationState( + parentSessionID, + messageCountAtStart, + ) + if (!clearedState) { + log(`[${HOOK_NAME}] Failed to restart loop after verification failure`, { + parentSessionID, + }) + return false + } + const committed = loopState.incrementIteration() if (!committed) { log(`[${HOOK_NAME}] Failed to commit iteration after verification restart`, { parentSessionID }) diff --git a/src/hooks/runtime-fallback/auto-retry.ts b/src/hooks/runtime-fallback/auto-retry.ts index 7e8a7fd65..aa875b4ef 100644 --- a/src/hooks/runtime-fallback/auto-retry.ts +++ b/src/hooks/runtime-fallback/auto-retry.ts @@ -147,7 +147,6 @@ export function createAutoRetryHelpers(deps: HookDeps) { sessionID, source: `runtime-fallback:${source}`, settleMs: 0, - postDispatchHoldMs: 0, input: { path: { id: sessionID }, body: { diff --git a/src/hooks/runtime-fallback/message-update-handler.ts b/src/hooks/runtime-fallback/message-update-handler.ts index b054ca5d8..348cf2dee 100644 --- a/src/hooks/runtime-fallback/message-update-handler.ts +++ b/src/hooks/runtime-fallback/message-update-handler.ts @@ -66,14 +66,14 @@ export function createMessageUpdateHandler(deps: HookDeps, helpers: AutoRetryHel } if (sessionID && role === "assistant" && error) { - sessionAwaitingFallbackResult.delete(sessionID) + const wasAwaitingFallbackResult = sessionAwaitingFallbackResult.delete(sessionID) if (sessionRetryInFlight.has(sessionID) && !retrySignal) { log(`[${HOOK_NAME}] message.updated fallback skipped (retry in flight)`, { sessionID }) return } - if (retrySignal && sessionRetryInFlight.has(sessionID) && timeoutEnabled) { - log(`[${HOOK_NAME}] Overriding in-flight retry due to provider auto-retry signal`, { + if (retrySignal && timeoutEnabled && (sessionRetryInFlight.has(sessionID) || wasAwaitingFallbackResult)) { + log(`[${HOOK_NAME}] Overriding active retry due to provider auto-retry signal`, { sessionID, model, }) diff --git a/src/hooks/shared/prompt-async-gate.test.ts b/src/hooks/shared/prompt-async-gate.test.ts index cd943d3b4..a4f5e3e91 100644 --- a/src/hooks/shared/prompt-async-gate.test.ts +++ b/src/hooks/shared/prompt-async-gate.test.ts @@ -118,6 +118,42 @@ describe("promptAsyncAfterSessionIdle", () => { expect(promptCalls).toBe(0) }) + test("#given dispatch hold has expired #when the same session prompts again #then the next promptAsync is accepted", async () => { + // given + let promptCalls = 0 + const client = { + session: { + promptAsync: async () => { + promptCalls += 1 + }, + }, + } + + // when + const first = await promptAsyncAfterSessionIdle({ + client, + sessionID: "ses_expired_hold", + input: { path: { id: "ses_expired_hold" }, body: { parts: [] } }, + source: "test:expired:first", + settleMs: 0, + postDispatchHoldMs: 1, + }) + await new Promise((resolve) => setTimeout(resolve, 5)) + const second = await promptAsyncAfterSessionIdle({ + client, + sessionID: "ses_expired_hold", + input: { path: { id: "ses_expired_hold" }, body: { parts: [] } }, + source: "test:expired:second", + settleMs: 0, + postDispatchHoldMs: 0, + }) + + // then + expect(first.status).toBe("dispatched") + expect(second.status).toBe("dispatched") + expect(promptCalls).toBe(2) + }) + test("#given two internal prompt calls race for one idle session #when they dispatch concurrently #then only one prompt is accepted", async () => { // given let promptCalls = 0 diff --git a/src/plugin/event.ts b/src/plugin/event.ts index 840bbe994..c071caf91 100644 --- a/src/plugin/event.ts +++ b/src/plugin/event.ts @@ -42,7 +42,7 @@ import { createTeamIdleWakeHint } from "../hooks/team-session-events/team-idle-w import { createTeamLeadOrphanHandler } from "../hooks/team-session-events/team-lead-orphan-handler"; import { createTeamMemberErrorHandler } from "../hooks/team-session-events/team-member-error-handler"; import { createTeamMemberStatusHandler } from "../hooks/team-session-events/team-member-status-handler"; -import { promptAfterSessionIdle, promptAsyncAfterSessionIdle } from "../hooks/shared/prompt-async-gate"; +import { promptAfterSessionIdle, promptAsyncAfterSessionIdle, releasePromptAsyncReservation } from "../hooks/shared/prompt-async-gate"; import type { CreatedHooks } from "../create-hooks"; import type { Managers } from "../create-managers"; @@ -469,6 +469,7 @@ export function createEventHandler(args: { await pluginContext.client.session.abort({ path: { id: sessionID } }).catch((error) => { log("[event] model-fallback abort failed", { sessionID, source, error }); }); + releasePromptAsyncReservation(sessionID, `model-fallback-abort:${source}`); const launchAgent = fallbackContext?.agentName ? resolveRegisteredAgentName(fallbackContext.agentName) diff --git a/src/shared/model-suggestion-retry.test.ts b/src/shared/model-suggestion-retry.test.ts index fb2e3248a..019f87e85 100644 --- a/src/shared/model-suggestion-retry.test.ts +++ b/src/shared/model-suggestion-retry.test.ts @@ -266,6 +266,31 @@ describe("promptWithModelSuggestionRetry", () => { expect(results[1]?.status).toBe("rejected") }) + it("#given promptAsync retry just dispatched #when the same session is prompted again immediately #then the second caller is rejected by the gate", async () => { + // given + const promptMock = mock(async () => undefined) + const client = { + session: { + promptAsync: promptMock, + }, + } + const args = { + path: { id: "session-post-dispatch-hold" }, + body: { + parts: [{ type: "text", text: "hello" }], + model: { providerID: "anthropic", modelID: "claude-sonnet-4" }, + }, + } + + // when + await promptWithModelSuggestionRetry(unsafeTestValue(client), args) + const second = promptWithModelSuggestionRetry(unsafeTestValue(client), args) + + // then + await expect(second).rejects.toThrow("promptAsync skipped by gate: reserved") + expect(promptMock).toHaveBeenCalledTimes(1) + }) + it("should throw error from promptAsync directly on model-not-found error", async () => { // given a client that fails with model-not-found error const promptMock = mock().mockRejectedValueOnce({ @@ -436,6 +461,31 @@ describe("promptSyncWithModelSuggestionRetry", () => { expect(promptAsyncMock).toHaveBeenCalledTimes(0) }) + it("#given sync prompt retry just dispatched #when the same session is prompted again immediately #then the second caller is rejected by the gate", async () => { + // given + const promptMock = mock(async () => undefined) + const client = { + session: { + prompt: promptMock, + }, + } + const args = { + path: { id: "session-sync-post-dispatch-hold" }, + body: { + parts: [{ type: "text", text: "hello" }], + model: { providerID: "anthropic", modelID: "claude-sonnet-4" }, + }, + } + + // when + await promptSyncWithModelSuggestionRetry(unsafeTestValue(client), args) + const second = promptSyncWithModelSuggestionRetry(unsafeTestValue(client), args) + + // then + await expect(second).rejects.toThrow("prompt skipped by gate: reserved") + expect(promptMock).toHaveBeenCalledTimes(1) + }) + it("should abort and throw timeout error when sync prompt hangs", async () => { // given a client where sync prompt never resolves unless aborted let receivedSignal: AbortSignal | undefined diff --git a/src/shared/model-suggestion-retry.ts b/src/shared/model-suggestion-retry.ts index 184467e4d..ab0ab5365 100644 --- a/src/shared/model-suggestion-retry.ts +++ b/src/shared/model-suggestion-retry.ts @@ -105,7 +105,6 @@ export async function promptWithModelSuggestionRetry( } as Parameters[0], source: "model-suggestion-retry", settleMs: 0, - postDispatchHoldMs: 0, }) if (promptResult.status === "failed") { throw promptResult.error @@ -145,7 +144,6 @@ export async function promptSyncWithModelSuggestionRetry( } as Parameters[0], source: "model-suggestion-retry:sync", settleMs: 0, - postDispatchHoldMs: 0, checkStatus: false, }) if (promptResult.status === "failed") { @@ -198,7 +196,6 @@ export async function promptSyncWithModelSuggestionRetry( } as Parameters[0], source: "model-suggestion-retry:sync-retry", settleMs: 0, - postDispatchHoldMs: 0, checkStatus: false, }) if (promptResult.status === "failed") { diff --git a/src/shared/prompt-async-gate.ts b/src/shared/prompt-async-gate.ts index 7f2f57553..ca1bac8a6 100644 --- a/src/shared/prompt-async-gate.ts +++ b/src/shared/prompt-async-gate.ts @@ -33,6 +33,7 @@ type PromptAsyncReservation = { source: string reservedAt: number token: symbol + expiresAt?: number } export type PromptAsyncGateResult = @@ -44,6 +45,23 @@ export type PromptAsyncGateResult = const promptAsyncReservations = new Map() +function pruneExpiredReservations(now = Date.now()): void { + for (const [sessionID, reservation] of promptAsyncReservations) { + if (typeof reservation.expiresAt === "number" && reservation.expiresAt <= now) { + promptAsyncReservations.delete(sessionID) + log("[prompt-async-gate] expired reservation released", { + sessionID, + source: reservation.source, + }) + } + } +} + +function getActiveReservation(sessionID: string): PromptAsyncReservation | undefined { + pruneExpiredReservations() + return promptAsyncReservations.get(sessionID) +} + export async function promptAsyncAfterSessionIdle(args: { client: PromptAsyncClient sessionID: string @@ -67,7 +85,7 @@ export async function promptAsyncAfterSessionIdle(arg return { status: "unavailable" } } - const existing = promptAsyncReservations.get(sessionID) + const existing = getActiveReservation(sessionID) if (existing) { log("[prompt-async-gate] promptAsync skipped because session is reserved", { sessionID, @@ -84,6 +102,7 @@ export async function promptAsyncAfterSessionIdle(arg token: Symbol(source), } promptAsyncReservations.set(sessionID, reservation) + let holdReservationAfterDispatch = false try { const canReadStatus = args.checkStatus !== false && typeof client.session?.status === "function" @@ -99,7 +118,7 @@ export async function promptAsyncAfterSessionIdle(arg log("[prompt-async-gate] promptAsync dispatching", { sessionID, source }) const response = await client.session.promptAsync(input) if (postDispatchHoldMs > 0) { - await settleAfterSessionIdle(postDispatchHoldMs) + holdReservationAfterDispatch = true } log("[prompt-async-gate] promptAsync dispatched", { sessionID, source }) return { status: "dispatched", response } @@ -109,7 +128,11 @@ export async function promptAsyncAfterSessionIdle(arg } finally { const current = promptAsyncReservations.get(sessionID) if (current?.token === reservation.token) { - promptAsyncReservations.delete(sessionID) + if (holdReservationAfterDispatch && postDispatchHoldMs > 0) { + reservation.expiresAt = Date.now() + postDispatchHoldMs + } else { + promptAsyncReservations.delete(sessionID) + } } } } @@ -137,7 +160,7 @@ export async function promptAfterSessionIdle(args: { return { status: "unavailable" } } - const existing = promptAsyncReservations.get(sessionID) + const existing = getActiveReservation(sessionID) if (existing) { log("[prompt-async-gate] prompt skipped because session is reserved", { sessionID, @@ -154,6 +177,7 @@ export async function promptAfterSessionIdle(args: { token: Symbol(source), } promptAsyncReservations.set(sessionID, reservation) + let holdReservationAfterDispatch = false try { const canReadStatus = args.checkStatus !== false && typeof client.session?.status === "function" @@ -169,7 +193,7 @@ export async function promptAfterSessionIdle(args: { log("[prompt-async-gate] prompt dispatching", { sessionID, source }) const response = await client.session.prompt(input) if (postDispatchHoldMs > 0) { - await settleAfterSessionIdle(postDispatchHoldMs) + holdReservationAfterDispatch = true } log("[prompt-async-gate] prompt dispatched", { sessionID, source }) return { status: "dispatched", response } @@ -179,7 +203,11 @@ export async function promptAfterSessionIdle(args: { } finally { const current = promptAsyncReservations.get(sessionID) if (current?.token === reservation.token) { - promptAsyncReservations.delete(sessionID) + if (holdReservationAfterDispatch && postDispatchHoldMs > 0) { + reservation.expiresAt = Date.now() + postDispatchHoldMs + } else { + promptAsyncReservations.delete(sessionID) + } } } } diff --git a/src/shared/prompt-async-route-audit.test.ts b/src/shared/prompt-async-route-audit.test.ts new file mode 100644 index 000000000..0335b7c47 --- /dev/null +++ b/src/shared/prompt-async-route-audit.test.ts @@ -0,0 +1,78 @@ +import { describe, expect, test } from "bun:test" +import { readdir, readFile } from "node:fs/promises" +import path from "node:path" + +const SOURCE_ROOT = path.resolve(import.meta.dir, "..") +const PROMPT_GATE_FILE = path.join(SOURCE_ROOT, "shared", "prompt-async-gate.ts") + +async function listSourceFiles(directory: string): Promise { + const entries = await readdir(directory, { withFileTypes: true }) + const nestedFiles = await Promise.all(entries.map(async (entry) => { + const entryPath = path.join(directory, entry.name) + if (entry.isDirectory()) { + return listSourceFiles(entryPath) + } + if ( + entry.isFile() + && entry.name.endsWith(".ts") + && !entry.name.endsWith(".test.ts") + && !entry.name.endsWith(".d.ts") + ) { + return [entryPath] + } + return [] + })) + + return nestedFiles.flat() +} + +function relativeSourcePath(filePath: string): string { + return path.relative(SOURCE_ROOT, filePath) +} + +function uncommentedLines(contents: string): string[] { + return contents + .split("\n") + .map((line) => line.trimStart()) + .filter((line) => !line.startsWith("//") && !line.startsWith("*")) +} + +describe("production prompt injection routes", () => { + test("#given production TypeScript sources #when prompt routes are audited #then only the shared gate may call raw OpenCode prompt APIs", async () => { + // given + const files = await listSourceFiles(SOURCE_ROOT) + const offenders: string[] = [] + + // when + for (const filePath of files) { + if (filePath === PROMPT_GATE_FILE) { + continue + } + + const contents = uncommentedLines(await readFile(filePath, "utf8")).join("\n") + if (/\bsession\.promptAsync\s*\(/.test(contents) || /\bsession\.prompt\s*\(/.test(contents)) { + offenders.push(relativeSourcePath(filePath)) + } + } + + // then + expect(offenders).toEqual([]) + }) + + test("#given production TypeScript sources #when prompt gate callers are audited #then callers cannot disable the post-dispatch reservation hold", async () => { + // given + const files = await listSourceFiles(SOURCE_ROOT) + const offenders: string[] = [] + + // when + for (const filePath of files) { + const contents = uncommentedLines(await readFile(filePath, "utf8")).join("\n") + if (/postDispatchHoldMs\s*:\s*0\b/.test(contents)) { + offenders.push(relativeSourcePath(filePath)) + } + } + + // then + expect(offenders).toEqual([]) + }) +}) diff --git a/src/shared/session-route.test.ts b/src/shared/session-route.test.ts new file mode 100644 index 000000000..e4e0eae15 --- /dev/null +++ b/src/shared/session-route.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, mock, test } from "bun:test" + +import { unsafeTestValue } from "../../test-support/unsafe-test-value" +import { promptAsyncInDirectory } from "./session-route" + +describe("promptAsyncInDirectory", () => { + test("#given no session id is present #when routing a promptAsync request #then the helper rejects instead of using an ungated raw prompt", async () => { + // given + const promptAsync = mock(async () => ({ data: "sent" })) + const client = { + session: { + promptAsync, + }, + } + const args = { + body: { parts: [{ type: "text", text: "continue" }] }, + } + + // when, then + await expect( + promptAsyncInDirectory( + unsafeTestValue(client), + unsafeTestValue(args), + "/workspace/project", + ), + ).rejects.toThrow("session id is required for routed promptAsync") + expect(promptAsync).toHaveBeenCalledTimes(0) + }) + + test("#given a routed prompt just dispatched #when the same session is prompted again immediately #then the route keeps the session reserved", async () => { + // given + const promptAsync = mock(async () => ({ data: "sent" })) + const client = { + session: { + promptAsync, + }, + } + const args = { + path: { id: "ses_route_hold" }, + body: { parts: [{ type: "text", text: "continue" }] }, + } + + // when + const first = await promptAsyncInDirectory( + unsafeTestValue(client), + unsafeTestValue(args), + "/workspace/project", + ) + const second = promptAsyncInDirectory( + unsafeTestValue(client), + unsafeTestValue(args), + "/workspace/project", + ) + + // then + expect(first).toEqual({ data: "sent" }) + await expect(second).rejects.toThrow("promptAsync skipped by gate: reserved") + expect(promptAsync).toHaveBeenCalledTimes(1) + expect(promptAsync.mock.calls[0]?.[0].query).toEqual({ directory: "/workspace/project" }) + }) +}) diff --git a/src/shared/session-route.ts b/src/shared/session-route.ts index e6e5428dc..3a39277d6 100644 --- a/src/shared/session-route.ts +++ b/src/shared/session-route.ts @@ -56,7 +56,7 @@ export function promptAsyncInDirectory( const routedArgs = routeSessionPrompt(args, directory) const sessionID = routedArgs.path?.id if (!sessionID) { - return client.session.promptAsync(routedArgs) + return Promise.reject(new Error("session id is required for routed promptAsync")) } return promptAsyncAfterSessionIdle({ @@ -65,7 +65,6 @@ export function promptAsyncInDirectory( input: routedArgs, source: "session-route", settleMs: 0, - postDispatchHoldMs: 0, }).then((result) => { if (result.status === "failed") { throw result.error diff --git a/src/tools/call-omo-agent/sync-executor.test.ts b/src/tools/call-omo-agent/sync-executor.test.ts index c0bb0a8d3..fdd6cff0b 100644 --- a/src/tools/call-omo-agent/sync-executor.test.ts +++ b/src/tools/call-omo-agent/sync-executor.test.ts @@ -389,6 +389,35 @@ describe("executeSync", () => { expect(deps.processMessages).not.toHaveBeenCalled() }) + test("#given a reused sync session was just prompted #when executeSync is called again immediately #then the second prompt is rejected by the shared gate", async () => { + //#given + const executeSync = await importExecuteSync() + const deps = createDependencies({ + createOrGetSession: mock(async () => ({ sessionID: "ses-reused-hold", isNew: false })), + }) + const toolContext = createToolContext() + const recorder = createPromptAsyncRecorder() + const args = { + subagent_type: "explore", + description: "reused hold", + prompt: "find something", + run_in_background: false, + session_id: "ses-reused-hold", + } + const context = createContext(recorder.promptAsync) as never + + //#when + const first = await executeSync(args, toolContext, context, deps) + const second = await executeSync(args, toolContext, context, deps) + + //#then + expect(first).toContain("agent response") + expect(second).toContain("promptAsync skipped by gate: reserved") + expect(recorder.promptAsync).toHaveBeenCalledTimes(1) + expect(deps.waitForCompletion).toHaveBeenCalledTimes(1) + expect(deps.processMessages).toHaveBeenCalledTimes(1) + }) + test("commits reserved descendant quota after creating a new sync session", async () => { //#given const { executeSync } = require("./sync-executor") diff --git a/src/tools/call-omo-agent/sync-executor.ts b/src/tools/call-omo-agent/sync-executor.ts index 85c0c6213..640bdb15a 100644 --- a/src/tools/call-omo-agent/sync-executor.ts +++ b/src/tools/call-omo-agent/sync-executor.ts @@ -116,7 +116,6 @@ export async function executeSync( sessionID, source: "call-omo-agent:sync", settleMs: 0, - postDispatchHoldMs: 0, input: { path: { id: sessionID }, body: { diff --git a/test-setup.ts b/test-setup.ts index ccdfb0807..c8e8f8d42 100644 --- a/test-setup.ts +++ b/test-setup.ts @@ -6,6 +6,7 @@ import { _resetTaskToastManagerForTesting as resetTaskToastManager } from "./src import { _resetForTesting as resetModelFallbackState } from "./src/hooks/model-fallback/hook" import { _resetMemCacheForTesting as resetConnectedProvidersCache } from "./src/shared/connected-providers-cache" import { getOmoOpenCodeCacheDir } from "./src/shared/data-path" +import { releaseAllPromptAsyncReservationsForTesting } from "./src/shared/prompt-async-gate" import { installModuleMockLifecycle } from "./src/testing/module-mock-lifecycle" const { restoreModuleMocks } = installModuleMockLifecycle(mock) @@ -25,6 +26,7 @@ beforeEach(() => { resetTaskToastManager() resetModelFallbackState() resetConnectedProvidersCache() + releaseAllPromptAsyncReservationsForTesting() }) afterEach(() => { @@ -53,6 +55,7 @@ afterEach(() => { cleanupOmoCacheDir(getOmoOpenCodeCacheDir()) resetTaskToastManager() resetConnectedProvidersCache() + releaseAllPromptAsyncReservationsForTesting() mock.restore() restoreModuleMocks() })