Skip to content

fix(server): report delegated tasks stopped by a server restart #13345

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
saphid wants to merge 1 commit into
base: main
Choose a base branch
from
+602 −91
Open

fix(server): report delegated tasks stopped by a server restart #13345

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
313 changes: 313 additions & 0 deletions apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import * as EffectOutbox from "./EffectOutbox.ts";
import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts";
import * as NodeServices from "@effect/platform-node/NodeServices";
import { assert, it } from "@effect/vitest";
Expand All @@ -6,6 +7,7 @@ import {
EventId,
MessageId,
type ModelSelection,
type OrchestrationV2ThreadProjection,
NodeId,
ProjectId,
ProviderDriverKind,
Expand Down Expand Up @@ -34,7 +36,9 @@ import * as WorkspacePaths from "../workspace/WorkspacePaths.ts";
import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
import * as EventSink from "./EventSink.ts";
import * as Orchestrator from "./Orchestrator.ts";
import * as ProviderRuntimeRecoveryService from "./ProviderRuntimeRecoveryService.ts";
import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";
import { makeSubagentChildThread } from "./SubagentProjection.ts";
import {
OrchestrationV2EventSinkLayerLive,
OrchestrationV2LayerLive,
Expand Down Expand Up @@ -384,6 +388,315 @@ it.layer(TestLayer)("delegated completion delivery repairs", (it) => {
}),
);

it.effect(
"recovers a child that runtime recovery cancelled, deferring pending continuations",
() =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
const sink = yield* EventSink.EventSinkV2;
const now = yield* DateTime.now;
const threadId = ThreadId.make("thread:delegated-recovery");
const runId = RunId.make("run:delegated-recovery");
const taskId = NodeId.make("node:delegated-recovery-task");
const childThreadId = ThreadId.make("thread:delegated-recovery-child");
yield* seedParentWithTerminalTask({
threadId,
runId,
projectId: ProjectId.make("project:delegated-recovery"),
rootNodeId: NodeId.make("node:delegated-recovery-root"),
taskId,
deliveryState: "claimed",
completionWake: "always",
now,
});
const parent = yield* orchestrator.getThreadProjection(threadId);
const parentRun = parent.runs[0]!;
const childRunId = RunId.make("run:delegated-recovery-child");
yield* sink.write({
commandId: CommandId.make("command:delegated-recovery:seed"),
events: [
{
id: EventId.make("event:delegated-recovery:child"),
type: "thread.created",
threadId: childThreadId,
driver,
providerInstanceId: modelSelection.instanceId,
occurredAt: now,
payload: makeSubagentChildThread({
parentThread: parent.thread,
childThreadId,
parentNodeId: taskId,
activeProviderThreadId: null,
providerInstanceId: modelSelection.instanceId,
modelSelection,
title: "Recovered child",
now,
createdBy: "agent",
creationSource: "server",
}),
},
{
id: EventId.make("event:delegated-recovery:task"),
type: "subagent.updated",
threadId,
runId,
nodeId: taskId,
driver,
providerInstanceId: modelSelection.instanceId,
occurredAt: now,
payload: {
...parent.subagents[0]!,
childThreadId,
status: "running",
result: null,
completedAt: null,
completionDelivery: undefined,
},
},
],
});
yield* sink.write({
commandId: CommandId.make("command:delegated-recovery:running-child"),
events: [
{
id: EventId.make("event:delegated-recovery:child-run"),
type: "run.updated",
threadId: childThreadId,
runId: childRunId,
providerInstanceId: modelSelection.instanceId,
occurredAt: now,
payload: {
...parentRun,
id: childRunId,
threadId: childThreadId,
userMessageId: MessageId.make("message:delegated-recovery-child"),
rootNodeId: null,
activeAttemptId: null,
status: "running",
completedAt: null,
delegatedCompletion: undefined,
},
},
],
});
const resultTransfers = (projection: {
readonly contextTransfers: ReadonlyArray<{
readonly type: string;
readonly sourceThreadId: ThreadId;
}>;
}) =>
projection.contextTransfers.filter(
(row) => row.type === "subagent_result" && row.sourceThreadId === childThreadId,
);

// The report pass must run after real runtime recovery cancels the run.
const runtimeRecovery =
yield* ProviderRuntimeRecoveryService.ProviderRuntimeRecoveryService;
yield* orchestrator.recoverDelegatedTaskReports();
assert.lengthOf(resultTransfers(yield* orchestrator.getThreadProjection(threadId)), 0);
yield* runtimeRecovery.recover;
const child = yield* orchestrator.getThreadProjection(childThreadId);
assert.equal(child.runs.at(-1)?.status, "cancelled");
// The terminal reactor ignores runtime-reconcile events.
assert.lengthOf(resultTransfers(yield* orchestrator.getThreadProjection(threadId)), 0);
// An unreadable continuation must not publish a false cancellation.
yield* orchestrator.recoverDelegatedTaskReports(() =>
Effect.fail(
new EffectOutbox.EffectOutboxError({
operation: "get",
cause: new Error("outbox unavailable"),
}),
),
);
assert.lengthOf(resultTransfers(yield* orchestrator.getThreadProjection(threadId)), 0);
// A child whose restart continuation is still pending is left alone.
yield* orchestrator.recoverDelegatedTaskReports(() => Effect.succeed(true));
assert.lengthOf(resultTransfers(yield* orchestrator.getThreadProjection(threadId)), 0);

yield* orchestrator.recoverDelegatedTaskReports((run) =>
runtimeRecovery.isRestartContinuationPending(run.id),
);
const recovered = yield* orchestrator.getThreadProjection(threadId);
assert.lengthOf(resultTransfers(recovered), 1);
const task = recovered.subagents.find((row) => row.id === taskId);
assert.equal(task?.status, "cancelled");
assert.isDefined(task?.completionDelivery);

// Running the pass again publishes nothing new.
yield* orchestrator.recoverDelegatedTaskReports();
assert.lengthOf(resultTransfers(yield* orchestrator.getThreadProjection(threadId)), 1);
}),
);

