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

// A subagent's own background shell keeps the subagent waiting on it.
yield* Queue.offer(
harness.sdkMessages,
claudeSdkFrame({
type: "system",
subtype: "background_tasks_changed",
tasks: [
{
task_id: "subagent-shell",
task_type: "local_bash",
description: "sleep 150",
parent_task_id: "agent-1",
},
],
uuid: "00000000-0000-4000-8000-000000000203",
session_id: WAKE_NATIVE_SESSION,
}),
);
yield* awaitUntil(
() =>
providerThreadRosterEvents(harness.events).some(
(event) =>
event.providerThread.pendingBackgroundTasks?.[0]?.parentTaskId === "agent-1",
),
"subagent shell roster",
);

yield* Queue.offer(harness.sdkMessages, emptyRoster);
yield* awaitUntil(
() =>
Expand Down
7 changes: 7 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2065,6 +2065,8 @@ function claudePendingBackgroundTask(input: {
// that started it tells it apart from a background Bash command.
readonly startedByMonitor: boolean;
readonly description: string | undefined;
// Undeclared SDK field: the subagent whose own background work this is.
readonly parentTaskId: unknown;
}): OrchestrationV2PendingBackgroundTask {
return {
taskId: input.taskId,
Expand All @@ -2076,6 +2078,9 @@ function claudePendingBackgroundTask(input: {
...(input.description !== undefined && input.description.trim().length > 0
? { description: input.description }
: {}),
...(typeof input.parentTaskId === "string" && input.parentTaskId.trim().length > 0
? { parentTaskId: input.parentTaskId }
: {}),
};
}

Expand Down Expand Up @@ -2136,6 +2141,7 @@ function parseClaudeBackgroundTaskEntry(
taskType,
startedByMonitor: monitorTasks.has(taskId),
description: typeof description === "string" ? description : undefined,
parentTaskId: Reflect.get(entry, "parent_task_id"),
});
}

Expand Down Expand Up @@ -6608,6 +6614,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
}),
description:
typeof message.description === "string" ? message.description : undefined,
parentTaskId: Reflect.get(message, "parent_task_id"),
}),
);
rosterChanged = true;
Expand Down
15 changes: 12 additions & 3 deletions apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,11 @@ import { CommandPolicyV2, resolveMessageDispatchIntent } from "./CommandPolicy.t
import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts";
import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts";
import { notificationTurnItem } from "@t3tools/provider-core/server/notification";
import { isRestartContinuationSource, runOutcome } from "./RestartBackgroundNote.ts";
import {
cancelledBeforeStart,
isRestartContinuationSource,
runOutcome,
} from "./RestartBackgroundNote.ts";
import {
isBufferedProviderNotification,
isUndeliveredMailboxSteer,
Expand Down Expand Up @@ -4638,9 +4642,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
projection.thread.archivedAt !== null ||
projection.thread.deletedAt !== null ||
projection.thread.providerInstanceId !== source.providerInstanceId ||
// Held queued runs never started; they wait behind the continuation.
// Held queued runs never started; they wait behind the continuation,
// and a run the restart ended before it started did not supersede it.
projection.runs.some(
(run) => run.id !== source.id && run.status !== "queued" && runRanAfter(run, source),
(run) =>
run.id !== source.id &&
run.status !== "queued" &&
!cancelledBeforeStart(run) &&
runRanAfter(run, source),
) ||
(yield* stopReachedRun(command, command.threadId, source.id))
) {
Expand Down
78 changes: 78 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionRecovery.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -731,6 +731,84 @@ it.effect("reads the run that owns a background roster, not a queued or resumed
}).pipe(Effect.provide(layerTest)),
);

it.effect("wakes the run whose subagent lost its shell, past a wake cut before it started", () =>
Effect.gen(function* () {
const projections = yield* ProjectionStore.ProjectionStoreV2;
const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;
const sql = yield* SqlClient.SqlClient;
const now = yield* DateTime.now;
const threadId = yield* createThread("subagent-shell");
const providerThreadId = ProviderThreadId.make(`provider-thread:${threadId}`);
const providerSessionId = ProviderSessionId.make(`session:${threadId}`);
yield* projections.apply({
id: EventId.make(`event:${threadId}:session`),
type: "provider-session.attached",
threadId,
driver,
providerInstanceId,
occurredAt: now,
payload: {
id: providerSessionId,
driver,
providerInstanceId,
status: "ready",
cwd: "/workspace",
model: modelSelection.model,
capabilities: CodexProviderCapabilitiesV2,
createdAt: now,
updatedAt: now,
lastError: null,
},
});
const settledRun = yield* createRun(threadId, "completed", { providerThreadId });
yield* createRun(threadId, "starting", {
ordinal: 2,
providerThreadId,
startedAt: null,
completedAt: null,
});
yield* projections.apply({
id: EventId.make(`event:${threadId}:roster`),
type: "provider-thread.updated",
threadId,
driver,
providerInstanceId,
occurredAt: now,
payload: {
id: providerThreadId,
appThreadId: threadId,
ownerNodeId: null,
driver,
providerInstanceId,
providerSessionId,
nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" },
nativeConversationHeadRef: null,
status: "idle",
firstRunOrdinal: 1,
lastRunOrdinal: 2,
handoffIds: [],
forkedFrom: null,
createdAt: now,
updatedAt: now,
pendingBackgroundTasks: [
{ taskId: "shell", description: "sleep 150", kind: "command", parentTaskId: "agent" },
],
},
});

yield* recovery.reconcile("startup");

const effects = yield* sql<{ readonly effect_id: string }>`
SELECT effect_id FROM orchestration_v2_effect_outbox WHERE thread_id = ${threadId}
`;
assert.deepEqual(effects, [{ effect_id: `effect:restart-continuation:${settledRun}` }]);
const projection = yield* projections.getThreadProjection(threadId);
assert.deepEqual(projection.runs[0]?.restartCancelledBackgroundWork, [
{ kind: "subagent", label: "sleep 150 (subagent agent)", id: "agent" },
]);
}).pipe(Effect.provide(layerRecoveryTest)),
);

