From cffa8013e7fbdac9443348420627485ed876e738 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 19:46:35 +0000 Subject: [PATCH 1/5] test(server): a Claude Monitor event wake is tagged as a monitor update Failing repro for #36113: a per-line Monitor wake reaches the continuation with no notification, so it is tagged background_task/updated and starts a delegated follow-up that wakes the parent. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 103 ++++++++++++++++++ 1 file changed, 103 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 155370da1757..44bb8007baca 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11496,6 +11496,109 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + // Claude 2.1.323 wakes a session per Monitor event with only init, assistant and a + // task-notification result; just the Monitor's end sends a task_notification frame. + it.effect("tags a Claude Monitor event wake as a monitor update", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const taskId = "bi5kfx7xp"; + const toolUseId = "toolu_monitor_events"; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-monitor-event"), + text: "Watch the mutation run and report each survivor.", + attachments: [], + }), + ); + for (const frame of [ + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_monitor_events", + type: "message", + role: "assistant", + content: [ + { + type: "tool_use", + id: toolUseId, + name: "Monitor", + input: { description: "line monitor", command: "./mutate.sh" }, + }, + ], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 1, output_tokens: 1 }, + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000901", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: taskId, + tool_use_id: toolUseId, + description: "line monitor", + is_backgrounded: true, + task_type: "local_bash", + uuid: "00000000-0000-4000-8000-000000000902", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "user", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: toolUseId, + content: `Monitor started (task ${taskId}). You will be notified on each event.`, + }, + ], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000903", + session_id: WAKE_NATIVE_SESSION, + tool_use_result: { taskId, persistent: false }, + }), + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000904", result: "STARTED" }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* harness.awaitTerminal(1); + + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000000905", + text: "Covered. Waiting for the rest.", + }), + ); + yield* harness.awaitContinuation(1); + + const notification = harness.continuationRequests[0]?.notification; + assert.deepEqual( + { source: notification?.source, outcome: notification?.outcome }, + { source: { kind: "monitor" }, outcome: "updated" }, + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + const scheduleWakeupFrames = (input: { readonly toolUseId: string; readonly uuids: readonly [string, string]; From 2090a541ac5840d532b278b52a67486649adb133 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 19:54:51 +0000 Subject: [PATCH 2/5] fix(server): a Claude Monitor line no longer wakes a delegated child's parent Claude wakes a session for each Monitor line with a bare turn (init, assistant, task-notification result) and no task_notification frame, so the continuation carried no notification and fell back to background_task/updated, which starts a delegated follow-up and wakes the parent. The adapter now tags a wake turn that starts while a Monitor runs, with no recorded task report and no due ScheduleWakeup, as monitor/updated. A continuation carrying that tag replays only its own wake: a wake buffered behind it gets a continuation of its own, so its answer keeps its own tag. Fixes pingdotgg/t3code#36113 Co-Authored-By: Claude Fable 5.2 --- .../Adapters/ClaudeAdapterV2.test.ts | 55 +++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 78 +++++++++++++++++-- 2 files changed, 127 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 44bb8007baca..a3b1e140fc7c 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11587,6 +11587,61 @@ describe("ClaudeAdapterV2 background wake turns", () => { { source: notification?.source, outcome: notification?.outcome }, { source: { kind: "monitor" }, outcome: "updated" }, ); + + // A wake buffered behind the queued Monitor continuation must not run under its tag. + for (const frame of [ + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000906", + result: "Covered. Waiting for the rest.", + origin: { kind: "task-notification" }, + }), + claudeSdkFrame({ + type: "system", + subtype: "init", + uuid: "00000000-0000-4000-8000-000000000907", + session_id: WAKE_NATIVE_SESSION, + }), + makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000000908", + text: "Scheduled answer.", + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000909", + result: "Scheduled answer.", + }), + ]) { + yield* harness.offerAndWait(frame); + } + const assistantTexts = () => + harness.events.flatMap((event) => + event.type === "message.updated" && event.message.role === "assistant" + ? [event.message.text] + : [], + ); + const startContinuation = (attempt: string, providerTurnOrdinal: number) => + harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make(attempt), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* startContinuation("attempt-claude-monitor-event-2", 2); + yield* harness.awaitTerminal(2); + assert.notInclude(assistantTexts(), "Scheduled answer."); + yield* harness.awaitContinuation(2); + assert.isUndefined(harness.continuationRequests[1]?.notification); + + yield* startContinuation("attempt-claude-monitor-event-3", 3); + yield* harness.awaitTerminal(3); + assert.include(assistantTexts(), "Scheduled answer."); + assert.lengthOf(harness.continuationRequests, 2); }).pipe( Effect.provide( Layer.mergeAll( diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index b3c1fd1c1fed..7cda340c4a89 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5775,6 +5775,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const scheduledWakeupsByNativeThread = new Map(); // Native threads whose current turn Claude started by itself inside the early-fire margin. const possibleEarlyWakeNativeThreads = new Set(); + // Turn starts a due or nearly due ScheduleWakeup may have caused. + const possibleScheduledWakeStarts = new WeakSet(); + // A Monitor event wakes Claude with a bare turn and no task_notification frame, so the + // only evidence is a turn start while a Monitor runs that no ScheduleWakeup explains. + const monitorEventWakeStarts = new WeakSet(); const trackClaudeScheduledWakeup = Effect.fnUntraced(function* ( nativeThreadId: string, message: SDKMessage, @@ -5783,6 +5788,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const scheduledFor = scheduledWakeupsByNativeThread.get(nativeThreadId); if (scheduledFor === undefined) return; const now = DateTime.toEpochMillis(yield* DateTime.now); + if (now >= scheduledFor - CLAUDE_SCHEDULED_WAKEUP_EARLY_FIRE_MARGIN_MS) { + possibleScheduledWakeStarts.add(message); + } if (now >= scheduledFor) { scheduledWakeupsByNativeThread.delete(nativeThreadId); } else if ( @@ -5864,6 +5872,24 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( for (const [taskId] of endedIds) tasks.delete(taskId); return [tasks, { ...current, tasks }] as const; }); + const hasRunningClaudeMonitor = (nativeThreadId: string) => + Ref.get(claudeMonitors).pipe( + Effect.map((current) => + [...current.tasks.values()].some((task) => task.nativeThreadId === nativeThreadId), + ), + ); + // The buffer's first wake is a Monitor event: its result, once known, is a task notification's. + const isClaudeMonitorEventWake = (messages: ReadonlyArray) => { + const startIndex = messages.findIndex(isClaudeTurnStartMessage); + const start = messages[startIndex]; + if (start === undefined || !monitorEventWakeStarts.has(start)) return false; + const result = messages + .slice(startIndex + 1) + .find( + (entry) => entry.type === "result" && !isClaudeEmptyTaskNotificationSuccess(entry), + ); + return result === undefined || isClaudeTaskNotificationOriginResult(result); + }; const claudeTaskOutcome = (status: "completed" | "failed" | "stopped") => status === "completed" ? "completed" : status === "stopped" ? "cancelled" : "failed"; // Reads the roster, so call it before the notification clears the task from it. @@ -5905,6 +5931,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( readonly token: symbol; readonly end: SDKMessage | undefined; readonly retired: "unopened" | "opened" | null; + // Offered as Monitor progress, which starts no delegated follow-up. + readonly monitorEvent: boolean; } >(), ); @@ -9379,6 +9407,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (isClaudeRootTurnFrame(message)) { yield* forgetAbsorbedWake(wakeInput.nativeThreadId); } + if ( + isWakeTurnStart && + !possibleScheduledWakeStarts.has(message) && + (yield* hasRunningClaudeMonitor(wakeInput.nativeThreadId)) + ) { + monitorEventWakeStarts.add(message); + } yield* Ref.update(wakeBuffers, (current) => { const existing = current.get(wakeInput.nativeThreadId); const updated = new Map(current); @@ -9435,6 +9470,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } const buffered = (yield* Ref.get(wakeBuffers)).get(nativeThreadId); const token = Symbol(nativeThreadId); + // A recorded report names finished work, which outranks Monitor progress. + const monitorEvent = + ((yield* Ref.get(wakeReportsByNativeThread)).reports.get(nativeThreadId)?.size ?? 0) === + 0 && isClaudeMonitorEventWake(buffered?.messages ?? []); const shouldOffer = yield* Ref.modify(requestedContinuations, (current) => { if (current.has(nativeThreadId)) { return [false, current] as const; @@ -9446,6 +9485,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( (entry) => entry.type === "result" && !isClaudeEmptyTaskNotificationSuccess(entry), ), retired: null, + monitorEvent, }); return [true, updated] as const; }); @@ -9454,7 +9494,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } const reports = yield* takeWakeReports(nativeThreadId); const notification = backgroundWorkNotification( - [...(reports?.values() ?? [])].map((entry) => entry.report), + monitorEvent + ? [{ kind: "monitor", outcome: "updated" }] + : [...(reports?.values() ?? [])].map((entry) => entry.report), ); yield* Effect.logInfo("orchestration-v2.claude-wake-turn-detected", { providerSessionId: input.providerSessionId, @@ -9610,9 +9652,17 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // After a retired wake releases its latch, a wake buffered behind it has had no offer of its own. const offerBufferedWake = Effect.fnUntraced(function* (nativeThreadId: string) { const messages = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; + if (yield* holdsBufferedWake(nativeThreadId, messages)) + yield* offerWakeContinuation(nativeThreadId); + }); + + const holdsBufferedWake = Effect.fnUntraced(function* ( + nativeThreadId: string, + messages: ReadonlyArray, + ) { const reports = (yield* Ref.get(wakeReportsByNativeThread)).reports.get(nativeThreadId); // A lifecycle result ends no turn, and closes the root frames before it. - const hasWake = + return ( messages .slice(messages.findLastIndex(isClaudeEmptyTaskNotificationSuccess) + 1) .some(isClaudeRootTurnFrame) || @@ -9622,8 +9672,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( (entry.type === "system" && entry.subtype === "task_notification" && reports?.get(entry.task_id)?.report.kind === "subagent"), - ); - if (hasWake) yield* offerWakeContinuation(nativeThreadId); + ) + ); }); // Untraced like the roster message handling: it runs for every streamed roster change. @@ -12615,13 +12665,27 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* sdkMessageLock.withPermits(1)( Effect.gen(function* () { yield* attachTurn; - const drained = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; + const buffered = (yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages ?? []; + const latch = (yield* Ref.get(requestedContinuations)).get(nativeThreadId); + // This run carries the Monitor tag, which starts no delegated follow-up, so a wake + // buffered behind the Monitor event is left for a continuation of its own. + const ownEndIndex = + latch?.monitorEvent === true && latch.end !== undefined + ? buffered.indexOf(latch.end) + : -1; + const leavesLaterWake = + ownEndIndex !== -1 && + (yield* holdsBufferedWake(nativeThreadId, buffered.slice(ownEndIndex + 1))); + const drained = leavesLaterWake ? buffered.slice(0, ownEndIndex + 1) : buffered; yield* Ref.update(requestedContinuations, (current) => { const updated = new Map(current); updated.delete(nativeThreadId); return updated; }); - const wakeAbsorbed = (yield* Ref.get(absorbedWakes)).has(nativeThreadId); + // A split wake that a user turn ended gets no result of its own. + const wakeAbsorbed = + (yield* Ref.get(absorbedWakes)).has(nativeThreadId) || + (leavesLaterWake && latch?.end?.type !== "result"); yield* forgetAbsorbedWake(nativeThreadId); if (drained.length === 0) { const completedAt = yield* DateTime.now; @@ -12683,11 +12747,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } if (lastResult !== undefined && !emptyTurnBeforeNextTurn) { yield* handleSdkMessage({ query: querySession.query, message: lastResult }); + if (leavesLaterWake) yield* offerBufferedWake(nativeThreadId); return; } if (wakeAbsorbed || (hasOpaqueTaskNotification && !context.nativeWakeStarted)) { const completedAt = yield* DateTime.now; yield* finalizeActiveTurn({ context, status: "completed", completedAt }); + if (leavesLaterWake) yield* offerBufferedWake(nativeThreadId); } else if (!context.nativeWakeStarted) { yield* Effect.sleep(CLAUDE_WAKE_TURN_START_DEADLINE).pipe( Effect.andThen( From 6fc0f8568142d75ddb4812013c46ba4f416a4381 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 20:15:57 +0000 Subject: [PATCH 3/5] fix(server): a wake absorbed behind a Claude Monitor event still settles When a Monitor continuation replayed only its own wake, it cleared the thread's absorbed marker, which by then belonged to the later wake a user turn had ended. That wake's continuation then never reached a terminal event. The marker is now kept when the drained wake ended with its own result, so the later wake consumes it on its own drain. Co-Authored-By: Claude Fable 5.2 --- .../Adapters/ClaudeAdapterV2.test.ts | 316 ++++++++++-------- .../Adapters/ClaudeAdapterV2.ts | 4 +- 2 files changed, 176 insertions(+), 144 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index a3b1e140fc7c..7fab431b9210 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11498,160 +11498,190 @@ describe("ClaudeAdapterV2 background wake turns", () => { // Claude 2.1.323 wakes a session per Monitor event with only init, assistant and a // task-notification result; just the Monitor's end sends a task_notification frame. - it.effect("tags a Claude Monitor event wake as a monitor update", () => - Effect.gen(function* () { - const harness = yield* makeWakeHarness; - const now = yield* DateTime.now; - const taskId = "bi5kfx7xp"; - const toolUseId = "toolu_monitor_events"; - - yield* harness.runtime.startTurn( - makeClaudeTestTurnInput({ - threadId: harness.threadId, - providerThread: harness.providerThread, - now, - attemptId: RunAttemptId.make("attempt-claude-monitor-event"), - text: "Watch the mutation run and report each survivor.", - attachments: [], - }), - ); - for (const frame of [ - claudeSdkFrame({ - type: "assistant", - message: { - model: "claude-sonnet-4-6", - id: "msg_monitor_events", - type: "message", - role: "assistant", - content: [ - { - type: "tool_use", - id: toolUseId, - name: "Monitor", - input: { description: "line monitor", command: "./mutate.sh" }, - }, - ], - stop_reason: null, - stop_sequence: null, - usage: { input_tokens: 1, output_tokens: 1 }, - }, - parent_tool_use_id: null, - uuid: "00000000-0000-4000-8000-000000000901", - session_id: WAKE_NATIVE_SESSION, - }), - claudeSdkFrame({ - type: "system", - subtype: "task_started", - task_id: taskId, - tool_use_id: toolUseId, - description: "line monitor", - is_backgrounded: true, - task_type: "local_bash", - uuid: "00000000-0000-4000-8000-000000000902", - session_id: WAKE_NATIVE_SESSION, - }), - claudeSdkFrame({ - type: "user", - message: { - role: "user", - content: [ - { - type: "tool_result", - tool_use_id: toolUseId, - content: `Monitor started (task ${taskId}). You will be notified on each event.`, - }, - ], - }, - parent_tool_use_id: null, - uuid: "00000000-0000-4000-8000-000000000903", - session_id: WAKE_NATIVE_SESSION, - tool_use_result: { taskId, persistent: false }, - }), - makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000904", result: "STARTED" }), - ]) { - yield* Queue.offer(harness.sdkMessages, frame); - } - yield* harness.awaitTerminal(1); - - yield* harness.offerAndWait(wakeTurnInit); - yield* harness.offerAndWait( - makeAssistantTextFrame({ - uuid: "00000000-0000-4000-8000-000000000905", - text: "Covered. Waiting for the rest.", - }), - ); - yield* harness.awaitContinuation(1); - - const notification = harness.continuationRequests[0]?.notification; - assert.deepEqual( - { source: notification?.source, outcome: notification?.outcome }, - { source: { kind: "monitor" }, outcome: "updated" }, - ); + it.effect.each([ + { laterWake: "ends with its own result" }, + { laterWake: "is absorbed by a user turn" }, + ] as const)( + "tags a Claude Monitor event wake as a monitor update (later wake $laterWake)", + ({ laterWake }) => + Effect.gen(function* () { + const absorbed = laterWake === "is absorbed by a user turn"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const taskId = "bi5kfx7xp"; + const toolUseId = "toolu_monitor_events"; - // A wake buffered behind the queued Monitor continuation must not run under its tag. - for (const frame of [ - makeResultFrame({ - uuid: "00000000-0000-4000-8000-000000000906", - result: "Covered. Waiting for the rest.", - origin: { kind: "task-notification" }, - }), - claudeSdkFrame({ - type: "system", - subtype: "init", - uuid: "00000000-0000-4000-8000-000000000907", - session_id: WAKE_NATIVE_SESSION, - }), - makeAssistantTextFrame({ - uuid: "00000000-0000-4000-8000-000000000908", - text: "Scheduled answer.", - }), - makeResultFrame({ - uuid: "00000000-0000-4000-8000-000000000909", - result: "Scheduled answer.", - }), - ]) { - yield* harness.offerAndWait(frame); - } - const assistantTexts = () => - harness.events.flatMap((event) => - event.type === "message.updated" && event.message.role === "assistant" - ? [event.message.text] - : [], - ); - const startContinuation = (attempt: string, providerTurnOrdinal: number) => - harness.runtime.startTurn( + yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ threadId: harness.threadId, providerThread: harness.providerThread, now, - attemptId: RunAttemptId.make(attempt), - text: "Background task completed.", + attemptId: RunAttemptId.make("attempt-claude-monitor-event"), + text: "Watch the mutation run and report each survivor.", attachments: [], - providerTurnOrdinal, - messageCreatedBy: "agent", - messageCreationSource: "provider", }), ); - yield* startContinuation("attempt-claude-monitor-event-2", 2); - yield* harness.awaitTerminal(2); - assert.notInclude(assistantTexts(), "Scheduled answer."); - yield* harness.awaitContinuation(2); - assert.isUndefined(harness.continuationRequests[1]?.notification); + for (const frame of [ + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_monitor_events", + type: "message", + role: "assistant", + content: [ + { + type: "tool_use", + id: toolUseId, + name: "Monitor", + input: { description: "line monitor", command: "./mutate.sh" }, + }, + ], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 1, output_tokens: 1 }, + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000901", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: taskId, + tool_use_id: toolUseId, + description: "line monitor", + is_backgrounded: true, + task_type: "local_bash", + uuid: "00000000-0000-4000-8000-000000000902", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "user", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: toolUseId, + content: `Monitor started (task ${taskId}). You will be notified on each event.`, + }, + ], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000903", + session_id: WAKE_NATIVE_SESSION, + tool_use_result: { taskId, persistent: false }, + }), + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000904", result: "STARTED" }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* harness.awaitTerminal(1); - yield* startContinuation("attempt-claude-monitor-event-3", 3); - yield* harness.awaitTerminal(3); - assert.include(assistantTexts(), "Scheduled answer."); - assert.lengthOf(harness.continuationRequests, 2); - }).pipe( - Effect.provide( - Layer.mergeAll( - IdAllocator.layer, - McpProviderSessions.layer, - NodeServices.layer, - NodeCrypto.layer, + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000000905", + text: "Covered. Waiting for the rest.", + }), + ); + yield* harness.awaitContinuation(1); + + const notification = harness.continuationRequests[0]?.notification; + assert.deepEqual( + { source: notification?.source, outcome: notification?.outcome }, + { source: { kind: "monitor" }, outcome: "updated" }, + ); + + // A wake buffered behind the queued Monitor continuation must not run under its tag. + for (const frame of [ + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000906", + result: "Covered. Waiting for the rest.", + origin: { kind: "task-notification" }, + }), + claudeSdkFrame({ + type: "system", + subtype: "init", + uuid: "00000000-0000-4000-8000-000000000907", + session_id: WAKE_NATIVE_SESSION, + }), + makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000000908", + text: "Scheduled answer.", + }), + ]) { + yield* harness.offerAndWait(frame); + } + const laterResult = makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000909", + result: absorbed ? "No survivors yet." : "Scheduled answer.", + }); + if (absorbed) { + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-monitor-event-user"), + text: "Any survivors yet?", + attachments: [], + providerTurnOrdinal: 2, + }), + ); + yield* harness.until(() => harness.offeredMessages.length === 2); + yield* Queue.offer(harness.sdkMessages, laterResult); + yield* harness.awaitTerminal(2); + } else { + yield* harness.offerAndWait(laterResult); + } + const settled = absorbed ? 2 : 1; + const assistantTexts = () => + harness.events.flatMap((event) => + event.type === "message.updated" && event.message.role === "assistant" + ? [event.message.text] + : [], + ); + const startContinuation = (attempt: string, providerTurnOrdinal: number) => + harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make(attempt), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* startContinuation("attempt-claude-monitor-event-2", settled + 1); + yield* harness.awaitTerminal(settled + 1); + assert.notInclude(assistantTexts(), "Scheduled answer."); + yield* harness.awaitContinuation(2); + // With no result of its own in the buffer, the later wake reads as another Monitor event. + assert.equal( + harness.continuationRequests[1]?.notification?.source.kind, + absorbed ? "monitor" : undefined, + ); + + yield* startContinuation("attempt-claude-monitor-event-3", settled + 2); + yield* harness.awaitTerminal(settled + 2); + assert.equal(harness.terminalEvents()[settled + 1]?.status, "completed"); + assert.include(assistantTexts(), "Scheduled answer."); + assert.lengthOf(harness.continuationRequests, 2); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), ), ), - ), ); const scheduleWakeupFrames = (input: { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 7cda340c4a89..86f576cb89b9 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -12686,7 +12686,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const wakeAbsorbed = (yield* Ref.get(absorbedWakes)).has(nativeThreadId) || (leavesLaterWake && latch?.end?.type !== "result"); - yield* forgetAbsorbedWake(nativeThreadId); + // After this wake's own result, the absorption recorded is the later wake's. + if (!leavesLaterWake || latch?.end?.type !== "result") + yield* forgetAbsorbedWake(nativeThreadId); if (drained.length === 0) { const completedAt = yield* DateTime.now; yield* finalizeActiveTurn({ context, status: "completed", completedAt }); From ff0698243f5ac0311333c22975fc90d29895ea24 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 20:22:59 +0000 Subject: [PATCH 4/5] fix(server): two wakes absorbed behind a Claude Monitor event both settle A root frame clears the absorbed marker, so when a Monitor continuation leaves a later wake buffered the marker is always that later wake's. The split drain now keeps it in every such case, and reads its own absorption from its end frame alone. Co-Authored-By: Claude Fable 5.2 --- .../Adapters/ClaudeAdapterV2.test.ts | 63 ++++++++++--------- .../Adapters/ClaudeAdapterV2.ts | 13 ++-- 2 files changed, 41 insertions(+), 35 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 7fab431b9210..f537fd8ec83d 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11499,13 +11499,15 @@ describe("ClaudeAdapterV2 background wake turns", () => { // Claude 2.1.323 wakes a session per Monitor event with only init, assistant and a // task-notification result; just the Monitor's end sends a task_notification frame. it.effect.each([ - { laterWake: "ends with its own result" }, - { laterWake: "is absorbed by a user turn" }, + { monitorWake: "ends with its own result", laterWake: "ends with its own result" }, + { monitorWake: "ends with its own result", laterWake: "is absorbed by a user turn" }, + { monitorWake: "is absorbed by a user turn", laterWake: "is absorbed by a user turn" }, ] as const)( - "tags a Claude Monitor event wake as a monitor update (later wake $laterWake)", - ({ laterWake }) => + "tags a Claude Monitor event wake as a monitor update (it $monitorWake, later wake $laterWake)", + ({ monitorWake, laterWake }) => Effect.gen(function* () { const absorbed = laterWake === "is absorbed by a user turn"; + const monitorWakeAbsorbed = monitorWake === "is absorbed by a user turn"; const harness = yield* makeWakeHarness; const now = yield* DateTime.now; const taskId = "bi5kfx7xp"; @@ -11594,13 +11596,35 @@ describe("ClaudeAdapterV2 background wake turns", () => { { source: { kind: "monitor" }, outcome: "updated" }, ); + let settled = 1; + // Claude folds a prompt into the wake turn it already opened, so one result ends both. + const absorbWithUserTurn = Effect.fnUntraced(function* (result: SDKMessage) { + settled += 1; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make(`attempt-claude-monitor-event-user-${settled}`), + text: "Any survivors yet?", + attachments: [], + providerTurnOrdinal: settled, + }), + ); + yield* harness.until(() => harness.offeredMessages.length === settled); + yield* Queue.offer(harness.sdkMessages, result); + yield* harness.awaitTerminal(settled); + }); + const monitorWakeResult = makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000906", + result: "Covered. Waiting for the rest.", + ...(monitorWakeAbsorbed ? {} : { origin: { kind: "task-notification" } as const }), + }); + if (monitorWakeAbsorbed) yield* absorbWithUserTurn(monitorWakeResult); + else yield* harness.offerAndWait(monitorWakeResult); + // A wake buffered behind the queued Monitor continuation must not run under its tag. for (const frame of [ - makeResultFrame({ - uuid: "00000000-0000-4000-8000-000000000906", - result: "Covered. Waiting for the rest.", - origin: { kind: "task-notification" }, - }), claudeSdkFrame({ type: "system", subtype: "init", @@ -11618,25 +11642,8 @@ describe("ClaudeAdapterV2 background wake turns", () => { uuid: "00000000-0000-4000-8000-000000000909", result: absorbed ? "No survivors yet." : "Scheduled answer.", }); - if (absorbed) { - yield* harness.runtime.startTurn( - makeClaudeTestTurnInput({ - threadId: harness.threadId, - providerThread: harness.providerThread, - now, - attemptId: RunAttemptId.make("attempt-claude-monitor-event-user"), - text: "Any survivors yet?", - attachments: [], - providerTurnOrdinal: 2, - }), - ); - yield* harness.until(() => harness.offeredMessages.length === 2); - yield* Queue.offer(harness.sdkMessages, laterResult); - yield* harness.awaitTerminal(2); - } else { - yield* harness.offerAndWait(laterResult); - } - const settled = absorbed ? 2 : 1; + if (absorbed) yield* absorbWithUserTurn(laterResult); + else yield* harness.offerAndWait(laterResult); const assistantTexts = () => harness.events.flatMap((event) => event.type === "message.updated" && event.message.role === "assistant" diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 86f576cb89b9..c4a9d213c41d 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -12682,13 +12682,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( updated.delete(nativeThreadId); return updated; }); - // A split wake that a user turn ended gets no result of its own. - const wakeAbsorbed = - (yield* Ref.get(absorbedWakes)).has(nativeThreadId) || - (leavesLaterWake && latch?.end?.type !== "result"); - // After this wake's own result, the absorption recorded is the later wake's. - if (!leavesLaterWake || latch?.end?.type !== "result") - yield* forgetAbsorbedWake(nativeThreadId); + // A root frame clears the absorbed marker, so behind a split it is the later wake's + // and stays; the split wake was absorbed when its end is not a result. + const wakeAbsorbed = leavesLaterWake + ? latch?.end?.type !== "result" + : (yield* Ref.get(absorbedWakes)).has(nativeThreadId); + if (!leavesLaterWake) yield* forgetAbsorbedWake(nativeThreadId); if (drained.length === 0) { const completedAt = yield* DateTime.now; yield* finalizeActiveTurn({ context, status: "completed", completedAt }); From 684712029b8d7ccd11dac8d499283e943c345c20 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 20:59:40 +0000 Subject: [PATCH 5/5] fix(server): a delegated child's Monitor ending reaches the parent With per-line Monitor wakes now tagged as monitor updates, the follow-up gate dropped every wake of a Monitor-using child, including the one where the Monitor ends, so the parent never heard the child's final answer. The gate now refuses only monitor updates. A Monitor that ends starts a delegated follow-up the way a background command's ending does. Co-Authored-By: Claude Fable 5.2 --- .../DelegatedCompletionDelivery.test.ts | 18 +++++++++++++----- .../src/orchestration-v2/Orchestrator.ts | 5 ++++- 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 1daeb72e7ece..b0886435d0b5 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -2635,11 +2635,19 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( threadId: child.childThreadId, messageId: wakeMessageId, text: "Background activity updated", - notification: { - source: { kind: "command" }, - outcome: "completed", - summary: 'Command "sleep 60" finished', - }, + // A monitor that ends is finished work too; only its per-line updates are progress. + notification: + delivery === "live" + ? { + source: { kind: "monitor" }, + outcome: "completed", + summary: 'Monitor "sleep 60" finished', + } + : { + source: { kind: "command" }, + outcome: "completed", + summary: 'Command "sleep 60" finished', + }, attachments: [], dispatchMode: { type: "queue_after_active" }, }); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 00f9dfae8d9e..658bc3f6332e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -765,7 +765,10 @@ function mayStartSubagentFollowup( ): boolean { if (command.scheduledTaskId !== undefined || command.restartContinuationOfRunId !== undefined) return false; - if (command.notification !== undefined) return command.notification.source.kind !== "monitor"; + if (command.notification !== undefined) + return ( + command.notification.source.kind !== "monitor" || command.notification.outcome !== "updated" + ); return ( command.createdBy === "user" || command.creationSource === "mcp" ||