Skip to content
169 changes: 131 additions & 38 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5295,45 +5295,57 @@ describe("ClaudeAdapterV2 background wake turns", () => {
];
}

const makeTaskToolHarness = Effect.gen(function* () {
const harness = yield* makeWakeHarness;
let sequence = 600;
let turn = 0;
const nextUuid = () => `00000000-0000-4000-8000-${String(++sequence).padStart(12, "0")}`;
const startTurn = Effect.fnUntraced(function* () {
turn++;
yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now: yield* DateTime.now,
attemptId: RunAttemptId.make(`attempt-claude-task-tools-${turn}`),
providerTurnOrdinal: turn,
text: "Update the task list.",
attachments: [],
}),
);
});
const call = Effect.fnUntraced(function* (
tool: TaskToolCall,
options?: { parentToolUseId?: string; isError?: boolean; extraToolResult?: boolean },
) {
for (const frame of taskToolFrames(tool, nextUuid, options)) {
yield* harness.offerAndWait(frame);
}
});
const finishTurn = Effect.fnUntraced(function* () {
yield* harness.offerAndWait(
makeResultFrame({ uuid: nextUuid(), result: "Task list updated." }),
);
yield* Queue.take(harness.terminalReceipts);
const makeTaskToolHarnessWithOptions = (environment: NodeJS.ProcessEnv = {}) =>
Effect.gen(function* () {
const configDir = yield* (yield* FileSystem.FileSystem).makeTempDirectoryScoped({
prefix: "t3-claude-tasks-",
});
const path = yield* Path.Path;
const harness = yield* makeWakeHarnessWithOptions({
environment: {
CLAUDE_CONFIG_DIR: path.relative(CLAUDE_TEST_RUNTIME_POLICY.cwd ?? ".", configDir),
...environment,
},
});
let sequence = 600;
let turn = 0;
const nextUuid = () => `00000000-0000-4000-8000-${String(++sequence).padStart(12, "0")}`;
const startTurn = Effect.fnUntraced(function* () {
turn++;
yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now: yield* DateTime.now,
attemptId: RunAttemptId.make(`attempt-claude-task-tools-${turn}`),
providerTurnOrdinal: turn,
text: "Update the task list.",
attachments: [],
}),
);
});
const call = Effect.fnUntraced(function* (
tool: TaskToolCall,
options?: { parentToolUseId?: string; isError?: boolean; extraToolResult?: boolean },
) {
for (const frame of taskToolFrames(tool, nextUuid, options)) {
yield* harness.offerAndWait(frame);
}
});
const finishTurn = Effect.fnUntraced(function* () {
yield* harness.offerAndWait(
makeResultFrame({ uuid: nextUuid(), result: "Task list updated." }),
);
return yield* Queue.take(harness.terminalReceipts);
});
const plans = () =>
harness.events.flatMap((event) =>
event.type === "plan.updated" && event.plan.kind === "todo_list" ? [event.plan] : [],
);
return { configDir, startTurn, call, finishTurn, plans };
});
const plans = () =>
harness.events.flatMap((event) =>
event.type === "plan.updated" && event.plan.kind === "todo_list" ? [event.plan] : [],
);
return { startTurn, call, finishTurn, plans };
});

const makeTaskToolHarness = makeTaskToolHarnessWithOptions();