it.effect("marks fork descendants unreadable when their source is missing or corrupt", () =>
Effect.gen(function* () {
const projections = yield* ProjectionStore.ProjectionStoreV2;
Expand Down
5 changes: 5 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4107,6 +4107,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
WHERE ended.thread_id = ${threadId}
AND ended.provider_thread_id = roster.provider_thread_id
AND ended.status NOT IN ('queued', 'rolled_back')
-- A run that never started (see neverStarted) owns no work.
AND NOT (
json_type(ended.payload_json, '$.startedAt') = 'null'
AND json_type(ended.payload_json, '$.restartContinuationOfRunId') IS NULL
)
ORDER BY ended.completed_at IS NULL DESC, ended.completed_at DESC,
ended.ordinal DESC
LIMIT 1
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import {
cancelledRosterTaskWork,
cancelledTurnItemWork,
mergeRestartCancelledBackgroundWork,
neverStarted,
} from "./RestartBackgroundNote.ts";

export class ProviderRuntimeRecoveryError extends Schema.TaggedError<ProviderRuntimeRecoveryError>()(
Expand Down Expand Up @@ -174,6 +175,7 @@ function latestStartedRun(
run.providerThreadId === providerThreadId &&
run.status !== "queued" &&
run.status !== "rolled_back" &&
!neverStarted(run) &&
(latest === undefined || runRanAfter(run, latest))
? run
: latest,
Expand Down Expand Up @@ -557,7 +559,13 @@ export const make = Effect.gen(function* () {
}
if (providerThread.ownerNodeId === null) {
for (const task of providerThread.pendingBackgroundTasks ?? []) {
if (cancelledBackgroundNativeIds.has(task.taskId)) continue;
// A subagent already recorded covers the shells it started.
if (
cancelledBackgroundNativeIds.has(task.taskId) ||
(task.parentTaskId !== undefined &&
cancelledBackgroundNativeIds.has(task.parentTaskId))
)
continue;
cancelledBackgroundNativeIds.add(task.taskId);
recordCancelledBackgroundWork(providerThread.id, cancelledRosterTaskWork(task));
}
Expand Down
42 changes: 42 additions & 0 deletions apps/server/src/orchestration-v2/RestartBackgroundNote.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import * as DateTime from "effect/DateTime";
import {
cancelledRosterTaskWork,
cancelledTurnItemWork,
isAwaitedWork,
isRestartNoteContinuation,
pendingRestartCancelledBackgroundWork,
restartCancelledBackgroundWorkNote,
Expand Down Expand Up @@ -117,6 +118,47 @@ it("bounds roster task labels including a long task id", () => {
assert.lengthOf(idOnly.label, 160);
});

it("names a lost subagent by the id the agent resumes it with", () => {
const ownShell = cancelledRosterTaskWork({
kind: "command",
taskId: "shell-1",
description: "sleep 150",
parentTaskId: "agent-1",
});
// The agent waits on the subagent, which waits on its shell.
assert.deepEqual(ownShell, {
kind: "subagent",
label: "sleep 150 (subagent agent-1)",
id: "agent-1",
});
assert.isTrue(isAwaitedWork(ownShell));
const long = cancelledRosterTaskWork({
kind: "command",
taskId: "shell-1",
description: "d".repeat(400),
parentTaskId: "agent-1",
}).label;
assert.lengthOf(long, 160);
assert.isTrue(long.endsWith("(subagent agent-1)"));
const longId = cancelledRosterTaskWork({
kind: "command",
taskId: "shell-1",
parentTaskId: "a".repeat(400),
}).label;
assert.lengthOf(longId, 160);
assert.isFalse(isAwaitedWork(cancelledRosterTaskWork({ kind: "command", taskId: "shell-2" })));
assert.equal(
cancelledTurnItemWork({
id: "item-1",
type: "subagent",
origin: "provider_native",
title: "Review",
nativeItemRef: { driver: "claudeAgent", nativeId: "agent-1", strength: "strong" },
} as never)?.label,
"Review (id agent-1)",
);
});

it("keeps separate cancelled tasks that share a kind and label", () => {
const first = { kind: "shell" as const, label: "sleep 20", id: "item-1" };
const second = { kind: "shell" as const, label: "sleep 20", id: "item-2" };
Expand Down
58 changes: 52 additions & 6 deletions apps/server/src/orchestration-v2/RestartBackgroundNote.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,22 +13,36 @@ type Attempt = Pick<OrchestrationV2RunAttempt, "id" | "runId">;

const MAX_LABEL_LENGTH = 160;

function compactLabel(value: unknown): string | undefined {
function compactLabel(value: unknown, maxLength = MAX_LABEL_LENGTH): string | undefined {
if (typeof value !== "string") return undefined;
const text = value.replaceAll(/\s+/g, " ").trim();
if (text.length === 0) return undefined;
return text.length > MAX_LABEL_LENGTH ? `${text.slice(0, MAX_LABEL_LENGTH - 1)}…` : text;
return text.length > maxLength ? `${text.slice(0, maxLength - 1)}…` : text;
}

/** A bounded label that keeps its id suffix whole whenever it fits. */
function labelWithId(name: unknown, fallback: string, suffix: string): string {
const budget = MAX_LABEL_LENGTH - suffix.length;
if (budget < 2) return compactLabel(`${fallback}${suffix}`) ?? fallback;
return `${compactLabel(name, budget) ?? compactLabel(fallback, budget)}${suffix}`;
}

/** Describes a background-capable turn item that restart recovery cancels. */
export function cancelledTurnItemWork(item: OrchestrationV2TurnItem): Work | undefined {
switch (item.type) {
case "subagent":
case "subagent": {
// The native id is what the agent resumes the subagent by.
const nativeId = item.origin === "provider_native" ? item.nativeItemRef?.nativeId : undefined;
return {
kind: "subagent",
label: compactLabel(item.title) ?? compactLabel(item.prompt) ?? "subagent",
label: labelWithId(
compactLabel(item.title) === undefined ? item.prompt : item.title,
"subagent",
nativeId === undefined ? "" : ` (id ${nativeId})`,
),
id: item.id,
};
}
case "command_execution":
return {
kind: "shell",
Expand Down Expand Up @@ -67,8 +81,19 @@ function restartWorkKind(task: OrchestrationV2PendingBackgroundTask): Work["kind
}
}

/** Describes a provider-reported background task (the provider-thread roster). */
/**
* Describes a provider-reported background task (the provider-thread roster).
* Work a subagent started and waits on is reported as that subagent, since the
* agent is waiting for the subagent's result and resumes it by that id.
*/
export function cancelledRosterTaskWork(task: OrchestrationV2PendingBackgroundTask): Work {
if (task.parentTaskId !== undefined) {
return {
kind: "subagent",
label: labelWithId(task.description, "background task", ` (subagent ${task.parentTaskId})`),
id: task.parentTaskId,
};
}
const kind = restartWorkKind(task);
const description = compactLabel(task.description);
return {
Expand All @@ -90,7 +115,7 @@ const MAX_NOTE_ENTRIES = 10;
export function restartCancelledBackgroundWorkNote(work: ReadonlyArray<Work>): string {
const omitted = work.length - MAX_NOTE_ENTRIES;
return [
"Note: the T3 server restarted, and this background work was cancelled before it finished. It will not report back:",
"Note: the T3 server restarted, and this background work was cancelled before it finished. It will not report back on its own, so resume or restart what you still need and wait for its result:",
...work.slice(0, MAX_NOTE_ENTRIES).map((entry) => `- ${entry.kind}: ${entry.label}`),
...(omitted > 0 ? [`- and ${omitted} more`] : []),
].join("\n");
Expand Down Expand Up @@ -148,6 +173,27 @@ export function runOutcome(
: run.status;
}

/**
* A run that never reached its provider, such as a provider wake the restart
* cut while it was starting: it lost no work and has no turn to continue. A
* prepared restart continuation still counts, since it carries its source.
*/
export function neverStarted(
run: Pick<OrchestrationV2Run, "startedAt" | "restartContinuationOfRunId">,
): boolean {
return run.startedAt === null && run.restartContinuationOfRunId === undefined;
}

/**
* A run the restart cancelled before it started, so it supersedes nothing. A
* run the user stopped before it started still does.
*/
export function cancelledBeforeStart(
run: Pick<OrchestrationV2Run, "status" | "startedAt" | "restartContinuationOfRunId">,
): boolean {
return run.status === "cancelled" && neverStarted(run);
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

/**
* Whether a restart continuation may prompt from `source`: a turn the restart
* cut mid-way, or a settled turn that lost awaited work in the restart.
Expand Down
Loading
Loading