# Files changed · PR \#21725 · pingdotgg/t3code
[View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/21725/diffs)
Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code)
Visibility: public
[fix\(claude\): reconcile stopped tasks after session replacement](https://git.cafe/pingdotgg/t3code/pull/21725.md)
Version: 8
Head commit: 7d26aae48cf644526eff70b6d4ae0bdcd57b6620
Base commit: ac83029f2f56a593085120c2ccacc3cae79ba57e
Comparison base: ac83029f2f56a593085120c2ccacc3cae79ba57e
Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing.
## apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts
Status: modified; +182; −0
```diff
@@ -10597,6 +10597,188 @@
),
);
+ it.effect("preserves transcript receipt source frames and excludes known shell tasks", () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const harness = yield* makeWakeHarness;
+ const readThrough = Effect.fnUntraced(function* (uuid: string) {
+ yield* harness.offerAndWait(
+ claudeSdkFrame({
+ type: "system",
+ subtype: "informational",
+ level: "warning",
+ content: uuid,
+ uuid,
+ session_id: WAKE_NATIVE_SESSION,
+ }),
+ );
+ yield* Queue.take(harness.systemNoticeReceipts);
+ });
+ const objective = "all tests pass";
+ const receiptText = (taskId: string) =>
+ `${taskId}stoppedPrevious session ended`;
+ const transcript = (taskId: string, uuid: string) =>
+ claudeSdkFrame({
+ type: "user",
+ isSynthetic: true,
+ parent_tool_use_id: null,
+ session_id: WAKE_NATIVE_SESSION,
+ uuid,
+ message: { role: "user", content: receiptText(taskId) },
+ });
+ yield* harness.runtime.startTurn(
+ makeClaudeTestTurnInput({
+ threadId: harness.threadId,
+ providerThread: {
+ ...harness.providerThread,
+ goal: { objective, status: "active", checks: 0 },
+ },
+ now: yield* DateTime.now,
+ attemptId: RunAttemptId.make("transcript-source"),
+ text: "Continue",
+ attachments: [],
+ }),
+ );
+ yield* harness.offerAndWait(
+ claudeSdkFrame({
+ ...transcript("old-task", "goal-receipt"),
+ user_message_uuid: harness.promptUuid(0),
+ message: {
+ role: "user",
+ content: `Stop hook feedback:\n[${objective}]: One test still fails.\n${receiptText("old-task")}`,
+ },
+ }),
+ );
+ yield* harness.offerAndWait(
+ claudeSdkFrame({
+ ...transcript("old-task", "user-quoted-receipt"),
+ isSynthetic: false,
+ }),
+ );
+ const taskId = "registered-task";
+ const taskStatuses = () =>
+ harness.events.flatMap((event) =>
+ event.type === "subagent.updated" && event.subagent.nativeTaskRef?.nativeId === taskId
+ ? [event.subagent.status]
+ : [],
+ );
+ yield* harness.offerAndWait(
+ makeSubagentTaskStartedFrame({ taskId, toolUseId: "launch-task", uuid: "task-started" }),
+ );
+ yield* harness.offerAndWait(
+ claudeSdkFrame({
+ ...makeSubagentNotificationFrame({
+ taskId,
+ toolUseId: "launch-task",
+ summary: "Previous session ended",
+ uuid: "task-stopped",
+ }),
+ status: "stopped",
+ }),
+ );
+ yield* harness.offerAndWait(
+ makeSubagentTaskStartedFrame({ taskId, toolUseId: "resume-task", uuid: "task-resumed" }),
+ );
+ yield* readThrough("before-registered-replay");
+ assert.deepEqual(taskStatuses(), ["running", "cancelled", "running"]);
+ yield* harness.offerAndWait(transcript(taskId, "registered-receipt"));
+ yield* harness.offerAndWait(transcript(taskId, "registered-receipt-replayed"));
+ yield* readThrough("after-registered-replay");
+ assert.deepEqual(taskStatuses(), ["running", "cancelled", "running"]);
+ yield* harness.offerAndWait(
+ makeSubagentNotificationFrame({
+ taskId,
+ toolUseId: "resume-task",
+ summary: "Resumed work completed",
+ uuid: "task-completed",
+ }),
+ );
+ yield* harness.offerAndWait(wakeTaskStarted);
+ yield* harness.offerAndWait(transcript(WAKE_TASK_ID, "shell-receipt"));
+ yield* harness.offerAndWait(
+ claudeSdkFrame({
+ ...wakeTaskStarted,
+ task_id: "other-shell",
+ tool_use_id: "other-shell-tool",
+ uuid: "other-shell-start",
+ }),
+ );
+ yield* harness.offerAndWait(
+ claudeSdkFrame({
+ type: "system",
+ subtype: "task_notification",
+ task_id: "other-shell",
+ status: "stopped",
+ summary: "Shell ended",
+ session_id: WAKE_NATIVE_SESSION,
+ uuid: "shell-sdk-receipt",
+ }),
+ );
+ yield* harness.offerAndWait(turnOneResult);
+ yield* Queue.take(harness.terminalReceipts);
+ assert.deepEqual(taskStatuses(), ["running", "cancelled", "running", "completed"]);
+ assert.deepEqual(
+ providerThreadRosterEvents(harness.events).at(-1)?.providerThread.pendingBackgroundTasks,
+ [],
+ );
+ assert.deepEqual(
+ harness.events.flatMap((event) =>
+ event.type === "subagent.native-task-ended" ? [event.nativeTaskId] : [],
+ ),
+ ["old-task"],
+ );
+ assert.isTrue(
+ harness.events.some(
+ (event) =>
+ event.type === "provider_thread.updated" && event.providerThread.goal?.checks === 1,
+ ),
+ );
+ yield* harness.runtime.startTurn(
+ makeClaudeTestTurnInput({
+ threadId: harness.threadId,
+ providerThread: harness.providerThread,
+ now: yield* DateTime.now,
+ attemptId: RunAttemptId.make("transcript-echo"),
+ text: "Continue again",
+ attachments: [],
+ providerTurnOrdinal: 2,
+ }),
+ );
+ const hasReply = () =>
+ harness.events.some(
+ (event) =>
+ event.type === "message.updated" &&
+ event.message.text === "Echo released this reply.",
+ );
+ yield* harness.offerAndWait(
+ makeAssistantTextFrame({
+ uuid: "held-reply",
+ text: "Echo released this reply.",
+ }),
+ );
+ yield* readThrough("before-echo");
+ assert.isFalse(hasReply());
+ yield* harness.offerAndWait(
+ claudeSdkFrame({
+ ...transcript("echo-task", "echo-receipt"),
+ user_message_uuid: harness.promptUuid(1),
+ }),
+ );
+ yield* readThrough("after-echo");
+ assert.isTrue(hasReply());
+ yield* harness.offerAndWait(makeResultFrame({ uuid: "echo-result", result: "" }));
+ yield* Queue.take(harness.terminalReceipts);
+ assert.isFalse(yield* harness.hasPendingBackgroundWork);
+ yield* harness.offerAndWait(transcript("another-old-task", "idle-receipt"));
+ assert.isTrue(yield* harness.hasPendingBackgroundWork);
+ }).pipe(
+ Effect.provide(
+ Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
+ ),
+ ),
+ ),
+ );
+
it.effect("extracts text from direct content-block subagent results", () =>
Effect.scoped(
Effect.gen(function* () {
```
## apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Status: modified; +72; −0
```diff
@@ -1305,6 +1305,46 @@
return content.flatMap((part) => (part.type === "text" ? [part.text] : [])).join("");
}
+/** Resume transcripts can report ended tasks as synthetic XML user messages. */
+function claudeTranscriptTaskNotifications(
+ message: SDKMessage,
+): ReadonlyArray> {
+ if (
+ message.type !== "user" ||
+ message.isSynthetic !== true ||
+ message.parent_tool_use_id !== null ||
+ message.session_id === undefined ||
+ message.uuid === undefined
+ )
+ return [];
+ const content = message.message.content;
+ const text =
+ typeof content === "string"
+ ? content
+ : content.flatMap((part) => (part.type === "text" ? [part.text] : [])).join("\n");
+ return Array.from(text.matchAll(/([\s\S]*?)<\/task-notification>/g)).flatMap(
+ (match) => {
+ const body = match[1]!;
+ const taskId = /([^<]+)<\/task-id>/.exec(body)?.[1]?.trim();
+ const status = /(completed|failed|stopped)<\/status>/.exec(body)?.[1];
+ const summary = /([\s\S]*?)<\/summary>/.exec(body)?.[1];
+ if (!taskId || !status || summary === undefined) return [];
+ return [
+ {
+ type: "system" as const,
+ subtype: "task_notification" as const,
+ task_id: taskId,
+ status: status as "completed" | "failed" | "stopped",
+ summary,
+ output_file: "",
+ uuid: message.uuid!,
+ session_id: message.session_id!,
+ },
+ ];
+ },
+ );
+}
+
// Claude 5.1 and later return the text written ahead of a tool call as a
// thinking block, at most one per call, instead of a text block. Summarized
// display makes it look like reasoning, so only the second of two thinking
@@ -7967,11 +8007,43 @@
}
});
+ const emitUntrackedTaskReceipt = Effect.fnUntraced(function* (message: SDKMessage) {
+ if (
+ message.type !== "system" ||
+ message.subtype !== "task_notification" ||
+ (yield* Ref.get(sessionSubagentsByTaskId)).has(message.task_id)
+ )
+ return;
+ const live = yield* Ref.get(queryContext);
+ if (
+ live === null ||
+ (yield* isKnownOpaqueBackgroundTaskOnNativeThread(live.nativeThreadId, message.task_id))
+ )
+ return;
+ yield* emitProviderEvent({
+ type: "subagent.native-task-ended",
+ driver: CLAUDE_PROVIDER,
+ threadId: input.threadId,
+ nativeTaskId: message.task_id,
+ status: claudeTaskOutcome(message.status),
+ result: message.summary,
+ });
+ });
+
const handleSdkMessage = Effect.fnUntraced(function* (input: {
readonly query: ClaudeAgentSdkQuerySession;
readonly message: SDKMessage;
}) {
const message = input.message;
+ const transcriptNotifications = claudeTranscriptTaskNotifications(message);
+ for (const notification of transcriptNotifications) {
+ // Transcript outcomes can predate a registered subagent's latest resume.
+ if ((yield* Ref.get(sessionSubagentsByTaskId)).has(notification.task_id)) continue;
+ yield* emitUntrackedTaskReceipt(notification);
+ yield* handleRoutedSdkMessage({ query: input.query, message: notification });
+ }
+ // The source frame still carries prompt echoes, goal updates, and wake evidence.
+ if (transcriptNotifications.length === 0) yield* emitUntrackedTaskReceipt(message);
const context = yield* Ref.get(activeTurn);
const liveQuery = yield* Ref.get(queryContext);
if (
```
## apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts
Status: modified; +233; −0
```diff
@@ -40,6 +40,7 @@
import * as ThreadLaunchService from "./ThreadLaunchService.ts";
import * as ThreadManagementService from "./ThreadManagementService.ts";
import * as ServerSettings from "../serverSettings.ts";
+import * as ProviderSessionManager from "./ProviderSessionManager.ts";
import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
@@ -478,3 +479,235 @@
),
),
);
+
+it.effect.each(["transcript", "sdk"] as const)(
+ "reconciles a %s stopped receipt through a replacement Claude process and durable projections",
+ (receiptKind) =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const cwd = yield* checkpointWorkspace("claude-stopped-task");
+ const processes: Array> = [];
+ const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({
+ instanceId: modelSelection.instanceId,
+ settings,
+ environment: {},
+ attachmentsDir: cwd,
+ fileSystem: yield* FileSystem.FileSystem,
+ path: yield* Path.Path,
+ crypto: yield* Crypto.Crypto,
+ idAllocator: yield* IdAllocator.IdAllocatorV2,
+ queryRunner: {
+ allocateSessionId: Effect.succeed(sessionId),
+ assertComplete: Effect.void,
+ subagentLaunchToolUseId: () => Effect.succeed(null),
+ open: () =>
+ Effect.gen(function* () {
+ const messages = yield* Queue.unbounded();
+ processes.push(messages);
+ return {
+ messages: Stream.fromQueue(messages),
+ offer: () => Effect.void,
+ setModel: () => Effect.void,
+ setPermissionMode: () => Effect.void,
+ applyFlagSettings: () => Effect.void,
+ backgroundTasks: () => Effect.succeed(false),
+ stopTask: () => Effect.void,
+ interrupt: Effect.void,
+ close: Queue.shutdown(messages),
+ };
+ }),
+ forkSession: () => Effect.die("unused forkSession"),
+ },
+ });
+ yield* Effect.gen(function* () {
+ const orchestrator = yield* Orchestrator.OrchestratorV2;
+ const sink = yield* EventSink.EventSinkV2;
+ const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
+ const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
+ const threadId = ThreadId.make("thread:stopped-task");
+ const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
+ Effect.gen(function* () {
+ // Capture the cursor before offering an SDK frame; a lazily opened
+ // live tail can miss a fast receipt that already committed.
+ const afterSequence = yield* sink.latestSequence();
+ return yield* sink.stream({ afterSequence }).pipe(
+ Stream.filter((stored) => predicate(stored.event)),
+ Stream.take(1),
+ Stream.runDrain,
+ Effect.forkScoped,
+ );
+ });
+ const dispatch = (id: string) =>
+ orchestrator.dispatch({
+ type: "message.dispatch",
+ commandId: CommandId.make(id),
+ threadId,
+ messageId: MessageId.make(id),
+ text: "Continue the review.",
+ attachments: [],
+ dispatchMode: { type: "start_immediately" },
+ createdBy: "user",
+ creationSource: "web",
+ });
+ yield* orchestrator.dispatch({
+ type: "thread.create",
+ commandId: CommandId.make("create"),
+ threadId,
+ projectId: ProjectId.make("project:stopped-task"),
+ title: "Stopped task",
+ modelSelection,
+ runtimeMode: "full-access",
+ interactionMode: "default",
+ branch: null,
+ worktreePath: cwd,
+ createdBy: "user",
+ creationSource: "web",
+ });
+ yield* dispatch("first");
+ yield* worker.drain();
+ assert.lengthOf(processes, 1);
+ const first = processes[0]!;
+ const spawned = yield* watch(
+ (event) =>
+ event.type === "subagent.updated" &&
+ event.payload.nativeTaskRef?.nativeId === "old-task",
+ );
+ yield* Queue.offer(
+ first,
+ frame({
+ type: "system",
+ subtype: "task_started",
+ task_id: "old-task",
+ tool_use_id: "old-tool",
+ description: "Review recent changes",
+ subagent_type: "general-purpose",
+ is_backgrounded: true,
+ task_type: "local_agent",
+ prompt: "Review recent changes.",
+ uuid: "spawn-old-task",
+ session_id: sessionId,
+ }),
+ );
+ yield* Fiber.join(spawned);
+ const finished = yield* watch(
+ (event) =>
+ event.type === "provider-turn.updated" && event.payload.status === "completed",
+ );
+ yield* Queue.offer(first, result("first-completed", false));
+ yield* Fiber.join(finished);
+ yield* worker.drain();
+ const before = yield* orchestrator.getThreadProjection(threadId);
+ const task = before.subagents.find(
+ (entry) => entry.nativeTaskRef?.nativeId === "old-task",
+ );
+ assert.isDefined(task);
+ assert.equal(task!.status, "running");
+ const providerSessionId = before.providerSessions.at(-1)!.id;
+ // Destroy the owning runtime; the replacement has an empty native-task registry.
+ yield* manager.release({
+ providerSessionId,
+ reason: "runtime_error",
+ detail: "fixture process exited",
+ });
+ yield* dispatch("resume");
+ yield* worker.drain();
+ assert.lengthOf(processes, 2);
+ assert.equal(
+ (yield* orchestrator.getThreadProjection(threadId)).subagents.find(
+ (entry) => entry.id === task!.id,
+ )?.status,
+ "running",
+ );
+ const second = processes[1]!;
+ yield* Queue.offer(
+ second,
+ frame({
+ type: "system",
+ subtype: "init",
+ uuid: "replacement-init",
+ session_id: sessionId,
+ }),
+ );
+ // A later persisted SDK frame proves the receipt has been ingested,
+ // even when the regression leaves the task running.
+ const processed = yield* watch(
+ (event) =>
+ event.type === "message.updated" && event.payload.text === "Receipt replay finished.",
+ );
+ const notification =
+ receiptKind === "transcript"
+ ? {
+ type: "user",
+ isSynthetic: true,
+ parent_tool_use_id: null,
+ session_id: sessionId,
+ uuid: "stopped-notification",
+ message: {
+ role: "user",
+ content:
+ "old-taskstoppedPrevious session ended",
+ },
+ }
+ : {
+ type: "system",
+ subtype: "task_notification",
+ task_id: "old-task",
+ status: "stopped",
+ summary: "Previous session ended",
+ session_id: sessionId,
+ uuid: "stopped-notification",
+ };
+ yield* Queue.offer(second, frame(notification));
+ yield* Queue.offer(
+ second,
+ frame({
+ type: "assistant",
+ uuid: "receipt-barrier",
+ session_id: sessionId,
+ parent_tool_use_id: null,
+ message: {
+ id: "receipt-barrier",
+ type: "message",
+ role: "assistant",
+ model: modelSelection.model,
+ content: [{ type: "text", text: "Receipt replay finished." }],
+ stop_reason: null,
+ stop_sequence: null,
+ usage: { input_tokens: 1, output_tokens: 1 },
+ },
+ }),
+ );
+ yield* Fiber.join(processed);
+ const after = yield* orchestrator.getThreadProjection(threadId);
+ const updated = after.subagents.find((entry) => entry.id === task!.id)!;
+ assert.equal(updated.status, "cancelled");
+ assert.equal(after.nodes.find((node) => node.id === task!.id)?.status, "cancelled");
+ assert.equal(updated.providerInstanceId, task!.providerInstanceId);
+ assert.equal(updated.result, "Previous session ended");
+ assert.isNotNull(updated.completedAt);
+ const items = after.turnItems.filter((item) => item.nodeId === task!.id);
+ assert.isAbove(items.length, 0);
+ assert.isTrue(
+ items.every((item) => item.status === "cancelled" && item.completedAt !== null),
+ );
+ assert.isNotNull(task!.childThreadId);
+ const child = yield* orchestrator.getThreadProjection(task!.childThreadId!);
+ const roots = child.nodes.filter((node) => node.nativeItemRef?.nativeId === "old-task");
+ assert.isAbove(roots.length, 0);
+ assert.isTrue(roots.every((node) => node.status === "cancelled"));
+ }).pipe(
+ Effect.provide(
+ ProviderReplayHarness.layerWithRegistry(
+ { name: "claude-stopped-task" },
+ ProviderAdapterRegistry.layerSingle(adapter),
+ { runEffectWorker: false },
+ ),
+ ),
+ );
+ }).pipe(
+ Effect.provide(
+ Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
+ ),
+ ),
+ ),
+);
```