diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 8a06dddeeec10..018550c008e4a 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -2334,6 +2334,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { readonly uuid: string; readonly result: string; readonly numTurns?: number; + readonly permissionDenials?: SDKResultMessage["permission_denials"]; readonly origin?: { readonly kind: "task-notification" }; readonly subtype?: string; readonly isError?: boolean; @@ -2365,7 +2366,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { cache_read_input_tokens: 0, }, modelUsage: {}, - permission_denials: [], + permission_denials: input.permissionDenials ?? [], uuid: input.uuid, session_id: WAKE_NATIVE_SESSION, ...(input.origin === undefined ? {} : { origin: input.origin }), @@ -2476,6 +2477,10 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* Queue.unbounded< Extract >(); + const runtimeRequestReceipts = + yield* Queue.unbounded< + Extract + >(); const systemNoticeReceipts = yield* Queue.unbounded< Extract @@ -2581,6 +2586,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { Stream.runForEach((event) => Effect.gen(function* () { events.push(event); + if (event.type === "runtime_request.updated") { + yield* Queue.offer(runtimeRequestReceipts, event); + } if (event.type === "subagent.updated") { yield* Queue.offer(subagentReceipts, event); } @@ -2626,6 +2634,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { terminalReceipts, subagentReceipts, systemNoticeReceipts, + runtimeRequestReceipts, getOpenedOptions: () => openedOptions, terminalEvents, hasPendingBackgroundWork, @@ -3123,8 +3132,32 @@ describe("ClaudeAdapterV2 background wake turns", () => { attachments: [], }); yield* harness.runtime.startTurn(input); - if (steered) { - yield* harness.runtime.steerTurn({ + const rawReason = "Auto policy blocked this command: it writes outside the workspace."; + const permissionDenials = [ + { tool_name: "Bash", tool_use_id: "steering-denied-bash", tool_input: {} }, + { tool_name: "Write", tool_use_id: "steering-denied-write", tool_input: {} }, + ]; + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "user", + uuid: "steering-denial-reason", + session_id: WAKE_NATIVE_SESSION, + parent_tool_use_id: null, + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: "steering-denied-bash", + is_error: true, + content: rawReason, + }, + ], + }, + }), + ); + const steer = (messageId: string) => + harness.runtime.steerTurn({ threadId: harness.threadId, runId: input.runId, providerThread: harness.providerThread, @@ -3135,19 +3168,21 @@ describe("ClaudeAdapterV2 background wake turns", () => { message: { createdBy: "user", creationSource: "web", - messageId: MessageId.make("message-steering-abort"), + messageId: MessageId.make(messageId), text: "Include the hierarchy mock.", attachments: [], }, }); + if (steered) { + yield* steer("message-steering-abort"); assert.equal(harness.offeredMessages[1]?.priority, "now"); } - yield* Queue.offer( - harness.sdkMessages, + yield* harness.offerAndWait( makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000901", result: "", terminalReason, + permissionDenials, usage: { input_tokens: 600, output_tokens: 40, @@ -3157,6 +3192,28 @@ describe("ClaudeAdapterV2 background wake turns", () => { }), ); if (steered) { + yield* steer("message-steering-abort-again"); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "steering-abort-repeated-denial", + result: "", + terminalReason, + permissionDenials: permissionDenials.slice(0, 1), + usage: { + input_tokens: 0, + output_tokens: 0, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }), + ); + assert.lengthOf(harness.terminalEvents(), 0); + assert.isFalse( + harness.events.some( + (event) => + event.type === "turn_item.updated" && event.turnItem.type === "system_notice", + ), + ); yield* Queue.offer(harness.sdkMessages, wakeAssistant); yield* Queue.offer( harness.sdkMessages, @@ -3202,6 +3259,25 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); } assert.lengthOf(harness.terminalEvents(), 1); + const notices = harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "system_notice" + ? [event.turnItem] + : [], + ); + assert.lengthOf(notices, 1); + const notice = notices[0]!; + assert.match(notice.title ?? "", /\b2 actions\b/); + assert.include(notice.message, "Bash"); + assert.include(notice.message, "Write"); + assert.include(notice.message, rawReason); + assert.isTrue(notice.needsAttention); + assert.equal(notice.runId, input.runId); + assert.isBelow( + harness.events.findIndex( + (event) => event.type === "turn_item.updated" && event.turnItem.id === notice.id, + ), + harness.events.indexOf(terminal), + ); }).pipe( Effect.provide( Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), @@ -3210,6 +3286,198 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("reports SDK-only denials with raw reasons once, scoped to their turn", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const start = (ordinal: number) => + harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make(`attempt-denials-${ordinal}`), + providerTurnOrdinal: ordinal, + text: "Continue.", + attachments: [], + runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "auto", + interactionMode: "default", + cwd: "/workspace", + }), + }), + ); + const notices = () => + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "system_notice" + ? [event.turnItem] + : [], + ); + yield* start(1); + const capturedPlan = yield* Effect.promise(() => + harness.getOpenedOptions()!.canUseTool!( + "ExitPlanMode", + {}, + { + signal: new AbortController().signal, + toolUseID: "captured-plan", + requestId: "captured-plan", + }, + ), + ); + assert.equal(capturedPlan?.behavior, "deny"); + const rawReason = "Auto policy blocked this command: it writes outside the workspace."; + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "user", + uuid: "denial-reason", + session_id: WAKE_NATIVE_SESSION, + parent_tool_use_id: null, + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: "sdk-bash", + is_error: true, + content: rawReason, + }, + ], + }, + tool_use_result: { message: "MISLEADING_STRUCTURED_RESULT" }, + }), + ); + const denials = [ + { tool_name: "ExitPlanMode", tool_use_id: "captured-plan", tool_input: {} }, + { tool_name: "Bash", tool_use_id: "sdk-bash", tool_input: {} }, + { tool_name: "Write", tool_use_id: "sdk-write", tool_input: {} }, + ]; + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "denials-1", + result: "Done.", + permissionDenials: denials, + }), + ); + const terminal = yield* Queue.take(harness.terminalReceipts); + assert.equal(terminal.status, "completed"); + assert.lengthOf(notices(), 1); + const notice = notices()[0]!; + assert.include(notice.message, rawReason); + assert.include(notice.message, "Bash"); + assert.include(notice.message, "Write"); + assert.notInclude(notice.message, "ExitPlanMode"); + assert.notInclude(notice.message, "MISLEADING_STRUCTURED_RESULT"); + assert.isTrue(notice.needsAttention); + assert.equal(notice.runId, RunId.make("run-attempt-denials-1")); + assert.isBelow( + harness.events.findIndex( + (event) => event.type === "turn_item.updated" && event.turnItem.id === notice.id, + ), + harness.events.indexOf(terminal), + ); + assert.isFalse(harness.events.some((event) => event.type === "runtime_request.updated")); + + // Reusing IDs exposes denial markers or reasons that escape their owning turn. + yield* start(2); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "denials-2", + result: "Done.", + permissionDenials: denials, + }), + ); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + assert.lengthOf(notices(), 2); + assert.include(notices()[1]!.message, "ExitPlanMode"); + assert.notInclude(notices()[1]!.message, rawReason); + assert.equal(notices()[1]!.runId, RunId.make("run-attempt-denials-2")); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect.each(["declined approval", "cancelled question"] as const)( + "does not announce a %s as an SDK permission denial", + (scenario) => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-client-denial"), + text: "Continue.", + attachments: [], + runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "approval-required", + interactionMode: "default", + cwd: "/workspace", + }), + }), + ); + const controller = new AbortController(); + const toolName = scenario === "cancelled question" ? "AskUserQuestion" : "Bash"; + const permission = yield* Effect.promise(() => + harness.getOpenedOptions()!.canUseTool!( + toolName, + toolName === "Bash" + ? { command: "pwd" } + : { + questions: [ + { question: "Proceed?", header: "Continue", options: [], multiSelect: false }, + ], + }, + { signal: controller.signal, toolUseID: "client-denied", requestId: "client-denied" }, + ), + ).pipe(Effect.forkScoped); + const request = yield* Queue.take(harness.runtimeRequestReceipts); + if (scenario === "declined approval") { + yield* harness.runtime.respondToRuntimeRequest({ + requestId: request.runtimeRequest.id, + decision: "decline", + }); + } else { + controller.abort(); + } + assert.equal((yield* Fiber.join(permission))?.behavior, "deny"); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "client-denial-result", + result: "Done.", + permissionDenials: [ + { tool_name: toolName, tool_use_id: "client-denied", tool_input: {} }, + ], + }), + ); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + assert.isFalse( + harness.events.some( + (event) => + event.type === "turn_item.updated" && event.turnItem.type === "system_notice", + ), + ); + assert.lengthOf( + harness.events.filter((event) => event.type === "runtime_request.updated"), + 1, + ); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect("announces usage-limit pauses once per window and again on a new turn", () => Effect.gen(function* () { const harness = yield* makeWakeHarness; @@ -6306,7 +6574,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { 0, ), ); - yield* awaitUntil(() => harness.terminalEvents().length === 1, "first turn terminal"); + yield* Queue.take(harness.terminalReceipts); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -6321,14 +6589,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { }), ); yield* Queue.offer(harness.sdkMessages, wakeNotification); - yield* Queue.offer(harness.sdkMessages, planToolUse); - // The SDK asks for permission only after streaming the tool_use. - yield* awaitUntil( - () => Queue.sizeUnsafe(harness.sdkMessages) === 0, - "the tool_use frame to be consumed", - ); - let settleYields = 0; - yield* awaitUntil(() => settleYields++ >= 50, "the tool_use frame to be handled"); + yield* harness.offerAndWait(planToolUse); const canUseTool = harness.getOpenedOptions()?.canUseTool; assert.isFunction(canUseTool); const callback = yield* Effect.promise(() => @@ -6348,11 +6609,35 @@ describe("ClaudeAdapterV2 background wake turns", () => { "The client captured your proposed plan", ); yield* Queue.offer(harness.sdkMessages, planToolResult); - yield* Queue.offer(harness.sdkMessages, wakeResult); - yield* awaitUntil( - () => harness.continuationRequests.length === 1, - "continuation request", + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + ...planToolResult, + uuid: "wake-sdk-denial-reason", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: "wake-sdk-denied", + is_error: true, + content: "Wake command blocked by policy.", + }, + ], + }, + }), ); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + ...wakeResult, + permission_denials: [ + { tool_name: "ExitPlanMode", tool_use_id: planToolUseId, tool_input: {} }, + { tool_name: "Bash", tool_use_id: "wake-sdk-denied", tool_input: {} }, + ], + }), + ); + yield* Queue.take(harness.continuationReceipts); yield* Queue.offer( harness.sdkMessages, @@ -6374,7 +6659,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { 1, ), ); - yield* awaitUntil(() => harness.terminalEvents().length === 2, "user turn terminal"); + yield* Queue.take(harness.terminalReceipts); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -6390,7 +6675,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { messageCreationSource: "provider", }), ); - yield* awaitUntil(() => harness.terminalEvents().length === 3, "continuation terminal"); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); const latestItems = new Map( harness.events.flatMap((event) => @@ -6403,6 +6688,12 @@ describe("ClaudeAdapterV2 background wake turns", () => { (item) => item.type === "dynamic_tool" && item.nativeItemRef?.nativeId === planToolUseId, ); + const notices = [...latestItems.values()].filter((item) => item.type === "system_notice"); + assert.lengthOf(notices, 1); + assert.equal(notices[0]!.runId, runOf(continuationAttempt)); + assert.include(notices[0]!.message, "Wake command blocked by policy."); + assert.notInclude(notices[0]!.message, "ExitPlanMode"); + assert.isTrue(notices[0]!.needsAttention); const proposedPlans = [...latestItems.values()].filter( (item) => item.type === "proposed_plan", ); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index be276af5f0edf..8c02549aafb97 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -3044,6 +3044,9 @@ interface ActiveClaudeTurnContext { // compaction makes the usage stale (null) without forgetting the request. streamUsage: { readonly messageId: string; usage: ClaudeMessageUsage | null } | null; readonly toolCalls: Map; + readonly deniedByT3ToolUseIds: Set; + readonly permissionDenials: Map; + readonly toolErrorTexts: Map; readonly ignoredTaskIds: Set; readonly announcedUsageLimits: Set; authenticationFailureMessage: string | undefined; @@ -3753,6 +3756,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); const requestedContinuations = yield* Ref.make(new Set()); const heldProposedPlansByToolUseId = new Map(); + // Held frames can drain into a different turn. + const heldDeniedByT3Tools = new WeakSet(); const runtimeContext = yield* Effect.context(); const runPromise = Effect.runPromiseWith(runtimeContext); @@ -7596,6 +7601,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const toolPresentations = claudeToolPresentationsFromAssistantMessage(message); for (const toolUse of claudeToolUseBlocksFromAssistantMessage(message)) { + if (heldDeniedByT3Tools.delete(toolUse)) { + context.deniedByT3ToolUseIds.add(toolUse.id); + } const nativeToolInput = claudeNativeToolInputFromUnknown(toolUse.input); if (toolUse.name === "Agent") { rememberClaudeSubagentLaunch( @@ -7641,6 +7649,15 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } for (const { toolResult, output } of claudeToolResultEntriesFromMessage(message)) { + if (isClaudeToolResultError(toolResult)) { + context.toolErrorTexts.set( + toolResult.tool_use_id, + claudeSubagentResultText(claudeNativeToolOutputFromToolResult(toolResult)) + .replace(/\s+/gu, " ") + .trim() + .slice(0, 500), + ); + } const subagent = yield* resolveSubagentByToolUseId(context, toolResult.tool_use_id); const runHandles = parseClaudeWorkflowRunHandles(claudeNativeToolOutputValue(output)); if (subagent !== undefined && runHandles !== undefined) { @@ -7832,6 +7849,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } if (message.type === "result") { + for (const denial of message.permission_denials) { + context.permissionDenials.set(denial.tool_use_id, denial); + } const completedAt = yield* DateTime.now; const interrupted = (yield* Ref.get(interruptedTurns)).has(context.providerTurnId); const wasSteered = (yield* Ref.get(steeredTurns)).has(context.providerTurnId); @@ -7888,6 +7908,54 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( resultFailure?.class === "usage_limit" ? { ...resultFailure, resetAt: resetAt ?? resultFailure.resetAt ?? null } : resultFailure; + const denials = Array.from(context.permissionDenials.values()).filter( + (denial) => !context.deniedByT3ToolUseIds.has(denial.tool_use_id), + ); + if (denials.length > 0) { + const tools = [...new Set(denials.map((denial) => denial.tool_name))].join(", "); + const reasons = [ + ...new Set( + denials.flatMap((denial) => { + const reason = context.toolErrorTexts.get(denial.tool_use_id); + return reason ? [reason] : []; + }), + ), + ] + .join("; ") + .slice(0, 1500); + const title = `Claude denied ${denials.length} action${denials.length === 1 ? "" : "s"} (${tools})`; + const nativeItemId = `${context.providerTurnId}:permission-denials`; + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CLAUDE_PROVIDER, + turnItem: { + id: idAllocator.derive.turnItemFromProviderItem({ + driver: CLAUDE_PROVIDER, + nativeItemId, + }), + threadId: context.input.threadId, + runId: context.input.runId, + nodeId: context.input.rootNodeId, + providerThreadId: context.input.providerThread.id, + providerTurnId: context.providerTurnId, + nativeItemRef: { + driver: CLAUDE_PROVIDER, + nativeId: nativeItemId, + strength: "weak", + }, + parentItemId: null, + ordinal: yield* resolveItemOrdinal(context, nativeItemId), + type: "system_notice", + status: "completed", + title, + message: reasons.length > 0 ? `${title}: ${reasons}` : title, + needsAttention: true, + startedAt: completedAt, + completedAt, + updatedAt: completedAt, + }, + }); + } yield* finalizeActiveTurn({ context, status: interrupted @@ -8317,7 +8385,24 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); const canUseTool: CanUseTool = (toolName, toolInput, callbackOptions) => - runPromise(canUseToolEffect(toolName, toolInput, callbackOptions)); + runPromise( + Effect.gen(function* () { + const context = yield* Ref.get(activeTurn); + const result = yield* canUseToolEffect(toolName, toolInput, callbackOptions); + if (result.behavior === "deny" && context !== null) { + // Held output may belong to a queued wake rather than this prompt. + const heldTool = context.heldRootFrames + .flatMap(claudeToolUseBlocksFromAssistantMessage) + .find((tool) => tool.id === callbackOptions.toolUseID); + if (heldTool !== undefined) { + heldDeniedByT3Tools.add(heldTool); + } else { + context.deniedByT3ToolUseIds.add(callbackOptions.toolUseID); + } + } + return result; + }), + ); const onUserDialog: NonNullable = ( request, @@ -8713,6 +8798,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }, streamUsage: null, toolCalls: new Map(), + deniedByT3ToolUseIds: new Set(), + permissionDenials: new Map(), + toolErrorTexts: new Map(), ignoredTaskIds: new Set(), announcedUsageLimits: new Set(), authenticationFailureMessage: undefined, diff --git a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts b/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts index 0c31bcafc9b1e..7b67c9ab790b0 100644 --- a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts @@ -76,6 +76,7 @@ const createRun = Effect.fn(function* ( threadId: ThreadId, status: OrchestrationV2Run["status"] = "completed", ordinal = 1, + overrides: Partial = {}, ) { const store = yield* ProjectionStore.ProjectionStoreV2; const runId = RunId.make(`run:${threadId}:${ordinal}`); @@ -100,6 +101,7 @@ const createRun = Effect.fn(function* ( completedAt: ["completed", "rolled_back", "failed"].includes(status) ? old : null, checkpointId: null, contextHandoffId: null, + ...overrides, }, }); return runId; @@ -353,6 +355,69 @@ it.effect.each([ }).pipe(Effect.provide(testLayer)), ); +it.effect.each([ + ["sql", layerSql], + ["memory", ProjectionStore.layerMemory], +] as const)( + "%s: attention notices hold settlement until another run actually starts", + ([, testLayer]) => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const later = DateTime.add(old, { hours: 1 }); + for (const [name, needsAttention, nextRun, eligible] of [ + ["attention", true, null, false], + ["ordinary-notice", undefined, null, true], + ["queued", true, { status: "queued", startedAt: null }, false], + ["canceled-before-start", true, { status: "cancelled", startedAt: null }, false], + ["later-start", true, { startedAt: later }, true], + ] satisfies ReadonlyArray< + readonly [string, boolean | undefined, Partial | null, boolean] + >) { + const threadId = yield* createThread(name); + const runId = yield* createRun(threadId, "completed", 2); + yield* store.apply({ + id: EventId.make(`event:notice:${runId}`), + type: "turn-item.updated", + threadId, + occurredAt: old, + payload: { + id: TurnItemId.make(`notice:${runId}`), + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + type: "system_notice", + status: "completed", + title: null, + startedAt: old, + completedAt: old, + updatedAt: old, + message: "Needs attention", + ...(needsAttention === undefined ? {} : { needsAttention }), + }, + }); + if (nextRun !== null) { + yield* createRun(threadId, "completed", 3, nextRun); + } + const notice = (yield* store.getThreadProjection(threadId)).turnItems.find( + (item) => item.id === TurnItemId.make(`notice:${runId}`), + ); + assert.equal(notice?.type, "system_notice"); + if (notice?.type !== "system_notice") return; + assert.equal(notice.needsAttention, needsAttention, name); + assert.deepEqual( + (yield* store.getSettlementCandidates(threadId)).map((candidate) => candidate.id), + eligible ? [threadId] : [], + name, + ); + } + }).pipe(Effect.provide(testLayer)), +); + const pullRequestLink = (number: number) => ({ host: "github.com", repository: "owner/repository", @@ -458,7 +523,7 @@ it.effect( ); assert.isTrue(queries.every(([query]) => !query.includes("COUNT("))); const pendingQuery = queries.find(([query]) => - query.includes("FROM orchestration_v2_projection_turn_items"), + query.includes("FROM orchestration_v2_projection_turn_items i"), ); assert.isDefined(pendingQuery); const metadata = yield* store.getThread(threadId); @@ -469,6 +534,19 @@ it.effect( pendingQuery![1], ); assert.isTrue(plan.some((row) => row.detail.includes("turn_items_shell_pending_idx"))); + const candidateQuery = queries.find(([query]) => query.includes("AS latest_run_id")); + assert.isDefined(candidateQuery); + const candidatePlan = yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${candidateQuery![0]}`, + candidateQuery![1], + ); + assert.isTrue( + candidatePlan.some((row) => + row.detail.startsWith( + "SEARCH notice USING INDEX orchestration_v2_projection_turn_items_attention_notice_idx (thread_id=? AND run_id=?)", + ), + ), + ); const countsPlan = yield* sql<{ readonly detail: string }>`EXPLAIN QUERY PLAN SELECT thread_id, run_id, COUNT(*) FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} AND run_id IS NOT NULL GROUP BY thread_id, run_id`; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 002ef7bafa950..c7553d59fcd31 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -5648,6 +5648,21 @@ export const layer: Layer.Layer = SELECT 1 FROM orchestration_v2_projection_runtime_requests request WHERE request.thread_id = t.thread_id AND request.status = 'pending' ) + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_turn_items notice + INDEXED BY orchestration_v2_projection_turn_items_attention_notice_idx + WHERE notice.thread_id = t.thread_id + AND notice.run_id = ( + SELECT started.run_id FROM orchestration_v2_projection_runs started + WHERE started.thread_id = t.thread_id + AND json_extract(started.payload_json, '$.startedAt') IS NOT NULL + ORDER BY json_extract(started.payload_json, '$.startedAt') DESC, + started.ordinal DESC + LIMIT 1 + ) + AND notice.type = 'system_notice' + AND json_extract(notice.payload_json, '$.needsAttention') = 1 + ) ORDER BY t.updated_at ASC, t.thread_id ASC `; if (rows.length === 0) return []; @@ -6226,6 +6241,29 @@ export const layerMemory: Layer.Layer = Layer.effect( !runs.some(isActivityRunForShell) && !runtimeRequests.some((request) => request.status === "pending"), ) + .filter(({ runs, turnItems }) => { + // Queue order can differ from the order runs actually start. + let latestStartedRun: OrchestrationV2Run | undefined; + let latestStartedAt = -Infinity; + for (const run of runs) { + if (run.startedAt === null) continue; + const startedAt = DateTime.toEpochMillis(run.startedAt); + if ( + latestStartedRun === undefined || + startedAt > latestStartedAt || + (startedAt === latestStartedAt && run.ordinal > latestStartedRun.ordinal) + ) { + latestStartedRun = run; + latestStartedAt = startedAt; + } + } + return !turnItems.some( + (item) => + item.runId === latestStartedRun?.id && + item.type === "system_notice" && + item.needsAttention === true, + ); + }) .map((projection) => { const shell = threadShellFromProjection(projection); return { diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 2b4981e04d1b9..a56541229dd9a 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -67,6 +67,7 @@ import Migration0061 from "./Migrations/061_PullRequestReviewGuides.ts"; import Migration0062 from "./Migrations/062_ThreadFailureLookupIndex.ts"; import Migration0063 from "./Migrations/063_DropUnusedV1Projections.ts"; import Migration0064 from "./Migrations/064_CanonicalizeThreadPullRequestLinks.ts"; +import Migration0065 from "./Migrations/065_ThreadAttentionNoticeLookupIndex.ts"; export const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -134,6 +135,7 @@ export const migrationEntries = [ [62, "ThreadFailureLookupIndex", Migration0062], [63, "DropUnusedV1Projections", Migration0063], [64, "CanonicalizeThreadPullRequestLinks", Migration0064], + [65, "ThreadAttentionNoticeLookupIndex", Migration0065], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts index 447c365082f27..26bec0293f052 100644 --- a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts +++ b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts @@ -13,7 +13,7 @@ layer("055_OrchestrationV2", (it) => { Effect.sync(() => { assert.deepStrictEqual( migrationEntries.map(([id]) => id), - Array.from({ length: 64 }, (_, index) => index + 1), + Array.from({ length: 65 }, (_, index) => index + 1), ); }), ); @@ -36,6 +36,7 @@ layer("055_OrchestrationV2", (it) => { [62, "ThreadFailureLookupIndex"], [63, "DropUnusedV1Projections"], [64, "CanonicalizeThreadPullRequestLinks"], + [65, "ThreadAttentionNoticeLookupIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); @@ -66,6 +67,7 @@ layer("055_OrchestrationV2", (it) => { { migration_id: 62, name: "ThreadFailureLookupIndex" }, { migration_id: 63, name: "DropUnusedV1Projections" }, { migration_id: 64, name: "CanonicalizeThreadPullRequestLinks" }, + { migration_id: 65, name: "ThreadAttentionNoticeLookupIndex" }, ]); const tables = yield* sql<{ readonly name: string }>` diff --git a/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts b/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts index 706cc02760eac..69eb8eff57027 100644 --- a/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts +++ b/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts @@ -123,8 +123,10 @@ layer("064_CanonicalizeThreadPullRequestLinks", (it) => { 'user', ${v1Payload}, '{}', 1) `; - assert.deepEqual(yield* runMigrations(), [[64, "CanonicalizeThreadPullRequestLinks"]]); - assert.deepEqual(yield* runMigrations(), []); + assert.deepEqual(yield* runMigrations({ toMigrationInclusive: 64 }), [ + [64, "CanonicalizeThreadPullRequestLinks"], + ]); + assert.deepEqual(yield* runMigrations({ toMigrationInclusive: 64 }), []); const rows = yield* sql<{ readonly thread_id: string; readonly payload_json: string }>` SELECT thread_id, payload_json FROM orchestration_v2_projection_threads `; diff --git a/apps/server/src/persistence/Migrations/065_ThreadAttentionNoticeLookupIndex.ts b/apps/server/src/persistence/Migrations/065_ThreadAttentionNoticeLookupIndex.ts new file mode 100644 index 0000000000000..991285faf8578 --- /dev/null +++ b/apps/server/src/persistence/Migrations/065_ThreadAttentionNoticeLookupIndex.ts @@ -0,0 +1,12 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_attention_notice_idx + ON orchestration_v2_projection_turn_items(thread_id, run_id) + WHERE type = 'system_notice' AND json_extract(payload_json, '$.needsAttention') = 1 + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 802c5110490ac..b70f30fb8ca85 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -44,6 +44,7 @@ describe("V2 preview upgrade", () => { [62, "ThreadFailureLookupIndex"], [63, "DropUnusedV1Projections"], [64, "CanonicalizeThreadPullRequestLinks"], + [65, "ThreadAttentionNoticeLookupIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -131,6 +132,7 @@ describe("V2 preview upgrade", () => { [62, "ThreadFailureLookupIndex"], [63, "DropUnusedV1Projections"], [64, "CanonicalizeThreadPullRequestLinks"], + [65, "ThreadAttentionNoticeLookupIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index f98d9e9143703..c76726e271813 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -149,8 +149,9 @@ Press `Escape` while dragging to cancel. By default, environments settle inactive threads after three days and settle threads whose pull request merged. A closed pull request can also settle an idle thread. Work in progress, pending questions or approvals, and live background work -prevent automatic settlement. An open pull request does not prevent inactivity -settlement unless you turn on **Keep threads with a PR** in **Settings → General**, +prevent automatic settlement. A notice that Claude denied an action also prevents +it until another turn starts, including an automatic continuation. An open pull +request does not prevent inactivity settlement unless you turn on **Keep threads with a PR** in **Settings → General**, but an old closed or merged pull request does not settle work you resumed after it closed. Only your own messages count as resuming. A turn that finished background work or a pull request watch starts on its own does not. diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 94a9eca331d13..339d6abe85fff 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1447,6 +1447,7 @@ export const OrchestrationV2TurnItem = Schema.Union([ ...OrchestrationV2TurnItemBaseFields, type: Schema.Literal("system_notice"), message: Schema.String, + needsAttention: Schema.optional(Schema.Boolean), }), Schema.Struct({ ...OrchestrationV2TurnItemBaseFields, @@ -2203,6 +2204,7 @@ export const OrchestrationV2TurnItemJson = Schema.Union([ ...OrchestrationV2TurnItemJsonBaseFields, type: Schema.Literal("system_notice"), message: Schema.String, + needsAttention: Schema.optional(Schema.Boolean), }), Schema.Struct({ ...OrchestrationV2TurnItemJsonBaseFields,