authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-10-03 20:46:32-03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-03 20:46:32-03:00
log5108c978b1d9a8bd029e5ea96781d8cbc90b3877
treec6258fd10d79c90283ce0103e03a966d34e60ec8
parentf786ff3548aef904b7bdd47f263e8704ba20ca4b
signature Signed by PGP key B5690EEEBB952194

fix(server): restarts keep delegated tasks, queued threads, and stops intact (#15323)


29 files changed, 2494 insertions(+), 172 deletions(-)

apps/server/src/mcp/OrchestratorMcpService.test.ts+101-1
......@@ -11,10 +11,12 @@ import {
1111 type OrchestrationV2ThreadProjection,
1212 type ServerProvider,
1313} from "@t3tools/contracts";
14import * as DateTime from "effect/DateTime";
1415import * as Effect from "effect/Effect";
1516import * as Layer from "effect/Layer";
1617import * as Ref from "effect/Ref";
1718
19import { OrchestratorProjectionError } from "../orchestration-v2/Orchestrator.ts";
1820import type { ProviderAdapterV2Shape } from "../orchestration-v2/ProviderAdapter.ts";
1921import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
2022import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
......@@ -52,7 +54,22 @@ describe("OrchestratorMcpService", () => {
5254 } as unknown as OrchestrationV2ThreadProjection;
5355 const childProjection = {
5456 thread: { id: childThreadId },
55 runs: [{ id: childRunId, ordinal: 1, status: "completed" }],
57 runs: [
58 {
59 id: childRunId,
60 ordinal: 1,
61 status: "completed",
62 startedAt: DateTime.makeUnsafe("2026-10-03T10:00:00Z"),
63 completedAt: DateTime.makeUnsafe("2026-10-03T10:10:00Z"),
64 },
65 {
66 id: RunId.make("run:mcp-ack-continuation"),
67 ordinal: 2,
68 status: "failed",
69 startedAt: DateTime.makeUnsafe("2026-10-03T10:02:00Z"),
70 completedAt: DateTime.makeUnsafe("2026-10-03T10:05:00Z"),
71 },
72 ],
5673 contextTransfers: [],
5774 messages: [],
5875 subagents: [],
......@@ -126,6 +143,8 @@ describe("OrchestratorMcpService", () => {
126143 const result = yield* service.taskStatus(scope, taskId);
127144 assert.equal(result.status, "completed");
128145 assert.equal(result.summary, "terminal result");
146 assert.equal(result.latestTerminalRunId, childRunId);
147 assert.equal(result.latestTerminalStatus, "completed");
129148 const commandIds = yield* Ref.get(acknowledgementCommandIds);
130149 assert.equal(commandIds.length, 2);
131150 assert.notEqual(commandIds[0], commandIds[1]);
......@@ -133,6 +152,87 @@ describe("OrchestratorMcpService", () => {
133152 }),
134153 );
135154
155 it.effect("reports a restart-cut child as working until its continuation settles", () =>
156 Effect.gen(function* () {
157 const parentThreadId = ThreadId.make("thread:mcp-restart-parent");
158 const childThreadId = ThreadId.make("thread:mcp-restart-child");
159 const taskId = NodeId.make("node:mcp-restart-task");
160 const dispatched = yield* Ref.make(0);
161 let awaitsRestart = true;
162 let readFails = true;
163 const parentProjection = {
164 thread: { id: parentThreadId },
165 runs: [],
166 contextTransfers: [],
167 subagents: [
168 {
169 id: taskId,
170 threadId: parentThreadId,
171 origin: "app_owned",
172 childThreadId,
173 driver: "codex",
174 model: "gpt-5.6-terra",
175 status: "running",
176 result: null,
177 completionDelivery: { state: "pending" },
178 },
179 ],
180 } as unknown as OrchestrationV2ThreadProjection;
181 const childProjection = {
182 thread: { id: childThreadId },
183 runs: [{ id: RunId.make("run:mcp-restart-child"), ordinal: 1, status: "cancelled" }],
184 contextTransfers: [],
185 messages: [],
186 subagents: [],
187 providerThreads: [],
188 turnItems: [],
189 } as unknown as OrchestrationV2ThreadProjection;
190 const dependencies = Layer.mergeAll(
191 NodeServices.layer,
192 Layer.mock(ThreadManagementService.ThreadManagementService)({
193 getThreadRecords: (threadId) =>
194 Effect.succeed(threadId === parentThreadId ? parentProjection : childProjection),
195 delegatedTaskResultPending: () =>
196 readFails
197 ? Effect.fail(new OrchestratorProjectionError({ threadId: childThreadId }))
198 : Effect.succeed(awaitsRestart),
199 dispatch: () => Ref.update(dispatched, (count) => count + 1).pipe(Effect.as({} as never)),
200 }),
201 Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }),
202 Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({
203 list: () => Effect.succeed([]),
204 }),
205 Layer.mock(ScheduledTaskService.ScheduledTaskService)({}),
206 );
207 const scope: McpInvocationScope = {
208 environmentId: EnvironmentId.make("environment:mcp-restart"),
209 threadId: parentThreadId,
210 providerSessionId: "provider-session:mcp-restart",
211 providerInstanceId: ProviderInstanceId.make("codex"),
212 capabilities: new Set(["orchestration"]),
213 issuedAt: 1,
214 };
215
216 yield* Effect.gen(function* () {
217 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
218 const failed = yield* service.taskStatus(scope, taskId).pipe(Effect.flip);
219 assert.equal(failed.code, "orchestration_error");
220 assert.equal(yield* Ref.get(dispatched), 0);
221 readFails = false;
222 const held = yield* service.taskStatus(scope, taskId);
223 assert.equal(held.status, "running");
224 assert.equal(held.workState, "working");
225 assert.isNull(held.summary);
226 // Acknowledging the cut run would suppress the real result's wake.
227 assert.equal(yield* Ref.get(dispatched), 0);
228 awaitsRestart = false;
229 const settled = yield* service.taskStatus(scope, taskId);
230 assert.equal(settled.status, "cancelled");
231 assert.equal(yield* Ref.get(dispatched), 1);
232 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(dependencies))));
233 }),
234 );
235
136236 it.effect("does not dispose delivery when a nonterminal task has no active child run", () =>
137237 Effect.gen(function* () {
138238 const parentThreadId = ThreadId.make("thread:mcp-cancel-parent");
apps/server/src/mcp/OrchestratorMcpService.ts+15-2
......@@ -52,6 +52,7 @@ import {
5252 type ServerProvider,
5353 ThreadId,
5454} from "@t3tools/contracts";
55import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
5556import * as Context from "effect/Context";
5657import * as Crypto from "effect/Crypto";
5758import * as DateTime from "effect/DateTime";
......@@ -412,7 +413,10 @@ function latestTerminalResultRun(
412413 run.status !== "rolled_back" &&
413414 (run.id === delegatedRun?.id || run.startedAt !== null),
414415 )
415 .toSorted((left, right) => right.ordinal - left.ordinal)[0];
416 .reduce<OrchestrationV2Run | undefined>(
417 (latest, run) => (latest === undefined || runRanAfter(run, latest) ? run : latest),
418 undefined,
419 );
416420}
417421
418422function canExposeTaskRunResult(run: OrchestrationV2Run | undefined): run is OrchestrationV2Run {
......@@ -1055,7 +1059,16 @@ const make = Effect.gen(function* () {
10551059 messages: [...childControls.messages, ...resultRecords.messages],
10561060 turnItems: resultRecords.turnItems,
10571061 };
1058 const workState = task.result !== null ? "result_available" : progress.state;
1062 // A restart cut the child's run and its continuation has not settled, or
1063 // the child started working again after this read.
1064 const heldForRestart =
1065 task.result === null &&
1066 progress.state === "result_available" &&
1067 (yield* threadManagement
1068 .delegatedTaskResultPending(task.childThreadId)
1069 .pipe(Effect.mapError(threadManagementFailure)));
1070 const workState =
1071 task.result !== null ? "result_available" : heldForRestart ? "working" : progress.state;
10591072 const status =
10601073 task.result !== null
10611074 ? taskStatusForRun(
apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts+624
......@@ -7,6 +7,7 @@ import {
77 MessageId,
88 type ModelSelection,
99 NodeId,
10 type OrchestrationV2Run,
1011 ProjectId,
1112 ProviderDriverKind,
1213 ProviderInstanceId,
......@@ -35,6 +36,7 @@ import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
3536import * as EventSink from "./EventSink.ts";
3637import * as Orchestrator from "./Orchestrator.ts";
3738import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";
39import { continueRestartedRun } from "./RestartContinuation.ts";
3840import {
3941 OrchestrationV2EventSinkLayerLive,
4042 OrchestrationV2LayerLive,
......@@ -384,6 +386,98 @@ it.layer(TestLayer)("delegated completion delivery repairs", (it) => {
384386 }),
385387 );
386388
389 it.effect("acceptance batches a settled_only sibling once its spawning run ended", () =>
390 Effect.gen(function* () {
391 const orchestrator = yield* Orchestrator.OrchestratorV2;
392 const sink = yield* EventSink.EventSinkV2;
393 const now = yield* DateTime.now;
394 const threadId = ThreadId.make("mailbox-mixed");
395 const runId = RunId.make("mailbox-mixed-parent");
396 const taskId = NodeId.make("mailbox-mixed-always");
397 const siblingId = NodeId.make("mailbox-mixed-settled-only");
398 const messageId = MessageId.make(`message:delegated-delivery:${threadId}`);
399 yield* seedParentWithTerminalTask({
400 threadId,
401 runId,
402 projectId: ProjectId.make("mailbox-mixed-project"),
403 rootNodeId: NodeId.make("mailbox-mixed-root"),
404 taskId,
405 deliveryState: "claimed",
406 completionWake: "always",
407 deliveryTaskIds: [taskId],
408 now,
409 });
410 const projection = yield* orchestrator.getThreadProjection(threadId);
411 const task = projection.subagents[0]!;
412 const spawningRun = projection.runs.find((row) => row.id === runId)!;
413 // A restart cut the spawning run; its continuation is the live run now.
414 yield* sink.write({
415 events: [
416 {
417 ...runEvent({ threadId, runId, ordinal: 1, status: "cancelled", now }),
418 payload: { ...spawningRun, status: "cancelled" as const, completedAt: now },
419 },
420 runEvent({
421 threadId,
422 runId: RunId.make("mailbox-mixed-continuation"),
423 ordinal: 2,
424 status: "running",
425 now,
426 }),
427 {
428 id: EventId.make("mailbox-mixed-message"),
429 type: "message.updated",
430 threadId,
431 runId,
432 occurredAt: now,
433 payload: {
434 id: messageId,
435 threadId,
436 runId,
437 nodeId: task.parentNodeId,
438 role: "user",
439 text: "Background task finished",
440 attachments: [],
441 streaming: false,
442 createdBy: "agent",
443 creationSource: "server",
444 createdAt: now,
445 updatedAt: now,
446 delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] },
447 },
448 },
449 {
450 id: EventId.make(`event:${siblingId}`),
451 type: "subagent.updated" as const,
452 threadId,
453 runId,
454 nodeId: siblingId,
455 occurredAt: now,
456 payload: {
457 ...task,
458 id: siblingId,
459 completionWake: "settled_only" as const,
460 completionDelivery: { state: "pending" as const, observedByRunId: null },
461 },
462 },
463 ],
464 });
465 yield* orchestrator.dispatch({
466 type: "notification.delivery.accept",
467 commandId: CommandId.make("accept-mixed"),
468 threadId,
469 messageId,
470 });
471 const accepted = yield* orchestrator.getThreadProjection(threadId);
472 const cohort = accepted.runs.find((row) => row.id === runId)?.delegatedCompletion;
473 assert.deepEqual(cohort?.delivery?.taskIds, [siblingId]);
474 assert.equal(
475 accepted.subagents.find((row) => row.id === siblingId)?.completionDelivery?.state,
476 "claimed",
477 );
478 }),
479 );
480
387481 it.effect("builds completion text and metadata from the same live cohort", () =>
388482 Effect.gen(function* () {
389483 const orchestrator = yield* Orchestrator.OrchestratorV2;
......@@ -652,3 +746,533 @@ it.layer(TestLayer)("delegated completion delivery repairs", (it) => {
652746 }),
653747 );
654748});
749
750// Runtime reconciliation writes restart cancellations under this command
751// prefix, which the live terminal-run listener skips.
752const reconcileCommandId = (name: string) => CommandId.make(`command:runtime-reconcile:${name}`);
753
754const parentProviderThreadId = (threadId: ThreadId) =>
755 ProviderThreadId.make(`provider-thread:${String(threadId).replace("thread:", "")}`);
756
757const runEvent = (input: {
758 readonly threadId: ThreadId;
759 readonly runId: RunId;
760 readonly ordinal: number;
761 readonly status: OrchestrationV2Run["status"];
762 readonly now: DateTime.Utc;
763 readonly providerThreadId?: ProviderThreadId;
764 readonly userMessageId?: MessageId;
765 readonly delegatedCompletion?: OrchestrationV2Run["delegatedCompletion"];
766}) => ({
767 id: EventId.make(`event:${input.runId}:${input.status}`),
768 type: "run.updated" as const,
769 threadId: input.threadId,
770 runId: input.runId,
771 providerInstanceId: modelSelection.instanceId,
772 occurredAt: input.now,
773 payload: {
774 id: input.runId,
775 threadId: input.threadId,
776 ordinal: input.ordinal,
777 providerInstanceId: modelSelection.instanceId,
778 modelSelection,
779 providerThreadId: input.providerThreadId ?? null,
780 userMessageId: input.userMessageId ?? MessageId.make(`message:${input.runId}`),
781 rootNodeId: null,
782 activeAttemptId: null,
783 status: input.status,
784 requestedAt: input.now,
785 startedAt: input.now,
786 completedAt: input.status === "running" ? null : input.now,
787 checkpointId: null,
788 contextHandoffId: null,
789 ...(input.delegatedCompletion === undefined
790 ? {}
791 : { delegatedCompletion: input.delegatedCompletion }),
792 },
793});
794
795/** A running app-owned task whose child thread's first run the restart cancelled. */
796const seedRestartCancelledChild = (input: {
797 readonly parentThreadId: ThreadId;
798 readonly projectId: ProjectId;
799 readonly parentRunId: RunId;
800 readonly rootNodeId: NodeId;
801 readonly name: string;
802 readonly completionWake: "always" | "settled_only";
803 readonly continuationPending: boolean;
804 /** "completed" seeds a settled turn whose background work the restart cancelled. */
805 readonly runStatus?: "cancelled" | "completed";
806 readonly now: DateTime.Utc;
807}) =>
808 Effect.gen(function* () {
809 const eventSink = yield* EventSink.EventSinkV2;
810 const taskId = NodeId.make(`node:${input.name}`);
811 const childThreadId = ThreadId.make(`thread:${input.name}`);
812 const childRunId = RunId.make(`run:${input.name}:1`);
813 yield* eventSink.write({
814 commandId: CommandId.make(`command:seed-child:${input.name}`),
815 events: [
816 {
817 id: EventId.make(`event:${input.name}:thread`),
818 type: "thread.created",
819 threadId: childThreadId,
820 occurredAt: input.now,
821 payload: {
822 createdBy: "agent",
823 creationSource: "server",
824 id: childThreadId,
825 projectId: input.projectId,
826 title: input.name,
827 providerInstanceId: modelSelection.instanceId,
828 modelSelection,
829 runtimeMode: "full-access",
830 interactionMode: "default",
831 branch: null,
832 worktreePath: null,
833 activeProviderThreadId: null,
834 lineage: {
835 parentThreadId: input.parentThreadId,
836 relationshipToParent: "subagent",
837 rootThreadId: input.parentThreadId,
838 },
839 forkedFrom: { type: "node", nodeId: taskId },
840 createdAt: input.now,
841 updatedAt: input.now,
842 archivedAt: null,
843 settledOverride: null,
844 settledAt: null,
845 lastVisitedAt: null,
846 deletedAt: null,
847 },
848 },
849 {
850 id: EventId.make(`event:${input.name}:task`),
851 type: "subagent.updated",
852 threadId: input.parentThreadId,
853 runId: input.parentRunId,
854 nodeId: taskId,
855 driver,
856 providerInstanceId: modelSelection.instanceId,
857 occurredAt: input.now,
858 payload: {
859 id: taskId,
860 threadId: input.parentThreadId,
861 runId: input.parentRunId,
862 parentNodeId: input.rootNodeId,
863 origin: "app_owned",
864 createdBy: "agent",
865 driver,
866 providerInstanceId: modelSelection.instanceId,
867 providerThreadId: null,
868 childThreadId,
869 nativeTaskRef: null,
870 prompt: `Run ${input.name}.`,
871 title: null,
872 model: null,
873 completionWake: input.completionWake,
874 status: "running",
875 result: null,
876 startedAt: input.now,
877 completedAt: null,
878 updatedAt: input.now,
879 },
880 },
881 ],
882 });
883 yield* eventSink.writeWithEffects({
884 commandId: reconcileCommandId(input.name),
885 events: [
886 runEvent({
887 threadId: childThreadId,
888 runId: childRunId,
889 ordinal: 1,
890 status: input.runStatus ?? "cancelled",
891 now: input.now,
892 }),
893 ],
894 effects: input.continuationPending
895 ? [
896 {
897 id: `effect:restart-continuation:${childRunId}`,
898 commandId: reconcileCommandId(input.name),
899 threadId: childThreadId,
900 request: { type: "provider-runtime.continue", sourceRunId: childRunId },
901 },
902 ]
903 : [],
904 });
905 return { taskId, childThreadId, childRunId };
906 });
907
908it.layer(TestLayer)("delegated tasks across a server restart", (it) => {
909 it.effect("holds a restart-cancelled child for its continuation's result", () =>
910 Effect.gen(function* () {
911 const orchestrator = yield* Orchestrator.OrchestratorV2;
912 const eventSink = yield* EventSink.EventSinkV2;
913 const now = yield* DateTime.now;
914 const threadId = ThreadId.make("thread:restart-parent");
915 const projectId = ProjectId.make("project:restart-parent");
916 const runId = RunId.make("run:restart-parent");
917 const rootNodeId = NodeId.make("node:restart-parent-root");
918 yield* seedParentWithTerminalTask({
919 threadId,
920 projectId,
921 runId,
922 rootNodeId,
923 taskId: NodeId.make("node:restart-parent-settled"),
924 deliveryState: "delivered",
925 now,
926 });
927 const child = (
928 name: string,
929 continuationPending: boolean,
930 runStatus?: "cancelled" | "completed",
931 ) =>
932 seedRestartCancelledChild({
933 parentThreadId: threadId,
934 projectId,
935 parentRunId: runId,
936 rootNodeId,
937 name,
938 completionWake: "always",
939 continuationPending,
940 ...(runStatus === undefined ? {} : { runStatus }),
941 now,
942 });
943 const resumed = yield* child("restart-resumed-child", true);
944 const stopped = yield* child("restart-stopped-child", false);
945 // Settled with only background work left: its interim reply is not the result.
946 const backgrounded = yield* child("restart-backgrounded-child", true, "completed");
947 // A second restart cut the first continuation before it started.
948 const recut = yield* child("restart-recut-child", false);
949 const recutContinuationId = RunId.make("run:restart-recut-child:2");
950 const recutCommandId = CommandId.make("command:restart-recut-child:reconcile");
951 const recutRun = runEvent({
952 threadId: recut.childThreadId,
953 runId: recutContinuationId,
954 ordinal: 2,
955 status: "cancelled",
956 now,
957 });
958 yield* eventSink.writeWithEffects({
959 commandId: recutCommandId,
960 events: [
961 {
962 ...recutRun,
963 payload: {
964 ...recutRun.payload,
965 startedAt: null,
966 restartContinuationOfRunId: recut.childRunId,
967 },
968 },
969 ],
970 effects: [
971 {
972 id: `effect:restart-continuation:${recutContinuationId}`,
973 commandId: recutCommandId,
974 threadId: recut.childThreadId,
975 request: { type: "provider-runtime.continue", sourceRunId: recutContinuationId },
976 },
977 ],
978 });
979
980 yield* orchestrator.recoverDelegatedTasks;
981
982 const recovered = yield* orchestrator.getThreadProjection(threadId);
983 const task = (id: NodeId) => recovered.subagents.find((row) => row.id === id);
984 assert.equal(task(stopped.taskId)?.status, "cancelled");
985 assert.equal(task(stopped.taskId)?.completionDelivery?.state, "claimed");
986 assert.equal(task(resumed.taskId)?.status, "running");
987 assert.isNull(task(resumed.taskId)?.result ?? null);
988 assert.equal(task(backgrounded.taskId)?.status, "running");
989 assert.isNull(task(backgrounded.taskId)?.result ?? null);
990 assert.equal(task(recut.taskId)?.status, "running");
991 assert.isTrue(yield* orchestrator.delegatedTaskResultPending(recut.childThreadId));
992 assert.isTrue(yield* orchestrator.delegatedTaskResultPending(resumed.childThreadId));
993 assert.isFalse(yield* orchestrator.delegatedTaskResultPending(stopped.childThreadId));
994 // A replayed first continuation settling must not release the second one's hold.
995 yield* orchestrator.recoverDelegatedTask(recut.childThreadId, recut.childRunId);
996 const replayed = yield* orchestrator.getThreadProjection(threadId);
997 assert.equal(replayed.subagents.find((row) => row.id === recut.taskId)?.status, "running");
998 // A caller that read the cancelled run before the child resumed sees it as pending.
999 yield* eventSink.write({
1000 commandId: CommandId.make("command:restart-stopped-child:resumed"),
1001 events: [
1002 runEvent({
1003 threadId: stopped.childThreadId,
1004 runId: RunId.make("run:restart-stopped-child:2"),
1005 ordinal: 2,
1006 status: "running",
1007 now,
1008 }),
1009 ],
1010 });
1011 assert.isTrue(yield* orchestrator.delegatedTaskResultPending(stopped.childThreadId));
1012 assert.isFalse(
1013 recovered.contextTransfers.some(
1014 (transfer) => transfer.sourceThreadId === resumed.childThreadId,
1015 ),
1016 );
1017
1018 // The continuation's own run finishing settles the task with its result.
1019 const afterSequence = yield* eventSink.latestSequence();
1020 const continuationRunId = RunId.make("run:restart-resumed-child:2");
1021 yield* eventSink.write({
1022 commandId: CommandId.make("command:restart-resumed-child:completed"),
1023 events: [
1024 {
1025 id: EventId.make("event:restart-resumed-child:result"),
1026 type: "message.updated",
1027 threadId: resumed.childThreadId,
1028 runId: continuationRunId,
1029 occurredAt: now,
1030 payload: {
1031 id: MessageId.make("message:restart-resumed-child:result"),
1032 threadId: resumed.childThreadId,
1033 runId: continuationRunId,
1034 nodeId: null,
1035 role: "assistant",
1036 text: "Finished after the restart.",
1037 attachments: [],
1038 streaming: false,
1039 createdBy: "agent",
1040 creationSource: "server",
1041 createdAt: now,
1042 updatedAt: now,
1043 },
1044 },
1045 runEvent({
1046 threadId: resumed.childThreadId,
1047 runId: continuationRunId,
1048 ordinal: 2,
1049 status: "completed",
1050 now,
1051 }),
1052 ],
1053 });
1054 const settled = yield* eventSink
1055 .stream({ afterSequence, eventType: "subagent.updated" })
1056 .pipe(
1057 Stream.filter(
1058 (stored) =>
1059 stored.event.type === "subagent.updated" &&
1060 stored.event.payload.id === resumed.taskId,
1061 ),
1062 Stream.take(1),
1063 Stream.runHead,
1064 );
1065 assert.isTrue(settled._tag === "Some");
1066 const finished = yield* orchestrator.getThreadProjection(threadId);
1067 const finishedTask = finished.subagents.find((row) => row.id === resumed.taskId);
1068 assert.equal(finishedTask?.status, "completed");
1069 assert.equal(finishedTask?.result, "Finished after the restart.");
1070 }),
1071 );
1072
1073 it.effect("settles a restart-cancelled child whose continuation declines to start", () =>
1074 Effect.gen(function* () {
1075 const orchestrator = yield* Orchestrator.OrchestratorV2;
1076 const now = yield* DateTime.now;
1077 const threadId = ThreadId.make("thread:restart-declined-parent");
1078 const projectId = ProjectId.make("project:restart-declined-parent");
1079 const runId = RunId.make("run:restart-declined-parent");
1080 const rootNodeId = NodeId.make("node:restart-declined-parent-root");
1081 yield* seedParentWithTerminalTask({
1082 threadId,
1083 projectId,
1084 runId,
1085 rootNodeId,
1086 taskId: NodeId.make("node:restart-declined-parent-settled"),
1087 deliveryState: "delivered",
1088 now,
1089 });
1090 const child = yield* seedRestartCancelledChild({
1091 parentThreadId: threadId,
1092 projectId,
1093 parentRunId: runId,
1094 rootNodeId,
1095 name: "restart-declined-child",
1096 completionWake: "always",
1097 continuationPending: true,
1098 now,
1099 });
1100 yield* orchestrator.recoverDelegatedTasks;
1101 const held = yield* orchestrator.getThreadProjection(threadId);
1102 assert.equal(held.subagents.find((row) => row.id === child.taskId)?.status, "running");
1103
1104 yield* continueRestartedRun({
1105 threadId: child.childThreadId,
1106 sourceRunId: child.childRunId,
1107 }).pipe(
1108 Effect.provide(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: false })),
1109 );
1110
1111 const settled = yield* orchestrator.getThreadProjection(threadId);
1112 const task = settled.subagents.find((row) => row.id === child.taskId);
1113 assert.equal(task?.status, "cancelled");
1114 assert.equal(task?.completionDelivery?.state, "claimed");
1115 }),
1116 );
1117
1118 // A running provider turn does not prove the provider consumed the delivery
1119 // (Claude marks the turn running before it reads the prompt), so a cut
1120 // delivery is offered again even when a continuation resumes that turn.
1121 it.effect("re-offers a cut delivery even when a continuation resumes its turn", () =>
1122 Effect.gen(function* () {
1123 const orchestrator = yield* Orchestrator.OrchestratorV2;
1124 const eventSink = yield* EventSink.EventSinkV2;
1125 const now = yield* DateTime.now;
1126 const seed = (name: string, continuationPending: boolean) =>
1127 Effect.gen(function* () {
1128 const threadId = ThreadId.make(`thread:${name}`);
1129 const runId = RunId.make(`run:${name}`);
1130 const taskId = NodeId.make(`node:${name}-task`);
1131 const messageId = MessageId.make(`message:delegated-delivery:${threadId}`);
1132 const deliveryRunId = RunId.make(`run:${name}:delivery`);
1133 yield* seedParentWithTerminalTask({
1134 threadId,
1135 projectId: ProjectId.make(`project:${name}`),
1136 runId,
1137 rootNodeId: NodeId.make(`node:${name}-root`),
1138 taskId,
1139 deliveryState: "claimed",
1140 completionWake: "always",
1141 deliveryTaskIds: [taskId],
1142 now,
1143 });
1144 yield* eventSink.writeWithEffects({
1145 commandId: reconcileCommandId(name),
1146 events: [
1147 {
1148 id: EventId.make(`event:${name}:delivery-message`),
1149 type: "message.updated",
1150 threadId,
1151 runId: deliveryRunId,
1152 occurredAt: now,
1153 payload: {
1154 id: messageId,
1155 threadId,
1156 runId: deliveryRunId,
1157 nodeId: null,
1158 role: "user",
1159 text: `Delegated task ${taskId} reached a terminal state.`,
1160 attachments: [],
1161 streaming: false,
1162 createdBy: "agent",
1163 creationSource: "server",
1164 createdAt: now,
1165 updatedAt: now,
1166 delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] },
1167 },
1168 },
1169 runEvent({
1170 threadId,
1171 runId: deliveryRunId,
1172 ordinal: 2,
1173 status: "cancelled",
1174 now,
1175 providerThreadId: parentProviderThreadId(threadId),
1176 userMessageId: messageId,
1177 }),
1178 ],
1179 effects: continuationPending
1180 ? [
1181 {
1182 id: `effect:restart-continuation:${deliveryRunId}`,
1183 commandId: reconcileCommandId(name),
1184 threadId,
1185 request: { type: "provider-runtime.continue", sourceRunId: deliveryRunId },
1186 },
1187 ]
1188 : [],
1189 });
1190 return { threadId, runId, taskId };
1191 });
1192 const cut = yield* seed("delivery-cut", false);
1193 const resumed = yield* seed("delivery-resumed", true);
1194
1195 yield* orchestrator.recoverDelegatedTasks;
1196
1197 for (const seeded of [cut, resumed]) {
1198 const projection = yield* orchestrator.getThreadProjection(seeded.threadId);
1199 assert.deepEqual(
1200 projection.subagents.find((row) => row.id === seeded.taskId)?.completionDelivery?.state,
1201 "claimed",
1202 );
1203 const delivery = projection.runs.find((row) => row.id === seeded.runId)?.delegatedCompletion
1204 ?.delivery;
1205 assert.equal(delivery?.generation, 2);
1206 assert.deepEqual(delivery?.taskIds, [seeded.taskId]);
1207 }
1208 }),
1209 );
1210
1211 it.effect("wakes the parent for a settled_only task once its spawning turn ended", () =>
1212 Effect.gen(function* () {
1213 const orchestrator = yield* Orchestrator.OrchestratorV2;
1214 const eventSink = yield* EventSink.EventSinkV2;
1215 const now = yield* DateTime.now;
1216 const threadId = ThreadId.make("thread:settled-only-restart");
1217 const projectId = ProjectId.make("project:settled-only-restart");
1218 const runId = RunId.make("run:settled-only-restart");
1219 const rootNodeId = NodeId.make("node:settled-only-restart-root");
1220 yield* seedParentWithTerminalTask({
1221 threadId,
1222 projectId,
1223 runId,
1224 rootNodeId,
1225 taskId: NodeId.make("node:settled-only-restart-settled"),
1226 deliveryState: "delivered",
1227 now,
1228 });
1229 // The restart cut the turn blocked in delegate_task(wait); its
1230 // continuation is a new, live run.
1231 yield* eventSink.write({
1232 commandId: reconcileCommandId("settled-only-restart"),
1233 events: [
1234 runEvent({
1235 threadId,
1236 runId,
1237 ordinal: 1,
1238 status: "cancelled",
1239 now,
1240 providerThreadId: parentProviderThreadId(threadId),
1241 userMessageId: MessageId.make(`message:seed-user:${threadId}`),
1242 delegatedCompletion: { disposition: "open", nextGeneration: 2, delivery: null },
1243 }),
1244 runEvent({
1245 threadId,
1246 runId: RunId.make("run:settled-only-restart:continuation"),
1247 ordinal: 2,
1248 status: "running",
1249 now,
1250 providerThreadId: parentProviderThreadId(threadId),
1251 }),
1252 ],
1253 });
1254 const child = yield* seedRestartCancelledChild({
1255 parentThreadId: threadId,
1256 projectId,
1257 parentRunId: runId,
1258 rootNodeId,
1259 name: "settled-only-restart-child",
1260 completionWake: "settled_only",
1261 continuationPending: false,
1262 now,
1263 });
1264
1265 yield* orchestrator.recoverDelegatedTasks;
1266
1267 const projection = yield* orchestrator.getThreadProjection(threadId);
1268 assert.equal(
1269 projection.subagents.find((row) => row.id === child.taskId)?.completionDelivery?.state,
1270 "claimed",
1271 );
1272 assert.deepEqual(
1273 projection.runs.find((row) => row.id === runId)?.delegatedCompletion?.delivery?.taskIds,
1274 [child.taskId],
1275 );
1276 }),
1277 );
1278});
apps/server/src/orchestration-v2/EffectWorker.test.ts+49-2
......@@ -79,6 +79,8 @@ function restartEffect(
7979function makeExecutorLayer(input: {
8080 readonly events: Ref.Ref<ReadonlyArray<string>>;
8181 readonly failFirstStart?: Ref.Ref<boolean>;
82 readonly threads?: Partial<ThreadManagementService.ThreadManagementService["Service"]>;
83 readonly continueAfterRestart?: boolean;
8284}) {
8385 const record = (event: string) => Ref.update(input.events, (events) => [...events, event]);
8486 const dependencies = Layer.mergeAll(
......@@ -148,8 +150,10 @@ function makeExecutorLayer(input: {
148150 Layer.provide(
149151 Layer.mergeAll(
150152 dependencies,
151 Layer.mock(ThreadManagementService.ThreadManagementService)({}),
152 ServerSettings.layerTest(),
153 Layer.mock(ThreadManagementService.ThreadManagementService)(input.threads ?? {}),
154 ServerSettings.layerTest(
155 input.continueAfterRestart === true ? { continueThreadsAfterServerUpdate: true } : {},
156 ),
153157 ),
154158 ),
155159 );
......@@ -753,3 +757,46 @@ it.effect("safely retries after replacement cleanup succeeds and start fails", (
753757 ]);
754758 }),
755759);
760
761it.effect("settles a delegated child once its restart continuation fails for good", () =>
762 Effect.gen(function* () {
763 const timestamp = DateTime.formatIso(yield* DateTime.now);
764 const events = yield* Ref.make<ReadonlyArray<string>>([]);
765 const recovered = yield* Ref.make<ReadonlyArray<ThreadId>>([]);
766 const layer = makeExecutorLayer({
767 events,
768 continueAfterRestart: true,
769 threads: {
770 getThreadRecords: () => Effect.fail(new Error("provider instance removed") as never),
771 recoverDelegatedTask: (childThreadId) =>
772 Ref.update(recovered, (ids) => [...ids, childThreadId]),
773 },
774 });
775 const effect: EffectOutbox.OrchestrationEffectV2 = {
776 id: `effect:restart-continuation:${runId}`,
777 commandId: CommandId.make("command:restart-continuation-failure"),
778 threadId,
779 request: { type: "provider-runtime.continue", sourceRunId: runId },
780 status: "running",
781 attemptCount: 1,
782 availableAt: timestamp,
783 leaseOwner: "test-worker",
784 leaseExpiresAt: timestamp,
785 createdAt: timestamp,
786 updatedAt: timestamp,
787 completedAt: null,
788 lastError: null,
789 };
790 yield* Effect.gen(function* () {
791 const executor = yield* EffectWorker.OrchestrationEffectExecutorV2;
792 assert.isTrue(
793 Exit.isFailure(yield* Effect.exit(executor.execute(effect, { willRetry: true }))),
794 );
795 assert.deepEqual(yield* Ref.get(recovered), []);
796 assert.isTrue(
797 Exit.isFailure(yield* Effect.exit(executor.execute(effect, { willRetry: false }))),
798 );
799 assert.deepEqual(yield* Ref.get(recovered), [threadId]);
800 }).pipe(Effect.provide(layer));
801 }),
802);
apps/server/src/orchestration-v2/EffectWorker.ts+10-5
......@@ -108,13 +108,17 @@ export const executorLayer: Layer.Layer<
108108 execute: (effect, options) => {
109109 const willRetry = options?.willRetry ?? false;
110110 switch (effect.request.type) {
111 case "provider-runtime.continue":
112 return continueRestartedRun({
113 threadId: effect.threadId,
114 sourceRunId: effect.request.sourceRunId,
115 }).pipe(
111 case "provider-runtime.continue": {
112 const sourceRunId = effect.request.sourceRunId;
113 return continueRestartedRun({ threadId: effect.threadId, sourceRunId }).pipe(
116114 Effect.provideService(ThreadManagementService.ThreadManagementService, threads),
117115 Effect.provideService(ServerSettings.ServerSettingsService, settings),
116 // A continuation that will never run still owes a delegated parent a result.
117 Effect.tapError(() =>
118 willRetry
119 ? Effect.void
120 : threads.recoverDelegatedTask(effect.threadId, sourceRunId),
121 ),
118122 Effect.mapError(
119123 (cause) =>
120124 new OrchestrationEffectExecutionError({
......@@ -124,6 +128,7 @@ export const executorLayer: Layer.Layer<
124128 }),
125129 ),
126130 );
131 }
127132 case "provider-session.detach":
128133 return providerSessions
129134 .detach({
apps/server/src/orchestration-v2/Orchestrator.ts+209-94
......@@ -1,6 +1,7 @@
11import {
22 latestExecutedRun,
33 latestRootProviderFailure,
4 runRanAfter,
45 usageLimitBlockedRun,
56} from "@t3tools/shared/orchestrationV2ThreadError";
67import { threadPullRequestsOf } from "@t3tools/shared/threadPullRequests";
......@@ -77,6 +78,7 @@ import { notificationTurnItem } from "./Notification.ts";
7778import { isRestartNoteSource } from "./RestartBackgroundNote.ts";
7879import { isUndeliveredMailboxSteer } from "./NotificationMailbox.ts";
7980import { EventSinkV2 } from "./EventSink.ts";
81import * as EffectOutbox from "./EffectOutbox.ts";
8082import type { OrchestrationEffectRequestV2, PendingOrchestrationEffectV2 } from "./EffectOutbox.ts";
8183import { IdAllocatorV2 } from "./IdAllocator.ts";
8284import {
......@@ -242,6 +244,17 @@ export interface OrchestratorV2DispatchResult {
242244
243245export interface OrchestratorV2Shape {
244246 readonly resumeQueuedRuns: Effect.Effect<number, OrchestratorV2Error>;
247 /** Startup pass that settles delegated-task results and deliveries runs left behind. */
248 readonly recoverDelegatedTasks: Effect.Effect<void>;
249 /** Settles a delegated child whose restart continuation of `sourceRunId` declined or failed. */
250 readonly recoverDelegatedTask: (threadId: ThreadId, sourceRunId: RunId) => Effect.Effect<void>;
251 /**
252 * Whether a delegated child's apparent result is not final yet: a restart
253 * continuation is pending, or the child is working again.
254 */
255 readonly delegatedTaskResultPending: (
256 childThreadId: ThreadId,
257 ) => Effect.Effect<boolean, OrchestratorProjectionError>;
245258 readonly dispatch: (
246259 command: OrchestrationV2ServerCommand,
247260 ) => Effect.Effect<OrchestratorV2DispatchResult, OrchestratorV2Error>;
......@@ -338,7 +351,8 @@ function wakeWorkStartedAt(
338351 return previous === undefined ? {} : { workStartedAt: orchestrationV2RunWorkStartedAt(previous) };
339352}
340353
341function isNativeMaintenanceCommand(message: {
354/** A native /compact or /logout turn: provider maintenance, not agent work. */
355export function isNativeMaintenanceCommand(message: {
342356 readonly text: string;
343357 readonly attachments: ReadonlyArray<ChatAttachment>;
344358 readonly context?: import("@t3tools/contracts").OrchestrationMessageContext | undefined;
......@@ -718,6 +732,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
718732 const idAllocator = yield* IdAllocatorV2;
719733 const projects = yield* ProjectStore.ProjectStoreV2;
720734 const projectionStore = yield* ProjectionStoreV2;
735 const effectOutbox = yield* EffectOutbox.EffectOutboxV2;
721736 const nextTurnItemOrdinal = (
722737 projection: Pick<OrchestrationV2ThreadProjection, "thread"> &
723738 Partial<Pick<OrchestrationV2ThreadProjection, "turnItems">>,
......@@ -4357,7 +4372,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
43574372 projection.thread.archivedAt !== null ||
43584373 projection.thread.deletedAt !== null ||
43594374 projection.thread.providerInstanceId !== source.providerInstanceId ||
4360 projection.runs.some((run) => run.ordinal > source.ordinal)
4375 // Held queued runs never started; they wait behind the continuation.
4376 projection.runs.some(
4377 (run) => run.id !== source.id && run.status !== "queued" && runRanAfter(run, source),
4378 )
43614379 ) {
43624380 // Preserve the current row so stale automatic deliveries receive an
43634381 // accepted receipt without changing work or repeatedly retrying.
......@@ -6594,12 +6612,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
65946612 // policy when the child terminalizes. Both writers hold this parent lock,
65956613 // so a terminal task means finalize already committed the terminal row and
65966614 // already made its offer decision under the pre-upgrade policy: under
6597 // settled_only it offered iff the parent had no live run. Plan a delivery
6598 // only when the parent has a live run now, which is precisely the case
6615 // settled_only it offered iff the spawning run was not live. Plan a
6616 // delivery only when that run is live now, which is precisely the case
65996617 // where finalize skipped. The mailbox steers a capable active session or
6600 // queues behind that run. When the parent is not live, finalize already
6601 // offered and a second
6602 // offer would wake the parent twice. (If the parent settled in between,
6618 // queues behind that run. When that run is not live, finalize already
6619 // offered and a second offer would wake the parent twice. (If the parent settled in between,
66036620 // this skips a wake that finalize also skipped; a missed wake is cheaper
66046621 // than a duplicate one, and the result is already in the projection.)
66056622 const parentRun =
......@@ -6609,7 +6626,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
66096626 const completionPlan =
66106627 command.completionWake === "always" &&
66116628 isTerminalDelegatedTaskStatus(task.status) &&
6612 hasLiveRun(parentProjection)
6629 parentRun !== undefined &&
6630 hasLiveRun({ runs: [parentRun] })
66136631 ? yield* planDelegatedCompletionDelivery({
66146632 parentProjection,
66156633 parentRun,
......@@ -8494,6 +8512,39 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
84948512 : undefined;
84958513 });
84968514
8515 /**
8516 * A run with a pending restart continuation has no outcome yet: restart
8517 * reconciliation cancelled it mid-turn, or cancelled the background work a
8518 * settled turn was waiting on. The continuation's own run settles it, or
8519 * RestartContinuation recovers the thread when it declines to continue.
8520 */
8521 const awaitsRestartContinuation = (run: OrchestrationV2Run) =>
8522 effectOutbox
8523 .get(`effect:restart-continuation:${run.id}`)
8524 .pipe(
8525 Effect.map(
8526 Option.exists((effect) => effect.status === "pending" || effect.status === "running"),
8527 ),
8528 );
8529
8530 /**
8531 * A delegated child's result is held while its result run, or a run after it
8532 * that a second restart cut before it started, still has a continuation pending.
8533 */
8534 const childAwaitsRestartContinuation = (
8535 runs: ReadonlyArray<OrchestrationV2Run>,
8536 resultRun: OrchestrationV2Run,
8537 settledContinuationOf?: RunId,
8538 ) =>
8539 Effect.forEach(
8540 runs.filter(
8541 (run) =>
8542 run.id !== settledContinuationOf &&
8543 (run.id === resultRun.id || runRanAfter(run, resultRun)),
8544 ),
8545 awaitsRestartContinuation,
8546 ).pipe(Effect.map((pending) => pending.includes(true)));
8547
84978548 const planDelegatedCompletionDelivery = Effect.fn(
84988549 "orchestrationV2.planDelegatedCompletionDelivery",
84998550 )(function* (input: {
......@@ -8541,9 +8592,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
85418592 offer: false,
85428593 };
85438594 }
8595 // settled_only defers to a blocking delegate_task wait, which lives only as
8596 // long as the turn that spawned the task. Once that turn is over (a restart
8597 // continues it as a new run), nothing else delivers the result.
85448598 if (
85458599 (input.task.completionWake ?? "settled_only") === "settled_only" &&
8546 hasLiveRun(input.parentProjection)
8600 hasLiveRun({ runs: [input.parentRun] })
85478601 ) {
85488602 return {
85498603 task: input.updatedTask,
......@@ -8694,9 +8748,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
86948748 * delegated_task.wake-policy handler rewrites the same subagent row under
86958749 * that lock with a full-row payload, and unserialized writers clobber each
86968750 * other (stale policy on the terminal row, or a terminal row regressed to
8697 * running).
8751 * running). `settledContinuationOf` names the restart continuation that just
8752 * declined or failed, so its own still-running effect does not hold the child.
86988753 */
8699 const finalizeAppOwnedSubagent = (childThreadId: ThreadId) =>
8754 const finalizeAppOwnedSubagent = (
8755 childThreadId: ThreadId,
8756 options?: { readonly settledContinuationOf?: RunId },
8757 ) =>
87008758 Effect.gen(function* () {
87018759 const childControls = yield* projectionStore.getThreadRecords(
87028760 childThreadId,
......@@ -8719,6 +8777,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
87198777 if (terminalStatus === null) {
87208778 return;
87218779 }
8780 if (
8781 yield* childAwaitsRestartContinuation(
8782 childControls.runs,
8783 childRun,
8784 options?.settledContinuationOf,
8785 )
8786 ) {
8787 return;
8788 }
87228789
87238790 const childResult = yield* projectionStore.getThreadRecords(
87248791 childThreadId,
......@@ -9147,7 +9214,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
91479214 });
91489215 return;
91499216 }
9150 const parentIsLive = hasLiveRun(projection);
9217 // settled_only tasks defer only to a blocking wait in their spawning run.
9218 const parentIsLive = hasLiveRun({ runs: [parentRun] });
91519219 const pendingTaskIds =
91529220 projection.thread.archivedAt === null && projection.thread.deletedAt === null
91539221 ? projection.subagents
......@@ -9749,97 +9817,140 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
97499817 Effect.forkDetach,
97509818 );
97519819
9752 // Recover child results from projections. Queue recovery instead holds
9753 // unstarted runs until an explicit queue.resume command arrives.
9754 yield* projectionStore.getRecoveryThreadIds("subagent-results").pipe(
9755 Effect.flatMap((threadIds) =>
9756 Effect.forEach(
9757 threadIds,
9758 (threadId) =>
9759 Effect.gen(function* () {
9760 const thread = yield* projectionStore.getThreadShell(threadId);
9761 const parentThreadId = thread?.lineage.parentThreadId;
9762 if (parentThreadId === undefined || parentThreadId === null) return;
9763 yield* threadDispatch.withLock(parentThreadId, finalizeAppOwnedSubagent(threadId));
9764 }).pipe(
9765 Effect.catchCause((cause) =>
9766 Effect.logWarning("Failed to recover terminal app-owned subagent", {
9767 childThreadId: threadId,
9768 cause,
9769 }),
9770 ),
9771 ),
9772 { concurrency: 8, discard: true },
9773 ),
9774 ),
9775 Effect.catchCause((cause) =>
9776 Effect.logWarning("Failed to inspect app-owned subagents during recovery", {
9777 cause,
9778 }),
9779 ),
9780 );
9781 yield* projectionStore.getRecoveryThreadIds("delegated-completions").pipe(
9782 Effect.flatMap((threadIds) =>
9783 Effect.forEach(
9784 threadIds,
9785 (threadId) =>
9786 threadDispatch
9787 .withLock(
9788 threadId,
9789 Effect.gen(function* () {
9790 const projection = yield* projectionStore.getThreadRecords(threadId, [
9791 "runs",
9792 "messages",
9793 ]);
9794 const terminalDeliveryRunIds = projection.runs
9795 .filter((run) => delegatedTaskTerminalStatus(run.status) !== null)
9796 .filter((run) =>
9797 projection.messages.some(
9798 (message) =>
9799 message.id === run.userMessageId &&
9800 message.delegatedCompletion !== undefined,
9801 ),
9802 )
9803 .map((run) => run.id);
9804 for (const runId of terminalDeliveryRunIds) {
9805 yield* finalizeDelegatedCompletionDelivery(threadId, runId);
9806 }
9807 const refreshed =
9808 terminalDeliveryRunIds.length === 0
9809 ? projection
9810 : yield* projectionStore.getThreadRecords(threadId, ["runs", "messages"], {
9811 messageRoles: ["user"],
9812 });
9813 for (const run of refreshed.runs) {
9814 if (
9815 run.delegatedCompletion?.delivery !== null &&
9816 run.delegatedCompletion !== undefined
9817 ) {
9818 yield* offerDelegatedCompletionDelivery(threadId, run.id);
9819 }
9820 }
9821 }),
9822 )
9823 .pipe(
9820 // Settles child results and completion deliveries whose runs ended without
9821 // the listener above: before this boot, or in runtime reconciliation, which
9822 // it skips. Startup runs this after reconciliation and before the effect
9823 // worker. Queue recovery instead holds unstarted runs until an explicit
9824 // queue.resume command arrives.
9825 const recoverDelegatedTasks = Effect.gen(function* () {
9826 yield* projectionStore.getRecoveryThreadIds("subagent-results").pipe(
9827 Effect.flatMap((threadIds) =>
9828 Effect.forEach(
9829 threadIds,
9830 (threadId) =>
9831 Effect.gen(function* () {
9832 const thread = yield* projectionStore.getThreadShell(threadId);
9833 const parentThreadId = thread?.lineage.parentThreadId;
9834 if (parentThreadId === undefined || parentThreadId === null) return;
9835 yield* threadDispatch.withLock(parentThreadId, finalizeAppOwnedSubagent(threadId));
9836 }).pipe(
98249837 Effect.catchCause((cause) =>
9825 Effect.logWarning("Failed to recover delegated completion delivery", {
9826 threadId,
9838 Effect.logWarning("Failed to recover terminal app-owned subagent", {
9839 childThreadId: threadId,
98279840 cause,
98289841 }),
98299842 ),
98309843 ),
9831 { concurrency: 8, discard: true },
9844 { concurrency: 8, discard: true },
9845 ),
98329846 ),
9833 ),
9834 Effect.catchCause((cause) =>
9835 Effect.logWarning("Failed to inspect delegated completion delivery during recovery", {
9836 cause,
9837 }),
9838 ),
9839 );
9847 Effect.catchCause((cause) =>
9848 Effect.logWarning("Failed to inspect app-owned subagents during recovery", {
9849 cause,
9850 }),
9851 ),
9852 );
9853 yield* projectionStore.getRecoveryThreadIds("delegated-completions").pipe(
9854 Effect.flatMap((threadIds) =>
9855 Effect.forEach(
9856 threadIds,
9857 (threadId) =>
9858 threadDispatch
9859 .withLock(
9860 threadId,
9861 Effect.gen(function* () {
9862 const projection = yield* projectionStore.getThreadRecords(threadId, [
9863 "runs",
9864 "messages",
9865 ]);
9866 const terminalDeliveryRunIds = projection.runs
9867 .filter((run) => delegatedTaskTerminalStatus(run.status) !== null)
9868 .filter((run) =>
9869 projection.messages.some(
9870 (message) =>
9871 message.id === run.userMessageId &&
9872 message.delegatedCompletion !== undefined,
9873 ),
9874 )
9875 .map((run) => run.id);
9876 for (const runId of terminalDeliveryRunIds) {
9877 yield* finalizeDelegatedCompletionDelivery(threadId, runId);
9878 }
9879 const refreshed =
9880 terminalDeliveryRunIds.length === 0
9881 ? projection
9882 : yield* projectionStore.getThreadRecords(threadId, ["runs", "messages"], {
9883 messageRoles: ["user"],
9884 });
9885 for (const run of refreshed.runs) {
9886 if (
9887 run.delegatedCompletion?.delivery !== null &&
9888 run.delegatedCompletion !== undefined
9889 ) {
9890 yield* offerDelegatedCompletionDelivery(threadId, run.id);
9891 }
9892 }
9893 }),
9894 )
9895 .pipe(
9896 Effect.catchCause((cause) =>
9897 Effect.logWarning("Failed to recover delegated completion delivery", {
9898 threadId,
9899 cause,
9900 }),
9901 ),
9902 ),
9903 { concurrency: 8, discard: true },
9904 ),
9905 ),
9906 Effect.catchCause((cause) =>
9907 Effect.logWarning("Failed to inspect delegated completion delivery during recovery", {
9908 cause,
9909 }),
9910 ),
9911 );
9912 });
9913
9914 const delegatedTaskResultPending = (childThreadId: ThreadId) =>
9915 Effect.gen(function* () {
9916 const child = yield* projectionStore.getThreadRecords(
9917 childThreadId,
9918 ["runs", "messages", "subagents", "providerThreads", "providerTurns", "attempts"],
9919 { messageRoles: ["user"] },
9920 );
9921 const progress = delegatedTaskProgress(child);
9922 // A caller's older read saw a result; newer work since then means it is not final.
9923 if (progress.state !== "result_available") return true;
9924 if (progress.resultRun === undefined) return false;
9925 return yield* childAwaitsRestartContinuation(child.runs, progress.resultRun);
9926 }).pipe(
9927 Effect.mapError(
9928 (cause) => new OrchestratorProjectionError({ threadId: childThreadId, cause }),
9929 ),
9930 );
9931
9932 const recoverDelegatedTask = (threadId: ThreadId, sourceRunId: RunId) =>
9933 Effect.gen(function* () {
9934 const parentThreadId = yield* appOwnedSubagentParentThreadId(threadId);
9935 if (parentThreadId === undefined) return;
9936 yield* threadDispatch.withLock(
9937 parentThreadId,
9938 finalizeAppOwnedSubagent(threadId, { settledContinuationOf: sourceRunId }),
9939 );
9940 }).pipe(
9941 Effect.catchCause((cause) =>
9942 Effect.logWarning("Failed to recover delegated task after restart", {
9943 childThreadId: threadId,
9944 cause,
9945 }),
9946 ),
9947 );
98409948
98419949 return OrchestratorV2.of({
98429950 resumeQueuedRuns,
9951 recoverDelegatedTasks,
9952 recoverDelegatedTask,
9953 delegatedTaskResultPending,
98439954 dispatch: dispatchWithReceipt,
98449955 getTimelinePage: (threadId, options) =>
98459956 projectionStore
......@@ -9933,6 +10044,7 @@ export const layer: Layer.Layer<
993310044 | CommandPolicyV2
993410045 | CommandReceiptStoreV2
993510046 | ContextHandoffServiceV2
10047 | EffectOutbox.EffectOutboxV2
993610048 | EventSinkV2
993710049 | IdAllocatorV2
993810050 | ProjectStore.ProjectStoreV2
......@@ -9956,6 +10068,9 @@ const layerUnavailable: Layer.Layer<OrchestratorV2> = Layer.succeed(
995610068 cause: "Orchestration V2 live runtime is not configured.",
995710069 }),
995810070 ),
10071 recoverDelegatedTasks: Effect.void,
10072 recoverDelegatedTask: () => Effect.void,
10073 delegatedTaskResultPending: () => Effect.succeed(false),
995910074 dispatch: (command) =>
996010075 Effect.fail(
996110076 new OrchestratorDispatchError({
apps/server/src/orchestration-v2/ProjectionRecovery.test.ts+67
......@@ -420,6 +420,73 @@ it.effect("includes shared sessions and provider-owned background rosters in rec
420420 }).pipe(Effect.provide(TestLayer)),
421421);
422422
423it.effect("reads the run that owns a background roster, not a queued or resumed one", () =>
424 Effect.gen(function* () {
425 const projections = yield* ProjectionStore.ProjectionStoreV2;
426 const now = yield* DateTime.now;
427 const withRoster = Effect.fn(function* (threadId: ThreadId) {
428 yield* projections.apply({
429 id: EventId.make(`event:${threadId}:roster`),
430 type: "provider-thread.updated",
431 threadId,
432 driver,
433 providerInstanceId,
434 occurredAt: now,
435 payload: {
436 id: ProviderThreadId.make(`provider-thread:${threadId}`),
437 appThreadId: threadId,
438 ownerNodeId: null,
439 driver,
440 providerInstanceId,
441 providerSessionId: null,
442 nativeThreadRef: null,
443 nativeConversationHeadRef: null,
444 status: "idle",
445 firstRunOrdinal: null,
446 lastRunOrdinal: null,
447 handoffIds: [],
448 forkedFrom: null,
449 createdAt: now,
450 updatedAt: now,
451 pendingBackgroundTasks: [
452 { taskId: "background", description: "Still running", kind: "command" },
453 ],
454 },
455 });
456 });
457 const runIds = (threadId: ThreadId) =>
458 projections
459 .getRuntimeRecoveryProjection(threadId)
460 .pipe(Effect.map((state) => state.runs.map((run) => run.id)));
461 // Settled with background work left, then a queued follow-up.
462 const settled = yield* createThread("roster-before-queue");
463 const settledRun = yield* createRun(settled, "completed", {
464 providerThreadId: ProviderThreadId.make(`provider-thread:${settled}`),
465 });
466 const queuedRun = yield* createRun(settled, "queued", { ordinal: 2, completedAt: null });
467 yield* withRoster(settled);
468 assert.deepEqual(yield* runIds(settled), [settledRun, queuedRun]);
469 // A resumed queued run (ordinal 1) ended after a continuation (ordinal 2).
470 const resumed = yield* createThread("roster-after-resume");
471 const resumedRun = yield* createRun(resumed, "completed", {
472 providerThreadId: ProviderThreadId.make(`provider-thread:${resumed}`),
473 completedAt: DateTime.makeUnsafe("2026-10-03T10:05:00.000Z"),
474 });
475 const continuationRun = yield* createRun(resumed, "completed", {
476 providerThreadId: ProviderThreadId.make(`provider-thread:${resumed}`),
477 ordinal: 2,
478 completedAt: DateTime.makeUnsafe("2026-10-03T10:00:00.000Z"),
479 });
480 yield* withRoster(resumed);
481 assert.deepEqual(yield* runIds(resumed), [resumedRun, continuationRun]);
482 // Without a roster, settled history is not read.
483 const quiet = yield* createThread("no-roster-before-queue");
484 yield* createRun(quiet, "completed");
485 const quietQueued = yield* createRun(quiet, "queued", { ordinal: 2, completedAt: null });
486 assert.deepEqual(yield* runIds(quiet), [quietQueued]);
487 }).pipe(Effect.provide(TestLayer)),
488);
489
423490it.effect("marks fork descendants unreadable when their source is missing or corrupt", () =>
424491 Effect.gen(function* () {
425492 const projections = yield* ProjectionStore.ProjectionStoreV2;
apps/server/src/orchestration-v2/ProjectionStore.test.ts+55
......@@ -2276,8 +2276,33 @@ it.layer(TestLayer)("ProjectionStoreV2", (it) => {
22762276 },
22772277 });
22782278 yield* assertSummary("Plan limit reached.", "usage_limit");
2279 // A restart continuation ran ahead of the held queue and ended before the
2280 // resumed failed run: the failure is still the latest executed run.
2281 const aheadRunId = RunId.make("run:limit-shell:ran-ahead");
2282 yield* store.apply({
2283 id: EventId.make("event:limit-shell:ran-ahead"),
2284 type: "run.created",
2285 threadId,
2286 runId: aheadRunId,
2287 nodeId: NodeId.make("node:limit-shell:ran-ahead"),
2288 driver,
2289 providerInstanceId,
2290 occurredAt: now,
2291 payload: {
2292 ...original,
2293 id: aheadRunId,
2294 ordinal: original.ordinal + 3,
2295 rootNodeId: NodeId.make("node:limit-shell:ran-ahead"),
2296 userMessageId: MessageId.make("message:limit-shell:ran-ahead"),
2297 status: "completed",
2298 startedAt: DateTime.subtract(now, { minutes: 10 }),
2299 completedAt: DateTime.subtract(now, { minutes: 5 }),
2300 },
2301 });
2302 yield* assertSummary("Plan limit reached.", "usage_limit");
22792303 const sql = yield* SqlClient.SqlClient;
22802304 // The rest of this case treats the failed run as the latest run.
2305 yield* sql`DELETE FROM orchestration_v2_projection_runs WHERE run_id = ${aheadRunId}`;
22812306 yield* sql`DELETE FROM orchestration_v2_projection_runs WHERE run_id = ${queuedRunId}`;
22822307 yield* assertSummary("Plan limit reached.", "usage_limit");
22832308 yield* sql`DELETE FROM orchestration_v2_projection_runs WHERE run_id = ${cancelledRunId}`;
......@@ -4372,6 +4397,36 @@ it.layer(TestLayer)("ProjectionStoreV2", (it) => {
43724397 released && projectThreadAwarenessV2({ environmentId, project, thread: released });
43734398 assert.equal(state?.phase, "completed");
43744399 assert.equal(state?.updatedAt, DateTime.formatIso(releasedAt));
4400
4401 // A later provider must not hide this owner's roster from restart recovery.
4402 yield* store.apply({
4403 id: EventId.make("event:held-completion:monitor-restarted"),
4404 type: "provider-thread.updated",
4405 threadId,
4406 occurredAt: releasedAt,
4407 payload: { ...providerThread, updatedAt: releasedAt },
4408 });
4409 const newerRunId = RunId.make("run:held-completion:new-provider");
4410 yield* store.apply({
4411 id: EventId.make("event:held-completion:new-provider"),
4412 type: "run.created",
4413 threadId,
4414 runId: newerRunId,
4415 occurredAt: releasedAt,
4416 payload: {
4417 ...(yield* store.getThreadProjection(threadId)).runs[0]!,
4418 id: newerRunId,
4419 ordinal: 2,
4420 providerThreadId: ProviderThreadId.make("provider-thread:held-completion:new-provider"),
4421 requestedAt: releasedAt,
4422 startedAt: releasedAt,
4423 completedAt: releasedAt,
4424 },
4425 });
4426 assert.deepEqual(
4427 (yield* store.getRuntimeRecoveryProjection(threadId)).runs.map((run) => run.id),
4428 [runId, newerRunId],
4429 );
43754430 }),
43764431 );
43774432});
apps/server/src/orchestration-v2/ProjectionStore.ts+39-5
......@@ -522,8 +522,10 @@ function needsRecovery(
522522 }
523523 case "runtime":
524524 return (
525 projection.runs.some((run) =>
526 ["queued", "preparing", "starting", "running", "waiting"].includes(run.status),
525 projection.runs.some(
526 (run) =>
527 ["preparing", "starting", "running", "waiting"].includes(run.status) ||
528 (run.status === "queued" && run.queueHeld !== true),
527529 ) ||
528530 projection.runtimeRequests.some((request) => request.status === "pending") ||
529531 projection.providerSessions.some(
......@@ -3307,7 +3309,10 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
33073309 latest.status = 'cancelled'
33083310 AND json_extract(latest.payload_json, '$.startedAt') IS NULL
33093311 )
3310 ORDER BY latest.ordinal DESC, latest.run_id DESC LIMIT 1
3312 -- latestExecutedRun: the run that ended last (runRanAfter).
3313 ORDER BY latest.completed_at IS NULL DESC, latest.completed_at DESC,
3314 latest.ordinal DESC, latest.run_id DESC
3315 LIMIT 1
33113316 ) AND r.status = 'failed'
33123317 INNER JOIN orchestration_v2_projection_turn_items item ON item.turn_item_id = (
33133318 SELECT error.turn_item_id FROM orchestration_v2_projection_turn_items error
......@@ -3433,7 +3438,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
34333438 ELSE 0 END
34343439 )
34353440 SELECT thread_id FROM orchestration_v2_projection_runs
3436 WHERE status IN ('queued', 'preparing', 'starting', 'running', 'waiting')
3441 WHERE status IN ('preparing', 'starting', 'running', 'waiting')
3442 UNION
3443 -- A held queue already went through recovery; rereading it on
3444 -- every boot costs a projection read per held thread.
3445 SELECT thread_id FROM orchestration_v2_projection_runs
3446 WHERE status = 'queued'
3447 AND CASE WHEN json_valid(payload_json)
3448 THEN json_extract(payload_json, '$.queueHeld') IS NOT 1
3449 ELSE 1 END
34373450 UNION
34383451 SELECT thread_id FROM orchestration_v2_projection_runtime_requests
34393452 WHERE status = 'pending'
......@@ -3805,6 +3818,24 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
38053818 WHERE latest.thread_id = ${threadId}
38063819 ORDER BY latest.ordinal DESC LIMIT 1
38073820 )
3821 -- Retain the latest run for each provider thread with lost
3822 -- background work, including owners used before a handoff.
3823 OR run.run_id IN (
3824 SELECT (
3825 SELECT ended.run_id FROM orchestration_v2_projection_runs AS ended
3826 WHERE ended.thread_id = ${threadId}
3827 AND ended.provider_thread_id = roster.provider_thread_id
3828 AND ended.status NOT IN ('queued', 'rolled_back')
3829 ORDER BY ended.completed_at IS NULL DESC, ended.completed_at DESC,
3830 ended.ordinal DESC
3831 LIMIT 1
3832 )
3833 FROM orchestration_v2_projection_provider_threads AS roster
3834 WHERE roster.thread_id = ${threadId}
3835 AND CASE WHEN json_valid(roster.payload_json)
3836 THEN json_array_length(roster.payload_json, '$.pendingBackgroundTasks') > 0
3837 ELSE 0 END
3838 )
38083839 OR run.run_id IN (
38093840 SELECT item.run_id FROM orchestration_v2_projection_turn_items AS item
38103841 WHERE item.thread_id = ${threadId}
......@@ -4899,7 +4930,10 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
48994930 candidate.status = 'cancelled'
49004931 AND json_extract(candidate.payload_json, '$.startedAt') IS NULL
49014932 )
4902 ORDER BY candidate.ordinal DESC, candidate.run_id DESC LIMIT 1
4933 -- latestExecutedRun: the run that ended last (runRanAfter).
4934 ORDER BY candidate.completed_at IS NULL DESC, candidate.completed_at DESC,
4935 candidate.ordinal DESC, candidate.run_id DESC
4936 LIMIT 1
49034937 ) AND blocked.status = 'failed'
49044938 WHERE t.deleted_at IS NULL${threadId === undefined ? sql`` : sql` AND t.thread_id = ${threadId}`}${
49054939 location === "active"
apps/server/src/orchestration-v2/ProviderRuntimeRecoveryPerformance.test.ts created+510
......@@ -0,0 +1,510 @@
1import { assert, it } from "@effect/vitest";
2import {
3 CheckpointScopeId,
4 CommandId,
5 EventId,
6 MessageId,
7 NodeId,
8 type OrchestrationV2AppThread,
9 type OrchestrationV2DomainEvent,
10 type OrchestrationV2Run,
11 ProjectId,
12 ProviderDriverKind,
13 ProviderInstanceId,
14 ProviderSessionId,
15 ProviderThreadId,
16 ProviderTurnId,
17 RunAttemptId,
18 RunId,
19 ThreadId,
20 TurnItemId,
21} from "@t3tools/contracts";
22import * as Console from "effect/Console";
23import * as DateTime from "effect/DateTime";
24import * as Effect from "effect/Effect";
25import * as Layer from "effect/Layer";
26import * as SqlClient from "effect/unstable/sql/SqlClient";
27
28import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
29import * as ServerSettings from "../serverSettings.ts";
30import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
31import * as EffectOutbox from "./EffectOutbox.ts";
32import * as EventSink from "./EventSink.ts";
33import * as EventStore from "./EventStore.ts";
34import * as IdAllocator from "./IdAllocator.ts";
35import * as ProjectionStore from "./ProjectionStore.ts";
36import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
37
38// Startup recovery cost for restart continuation. The default case is small
39// enough for CI and asserts correctness; T3_BENCH_RECOVERY=1 adds the matrix.
40// `it.live` keeps a real clock: each reconcile gets a fresh command id, so the
41// second recover cannot hide behind command receipt dedup.
42
43const stores = Layer.mergeAll(
44 EventStore.layer,
45 ProjectionStore.layer,
46 EffectOutbox.layer,
47 IdAllocator.layer,
48).pipe(Layer.provideMerge(SqlitePersistenceMemory));
49const TestLayer = ProviderRuntimeRecovery.layer.pipe(
50 Layer.provideMerge(EventSink.layer.pipe(Layer.provideMerge(stores))),
51 Layer.provideMerge(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true })),
52);
53
54const providerInstanceId = ProviderInstanceId.make("codex");
55const driver = ProviderDriverKind.make("codex");
56const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.4" };
57const projectId = ProjectId.make("project:recovery-bench");
58const ITEMS_PER_SETTLED_THREAD = 10;
59const OPEN_ITEMS_PER_ACTIVE_THREAD = 3;
60const WAITING_THREADS = 3;
61const SEED_BATCH = 500;
62
63type UnstampedEvent = OrchestrationV2DomainEvent extends infer Event
64 ? Event extends OrchestrationV2DomainEvent
65 ? Omit<Event, "id" | "occurredAt">
66 : never
67 : never;
68
69interface Scenario {
70 readonly settled: number;
71 readonly active: number;
72}
73
74interface Seeded {
75 readonly unheldQueuedThreads: number;
76 readonly waitingThreads: number;
77}
78
79const seedScenario = Effect.fn(function* (scenario: Scenario) {
80 const projections = yield* ProjectionStore.ProjectionStoreV2;
81 const outbox = yield* EffectOutbox.EffectOutboxV2;
82 const sql = yield* SqlClient.SqlClient;
83 const now = yield* DateTime.now;
84 let eventIndex = 0;
85 const apply = (event: UnstampedEvent) =>
86 projections.apply({
87 ...event,
88 id: EventId.make(`event:bench:${eventIndex++}`),
89 occurredAt: now,
90 } as OrchestrationV2DomainEvent);
91
92 const thread = (id: ThreadId, overrides: Partial<OrchestrationV2AppThread> = {}) =>
93 apply({
94 type: "thread.created",
95 threadId: id,
96 payload: {
97 createdBy: "user",
98 creationSource: "web",
99 id,
100 projectId,
101 title: id,
102 providerInstanceId,
103 modelSelection,
104 runtimeMode: "full-access",
105 interactionMode: "default",
106 branch: null,
107 worktreePath: null,
108 activeProviderThreadId: null,
109 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: id },
110 forkedFrom: null,
111 createdAt: now,
112 updatedAt: now,
113 archivedAt: null,
114 settledOverride: null,
115 settledAt: null,
116 lastVisitedAt: null,
117 deletedAt: null,
118 ...overrides,
119 },
120 });
121
122 const run = (
123 threadId: ThreadId,
124 ordinal: number,
125 status: OrchestrationV2Run["status"],
126 overrides: Partial<OrchestrationV2Run> = {},
127 ) => {
128 const runId = RunId.make(`run:${threadId}:${ordinal}`);
129 return apply({
130 type: "run.created",
131 threadId,
132 runId,
133 payload: {
134 id: runId,
135 threadId,
136 ordinal,
137 providerInstanceId,
138 modelSelection,
139 providerThreadId: null,
140 userMessageId: MessageId.make(`message:${runId}`),
141 rootNodeId: null,
142 activeAttemptId: null,
143 status,
144 requestedAt: now,
145 startedAt: status === "queued" ? null : now,
146 completedAt: status === "completed" ? now : null,
147 checkpointId: null,
148 contextHandoffId: null,
149 ...overrides,
150 },
151 }).pipe(Effect.as(runId));
152 };
153
154 const commandItem = (input: {
155 readonly threadId: ThreadId;
156 readonly runId: RunId;
157 readonly ordinal: number;
158 readonly status: "running" | "completed";
159 readonly nodeId?: NodeId;
160 readonly providerThreadId?: ProviderThreadId;
161 readonly providerTurnId?: ProviderTurnId;
162 }) =>
163 apply({
164 type: "turn-item.updated",
165 threadId: input.threadId,
166 runId: input.runId,
167 payload: {
168 id: TurnItemId.make(`item:${input.runId}:${input.ordinal}`),
169 threadId: input.threadId,
170 runId: input.runId,
171 nodeId: input.nodeId ?? null,
172 providerThreadId: input.providerThreadId ?? null,
173 providerTurnId: input.providerTurnId ?? null,
174 nativeItemRef: null,
175 parentItemId: null,
176 ordinal: input.ordinal,
177 type: "command_execution",
178 status: input.status,
179 title: `Command ${input.ordinal}`,
180 input: `echo ${input.ordinal}`,
181 output: input.status === "completed" ? `${input.ordinal}\n` : "",
182 ...(input.status === "completed" ? { exitCode: 0 } : {}),
183 startedAt: now,
184 completedAt: input.status === "completed" ? now : null,
185 updatedAt: now,
186 },
187 });
188
189 const settledThread = Effect.fn(function* (index: number) {
190 const threadId = ThreadId.make(`thread:bench:settled:${index}`);
191 yield* thread(threadId);
192 const runId = yield* run(threadId, 1, "completed");
193 for (let ordinal = 1; ordinal <= ITEMS_PER_SETTLED_THREAD; ordinal += 1) {
194 yield* commandItem({ threadId, runId, ordinal, status: "completed" });
195 }
196 // ~5% of threads carry a queued follow-up; half are already held, and an
197 // earlier recovery already handled those, so only unheld ones are candidates.
198 if (index % 20 === 0) {
199 const queueHeld = index % 40 === 0;
200 yield* run(threadId, 2, "queued", { queueHeld });
201 return !queueHeld;
202 }
203 return false;
204 });
205
206 const activeThread = Effect.fn(function* (index: number) {
207 const threadId = ThreadId.make(`thread:bench:active:${index}`);
208 const providerSessionId = ProviderSessionId.make(`session:bench:${index}`);
209 const providerThreadId = ProviderThreadId.make(`provider-thread:bench:${index}`);
210 const runId = RunId.make(`run:${threadId}:1`);
211 const attemptId = RunAttemptId.make(`attempt:bench:${index}`);
212 const nodeId = NodeId.make(`node:bench:${index}`);
213 const providerTurnId = ProviderTurnId.make(`provider-turn:bench:${index}`);
214 yield* thread(threadId, { activeProviderThreadId: providerThreadId });
215 yield* apply({
216 type: "provider-session.attached",
217 threadId,
218 driver,
219 providerInstanceId,
220 payload: {
221 id: providerSessionId,
222 driver,
223 providerInstanceId,
224 status: "ready",
225 cwd: "/workspace",
226 model: modelSelection.model,
227 capabilities: CodexProviderCapabilitiesV2,
228 createdAt: now,
229 updatedAt: now,
230 lastError: null,
231 },
232 });
233 yield* apply({
234 type: "provider-thread.updated",
235 threadId,
236 driver,
237 providerInstanceId,
238 payload: {
239 id: providerThreadId,
240 appThreadId: threadId,
241 ownerNodeId: null,
242 driver,
243 providerInstanceId,
244 providerSessionId,
245 nativeThreadRef: { driver, nativeId: `native:bench:${index}`, strength: "strong" },
246 nativeConversationHeadRef: null,
247 status: "active",
248 firstRunOrdinal: 1,
249 lastRunOrdinal: 1,
250 handoffIds: [],
251 forkedFrom: null,
252 createdAt: now,
253 updatedAt: now,
254 pendingBackgroundTasks: [],
255 },
256 });
257 yield* run(threadId, 1, "running", {
258 providerThreadId,
259 rootNodeId: nodeId,
260 activeAttemptId: attemptId,
261 });
262 yield* apply({
263 type: "run-attempt.created",
264 threadId,
265 runId,
266 payload: {
267 id: attemptId,
268 runId,
269 attemptOrdinal: 1,
270 rootNodeId: nodeId,
271 providerInstanceId,
272 providerThreadId,
273 providerTurnId,
274 reason: "initial",
275 status: "running",
276 startedAt: now,
277 completedAt: null,
278 },
279 });
280 yield* apply({
281 type: "node.updated",
282 threadId,
283 runId,
284 nodeId,
285 providerInstanceId,
286 payload: {
287 id: nodeId,
288 threadId,
289 runId,
290 parentNodeId: null,
291 rootNodeId: nodeId,
292 kind: "root_turn",
293 status: "running",
294 countsForRun: true,
295 providerThreadId,
296 providerTurnId,
297 nativeItemRef: null,
298 runtimeRequestId: null,
299 checkpointScopeId: null,
300 startedAt: now,
301 completedAt: null,
302 },
303 });
304 yield* apply({
305 type: "provider-turn.updated",
306 threadId,
307 runId,
308 nodeId,
309 providerInstanceId,
310 payload: {
311 id: providerTurnId,
312 providerThreadId,
313 nodeId,
314 runAttemptId: attemptId,
315 nativeTurnRef: { driver, nativeId: `native-turn:bench:${index}`, strength: "strong" },
316 ordinal: 1,
317 status: "running",
318 startedAt: now,
319 completedAt: null,
320 },
321 });
322 for (let ordinal = 1; ordinal <= OPEN_ITEMS_PER_ACTIVE_THREAD; ordinal += 1) {
323 yield* commandItem({
324 threadId,
325 runId,
326 ordinal,
327 status: "running",
328 nodeId,
329 providerThreadId,
330 providerTurnId,
331 });
332 }
333 });
334
335 const waitingThread = Effect.fn(function* (index: number) {
336 const threadId = ThreadId.make(`thread:bench:waiting:${index}`);
337 yield* thread(threadId);
338 yield* run(threadId, 1, "completed");
339 const runId = yield* run(threadId, 2, "waiting");
340 yield* outbox.enqueue([
341 {
342 id: `effect:bench:checkpoint:${runId}`,
343 commandId: CommandId.make(`command:effect:checkpoint.capture:${runId}`),
344 threadId,
345 request: {
346 type: "checkpoint.capture",
347 runId,
348 scopeId: CheckpointScopeId.make(`scope:bench:${index}`),
349 },
350 },
351 ]);
352 });
353
354 let unheldQueuedThreads = 0;
355 for (let start = 0; start < scenario.settled; start += SEED_BATCH) {
356 const end = Math.min(scenario.settled, start + SEED_BATCH);
357 yield* sql.withTransaction(
358 Effect.gen(function* () {
359 for (let index = start; index < end; index += 1) {
360 if (yield* settledThread(index)) unheldQueuedThreads += 1;
361 }
362 }),
363 );
364 }
365 yield* sql.withTransaction(
366 Effect.gen(function* () {
367 for (let index = 0; index < scenario.active; index += 1) yield* activeThread(index);
368 for (let index = 0; index < WAITING_THREADS; index += 1) yield* waitingThread(index);
369 }),
370 );
371 return { unheldQueuedThreads, waitingThreads: WAITING_THREADS } satisfies Seeded;
372});
373
374const timed = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
375 Effect.gen(function* () {
376 const start = performance.now();
377 const value = yield* effect;
378 return [value, performance.now() - start] as const;
379 });
380
381const continuationEffectCount = Effect.gen(function* () {
382 const sql = yield* SqlClient.SqlClient;
383 const rows = yield* sql<{ readonly count: number }>`
384 SELECT COUNT(*) AS count FROM orchestration_v2_effect_outbox
385 WHERE effect_id LIKE 'effect:restart-continuation:%' AND status = 'pending'
386 `;
387 return Number(rows[0]?.count ?? 0);
388});
389
390interface Measurement {
391 readonly settled: number;
392 readonly active: number;
393 readonly seedMs: number;
394 readonly candidates: number;
395 readonly selectMs: number;
396 readonly prepareMs: number;
397 readonly shutdownMs: number;
398 readonly startupAfterShutdownMs: number;
399 readonly crashRecoverMs: number;
400 readonly secondRecoverMs: number;
401}
402
403/** A graceful restart: prepare intent, reconcile on shutdown, recover on boot. */
404const measureGraceful = Effect.fn(function* (scenario: Scenario) {
405 const [seeded, seedMs] = yield* timed(seedScenario(scenario));
406 const projections = yield* ProjectionStore.ProjectionStoreV2;
407 const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;
408 const [candidates, selectMs] = yield* timed(projections.getRecoveryThreadIds("runtime"));
409 assert.equal(
410 candidates.length,
411 scenario.active + seeded.unheldQueuedThreads + seeded.waitingThreads,
412 "runtime recovery candidates",
413 );
414 const [, prepareMs] = yield* timed(recovery.prepareForShutdown);
415 assert.equal(yield* continuationEffectCount, scenario.active, "prepared continuations");
416 const [, shutdownMs] = yield* timed(recovery.reconcile("shutdown"));
417 const [, startupAfterShutdownMs] = yield* timed(recovery.recover);
418 assert.equal(yield* continuationEffectCount, scenario.active, "continuations after boot");
419 return {
420 seedMs,
421 candidates: candidates.length,
422 selectMs,
423 prepareMs,
424 shutdownMs,
425 startupAfterShutdownMs,
426 };
427});
428
429/** A crash: no shutdown hook ran, so boot recovery records the continuations. */
430const measureCrash = Effect.fn(function* (scenario: Scenario) {
431 yield* seedScenario(scenario);
432 const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;
433 const [summary, crashRecoverMs] = yield* timed(recovery.recover);
434 assert.equal(summary.terminalizedRuns, scenario.active, "terminalized active runs");
435 assert.equal(summary.stoppedSessions, scenario.active, "stopped active sessions");
436 assert.equal(yield* continuationEffectCount, scenario.active, "recorded continuations");
437 const [, secondRecoverMs] = yield* timed(recovery.recover);
438 assert.equal(yield* continuationEffectCount, scenario.active, "second recover is idempotent");
439 return { crashRecoverMs, secondRecoverMs };
440});
441
442const measure = Effect.fn(function* (scenario: Scenario) {
443 const graceful = yield* measureGraceful(scenario).pipe(Effect.provide(Layer.fresh(TestLayer)));
444 const crash = yield* measureCrash(scenario).pipe(Effect.provide(Layer.fresh(TestLayer)));
445 return { ...scenario, ...graceful, ...crash } satisfies Measurement;
446});
447
448const formatTable = (rows: ReadonlyArray<Measurement>) => {
449 const ms = (value: number) => value.toFixed(1);
450 const perActive = (value: number, active: number) => (value / Math.max(active, 1)).toFixed(2);
451 const header = [
452 "N",
453 "A",
454 "seed",
455 "cand",
456 "select",
457 "prepare",
458 "prep/A",
459 "shutdown",
460 "boot-after",
461 "crash-recover",
462 "recover/A",
463 "recover-2",
464 ];
465 const lines = rows.map((row) => [
466 String(row.settled),
467 String(row.active),
468 ms(row.seedMs),
469 String(row.candidates),
470 ms(row.selectMs),
471 ms(row.prepareMs),
472 perActive(row.prepareMs, row.active),
473 ms(row.shutdownMs),
474 ms(row.startupAfterShutdownMs),
475 ms(row.crashRecoverMs),
476 perActive(row.crashRecoverMs, row.active),
477 ms(row.secondRecoverMs),
478 ]);
479 const widths = header.map((cell, column) =>
480 Math.max(cell.length, ...lines.map((line) => line[column]!.length)),
481 );
482 return [header, ...lines]
483 .map((line) => line.map((cell, column) => cell.padStart(widths[column]!)).join(" "))
484 .join("\n");
485};
486
487it.live(
488 "restart recovery records one continuation per active thread and is idempotent",
489 () =>
490 Effect.gen(function* () {
491 const row = yield* measure({ settled: 500, active: 20 });
492 yield* Console.log(`provider runtime recovery (ms)\n${formatTable([row])}`);
493 }),
494 60_000,
495);
496
497it.live.skipIf(process.env.T3_BENCH_RECOVERY !== "1")(
498 "restart recovery scales with active threads, not settled history",
499 () =>
500 Effect.gen(function* () {
501 const rows: Array<Measurement> = [];
502 for (const settled of [5_000, 50_000]) {
503 for (const active of [10, 100, 1_000]) {
504 rows.push(yield* measure({ settled, active }));
505 yield* Console.log(`provider runtime recovery (ms)\n${formatTable(rows)}`);
506 }
507 }
508 }),
509 3_600_000,
510);
apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts+148
......@@ -1272,3 +1272,151 @@ it.effect(
12721272 }).pipe(Effect.provide(layer));
12731273 },
12741274);
1275
1276it.effect("leaves delegated tasks to their own child threads after process loss", () => {
1277 const threadId = ThreadId.make("thread_recovery_delegation");
1278 const settledRunId = RunId.make("run_recovery_delegation_settled");
1279 const runningRunId = RunId.make("run_recovery_delegation_running");
1280 const providerThreadId = ProviderThreadId.make("provider_thread_recovery_delegation");
1281 const claudeInstanceId = ProviderInstanceId.make("claude");
1282 const driver = ProviderDriverKind.make("claude");
1283 const settledTaskId = NodeId.make("node_recovery_delegation_settled");
1284 const runningTaskId = NodeId.make("node_recovery_delegation_running");
1285 const nativeSubagentId = NodeId.make("node_recovery_native_subagent");
1286 const delegatedTask = (id: NodeId, runId: RunId) => ({
1287 id,
1288 runId,
1289 origin: "app_owned",
1290 childThreadId: ThreadId.make(`thread:delegated-task:${id}`),
1291 driver,
1292 providerInstanceId: claudeInstanceId,
1293 status: "running",
1294 });
1295 const subagentItem = (subagentId: NodeId, runId: RunId, origin: string) => ({
1296 id: TurnItemId.make(`turn_item:${subagentId}`),
1297 runId,
1298 nodeId: subagentId,
1299 providerThreadId,
1300 type: "subagent",
1301 status: "running",
1302 subagentId,
1303 origin,
1304 childThreadId:
1305 origin === "app_owned" ? ThreadId.make(`thread:delegated-task:${subagentId}`) : null,
1306 providerInstanceId: claudeInstanceId,
1307 title: `subagent ${subagentId}`,
1308 });
1309 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
1310 null;
1311 const projection = {
1312 thread: { id: threadId, providerInstanceId: claudeInstanceId },
1313 runtimeRequests: [],
1314 providerSessions: [],
1315 providerThreads: [
1316 {
1317 id: providerThreadId,
1318 driver,
1319 providerInstanceId: claudeInstanceId,
1320 ownerNodeId: null,
1321 status: "idle",
1322 pendingBackgroundTasks: [],
1323 },
1324 ],
1325 providerTurns: [],
1326 runs: [
1327 {
1328 id: settledRunId,
1329 ordinal: 1,
1330 status: "completed",
1331 providerThreadId,
1332 providerInstanceId: claudeInstanceId,
1333 },
1334 {
1335 id: runningRunId,
1336 ordinal: 2,
1337 status: "running",
1338 providerThreadId,
1339 providerInstanceId: claudeInstanceId,
1340 },
1341 ],
1342 attempts: [],
1343 nodes: [
1344 { id: settledTaskId, runId: settledRunId, status: "running", kind: "subagent" },
1345 { id: runningTaskId, runId: runningRunId, status: "running", kind: "subagent" },
1346 { id: nativeSubagentId, runId: settledRunId, status: "running", kind: "subagent" },
1347 ],
1348 subagents: [
1349 delegatedTask(settledTaskId, settledRunId),
1350 delegatedTask(runningTaskId, runningRunId),
1351 {
1352 ...delegatedTask(nativeSubagentId, settledRunId),
1353 origin: "provider_native",
1354 childThreadId: null,
1355 },
1356 ],
1357 messages: [],
1358 turnItems: [
1359 subagentItem(settledTaskId, settledRunId, "app_owned"),
1360 subagentItem(runningTaskId, runningRunId, "app_owned"),
1361 subagentItem(nativeSubagentId, settledRunId, "provider_native"),
1362 ],
1363 } as unknown as OrchestrationV2ThreadProjection;
1364 const layer = ProviderRuntimeRecovery.layer.pipe(
1365 Layer.provide(ServerSettings.layerTest()),
1366 Layer.provide(
1367 Layer.mergeAll(
1368 Layer.mock(ProjectionStore.ProjectionStoreV2)({
1369 getRecoveryThreadIds: () => Effect.succeed([threadId]),
1370 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
1371 }),
1372 Layer.mock(EventSink.EventSinkV2)({
1373 commitCommand: (input) => {
1374 committedInput = input;
1375 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
1376 },
1377 }),
1378 IdAllocator.layer,
1379 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
1380 runRecoveryOnce: Effect.succeed(false),
1381 }),
1382 Layer.mock(EffectOutbox.EffectOutboxV2)({
1383 listByCommandId: () => Effect.succeed([]),
1384 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
1385 }),
1386 ),
1387 ),
1388 );
1389
1390 return Effect.gen(function* () {
1391 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
1392 const events = committedInput?.events ?? [];
1393 const touched = events.flatMap((event) =>
1394 event.type === "turn-item.updated"
1395 ? event.payload.type === "subagent"
1396 ? [event.payload.subagentId]
1397 : []
1398 : event.type === "subagent.updated" || event.type === "node.updated"
1399 ? [event.payload.id]
1400 : [],
1401 );
1402 // The cut run itself is cancelled; only the provider-native subagent dies with it.
1403 assert.isTrue(
1404 events.some(
1405 (event) =>
1406 event.type === "run.updated" &&
1407 event.payload.id === runningRunId &&
1408 event.payload.status === "cancelled",
1409 ),
1410 );
1411 assert.notInclude(touched, settledTaskId);
1412 assert.notInclude(touched, runningTaskId);
1413 assert.include(touched, nativeSubagentId);
1414 // The note lists only provider-native work: delegated tasks report back on their own.
1415 const noted = events.flatMap((event) =>
1416 event.type === "run.background-work-cancelled"
1417 ? event.payload.restartCancelledBackgroundWork.map((work) => work.label)
1418 : [],
1419 );
1420 assert.deepEqual(noted, [`subagent ${nativeSubagentId}`]);
1421 }).pipe(Effect.provide(layer));
1422});
apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts+71-28
......@@ -1,12 +1,15 @@
1import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
12import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
23import {
34 CommandId,
45 type OrchestrationV2DomainEvent,
56 type ProviderThreadId,
67 type OrchestrationV2RestartCancelledBackgroundWork,
8 type OrchestrationV2Subagent,
79 type OrchestrationV2ThreadProjection,
810 ThreadId,
911} from "@t3tools/contracts";
12import * as Cause from "effect/Cause";
1013import * as Context from "effect/Context";
1114import * as DateTime from "effect/DateTime";
1215import * as Effect from "effect/Effect";
......@@ -93,6 +96,23 @@ function isNonterminalNodeStatus(status: string): boolean {
9396 return status === "pending" || status === "running" || status === "waiting";
9497}
9598
99/**
100 * A delegate_task child. Its own thread is reconciled and continued on its own
101 * and reports back through the app, so it is not provider background work.
102 */
103function isAppOwnedDelegation(task: {
104 readonly origin: OrchestrationV2Subagent["origin"];
105 readonly childThreadId: ThreadId | null;
106}): boolean {
107 return task.origin === "app_owned" && task.childThreadId !== null;
108}
109
110function isAppOwnedDelegationItem(
111 item: OrchestrationV2ThreadProjection["turnItems"][number],
112): boolean {
113 return item.type === "subagent" && isAppOwnedDelegation(item);
114}
115
96116function providerThreadHasPendingBackgroundTasks(
97117 providerThread: OrchestrationV2ThreadProjection["providerThreads"][number],
98118): boolean {
......@@ -138,7 +158,11 @@ function providerThreadsWithOpenBackgroundWork(
138158): ReadonlySet<ProviderThreadId> {
139159 const ids = new Set<ProviderThreadId>();
140160 for (const item of projection.turnItems ?? []) {
141 if (!isBackgroundCapableTurnItemType(item.type) || !isNonterminalTurnItemStatus(item.status))
161 if (
162 !isBackgroundCapableTurnItemType(item.type) ||
163 !isNonterminalTurnItemStatus(item.status) ||
164 isAppOwnedDelegationItem(item)
165 )
142166 continue;
143167 const providerThreadId =
144168 item.providerThreadId ??
......@@ -166,7 +190,7 @@ function latestStartedRun(
166190 run.providerThreadId === providerThreadId &&
167191 run.status !== "queued" &&
168192 run.status !== "rolled_back" &&
169 (latest === undefined || run.ordinal > latest.ordinal)
193 (latest === undefined || runRanAfter(run, latest))
170194 ? run
171195 : latest,
172196 undefined,
......@@ -222,6 +246,13 @@ export const make = Effect.gen(function* () {
222246 const requests = projection.runtimeRequests.filter(
223247 (request) => request.status === "pending" && request.responseCapability.type !== "message",
224248 );
249 // Delegated task rows, items and nodes stay open: the child settles them.
250 const delegatedTaskNodeIds = new Set<string>([
251 ...(projection.subagents ?? []).filter(isAppOwnedDelegation).map((subagent) => subagent.id),
252 ...(projection.turnItems ?? []).flatMap((item) =>
253 item.type === "subagent" && isAppOwnedDelegation(item) ? [item.subagentId] : [],
254 ),
255 ]);
225256 const detail = `Cancelled because the server ${trigger === "startup" ? "restarted" : "shut down"} before the provider work completed.`;
226257 const commandId = CommandId.make(
227258 `command:runtime-reconcile:${trigger}:${projection.thread.id}:${DateTime.formatIso(now)}`,
......@@ -333,6 +364,7 @@ export const make = Effect.gen(function* () {
333364 (candidate) =>
334365 candidate.runId === run.id &&
335366 !messageRequestNodeIds.has(candidate.id) &&
367 !delegatedTaskNodeIds.has(candidate.id) &&
336368 (candidate.status === "pending" ||
337369 candidate.status === "running" ||
338370 candidate.status === "waiting"),
......@@ -351,6 +383,7 @@ export const make = Effect.gen(function* () {
351383 for (const subagent of projection.subagents.filter(
352384 (candidate) =>
353385 candidate.runId === run.id &&
386 !isAppOwnedDelegation(candidate) &&
354387 (candidate.status === "pending" ||
355388 candidate.status === "running" ||
356389 candidate.status === "waiting"),
......@@ -404,6 +437,7 @@ export const make = Effect.gen(function* () {
404437 (candidate) =>
405438 candidate.runId === run.id &&
406439 (candidate.nodeId === null || !messageRequestNodeIds.has(candidate.nodeId)) &&
440 !isAppOwnedDelegationItem(candidate) &&
407441 (candidate.status === "pending" ||
408442 candidate.status === "running" ||
409443 candidate.status === "waiting"),
......@@ -436,7 +470,7 @@ export const make = Effect.gen(function* () {
436470 if (!isBackgroundCapableTurnItemType(item.type)) {
437471 continue;
438472 }
439 if (!isNonterminalTurnItemStatus(item.status)) {
473 if (!isNonterminalTurnItemStatus(item.status) || isAppOwnedDelegationItem(item)) {
440474 continue;
441475 }
442476 const providerInstanceId = resolveStaleBackgroundItemProviderInstanceId(item, projection);
......@@ -763,32 +797,41 @@ export const make = Effect.gen(function* () {
763797 if (!enabled) return;
764798 const threadIds = yield* projections.getRecoveryThreadIds("runtime");
765799 for (const threadId of threadIds) {
766 const projection = yield* projections.getRuntimeRecoveryProjection(threadId);
767 if (
768 !resolveProjectSettings(enabled, projection.thread.projectId).settings
769 .continueThreadsAfterServerUpdate
770 )
771 continue;
772 // Shutdown reconciliation cancels the background work below, so a
773 // settled thread's continuation must be captured while it is still open.
774 const run = restartContinuationRun(
775 projection,
776 providerThreadsWithOpenBackgroundWork(projection),
800 yield* Effect.gen(function* () {
801 const projection = yield* projections.getRuntimeRecoveryProjection(threadId);
802 if (
803 !resolveProjectSettings(enabled, projection.thread.projectId).settings
804 .continueThreadsAfterServerUpdate
805 )
806 return;
807 // Shutdown reconciliation cancels the background work below, so a
808 // settled thread's continuation must be captured while it is still open.
809 const run = restartContinuationRun(
810 projection,
811 providerThreadsWithOpenBackgroundWork(projection),
812 );
813 if (!run) return;
814 const commandId = CommandId.make(`command:restart-prepare:${run.id}`);
815 yield* eventSink.writeWithEffects({
816 commandId,
817 events: [],
818 effects: [
819 {
820 id: `effect:restart-continuation:${run.id}`,
821 commandId,
822 threadId,
823 request: { type: "provider-runtime.continue", sourceRunId: run.id },
824 },
825 ],
826 });
827 }).pipe(
828 // One failing thread must not cost the threads after it their continuation.
829 Effect.catchCauseIf(
830 (cause) => !Cause.hasInterruptsOnly(cause),
831 (cause) =>
832 Effect.logWarning("Failed to prepare a restart continuation", { threadId, cause }),
833 ),
777834 );
778 if (!run) continue;
779 const commandId = CommandId.make(`command:restart-prepare:${run.id}`);
780 yield* eventSink.writeWithEffects({
781 commandId,
782 events: [],
783 effects: [
784 {
785 id: `effect:restart-continuation:${run.id}`,
786 commandId,
787 threadId,
788 request: { type: "provider-runtime.continue", sourceRunId: run.id },
789 },
790 ],
791 });
792835 }
793836 }).pipe(
794837 Effect.mapError((cause) => new ProviderRuntimeRecoveryError({ operation: "reconcile", cause })),
apps/server/src/orchestration-v2/ProviderSessionManager.ts+22
......@@ -362,6 +362,25 @@ export const layerWithOptions = (
362362 },
363363 );
364364 const layerScope = yield* Effect.scope;
365 // Ctrl+C, or a stop that signals the whole process group, reaches the
366 // provider CLIs with the server. They report their own background work
367 // stopped before shutdown captures restart continuations, so provider
368 // events after the signal are dropped; restart recovery owns that state.
369 const shutdownSignal = { received: false };
370 const onShutdownSignal = () => {
371 shutdownSignal.received = true;
372 };
373 yield* Effect.acquireRelease(
374 Effect.sync(() => {
375 process.on("SIGINT", onShutdownSignal);
376 process.on("SIGTERM", onShutdownSignal);
377 }),
378 () =>
379 Effect.sync(() => {
380 process.off("SIGINT", onShutdownSignal);
381 process.off("SIGTERM", onShutdownSignal);
382 }),
383 );
365384 const sessions = yield* Ref.make(new Map<string, LiveSessionEntry>());
366385 // One retry per released entry, so a later release with the same id
367386 // cannot drop cleanup for threads only the earlier session served.
......@@ -1542,6 +1561,7 @@ export const layerWithOptions = (
15421561 let stoppedByProvider = false;
15431562 return entry.runtime.events.pipe(
15441563 Stream.runForEach((event) => {
1564 if (shutdownSignal.received) return Effect.void;
15451565 if (
15461566 event.type === "provider_session.updated" &&
15471567 event.providerSession.status === "stopped"
......@@ -1600,6 +1620,8 @@ export const layerWithOptions = (
16001620 Effect.exit,
16011621 Effect.flatMap((exit) =>
16021622 Effect.gen(function* () {
1623 // A provider that exits on the shutdown signal is released by shutdown.
1624 if (shutdownSignal.received) return;
16031625 const current = (yield* Ref.get(sessions)).get(
16041626 sessionKey(entry.runtime.providerSessionId),
16051627 );
apps/server/src/orchestration-v2/ProviderTurnStartService.ts+2-3
......@@ -953,6 +953,7 @@ export const layer: Layer.Layer<
953953 run,
954954 projection.runs,
955955 projection.providerTurns,
956 projection.attempts,
956957 );
957958 const restartCancelledWork = pendingRestartCancelledBackgroundWork({
958959 runs: projection.runs,
......@@ -967,9 +968,7 @@ export const layer: Layer.Layer<
967968 .map((candidate) => candidate.id),
968969 ),
969970 run,
970 runAttemptIds: projection.attempts
971 .filter((candidate) => candidate.runId === run.id)
972 .map((candidate) => candidate.id),
971 attempts: projection.attempts,
973972 });
974973 const restartNote =
975974 restartCancelledWork.length === 0
apps/server/src/orchestration-v2/RestartBackgroundNote.test.ts+106-2
......@@ -6,12 +6,15 @@ import {
66 RunId,
77 type OrchestrationV2Run,
88} from "@t3tools/contracts";
9import * as DateTime from "effect/DateTime";
910
1011import {
1112 cancelledRosterTaskWork,
1213 cancelledTurnItemWork,
14 isRestartNoteContinuation,
1315 pendingRestartCancelledBackgroundWork,
1416 restartCancelledBackgroundWorkNote,
17 restartContinuationNote,
1518 mergeRestartCancelledBackgroundWork,
1619} from "./RestartBackgroundNote.ts";
1720
......@@ -51,7 +54,7 @@ it("keeps the note for the provider thread that lost the work across a provider
5154 providerTurns: runs.filter((source) => source.id !== target.id).map(turnFor),
5255 compactionMessageIds: new Set(),
5356 run: target,
54 runAttemptIds: target.activeAttemptId === null ? [] : [target.activeAttemptId],
57 attempts: [{ id: target.activeAttemptId!, runId: target.id }],
5558 });
5659
5760 // Codex never lost the work, so it neither receives nor consumes the note.
......@@ -62,6 +65,30 @@ it("keeps the note for the provider thread that lost the work across a provider
6265 assert.deepEqual(pending(later, [root, onCodex, backOnClaude, later]), []);
6366});
6467
68it("delivers a resumed queued run's note even after a higher-ordinal run", () => {
69 // Run 3 ran ahead of held run 2; run 2 then resumed, lost background work in
70 // a second restart, and its continuation was superseded by a new message.
71 const ranFirst = run(3, claudeThread, {
72 completedAt: DateTime.makeUnsafe("2026-10-03T10:00:00.000Z"),
73 });
74 const resumed = run(2, claudeThread, {
75 completedAt: DateTime.makeUnsafe("2026-10-03T10:05:00.000Z"),
76 restartCancelledBackgroundWork: lost,
77 });
78 const next = run(4, claudeThread);
79 const runs = [resumed, ranFirst, next];
80 assert.deepEqual(
81 pendingRestartCancelledBackgroundWork({
82 runs,
83 providerTurns: [resumed, ranFirst].map(turnFor),
84 compactionMessageIds: new Set(),
85 run: next,
86 attempts: [{ id: next.activeAttemptId!, runId: next.id }],
87 }),
88 lost,
89 );
90});
91
6592it("bounds the note so it cannot crowd out the turn's context", () => {
6693 const work = Array.from({ length: 25 }, (_, index) => ({
6794 kind: "shell" as const,
......@@ -123,7 +150,7 @@ it("does not repeat the note when a steer restarts the run on a new attempt", ()
123150 ],
124151 compactionMessageIds: new Set(),
125152 run: steered,
126 runAttemptIds,
153 attempts: runAttemptIds.map((id) => ({ id, runId: steered.id })),
127154 });
128155
129156 // The first attempt reached the provider with the note; its replacement must not repeat it.
......@@ -131,3 +158,80 @@ it("does not repeat the note when a steer restarts the run on a new attempt", ()
131158 // A first attempt that never reached the provider did not deliver it.
132159 assert.deepEqual(pending([firstAttempt, RunAttemptId.make("attempt:2b")], false), lost);
133160});
161
162it("counts a prompted mid-turn or chained continuation as delivering the note", () => {
163 const cut = run(1, claudeThread, { status: "cancelled", restartCancelledBackgroundWork: lost });
164 const cutTurn = { ...turnFor(cut), status: "cancelled" as const };
165 const continuation = run(2, claudeThread, { restartContinuationOfRunId: cut.id });
166 // A turn cut mid-way that lost work is prompted with the note, not resumed natively.
167 assert.isTrue(isRestartNoteContinuation(continuation, [cut, continuation], [cutTurn], []));
168 assert.isFalse(
169 isRestartNoteContinuation(
170 continuation,
171 [{ ...cut, restartCancelledBackgroundWork: [] }, continuation],
172 [cutTurn],
173 [],
174 ),
175 );
176 const later = run(3, claudeThread);
177 const pending = (delivered: boolean) =>
178 pendingRestartCancelledBackgroundWork({
179 runs: [cut, continuation, later],
180 providerTurns: [cutTurn, ...(delivered ? [turnFor(continuation)] : [])],
181 compactionMessageIds: new Set(),
182 run: later,
183 attempts: [{ id: later.activeAttemptId!, runId: later.id }],
184 });
185 assert.deepEqual(pending(true), []);
186 // A continuation that never reached the provider delivered nothing.
187 assert.deepEqual(pending(false), lost);
188
189 // Its own continuation carries the original note forward.
190 const unstarted = { ...continuation, status: "cancelled" as const };
191 const chained = run(3, claudeThread, { restartContinuationOfRunId: unstarted.id });
192 assert.deepEqual(restartContinuationNote(unstarted, [cut, unstarted], [cutTurn], []), {
193 work: lost,
194 settled: false,
195 });
196 assert.isTrue(isRestartNoteContinuation(chained, [cut, unstarted, chained], [cutTurn], []));
197 // Claude announces running before accepting the prompt. A second restart
198 // can cancel that turn without delivering anything to the provider.
199 for (const status of ["running", "cancelled"] as const) {
200 const turns = [cutTurn, { ...turnFor(unstarted), status }];
201 assert.deepEqual(restartContinuationNote(unstarted, [cut, unstarted], turns, []), {
202 work: lost,
203 settled: false,
204 });
205 assert.deepEqual(
206 pendingRestartCancelledBackgroundWork({
207 runs: [cut, unstarted, later],
208 providerTurns: turns,
209 compactionMessageIds: new Set(),
210 run: later,
211 attempts: [{ id: later.activeAttemptId!, runId: later.id }],
212 }),
213 lost,
214 );
215 }
216
217 // A steer replaced the attempt after its completed turn delivered the note.
218 const steered = { ...unstarted, activeAttemptId: RunAttemptId.make("attempt:replacement") };
219 const attempts = [{ id: unstarted.activeAttemptId!, runId: steered.id }];
220 const turns = [
221 cutTurn,
222 turnFor(unstarted),
223 { ...turnFor(steered), status: "cancelled" as const },
224 ];
225 assert.deepEqual(restartContinuationNote(steered, [cut, steered], turns, attempts).work, []);
226 assert.isFalse(isRestartNoteContinuation(chained, [cut, steered, chained], turns, attempts));
227 assert.deepEqual(
228 pendingRestartCancelledBackgroundWork({
229 runs: [cut, steered, later],
230 providerTurns: turns,
231 compactionMessageIds: new Set(),
232 run: later,
233 attempts,
234 }),
235 [],
236 );
237});
apps/server/src/orchestration-v2/RestartBackgroundNote.ts+67-13
......@@ -3,10 +3,13 @@ import type {
33 OrchestrationV2ProviderTurn,
44 OrchestrationV2RestartCancelledBackgroundWork,
55 OrchestrationV2Run,
6 OrchestrationV2RunAttempt,
67 OrchestrationV2TurnItem,
78} from "@t3tools/contracts";
9import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
810
911type Work = OrchestrationV2RestartCancelledBackgroundWork;
12type Attempt = Pick<OrchestrationV2RunAttempt, "id" | "runId">;
1013
1114const MAX_LABEL_LENGTH = 160;
1215
......@@ -120,25 +123,69 @@ export function isRestartNoteSource(
120123 );
121124}
122125
123/** A restart continuation whose prompt is the note rather than a resume. */
126/**
127 * The note a restart continuation of `source` carries, and whether the turn it
128 * continues had settled (the note is then the whole prompt). A continuation cut
129 * before its provider turn completed may not have delivered its own note:
130 * adapters can announce a running turn before accepting the prompt. Carry the
131 * note forward until a completed turn proves delivery.
132 */
133export function restartContinuationNote(
134 source: OrchestrationV2Run,
135 runs: ReadonlyArray<OrchestrationV2Run>,
136 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,
137 attempts: ReadonlyArray<Attempt>,
138): { readonly work: ReadonlyArray<Work>; readonly settled: boolean } {
139 const completedAttempts = new Set(
140 providerTurns.filter((turn) => turn.status === "completed").map((turn) => turn.runAttemptId),
141 );
142 const completedRuns = new Set(
143 attempts.filter((attempt) => completedAttempts.has(attempt.id)).map((attempt) => attempt.runId),
144 );
145 let current = source;
146 let work = current.restartCancelledBackgroundWork ?? [];
147 const visited = new Set([current.id]);
148 while (
149 current.restartContinuationOfRunId !== undefined &&
150 !completedRuns.has(current.id) &&
151 !(current.activeAttemptId !== null && completedAttempts.has(current.activeAttemptId))
152 ) {
153 const previous = runs.find((candidate) => candidate.id === current.restartContinuationOfRunId);
154 if (previous === undefined || visited.has(previous.id)) break;
155 visited.add(previous.id);
156 current = previous;
157 work = mergeRestartCancelledBackgroundWork(current.restartCancelledBackgroundWork ?? [], work);
158 }
159 return { work, settled: isRestartNoteSource(current, providerTurns) };
160}
161
162/**
163 * A restart continuation prompted with the note rather than a native resume.
164 * A turn cut mid-way that lost background work is prompted too, so the note
165 * is delivered with it; only a continuation without a note resumes natively.
166 */
124167export function isRestartNoteContinuation(
125168 run: Pick<OrchestrationV2Run, "restartContinuationOfRunId">,
126169 runs: ReadonlyArray<OrchestrationV2Run>,
127170 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,
171 attempts: ReadonlyArray<Attempt>,
128172): boolean {
129173 const source =
130174 run.restartContinuationOfRunId === undefined
131175 ? undefined
132176 : runs.find((candidate) => candidate.id === run.restartContinuationOfRunId);
133 return source !== undefined && isRestartNoteSource(source, providerTurns);
177 return (
178 source !== undefined &&
179 restartContinuationNote(source, runs, providerTurns, attempts).work.length > 0
180 );
134181}
135182
136183/**
137184 * Work cancelled by a restart that the run's provider thread has not been told
138185 * about yet. The note belongs to the provider thread that lost the work: turns
139186 * on another provider (after a switch) neither owe it nor deliver it. A later
140 * run on the same provider thread delivers it once its attempt reaches the
141 * provider, so the pending set is derived rather than cleared. Compactions and
187 * completed turn on the same provider thread proves delivery, so the pending
188 * set is derived rather than cleared. Compactions and
142189 * resumed turns carry no note, and a rolled-back run left native history, so
143190 * none of them counts as delivery. A note continuation's own prompt is the note.
144191 */
......@@ -150,13 +197,14 @@ export function pendingRestartCancelledBackgroundWork(input: {
150197 OrchestrationV2Run,
151198 | "id"
152199 | "ordinal"
200 | "completedAt"
153201 | "userMessageId"
154202 | "providerThreadId"
155203 | "restartContinuationOfRunId"
156204 | "activeAttemptId"
157205 >;
158 /** Every attempt id of `run`; a steer replaces the attempt but not the run. */
159 readonly runAttemptIds: ReadonlyArray<OrchestrationV2Run["activeAttemptId"] & string>;
206 /** Include earlier attempts: a steer replaces the attempt but not the run. */
207 readonly attempts: ReadonlyArray<Attempt>;
160208}): ReadonlyArray<Work> {
161209 const isCompaction = (run: typeof input.run) => input.compactionMessageIds.has(run.userMessageId);
162210 // The current run prepends the note unless it is a compaction or a
......@@ -170,7 +218,7 @@ export function pendingRestartCancelledBackgroundWork(input: {
170218 const providerThreadId = input.run.providerThreadId;
171219 const deliveredAttemptIds = new Set(
172220 input.providerTurns
173 .filter((turn) => turn.providerThreadId === providerThreadId)
221 .filter((turn) => turn.providerThreadId === providerThreadId && turn.status === "completed")
174222 .map((turn) => turn.runAttemptId),
175223 );
176224 const sameThread = input.runs.filter((run) => run.providerThreadId === providerThreadId);
......@@ -179,23 +227,29 @@ export function pendingRestartCancelledBackgroundWork(input: {
179227 candidate.id !== input.run.id &&
180228 candidate.activeAttemptId !== null &&
181229 candidate.status !== "rolled_back" &&
182 deliveredAttemptIds.has(candidate.activeAttemptId) &&
230 (deliveredAttemptIds.has(candidate.activeAttemptId) ||
231 input.attempts.some(
232 (attempt) => attempt.runId === candidate.id && deliveredAttemptIds.has(attempt.id),
233 )) &&
183234 !isCompaction(candidate) &&
184235 (candidate.restartContinuationOfRunId === undefined ||
185 isRestartNoteContinuation(candidate, input.runs, input.providerTurns)),
236 isRestartNoteContinuation(candidate, input.runs, input.providerTurns, input.attempts)),
186237 );
187238 // A steer restarts this run on a new attempt: an earlier attempt that already
188239 // reached the provider delivered the note, so the replacement must not repeat it.
189 const alreadyDelivered = input.runAttemptIds.some(
190 (attemptId) => attemptId !== input.run.activeAttemptId && deliveredAttemptIds.has(attemptId),
240 const alreadyDelivered = input.attempts.some(
241 (attempt) =>
242 attempt.runId === input.run.id &&
243 attempt.id !== input.run.activeAttemptId &&
244 deliveredAttemptIds.has(attempt.id),
191245 );
192246 if (alreadyDelivered) return [];
193247 return sameThread
194248 .filter(
195249 (source) =>
196 source.ordinal < input.run.ordinal &&
250 runRanAfter(input.run, source) &&
197251 (source.restartCancelledBackgroundWork?.length ?? 0) > 0 &&
198 !prompted.some((later) => later.ordinal > source.ordinal),
252 !prompted.some((later) => runRanAfter(later, source)),
199253 )
200254 .reduce<ReadonlyArray<Work>>(
201255 (work, source) =>
apps/server/src/orchestration-v2/RestartContinuation.test.ts+217
......@@ -1,5 +1,6 @@
11import { assert, it } from "@effect/vitest";
22import {
3 MessageId,
34 ProjectId,
45 ProviderDriverKind,
56 ProviderInstanceId,
......@@ -11,6 +12,7 @@ import {
1112 ThreadId,
1213 type OrchestrationV2ThreadProjection,
1314} from "@t3tools/contracts";
15import * as DateTime from "effect/DateTime";
1416import * as Effect from "effect/Effect";
1517import * as Layer from "effect/Layer";
1618import * as ServerSettings from "../serverSettings.ts";
......@@ -201,6 +203,7 @@ it.effect("prompts a settled thread's continuation with the note of its lost wor
201203 Layer.merge(
202204 Layer.mock(ThreadManagementService.ThreadManagementService)({
203205 getThreadRecords: () => Effect.succeed(projection),
206 recoverDelegatedTask: () => Effect.void,
204207 dispatch: (command) => {
205208 commands.push(command);
206209 return Effect.succeed({} as never);
......@@ -246,6 +249,7 @@ it.effect("does not continue a failed run that lost background work", () =>
246249 Layer.merge(
247250 Layer.mock(ThreadManagementService.ThreadManagementService)({
248251 getThreadRecords: () => Effect.succeed(projection),
252 recoverDelegatedTask: () => Effect.void,
249253 dispatch: (command) => {
250254 commands.push(command);
251255 return Effect.succeed({} as never);
......@@ -328,6 +332,7 @@ it.effect("does not duplicate delivery and yields to newer user work or opt-out"
328332 >[0][] = [];
329333 const threads = Layer.mock(ThreadManagementService.ThreadManagementService)({
330334 getThreadRecords: () => Effect.succeed(projection),
335 recoverDelegatedTask: () => Effect.void,
331336 dispatch: (command) => {
332337 commands.push(command);
333338 if (command.type === "message.dispatch")
......@@ -499,6 +504,7 @@ it.effect("does not cancel or resume a run that completes while shutdown intent
499504 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
500505 Layer.mock(ThreadManagementService.ThreadManagementService)({
501506 getThreadRecords: () => Effect.succeed(projection),
507 recoverDelegatedTask: () => Effect.void,
502508 dispatch: () =>
503509 Effect.sync(() => {
504510 dispatched = true;
......@@ -511,3 +517,214 @@ it.effect("does not cancel or resume a run that completes while shutdown intent
511517 assert.isFalse(dispatched);
512518 }),
513519);
520
521const continuationTexts = (projection: OrchestrationV2ThreadProjection) =>
522 Effect.gen(function* () {
523 const texts: Array<string> = [];
524 yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
525 Effect.provide(
526 Layer.merge(
527 Layer.mock(ThreadManagementService.ThreadManagementService)({
528 getThreadRecords: () => Effect.succeed(projection),
529 recoverDelegatedTask: () => Effect.void,
530 dispatch: (command) => {
531 if (command.type === "message.dispatch") texts.push(command.text);
532 return Effect.succeed({} as never);
533 },
534 }),
535 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
536 ),
537 ),
538 );
539 return texts;
540 });
541
542const cutMidTurn = (extra: Record<string, unknown> = {}) => {
543 const base = makeProjection();
544 return {
545 ...base,
546 runs: [
547 {
548 ...base.runs[0]!,
549 status: "cancelled",
550 userMessageId: MessageId.make("message:user"),
551 ...extra,
552 },
553 ],
554 providerTurns: [{ ...base.providerTurns[0]!, status: "cancelled" }],
555 } as unknown as OrchestrationV2ThreadProjection;
556};
557
558const queuedFollowUp = {
559 id: RunId.make("run:queued"),
560 ordinal: 2,
561 providerInstanceId: instanceId,
562 providerThreadId,
563 status: "queued",
564 queueHeld: true,
565};
566
567it.effect("continues a cut run past queued follow-ups, which stay held", () =>
568 Effect.gen(function* () {
569 const live = makeProjection();
570 assert.equal(
571 restartContinuationRun({
572 ...live,
573 runs: [...live.runs, queuedFollowUp],
574 } as unknown as OrchestrationV2ThreadProjection)?.id,
575 runId,
576 );
577 const cut = cutMidTurn();
578 const texts = yield* continuationTexts({
579 ...cut,
580 runs: [...cut.runs, queuedFollowUp],
581 } as unknown as OrchestrationV2ThreadProjection);
582 assert.deepEqual(texts, ["Continue where you left off."]);
583 }),
584);
585
586it.effect("continues a resumed queued run that ran after an earlier continuation", () =>
587 Effect.gen(function* () {
588 // The continuation (ordinal 3) ran ahead of the held queue, then the user
589 // resumed the queued run (ordinal 1 here) and the server restarted again.
590 const finishedContinuation = {
591 id: RunId.make("run:earlier-continuation"),
592 ordinal: 3,
593 providerInstanceId: instanceId,
594 providerThreadId,
595 status: "completed",
596 completedAt: DateTime.makeUnsafe("2026-10-03T10:00:00.000Z"),
597 };
598 const live = makeProjection();
599 assert.equal(
600 restartContinuationRun({
601 ...live,
602 runs: [...live.runs, finishedContinuation],
603 } as unknown as OrchestrationV2ThreadProjection)?.id,
604 runId,
605 );
606 const cut = cutMidTurn({ completedAt: DateTime.makeUnsafe("2026-10-03T10:05:00.000Z") });
607 const texts = yield* continuationTexts({
608 ...cut,
609 runs: [...cut.runs, finishedContinuation],
610 } as unknown as OrchestrationV2ThreadProjection);
611 assert.deepEqual(texts, ["Continue where you left off."]);
612 }),
613);
614
615it.effect("does not continue a run the user asked to stop before the restart", () =>
616 Effect.gen(function* () {
617 const texts = yield* continuationTexts({
618 ...cutMidTurn(),
619 turnItems: [{ id: "turn-item:interrupt", runId, type: "run_interrupt_request" }],
620 } as unknown as OrchestrationV2ThreadProjection);
621 assert.deepEqual(texts, []);
622 }),
623);
624
625it.effect("does not continue a cut /compact or /logout turn", () =>
626 Effect.gen(function* () {
627 for (const text of ["/compact", " /LOGOUT "]) {
628 const texts = yield* continuationTexts({
629 ...cutMidTurn(),
630 messages: [{ id: MessageId.make("message:user"), text, attachments: [] }],
631 } as unknown as OrchestrationV2ThreadProjection);
632 assert.deepEqual(texts, [], text);
633 }
634 const texts = yield* continuationTexts({
635 ...cutMidTurn(),
636 messages: [{ id: MessageId.make("message:user"), text: "/compact later", attachments: [] }],
637 } as unknown as OrchestrationV2ThreadProjection);
638 assert.lengthOf(texts, 1);
639 }),
640);
641
642it.effect("tells a turn cut mid-way about the background work it lost", () =>
643 Effect.gen(function* () {
644 const texts = yield* continuationTexts(
645 cutMidTurn({
646 restartCancelledBackgroundWork: [{ kind: "subagent", label: "Background reviewer" }],
647 }),
648 );
649 assert.lengthOf(texts, 1);
650 assert.include(texts[0]!, "Background reviewer");
651 assert.isTrue(texts[0]!.endsWith("Continue where you left off."));
652 }),
653);
654
655it.effect(
656 "carries the note forward when its continuation was cut before reaching the provider",
657 () =>
658 Effect.gen(function* () {
659 const base = makeProjection();
660 const original = {
661 ...base.runs[0]!,
662 id: RunId.make("run:original"),
663 status: "completed",
664 activeAttemptId: RunAttemptId.make("attempt:original"),
665 restartCancelledBackgroundWork: [{ kind: "shell", label: "sleep 25 && echo DONE" }],
666 };
667 const projection = {
668 ...base,
669 runs: [
670 original,
671 {
672 ...base.runs[0]!,
673 ordinal: 2,
674 status: "cancelled",
675 restartContinuationOfRunId: original.id,
676 },
677 ],
678 // The original turn settled; the continuation's attempt never started one.
679 providerTurns: [
680 {
681 ...base.providerTurns[0]!,
682 runAttemptId: original.activeAttemptId,
683 status: "completed",
684 },
685 ],
686 } as unknown as OrchestrationV2ThreadProjection;
687 const texts = yield* continuationTexts(projection);
688 assert.lengthOf(texts, 1);
689 assert.include(texts[0]!, "sleep 25 && echo DONE");
690 assert.notInclude(texts[0]!, "Continue where");
691 }),
692);
693
694it.effect("prepares later threads' continuations when one thread fails", () =>
695 Effect.gen(function* () {
696 const brokenThreadId = ThreadId.make("thread:broken");
697 const writes: Parameters<EventSink.EventSinkV2["Service"]["writeWithEffects"]>[0][] = [];
698 const recovery = yield* ProviderRuntimeRecovery.make.pipe(
699 Effect.provide(
700 Layer.mergeAll(
701 ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
702 Layer.mock(ProjectionStore.ProjectionStoreV2)({
703 getRecoveryThreadIds: () => Effect.succeed([brokenThreadId, threadId]),
704 getRuntimeRecoveryProjection: (id) =>
705 id === brokenThreadId
706 ? Effect.fail(
707 new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId: id }),
708 )
709 : Effect.succeed(makeProjection()),
710 }),
711 Layer.mock(EventSink.EventSinkV2)({
712 writeWithEffects: (input) =>
713 Effect.sync(() => {
714 writes.push(input);
715 return [];
716 }),
717 }),
718 IdAllocator.layer,
719 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}),
720 Layer.mock(EffectOutbox.EffectOutboxV2)({}),
721 ),
722 ),
723 );
724 yield* recovery.prepareForShutdown;
725 assert.deepEqual(
726 writes.map((write) => write.effects[0]?.request),
727 [{ type: "provider-runtime.continue", sourceRunId: runId }],
728 );
729 }),
730);
apps/server/src/orchestration-v2/RestartContinuation.ts+64-7
......@@ -1,3 +1,4 @@
1import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
12import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
23import {
34 CommandId,
......@@ -11,12 +12,16 @@ import * as Effect from "effect/Effect";
1112import type { ProjectionRuntimeRecoveryState } from "./ProjectionStore.ts";
1213
1314import * as ServerSettings from "../serverSettings.ts";
15import { isNativeMaintenanceCommand } from "./Orchestrator.ts";
1416import * as ThreadManagementService from "./ThreadManagementService.ts";
1517import {
1618 isRestartNoteSource,
1719 restartCancelledBackgroundWorkNote,
20 restartContinuationNote,
1821} from "./RestartBackgroundNote.ts";
1922
23const CONTINUE_PROMPT = "Continue where you left off.";
24
2025/**
2126 * The run a restart continuation resumes, if any: an unfinished root run, or a
2227 * settled one whose own provider thread lost background work in the restart
......@@ -30,8 +35,12 @@ export function restartContinuationRun(
3035 cancelledWorkProviderThreadIds: ReadonlySet<ProviderThreadId> = new Set(),
3136): OrchestrationV2Run | undefined {
3237 if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) return;
38 // Queued runs never started; recovery holds them behind the cut run.
3339 const run = projection.runs.reduce<OrchestrationV2Run | undefined>(
34 (latest, candidate) => (!latest || candidate.ordinal > latest.ordinal ? candidate : latest),
40 (latest, candidate) =>
41 candidate.status !== "queued" && (!latest || runRanAfter(candidate, latest))
42 ? candidate
43 : latest,
3544 undefined,
3645 );
3746 if (!run) return;
......@@ -98,7 +107,7 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta
98107 const messageId = MessageId.make(`message:restart-continuation:${input.sourceRunId}`);
99108 const projection = yield* threads.getThreadRecords(
100109 input.threadId,
101 ["messages", "runs", "providerTurns"],
110 ["messages", "runs", "providerTurns", "attempts"],
102111 { messageIds: [messageId] },
103112 );
104113 if (
......@@ -114,17 +123,54 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta
114123 const noteSource =
115124 source !== undefined && isRestartNoteSource(source, projection.providerTurns);
116125 if (!source || (source.status !== "cancelled" && !noteSource)) return;
117 // A user submission after reconciliation takes precedence over an automatic prompt.
118 if (projection.runs.some((run) => run.ordinal > source.ordinal)) return;
126 // A user submission after reconciliation takes precedence over an automatic
127 // prompt. Queued runs never started and stay held behind this one.
128 if (
129 projection.runs.some(
130 (run) => run.id !== source.id && run.status !== "queued" && runRanAfter(run, source),
131 )
132 )
133 return;
119134 if (projection.thread.providerInstanceId !== source.providerInstanceId) return;
135 const sourceRecords = yield* threads.getThreadRecords(
136 input.threadId,
137 ["messages", "turnItems"],
138 {
139 messageIds: [source.userMessageId],
140 turnItemRunIds: [source.id],
141 turnItemTypes: ["run_interrupt_request"],
142 },
143 );
144 // The user asked this run to stop before the restart cut it.
145 if (
146 sourceRecords.turnItems.some(
147 (item) => item.runId === source.id && item.type === "run_interrupt_request",
148 )
149 )
150 return;
151 const sourceMessage = sourceRecords.messages.find(
152 (message) => message.id === source.userMessageId,
153 );
154 if (sourceMessage !== undefined && isNativeMaintenanceCommand(sourceMessage)) return;
155 const note = restartContinuationNote(
156 source,
157 projection.runs,
158 projection.providerTurns,
159 projection.attempts,
160 );
161 const noteText =
162 note.work.length === 0 ? undefined : restartCancelledBackgroundWorkNote(note.work);
120163 yield* threads.dispatch({
121164 type: "message.dispatch",
122165 commandId: CommandId.make(`command:restart-continuation:${input.sourceRunId}`),
123166 threadId: input.threadId,
124167 messageId,
125 text: noteSource
126 ? restartCancelledBackgroundWorkNote(source.restartCancelledBackgroundWork ?? [])
127 : "Continue where you left off.",
168 text:
169 noteText === undefined
170 ? CONTINUE_PROMPT
171 : note.settled
172 ? noteText
173 : `${noteText}\n\n${CONTINUE_PROMPT}`,
128174 attachments: [],
129175 modelSelection: source.modelSelection,
130176 dispatchMode: { type: "start_immediately" },
......@@ -133,4 +179,15 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta
133179 restartContinuationOfRunId: input.sourceRunId,
134180 });
135181 },
182 // A delegated child this declined to continue still owes its parent a
183 // result. Once a continuation run exists this is a no-op; that run settles it.
184 (effect, input) =>
185 effect.pipe(
186 Effect.andThen(
187 Effect.gen(function* () {
188 const threads = yield* ThreadManagementService.ThreadManagementService;
189 yield* threads.recoverDelegatedTask(input.threadId, input.sourceRunId);
190 }),
191 ),
192 ),
136193);
apps/server/src/orchestration-v2/SubagentProjection.test.ts+22
......@@ -166,6 +166,28 @@ function taskFixture() {
166166 return { projection: { ...projection, runs: [run] }, run };
167167}
168168
169it("reports the run that ended last, not the highest ordinal", () => {
170 const { projection, run } = taskFixture();
171 // A restart continuation (ordinal 4) ran ahead of held queued runs 2 and 3.
172 const ended = (ordinal: number, completedAt: string): OrchestrationV2Run => ({
173 ...run,
174 id: RunId.make(`run:${ordinal}`),
175 ordinal,
176 completedAt: DateTime.makeUnsafe(completedAt),
177 });
178 const progress = delegatedTaskProgress({
179 ...projection,
180 runs: [
181 { ...run, status: "cancelled" },
182 ended(4, "2026-07-24T10:00:00.000Z"),
183 ended(2, "2026-07-24T10:05:00.000Z"),
184 ended(3, "2026-07-24T10:10:00.000Z"),
185 ],
186 });
187 assert.equal(progress.state, "result_available");
188 assert.equal(progress.resultRun?.ordinal, 3);
189});
190
169191it("waits for nested work and retains the report across monitor acknowledgements", () => {
170192 const { projection, run } = taskFixture();
171193 assert.equal(
apps/server/src/orchestration-v2/SubagentProjection.ts+2-1
......@@ -16,6 +16,7 @@ import type {
1616 ThreadId,
1717 TurnItemId,
1818} from "@t3tools/contracts";
19import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
1920import * as DateTime from "effect/DateTime";
2021import { isOrchestrationV2WorkActive } from "@t3tools/contracts";
2122
......@@ -239,7 +240,7 @@ export function delegatedTaskProgress(projection: {
239240 projection.providerThreads.some((thread) => (thread.pendingBackgroundTasks?.length ?? 0) > 0);
240241 const resultRun = workRuns
241242 .filter((run) => terminal(run.status) && (run.startedAt !== null || run.ordinal === 1))
242 .toSorted((a, b) => b.ordinal - a.ordinal)[0];
243 .toSorted((a, b) => (runRanAfter(a, b) ? -1 : runRanAfter(b, a) ? 1 : 0))[0];
243244 return {
244245 state:
245246 active || resultRun === undefined
apps/server/src/orchestration-v2/ThreadManagementService.ts+4
......@@ -313,6 +313,8 @@ export interface ThreadManagementServiceShape {
313313 input: ThreadManagementInterruptInput,
314314 ) => Effect.Effect<ThreadManagementInterruptResult, ThreadManagementFailure>;
315315 readonly getThreadEventSequence: Orchestrator.OrchestratorV2["Service"]["getThreadEventSequence"];
316 readonly recoverDelegatedTask: Orchestrator.OrchestratorV2["Service"]["recoverDelegatedTask"];
317 readonly delegatedTaskResultPending: Orchestrator.OrchestratorV2["Service"]["delegatedTaskResultPending"];
316318 readonly streamStoredEvents: Orchestrator.OrchestratorV2["Service"]["streamStoredEvents"];
317319 readonly streamStoredEventsFrom: Orchestrator.OrchestratorV2["Service"]["streamStoredEventsFrom"];
318320 readonly streamDomainEvents: Orchestrator.OrchestratorV2["Service"]["streamDomainEvents"];
......@@ -736,6 +738,8 @@ const make = Effect.gen(function* () {
736738 waitForThread,
737739 interruptThread,
738740 getThreadEventSequence: orchestrator.getThreadEventSequence,
741 recoverDelegatedTask: orchestrator.recoverDelegatedTask,
742 delegatedTaskResultPending: orchestrator.delegatedTaskResultPending,
739743 streamStoredEvents: orchestrator.streamStoredEvents,
740744 streamStoredEventsFrom: orchestrator.streamStoredEventsFrom,
741745 streamDomainEvents: orchestrator.streamDomainEvents,
apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts+4
......@@ -433,6 +433,8 @@ export function makeOrchestratorV2ReplayLayerWithRegistry<Error>(
433433 dispatch: orchestrator.dispatch,
434434 getThreadRecords: orchestrator.getThreadRecords,
435435 getThreadProjection: orchestrator.getThreadProjection,
436 recoverDelegatedTask: orchestrator.recoverDelegatedTask,
437 delegatedTaskResultPending: orchestrator.delegatedTaskResultPending,
436438 });
437439 }),
438440 ).pipe(Layer.provide(orchestratorProvided));
......@@ -497,6 +499,8 @@ export function makeOrchestratorV2ReplayLayerWithRegistry<Error>(
497499 Orchestrator.OrchestratorV2,
498500 Effect.gen(function* () {
499501 const orchestrator = yield* Orchestrator.OrchestratorV2;
502 // As in serverRuntimeStartup: after runtime recovery, before the worker.
503 yield* orchestrator.recoverDelegatedTasks;
500504 yield* EffectWorker.runDaemon.pipe(Effect.forkScoped);
501505 return orchestrator;
502506 }),
apps/server/src/relay/AgentAwarenessRelay.test.ts+2
......@@ -206,6 +206,8 @@ const makeTestRelay = Effect.fnUntraced(function* (
206206 waitForThread: unused,
207207 interruptThread: unused,
208208 getThreadEventSequence: unused,
209 recoverDelegatedTask: unused,
210 delegatedTaskResultPending: unused,
209211 streamStoredEvents: Stream.empty,
210212 streamStoredEventsFrom: () => Stream.empty,
211213 streamDomainEvents: options.domainEvents ?? Stream.empty,
apps/server/src/scheduling/Scheduler.test.ts+24
......@@ -6,6 +6,7 @@ import * as Queue from "effect/Queue";
66import * as Ref from "effect/Ref";
77import * as TestClock from "effect/testing/TestClock";
88
9import * as ServerActivation from "../serverActivation.ts";
910import * as Scheduler from "./Scheduler.ts";
1011
1112it.effect("keeps other sources running after a source defects", () =>
......@@ -84,3 +85,26 @@ it.effect("unregisters closed sources and interrupts their in-flight work", () =
8485 assert.equal(yield* Ref.get(runs), 1);
8586 }).pipe(Effect.provide(Scheduler.layer)),
8687);
88
89it.effect("holds due work and the clock until startup recovery activates the server", () =>
90 Effect.gen(function* () {
91 const activation = yield* Deferred.make<void>();
92 const runs = yield* Ref.make(0);
93 const ran = yield* Deferred.make<void>();
94 yield* Effect.gen(function* () {
95 const scheduler = yield* Scheduler.Scheduler;
96 yield* scheduler.register(
97 "due",
98 Ref.update(runs, (n) => n + 1).pipe(Effect.andThen(Deferred.succeed(ran, undefined))),
99 );
100 yield* TestClock.adjust("10 seconds");
101 assert.equal(yield* Ref.get(runs), 0);
102 yield* Deferred.succeed(activation, undefined);
103 yield* Deferred.await(ran);
104 }).pipe(
105 Effect.provide(Scheduler.layer),
106 Effect.provideService(ServerActivation.ServerActivation, Deferred.await(activation)),
107 );
108 assert.isAtLeast(yield* Ref.get(runs), 1);
109 }),
110);
apps/server/src/scheduling/Scheduler.ts+5-2
......@@ -6,6 +6,8 @@ import * as Ref from "effect/Ref";
66import * as Scope from "effect/Scope";
77import * as Semaphore from "effect/Semaphore";
88
9import { forkParked } from "../serverActivation.ts";
10
911/** Sources load due work from their durable state; registration owns its execution lifetime. */
1012export class Scheduler extends Context.Service<
1113 Scheduler,
......@@ -47,14 +49,15 @@ const make = Effect.gen(function* () {
4749 return next;
4850 }),
4951 );
50 yield* run;
52 // Due work waits for startup recovery, which would cancel runs it started.
53 yield* forkParked(run);
5154 });
5255 const tick = Ref.get(sources).pipe(
5356 Effect.flatMap((current) => Effect.forEach(current.values(), (run) => run, { discard: true })),
5457 );
5558 // One clock for all due-work sources. A slow source cannot block another
5659 // source or overlap itself, and no extra missed-tick backlog is queued.
57 yield* Effect.sleep("5 seconds").pipe(Effect.andThen(tick), Effect.forever, Effect.forkScoped);
60 yield* forkParked(Effect.sleep("5 seconds").pipe(Effect.andThen(tick), Effect.forever));
5861 return Scheduler.of({ register });
5962});
6063
apps/server/src/serverRuntimeStartup.test.ts+10-1
......@@ -31,11 +31,20 @@ it.effect("starts without scanning or rebuilding projection history", () =>
3131 const result = yield* ServerRuntimeStartup.runOrderedV2StartupPhases({
3232 importLegacyShells: record("import"),
3333 recover: record("recover").pipe(Effect.as({ closedRequests: 2 })),
34 recoverDelegatedTasks: record("delegated"),
3435 startEffectWorker: record("worker"),
3536 autoBootstrap: record("bootstrap").pipe(Effect.as({ projectId: "project-1" })),
3637 });
3738
38 assert.deepEqual(yield* Ref.get(calls), ["import", "recover", "worker", "bootstrap"]);
39 // Delegated recovery reads the runs recovery terminalizes, and settles them
40 // before the worker runs restart continuations that would otherwise race it.
41 assert.deepEqual(yield* Ref.get(calls), [
42 "import",
43 "recover",
44 "delegated",
45 "worker",
46 "bootstrap",
47 ]);
3948 assert.deepEqual(result, {
4049 recovery: { closedRequests: 2 },
4150 bootstrap: { projectId: "project-1" },
apps/server/src/serverRuntimeStartup.ts+11
......@@ -32,6 +32,7 @@ import * as Keybindings from "./keybindings.ts";
3232import * as ExternalLauncher from "./process/externalLauncher.ts";
3333import * as EffectWorker from "./orchestration-v2/EffectWorker.ts";
3434import * as LegacyV1ThreadImporter from "./orchestration-v2/legacy/LegacyV1ThreadImporter.ts";
35import * as Orchestrator from "./orchestration-v2/Orchestrator.ts";
3536import * as ProviderRuntimeRecovery from "./orchestration-v2/ProviderRuntimeRecoveryService.ts";
3637import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts";
3738import * as ThreadLaunch from "./orchestration-v2/ThreadLaunchService.ts";
......@@ -386,21 +387,26 @@ export function runOrderedV2StartupPhases<
386387 Bootstrap,
387388 ImportError,
388389 RecoveryError,
390 DelegationError,
389391 WorkerError,
390392 BootstrapError,
391393 ImportContext,
392394 RecoveryContext,
395 DelegationContext,
393396 WorkerContext,
394397 BootstrapContext,
395398>(input: {
396399 readonly importLegacyShells: Effect.Effect<Import, ImportError, ImportContext>;
397400 readonly recover: Effect.Effect<Recovery, RecoveryError, RecoveryContext>;
401 /** Settles delegated tasks whose runs recovery just terminalized. */
402 readonly recoverDelegatedTasks: Effect.Effect<void, DelegationError, DelegationContext>;
398403 readonly startEffectWorker: Effect.Effect<void, WorkerError, WorkerContext>;
399404 readonly autoBootstrap: Effect.Effect<Bootstrap, BootstrapError, BootstrapContext>;
400405}) {
401406 return Effect.gen(function* () {
402407 yield* input.importLegacyShells;
403408 const recovery = yield* input.recover;
409 yield* input.recoverDelegatedTasks;
404410 yield* input.startEffectWorker;
405411 const bootstrap = yield* input.autoBootstrap;
406412 return { recovery, bootstrap } as const;
......@@ -413,6 +419,7 @@ const make = (options?: StartupOptions) =>
413419 const keybindings = yield* Keybindings.Keybindings;
414420 const legacyV1ThreadImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
415421 const providerRuntimeRecovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;
422 const orchestrator = yield* Orchestrator.OrchestratorV2;
416423 const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
417424 const agentAwarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay;
418425 const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents;
......@@ -509,6 +516,10 @@ const make = (options?: StartupOptions) =>
509516 ),
510517 ),
511518 recover: runStartupPhase("orchestration-v2.recovery", providerRuntimeRecovery.recover),
519 recoverDelegatedTasks: runStartupPhase(
520 "orchestration-v2.delegated-tasks.recover",
521 orchestrator.recoverDelegatedTasks,
522 ),
512523 startEffectWorker: runStartupPhase(
513524 "orchestration-v2.effect-worker.start",
514525 startEffectWorkerWithRelay({
docs/internals/server-updates.md+16-5
......@@ -61,7 +61,9 @@ stopped backends and replays the failure for the same token.
6161Restart continuation is an environment-owned preference, off by default. The
6262[v2 recovery service](../../apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts)
6363requires matching durable run, provider thread, session, and native resume identity.
64Ordinary queued work, finished runs, and background-only work do not qualify.
64Queued runs never started, so recovery holds them and continues the run they wait
65behind. A finished run qualifies only when the restart cancelled its background work;
66its continuation tells the provider what will not report back.
6567
6668Recovery retires effects tied to the lost process and records continuation intent
6769in the durable outbox. That intent survives another restart before provider startup.
......@@ -71,7 +73,16 @@ closing providers, then reconciles after ingestion has stopped so a late complet
7173cannot be overwritten by a stale cancellation.
7274
7375The [continuation handler](../../apps/server/src/orchestration-v2/RestartContinuation.ts)
74rechecks the preference, archive state, provider selection, and newer user work before
75dispatching. Stable command and message IDs prevent duplicate submissions after an
76outbox retry. Codex resumes without adding provider prompt text; other adapters receive
77the continuation message through their normal turn path.
76rechecks the preference, archive state, provider selection, newer user work, a stop
77the user requested, and maintenance turns such as `/compact` before dispatching. Stable
78command and message IDs prevent duplicate submissions after an outbox retry. Codex
79resumes without adding provider prompt text unless the turn lost background work;
80other adapters receive the continuation message through their normal turn path.
81
82Delegated tasks (`delegate_task` child threads) are reconciled as their own threads,
83never as the parent's background work. The orchestrator settles child results and
84completion deliveries in a startup pass after reconciliation, because the terminal-run
85listener ignores reconciliation's cancellations. A cancelled child whose restart
86continuation is still pending in the outbox is not a result yet; the continuation's run
87settles it, or the handler settles it when it declines to continue. Schedulers wait for
88activation so they cannot start runs that reconciliation would then cancel.
packages/shared/src/orchestrationV2ThreadError.ts+18-1
......@@ -55,11 +55,28 @@ export function latestExecutedRun(
5555 for (const run of runs) {
5656 if (run.status === "queued") continue;
5757 if (run.status === "cancelled" && run.startedAt === null) continue;
58 if (latest === null || run.ordinal > latest.ordinal) latest = run;
58 if (latest === null || runRanAfter(run, latest)) latest = run;
5959 }
6060 return latest;
6161}
6262
63/**
64 * Whether started `run` ran after `other`. Ordinals follow submission, but a
65 * run can start ahead of a held queue (a restart continuation, or a message
66 * sent while the queue is held), so a queued run resumed later can have a
67 * lower ordinal than one that already ended. An unfinished run is the latest.
68 */
69export function runRanAfter(
70 run: Pick<OrchestrationV2Run, "ordinal" | "completedAt">,
71 other: Pick<OrchestrationV2Run, "ordinal" | "completedAt">,
72): boolean {
73 const end = (candidate: typeof run) =>
74 !candidate.completedAt
75 ? Number.POSITIVE_INFINITY
76 : DateTime.toEpochMillis(candidate.completedAt);
77 return end(run) === end(other) ? run.ordinal > other.ordinal : end(run) > end(other);
78}
79
6380/**
6481 * The latest run that actually started, when it stopped because the
6582 * subscription limit was reached. Queued messages after that run must stay