Skip to content
Closed
121 changes: 118 additions & 3 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ import * as Queue from "effect/Queue";
import * as Schema from "effect/Schema";
import * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";
import * as TestClock from "effect/testing/TestClock";
import { Tool } from "effect/ai";
import { formatClaudeResumeCompactionQuestion } from "@t3tools/shared/claudeCompaction";
import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
Expand Down Expand Up @@ -2085,14 +2086,15 @@ describe("ClaudeAdapterV2 background wake turns", () => {
readonly close?: (sdkMessages: Queue.Queue<SDKMessage>) => Effect.Effect<void>;
readonly interrupt?: Effect.Effect<void>;
readonly environment?: NodeJS.ProcessEnv;
readonly freshMessageQueue?: boolean;
}) =>
Effect.gen(function* () {
const fileSystem = yield* FileSystem.FileSystem;
const idAllocator = yield* IdAllocator.IdAllocatorV2;
const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({
prefix: "t3-claude-v2-wake-",
});
const sdkMessages = yield* Queue.unbounded<SDKMessage>();
let sdkMessages = yield* Queue.unbounded<SDKMessage>();
const processedMessages = new WeakMap<SDKMessage, Deferred.Deferred<void>>();
const offerAndWait = Effect.fnUntraced(function* (message: SDKMessage) {
const processed = yield* Deferred.make<void>();
Expand Down Expand Up @@ -2126,7 +2128,8 @@ describe("ClaudeAdapterV2 background wake turns", () => {
queryRunner: {
allocateSessionId: Effect.succeed(WAKE_NATIVE_SESSION),
open: (input) =>
Effect.sync(() => {
Effect.gen(function* () {
if (options?.freshMessageQueue) sdkMessages = yield* Queue.unbounded<SDKMessage>();
openedOptions = input.options;
return {
messages: Stream.fromQueue(sdkMessages).pipe(
Expand Down Expand Up @@ -2204,7 +2207,9 @@ describe("ClaudeAdapterV2 background wake turns", () => {
runtime,
providerThread,
threadId,
sdkMessages,
get sdkMessages() {
return sdkMessages;
},
offerAndWait,
offeredMessages,
permissionModeChanges,
Expand Down Expand Up @@ -4996,6 +5001,116 @@ describe("ClaudeAdapterV2 background wake turns", () => {
),
);

it.effect("discards held plans when an interrupt finalizes the turn before its echo", () =>
Effect.gen(function* () {
const interrupted = yield* Deferred.make<void>();
const harness = yield* makeWakeHarnessWithOptions({
freshMessageQueue: true,
interrupt: Deferred.succeed(interrupted, undefined).pipe(Effect.asVoid),
});
const now = yield* DateTime.now;
const firstAttempt = RunAttemptId.make("held-plan-first");
const secondAttempt = RunAttemptId.make("held-plan-second");
const thirdAttempt = RunAttemptId.make("held-plan-third");
const stamp = Effect.fnUntraced(function* (frame: SDKMessage, attempt: RunAttemptId) {
return claudeSdkFrame({
...frame,
user_message_uuid: yield* ClaudeAdapterV2.claudePromptUuid(attempt),
});
});
const start = (attemptId: RunAttemptId, providerTurnOrdinal: number) =>
harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now,
attemptId,
providerTurnOrdinal,
text: "Plan the task.",
attachments: [],
}),
);
yield* start(firstAttempt, 1);
yield* harness.offerAndWait(
yield* stamp(
makeAssistantTextFrame({ uuid: "00000000-0000-4000-8000-000000000801", text: "One." }),
firstAttempt,
),
);
yield* harness.offerAndWait(
yield* stamp(
makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000802", result: "One." }),
firstAttempt,
),
);
yield* Queue.take(harness.terminalReceipts);
yield* start(secondAttempt, 2);
const toolUseId = "held-exit-plan";
const planFrame = claudeSdkFrame({
type: "assistant",
message: {
model: "claude-sonnet-4-6",
id: "msg_held_plan",
type: "message",
role: "assistant",
content: [{ type: "tool_use", id: toolUseId, name: "ExitPlanMode", input: {} }],
},
parent_tool_use_id: null,
uuid: "00000000-0000-4000-8000-000000000803",
session_id: WAKE_NATIVE_SESSION,
});
yield* harness.offerAndWait(planFrame);
const canUseTool = harness.getOpenedOptions()?.canUseTool;
assert.isFunction(canUseTool);
const signal = yield* Effect.abortSignal;
yield* Effect.promise(() =>
canUseTool!(
"ExitPlanMode",
{ plan: "OLD PLAN" },
{
signal,
toolUseID: toolUseId,
requestId: "held-plan-request",
},
),
);
const active = harness.events.findLast(
(event) =>
event.type === "provider_turn.updated" && event.providerTurn.status === "running",
);
assert.equal(active?.type, "provider_turn.updated");
if (active?.type !== "provider_turn.updated") return;
const stopping = yield* harness.runtime
.interruptTurn({
providerThread: harness.providerThread,
providerTurnId: active.providerTurn.id,
})
.pipe(Effect.forkScoped);
yield* Deferred.await(interrupted);
// This process never closes; exercise the adapter's forced-finalization path.
yield* TestClock.adjust("10 seconds");
yield* Fiber.join(stopping);
assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted");
yield* start(thirdAttempt, 3);
yield* harness.offerAndWait(yield* stamp(planFrame, thirdAttempt));
yield* harness.offerAndWait(
yield* stamp(
makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000804", result: "Three." }),
thirdAttempt,
),
);
yield* Queue.take(harness.terminalReceipts);
assert.isFalse(
harness.events.some(
(event) =>
event.type === "turn_item.updated" &&
event.turnItem.type === "proposed_plan" &&
event.turnItem.markdown === "OLD PLAN",
),
);
}).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
);

it.effect("buffers wake output and requests a single continuation run", () =>
Effect.scoped(
Effect.gen(function* () {
Expand Down
15 changes: 15 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3655,16 +3655,29 @@ export function makeClaudeAdapterV2(
yield* endClaudeMonitorTasks((_taskId, task) => task.nativeThreadId === nativeThreadId);
});

const clearHeldPlansForFrames = (frames: ReadonlyArray<SDKMessage>) => {
for (const frame of frames) {
for (const toolUse of claudeToolUseBlocksFromAssistantMessage(frame)) {
heldProposedPlansByToolUseId.delete(toolUse.id);
}
}
};
// Drop idle wake traffic for a dead native process so it cannot pin
// session-wide pending work after sibling query replacement.
const clearWakeStateForNativeThread = (nativeThreadId: string) =>
Effect.gen(function* () {
yield* clearWakeReports(nativeThreadId);
yield* Ref.update(wakeReportsByNativeThread, (current) => {
const userTurns = new Map(current.userTurns);
userTurns.delete(nativeThreadId);
return { userTurns, reports: current.reports };
});
yield* Ref.update(wakeBuffers, (current) => {
if (!current.has(nativeThreadId)) {
return current;
}
const updated = new Map(current);
clearHeldPlansForFrames(current.get(nativeThreadId)?.messages ?? []);
updated.delete(nativeThreadId);
return updated;
});
Expand Down Expand Up @@ -4902,6 +4915,8 @@ export function makeClaudeAdapterV2(
readonly threadDisposition?: "reusable" | "broken";
readonly result?: SDKResultMessage;
}) {
clearHeldPlansForFrames(input.context.heldRootFrames);
input.context.heldRootFrames.length = 0;
yield* reasoningDeltas.flushTurn(input.context.nativeTurnId);
for (const toolCall of input.context.toolCalls.values()) {
const artifacts = buildToolCallArtifacts({
Expand Down
Loading
Loading