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