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 getLimitRecoveryCandidates: () => Effect.die("unused getLimitRecoveryCandidates"),
213 getLimitRecoveryWakeAt: () => Effect.die("unused getLimitRecoveryWakeAt"),
214 getOpenRemoteDelegatedTasks: Effect.die("unused getOpenRemoteDelegatedTasks"),
215 getAgentAwarenessSnapshot: () => Effect.die("unused awareness snapshot"),
216 getShellSnapshot: () => Effect.die("unused getShellSnapshot"),
217 listActiveSubagentThreadIds: () => Effect.die("unused listActiveSubagentThreadIds"),
218 countThreads: () => Effect.die("unused countThreads"),
219 listActiveShellIndex: () => Effect.die("unused listActiveShellIndex"),
220 listOlderSettledThreads: () => Effect.die("unused listOlderSettledThreads"),
221 readShellSnapshot: () => Effect.die("unused readShellSnapshot"),
222 getThreadShell: () => Effect.die("unused getThreadShell"),
223 getThreadTitles: () => Effect.die("unused getThreadTitles"),
224 getThreadLinkMetadata: () => Effect.die("unused getThreadLinkMetadata"),
225 findGuestShareThreadId: () => Effect.die("unused findGuestShareThreadId"),
226 getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)),
227 getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"),
228 getThreadsWithPullRequests: () => Effect.die("unused getThreadsWithPullRequests"),
229 getThreadWorktreePaths: () => Effect.die("unused getThreadWorktreePaths"),
230 getThreadProjection: () => Effect.die("control effects must not load transcript"),
231 getTurnStartContext: () => Effect.die("unused"),
232 getTurnStartHistory: () => Effect.die("unused"),
233 getRuntimeRecoveryProjection: () => Effect.die("unused getRuntimeRecoveryProjection"),
234 getPlan: () => Effect.die("unused"),
235 hasUnpairedRunInterruptRequest: () => Effect.die("unused interrupt read"),
236 getThreadAttachmentIds: () => Effect.die("Unused attachment lookup"),
237 searchThread: () => Effect.die("unused"),
238 searchThreadStream: () => Stream.empty,
239 getThreadHistoryPage: () => Effect.die("unused"),
240 getTimelinePage: () => Effect.die("Unused timeline read"),
241 getMessageCount: () => Effect.die("unused message count"),
242 getTranscriptByteFloor: () => Effect.die("unused transcript byte floor"),
243 hasPendingDelegatedCompletion: () => Effect.die("unused hasPendingDelegatedCompletion"),
244 getSubagentCount: () => Effect.die("unused subagent count"),
245 listAppOwnedChildThreadIds: () => Effect.die("unused listAppOwnedChildThreadIds"),
246 getNextTurnItemOrdinal: () => Effect.die("unused ordinal read"),
247 getTurnItem: () => Effect.die("unused turn item read"),
248 getMessagesById: () => Effect.die("unused getMessagesById"),
249 getThreadRecords: () => Effect.die("unused record read"),
250 getRuntimeRequest: () => Effect.die("unused getRuntimeRequest"),
251 getRunningTurnContext: () => Effect.die("unused getRunningTurnContext"),
252 getThreadProviderContext: () => Effect.die("unused getThreadProviderContext"),
253 getRuntimeResponseContext: () => Effect.die("unused getRuntimeResponseContext"),
254 getPendingNativeUserInputs: () => Effect.die("unused getPendingNativeUserInputs"),
255 getProviderControlContext: (_threadId, target) =>
256 Ref.get(projection).pipe(
257 Effect.map((current) => ({
258 providerThread: current.providerThreads.find(
259 (thread) => thread.id === target.providerThreadId,
260 ),
261 providerTurn: current.providerTurns.find(
262 (turn) => turn.id === target.providerTurnId,
263 ),
264 attempt: current.attempts.find((attempt) => attempt.id === target.attemptId),
265 message: undefined,
266 run: undefined,
267 })),
268 ),
269 getCheckpointContext: () => Effect.die("not used"),
270 getCheckpointRollbackContext: () => Effect.die("unused rollback context"),
271 getCheckpointCaptureContext: () => Effect.die("not used"),
272 getRunMessage: () => Effect.die("not used"),
273 getRun: () => Effect.die("not used"),
274 isNodeStored: () => Effect.die("not used"),
275 canStartQueuedRun: () => Effect.die("not used"),
276 getHeldQueuedRuns: () => Effect.die("not used"),
277 getRecoveryThreadIds: () => Effect.die("unused getRecoveryThreadIds"),
278 getUnreadableThreadIds: () => Effect.die("unused getUnreadableThreadIds"),
279 getThreadSnapshot: () => Effect.die("unused getThreadSnapshot"),
280 getThreadSnapshotWindow: () => Effect.die("unused getThreadSnapshotWindow"),
281 getPullRequestSnapshot: () => Effect.die("unused getPullRequestSnapshot"),
282 }),
283 );
284 const layerSessionManager = Layer.succeed(
285 ProviderSessionManager.ProviderSessionManagerV2,
286 ProviderSessionManager.ProviderSessionManagerV2.of({
287 shutdown: Effect.void,
288 open: () => Effect.die("unused open"),
289 get: (providerSessionId) =>
290 Effect.succeed(
291 providerSessionId === oldSessionId ? Option.some(runtime) : Option.none(),
292 ),
293 isLive: (providerSessionId) => Effect.succeed(providerSessionId === oldSessionId),
294 close: () => Effect.void,
295 closeInstance: () => Effect.void,
296 release: () => Effect.void,
297 detach: () => Effect.void,
298 otherWorkingThreads: () => Effect.succeed([]),
299 releaseUnlessBusy: () => Effect.succeed(true),
300 subscribePromptSuggestion: () => Stream.empty,
301 }),
302 );
303 const layerControl = ProviderTurnControlService.layer.pipe(
304 Layer.provide(Layer.merge(layerProjection, layerSessionManager)),
305 );
306
307 const [ordinaryInterrupt, unrelatedRestart] = yield* Effect.gen(function* () {
308 const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2;
309 const ordinary = yield* Effect.exit(
310 control.interrupt({
311 threadId,
312 providerSessionId: oldSessionId,
313 providerThreadId,
314 providerTurnId,
315 }),
316 );
317 const unrelated = yield* Effect.exit(
318 control.interruptAndAwaitTerminal({
319 threadId,
320 providerSessionId: oldSessionId,
321 replacementProviderSessionId: unrelatedSessionId,
322 providerThreadId,
323 providerTurnId,
324 interruptedAttemptId: attemptId,
325 }),
326 );
327 return [ordinary, unrelated] as const;
328 }).pipe(Effect.provide(layerControl));
329
330 assert.isTrue(Exit.isFailure(ordinaryInterrupt));
331 assert.isTrue(Exit.isFailure(unrelatedRestart));
332 assert.isNull(yield* Ref.get(interruptedThread));
333
334 yield* Effect.gen(function* () {
335 const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2;
336 yield* control.interruptAndAwaitTerminal({
337 threadId,
338 providerSessionId: oldSessionId,
339 replacementProviderSessionId: replacementSessionId,
340 providerThreadId,
341 providerTurnId,
342 interruptedAttemptId: attemptId,
343 });
344 }).pipe(Effect.provide(layerControl));
345
346 const interrupted = yield* Ref.get(interruptedThread);
347 assert.isNotNull(interrupted);
348 assert.equal(interrupted?.providerSessionId, oldSessionId);
349 assert.equal(interrupted?.id, providerThreadId);
350 assert.equal(interrupted?.nativeThreadRef?.nativeId, "native-thread:restart-session");
351 }),
352);