Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
120 changes: 120 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7708,6 +7708,126 @@ describe("ClaudeAdapterV2 background wake turns", () => {
),
);

// regression: #24690 a PermissionRequest hook that answers first withdraws the approval card
it.effect("cancels a pending approval when Claude withdraws its permission request", () =>
Effect.scoped(
Effect.gen(function* () {
const harness = yield* makeWakeHarness;
const bashToolUseId = "toolu_01WithdrawnApprovalBash";
yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now: yield* DateTime.now,
attemptId: RunAttemptId.make("attempt-claude-withdrawn-approval"),
text: "Clean up.",
attachments: [],
runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY,
}),
);
const controller = new AbortController();
const permission = yield* Effect.promise(() =>
harness.getOpenedOptions()!.canUseTool!(
"Bash",
{ command: "rm -f $F/*" },
{
signal: controller.signal,
toolUseID: bashToolUseId,
requestId: "request-withdrawn-approval",
},
),
).pipe(Effect.forkScoped);
const pending = yield* Queue.take(harness.runtimeRequestReceipts);
assert.equal(pending.runtimeRequest.status, "pending");

// The CLI sends control_cancel_request, which the SDK turns into this abort.
controller.abort();
assert.notEqual((yield* Fiber.join(permission))?.behavior, "allow");

const cancelled = yield* Queue.take(harness.runtimeRequestReceipts);
assert.equal(cancelled.runtimeRequest.id, pending.runtimeRequest.id);
assert.equal(cancelled.runtimeRequest.status, "cancelled");
assert.isNotNull(cancelled.runtimeRequest.resolvedAt);
yield* harness.until(() =>
harness.events.some(
(event) =>
event.type === "turn_item.updated" &&
event.turnItem.nodeId === pending.runtimeRequest.nodeId &&
event.turnItem.status === "cancelled",
),
);
const node = harness.events.findLast(
(event) =>
event.type === "node.updated" && event.node.id === pending.runtimeRequest.nodeId,
);
assert.equal(node?.type === "node.updated" ? node.node.status : undefined, "cancelled");
}).pipe(
Effect.provide(
Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
),
),
),
);

it.effect("keeps a user-cancelled approval resolved when Claude aborts afterwards", () =>
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-claude-user-cancelled-approval"),
text: "Clean up.",
attachments: [],
runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY,
}),
);
const controller = new AbortController();
const permission = yield* Effect.promise(() =>
harness.getOpenedOptions()!.canUseTool!(
"Bash",
{ command: "rm -f $F/*" },
{
signal: controller.signal,
toolUseID: "toolu_01UserCancelledApprovalBash",
requestId: "request-user-cancelled-approval",
},
),
).pipe(Effect.forkScoped);
const pending = yield* Queue.take(harness.runtimeRequestReceipts);
yield* harness.runtime.respondToRuntimeRequest({
requestId: pending.runtimeRequest.id,
decision: "cancel",
});
// The SDK aborts the signal when an interrupting deny ends the turn, racing the answer.
controller.abort();
assert.equal((yield* Fiber.join(permission))?.behavior, "deny");

yield* Effect.promise(() =>
harness.getOpenedOptions()!.canUseTool!(
"Bash",
{ command: "rm -f $G/*" },
{
signal: new AbortController().signal,
toolUseID: "toolu_01UserCancelledApprovalNext",
requestId: "request-user-cancelled-approval-next",
},
),
).pipe(Effect.forkScoped);
// Events are ordered, so a withdrawal update for the answered request would arrive first.
const next = yield* Queue.take(harness.runtimeRequestReceipts);
assert.notEqual(next.runtimeRequest.id, pending.runtimeRequest.id);
assert.equal(next.runtimeRequest.status, "pending");
}).pipe(
Effect.provide(
Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
),
),
),
);

it.effect("releases output held for a prompt echo when the stream ends", () =>
Effect.scoped(
Effect.gen(function* () {
Expand Down
33 changes: 33 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9092,6 +9092,39 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
}),
),
);
// An aborted signal means Claude withdrew the request, e.g. a PermissionRequest hook answered
// first. Claiming the decision tells that apart from a user answer the orchestrator already resolved.
if (
callbackOptions.signal.aborted &&
(yield* Deferred.succeed(decision, resolvedDecision))
) {
const now = yield* DateTime.now;
yield* Effect.all(
[
emitProviderEvent({
type: "node.updated",
driver: CLAUDE_PROVIDER,
node: { ...artifacts.node, status: "cancelled", completedAt: now },
}),
emitProviderEvent({
type: "runtime_request.updated",
driver: CLAUDE_PROVIDER,
runtimeRequest: { ...artifacts.request, status: "cancelled", resolvedAt: now },
}),
emitProviderEvent({
type: "turn_item.updated",
driver: CLAUDE_PROVIDER,
turnItem: {
...artifacts.turnItem,
status: "cancelled",
completedAt: now,
updatedAt: now,
},
}),
],
{ concurrency: 1 },
);
}

