luna/oh-my-opencode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(prompt-gate): harden sync and team prompt dispatch

Browse Source
This commit is contained in:
YeonGyu-Kim
2026-05-19 17:43:37 +09:00
parent 1492bffd20
commit bcea4a9d28
44 changed files with 688 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/features/background-agent/manager.test.ts
+1
View File
@@ -5704,6 +5704,7 @@ describe("BackgroundManager.handleEvent - session.error", () => {
{
info: {
role: "assistant",
finish: "end_turn",
time: { created: 2_000 },
},
parts: [{ type: "text", text: "wake was already accepted" }],
src/features/background-agent/parent-wake-notifier.ts
+1
View File
@@ -197,6 +197,7 @@ export class ParentWakeNotifier {
source: "background-agent-parent-wake",
settleMs: 0,
postDispatchHoldMs: 250,
queueBehavior: "defer",
checkToolState: !toolWaitDecision.skipPromptGateToolStateCheck,
input: {
path: { id: sessionID },
src/features/background-agent/parent-wake-user-message-race.test.ts
+47
View File
@@ -220,6 +220,53 @@ describe("ParentWakeNotifier — user message race guard (issue #4120)", () => {
releaseAllPromptAsyncReservationsForTesting()
})
test("#given all-complete wake arrives while prior assistant turn is still streaming #when the parent status is stale-idle #then the wake stays pending", async () => {
// given
const sessionMessages: SessionMessageStub[] = [
{
info: {
role: "user",
time: { created: Date.now() - 20_000 },
},
parts: [{ type: "text", text: "start work" }],
},
{
info: {
role: "assistant",
time: { created: Date.now() - 5_000 },
},
parts: [{ type: "reasoning", text: "still gathering background results" }],
},
{
info: {
role: "user",
time: { created: Date.now() - 4_000 },
},
parts: [{ type: "text", text: "partial wake\n<!-- OMO_INTERNAL_INITIATOR -->" }],
},
]
const { notifier, promptAsyncCalls } = createNotifier({
sessionStatuses: { "parent-stale-idle": { type: "idle" } },
sessionMessages,
})
notifier.queuePendingParentWake(
"parent-stale-idle",
"<system-reminder>\n[ALL BACKGROUND TASKS COMPLETE]\n</system-reminder>",
{ agent: "sisyphus" },
true,
)
// when
await notifier.flushPendingParentWake("parent-stale-idle")
// then
expect(promptAsyncCalls).toHaveLength(0)
expect(notifier.getPendingParentWakes().has("parent-stale-idle")).toBe(true)
notifier.shutdown()
releaseAllPromptAsyncReservationsForTesting()
})
test("#given latest message is a user message just added #when flushing pending wake #then dispatch is deferred (no promptAsync)", async () => {
// given
const { notifier, promptAsyncCalls } = createNotifier({
src/features/background-agent/session-route.test.ts
+74
View File
@@ -0,0 +1,74 @@
import { afterEach, describe, expect, mock, test } from "bun:test"
import { releaseAllPromptAsyncReservationsForTesting } from "../../shared/prompt-async-gate"
import { unsafeTestValue } from "../../../test-support/unsafe-test-value"
import { promptAsyncInDirectory, promptWithRetryInDirectory } from "./session-route"
describe("background-agent session routing", () => {
afterEach(() => {
releaseAllPromptAsyncReservationsForTesting()
})
test("#given a routed prompt just dispatched #when the same child session is prompted again immediately #then promptAsync routing defers instead of enqueueing", async () => {
// given
const promptAsync = mock(async () => ({ data: "sent" }))
const client = {
session: {
promptAsync,
},
}
const args = {
path: { id: "ses_background_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" })
})
test("#given a background retry prompt just dispatched #when the same child session is prompted again immediately #then retry routing defers instead of enqueueing", async () => {
// given
const promptAsync = mock(async () => undefined)
const client = {
session: {
promptAsync,
},
}
const args = {
path: { id: "ses_background_retry_route_hold" },
body: { parts: [{ type: "text", text: "continue" }] },
}
// when
await promptWithRetryInDirectory(
unsafeTestValue(client),
unsafeTestValue(args),
"/workspace/project",
)
const second = promptWithRetryInDirectory(
unsafeTestValue(client),
unsafeTestValue(args),
"/workspace/project",
)
// then
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" })
})
})
src/features/background-agent/session-route.ts
+2 -1
View File
@@ -42,6 +42,7 @@ export function promptAsyncInDirectory(
input: routedArgs,
source: "background-agent-session-route",
settleMs: 0,
queueBehavior: "defer",
}).then((result) => {
if (result.status === "failed") {
throw result.error
@@ -58,7 +59,7 @@ export function promptWithRetryInDirectory(
args: PromptRetryArgs,
directory: string,
): Promise<void> {
return promptWithModelSuggestionRetry(client, routePromptRetry(args, directory))
return promptWithModelSuggestionRetry(client, routePromptRetry(args, directory), { queueBehavior: "defer" })
}
export function messagesInDirectory(
src/features/team-mode/integration.test.ts
+6
View File
@@ -230,6 +230,12 @@ describe("team-mode integration", () => {
if (!leadMember?.sessionId || !workerMember?.sessionId) {
throw new Error("expected both team members to hold sessionIds")
}
await saveRuntimeState({
...runtime,
members: runtime.members.map((member) => (
member.name === "worker" ? { ...member, status: "idle" } : member
)),
}, config)
const { createTeamSendMessageTool } = await import("./tools/messaging")
const tool = createTeamSendMessageTool(config, recordingClient)
src/features/team-mode/tools/messaging.test.ts
+17 -10
View File
@@ -257,7 +257,7 @@ describe("createTeamSendMessageTool", () => {
expect(calls[0]?.directory).toBe("/tmp/team-worker-m2")
})
test("live-delivers to running recipients so active teammates receive messages immediately", async () => {
test("#given runtime marks recipient running #when team_send_message sends a peer message #then it leaves the unread mailbox path without promptAsync", async () => {
// given
const fixture = await createTeamFixture()
const { loadRuntimeState: loadState, saveRuntimeState: saveState } = await import("../team-state-store/store")
@@ -280,12 +280,16 @@ describe("createTeamSendMessageTool", () => {
// then
expect(parsedResult.deliveredTo).toEqual(["m2"])
expect(calls).toHaveLength(1)
expect(calls[0]?.sessionId).toBe(fixture.memberTwoSessionId)
expect(calls[0]?.directory).toBe(resolveBaseDir(fixture.config))
expect(calls).toHaveLength(0)
const unread = await listUnreadMessages(fixture.teamRunId, "m2", fixture.config)
expect(unread).toHaveLength(1)
expect(unread[0]?.body).toBe("ping")
const inboxEntries = await readdir(getInboxDir(resolveBaseDir(fixture.config), fixture.teamRunId, "m2"))
expect(inboxEntries.filter((entry) => entry.endsWith(".json") && !entry.startsWith("."))).toHaveLength(1)
expect(inboxEntries.some((entry) => entry.startsWith(".delivering-"))).toBe(false)
})
test("#given recipient OpenCode session is busy #when team_send_message attempts live delivery #then it reserves the message in the central prompt queue", async () => {
test("#given recipient OpenCode session is busy #when team_send_message attempts live delivery #then it releases the message for later mailbox injection", async () => {
// given
const fixture = await createTeamFixture()
let promptCalls = 0
@@ -309,10 +313,11 @@ describe("createTeamSendMessageTool", () => {
// then
expect(promptCalls).toBe(0)
const unread = await listUnreadMessages(fixture.teamRunId, "m2", fixture.config)
expect(unread).toHaveLength(0)
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 is queued instead of starting another reply", async () => {
test("#given rapid live deliveries to one recipient #when the first prompt just dispatched #then the next message waits for mailbox injection", async () => {
// given
const fixture = await createTeamFixture()
const { client, calls } = createRecordingClient()
@@ -334,10 +339,11 @@ describe("createTeamSendMessageTool", () => {
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(0)
expect(unread).toHaveLength(1)
expect(unread[0]?.body).toBe("second ping")
})
test("#given live delivery queued a rapid message #when recipient idle wake fires immediately #then the wake hint does not start a second reply", async () => {
test("#given live delivery deferred a rapid message #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()
@@ -371,7 +377,8 @@ describe("createTeamSendMessageTool", () => {
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(0)
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 () => {
src/features/team-mode/tools/messaging.ts
+36 -2
View File
@@ -16,7 +16,7 @@ import {
import { BroadcastNotPermittedError, sendMessage } from "../team-mailbox/send"
import { lookupTeamSession } from "../team-session-registry"
import { loadRuntimeState, transitionRuntimeState } from "../team-state-store/store"
import type { Message } from "../types"
import type { Message, RuntimeState } from "../types"
import { MessageSchema } from "../types"
const MESSAGE_TOOL_KINDS = ["message", "announcement"] as const
@@ -68,6 +68,14 @@ const TeamSendMessageArgsSchema = z.object({
})
type DeliveryReservation = Awaited<ReturnType<typeof reserveMessageForDelivery>>
type RuntimeMember = RuntimeState["members"][number]
function canPreReserveForLiveDelivery(member: RuntimeMember, senderName: string): boolean {
return member.name !== senderName
&& member.sessionId !== undefined
&& member.status === "idle"
&& member.pendingInjectedMessageIds.length === 0
}
async function resolveTeamRuntimeDetails(
teamRunId: string,
@@ -181,6 +189,31 @@ async function deliverLive(
continue
}
if (recipientMember.pendingInjectedMessageIds.length > 0) {
await releaseReservationSafely(reservation, {
teamRunId,
recipient: recipientName,
messageId: message.messageId,
})
continue
}
if (recipientMember.status !== "idle") {
log("[team-mailbox] live delivery unavailable, recipient is not idle", {
reason: "recipient-not-idle",
teamRunId,
recipient: recipientName,
status: recipientMember.status,
messageId: message.messageId,
})
await releaseReservationSafely(reservation, {
teamRunId,
recipient: recipientName,
messageId: message.messageId,
})
continue
}
const recipientSessionId = recipientMember.sessionId
if (!recipientSessionId) {
log("[team-mailbox] live delivery unavailable, falling back to inbox injection", {
@@ -205,6 +238,7 @@ async function deliverLive(
client,
sessionID: recipientSessionId,
source: "team-live-delivery",
queueBehavior: "defer",
input: {
path: { id: recipientSessionId },
body: buildMemberPromptBody(recipientMember, envelope),
@@ -315,7 +349,7 @@ export function createTeamSendMessageTool(
const runtimeState = await deps.loadRuntimeState(teamRuntime.teamRunId, config)
const reservedRecipients = new Set<string>(
runtimeState.members
.filter((member) => member.sessionId !== undefined && member.name !== teamRuntime.senderName)
.filter((member) => canPreReserveForLiveDelivery(member, teamRuntime.senderName))
.map((member) => member.name),
)