fix(background-agent): coalesce parent wake races
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
@@ -194,6 +194,12 @@ export class ParentWakeNotifier {
|
|||||||
if (promptResult.status === "failed") {
|
if (promptResult.status === "failed") {
|
||||||
throw promptResult.error
|
throw promptResult.error
|
||||||
}
|
}
|
||||||
|
if (promptResult.status === "reserved" && promptResult.reservedBy === "background-agent-parent-wake") {
|
||||||
|
log("[background-agent] Ignored duplicate parent wake flush already reserved by promptAsync gate:", {
|
||||||
|
sessionID,
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
if (promptResult.status !== "dispatched") {
|
if (promptResult.status !== "dispatched") {
|
||||||
this.requeueWake(sessionID, latestWake)
|
this.requeueWake(sessionID, latestWake)
|
||||||
this.schedulePendingParentWakeFlush(sessionID)
|
this.schedulePendingParentWakeFlush(sessionID)
|
||||||
@@ -451,7 +457,7 @@ export class ParentWakeNotifier {
|
|||||||
if (createdAt === undefined) {
|
if (createdAt === undefined) {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
return Date.now() - createdAt < this.options.userMessageInProgressWindowMs
|
return Date.now() - createdAt <= this.options.userMessageInProgressWindowMs
|
||||||
}
|
}
|
||||||
if (role === "assistant" || role === "tool") {
|
if (role === "assistant" || role === "tool") {
|
||||||
// An assistant/tool message is more recent than the last user message,
|
// An assistant/tool message is more recent than the last user message,
|
||||||
|
|||||||
@@ -67,6 +67,155 @@ function createNotifier(args: {
|
|||||||
}
|
}
|
||||||
|
|
||||||
describe("ParentWakeNotifier — user message race guard (issue #4120)", () => {
|
describe("ParentWakeNotifier — user message race guard (issue #4120)", () => {
|
||||||
|
test("#given user message was created exactly at the race-window boundary #when flushing pending wake #then dispatch is still deferred", async () => {
|
||||||
|
// given
|
||||||
|
const originalDateNow = Date.now
|
||||||
|
Date.now = () => 10_000
|
||||||
|
const { notifier, promptAsyncCalls } = createNotifier({
|
||||||
|
sessionMessages: [
|
||||||
|
{
|
||||||
|
info: {
|
||||||
|
role: "user",
|
||||||
|
time: { created: 8_000 },
|
||||||
|
},
|
||||||
|
},
|
||||||
|
],
|
||||||
|
userMessageInProgressWindowMs: 2_000,
|
||||||
|
})
|
||||||
|
notifier.queuePendingParentWake(
|
||||||
|
"parent-boundary",
|
||||||
|
"task complete",
|
||||||
|
{ agent: "sisyphus" },
|
||||||
|
true,
|
||||||
|
)
|
||||||
|
|
||||||
|
try {
|
||||||
|
// when
|
||||||
|
await notifier.flushPendingParentWake("parent-boundary")
|
||||||
|
|
||||||
|
// then
|
||||||
|
expect(promptAsyncCalls).toHaveLength(0)
|
||||||
|
expect(notifier.getPendingParentWakes().has("parent-boundary")).toBe(true)
|
||||||
|
} finally {
|
||||||
|
Date.now = originalDateNow
|
||||||
|
notifier.shutdown()
|
||||||
|
releaseAllPromptAsyncReservationsForTesting()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
test("#given user message is just outside the race-window boundary #when flushing pending wake #then dispatch proceeds", async () => {
|
||||||
|
// given
|
||||||
|
const originalDateNow = Date.now
|
||||||
|
Date.now = () => 10_001
|
||||||
|
const { notifier, promptAsyncCalls } = createNotifier({
|
||||||
|
sessionMessages: [
|
||||||
|
{
|
||||||
|
info: {
|
||||||
|
role: "user",
|
||||||
|
time: { created: 8_000 },
|
||||||
|
},
|
||||||
|
},
|
||||||
|
],
|
||||||
|
userMessageInProgressWindowMs: 2_000,
|
||||||
|
})
|
||||||
|
notifier.queuePendingParentWake(
|
||||||
|
"parent-boundary-open",
|
||||||
|
"task complete",
|
||||||
|
{ agent: "sisyphus" },
|
||||||
|
true,
|
||||||
|
)
|
||||||
|
|
||||||
|
try {
|
||||||
|
// when
|
||||||
|
await notifier.flushPendingParentWake("parent-boundary-open")
|
||||||
|
|
||||||
|
// then
|
||||||
|
expect(promptAsyncCalls).toHaveLength(1)
|
||||||
|
expect(notifier.getPendingParentWakes().has("parent-boundary-open")).toBe(false)
|
||||||
|
} finally {
|
||||||
|
Date.now = originalDateNow
|
||||||
|
notifier.shutdown()
|
||||||
|
releaseAllPromptAsyncReservationsForTesting()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
test("#given two flushes race for one pending wake #when both reach the prompt gate #then the skipped duplicate is not requeued", async () => {
|
||||||
|
// given
|
||||||
|
const { notifier, promptAsyncCalls } = createNotifier({
|
||||||
|
sessionMessages: [
|
||||||
|
{
|
||||||
|
info: {
|
||||||
|
role: "assistant",
|
||||||
|
finish: "stop",
|
||||||
|
time: { created: Date.now() - 10_000 },
|
||||||
|
},
|
||||||
|
},
|
||||||
|
],
|
||||||
|
})
|
||||||
|
notifier.queuePendingParentWake(
|
||||||
|
"parent-concurrent",
|
||||||
|
"task complete",
|
||||||
|
{ agent: "sisyphus" },
|
||||||
|
true,
|
||||||
|
)
|
||||||
|
|
||||||
|
// when
|
||||||
|
await Promise.all([
|
||||||
|
notifier.flushPendingParentWake("parent-concurrent"),
|
||||||
|
notifier.flushPendingParentWake("parent-concurrent"),
|
||||||
|
])
|
||||||
|
|
||||||
|
// then
|
||||||
|
expect(promptAsyncCalls).toHaveLength(1)
|
||||||
|
expect(notifier.getPendingParentWakes().has("parent-concurrent")).toBe(false)
|
||||||
|
expect(notifier.getPendingParentWakeTimers().has("parent-concurrent")).toBe(false)
|
||||||
|
|
||||||
|
notifier.shutdown()
|
||||||
|
releaseAllPromptAsyncReservationsForTesting()
|
||||||
|
})
|
||||||
|
|
||||||
|
test("#given burst notifications share a parent session #when the pending wake flushes #then one dispatch drains the coalesced wake", async () => {
|
||||||
|
// given
|
||||||
|
const { notifier, promptAsyncCalls } = createNotifier({
|
||||||
|
sessionMessages: [
|
||||||
|
{
|
||||||
|
info: {
|
||||||
|
role: "assistant",
|
||||||
|
finish: "stop",
|
||||||
|
time: { created: Date.now() - 10_000 },
|
||||||
|
},
|
||||||
|
},
|
||||||
|
],
|
||||||
|
})
|
||||||
|
notifier.queuePendingParentWake(
|
||||||
|
"parent-burst",
|
||||||
|
"task one complete",
|
||||||
|
{ agent: "sisyphus" },
|
||||||
|
false,
|
||||||
|
)
|
||||||
|
notifier.queuePendingParentWake(
|
||||||
|
"parent-burst",
|
||||||
|
"task two complete",
|
||||||
|
{ agent: "sisyphus" },
|
||||||
|
true,
|
||||||
|
)
|
||||||
|
|
||||||
|
// when
|
||||||
|
await Promise.all([
|
||||||
|
notifier.flushPendingParentWake("parent-burst"),
|
||||||
|
notifier.flushPendingParentWake("parent-burst"),
|
||||||
|
])
|
||||||
|
|
||||||
|
// then
|
||||||
|
expect(promptAsyncCalls).toHaveLength(1)
|
||||||
|
expect(promptAsyncCalls[0]?.body.noReply).toBe(false)
|
||||||
|
expect(notifier.getPendingParentWakes().has("parent-burst")).toBe(false)
|
||||||
|
expect(notifier.getPendingParentWakeTimers().has("parent-burst")).toBe(false)
|
||||||
|
|
||||||
|
notifier.shutdown()
|
||||||
|
releaseAllPromptAsyncReservationsForTesting()
|
||||||
|
})
|
||||||
|
|
||||||
test("#given latest message is a user message just added #when flushing pending wake #then dispatch is deferred (no promptAsync)", async () => {
|
test("#given latest message is a user message just added #when flushing pending wake #then dispatch is deferred (no promptAsync)", async () => {
|
||||||
// given
|
// given
|
||||||
const { notifier, promptAsyncCalls } = createNotifier({
|
const { notifier, promptAsyncCalls } = createNotifier({
|
||||||
|
|||||||
Reference in New Issue
Block a user