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 // The restart command has already projected this replacement binding
146 // before the process-bound restart effect executes.
147 providerSessionId: replacementSessionId,
148 appThreadId: threadId,
149 ownerNodeId: null,
150 nativeThreadRef: {
151 driver,
152 nativeId: "native-thread:restart-session",
153 strength: "strong",
154 },
155 nativeConversationHeadRef: null,
156 status: "not_loaded",
157 firstRunOrdinal: 1,
158 lastRunOrdinal: 1,
159 handoffIds: [],
160 forkedFrom: null,
161 createdAt: now,
162 updatedAt: now,
163 };
164 const projection = yield* Ref.make(
165 makeProjection({ now, threadId, providerThread, providerTurnId, attemptId }),
166 );
167 const interruptedThread = yield* Ref.make<OrchestrationV2ProviderThread | null>(null);
168 const providerSession = {
169 id: oldSessionId,
170 driver,
171 providerInstanceId,
172 status: "running" as const,
173 cwd: "/workspace",
174 model: modelSelection.model,
175 capabilities: CodexProviderCapabilitiesV2,
176 createdAt: now,
177 updatedAt: now,
178 lastError: null,
179 };
180 const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = {
181 instanceId: providerInstanceId,
182 driver,
183 providerSessionId: oldSessionId,
184 providerSession,
185 events: Stream.empty,
186 ensureThread: () => Effect.die("unused ensureThread"),
187 resumeThread: () => Effect.die("unused resumeThread"),
188 startTurn: () => Effect.die("unused startTurn"),
189 steerTurn: () => Effect.die("unused steerTurn"),
190 interruptTurn: ({ providerThread: target }) =>
191 Effect.all(
192 [
193 Ref.set(interruptedThread, target),
194 Ref.update(projection, (current) => ({
195 ...current,
196 providerTurns: current.providerTurns.map((turn) =>
197 turn.id === providerTurnId
198 ? { ...turn, status: "interrupted" as const, completedAt: now }
199 : turn,
200 ),
201 })),
202 ],
203 { discard: true },
204 ),
205 respondToRuntimeRequest: () => Effect.die("unused respondToRuntimeRequest"),
206 readThreadSnapshot: () => Effect.die("unused readThreadSnapshot"),
207 rollbackThread: () => Effect.die("unused rollbackThread"),
208 forkThread: () => Effect.die("unused forkThread"),
209 };
210 const layerProjection = Layer.succeed(
211 ProjectionStore.ProjectionStoreV2,
212 ProjectionStore.ProjectionStoreV2.of({
213 apply: () => Effect.void,
214 getLimitRecoveryCandidates: () => Effect.die("unused getLimitRecoveryCandidates"),
215 getShellSnapshot: () => Effect.die("unused getShellSnapshot"),
216 readShellSnapshot: () => Effect.die("unused readShellSnapshot"),
217 getThreadShell: () => Effect.die("unused getThreadShell"),
218 getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)),
219 getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"),
220 getThreadsWithPullRequests: () => Effect.die("unused getThreadsWithPullRequests"),
221 getThreadWorktreePaths: () => Effect.die("unused getThreadWorktreePaths"),
222 getThreadProjection: () => Effect.die("control effects must not load transcript"),
223 getTurnStartContext: () => Effect.die("unused"),
224 getTurnStartHistory: () => Effect.die("unused"),
225 getRuntimeRecoveryProjection: () => Effect.die("unused getRuntimeRecoveryProjection"),
226 getPlan: () => Effect.die("unused"),
227 hasUnpairedRunInterruptRequest: () => Effect.die("unused interrupt read"),
228 getThreadAttachmentIds: () => Effect.die("Unused attachment lookup"),
229 searchThread: () => Effect.die("unused"),
230 searchThreadStream: () => Stream.empty,
231 getThreadHistoryPage: () => Effect.die("unused"),
232 getTimelinePage: () => Effect.die("Unused timeline read"),
233 getMessageCount: () => Effect.die("unused message count"),
234 getNextTurnItemOrdinal: () => Effect.die("unused ordinal read"),
235 getTurnItem: () => Effect.die("unused turn item read"),
236 getThreadRecords: () => Effect.die("unused record read"),
237 getRuntimeRequest: () => Effect.die("unused getRuntimeRequest"),
238 getRunningTurnContext: () => Effect.die("unused getRunningTurnContext"),
239 getThreadProviderContext: () => Effect.die("unused getThreadProviderContext"),
240 getRuntimeResponseContext: () => Effect.die("unused getRuntimeResponseContext"),
241 getPendingNativeUserInputs: () => Effect.die("unused getPendingNativeUserInputs"),
242 getProviderControlContext: (_threadId, target) =>
243 Ref.get(projection).pipe(
244 Effect.map((current) => ({
245 providerThread: current.providerThreads.find(
246 (thread) => thread.id === target.providerThreadId,
247 ),
248 providerTurn: current.providerTurns.find(
249 (turn) => turn.id === target.providerTurnId,
250 ),
251 attempt: current.attempts.find((attempt) => attempt.id === target.attemptId),
252 message: undefined,
253 run: undefined,
254 })),
255 ),
256 getCheckpointContext: () => Effect.die("not used"),
257 getCheckpointCaptureContext: () => Effect.die("not used"),
258 getRunMessage: () => Effect.die("not used"),
259 isNodeStored: () => Effect.die("not used"),
260 canStartQueuedRun: () => Effect.die("not used"),
261 getRecoveryThreadIds: () => Effect.die("unused getRecoveryThreadIds"),
262 getUnreadableThreadIds: () => Effect.die("unused getUnreadableThreadIds"),
263 getThreadSnapshot: () => Effect.die("unused getThreadSnapshot"),
264 getThreadSnapshotWindow: () => Effect.die("unused getThreadSnapshotWindow"),
265 }),
266 );
267 const layerSessionManager = Layer.succeed(
268 ProviderSessionManager.ProviderSessionManagerV2,
269 ProviderSessionManager.ProviderSessionManagerV2.of({
270 shutdown: Effect.void,
271 open: () => Effect.die("unused open"),
272 get: (providerSessionId) =>
273 Effect.succeed(
274 providerSessionId === oldSessionId ? Option.some(runtime) : Option.none(),
275 ),
276 isLive: (providerSessionId) => Effect.succeed(providerSessionId === oldSessionId),
277 close: () => Effect.void,
278 closeInstance: () => Effect.void,
279 release: () => Effect.void,
280 detach: () => Effect.void,
281 }),
282 );
283 const layerControl = ProviderTurnControlService.layer.pipe(
284 Layer.provide(Layer.merge(layerProjection, layerSessionManager)),
285 );
286
287 const [ordinaryInterrupt, unrelatedRestart] = yield* Effect.gen(function* () {
288 const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2;
289 const ordinary = yield* Effect.exit(
290 control.interrupt({
291 threadId,
292 providerSessionId: oldSessionId,
293 providerThreadId,
294 providerTurnId,
295 }),
296 );
297 const unrelated = yield* Effect.exit(
298 control.interruptAndAwaitTerminal({
299 threadId,
300 providerSessionId: oldSessionId,
301 replacementProviderSessionId: unrelatedSessionId,
302 providerThreadId,
303 providerTurnId,
304 interruptedAttemptId: attemptId,
305 }),
306 );
307 return [ordinary, unrelated] as const;
308 }).pipe(Effect.provide(layerControl));
309
310 assert.isTrue(Exit.isFailure(ordinaryInterrupt));
311 assert.isTrue(Exit.isFailure(unrelatedRestart));
312 assert.isNull(yield* Ref.get(interruptedThread));
313
314 yield* Effect.gen(function* () {
315 const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2;
316 yield* control.interruptAndAwaitTerminal({
317 threadId,
318 providerSessionId: oldSessionId,
319 replacementProviderSessionId: replacementSessionId,
320 providerThreadId,
321 providerTurnId,
322 interruptedAttemptId: attemptId,
323 });
324 }).pipe(Effect.provide(layerControl));
325
326 const interrupted = yield* Ref.get(interruptedThread);
327 assert.isNotNull(interrupted);
328 assert.equal(interrupted?.providerSessionId, oldSessionId);
329 assert.equal(interrupted?.id, providerThreadId);
330 assert.equal(interrupted?.nativeThreadRef?.nativeId, "native-thread:restart-session");
331 }),
332);