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 diff --git a/src/cli/run/runner.ts b/src/cli/run/runner.ts index 75e6e49da..81763668f 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,30 @@ 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, + 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/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 2a58b67ce..579033be0 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, 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) @@ -1161,27 +1234,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, + 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, + }) + this.restoreTaskAfterSkippedResume(existingTask, resumeSnapshot, promptResult.status) + } + }).catch(async (error) => { log("[background-agent] resume prompt error:", error) const errorInfo = { name: extractErrorName(error), @@ -1257,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) } @@ -1288,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) @@ -1320,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 @@ -1450,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 @@ -1562,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 } @@ -1577,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 } @@ -2328,15 +2543,43 @@ 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 }) + this.trackDispatchedParentWake(sessionID, latestWake) } catch (error) { this.queuePendingNotification(sessionID, notificationContent) log("[background-agent] Failed to send deferred parent wake:", { sessionID, error }) @@ -2693,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) @@ -2705,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/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) diff --git a/src/features/team-mode/tools/messaging.test.ts b/src/features/team-mode/tools/messaging.test.ts index 0faf67771..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 = { @@ -282,6 +283,98 @@ 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("#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/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/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/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, }) } 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/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/hooks/compaction-context-injector/recovery.ts b/src/hooks/compaction-context-injector/recovery.ts index ed69bf0e0..8b74294ea 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, releasePromptAsyncReservation } 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)) { @@ -103,6 +117,7 @@ export function createRecoveryLogic( hasTools: !!tools, recoveredPromptConfig, }) + releasePromptAsyncReservation(sessionID, "compaction-context-injector:incomplete-recovery") return false } 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 diff --git a/src/hooks/ralph-loop/continuation-prompt-injector.ts b/src/hooks/ralph-loop/continuation-prompt-injector.ts index dec80c5c6..62dc150e2 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 @@ -21,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 { @@ -66,6 +68,7 @@ export async function injectContinuationPrompt( directory: string apiTimeoutMs: number inheritFromSessionID?: string + idleSettleMs?: number }, ): Promise { let agent: string | undefined @@ -119,17 +122,36 @@ 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 === "active" || promptResult.status === "reserved") { + return { status: "deferred", reason: promptResult.status } + } + 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/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 fd3d2741d..6067f6a70 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 @@ -17,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 } @@ -45,7 +47,11 @@ export async function continueIteration( prompt: continuationPrompt, directory: options.directory, 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 } } @@ -73,7 +79,11 @@ export async function continueIteration( prompt: continuationPrompt, directory: options.directory, 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 0e832a9ef..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) } @@ -358,6 +360,7 @@ export function createRalphLoopEventHandler( previousSessionID: sessionID, directory: options.directory, apiTimeoutMs: options.apiTimeoutMs, + idleSettleMs: options.idleSettleMs, loopState: options.loopState, }) @@ -395,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() @@ -523,6 +530,7 @@ export function createRalphLoopEventHandler( previousSessionID: sessionID, directory: options.directory, apiTimeoutMs: options.apiTimeoutMs, + idleSettleMs: options.idleSettleMs, loopState: options.loopState, }) @@ -561,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 cbb3be2be..aa875b4ef 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,31 @@ 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, + 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", () => { 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/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 } 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..a4f5e3e91 --- /dev/null +++ b/src/hooks/shared/prompt-async-gate.test.ts @@ -0,0 +1,235 @@ +import { afterEach, describe, expect, test } from "bun:test" + +import { + promptAfterSessionIdle, + 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 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 + 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) + }) + + 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 + 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) + }) + + 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/hooks/shared/prompt-async-gate.ts b/src/hooks/shared/prompt-async-gate.ts new file mode 100644 index 000000000..68d44ab1b --- /dev/null +++ b/src/hooks/shared/prompt-async-gate.ts @@ -0,0 +1 @@ +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/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, 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) { 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/event.ts b/src/plugin/event.ts index 976c732ab..c071caf91 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 { promptAfterSessionIdle, promptAsyncAfterSessionIdle, releasePromptAsyncReservation } 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) @@ -467,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) @@ -495,19 +498,36 @@ 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; } - 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); @@ -898,13 +918,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.) 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 + } }, }, }, diff --git a/src/shared/model-suggestion-retry.test.ts b/src/shared/model-suggestion-retry.test.ts index e594ad6a3..019f87e85 100644 --- a/src/shared/model-suggestion-retry.test.ts +++ b/src/shared/model-suggestion-retry.test.ts @@ -230,6 +230,67 @@ 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("#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({ @@ -400,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 7047b8bb5..ab0ab5365 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,24 @@ 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, + }) + 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 +135,23 @@ 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, + 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 +187,23 @@ 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, + 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..ca1bac8a6 --- /dev/null +++ b/src/shared/prompt-async-gate.ts @@ -0,0 +1,231 @@ +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 + expiresAt?: number +} + +export type PromptAsyncGateResult = + | { status: "dispatched"; response: unknown } + | { status: "active" } + | { status: "reserved"; reservedBy: string } + | { status: "unavailable" } + | { status: "failed"; error: unknown } + +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 + 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 ?? DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS + + if (typeof client.session?.promptAsync !== "function") { + log("[prompt-async-gate] promptAsync unavailable", { sessionID, source }) + return { status: "unavailable" } + } + + const existing = getActiveReservation(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) + let holdReservationAfterDispatch = false + + 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 (postDispatchHoldMs > 0) { + holdReservationAfterDispatch = true + } + 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) { + if (holdReservationAfterDispatch && postDispatchHoldMs > 0) { + reservation.expiresAt = Date.now() + postDispatchHoldMs + } else { + 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 ?? DEFAULT_PROMPT_ASYNC_POST_DISPATCH_HOLD_MS + + if (typeof client.session?.prompt !== "function") { + log("[prompt-async-gate] prompt unavailable", { sessionID, source }) + return { status: "unavailable" } + } + + const existing = getActiveReservation(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) + let holdReservationAfterDispatch = false + + 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 (postDispatchHoldMs > 0) { + holdReservationAfterDispatch = true + } + 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) { + if (holdReservationAfterDispatch && postDispatchHoldMs > 0) { + reservation.expiresAt = Date.now() + postDispatchHoldMs + } else { + 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/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-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.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 53901cf23..3a39277d6 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,27 @@ 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 Promise.reject(new Error("session id is required for routed promptAsync")) + } + + return promptAsyncAfterSessionIdle({ + client, + sessionID, + input: routedArgs, + source: "session-route", + settleMs: 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( diff --git a/src/tools/call-omo-agent/sync-executor.test.ts b/src/tools/call-omo-agent/sync-executor.test.ts index 9b8392b93..fdd6cff0b 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,70 @@ 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("#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 31ae8beb8..640bdb15a 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,33 @@ 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, + 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) 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() })