it.effect("re-offers a parent wake that startup recovery cancelled with a sibling", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
const sink = yield* EventSink.EventSinkV2;
const now = yield* DateTime.now;
const threadId = ThreadId.make("thread:delegated-recovery-wake");
const runId = RunId.make("run:delegated-recovery-wake");
const taskId = NodeId.make("node:delegated-recovery-wake-first");
const siblingId = NodeId.make("node:delegated-recovery-wake-sibling");
const siblingThreadId = ThreadId.make("thread:delegated-recovery-wake-sibling");
const wakeMessageId = MessageId.make(`message:delegated-delivery:${threadId}`);
const wakeRunId = RunId.make("run:delegated-recovery-wake-delivery");
const siblingRunId = RunId.make("run:delegated-recovery-wake-sibling");
yield* seedParentWithTerminalTask({
threadId,
runId,
projectId: ProjectId.make("project:delegated-recovery-wake"),
rootNodeId: NodeId.make("node:delegated-recovery-wake-root"),
taskId,
deliveryState: "claimed",
completionWake: "always",
deliveryTaskIds: [taskId],
now,
});
const parent = yield* orchestrator.getThreadProjection(threadId);
const parentRun = parent.runs[0]!;
const runPayload = (
id: RunId,
thread: ThreadId,
messageId: MessageId,
status: "running" | "cancelled",
) => ({
...parentRun,
id,
threadId: thread,
ordinal: 2,
userMessageId: messageId,
rootNodeId: null,
activeAttemptId: null,
status,
completedAt: status === "cancelled" ? now : null,
delegatedCompletion: undefined,
});
// Task A's wake is running in the parent while sibling B still works.
yield* sink.write({
commandId: CommandId.make("command:delegated-recovery-wake:seed"),
events: [
{
id: EventId.make("event:delegated-recovery-wake:message"),
type: "message.updated",
threadId,
runId: wakeRunId,
occurredAt: now,
payload: {
id: wakeMessageId,
threadId,
runId: wakeRunId,
nodeId: null,
role: "user",
text: "Background task finished",
attachments: [],
streaming: false,
createdBy: "agent",
creationSource: "server",
createdAt: now,
updatedAt: now,
delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] },
},
},
{
id: EventId.make("event:delegated-recovery-wake:wake-run"),
type: "run.updated",
threadId,
runId: wakeRunId,
providerInstanceId: modelSelection.instanceId,
occurredAt: now,
payload: runPayload(wakeRunId, threadId, wakeMessageId, "running"),
},
{
id: EventId.make("event:delegated-recovery-wake:sibling-thread"),
type: "thread.created",
threadId: siblingThreadId,
driver,
providerInstanceId: modelSelection.instanceId,
occurredAt: now,
payload: makeSubagentChildThread({
parentThread: parent.thread,
childThreadId: siblingThreadId,
parentNodeId: siblingId,
activeProviderThreadId: null,
providerInstanceId: modelSelection.instanceId,
modelSelection,
title: "Sibling",
now,
createdBy: "agent",
creationSource: "server",
}),
},
{
id: EventId.make("event:delegated-recovery-wake:sibling-task"),
type: "subagent.updated",
threadId,
runId,
nodeId: siblingId,
driver,
providerInstanceId: modelSelection.instanceId,
occurredAt: now,
payload: {
...parent.subagents[0]!,
id: siblingId,
childThreadId: siblingThreadId,
status: "running",
result: null,
completedAt: null,
completionDelivery: undefined,
},
},
],
});
// Startup runtime recovery cancels both the wake and the sibling.
yield* sink.write({
commandId: CommandId.make("command:runtime-reconcile:delegated-recovery-wake"),
events: [
{
id: EventId.make("event:delegated-recovery-wake:wake-cancelled"),
type: "run.updated",
threadId,
runId: wakeRunId,
providerInstanceId: modelSelection.instanceId,
occurredAt: now,
payload: runPayload(wakeRunId, threadId, wakeMessageId, "cancelled"),
},
{
id: EventId.make("event:delegated-recovery-wake:sibling-cancelled"),
type: "run.updated",
threadId: siblingThreadId,
runId: siblingRunId,
providerInstanceId: modelSelection.instanceId,
occurredAt: now,
payload: runPayload(
siblingRunId,
siblingThreadId,
MessageId.make("message:delegated-recovery-wake-sibling"),
"cancelled",
),
},
],
});
const cohortOf = (projection: OrchestrationV2ThreadProjection) =>
projection.runs.find((run) => run.id === runId)?.delegatedCompletion;
// Until the startup pass runs, nothing re-offers the cancelled wake.
assert.equal(
cohortOf(yield* orchestrator.getThreadProjection(threadId))?.delivery?.generation,
1,
);

