1import { assert, it } from "@effect/vitest";
2import {
3 type ModelSelection,
4 NodeId,
5 type OrchestrationV2ProviderThread,
6 type OrchestrationV2ThreadProjection,
7 ProjectId,
8 ProviderDriverKind,
9 ProviderInstanceId,
10 ProviderSessionId,
11 ProviderThreadId,
12 ProviderTurnId,
13 RunAttemptId,
14 RunId,
15 ThreadId,
16} from "@t3tools/contracts";
17import * as DateTime from "effect/DateTime";
18import * as Effect from "effect/Effect";
19import * as Exit from "effect/Exit";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as Ref from "effect/Ref";
23import * as Stream from "effect/Stream";
24
25import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
26import * as ProjectionStore from "./ProjectionStore.ts";
27import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
28import * as ProviderSessionManager from "./ProviderSessionManager.ts";
29import * as ProviderTurnControlService from "./ProviderTurnControlService.ts";
30
31const driver = ProviderDriverKind.make("codex");
32const providerInstanceId = ProviderInstanceId.make("codex");
33const modelSelection = {
34 instanceId: providerInstanceId,
35 model: "gpt-5.4",
36} satisfies ModelSelection;
37
38function makeProjection(input: {
39 readonly now: DateTime.Utc;
40 readonly threadId: ThreadId;
41 readonly providerThread: OrchestrationV2ProviderThread;
42 readonly providerTurnId: ProviderTurnId;
43 readonly attemptId: RunAttemptId;
44}): OrchestrationV2ThreadProjection {
45 const runId = RunId.make("run:restart-session");
46 const nodeId = NodeId.make("node:restart-session");
47 return {
48 thread: {
49 createdBy: "user",
50 creationSource: "web",
51 id: input.threadId,
52 projectId: ProjectId.make("project:restart-session"),
53 title: "Restart session",
54 providerInstanceId,
55 modelSelection,
56 runtimeMode: "full-access",
57 interactionMode: "default",
58 branch: null,
59 worktreePath: "/workspace",
60 activeProviderThreadId: input.providerThread.id,
61 lineage: {
62 parentThreadId: null,
63 relationshipToParent: null,
64 rootThreadId: input.threadId,
65 },
66 forkedFrom: null,
67 createdAt: input.now,
68 updatedAt: input.now,
69 archivedAt: null,
70 settledOverride: null,
71 settledAt: null,
72 lastVisitedAt: null,
73 deletedAt: null,
74 },
75 runs: [],
76 attempts: [
77 {
78 id: input.attemptId,
79 runId,
80 attemptOrdinal: 1,
81 rootNodeId: nodeId,
82 providerInstanceId,
83 providerThreadId: input.providerThread.id,
84 providerTurnId: input.providerTurnId,
85 reason: "initial",
86 status: "superseded",
87 startedAt: input.now,
88 completedAt: input.now,
89 },
90 ],
91 nodes: [],
92 subagents: [],
93 providerSessions: [],
94 providerThreads: [input.providerThread],
95 providerTurns: [
96 {
97 id: input.providerTurnId,
98 providerThreadId: input.providerThread.id,
99 nodeId,
100 runAttemptId: input.attemptId,
101 nativeTurnRef: {
102 driver,
103 nativeId: "native-turn:restart-session",
104 strength: "strong",
105 },
106 ordinal: 1,
107 status: "running",
108 startedAt: input.now,
109 completedAt: null,
110 },
111 ],
112 runtimeRequests: [],
113 messages: [],
114 plans: [],
115 turnItems: [],
116 checkpointScopes: [],
117 checkpoints: [],
118 contextHandoffs: [],
119 contextTransfers: [],
120 visibleTurnItems: [],
121 updatedAt: input.now,
122 };
123}
124
125it.effect(
126 "interrupts the historical session only for the exact committed restart replacement",
127 () =>
128 Effect.gen(function* () {
129 const now = yield* DateTime.now;
130 const threadId = ThreadId.make("thread:restart-session");
131 const oldSessionId = ProviderSessionId.make("provider-session:restart-session:old");
132 const replacementSessionId = ProviderSessionId.make(
133 "provider-session:restart-session:replacement",
134 );
135 const unrelatedSessionId = ProviderSessionId.make(
136 "provider-session:restart-session:unrelated",
137 );
138 const providerThreadId = ProviderThreadId.make("provider-thread:restart-session");
139 const providerTurnId = ProviderTurnId.make("provider-turn:restart-session");
140 const attemptId = RunAttemptId.make("run-attempt:restart-session");
141 const providerThread: OrchestrationV2ProviderThread = {
142 id: providerThreadId,
143 driver,
144 providerInstanceId,
145 providerSessionId: replacementSessionId,
146 appThreadId: threadId,
147 ownerNodeId: null,
148 nativeThreadRef: {
149 driver,
150 nativeId: "native-thread:restart-session",
151 strength: "strong",
152 },
153 nativeConversationHeadRef: null,
154 status: "not_loaded",
155 firstRunOrdinal: 1,
156 lastRunOrdinal: 1,
157 handoffIds: [],
158 forkedFrom: null,
159 createdAt: now,
160 updatedAt: now,
161 };
162 const projection = yield* Ref.make(
163 makeProjection({ now, threadId, providerThread, providerTurnId, attemptId }),
164 );
165 const interruptedThread = yield* Ref.make<OrchestrationV2ProviderThread | null>(null);
166 const providerSession = {
167 id: oldSessionId,
168 driver,
169 providerInstanceId,
170 status: "running" as const,
171 cwd: "/workspace",
172 model: modelSelection.model,
173 capabilities: CodexProviderCapabilitiesV2,
174 createdAt: now,
175 updatedAt: now,
176 lastError: null,
177 };
178 const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = {
179 instanceId: providerInstanceId,
180 driver,
181 providerSessionId: oldSessionId,
182 providerSession,
183 events: Stream.empty,
184 ensureThread: () => Effect.die("unused ensureThread"),
185 resumeThread: () => Effect.die("unused resumeThread"),
186 startTurn: () => Effect.die("unused startTurn"),
187 steerTurn: () => Effect.die("unused steerTurn"),
188 interruptTurn: ({ providerThread: target }) =>
189 Effect.all(
190 [
191 Ref.set(interruptedThread, target),
192 Ref.update(projection, (current) => ({
193 ...current,
194 providerTurns: current.providerTurns.map((turn) =>
195 turn.id === providerTurnId
196 ? { ...turn, status: "interrupted" as const, completedAt: now }
197 : turn,
198 ),
199 })),
200 ],
201 { discard: true },
202 ),
203 respondToRuntimeRequest: () => Effect.die("unused respondToRuntimeRequest"),
204 readThreadSnapshot: () => Effect.die("unused readThreadSnapshot"),
205 rollbackThread: () => Effect.die("unused rollbackThread"),
206 forkThread: () => Effect.die("unused forkThread"),
207 };
208 const layerProjection = Layer.succeed(
209 ProjectionStore.ProjectionStoreV2,
210 ProjectionStore.ProjectionStoreV2.of({
211 apply: () => Effect.void,
212 carryProviderTurnContextUsage: () => Effect.die("unused carryProviderTurnContextUsage"),
213 getLimitRecoveryCandidates: () => Effect.die("unused getLimitRecoveryCandidates"),
214 getLimitRecoveryWakeAt: () => Effect.die("unused getLimitRecoveryWakeAt"),
215 getOpenRemoteDelegatedTasks: Effect.die("unused getOpenRemoteDelegatedTasks"),
216 getAgentAwarenessSnapshot: () => Effect.die("unused awareness snapshot"),
217 getShellSnapshot: () => Effect.die("unused getShellSnapshot"),
218 listActiveSubagentThreadIds: () => Effect.die("unused listActiveSubagentThreadIds"),
219 countThreads: () => Effect.die("unused countThreads"),
220 countArchivedThreadMessages: () => Effect.die("unused countArchivedThreadMessages"),
221 listActiveShellIndex: () => Effect.die("unused listActiveShellIndex"),
222 listOlderSettledThreads: () => Effect.die("unused listOlderSettledThreads"),
223 readShellSnapshot: () => Effect.die("unused readShellSnapshot"),
224 getThreadShell: () => Effect.die("unused getThreadShell"),
225 getThreadTitles: () => Effect.die("unused getThreadTitles"),
226 getThreadLinkMetadata: () => Effect.die("unused getThreadLinkMetadata"),
227 findGuestShareThreadId: () => Effect.die("unused findGuestShareThreadId"),
228 getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)),
229 getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"),
230 getThreadsWithPullRequests: () => Effect.die("unused getThreadsWithPullRequests"),
231 getThreadWorktreePaths: () => Effect.die("unused getThreadWorktreePaths"),
232 getThreadProjection: () => Effect.die("control effects must not load transcript"),
233 getTurnStartContext: () => Effect.die("unused"),
234 getTurnStartHistory: () => Effect.die("unused"),
235 getRuntimeRecoveryProjection: () => Effect.die("unused getRuntimeRecoveryProjection"),
236 getPlan: () => Effect.die("unused"),
237 hasUnpairedRunInterruptRequest: () => Effect.die("unused interrupt read"),
238 getThreadAttachmentIds: () => Effect.die("Unused attachment lookup"),
239 searchThread: () => Effect.die("unused"),
240 searchThreadStream: () => Stream.empty,
241 getThreadHistoryPage: () => Effect.die("unused"),
242 getTimelinePage: () => Effect.die("Unused timeline read"),
243 getMessageCount: () => Effect.die("unused message count"),
244 getTranscriptByteFloor: () => Effect.die("unused transcript byte floor"),
245 hasPendingDelegatedCompletion: () => Effect.die("unused hasPendingDelegatedCompletion"),
246 getSubagentCount: () => Effect.die("unused subagent count"),
247 listAppOwnedChildThreadIds: () => Effect.die("unused listAppOwnedChildThreadIds"),
248 getNextTurnItemOrdinal: () => Effect.die("unused ordinal read"),
249 getTurnItem: () => Effect.die("unused turn item read"),
250 getMessagesById: () => Effect.die("unused getMessagesById"),
251 getThreadRecords: () => Effect.die("unused record read"),
252 getRuntimeRequest: () => Effect.die("unused getRuntimeRequest"),
253 getRunningTurnContext: () => Effect.die("unused getRunningTurnContext"),
254 getThreadProviderContext: () => Effect.die("unused getThreadProviderContext"),
255 getRuntimeResponseContext: () => Effect.die("unused getRuntimeResponseContext"),
256 getPendingNativeUserInputs: () => Effect.die("unused getPendingNativeUserInputs"),
257 getProviderControlContext: (_threadId, target) =>
258 Ref.get(projection).pipe(
259 Effect.map((current) => ({
260 providerThread: current.providerThreads.find(
261 (thread) => thread.id === target.providerThreadId,
262 ),
263 providerTurn: current.providerTurns.find(
264 (turn) => turn.id === target.providerTurnId,
265 ),
266 attempt: current.attempts.find((attempt) => attempt.id === target.attemptId),
267 message: undefined,
268 run: undefined,
269 })),
270 ),
271 getCheckpointContext: () => Effect.die("not used"),
272 getCheckpointRollbackContext: () => Effect.die("unused rollback context"),
273 getCheckpointCaptureContext: () => Effect.die("not used"),
274 getRunMessage: () => Effect.die("not used"),
275 getRun: () => Effect.die("not used"),
276 isNodeStored: () => Effect.die("not used"),
277 canStartQueuedRun: () => Effect.die("not used"),
278 getHeldQueuedRuns: () => Effect.die("not used"),
279 getRecoveryThreadIds: () => Effect.die("unused getRecoveryThreadIds"),
280 getUnreadableThreadIds: () => Effect.die("unused getUnreadableThreadIds"),
281 getThreadSnapshot: () => Effect.die("unused getThreadSnapshot"),
282 getThreadSnapshotWindow: () => Effect.die("unused getThreadSnapshotWindow"),
283 getPullRequestSnapshot: () => Effect.die("unused getPullRequestSnapshot"),
284 }),
285 );
286 const layerSessionManager = Layer.succeed(
287 ProviderSessionManager.ProviderSessionManagerV2,
288 ProviderSessionManager.ProviderSessionManagerV2.of({
289 shutdown: Effect.void,
290 open: () => Effect.die("unused open"),
291 get: (providerSessionId) =>
292 Effect.succeed(
293 providerSessionId === oldSessionId ? Option.some(runtime) : Option.none(),
294 ),
295 isLive: (providerSessionId) => Effect.succeed(providerSessionId === oldSessionId),
296 close: () => Effect.void,
297 closeInstance: () => Effect.void,
298 release: () => Effect.void,
299 detach: () => Effect.void,
300 otherWorkingThreads: () => Effect.succeed([]),
301 releaseUnlessBusy: () => Effect.succeed(true),
302 subscribePromptSuggestion: () => Stream.empty,
303 }),
304 );
305 const layerControl = ProviderTurnControlService.layer.pipe(
306 Layer.provide(Layer.merge(layerProjection, layerSessionManager)),
307 );
308
309 const [ordinaryInterrupt, unrelatedRestart] = yield* Effect.gen(function* () {
310 const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2;
311 const ordinary = yield* Effect.exit(
312 control.interrupt({
313 threadId,
314 providerSessionId: oldSessionId,
315 providerThreadId,
316 providerTurnId,
317 }),
318 );
319 const unrelated = yield* Effect.exit(
320 control.interruptAndAwaitTerminal({
321 threadId,
322 providerSessionId: oldSessionId,
323 replacementProviderSessionId: unrelatedSessionId,
324 providerThreadId,
325 providerTurnId,
326 interruptedAttemptId: attemptId,
327 }),
328 );
329 return [ordinary, unrelated] as const;
330 }).pipe(Effect.provide(layerControl));
331
332 assert.isTrue(Exit.isFailure(ordinaryInterrupt));
333 assert.isTrue(Exit.isFailure(unrelatedRestart));
334 assert.isNull(yield* Ref.get(interruptedThread));
335
336 yield* Effect.gen(function* () {
337 const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2;
338 yield* control.interruptAndAwaitTerminal({
339 threadId,
340 providerSessionId: oldSessionId,
341 replacementProviderSessionId: replacementSessionId,
342 providerThreadId,
343 providerTurnId,
344 interruptedAttemptId: attemptId,
345 });
346 }).pipe(Effect.provide(layerControl));
347
348 const interrupted = yield* Ref.get(interruptedThread);
349 assert.isNotNull(interrupted);
350 assert.equal(interrupted?.providerSessionId, oldSessionId);
351 assert.equal(interrupted?.id, providerThreadId);
352 assert.equal(interrupted?.nativeThreadRef?.nativeId, "native-thread:restart-session");
353 }),
354);