| author | |
| committer | |
| log | b9ad3daf8d489604e0eca9d32a7e08af7c4e9fb7 |
| tree | bbd60aa1e15c1c4031e68fade5043a3c5aa02df9 |
| parent | 12b34e83af63da05368e97b7cd0bb7719b17a62d |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>10 files changed, 270 insertions(+), 38 deletions(-)
apps/server/src/orchestration-v2/FoundationPersistence.test.ts+2| ... | ... | @@ -2779,6 +2779,8 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { |
| 2779 | 2779 | { id: sessionId, driver: "codex", providerInstanceId, status: "running" }, |
| 2780 | 2780 | ], |
| 2781 | 2781 | providerTurns: [{ providerThreadId, runAttemptId: attemptId, status: "running" }], |
| 2782 | runtimeRequests: [], | |
| 2783 | turnItems: [], | |
| 2782 | 2784 | } as unknown as OrchestrationV2ThreadProjection; |
| 2783 | 2785 | const recovery = yield* ProviderRuntimeRecovery.make.pipe( |
| 2784 | 2786 | Effect.provide( |
apps/server/src/orchestration-v2/ProjectionRecovery.test.ts+1-1| ... | ... | @@ -502,7 +502,7 @@ it.effect("includes shared sessions and provider-owned background rosters in rec |
| 502 | 502 | preparedState.providerSessions.map((session) => session.id), |
| 503 | 503 | [preparedSessionId], |
| 504 | 504 | ); |
| 505 | assert.equal(restartContinuationRun(preparedState)?.id, preparedRunId); | |
| 505 | assert.equal(restartContinuationRun(preparedState, yield* DateTime.now)?.id, preparedRunId); | |
| 506 | 506 | assert.deepEqual(yield* projections.getUnreadableThreadIds(), []); |
| 507 | 507 | const sql = yield* SqlClient.SqlClient; |
| 508 | 508 | yield* sql` |
apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts+2-2| ... | ... | @@ -561,7 +561,7 @@ export const make = Effect.gen(function* () { |
| 561 | 561 | } |
| 562 | 562 | const continuationRun = |
| 563 | 563 | continueAfterRestart && trigger === "startup" |
| 564 | ? restartContinuationRun(projection) | |
| 564 | ? restartContinuationRun(projection, now) | |
| 565 | 565 | : undefined; |
| 566 | 566 | const effects: Array<EffectOutbox.PendingOrchestrationEffectV2> = continuationRun |
| 567 | 567 | ? [ |
| ... | ... | @@ -697,7 +697,7 @@ export const make = Effect.gen(function* () { |
| 697 | 697 | .continueThreadsAfterServerUpdate |
| 698 | 698 | ) |
| 699 | 699 | return; |
| 700 | const run = restartContinuationRun(projection); | |
| 700 | const run = restartContinuationRun(projection, yield* DateTime.now); | |
| 701 | 701 | if (!run) return; |
| 702 | 702 | const commandId = CommandId.make(`command:restart-prepare:${run.id}`); |
| 703 | 703 | yield* eventSink.writeWithEffects({ |
apps/server/src/orchestration-v2/RestartContinuation.test.ts+151-18| ... | ... | @@ -1,6 +1,7 @@ |
| 1 | 1 | import { assert, it } from "@effect/vitest"; |
| 2 | 2 | import { |
| 3 | 3 | MessageId, |
| 4 | NodeId, | |
| 4 | 5 | ProjectId, |
| 5 | 6 | ProviderDriverKind, |
| 6 | 7 | ProviderInstanceId, |
| ... | ... | @@ -9,6 +10,7 @@ import { |
| 9 | 10 | ProviderTurnId, |
| 10 | 11 | RunAttemptId, |
| 11 | 12 | RunId, |
| 13 | RuntimeRequestId, | |
| 12 | 14 | ThreadId, |
| 13 | 15 | type OrchestrationV2Command, |
| 14 | 16 | type OrchestrationV2ThreadProjection, |
| ... | ... | @@ -35,6 +37,7 @@ const sessionId = ProviderSessionId.make("session:restart"); |
| 35 | 37 | const attemptId = RunAttemptId.make("attempt:restart"); |
| 36 | 38 | // "No project" threads belong to the environment's Scratch project. |
| 37 | 39 | const scratchProjectId = ProjectId.make("project:scratch"); |
| 40 | const now = DateTime.makeUnsafe("2026-10-03T12:00:00.000Z"); | |
| 38 | 41 | |
| 39 | 42 | function makeProjection() { |
| 40 | 43 | return { |
| ... | ... | @@ -90,7 +93,7 @@ function makeProjection() { |
| 90 | 93 | |
| 91 | 94 | it("requires matching saved native state for an unfinished root run", () => { |
| 92 | 95 | const projection = makeProjection(); |
| 93 | assert.equal(restartContinuationRun(projection)?.id, runId); | |
| 96 | assert.equal(restartContinuationRun(projection, now)?.id, runId); | |
| 94 | 97 | for (const invalid of [ |
| 95 | 98 | { ...projection, thread: { ...projection.thread, archivedAt: {} } }, |
| 96 | 99 | { ...projection, thread: { ...projection.thread, deletedAt: {} } }, |
| ... | ... | @@ -123,7 +126,7 @@ it("requires matching saved native state for an unfinished root run", () => { |
| 123 | 126 | "interrupted", |
| 124 | 127 | ].map((status) => ({ ...projection, runs: [{ ...projection.runs[0]!, status }] })), |
| 125 | 128 | ]) |
| 126 | assert.isUndefined(restartContinuationRun(invalid as OrchestrationV2ThreadProjection)); | |
| 129 | assert.isUndefined(restartContinuationRun(invalid as OrchestrationV2ThreadProjection, now)); | |
| 127 | 130 | }); |
| 128 | 131 | |
| 129 | 132 | it("continues a live turn whose session the adapter never marked running", () => { |
| ... | ... | @@ -135,9 +138,9 @@ it("continues a live turn whose session the adapter never marked running", () => |
| 135 | 138 | providerSessions: [{ ...projection.providerSessions[0]!, status }], |
| 136 | 139 | }) as OrchestrationV2ThreadProjection; |
| 137 | 140 | for (const status of ["starting", "ready", "running", "waiting"]) |
| 138 | assert.equal(restartContinuationRun(withSessionStatus(status))?.id, runId, status); | |
| 141 | assert.equal(restartContinuationRun(withSessionStatus(status), now)?.id, runId, status); | |
| 139 | 142 | for (const status of ["stopped", "error"]) |
| 140 | assert.isUndefined(restartContinuationRun(withSessionStatus(status)), status); | |
| 143 | assert.isUndefined(restartContinuationRun(withSessionStatus(status), now), status); | |
| 141 | 144 | }); |
| 142 | 145 | |
| 143 | 146 | it("recovers an admitted continuation after another crash before provider start", () => { |
| ... | ... | @@ -155,7 +158,7 @@ it("recovers an admitted continuation after another crash before provider start" |
| 155 | 158 | providerSessions: [{ ...projection.providerSessions[0]!, status: "stopped" as const }], |
| 156 | 159 | providerTurns: [], |
| 157 | 160 | }; |
| 158 | assert.equal(restartContinuationRun(starting)?.id, runId); | |
| 161 | assert.equal(restartContinuationRun(starting, now)?.id, runId); | |
| 159 | 162 | }); |
| 160 | 163 | |
| 161 | 164 | it("does not continue settled root runs with restart-cancelled background work", () => { |
| ... | ... | @@ -176,11 +179,14 @@ it("does not continue settled root runs with restart-cancelled background work", |
| 176 | 179 | for (const projectId of [scratchProjectId, projection.thread.projectId]) |
| 177 | 180 | for (const status of ["completed", "waiting"] as const) |
| 178 | 181 | assert.isUndefined( |
| 179 | restartContinuationRun({ | |
| 180 | ...settled, | |
| 181 | thread: { ...settled.thread, projectId }, | |
| 182 | runs: [{ ...settled.runs[0]!, status }], | |
| 183 | }), | |
| 182 | restartContinuationRun( | |
| 183 | { | |
| 184 | ...settled, | |
| 185 | thread: { ...settled.thread, projectId }, | |
| 186 | runs: [{ ...settled.runs[0]!, status }], | |
| 187 | }, | |
| 188 | now, | |
| 189 | ), | |
| 184 | 190 | ); |
| 185 | 191 | }); |
| 186 | 192 | |
| ... | ... | @@ -663,10 +669,13 @@ it.effect("continues a cut run past queued follow-ups, which stay held", () => |
| 663 | 669 | Effect.gen(function* () { |
| 664 | 670 | const live = makeProjection(); |
| 665 | 671 | assert.equal( |
| 666 | restartContinuationRun({ | |
| 667 | ...live, | |
| 668 | runs: [...live.runs, queuedFollowUp], | |
| 669 | } as unknown as OrchestrationV2ThreadProjection)?.id, | |
| 672 | restartContinuationRun( | |
| 673 | { | |
| 674 | ...live, | |
| 675 | runs: [...live.runs, queuedFollowUp], | |
| 676 | } as unknown as OrchestrationV2ThreadProjection, | |
| 677 | now, | |
| 678 | )?.id, | |
| 670 | 679 | runId, |
| 671 | 680 | ); |
| 672 | 681 | const cut = cutMidTurn(); |
| ... | ... | @@ -692,10 +701,13 @@ it.effect("continues a resumed queued run that ran after an earlier continuation |
| 692 | 701 | }; |
| 693 | 702 | const live = makeProjection(); |
| 694 | 703 | assert.equal( |
| 695 | restartContinuationRun({ | |
| 696 | ...live, | |
| 697 | runs: [...live.runs, finishedContinuation], | |
| 698 | } as unknown as OrchestrationV2ThreadProjection)?.id, | |
| 704 | restartContinuationRun( | |
| 705 | { | |
| 706 | ...live, | |
| 707 | runs: [...live.runs, finishedContinuation], | |
| 708 | } as unknown as OrchestrationV2ThreadProjection, | |
| 709 | now, | |
| 710 | )?.id, | |
| 699 | 711 | runId, |
| 700 | 712 | ); |
| 701 | 713 | const cut = cutMidTurn({ completedAt: DateTime.makeUnsafe("2026-10-03T10:05:00.000Z") }); |
| ... | ... | @@ -835,3 +847,124 @@ it.effect("prepares later threads' continuations when one thread fails", () => |
| 835 | 847 | ); |
| 836 | 848 | }), |
| 837 | 849 | ); |
| 850 | ||
| 851 | it.effect("stops continuing after three automatic restart continuations in a row", () => | |
| 852 | Effect.gen(function* () { | |
| 853 | const first = RunId.make("run:first-continuation"); | |
| 854 | const second = RunId.make("run:second-continuation"); | |
| 855 | const cut = cutMidTurn({ restartContinuationOfRunId: second }); | |
| 856 | const withTwoLinks = { | |
| 857 | ...cut, | |
| 858 | runs: [ | |
| 859 | ...cut.runs, | |
| 860 | { ...cut.runs[0]!, id: second, restartContinuationOfRunId: first }, | |
| 861 | { ...cut.runs[0]!, id: first, restartContinuationOfRunId: undefined }, | |
| 862 | ], | |
| 863 | } as unknown as OrchestrationV2ThreadProjection; | |
| 864 | assert.lengthOf(yield* continuationTexts(withTwoLinks), 1); | |
| 865 | ||
| 866 | const withThreeLinks = { | |
| 867 | ...withTwoLinks, | |
| 868 | runs: withTwoLinks.runs.map((run) => | |
| 869 | run.id === first ? { ...run, restartContinuationOfRunId: RunId.make("run:root") } : run, | |
| 870 | ), | |
| 871 | } as unknown as OrchestrationV2ThreadProjection; | |
| 872 | assert.lengthOf(yield* continuationTexts(withThreeLinks), 0); | |
| 873 | }), | |
| 874 | ); | |
| 875 | ||
| 876 | it.effect.each([ | |
| 877 | ["user_input", "pending", "live", false], | |
| 878 | ["user_input", "expired", "not_resumable", false], | |
| 879 | ["user_input", "cancelled", "not_resumable", false], | |
| 880 | ["command", "pending", "live", false], | |
| 881 | ["command", "expired", "not_resumable", false], | |
| 882 | ["user_input", "resolved", "live", true], | |
| 883 | ["command", "cancelled", "live", true], | |
| 884 | ["user_input", "pending", "message", true], | |
| 885 | ["dynamic_tool_call", "pending", "live", true], | |
| 886 | ["auth_refresh", "pending", "live", true], | |
| 887 | ] as const)( | |
| 888 | "a %s request that is %s with %s capability allows continuation: %s", | |
| 889 | ([kind, status, capability, shouldContinue]) => | |
| 890 | Effect.gen(function* () { | |
| 891 | const base = makeProjection(); | |
| 892 | const runtimeRequests = [ | |
| 893 | { | |
| 894 | id: RuntimeRequestId.make("request:restart"), | |
| 895 | nodeId: NodeId.make("node:restart-question"), | |
| 896 | providerTurnId: base.providerTurns[0]!.id, | |
| 897 | kind, | |
| 898 | status, | |
| 899 | responseCapability: | |
| 900 | capability === "live" | |
| 901 | ? { type: capability, providerSessionId: sessionId } | |
| 902 | : capability === "not_resumable" | |
| 903 | ? { type: capability, reason: "The server restarted." } | |
| 904 | : { type: capability }, | |
| 905 | }, | |
| 906 | ]; | |
| 907 | const live = { ...base, runtimeRequests } as unknown as OrchestrationV2ThreadProjection; | |
| 908 | assert.equal(restartContinuationRun(live, now) !== undefined, shouldContinue); | |
| 909 | const cut = { | |
| 910 | ...cutMidTurn(), | |
| 911 | runtimeRequests, | |
| 912 | } as unknown as OrchestrationV2ThreadProjection; | |
| 913 | assert.lengthOf(yield* continuationTexts(cut), shouldContinue ? 1 : 0); | |
| 914 | }), | |
| 915 | ); | |
| 916 | ||
| 917 | it.effect("does not wake a snoozed or settled thread, but continues once a snooze elapsed", () => | |
| 918 | Effect.gen(function* () { | |
| 919 | const live = makeProjection(); | |
| 920 | for (const held of [ | |
| 921 | { snoozedUntil: DateTime.add(now, { days: 1 }) }, | |
| 922 | { settledOverride: "settled" as const }, | |
| 923 | ]) { | |
| 924 | const thread = { ...live.thread, ...held }; | |
| 925 | assert.isUndefined(restartContinuationRun({ ...live, thread }, now)); | |
| 926 | assert.deepEqual(yield* continuationTexts({ ...cutMidTurn(), thread }), []); | |
| 927 | } | |
| 928 | const thread = { ...live.thread, snoozedUntil: DateTime.subtract(now, { minutes: 1 }) }; | |
| 929 | assert.equal(restartContinuationRun({ ...live, thread }, now)?.id, runId); | |
| 930 | }), | |
| 931 | ); | |
| 932 | ||
| 933 | it.effect("does not continue a run waiting on a secret the restart closed", () => | |
| 934 | Effect.gen(function* () { | |
| 935 | const completedAt = DateTime.makeUnsafe("2026-10-03T10:05:00.000Z"); | |
| 936 | const secret = (secretStatus: string, at: DateTime.Utc | null) => | |
| 937 | ({ type: "secret_request", runId, secretStatus, completedAt: at }) as never; | |
| 938 | assert.isUndefined( | |
| 939 | restartContinuationRun({ ...makeProjection(), turnItems: [secret("pending", null)] }, now), | |
| 940 | ); | |
| 941 | const cut = cutMidTurn({ completedAt }); | |
| 942 | assert.deepEqual( | |
| 943 | yield* continuationTexts({ ...cut, turnItems: [secret("cancelled", completedAt)] }), | |
| 944 | [], | |
| 945 | ); | |
| 946 | // A secret the user dismissed earlier in the run does not hold it. | |
| 947 | assert.lengthOf( | |
| 948 | yield* continuationTexts({ | |
| 949 | ...cut, | |
| 950 | turnItems: [secret("cancelled", DateTime.subtract(completedAt, { seconds: 1 }))], | |
| 951 | }), | |
| 952 | 1, | |
| 953 | ); | |
| 954 | }), | |
| 955 | ); | |
| 956 | ||
| 957 | it.effect.each(["completed", "failed", "interrupted"] as const)( | |
| 958 | "does not continue when the provider turn %s before the app run settled", | |
| 959 | (status) => | |
| 960 | Effect.gen(function* () { | |
| 961 | const cut = cutMidTurn(); | |
| 962 | assert.deepEqual( | |
| 963 | yield* continuationTexts({ | |
| 964 | ...cut, | |
| 965 | providerTurns: [{ ...cut.providerTurns[0]!, status }], | |
| 966 | }), | |
| 967 | [], | |
| 968 | ); | |
| 969 | }), | |
| 970 | ); |
apps/server/src/orchestration-v2/RestartContinuation.ts+89-7| ... | ... | @@ -2,11 +2,13 @@ import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError"; |
| 2 | 2 | import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; |
| 3 | 3 | import { |
| 4 | 4 | CommandId, |
| 5 | latestProviderTurnForAttempt, | |
| 5 | 6 | MessageId, |
| 6 | 7 | type OrchestrationV2Run, |
| 7 | 8 | type RunId, |
| 8 | 9 | type ThreadId, |
| 9 | 10 | } from "@t3tools/contracts"; |
| 11 | import * as DateTime from "effect/DateTime"; | |
| 10 | 12 | import * as Effect from "effect/Effect"; |
| 11 | 13 | import type { ProjectionRuntimeRecoveryState } from "./ProjectionStore.ts"; |
| 12 | 14 | |
| ... | ... | @@ -21,6 +23,64 @@ import { |
| 21 | 23 | |
| 22 | 24 | const CONTINUE_PROMPT = "Continue where you left off."; |
| 23 | 25 | |
| 26 | /** Consecutive automatic continuations allowed before a person has to step in. */ | |
| 27 | const MAX_CONTINUATION_CHAIN = 3; | |
| 28 | ||
| 29 | /** | |
| 30 | * Whether a restart must leave this run for the user: the thread is put away, | |
| 31 | * the provider turn already ended, or the user owes an answer. Checked when | |
| 32 | * intent is recorded and again when the continuation is delivered. | |
| 33 | */ | |
| 34 | function restartContinuationBlocked( | |
| 35 | projection: Pick< | |
| 36 | ProjectionRuntimeRecoveryState, | |
| 37 | "thread" | "runtimeRequests" | "providerTurns" | "turnItems" | |
| 38 | >, | |
| 39 | source: OrchestrationV2Run, | |
| 40 | now: DateTime.Utc, | |
| 41 | ): boolean { | |
| 42 | const { thread } = projection; | |
| 43 | if ( | |
| 44 | thread.archivedAt !== null || | |
| 45 | thread.deletedAt !== null || | |
| 46 | thread.settledOverride === "settled" || | |
| 47 | (thread.snoozedUntil != null && DateTime.isGreaterThan(thread.snoozedUntil, now)) | |
| 48 | ) | |
| 49 | return true; | |
| 50 | const turn = latestProviderTurnForAttempt(projection.providerTurns, source.activeAttemptId); | |
| 51 | if ( | |
| 52 | turn !== undefined && | |
| 53 | (turn.status === "completed" || turn.status === "failed" || turn.status === "interrupted") | |
| 54 | ) | |
| 55 | return true; | |
| 56 | const waitingRequest = projection.runtimeRequests.some( | |
| 57 | (request) => | |
| 58 | // Tool calls and auth refreshes do not wait on a person, and a | |
| 59 | // message-capable question stays answerable after the restart. | |
| 60 | request.kind !== "dynamic_tool_call" && | |
| 61 | request.kind !== "auth_refresh" && | |
| 62 | request.responseCapability.type !== "message" && | |
| 63 | (request.status === "pending" || | |
| 64 | // Recovery closes live callbacks on the cut turn; it does not answer them. | |
| 65 | ((request.status === "expired" || request.status === "cancelled") && | |
| 66 | request.responseCapability.type === "not_resumable" && | |
| 67 | turn !== undefined && | |
| 68 | turn.id === request.providerTurnId)), | |
| 69 | ); | |
| 70 | if (waitingRequest) return true; | |
| 71 | return projection.turnItems.some( | |
| 72 | (item) => | |
| 73 | item.type === "secret_request" && | |
| 74 | item.runId === source.id && | |
| 75 | (item.secretStatus === "pending" || | |
| 76 | // Recovery cancels an open secret card and its run at the same instant. | |
| 77 | (item.secretStatus === "cancelled" && | |
| 78 | item.completedAt !== null && | |
| 79 | source.completedAt !== null && | |
| 80 | DateTime.Equivalence(item.completedAt, source.completedAt))), | |
| 81 | ); | |
| 82 | } | |
| 83 | ||
| 24 | 84 | /** |
| 25 | 85 | * Resume only an unfinished root turn. Leftover background work is cleaned up |
| 26 | 86 | * separately and reported on the next user turn; it must not wake a settled run. |
| ... | ... | @@ -28,10 +88,16 @@ const CONTINUE_PROMPT = "Continue where you left off."; |
| 28 | 88 | export function restartContinuationRun( |
| 29 | 89 | projection: Pick< |
| 30 | 90 | ProjectionRuntimeRecoveryState, |
| 31 | "thread" | "runs" | "providerThreads" | "providerSessions" | "providerTurns" | |
| 91 | | "thread" | |
| 92 | | "runs" | |
| 93 | | "providerThreads" | |
| 94 | | "providerSessions" | |
| 95 | | "providerTurns" | |
| 96 | | "runtimeRequests" | |
| 97 | | "turnItems" | |
| 32 | 98 | >, |
| 99 | now: DateTime.Utc, | |
| 33 | 100 | ): OrchestrationV2Run | undefined { |
| 34 | if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) return; | |
| 35 | 101 | // Queued runs never started; recovery holds them behind the cut run. |
| 36 | 102 | const run = projection.runs.reduce<OrchestrationV2Run | undefined>( |
| 37 | 103 | (latest, candidate) => |
| ... | ... | @@ -44,6 +110,7 @@ export function restartContinuationRun( |
| 44 | 110 | const preparedContinuation = |
| 45 | 111 | run.status === "starting" && run.restartContinuationOfRunId !== undefined; |
| 46 | 112 | if (run.status !== "running" && !preparedContinuation) return; |
| 113 | if (restartContinuationBlocked(projection, run, now)) return; | |
| 47 | 114 | const liveTurnRequired = !preparedContinuation; |
| 48 | 115 | if (projection.thread.providerInstanceId !== run.providerInstanceId) return; |
| 49 | 116 | const providerThread = projection.providerThreads.find( |
| ... | ... | @@ -96,16 +163,14 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta |
| 96 | 163 | const messageId = MessageId.make(`message:restart-continuation:${input.sourceRunId}`); |
| 97 | 164 | const projection = yield* threads.getThreadRecords( |
| 98 | 165 | input.threadId, |
| 99 | ["messages", "runs", "providerTurns", "attempts"], | |
| 100 | { messageIds: [messageId] }, | |
| 166 | ["messages", "runs", "providerTurns", "attempts", "runtimeRequests", "turnItems"], | |
| 167 | { messageIds: [messageId], turnItemTypes: ["secret_request"] }, | |
| 101 | 168 | ); |
| 102 | 169 | if ( |
| 103 | 170 | !resolveProjectSettings(enabled, projection.thread.projectId).settings |
| 104 | 171 | .continueThreadsAfterServerUpdate |
| 105 | 172 | ) |
| 106 | 173 | return; |
| 107 | if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) return; | |
| 108 | ||
| 109 | 174 | if (projection.messages.some((message) => message.id === messageId)) return; |
| 110 | 175 | const source = projection.runs.find((run) => run.id === input.sourceRunId); |
| 111 | 176 | // Pending effects from older versions may target settled background work, |
| ... | ... | @@ -113,7 +178,8 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta |
| 113 | 178 | if ( |
| 114 | 179 | !source || |
| 115 | 180 | source.status !== "cancelled" || |
| 116 | isRestartNoteSource(source, projection.providerTurns) | |
| 181 | isRestartNoteSource(source, projection.providerTurns) || | |
| 182 | restartContinuationBlocked(projection, source, yield* DateTime.now) | |
| 117 | 183 | ) |
| 118 | 184 | return; |
| 119 | 185 | // A user submission after reconciliation takes precedence over an automatic |
| ... | ... | @@ -125,6 +191,22 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta |
| 125 | 191 | ) |
| 126 | 192 | return; |
| 127 | 193 | if (projection.thread.providerInstanceId !== source.providerInstanceId) return; |
| 194 | // A turn that keeps crashing the server would otherwise restart it forever. | |
| 195 | let chainLength = 0; | |
| 196 | for ( | |
| 197 | let run: OrchestrationV2Run | undefined = source; | |
| 198 | run?.restartContinuationOfRunId !== undefined && chainLength < MAX_CONTINUATION_CHAIN; | |
| 199 | run = projection.runs.find((candidate) => candidate.id === run?.restartContinuationOfRunId) | |
| 200 | ) { | |
| 201 | chainLength += 1; | |
| 202 | } | |
| 203 | if (chainLength >= MAX_CONTINUATION_CHAIN) { | |
| 204 | yield* Effect.logWarning("Not continuing a run restarted too many times in a row", { | |
| 205 | threadId: input.threadId, | |
| 206 | sourceRunId: input.sourceRunId, | |
| 207 | }); | |
| 208 | return; | |
| 209 | } | |
| 128 | 210 | const sourceRecords = yield* threads.getThreadRecords( |
| 129 | 211 | input.threadId, |
| 130 | 212 | ["messages", "turnItems"], |
apps/server/src/serverSettings.test.ts+5-1| ... | ... | @@ -1118,7 +1118,8 @@ it.layer(NodeServices.layer)("server settings", (it) => { |
| 1118 | 1118 | const serverConfig = yield* ServerConfig.ServerConfig; |
| 1119 | 1119 | const fileSystem = yield* FileSystem.FileSystem; |
| 1120 | 1120 | const serverSettings = yield* ServerSettingsModule.ServerSettingsService; |
| 1121 | const raw = '{"addProjectBaseDirectory":42,"providers":{"codex":{"binaryPath":"/x"}}}'; | |
| 1121 | const raw = | |
| 1122 | '{"addProjectBaseDirectory":42,"continueThreadsAfterServerUpdate":"false","providers":{"codex":{"binaryPath":"/x"}}}'; | |
| 1122 | 1123 | yield* fileSystem.writeFileString(serverConfig.settingsPath, raw); |
| 1123 | 1124 | |
| 1124 | 1125 | const settings = yield* serverSettings.getSettings; |
| ... | ... | @@ -1128,6 +1129,8 @@ it.layer(NodeServices.layer)("server settings", (it) => { |
| 1128 | 1129 | driver: ProviderDriverKind.make("codex"), |
| 1129 | 1130 | config: { binaryPath: "/x" }, |
| 1130 | 1131 | }); |
| 1132 | // An unreadable file may hold an opt-out, so threads are not resumed on a guess. | |
| 1133 | assert.isFalse(settings.continueThreadsAfterServerUpdate); | |
| 1131 | 1134 | assert.equal(yield* fileSystem.readFileString(serverConfig.settingsPath), raw); |
| 1132 | 1135 | }).pipe(Effect.provide(layerServerSettings())), |
| 1133 | 1136 | ); |
| ... | ... | @@ -1969,6 +1972,7 @@ it.layer(NodeServices.layer)("server settings", (it) => { |
| 1969 | 1972 | |
| 1970 | 1973 | const settings = yield* serverSettings.getSettings; |
| 1971 | 1974 | assert.isFalse(settings.projectSettingsFolded); |
| 1975 | assert.isFalse(settings.continueThreadsAfterServerUpdate); | |
| 1972 | 1976 | assert.deepEqual(settings.projectSettingsOverrides, {}); |
| 1973 | 1977 | // The user's file is still there to repair; nothing was written over it. |
| 1974 | 1978 | assert.equal(yield* fileSystem.readFileString(serverConfig.settingsPath), broken); |
apps/server/src/serverSettings.ts+2| ... | ... | @@ -834,6 +834,8 @@ const make = Effect.gen(function* () { |
| 834 | 834 | if (decoded._tag === "Failure" || legacySettings._tag === "Failure") { |
| 835 | 835 | const failure = decoded._tag === "Failure" ? decoded : legacySettings; |
| 836 | 836 | settingsFileTrusted = false; |
| 837 | // The file may hold an opt-out we can no longer read; never resume threads on a guess. | |
| 838 | settings = { ...DEFAULT_SERVER_SETTINGS, continueThreadsAfterServerUpdate: false }; | |
| 837 | 839 | if (failure._tag === "Failure") { |
| 838 | 840 | yield* Effect.logWarning("failed to parse settings.json, using defaults", { |
| 839 | 841 | path: settingsPath, |
docs/internals/server-updates.md+10-3| ... | ... | @@ -58,7 +58,9 @@ stopped backends and replays the failure for the same token. |
| 58 | 58 | |
| 59 | 59 | ## Recovering interrupted threads |
| 60 | 60 | |
| 61 | Restart continuation is an environment-owned preference, off by default. The | |
| 61 | Restart continuation is an environment-owned preference, on by default. A | |
| 62 | `settings.json` that fails to decode turns it off, since the file may hold an | |
| 63 | opt-out. The | |
| 62 | 64 | [v2 recovery service](../../apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts) |
| 63 | 65 | requires matching durable run, provider thread, session, and native resume identity. |
| 64 | 66 | Queued runs never started, so recovery holds them and continues the run they wait |
| ... | ... | @@ -73,8 +75,13 @@ closing providers, then reconciles after ingestion has stopped so a late complet |
| 73 | 75 | cannot be overwritten by a stale cancellation. |
| 74 | 76 | |
| 75 | 77 | The [continuation handler](../../apps/server/src/orchestration-v2/RestartContinuation.ts) |
| 76 | rechecks the preference, archive state, provider selection, newer user work, a stop | |
| 77 | the user requested, and maintenance turns such as `/compact` before dispatching. Stable | |
| 78 | rechecks the preference, provider selection, newer user work, a stop the user requested, | |
| 79 | and maintenance turns such as `/compact` before dispatching. Intent and delivery both | |
| 80 | refuse a thread that is archived, settled, or still snoozed, a provider turn that already | |
| 81 | ended, and a run waiting on the user: an approval, question, or secret that was pending, | |
| 82 | or that recovery closed on the cut turn. Message-capable questions stay answerable, so | |
| 83 | they do not block. Delivery also stops after three consecutive continuations in one | |
| 84 | `restartContinuationOfRunId` chain, so a turn that crashes the server cannot loop. Stable | |
| 78 | 85 | command and message IDs prevent duplicate submissions after an outbox retry. Codex |
| 79 | 86 | resumes without adding provider prompt text unless the turn lost background work; |
| 80 | 87 | other adapters receive the continuation message through their normal turn path. |
docs/user/updating.md+5-4| ... | ... | @@ -10,16 +10,17 @@ notice. |
| 10 | 10 | Server updates restart the connection and can interrupt active agents and |
| 11 | 11 | terminal commands. Saved threads, settings, and project files remain. |
| 12 | 12 | |
| 13 | **Settings → General → Continue threads after restarts** is off by default. | |
| 14 | Enable it to resume supported active threads after an update, crash, or machine | |
| 13 | **Settings → General → Continue threads after restarts** is on by default. | |
| 14 | It resumes supported active threads after an update, crash, or machine | |
| 15 | 15 | restart. Changes are saved to connected environments that support this setting; |
| 16 | 16 | update older servers first. If a supported environment was offline or has a |
| 17 | 17 | different value, use **Apply to all** in Settings after it connects. |
| 18 | 18 | T3 Code must start again on that machine; |
| 19 | 19 | the setting does not enable automatic startup. Terminal commands may still be |
| 20 | 20 | interrupted, and threads without saved provider resume state need a new message. |
| 21 | If you previously enabled continuation for updates, enable this setting once | |
| 22 | to allow recovery without a connected client. | |
| 21 | A thread is not resumed while it waits for your approval, answer, or secret, or | |
| 22 | while it is archived, snoozed, or settled. After three automatic resumes in a row, | |
| 23 | the thread waits for your next message. | |
| 23 | 24 | |
| 24 | 25 | Updates from the previous orchestration system preserve conversation transcripts but cannot carry |
| 25 | 26 | every kind of runtime history forward. Read [Threads from older T3 Code versions](./thread-migration.md) |
packages/contracts/src/settings.ts+3-2| ... | ... | @@ -1107,9 +1107,10 @@ export const ServerSettings = Schema.Struct({ |
| 1107 | 1107 | Schema.withDecodingDefault(Effect.succeed("paragraph" as const)), |
| 1108 | 1108 | ), |
| 1109 | 1109 | enableProviderUpdateChecks: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), |
| 1110 | // Retain the update-era key; recovery now needs an environment-owned opt-in. | |
| 1110 | // Retain the update-era key. On by default so restarts resume the work they | |
| 1111 | // interrupted; recovery stops after three automatic continuations in a row. | |
| 1111 | 1112 | continueThreadsAfterServerUpdate: Schema.Boolean.pipe( |
| 1112 | Schema.withDecodingDefault(Effect.succeed(false)), | |
| 1113 | Schema.withDecodingDefault(Effect.succeed(true)), | |
| 1113 | 1114 | ), |
| 1114 | 1115 | /** |
| 1115 | 1116 | * Whether agents may drive the in-app preview browser. Turning this off |