const createTask = (id: string, subject: string): TaskToolCall => ({
name: "TaskCreate",
Expand Down Expand Up @@ -5494,6 +5506,87 @@ describe("ClaudeAdapterV2 background wake turns", () => {
),
);

it.effect.each([
{
tool: updateTask({ taskId: "2", status: "completed" }),
listId: undefined,
directory: WAKE_NATIVE_SESSION,
status: "completed",
},
{
tool: createTask("2", "Implement"),
listId: " list/ID ",
directory: "list-ID",
status: "pending",
},
] as const)(
"recovers persisted tasks on the first $tool.name after restart",
({ tool, listId, directory, status }) =>
Effect.gen(function* () {
const fs = yield* FileSystem.FileSystem;
const harness = yield* makeTaskToolHarnessWithOptions({ CLAUDE_CODE_TASK_LIST_ID: listId });
const tasksDir = `${harness.configDir}/tasks/${directory}`;
yield* fs.makeDirectory(tasksDir, { recursive: true });
// The native tool has already written its result before emitting the SDK frame.
for (const task of [
{ id: "1", subject: "Inspect", status: "completed" },
{ id: "2", subject: "Implement", status },
{ id: "3", subject: "Internal", status: "pending", metadata: { _internal: true } },
]) {
yield* fs.writeFileString(
`${tasksDir}/${task.id}.json`,
JSON.stringify({ description: "", blocks: [], blockedBy: [], ...task }),
);
}
yield* fs.writeFileString(`${tasksDir}/4.json`, "{");
yield* fs.makeDirectory(`${tasksDir}/5.json`);

yield* harness.startTurn();
yield* harness.call(tool);
assert.equal((yield* harness.finishTurn()).status, "completed");
assert.deepEqual(
harness.plans().map((plan) => plan.steps),
[
[
{ id: "task-1", text: "Inspect", status: "completed" },
{ id: "task-2", text: "Implement", status },
],
],
);
}).pipe(
Effect.scoped,
Effect.provide(
Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
),
),
);

it.effect(
"publishes an empty task list when the first TaskUpdate after restart deletes the last task",
() =>
Effect.gen(function* () {
const fs = yield* FileSystem.FileSystem;
const harness = yield* makeTaskToolHarness;
const tasksDir = `${harness.configDir}/tasks/${WAKE_NATIVE_SESSION}`;
// Claude has removed the task file before emitting the SDK frame.
yield* fs.makeDirectory(tasksDir, { recursive: true });
yield* fs.writeFileString(`${tasksDir}/.highwatermark`, "1");

yield* harness.startTurn();
yield* harness.call(updateTask({ taskId: "1", status: "deleted" }));
assert.equal((yield* harness.finishTurn()).status, "completed");
assert.deepEqual(
harness.plans().map((plan) => plan.steps),
[[]],
);
}).pipe(
Effect.scoped,
Effect.provide(
Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
),
),
);

it.effect.each(["success false", "error result", "unknown id"] as const)(
"does not emit a plan update for TaskUpdate with %s",
(reason) =>
Expand Down
91 changes: 80 additions & 11 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ import {
import {
type ChatAttachment,
ClaudeSettings,
TrimmedNonEmptyString,
defaultInstanceIdForDriver,
type ModelSelection,
type OrchestrationV2ConversationMessage,
Expand Down Expand Up @@ -128,6 +129,7 @@ import { signalProcessGroup } from "@t3tools/provider-core/server/processGroup";
import {
claudeSignedOutMessage,
makeClaudeEnvironment,
resolveClaudeHomePath,
} from "../../provider/Drivers/ClaudeHome.ts";
import {
BUNDLED_CLAUDE_MODEL_CATALOG,
Expand Down Expand Up @@ -3486,6 +3488,52 @@ function applyClaudeTaskTool(
return new Map(tasks).set(id, { text, status });
}

const CLAUDE_TASK_TOOL_NAMES = new Set(["TaskCreate", "TaskUpdate", "TaskList"]);
const decodeClaudeTaskFile = Schema.decodeUnknownOption(
Schema.fromJsonString(
Schema.Struct({
id: TrimmedNonEmptyString,
subject: TrimmedNonEmptyString,
status: Schema.Literals(["pending", "in_progress", "completed"]),
metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
}),
),
);

const readClaudeTaskList = Effect.fnUntraced(
function* (input: {
readonly environment: NodeJS.ProcessEnv;
readonly cwd: string | null | undefined;
readonly nativeThreadId: string;
}) {
const fileSystem = yield* FileSystem.FileSystem;
const path = yield* Path.Path;
const configDirOverride = input.environment.CLAUDE_CONFIG_DIR?.trim();
const configDir = configDirOverride
? path.resolve(input.cwd ?? ".", configDirOverride)
: yield* resolveClaudeHomePath({ homePath: "" });
const listId = input.environment.CLAUDE_CODE_TASK_LIST_ID?.trim() || input.nativeThreadId;
const directory = path.join(configDir, "tasks", listId.replace(/[^a-zA-Z0-9_-]/g, "-"));
const tasks = new Map<string, Omit<OrchestrationV2PlanStep, "id">>();
const names = (yield* fileSystem.readDirectory(directory))
.filter((name) => /^\d+\.json$/.test(name))
.toSorted((left, right) => Number.parseInt(left, 10) - Number.parseInt(right, 10));
for (const name of names) {
const contents = yield* fileSystem
.readFileString(path.join(directory, name))
.pipe(Effect.orElseSucceed(() => ""));
const task = Option.getOrUndefined(decodeClaudeTaskFile(contents));
if (task === undefined || task.metadata?._internal) continue;
tasks.set(task.id, {
text: task.subject,
status: task.status === "in_progress" ? "running" : task.status,
});
}
return tasks;
},
Effect.orElseSucceed((): ClaudeTaskList | undefined => undefined),
);

function claudeTaskSteps(tasks: ClaudeTaskList): ReadonlyArray<OrchestrationV2PlanStep> {
return [...tasks].map(([id, task]) => ({ id: `task-${id}`, ...task }));
}
Expand Down Expand Up @@ -3621,7 +3669,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
const openedNativeThreads = yield* Ref.make(new Set<string>());
const latestPlanByKind = yield* Ref.make(new Map<string, OrchestrationV2PlanArtifact>());
// Claude 5 task tools edit one list per native session, across turns.
const taskListsByNativeThread = yield* Ref.make(new Map<string, ClaudeTaskList>());
const taskListsByNativeThread = yield* Ref.make(
new Map<string, ClaudeTaskList | undefined>(),
);
const planIdsByNativeItem = yield* Ref.make(
new Map<string, OrchestrationV2PlanArtifact["id"]>(),
);
Expand Down Expand Up @@ -7862,19 +7912,38 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
output,
completedAt: yield* DateTime.now,
});
if (parentToolUseId === null && !isClaudeToolResultError(toolResult)) {
if (
parentToolUseId === null &&
!isClaudeToolResultError(toolResult) &&
CLAUDE_TASK_TOOL_NAMES.has(toolCall.toolName)
) {
const taskLists = yield* Ref.get(taskListsByNativeThread);
const taskList = applyClaudeTaskTool(
taskLists.get(liveQuery.nativeThreadId) ?? new Map(),
toolCall.toolName,
claudeNativeToolInputValue(toolCall.input),
claudeNativeToolOutputValue(output),
const knownTaskList = taskLists.get(liveQuery.nativeThreadId);
// TaskList already supplies a snapshot. Mutations need Claude's persisted
// list once; afterward the cache also records an unavailable store.
const storedTaskList =
!taskLists.has(liveQuery.nativeThreadId) && toolCall.toolName !== "TaskList"
? yield* readClaudeTaskList({
environment: adapterOptions.environment,
cwd: context.input.runtimePolicy.cwd,
nativeThreadId: liveQuery.nativeThreadId,
}).pipe(
Effect.provideService(FileSystem.FileSystem, fileSystem),
Effect.provideService(Path.Path, path),
)
: undefined;
const taskList =
applyClaudeTaskTool(
knownTaskList ?? storedTaskList ?? new Map(),
toolCall.toolName,
claudeNativeToolInputValue(toolCall.input),
claudeNativeToolOutputValue(output),
) ?? storedTaskList;
yield* Ref.set(
taskListsByNativeThread,
new Map(taskLists).set(liveQuery.nativeThreadId, taskList ?? knownTaskList),
);
if (taskList !== undefined) {
yield* Ref.set(
taskListsByNativeThread,
new Map(taskLists).set(liveQuery.nativeThreadId, taskList),
);
yield* emitClaudePlanProjection({
context,
nativeItemId: `claude-tasks:${context.providerTurnId}`,
Expand Down
Loading