yield* orchestrator.recoverDelegatedTaskReports(() => Effect.succeed(false));
const recovered = yield* orchestrator.getThreadProjection(threadId);
assert.isTrue(
recovered.contextTransfers.some(
(row) => row.type === "subagent_result" && row.sourceThreadId === siblingThreadId,
),
);
const delivery = cohortOf(recovered)?.delivery;
assert.equal(delivery?.generation, 2);
assert.deepEqual([...(delivery?.taskIds ?? [])].toSorted(), [siblingId, taskId].toSorted());
}),
);

it.effect("builds completion text and metadata from the same live cohort", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
Expand Down
9 changes: 9 additions & 0 deletions apps/server/src/orchestration-v2/EffectWorker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,10 +109,18 @@ export const executorLayer: Layer.Layer<
const willRetry = options?.willRetry ?? false;
switch (effect.request.type) {
case "provider-runtime.continue":
// A delegated child that will not be continued stays cancelled, so
// its parent must get that result now rather than at next startup.
return continueRestartedRun({
threadId: effect.threadId,
sourceRunId: effect.request.sourceRunId,
}).pipe(
Effect.tapCause(() =>
willRetry ? Effect.void : threads.reconcileAppOwnedSubagentResult(effect.threadId),
),
Effect.tap((started) =>
started ? Effect.void : threads.reconcileAppOwnedSubagentResult(effect.threadId),
),
Effect.provideService(ThreadManagementService.ThreadManagementService, threads),
Effect.provideService(ServerSettings.ServerSettingsService, settings),
Effect.mapError(
Expand All @@ -123,6 +131,7 @@ export const executorLayer: Layer.Layer<
cause,
}),
),
Effect.asVoid,
);
case "provider-session.detach":
return providerSessions
Expand Down
Loading
Loading