return permissionResultFromDecision({
toolName,
Expand Down
65 changes: 45 additions & 20 deletions apps/server/src/orchestration-v2/EventSink.ts
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ export interface EventSinkV2Shape {
OrchestrationV2ProviderThread,
"driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef"
>;
readonly guardPendingUserInputCancellations?: boolean;
readonly guardPendingRuntimeRequestCancellations?: boolean;
readonly commandId?: CommandId;
readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
}) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventSinkV2Error>;
Expand All @@ -87,13 +87,13 @@ export interface EventSinkV2Shape {
OrchestrationV2ProviderThread,
"driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef"
>;
readonly guardPendingUserInputCancellations?: boolean;
readonly guardPendingRuntimeRequestCancellations?: boolean;
readonly commandId?: CommandId;
readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
readonly effects: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
}) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventSinkV2Error>;
readonly writeIfRunCurrent: (input: {
readonly guardPendingUserInputCancellations?: boolean;
readonly guardPendingRuntimeRequestCancellations?: boolean;
readonly commandId?: CommandId;
readonly threadId: ThreadId;
readonly runId: RunId;
Expand All @@ -110,7 +110,7 @@ export interface EventSinkV2Shape {
>;
/** Commits only if the expected attempt and ordinal still own the provider thread, protecting newer attempts. */
readonly writeIfProviderThreadOwner: (input: {
readonly guardPendingUserInputCancellations?: boolean;
readonly guardPendingRuntimeRequestCancellations?: boolean;
readonly commandId?: CommandId;
readonly providerThreadId: ProviderThreadId;
readonly runId: RunId;
Expand Down Expand Up @@ -278,29 +278,53 @@ const layerBase: Layer.Layer<
);
});

// A user can answer after terminal normalization reads the pending request.
// A user can answer after terminal normalization reads the pending request, and an approval
// answer commits before its outbox delivery reaches a provider that may withdraw the request.
// Recheck inside the write transaction so stale cleanup cannot erase answers.
const guardUserInputCancellations =
const guardRuntimeRequestCancellations =
/* This helper runs for individual orchestration events; the enclosing operation owns the span. */
Effect.fnUntraced(function* (events: ReadonlyArray<OrchestrationV2DomainEvent>) {
const staleRequests = new Set<RuntimeRequestId>();
const staleNodes = new Set<NodeId>();
for (const event of events) {
// Providers can withdraw the node and item in writes of their own, so check each against its request.
if (
event.type !== "runtime-request.updated" ||
event.payload.kind !== "user_input" ||
event.payload.status !== "cancelled"
)
event.type === "node.updated" &&
event.payload.status === "cancelled" &&
event.payload.runtimeRequestId !== null
) {
const current = yield* projectionStore.getRuntimeRequest(
event.threadId,
event.payload.runtimeRequestId,
);
if (current?.status === "resolved") staleNodes.add(event.payload.id);
continue;
}
if (
event.type === "turn-item.updated" &&
(event.payload.type === "user_input_request" ||
event.payload.type === "approval_request") &&
event.payload.status === "cancelled"
) {
const current = yield* projectionStore.getRuntimeRequest(
event.threadId,
event.payload.requestId,
);
if (current?.status === "resolved") staleRequests.add(event.payload.requestId);
continue;
}
if (event.type !== "runtime-request.updated" || event.payload.status !== "cancelled")
continue;
const current = yield* projectionStore.getRuntimeRequest(
event.threadId,
event.payload.id,
);
if (
current?.status !== "pending" ||
current.kind !== "user_input" ||
current.providerTurnId !== event.payload.providerTurnId ||
current.responseCapability.type === "message"
(event.payload.kind === "user_input" &&
(current.kind !== "user_input" ||
current.providerTurnId !== event.payload.providerTurnId ||
current.responseCapability.type === "message"))
) {
staleRequests.add(event.payload.id);
staleNodes.add(event.payload.nodeId);
Expand All @@ -314,7 +338,8 @@ const layerBase: Layer.Layer<
return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id);
case "turn-item.updated":
return (
event.payload.type !== "user_input_request" ||
(event.payload.type !== "user_input_request" &&
event.payload.type !== "approval_request") ||
event.payload.status !== "cancelled" ||
!staleRequests.has(event.payload.requestId)
);
Expand Down Expand Up @@ -430,8 +455,8 @@ const layerBase: Layer.Layer<
)
: input.events;
const normalized = yield* normalizeEvents(
input.guardPendingUserInputCancellations === true
? yield* guardUserInputCancellations(events)
input.guardPendingRuntimeRequestCancellations === true
? yield* guardRuntimeRequestCancellations(events)
: events,
);
const committed = yield* eventStore.append({
Expand Down Expand Up @@ -489,8 +514,8 @@ const layerBase: Layer.Layer<
}

const normalized = yield* normalizeEvents(
input.guardPendingUserInputCancellations === true
? yield* guardUserInputCancellations(input.events)
input.guardPendingRuntimeRequestCancellations === true
? yield* guardRuntimeRequestCancellations(input.events)
: input.events,
);
const storedEvents = yield* eventStore.append({
Expand Down Expand Up @@ -554,8 +579,8 @@ const layerBase: Layer.Layer<
}

const normalized = yield* normalizeEvents(
input.guardPendingUserInputCancellations === true
? yield* guardUserInputCancellations(input.events)
input.guardPendingRuntimeRequestCancellations === true
? yield* guardRuntimeRequestCancellations(input.events)
: input.events,
);
const storedEvents = yield* eventStore.append({
Expand Down
Loading
Loading