diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 155370da1757..f537fd8ec83d 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -11496,6 +11496,201 @@ 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([ + { 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 (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"; + 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" }, + ); + + 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 [ + 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* absorbWithUserTurn(laterResult); + else yield* harness.offerAndWait(laterResult); + 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: { readonly toolUseId: string; readonly uuids: readonly [string, string]; diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index b3c1fd1c1fed..c4a9d213c41d 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,14 +12665,29 @@ 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); - 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 }); @@ -12683,11 +12748,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( 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" ||