Skip to content

fix(server): consumed Claude background wakes no longer block the thread #26553

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
shivamhwp wants to merge 1 commit into from
+118 −82
Closed

fix(server): consumed Claude background wakes no longer block the thread #26553

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
189 changes: 108 additions & 81 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9170,95 +9170,122 @@ describe("ClaudeAdapterV2 background wake turns", () => {
),
);

it.effect("leaves buffered wake messages for the continuation queued behind a user turn", () =>
Effect.scoped(
Effect.gen(function* () {
const harness = yield* makeWakeHarness;
const now = yield* DateTime.now;
it.effect.each([true, false])(
"leaves buffered wake messages for the continuation queued behind a user turn, buffered result=%s",
(bufferedResult) =>
Effect.scoped(
Effect.gen(function* () {
const harness = yield* makeWakeHarness;
const now = yield* DateTime.now;

yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now,
attemptId: RunAttemptId.make("attempt-claude-wake-4a"),
text: "Run the build in the background.",
attachments: [],
}),
);
yield* Queue.offer(harness.sdkMessages, wakeTaskStarted);
yield* Queue.offer(harness.sdkMessages, turnOneResult);
yield* harness.awaitTerminal(1);
yield* Queue.offer(harness.sdkMessages, wakeNotification);
yield* Queue.offer(harness.sdkMessages, wakeResult);
yield* harness.awaitContinuation(1);
yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now,
attemptId: RunAttemptId.make("attempt-claude-wake-4a"),
text: "Run the build in the background.",
attachments: [],
}),
);
yield* Queue.offer(harness.sdkMessages, wakeTaskStarted);
yield* Queue.offer(harness.sdkMessages, turnOneResult);
yield* harness.awaitTerminal(1);
yield* Queue.offer(harness.sdkMessages, wakeNotification);
yield* Queue.offer(harness.sdkMessages, bufferedResult ? wakeResult : wakeTurnInit);
yield* harness.awaitContinuation(1);

yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now,
attemptId: RunAttemptId.make("attempt-claude-wake-4b"),
text: "How is the build going?",
attachments: [],
providerTurnOrdinal: 2,
}),
);
yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now,
attemptId: RunAttemptId.make("attempt-claude-wake-4b"),
text: "How is the build going?",
attachments: [],
providerTurnOrdinal: 2,
}),
);

yield* harness.until(() => harness.offeredMessages.length === 2);
assert.lengthOf(harness.terminalEvents(), 1);
assert.isTrue(yield* harness.hasPendingBackgroundWork);
yield* harness.until(() => harness.offeredMessages.length === 2);
assert.lengthOf(harness.terminalEvents(), 1);
assert.equal(yield* harness.hasPendingBackgroundWork, bufferedResult);

yield* Queue.offer(
harness.sdkMessages,
makeResultFrame({
uuid: "00000000-0000-4000-8000-000000000105",
result: "The build passed; nothing else pending.",
}),
);
yield* harness.awaitTerminal(2);
assert.equal(harness.terminalEvents()[1]?.status, "completed");
assert.isFalse(
harness.events.some(
(event) => event.type === "message.updated" && event.message.text === WAKE_RESULT_TEXT,
),
);
yield* Queue.offer(
harness.sdkMessages,
makeResultFrame({
uuid: "00000000-0000-4000-8000-000000000105",
result: "The build passed; nothing else pending.",
}),
);
yield* harness.awaitTerminal(2);
assert.equal(harness.terminalEvents()[1]?.status, "completed");
assert.isFalse(
harness.events.some(
(event) =>
event.type === "message.updated" && event.message.text === WAKE_RESULT_TEXT,
),
);

yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now,
attemptId: RunAttemptId.make("attempt-claude-wake-4c"),
text: "Background task completed.",
attachments: [],
providerTurnOrdinal: 3,
messageCreatedBy: "agent",
messageCreationSource: "provider",
}),
);
yield* harness.awaitTerminal(3);
assert.equal(harness.terminalEvents()[2]?.status, "completed");
assert.lengthOf(harness.offeredMessages, 2);
assert.isTrue(
harness.events.some(
(event) => event.type === "message.updated" && event.message.text === WAKE_RESULT_TEXT,
),
);
assert.isFalse(
harness.events.some(
(event) =>
event.type !== "provider_thread.updated" &&
JSON.stringify(event).includes(WAKE_TASK_ID),
yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now,
attemptId: RunAttemptId.make("attempt-claude-wake-4c"),
text: "Background task completed.",
attachments: [],
providerTurnOrdinal: 3,
messageCreatedBy: "agent",
messageCreationSource: "provider",
}),
);
yield* TestClock.adjust("8 hours");
assert.lengthOf(harness.terminalEvents(), 3);
yield* harness.awaitTerminal(3);
assert.equal(harness.terminalEvents()[2]?.status, "completed");
assert.lengthOf(harness.offeredMessages, 2);
assert.equal(
harness.events.some(
(event) =>
event.type === "message.updated" && event.message.text === WAKE_RESULT_TEXT,
),
bufferedResult,
);
assert.isFalse(
harness.events.some(
(event) =>
event.type !== "provider_thread.updated" &&
JSON.stringify(event).includes(WAKE_TASK_ID),
),
);
assert.isFalse(yield* harness.hasPendingBackgroundWork);
yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now,
attemptId: RunAttemptId.make("attempt-claude-wake-4d"),
text: "Continue with the next task.",
attachments: [],
providerTurnOrdinal: 4,
}),
);
yield* harness.offerAndWait(
makeResultFrame({
uuid: "00000000-0000-4000-8000-000000000106",
result: "The next task completed.",
}),
);
yield* harness.awaitTerminal(4);
assert.equal(harness.terminalEvents()[3]?.status, "completed");
assert.lengthOf(harness.offeredMessages, 3);
}).pipe(
Effect.provide(
Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
),
);
assert.isFalse(yield* harness.hasPendingBackgroundWork);
}).pipe(
Effect.provide(
Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
),
),
),
);

it.effect("ignores a zero-turn task-notification origin result during a normal user turn", () =>
Expand Down
11 changes: 10 additions & 1 deletion apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4632,6 +4632,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
{ readonly messages: ReadonlyArray<SDKMessage>; readonly detail: string | null }
>(),
);
// Buffered root frames that a later live result closed; replaying them must not open a wake.
const endedWakeTurnFrames = new WeakSet<SDKMessage>();
// Wake results lack task ids and can follow queued prompts; retain reports for one user turn.
// Keep report stamps atomic with turn changes.
const wakeReportsByNativeThread = yield* Ref.make<{
Expand Down Expand Up @@ -9205,10 +9207,17 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
const message = input.message;
const context = yield* Ref.get(activeTurn);
const liveQuery = yield* Ref.get(queryContext);
if (message.type === "result" && liveQuery?.query === input.query) {
for (const buffered of (yield* Ref.get(wakeBuffers)).get(liveQuery.nativeThreadId)
?.messages ?? []) {
if (isClaudeRootTurnFrame(buffered)) endedWakeTurnFrames.add(buffered);
}
}
if (
context?.promptUuid === null &&
liveQuery?.query === input.query &&
isClaudeRootTurnFrame(message)
isClaudeRootTurnFrame(message) &&
!endedWakeTurnFrames.has(message)
) {
context.nativeWakeStarted = true;
}
Expand Down
Loading