authorgravatar for 269035359+t3-code[bot]@users.noreply.github.comt3-code[bot] <269035359+t3-code[bot]@users.noreply.github.com> 2026-08-20 16:26:24-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-20 16:26:24-04:00
log0929907ff9aecd2541050fe4361ad03199ffbf73
tree8df4e22116a90410e82378ea9fbcc5e4d4d00152
parentaa17ec6e7d77d2bcb8f86720e41b7e5402d6b74c
signature Signed by PGP key B5690EEEBB952194

fix(server): reconcile orphaned provider sessions (#7719)

Co-authored-by: t3-code[bot] <269035359+t3-code[bot]@users.noreply.github.com> Co-authored-by: maria <254055478+maria-rcks@users.noreply.github.com>

4 files changed, 743 insertions(+), 0 deletions(-)

.macroscope/check-run-agents/effect-service-conventions.md+1
......@@ -68,6 +68,7 @@ Review changed TypeScript and directly affected call sites for the conventions b
6868- Export direct schema predicates such as `export const isFoo = Schema.is(Foo)`. Flag a private `Schema.is` constant wrapped by a redundant function with the same signature.
6969- Do not introduce a large `switch` or lookup table in an error's `message` getter to model failures that deserve separate error classes.
7070- Catch statically known tagged failures with `Effect.catchTags({ ... })`, including when handling only one tag. Do not use `catchIf` with a schema predicate merely to recover one or more known `_tag` variants, and do not use `catchTag`. `Effect.catch` is appropriate when the entire error channel is intentionally handled; `catchIf` remains appropriate for genuinely structural predicates such as inspecting an underlying platform error code.
71- For startup reconciliation that repairs multiple independent entities, preserve interruption rather than reducing it to a warning. Retry a transient per-entity repair before readiness, then isolate a persistent failure so one bad entity cannot abort global startup or prevent later entities from being repaired. Require tests for both the retry-success path and persistent-failure continuation.
7172- Do not add a helper whose only behavior is `(...args) => new SomeError({ ...args })`, including curried aliases used once with `mapError`. Construct the error at the failure boundary so its attributes and cause remain visible. Keep a mapper only when it performs real normalization, passes through existing domain errors, or adds reusable context/control flow.
7273- When a reusable error-to-error translation clearly belongs to the target error type, prefer a descriptive static factory on that error class over a detached production-side switch. Do not force a static method for one-off inline mappings.
7374
apps/server/integration/orphanedProviderSessionStartup.integration.test.ts created+358
......@@ -0,0 +1,358 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import {
3 CommandId,
4 DEFAULT_PROVIDER_INTERACTION_MODE,
5 EnvironmentId,
6 MessageId,
7 ProjectId,
8 ProviderDriverKind,
9 ProviderInstanceId,
10 ThreadId,
11} from "@t3tools/contracts";
12import { assert, it } from "@effect/vitest";
13import * as DateTime from "effect/DateTime";
14import * as Effect from "effect/Effect";
15import * as Exit from "effect/Exit";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as Stream from "effect/Stream";
19import * as SqlClient from "effect/unstable/sql/SqlClient";
20import { HttpServer } from "effect/unstable/http";
21
22import * as EnvironmentAuth from "../src/auth/EnvironmentAuth.ts";
23import * as ServiceLauncherClient from "../src/cloud/serviceLauncherClient.ts";
24import * as ServerConfig from "../src/config.ts";
25import * as ServerEnvironment from "../src/environment/ServerEnvironment.ts";
26import * as Keybindings from "../src/keybindings.ts";
27import { OrchestrationLayerLive } from "../src/orchestration/runtimeLayer.ts";
28import * as OrchestrationEngine from "../src/orchestration/Services/OrchestrationEngine.ts";
29import * as OrchestrationReactor from "../src/orchestration/Services/OrchestrationReactor.ts";
30import * as ProjectionSnapshotQuery from "../src/orchestration/Services/ProjectionSnapshotQuery.ts";
31import { makeSqlitePersistenceLive } from "../src/persistence/Layers/Sqlite.ts";
32import * as ProviderSessionRuntime from "../src/persistence/ProviderSessionRuntime.ts";
33import * as ExternalLauncher from "../src/process/externalLauncher.ts";
34import { ProviderSessionDirectoryLive } from "../src/provider/Layers/ProviderSessionDirectory.ts";
35import * as ProviderService from "../src/provider/Services/ProviderService.ts";
36import * as ProviderSessionDirectory from "../src/provider/Services/ProviderSessionDirectory.ts";
37import * as ProviderSessionReaper from "../src/provider/Services/ProviderSessionReaper.ts";
38import * as RepositoryIdentityResolver from "../src/project/RepositoryIdentityResolver.ts";
39import * as ServerLifecycleEvents from "../src/serverLifecycleEvents.ts";
40import * as ServerRuntimeStartup from "../src/serverRuntimeStartup.ts";
41import * as ServerSettings from "../src/serverSettings.ts";
42import * as AnalyticsService from "../src/telemetry/AnalyticsService.ts";
43
44const providerInstanceId = ProviderInstanceId.make("codex");
45const projectId = ProjectId.make("project-startup-orphan");
46const threadId = ThreadId.make("thread-startup-orphan");
47const stoppedBindingThreadId = ThreadId.make("thread-startup-orphan-stopped-binding");
48const resumeCursor = { schemaVersion: 1, sessionId: "provider-session-before-restart" };
49const stoppedBindingResumeCursor = {
50 schemaVersion: 1,
51 sessionId: "provider-session-stopped-before-restart",
52};
53
54const makePersistedRuntimeLayer = (dbPath: string) => {
55 const persistence = makeSqlitePersistenceLive(dbPath);
56 const orchestration = OrchestrationLayerLive.pipe(
57 Layer.provideMerge(RepositoryIdentityResolver.layer),
58 Layer.provideMerge(persistence),
59 );
60 const directory = ProviderSessionDirectoryLive.pipe(
61 Layer.provide(ProviderSessionRuntime.layer),
62 Layer.provide(persistence),
63 );
64 return Layer.mergeAll(orchestration, directory);
65};
66
67const startupDependencies = Layer.mergeAll(
68 Layer.mock(Keybindings.Keybindings)({
69 start: Effect.void,
70 }),
71 ServerSettings.layerTest(),
72 Layer.succeed(OrchestrationReactor.OrchestrationReactor, {
73 start: () => Effect.void,
74 }),
75 Layer.succeed(ProviderSessionReaper.ProviderSessionReaper, {
76 start: () => Effect.void,
77 }),
78 ServerLifecycleEvents.layer,
79 Layer.succeed(ServerEnvironment.ServerEnvironment, {
80 getEnvironmentId: Effect.succeed(EnvironmentId.make("environment-startup-orphan")),
81 getDescriptor: Effect.succeed({
82 environmentId: EnvironmentId.make("environment-startup-orphan"),
83 label: "Startup orphan test",
84 version: "test",
85 platform: { os: "linux", arch: "x64" },
86 capabilities: {},
87 } as never),
88 }),
89 Layer.mock(EnvironmentAuth.EnvironmentAuth)({
90 issueStartupPairingUrl: (baseUrl: string) => Effect.succeed(`${baseUrl}/pair`),
91 }),
92 Layer.mock(ExternalLauncher.ExternalLauncher)({
93 launchBrowser: () => Effect.void,
94 }),
95 Layer.succeed(ServiceLauncherClient.ServiceLauncherClient, {
96 managed: false,
97 requestUpdate: () => Effect.die("unused"),
98 prepareTrial: Effect.sync(() => undefined),
99 }),
100 Layer.succeed(
101 HttpServer.HttpServer,
102 HttpServer.HttpServer.of({
103 address: { _tag: "TcpAddress", hostname: "127.0.0.1", port: 3773 },
104 serve: (() => Effect.void) as HttpServer.HttpServer["Service"]["serve"],
105 }),
106 ),
107 AnalyticsService.layerTest,
108 Layer.succeed(ProviderService.ProviderService, {
109 startSession: () => Effect.die("unused"),
110 sendTurn: () => Effect.die("unused"),
111 interruptTurn: () => Effect.die("unused"),
112 respondToRequest: () => Effect.die("unused"),
113 respondToUserInput: () => Effect.die("unused"),
114 stopSession: () => Effect.die("unused"),
115 listSessions: () => Effect.succeed([]),
116 getCapabilities: () => Effect.die("unused"),
117 getInstanceInfo: () => Effect.die("unused"),
118 rollbackConversation: () => Effect.die("unused"),
119 streamEvents: Stream.empty,
120 }),
121);
122
123it.effect(
124 "recovers a persisted starting session before opening the command gate after restart",
125 () =>
126 Effect.gen(function* () {
127 const config = yield* ServerConfig.ServerConfig;
128 const firstRuntime = makePersistedRuntimeLayer(config.dbPath);
129 const now = yield* DateTime.now;
130 const createdAt = DateTime.formatIso(now);
131
132 yield* Effect.gen(function* () {
133 const engine = yield* OrchestrationEngine.OrchestrationEngineService;
134 const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
135
136 yield* engine.dispatch({
137 type: "project.create",
138 commandId: CommandId.make("command-create-project"),
139 projectId,
140 title: "Startup orphan project",
141 workspaceRoot: "/tmp/startup-orphan-project",
142 defaultModelSelection: { instanceId: providerInstanceId, model: "gpt-5" },
143 createdAt,
144 });
145 yield* engine.dispatch({
146 type: "thread.create",
147 commandId: CommandId.make("command-create-thread"),
148 threadId,
149 projectId,
150 title: "Startup orphan thread",
151 modelSelection: { instanceId: providerInstanceId, model: "gpt-5" },
152 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
153 runtimeMode: "full-access",
154 branch: null,
155 worktreePath: null,
156 createdAt,
157 });
158 yield* engine.dispatch({
159 type: "thread.turn.start",
160 commandId: CommandId.make("command-start-pending-turn"),
161 threadId,
162 message: {
163 messageId: MessageId.make("message-pending-before-restart"),
164 role: "user",
165 text: "Persist this queued turn before restart",
166 attachments: [],
167 },
168 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
169 runtimeMode: "full-access",
170 createdAt,
171 });
172 yield* engine.dispatch({
173 type: "thread.session.set",
174 commandId: CommandId.make("command-mark-session-starting"),
175 threadId,
176 session: {
177 threadId,
178 status: "starting",
179 providerName: "codex",
180 providerInstanceId,
181 runtimeMode: "full-access",
182 activeTurnId: null,
183 lastError: null,
184 updatedAt: createdAt,
185 },
186 createdAt,
187 });
188 yield* directory.upsert({
189 threadId,
190 provider: ProviderDriverKind.make("codex"),
191 providerInstanceId,
192 status: "running",
193 resumeCursor,
194 runtimePayload: { activeTurnId: null, unrelated: "preserve-me" },
195 runtimeMode: "full-access",
196 });
197 yield* engine.dispatch({
198 type: "thread.create",
199 commandId: CommandId.make("command-create-stopped-binding-thread"),
200 threadId: stoppedBindingThreadId,
201 projectId,
202 title: "Startup orphan with stopped binding",
203 modelSelection: { instanceId: providerInstanceId, model: "gpt-5" },
204 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
205 runtimeMode: "full-access",
206 branch: null,
207 worktreePath: null,
208 createdAt,
209 });
210 yield* engine.dispatch({
211 type: "thread.turn.start",
212 commandId: CommandId.make("command-start-stopped-binding-pending-turn"),
213 threadId: stoppedBindingThreadId,
214 message: {
215 messageId: MessageId.make("message-stopped-binding-pending-before-restart"),
216 role: "user",
217 text: "Persist another queued turn before restart",
218 attachments: [],
219 },
220 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
221 runtimeMode: "full-access",
222 createdAt,
223 });
224 yield* engine.dispatch({
225 type: "thread.session.set",
226 commandId: CommandId.make("command-mark-stopped-binding-session-starting"),
227 threadId: stoppedBindingThreadId,
228 session: {
229 threadId: stoppedBindingThreadId,
230 status: "starting",
231 providerName: "codex",
232 providerInstanceId,
233 runtimeMode: "full-access",
234 activeTurnId: null,
235 lastError: null,
236 updatedAt: createdAt,
237 },
238 createdAt,
239 });
240 yield* directory.upsert({
241 threadId: stoppedBindingThreadId,
242 provider: ProviderDriverKind.make("codex"),
243 providerInstanceId,
244 status: "stopped",
245 resumeCursor: stoppedBindingResumeCursor,
246 runtimePayload: { activeTurnId: "stale", unrelated: "also-preserve-me" },
247 runtimeMode: "full-access",
248 });
249 }).pipe(Effect.provide(firstRuntime));
250
251 const secondRuntime = makePersistedRuntimeLayer(config.dbPath);
252 const startupLayer = ServerRuntimeStartup.layer.pipe(
253 Layer.provideMerge(secondRuntime),
254 Layer.provideMerge(startupDependencies),
255 );
256
257 const result = yield* Effect.gen(function* () {
258 const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup;
259 const engine = yield* OrchestrationEngine.OrchestrationEngineService;
260 const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
261 const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
262 const sql = yield* SqlClient.SqlClient;
263
264 yield* startup.markHttpListening;
265 yield* startup.awaitCommandReady;
266
267 const restartedThread = Option.getOrThrow(yield* query.getThreadDetailById(threadId));
268 const restartedStoppedBindingThread = Option.getOrThrow(
269 yield* query.getThreadDetailById(stoppedBindingThreadId),
270 );
271 const pendingRows = yield* sql<{ readonly threadId: string }>`
272 SELECT thread_id AS "threadId"
273 FROM projection_turns
274 WHERE thread_id IN (${threadId}, ${stoppedBindingThreadId})
275 AND turn_id IS NULL
276 AND state = 'pending'
277 `;
278 const settleExit = yield* Effect.exit(
279 engine.dispatch({
280 type: "thread.settle",
281 commandId: CommandId.make("command-settle-after-restart"),
282 threadId,
283 }),
284 );
285 const snoozeExit = yield* Effect.exit(
286 engine.dispatch({
287 type: "thread.snooze",
288 commandId: CommandId.make("command-snooze-after-restart"),
289 threadId,
290 snoozedUntil: DateTime.formatIso(DateTime.add(now, { hours: 1 })),
291 }),
292 );
293 const newTurnExit = yield* Effect.exit(
294 engine.dispatch({
295 type: "thread.turn.start",
296 commandId: CommandId.make("command-new-turn-after-restart"),
297 threadId,
298 message: {
299 messageId: MessageId.make("message-new-turn-after-restart"),
300 role: "user",
301 text: "Continue immediately after restart",
302 attachments: [],
303 },
304 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
305 runtimeMode: "full-access",
306 createdAt,
307 }),
308 );
309 const binding = Option.getOrThrow(yield* directory.getBinding(threadId));
310 const stoppedBinding = Option.getOrThrow(
311 yield* directory.getBinding(stoppedBindingThreadId),
312 );
313
314 return {
315 sessionStatus: restartedThread.session?.status,
316 activeTurnId: restartedThread.session?.activeTurnId,
317 latestTurn: restartedThread.latestTurn,
318 pendingTurnCount: pendingRows.length,
319 settleSucceeded: Exit.isSuccess(settleExit),
320 snoozeSucceeded: Exit.isSuccess(snoozeExit),
321 newTurnSucceeded: Exit.isSuccess(newTurnExit),
322 bindingStatus: binding.status,
323 resumeCursor: binding.resumeCursor,
324 runtimePayload: binding.runtimePayload,
325 stoppedBindingSessionStatus: restartedStoppedBindingThread.session?.status,
326 stoppedBindingStatus: stoppedBinding.status,
327 stoppedBindingResumeCursor: stoppedBinding.resumeCursor,
328 stoppedBindingRuntimePayload: stoppedBinding.runtimePayload,
329 };
330 }).pipe(Effect.provide(startupLayer));
331
332 assert.deepStrictEqual(result, {
333 sessionStatus: "error",
334 activeTurnId: null,
335 latestTurn: null,
336 pendingTurnCount: 0,
337 settleSucceeded: true,
338 snoozeSucceeded: true,
339 newTurnSucceeded: true,
340 bindingStatus: "stopped",
341 resumeCursor,
342 runtimePayload: { activeTurnId: null, unrelated: "preserve-me" },
343 stoppedBindingSessionStatus: "error",
344 stoppedBindingStatus: "stopped",
345 stoppedBindingResumeCursor,
346 stoppedBindingRuntimePayload: {
347 activeTurnId: null,
348 unrelated: "also-preserve-me",
349 },
350 });
351 }).pipe(
352 Effect.provide(
353 ServerConfig.layerTest(process.cwd(), {
354 prefix: "t3-orphaned-provider-session-startup-",
355 }).pipe(Layer.provideMerge(NodeServices.layer)),
356 ),
357 ),
358);
apps/server/src/serverRuntimeStartup.reconcile.test.ts created+296
......@@ -0,0 +1,296 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import {
3 type OrchestrationCommand,
4 ProviderDriverKind,
5 ProviderInstanceId,
6 ThreadId,
7 TurnId,
8} from "@t3tools/contracts";
9import { assert, it } from "@effect/vitest";
10import * as Effect from "effect/Effect";
11import * as Option from "effect/Option";
12import * as Stream from "effect/Stream";
13
14import { OrchestrationCommandInvariantError } from "./orchestration/Errors.ts";
15import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts";
16import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";
17import { ProviderSessionDirectoryPersistenceError } from "./provider/Errors.ts";
18import * as ProviderService from "./provider/Services/ProviderService.ts";
19import * as ProviderSessionDirectory from "./provider/Services/ProviderSessionDirectory.ts";
20import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts";
21
22const providerInstanceId = ProviderInstanceId.make("codex");
23const updatedAt = "2026-08-20T12:00:00.000Z";
24
25const makeThread = (
26 id: string,
27 status: "starting" | "running" | "ready" | "stopped" | "error",
28 activeTurnId: TurnId | null = null,
29 archivedAt: string | null = null,
30) => ({
31 id: ThreadId.make(id),
32 archivedAt,
33 deletedAt: null,
34 session: {
35 threadId: ThreadId.make(id),
36 status,
37 providerName: "codex" as const,
38 providerInstanceId,
39 runtimeMode: "full-access" as const,
40 activeTurnId,
41 lastError: null,
42 updatedAt,
43 },
44});
45
46const makeProviderService = (liveThreadIds: ReadonlyArray<ThreadId> = []) =>
47 ({
48 startSession: () => Effect.die("unused"),
49 sendTurn: () => Effect.die("unused"),
50 interruptTurn: () => Effect.die("unused"),
51 respondToRequest: () => Effect.die("unused"),
52 respondToUserInput: () => Effect.die("unused"),
53 stopSession: () => Effect.die("unused"),
54 listSessions: () => Effect.succeed(liveThreadIds.map((threadId) => ({ threadId }) as never)),
55 getCapabilities: () => Effect.die("unused"),
56 getInstanceInfo: () => Effect.die("unused"),
57 rollbackConversation: () => Effect.die("unused"),
58 streamEvents: Stream.empty,
59 }) satisfies ProviderService.ProviderService["Service"];
60
61const queryWithThreads = (threads: ReadonlyArray<ReturnType<typeof makeThread>>) =>
62 ({
63 getCommandReadModel: () => Effect.succeed({ threads } as never),
64 }) as unknown as ProjectionSnapshotQuery.ProjectionSnapshotQuery["Service"];
65
66const runReconciliation = (input: {
67 readonly threads: ReadonlyArray<ReturnType<typeof makeThread>>;
68 readonly liveThreadIds?: ReadonlyArray<ThreadId>;
69 readonly directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"];
70 readonly dispatch: OrchestrationEngine.OrchestrationEngineService["Service"]["dispatch"];
71}) =>
72 ServerRuntimeStartup.reconcileProviderSessions.pipe(
73 Effect.provideService(
74 ProjectionSnapshotQuery.ProjectionSnapshotQuery,
75 queryWithThreads(input.threads),
76 ),
77 Effect.provideService(
78 ProviderService.ProviderService,
79 makeProviderService(input.liveThreadIds),
80 ),
81 Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, input.directory),
82 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
83 readEvents: () => Stream.empty,
84 dispatch: input.dispatch,
85 streamDomainEvents: Stream.empty,
86 latestSequence: Effect.succeed(0),
87 }),
88 Effect.provide(NodeServices.layer),
89 );
90
91it.effect("reconciles multiple active and archived orphans but skips live sessions", () => {
92 const starting = makeThread("thread-starting", "starting");
93 const running = makeThread("thread-running", "running", TurnId.make("turn-running"));
94 const staleActiveTurn = makeThread(
95 "thread-stale-active-turn",
96 "ready",
97 TurnId.make("turn-stale-active"),
98 );
99 const archived = makeThread(
100 "thread-archived",
101 "running",
102 TurnId.make("turn-archived"),
103 updatedAt,
104 );
105 const live = makeThread("thread-live", "running", TurnId.make("turn-live"));
106 const settled = makeThread("thread-ready", "ready");
107 const dispatched: OrchestrationCommand[] = [];
108 const bindingReads: ThreadId[] = [];
109 const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = [];
110
111 return runReconciliation({
112 threads: [starting, running, staleActiveTurn, archived, live, settled],
113 liveThreadIds: [live.id],
114 directory: {
115 getBinding: (candidate) =>
116 Effect.sync(() => bindingReads.push(candidate)).pipe(
117 Effect.as(
118 Option.some({
119 threadId: candidate,
120 provider: ProviderDriverKind.make("codex"),
121 providerInstanceId,
122 status: "running" as const,
123 resumeCursor: { cursor: candidate },
124 runtimePayload: { activeTurnId: "stale", unrelated: candidate },
125 }),
126 ),
127 ),
128 upsert: (binding) => Effect.sync(() => upserts.push(binding)),
129 getProvider: () => Effect.die("unused"),
130 listThreadIds: () => Effect.die("unused"),
131 listBindings: () => Effect.die("unused"),
132 },
133 dispatch: (command) =>
134 Effect.sync(() => dispatched.push(command)).pipe(Effect.as({ sequence: dispatched.length })),
135 }).pipe(
136 Effect.tap(() =>
137 Effect.sync(() => {
138 const orphanIds = [starting.id, running.id, staleActiveTurn.id, archived.id];
139 assert.deepStrictEqual(bindingReads, orphanIds);
140 assert.deepStrictEqual(
141 dispatched.map((command) => command.type === "thread.session.set" && command.threadId),
142 orphanIds,
143 );
144 assert.deepStrictEqual(
145 dispatched.map((command) =>
146 command.type === "thread.session.set"
147 ? {
148 status: command.session.status,
149 activeTurnId: command.session.activeTurnId,
150 }
151 : null,
152 ),
153 orphanIds.map(() => ({ status: "error" as const, activeTurnId: null })),
154 );
155 assert.equal(upserts.length, orphanIds.length);
156 for (const binding of upserts) {
157 assert.equal(binding.status, "stopped");
158 assert.deepStrictEqual(binding.runtimePayload, { activeTurnId: null });
159 assert.deepStrictEqual(binding.resumeCursor, { cursor: binding.threadId });
160 }
161 }),
162 ),
163 );
164});
165
166it.effect(
167 "settles projections when directory bindings are absent, corrupt, or fail to upsert",
168 () => {
169 const absent = makeThread("thread-binding-absent", "starting");
170 const corrupt = makeThread("thread-binding-corrupt", "running");
171 const upsertFailure = makeThread("thread-binding-upsert-failure", "running");
172 const dispatched: OrchestrationCommand[] = [];
173 const corruptFailure = new ProviderSessionDirectoryPersistenceError({
174 operation: "ProviderSessionDirectory.getBinding",
175 detail: "corrupt persisted binding",
176 });
177 const writeFailure = new ProviderSessionDirectoryPersistenceError({
178 operation: "ProviderSessionDirectory.upsert",
179 detail: "failed binding write",
180 });
181
182 return runReconciliation({
183 threads: [absent, corrupt, upsertFailure],
184 directory: {
185 getBinding: (candidate) =>
186 candidate === absent.id
187 ? Effect.succeed(Option.none())
188 : candidate === corrupt.id
189 ? Effect.fail(corruptFailure)
190 : Effect.succeed(
191 Option.some({
192 threadId: candidate,
193 provider: ProviderDriverKind.make("codex"),
194 providerInstanceId,
195 }),
196 ),
197 upsert: () => Effect.fail(writeFailure),
198 getProvider: () => Effect.die("unused"),
199 listThreadIds: () => Effect.die("unused"),
200 listBindings: () => Effect.die("unused"),
201 },
202 dispatch: (command) =>
203 Effect.sync(() => dispatched.push(command)).pipe(
204 Effect.as({ sequence: dispatched.length }),
205 ),
206 }).pipe(
207 Effect.tap(() =>
208 Effect.sync(() => {
209 assert.deepStrictEqual(
210 dispatched.map((command) => command.type === "thread.session.set" && command.threadId),
211 [absent.id, corrupt.id, upsertFailure.id],
212 );
213 }),
214 ),
215 );
216 },
217);
218
219it.effect("retries failed projections and continues after a persistent failure", () => {
220 const transient = makeThread("thread-dispatch-transient-failure", "running");
221 const persistent = makeThread("thread-dispatch-persistent-failure", "running");
222 const later = makeThread("thread-dispatch-success", "running");
223 const attempted: ThreadId[] = [];
224 let transientAttempts = 0;
225 const failure = new OrchestrationCommandInvariantError({
226 commandType: "thread.session.set",
227 detail: "simulated startup reconciliation failure",
228 });
229
230 return runReconciliation({
231 threads: [transient, persistent, later],
232 directory: {
233 getBinding: () => Effect.succeed(Option.none()),
234 upsert: () => Effect.void,
235 getProvider: () => Effect.die("unused"),
236 listThreadIds: () => Effect.die("unused"),
237 listBindings: () => Effect.die("unused"),
238 },
239 dispatch: (command) => {
240 if (command.type !== "thread.session.set") {
241 return Effect.die("unexpected command");
242 }
243 attempted.push(command.threadId);
244 if (command.threadId === transient.id && transientAttempts++ === 0) {
245 return Effect.fail(failure);
246 }
247 return command.threadId === persistent.id
248 ? Effect.fail(failure)
249 : Effect.succeed({ sequence: attempted.length });
250 },
251 }).pipe(
252 Effect.tap(() =>
253 Effect.sync(() =>
254 assert.deepStrictEqual(attempted, [
255 transient.id,
256 transient.id,
257 persistent.id,
258 persistent.id,
259 later.id,
260 ]),
261 ),
262 ),
263 );
264});
265
266it.effect("does not fail startup when the live provider session inventory cannot be read", () => {
267 let queried = false;
268 return ServerRuntimeStartup.reconcileProviderSessions.pipe(
269 Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
270 getCommandReadModel: () =>
271 Effect.sync(() => {
272 queried = true;
273 return { threads: [] } as never;
274 }),
275 } as unknown as ProjectionSnapshotQuery.ProjectionSnapshotQuery["Service"]),
276 Effect.provideService(ProviderService.ProviderService, {
277 ...makeProviderService(),
278 listSessions: () => Effect.die("provider inventory unavailable"),
279 }),
280 Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, {
281 getBinding: () => Effect.die("unused"),
282 upsert: () => Effect.die("unused"),
283 getProvider: () => Effect.die("unused"),
284 listThreadIds: () => Effect.die("unused"),
285 listBindings: () => Effect.die("unused"),
286 }),
287 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
288 readEvents: () => Stream.empty,
289 dispatch: () => Effect.die("unused"),
290 streamDomainEvents: Stream.empty,
291 latestSequence: Effect.succeed(0),
292 }),
293 Effect.provide(NodeServices.layer),
294 Effect.tap(() => Effect.sync(() => assert.equal(queried, false))),
295 );
296});
apps/server/src/serverRuntimeStartup.ts+88
......@@ -7,6 +7,7 @@ import {
77 ProviderInstanceId,
88 ThreadId,
99} from "@t3tools/contracts";
10import * as Cause from "effect/Cause";
1011import * as Console from "effect/Console";
1112import * as Context from "effect/Context";
1213import * as Crypto from "effect/Crypto";
......@@ -33,6 +34,8 @@ import * as ServerSettings from "./serverSettings.ts";
3334import * as AnalyticsService from "./telemetry/AnalyticsService.ts";
3435import * as ServerEnvironment from "./environment/ServerEnvironment.ts";
3536import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";
37import * as ProviderService from "./provider/Services/ProviderService.ts";
38import * as ProviderSessionDirectory from "./provider/Services/ProviderSessionDirectory.ts";
3639import * as ProviderSessionReaper from "./provider/Services/ProviderSessionReaper.ts";
3740import { forkParked } from "./serverActivation.ts";
3841import * as ServiceLauncherClient from "./cloud/serviceLauncherClient.ts";
......@@ -290,6 +293,89 @@ const runStartupPhase = <A, E, R>(phase: string, effect: Effect.Effect<A, E, R>)
290293 Effect.withSpan(`server.startup.${phase}`),
291294 );
292295
296const ORPHANED_PROVIDER_SESSION_ERROR =
297 "Provider session did not survive a server restart. Send a new message to continue.";
298
299export const reconcileProviderSessions = Effect.gen(function* () {
300 const crypto = yield* Crypto.Crypto;
301 const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
302 const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService;
303 const providerService = yield* ProviderService.ProviderService;
304 const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
305
306 const liveThreadIds = new Set(
307 (yield* providerService.listSessions()).map((session) => session.threadId),
308 );
309 const { threads } = yield* query.getCommandReadModel();
310 const orphanedThreads = threads.filter(
311 (thread) =>
312 thread.session !== null &&
313 (thread.session.status === "starting" ||
314 thread.session.status === "running" ||
315 thread.session.activeTurnId !== null) &&
316 !liveThreadIds.has(thread.id),
317 );
318
319 for (const thread of orphanedThreads) {
320 const session = thread.session;
321 if (session === null) {
322 continue;
323 }
324 yield* Effect.gen(function* () {
325 const binding = yield* directory.getBinding(thread.id);
326 if (Option.isSome(binding)) {
327 yield* directory.upsert({
328 ...binding.value,
329 status: "stopped",
330 runtimePayload: { activeTurnId: null },
331 });
332 }
333 }).pipe(
334 Effect.catchCause((cause) =>
335 Cause.hasInterrupts(cause)
336 ? Effect.failCause(cause)
337 : Effect.logWarning("failed to reconcile orphaned provider session directory binding", {
338 threadId: thread.id,
339 cause,
340 }),
341 ),
342 );
343
344 yield* Effect.gen(function* () {
345 const reconciledAt = DateTime.formatIso(yield* DateTime.now);
346 yield* orchestrationEngine.dispatch({
347 type: "thread.session.set",
348 commandId: CommandId.make(yield* crypto.randomUUIDv4),
349 threadId: thread.id,
350 session: {
351 ...session,
352 status: "error",
353 activeTurnId: null,
354 lastError: ORPHANED_PROVIDER_SESSION_ERROR,
355 updatedAt: reconciledAt,
356 },
357 createdAt: reconciledAt,
358 });
359 }).pipe(
360 Effect.retry({ times: 1 }),
361 Effect.catchCause((cause) =>
362 Cause.hasInterrupts(cause)
363 ? Effect.failCause(cause)
364 : Effect.logWarning("failed to settle orphaned provider session projection", {
365 threadId: thread.id,
366 cause,
367 }),
368 ),
369 );
370 }
371}).pipe(
372 Effect.catchCause((cause) =>
373 Cause.hasInterrupts(cause)
374 ? Effect.failCause(cause)
375 : Effect.logWarning("provider session startup reconciliation failed", { cause }),
376 ),
377);
378
293379interface StartupOptions {
294380 readonly activate?: Effect.Effect<void>;
295381 readonly awaitAuxiliaryParked?: Effect.Effect<void>;
......@@ -354,6 +440,8 @@ export const make = (options?: StartupOptions) =>
354440 }),
355441 );
356442
443 yield* runStartupPhase("provider-sessions.reconcile", reconcileProviderSessions);
444
357445 const welcomeBase = yield* resolveWelcomeBase;
358446 const environment = yield* serverEnvironment.getDescriptor;
359447 yield* Effect.logDebug("startup phase: preparing welcome payload");