1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { expect, it, vi } from "vite-plus/test";
3import { it as effectIt } from "@effect/vitest";
4import {
5 CheckpointScopeId,
6 MessageId,
7 NodeId,
8 ProviderSessionId,
9 ProviderThreadId,
10 ProviderDriverKind,
11 ProviderInstanceId,
12 ProviderSetupError,
13 RunAttemptId,
14 RunId,
15 ThreadId,
16 ProjectId,
17 type OrchestrationV2ThreadProjection,
18 OrchestrationV2DomainEvent,
19} from "@t3tools/contracts";
20import * as Cause from "effect/Cause";
21import * as DateTime from "effect/DateTime";
22import * as Deferred from "effect/Deferred";
23import * as Effect from "effect/Effect";
24import * as FileSystem from "effect/FileSystem";
25import * as Fiber from "effect/Fiber";
26import * as Layer from "effect/Layer";
27import * as Option from "effect/Option";
28import * as Path from "effect/Path";
29import * as Schema from "effect/Schema";
30
31import { withWorkspaceLease } from "../workspace/workspaceLease.ts";
32import * as GitWorkflow from "../git/GitWorkflowService.ts";
33import * as ProjectService from "../project/ProjectService.ts";
34import * as ProviderAuthService from "../provider/ProviderAuthService.ts";
35import * as ContextHandoffService from "./ContextHandoffService.ts";
36import * as EventSink from "./EventSink.ts";
37import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
38import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
39import * as ProjectionStore from "./ProjectionStore.ts";
40import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
41import * as ProviderSessionManager from "./ProviderSessionManager.ts";
42import * as ProviderTurnStart from "./ProviderTurnStartService.ts";
43import * as RunExecutionService from "./RunExecutionService.ts";
44import * as RuntimePolicy from "./RuntimePolicy.ts";
45
46const isDomainEvent = Schema.is(OrchestrationV2DomainEvent);
47
48it("does not commit running state when inherited background routing cannot be read", async () => {
49 const threadId = ThreadId.make("thread_provider_turn_start_projection_failure");
50 const runId = RunId.make("run_provider_turn_start_projection_failure");
51 const attemptId = RunAttemptId.make("attempt_provider_turn_start_projection_failure");
52 const rootNodeId = NodeId.make("node_provider_turn_start_projection_failure");
53 const providerThreadId = ProviderThreadId.make(
54 "provider_thread_provider_turn_start_projection_failure",
55 );
56 const providerSessionId = ProviderSessionId.make(
57 "provider_session_provider_turn_start_projection_failure",
58 );
59 const messageId = MessageId.make("message_provider_turn_start_projection_failure");
60 const checkpointScopeId = CheckpointScopeId.make(
61 "checkpoint_scope_provider_turn_start_projection_failure",
62 );
63 const projection = {
64 thread: {
65 id: threadId,
66 projectId: ProjectId.make("project_provider_turn_start_projection_failure"),
67 branch: "feature/restore",
68 worktreePath: "/tmp/missing-provider-turn-start-worktree",
69 },
70 runs: [
71 {
72 id: runId,
73 status: "starting",
74 rootNodeId,
75 activeAttemptId: attemptId,
76 providerThreadId,
77 userMessageId: messageId,
78 ordinal: 2,
79 },
80 ],
81 nodes: [{ id: rootNodeId, checkpointScopeId }],
82 attempts: [{ id: attemptId }],
83 providerThreads: [{ id: providerThreadId, providerSessionId }],
84 messages: [{ id: messageId, text: "Continue", attachments: [] }],
85 checkpointScopes: [{ id: checkpointScopeId }],
86 contextHandoffs: [],
87 contextTransfers: [],
88 turnItems: [],
89 } as unknown as OrchestrationV2ThreadProjection;
90 let projectionReadCount = 0;
91 const writeIfRunCurrent = vi.fn(() =>
92 Effect.succeed({ committed: true, storedEvents: [] } as never),
93 );
94 const startRootRun = vi.fn(() => Effect.void);
95 const pruneWorktrees = vi.fn(() => Effect.void);
96 const createWorktree = vi.fn(() => Effect.succeed({} as never));
97 const layer = ProviderTurnStart.layer.pipe(
98 Layer.provide(
99 Layer.mergeAll(
100 Layer.mock(ContextHandoffService.ContextHandoffServiceV2)({}),
101 Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent }),
102 IdAllocator.layer,
103 Path.layer,
104 FileSystem.layerNoop({ exists: () => Effect.succeed(false) }),
105 Layer.mock(GitWorkflow.GitWorkflowService)({ pruneWorktrees, createWorktree }),
106 Layer.mock(ProjectService.ProjectService)({
107 getById: () =>
108 Effect.succeed(
109 Option.some({ workspaceRoot: "/tmp/provider-turn-start-project" } as never),
110 ),
111 }),
112 Layer.mock(ProjectionStore.ProjectionStoreV2)({
113 getThreadRecords: () => {
114 projectionReadCount += 1;
115 return Effect.succeed(projection);
116 },
117 getTurnStartContext: () => {
118 projectionReadCount += 1;
119 return Effect.succeed({
120 ...projection,
121 hasConversation: projection.messages.some(
122 (m) =>
123 m.role === "user" &&
124 (m.text.trim().toLowerCase() !== "/compact" || m.attachments.length > 0),
125 ),
126 });
127 },
128 getRuntimeRecoveryProjection: () => {
129 projectionReadCount += 1;
130 return Effect.fail(
131 new ProjectionStore.ProjectionStoreReadError({
132 threadId,
133 cause: "simulated inherited-background projection failure",
134 }),
135 );
136 },
137 }),
138 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}),
139 Layer.mock(ProviderAuthService.ProviderAuthService)({
140 tryHandlePromptCommand: () => Effect.succeed(false),
141 }),
142 Layer.mock(RunExecutionService.RunExecutionServiceV2)({ startRootRun }),
143 Layer.mock(RuntimePolicy.RuntimePolicyV2)({}),
144 ),
145 ),
146 );
147
148 await Effect.gen(function* () {
149 const error = yield* (yield* ProviderTurnStart.ProviderTurnStartServiceV2)
150 .start({ threadId, runId })
151 .pipe(Effect.flip);
152
153 expect(error._tag).toBe("ProviderTurnStartError");
154 expect(projectionReadCount).toBe(3);
155 expect(pruneWorktrees).toHaveBeenCalledWith({ cwd: "/tmp/provider-turn-start-project" });
156 expect(createWorktree).toHaveBeenCalledWith({
157 cwd: "/tmp/provider-turn-start-project",
158 refName: "feature/restore",
159 path: "/tmp/missing-provider-turn-start-worktree",
160 });
161 expect(writeIfRunCurrent).not.toHaveBeenCalled();
162 expect(startRootRun).not.toHaveBeenCalled();
163 }).pipe(Effect.provide(layer), Effect.runPromise);
164});
165
166function makeLocalCommandHarness(input: {
167 readonly text: string;
168 readonly worktreePath?: string;
169 readonly fileSystem?: FileSystem.FileSystem;
170 readonly previousNativeSession?: boolean;
171 readonly previousMessages?: ReadonlyArray<string>;
172 readonly logoutFailure?: string;
173 readonly openFailure?: unknown;
174 /** Opens the session, then fails loading its provider thread. */
175 readonly ensureThreadFailure?: unknown;
176 /**
177 * Resumes a thread that has a native ref: resume fails, the fresh-thread
178 * fallback succeeds, then reading history for its handoff fails.
179 */
180 readonly historyReadFailureAfterFallback?: unknown;
181 readonly interruptOpen?: boolean;
182 readonly interruptRunBeforeOpenFailure?: boolean;
183 readonly writeFailure?: unknown;
184 /** Loads the thread and starts the run, then fails every later state read. */
185 readonly failReadsAfterRunning?: boolean;
186}) {
187 const now = DateTime.makeUnsafe("2026-09-04T12:00:00Z");
188 const threadId = ThreadId.make("thread-native-account-command");
189 const runId = RunId.make("run-native-account-command");
190 const rootNodeId = NodeId.make("root-native-account-command");
191 const attemptId = RunAttemptId.make("attempt-native-account-command");
192 const providerThreadId = ProviderThreadId.make("new-provider-thread");
193 const providerSessionId = ProviderSessionId.make("new-provider-session");
194 const oldProviderThreadId = ProviderThreadId.make("existing-native-provider-thread");
195 const oldInstanceId = ProviderInstanceId.make("antigravity-personal");
196 const newInstanceId = ProviderInstanceId.make("codex-personal");
197 const checkpointScopeId = CheckpointScopeId.make("scope-native-account-command");
198 const messageId = MessageId.make("message-native-account-command");
199 const run: OrchestrationV2ThreadProjection["runs"][number] = {
200 id: runId,
201 threadId,
202 ordinal: 2,
203 providerInstanceId: newInstanceId,
204 modelSelection: { instanceId: newInstanceId, model: "gpt-5.4" },
205 providerThreadId,
206 userMessageId: messageId,
207 rootNodeId,
208 activeAttemptId: attemptId,
209 status: "starting",
210 requestedAt: now,
211 startedAt: null,
212 completedAt: null,
213 checkpointId: null,
214 contextHandoffId: null,
215 };
216 const providerThread: OrchestrationV2ThreadProjection["providerThreads"][number] = {
217 id: providerThreadId,
218 driver: ProviderDriverKind.make("codex"),
219 providerInstanceId: newInstanceId,
220 providerSessionId,
221 appThreadId: threadId,
222 ownerNodeId: null,
223 nativeThreadRef: null,
224 nativeConversationHeadRef: null,
225 status: "not_loaded",
226 firstRunOrdinal: 2,
227 lastRunOrdinal: 2,
228 handoffIds: [],
229 forkedFrom: null,
230 createdAt: now,
231 updatedAt: now,
232 };
233 const message: OrchestrationV2ThreadProjection["messages"][number] = {
234 id: messageId,
235 threadId,
236 runId,
237 nodeId: rootNodeId,
238 role: "user",
239 text: input.text,
240 attachments: [],
241 streaming: false,
242 createdBy: "user",
243 creationSource: "web",
244 createdAt: now,
245 updatedAt: now,
246 };
247 let projection: OrchestrationV2ThreadProjection = {
248 thread: {
249 id: threadId,
250 activeProviderThreadId: providerThreadId,
251 projectId: ProjectId.make("project-native-account-command"),
252 branch: input.worktreePath === undefined ? null : "feature/restore",
253 worktreePath: input.worktreePath ?? null,
254 } as OrchestrationV2ThreadProjection["thread"],
255 runs: [
256 ...(input.previousNativeSession
257 ? [
258 {
259 ...run,
260 id: RunId.make("previous-native-run"),
261 ordinal: 1,
262 status: "completed" as const,
263 providerInstanceId: oldInstanceId,
264 providerThreadId: oldProviderThreadId,
265 },
266 ]
267 : []),
268 run,
269 ],
270 attempts: [
271 {
272 id: attemptId,
273 runId,
274 rootNodeId,
275 attemptOrdinal: 1,
276 providerInstanceId: newInstanceId,
277 providerThreadId,
278 providerTurnId: null,
279 reason: "initial",
280 status: "pending",
281 startedAt: null,
282 completedAt: null,
283 },
284 ],
285 nodes: [
286 {
287 id: rootNodeId,
288 threadId,
289 runId,
290 parentNodeId: null,
291 rootNodeId,
292 kind: "root_turn",
293 status: "pending",
294 countsForRun: true,
295 providerThreadId,
296 providerTurnId: null,
297 nativeItemRef: null,
298 runtimeRequestId: null,
299 checkpointScopeId,
300 startedAt: null,
301 completedAt: null,
302 },
303 ],
304 providerThreads: [
305 ...(input.previousNativeSession
306 ? [
307 {
308 ...providerThread,
309 id: oldProviderThreadId,
310 providerInstanceId: oldInstanceId,
311 driver: ProviderDriverKind.make("antigravity"),
312 lastRunOrdinal: 1,
313 nativeThreadRef: {
314 driver: ProviderDriverKind.make("antigravity"),
315 nativeId: "existing-session",
316 strength: "strong" as const,
317 },
318 },
319 ]
320 : []),
321 providerThread,
322 ],
323 messages: [
324 ...(input.previousMessages ?? []).map((text, index) => ({
325 ...message,
326 id: MessageId.make(`previous-message-${index}`),
327 text,
328 })),
329 message,
330 ],
331 checkpointScopes: [
332 {
333 id: checkpointScopeId,
334 threadId,
335 runId,
336 nodeId: rootNodeId,
337 parentScopeId: null,
338 providerThreadId,
339 kind: "root_run",
340 ordinalWithinParent: 0,
341 advancesAppRunCount: true,
342 cwd: "/tmp/native-account-command",
343 createdAt: now,
344 },
345 ],
346 providerSessions: [],
347 providerTurns: [],
348 contextHandoffs: [],
349 contextTransfers: [],
350 turnItems: [],
351 visibleTurnItems: [],
352 runtimeRequests: [],
353 subagents: [],
354 plans: [],
355 checkpoints: [],
356 updatedAt: now,
357 };
358 if ("historyReadFailureAfterFallback" in input) {
359 const nativeThreadRef = {
360 driver: providerThread.driver,
361 nativeId: "native-resume-thread",
362 strength: "strong" as const,
363 };
364 projection = {
365 ...projection,
366 providerThreads: projection.providerThreads.map((candidate) =>
367 candidate.id === providerThreadId ? { ...candidate, nativeThreadRef } : candidate,
368 ),
369 };
370 }
371 const events: Array<OrchestrationV2DomainEvent> = [];
372 const interruptRun = () => {
373 projection = {
374 ...projection,
375 runs: projection.runs.map((candidate) =>
376 candidate.id === runId
377 ? { ...candidate, status: "interrupted", completedAt: now }
378 : candidate,
379 ),
380 };
381 };
382 const ensureThread = vi.fn(() =>
383 Effect.sync(() => {
384 if (input.interruptRunBeforeOpenFailure === true) interruptRun();
385 }).pipe(
386 Effect.andThen(
387 Effect.fail(
388 new ProviderAdapter.ProviderAdapterEventStreamError({
389 driver: providerThread.driver,
390 providerSessionId,
391 cause: input.ensureThreadFailure,
392 }),
393 ),
394 ),
395 ),
396 );
397 const resumeFallbackSession = {
398 driver: providerThread.driver,
399 resumeThread: () =>
400 Effect.fail(
401 new ProviderAdapter.ProviderAdapterEventStreamError({
402 driver: providerThread.driver,
403 providerSessionId,
404 cause: "native thread is gone",
405 }),
406 ),
407 ensureThread: () => Effect.succeed(providerThread),
408 };
409 const open = vi.fn(() =>
410 input.interruptOpen === true
411 ? Effect.interrupt
412 : "historyReadFailureAfterFallback" in input
413 ? Effect.succeed(resumeFallbackSession as never)
414 : "ensureThreadFailure" in input
415 ? Effect.succeed({ driver: providerThread.driver, ensureThread } as never)
416 : "openFailure" in input
417 ? Effect.sync(() => {
418 if (input.interruptRunBeforeOpenFailure === true) interruptRun();
419 }).pipe(
420 Effect.andThen(
421 Effect.fail(
422 new ProviderSessionManager.ProviderSessionOpenError({
423 instanceId: newInstanceId,
424 providerSessionId,
425 cause: input.openFailure,
426 }),
427 ),
428 ),
429 )
430 : input.failReadsAfterRunning === true
431 ? Effect.succeed({
432 driver: providerThread.driver,
433 providerSession: {
434 id: providerSessionId,
435 driver: providerThread.driver,
436 providerInstanceId: newInstanceId,
437 status: "ready",
438 cwd: "/tmp/native-account-command",
439 model: null,
440 capabilities: CodexProviderCapabilitiesV2,
441 createdAt: now,
442 updatedAt: now,
443 lastError: null,
444 },
445 ensureThread: () => Effect.succeed(providerThread),
446 } as never)
447 : Effect.die("A local command must not open a native session."),
448 );
449 const startRootRun = vi.fn<
450 (input: RunExecutionService.RunExecutionServiceV2StartRootRunInput) => Effect.Effect<void>
451 >(() =>
452 input.failReadsAfterRunning === true
453 ? Effect.void
454 : Effect.die("A local command must not start a native turn."),
455 );
456 const failReadIfRunning = Effect.suspend(() =>
457 input.failReadsAfterRunning === true &&
458 projection.runs.find((candidate) => candidate.id === runId)?.status === "running"
459 ? Effect.fail(
460 new ProjectionStore.ProjectionStoreReadError({ threadId, cause: "database unavailable" }),
461 )
462 : Effect.void,
463 );
464 const tryHandlePromptCommand = vi.fn(() =>
465 input.logoutFailure === undefined
466 ? Effect.succeed(true)
467 : Effect.fail(
468 new ProviderSetupError({
469 instanceId: oldInstanceId,
470 operation: "logout",
471 detail: input.logoutFailure,
472 }),
473 ),
474 );
475 const writeIfRunCurrent = vi.fn(({ events: incoming, activeAttemptId, expectedStatus }) =>
476 "writeFailure" in input
477 ? Effect.fail(
478 new EventSink.EventSinkWriteError({
479 eventCount: incoming.length,
480 cause: input.writeFailure,
481 }),
482 )
483 : Effect.sync(() => {
484 const current = projection.runs.find((candidate) => candidate.id === runId);
485 const committed =
486 current !== undefined &&
487 current.activeAttemptId === activeAttemptId &&
488 current.status === expectedStatus;
489 if (committed) {
490 for (const event of incoming) {
491 expect(isDomainEvent(event)).toBe(true);
492 events.push(event);
493 projection = ProjectionStore.applyToProjection(projection, event);
494 }
495 }
496 return { committed, storedEvents: [] };
497 }),
498 );
499 const resolveRuntimePolicy = vi.fn(
500 ({ thread }: Parameters<RuntimePolicy.RuntimePolicyV2Shape["resolve"]>[0]) =>
501 Effect.succeed({ cwd: thread.worktreePath ?? "/tmp/native-account-command" } as never),
502 );
503 const createWorktree = vi.fn(() => Effect.succeed({} as never));
504 const getThreadRecords = vi.fn<ProjectionStore.ProjectionStoreV2Shape["getThreadRecords"]>(() =>
505 Effect.succeed(projection),
506 );
507 const getTurnStartContext = vi.fn(() =>
508 Effect.succeed({
509 ...projection,
510 hasConversation: projection.messages.some(
511 (m) =>
512 m.role === "user" &&
513 (m.text.trim().toLowerCase() !== "/compact" || m.attachments.length > 0),
514 ),
515 }),
516 );
517 const layer = ProviderTurnStart.layer.pipe(
518 Layer.provide(
519 Layer.mergeAll(
520 Layer.mock(ContextHandoffService.ContextHandoffServiceV2)({
521 prepareProviderHandoff: () => Effect.die("history read must fail first"),
522 }),
523 Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent }),
524 IdAllocator.layer,
525 Path.layer,
526 input.fileSystem === undefined
527 ? FileSystem.layerNoop({ exists: () => Effect.succeed(true) })
528 : Layer.succeed(FileSystem.FileSystem, input.fileSystem),
529 Layer.mock(GitWorkflow.GitWorkflowService)({
530 pruneWorktrees: () => Effect.void,
531 createWorktree,
532 }),
533 Layer.mock(ProjectService.ProjectService)({
534 getById: () =>
535 Effect.succeed(Option.some({ workspaceRoot: "/tmp/native-account-project" } as never)),
536 }),
537 Layer.mock(ProjectionStore.ProjectionStoreV2)({
538 getThreadRecords,
539 getTurnStartContext,
540 getRuntimeRecoveryProjection: () =>
541 Effect.as(failReadIfRunning, {
542 ...projection,
543 hasConversation: projection.messages.some(
544 (m) =>
545 m.role === "user" &&
546 (m.text.trim().toLowerCase() !== "/compact" || m.attachments.length > 0),
547 ),
548 }),
549 getTurnStartHistory: () =>
550 Effect.fail(
551 new ProjectionStore.ProjectionStoreReadError({
552 threadId,
553 cause: input.historyReadFailureAfterFallback,
554 }),
555 ),
556 }),
557 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ open }),
558 Layer.mock(ProviderAuthService.ProviderAuthService)({ tryHandlePromptCommand }),
559 Layer.mock(RunExecutionService.RunExecutionServiceV2)({ startRootRun }),
560 Layer.mock(RuntimePolicy.RuntimePolicyV2)({
561 resolve: resolveRuntimePolicy,
562 }),
563 ),
564 ),
565 );
566 return {
567 getThreadRecords,
568 getTurnStartContext,
569 setProjection: (next: OrchestrationV2ThreadProjection) => {
570 projection = next;
571 },
572 resolveRuntimePolicy,
573 createWorktree,
574 open,
575 writeIfRunCurrent,
576 startRootRun,
577 tryHandlePromptCommand,
578 events,
579 oldInstanceId,
580 newInstanceId,
581 attemptId,
582 projection: () => projection,
583 start: Effect.gen(function* () {
584 yield* (yield* ProviderTurnStart.ProviderTurnStartServiceV2).start({ threadId, runId });
585 }).pipe(Effect.provide(layer)),
586 startWithRetry: Effect.gen(function* () {
587 yield* (yield* ProviderTurnStart.ProviderTurnStartServiceV2).start({
588 threadId,
589 runId,
590 willRetry: true,
591 });
592 }).pipe(Effect.provide(layer)),
593 };
594}
595
596effectIt.effect.each(["cleanup-first", "startup-first"])(
597 "coordinates provider startup with %s",
598 (order) =>
599 Effect.gen(function* () {
600 const fileSystem = yield* FileSystem.FileSystem;
601 const path = yield* Path.Path;
602 const workspaceRoot = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-start-" });
603 const worktreePath = path.join(workspaceRoot, "worktree");
604 yield* fileSystem.makeDirectory(worktreePath);
605 const release = yield* Deferred.make<void>();
606 const openEntered = yield* Deferred.make<void>();
607 const harness = makeLocalCommandHarness({
608 text: "Continue",
609 worktreePath,
610 fileSystem,
611 failReadsAfterRunning: true,
612 });
613 harness.createWorktree.mockImplementation(() =>
614 fileSystem.makeDirectory(worktreePath).pipe(Effect.as({} as never), Effect.orDie),
615 );
616 const open = harness.open.getMockImplementation()!;
617 harness.open.mockImplementation(() =>
618 Effect.gen(function* () {
619 expect(yield* fileSystem.exists(worktreePath).pipe(Effect.orDie)).toBe(true);
620 yield* Deferred.succeed(openEntered, undefined);
621 if (order === "startup-first") yield* Deferred.await(release);
622 return yield* open();
623 }),
624 );
625 let cleanupEntered = false;
626 const cleanupEffect = withWorkspaceLease(
627 worktreePath,
628 Effect.gen(function* () {
629 cleanupEntered = true;
630 if (order === "cleanup-first") yield* Deferred.await(release);
631 else {
632 expect(harness.projection().providerSessions).toHaveLength(1);
633 expect(harness.projection().runs.at(-1)?.status).toBe("running");
634 }
635 yield* fileSystem.remove(worktreePath, { recursive: true, force: true });
636 }),
637 );
638 const cleanup =
639 order === "cleanup-first"
640 ? yield* cleanupEffect.pipe(Effect.forkChild({ startImmediately: true }))
641 : undefined;
642 const startup = yield* harness.start.pipe(Effect.forkChild({ startImmediately: true }));
643 if (order === "cleanup-first") expect(harness.open).not.toHaveBeenCalled();
644 if (order === "startup-first") yield* Deferred.await(openEntered);
645 const laterCleanup =
646 order === "startup-first"
647 ? yield* cleanupEffect.pipe(Effect.forkChild({ startImmediately: true }))
648 : undefined;
649 if (order === "startup-first") expect(cleanupEntered).toBe(false);
650 yield* Deferred.succeed(release, undefined);
651 if (cleanup !== undefined) yield* Fiber.join(cleanup);
652 yield* Fiber.join(startup);
653 if (laterCleanup !== undefined) yield* Fiber.join(laterCleanup);
654 if (order === "cleanup-first") {
655 expect(harness.createWorktree).toHaveBeenCalledWith({
656 cwd: "/tmp/native-account-project",
657 refName: "feature/restore",
658 path: worktreePath,
659 });
660 }
661 expect(harness.resolveRuntimePolicy).toHaveBeenCalledWith(
662 expect.objectContaining({
663 thread: expect.objectContaining({ worktreePath }),
664 }),
665 );
666 expect(harness.open).toHaveBeenCalledWith(
667 expect.objectContaining({
668 runtimePolicy: expect.objectContaining({ cwd: worktreePath }),
669 }),
670 );
671 expect(harness.startRootRun).toHaveBeenCalledWith(
672 expect.objectContaining({
673 appThread: expect.objectContaining({ worktreePath }),
674 runtimePolicy: expect.objectContaining({ cwd: worktreePath }),
675 }),
676 );
677 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
678);
679
680effectIt.effect("reads only thread state before taking the startup lease", () =>
681 Effect.gen(function* () {
682 const harness = makeLocalCommandHarness({ text: "/logout", previousNativeSession: true });
683 yield* harness.start;
684 expect(harness.getThreadRecords).toHaveBeenCalledExactlyOnceWith(
685 harness.projection().thread.id,
686 [],
687 );
688 expect(harness.getTurnStartContext).toHaveBeenCalledOnce();
689 expect(harness.projection().runs.at(-1)?.status).toBe("completed");
690 }),
691);
692
693effectIt.effect("reads current startup state after waiting for the workspace lease", () =>
694 Effect.gen(function* () {
695 const worktreePath = "/tmp/provider-start-fresh-state";
696 const release = yield* Deferred.make<void>();
697 const readEntered = yield* Deferred.make<void>();
698 const harness = makeLocalCommandHarness({
699 text: "Continue",
700 worktreePath,
701 failReadsAfterRunning: true,
702 });
703 harness.getThreadRecords.mockImplementationOnce(() =>
704 Deferred.succeed(readEntered, undefined).pipe(Effect.as(harness.projection())),
705 );
706 const holder = yield* withWorkspaceLease(worktreePath, Deferred.await(release)).pipe(
707 Effect.forkChild({ startImmediately: true }),
708 );
709 const startup = yield* harness.start.pipe(Effect.forkChild({ startImmediately: true }));
710 yield* Deferred.await(readEntered);
711 expect(harness.getTurnStartContext).not.toHaveBeenCalled();
712 const projection = harness.projection();
713 harness.setProjection({
714 ...projection,
715 runs: projection.runs.map((run) => ({ ...run, status: "running" })),
716 });
717 yield* Deferred.succeed(release, undefined);
718 yield* Fiber.join(holder);
719 yield* Fiber.join(startup);
720 expect(harness.open).not.toHaveBeenCalled();
721 expect(harness.startRootRun).not.toHaveBeenCalled();
722 }),
723);
724
725effectIt.effect("starts another provider in the same checkout while a turn is pending", () =>
726 Effect.gen(function* () {
727 const worktreePath = "/tmp/provider-start-shared-checkout";
728 const turnEntered = yield* Deferred.make<void>();
729 const releaseTurn = yield* Deferred.make<void>();
730 const first = makeLocalCommandHarness({
731 text: "Continue",
732 worktreePath,
733 failReadsAfterRunning: true,
734 });
735 const second = makeLocalCommandHarness({
736 text: "Continue",
737 worktreePath,
738 failReadsAfterRunning: true,
739 });
740 first.startRootRun.mockImplementation(() =>
741 Deferred.succeed(turnEntered, undefined).pipe(Effect.andThen(Deferred.await(releaseTurn))),
742 );
743 const pending = yield* first.start.pipe(Effect.forkChild({ startImmediately: true }));
744 yield* Deferred.await(turnEntered);
745 yield* second.start;
746 expect(second.startRootRun).toHaveBeenCalledOnce();
747 expect(second.projection().runs.at(-1)?.status).toBe("running");
748 yield* Deferred.succeed(releaseTurn, undefined);
749 yield* Fiber.join(pending);
750 }),
751);
752
753effectIt.effect("terminalizes a starting run when its provider session cannot open", () =>
754 Effect.gen(function* () {
755 const harness = makeLocalCommandHarness({
756 text: "Continue",
757 openFailure: new Error("DESCRIPTION is not valid ACP JSON"),
758 });
759
760 yield* harness.start;
761
762 expect(harness.open).toHaveBeenCalledOnce();
763 expect(harness.startRootRun).not.toHaveBeenCalled();
764 expect(harness.writeIfRunCurrent).toHaveBeenCalledWith(
765 expect.objectContaining({
766 activeAttemptId: harness.attemptId,
767 expectedStatus: "starting",
768 }),
769 );
770 const projection = harness.projection();
771 expect(projection.runs.at(-1)).toMatchObject({ status: "failed", startedAt: null });
772 expect(projection.attempts[0]).toMatchObject({ status: "failed", startedAt: null });
773 expect(projection.nodes[0]).toMatchObject({ status: "failed", startedAt: null });
774 expect(projection.turnItems).toMatchObject([
775 {
776 type: "error",
777 status: "failed",
778 failure: {
779 class: "provider_error",
780 message: "DESCRIPTION is not valid ACP JSON",
781 },
782 },
783 ]);
784 }),
785);
786
787effectIt.effect("leaves the run starting when a session-open failure will be retried", () =>
788 Effect.gen(function* () {
789 const harness = makeLocalCommandHarness({
790 text: "Continue",
791 openFailure: new Error("provider session rejected"),
792 });
793
794 const error = yield* harness.startWithRetry.pipe(Effect.flip);
795
796 expect(error._tag).toBe("ProviderTurnStartError");
797 expect(harness.writeIfRunCurrent).not.toHaveBeenCalled();
798 expect(harness.projection().runs.at(-1)?.status).toBe("starting");
799 }),
800);
801
802effectIt.effect("keeps a session-open failure retryable when terminal persistence fails", () =>
803 Effect.gen(function* () {
804 const harness = makeLocalCommandHarness({
805 text: "Continue",
806 openFailure: new Error("provider session rejected"),
807 writeFailure: new Error("database unavailable"),
808 });
809
810 const error = yield* harness.start.pipe(Effect.flip);
811
812 expect(error._tag).toBe("ProviderTurnStartError");
813 expect(harness.writeIfRunCurrent).toHaveBeenCalledOnce();
814 expect(harness.startRootRun).not.toHaveBeenCalled();
815 expect(harness.projection().runs.at(-1)?.status).toBe("starting");
816 expect(harness.events).toEqual([]);
817 }),
818);
819
820effectIt.effect("does not terminalize a provider-session open interruption", () =>
821 Effect.gen(function* () {
822 const harness = makeLocalCommandHarness({ text: "Continue", interruptOpen: true });
823
824 const exit = yield* Effect.exit(harness.start);
825
826 expect(exit._tag).toBe("Failure");
827 if (exit._tag === "Failure") {
828 expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true);
829 }
830 expect(harness.writeIfRunCurrent).not.toHaveBeenCalled();
831 expect(harness.startRootRun).not.toHaveBeenCalled();
832 expect(harness.projection().runs.at(-1)?.status).toBe("starting");
833 expect(harness.events).toEqual([]);
834 }),
835);
836
837effectIt.effect("does not overwrite a run interrupted while its provider session opens", () =>
838 Effect.gen(function* () {
839 const harness = makeLocalCommandHarness({
840 text: "Continue",
841 openFailure: new Error("provider session rejected"),
842 interruptRunBeforeOpenFailure: true,
843 });
844
845 yield* harness.start;
846
847 expect(harness.writeIfRunCurrent).toHaveBeenCalledOnce();
848 expect(harness.startRootRun).not.toHaveBeenCalled();
849 const projection = harness.projection();
850 expect(projection.runs.at(-1)?.status).toBe("interrupted");
851 expect(projection.attempts[0]?.status).toBe("pending");
852 expect(projection.nodes[0]?.status).toBe("pending");
853 expect(projection.turnItems).toEqual([]);
854 expect(harness.events).toEqual([]);
855 }),
856);
857
858effectIt.effect("fails a starting run when its last start attempt cannot load the thread", () =>
859 Effect.gen(function* () {
860 const harness = makeLocalCommandHarness({
861 text: "Continue",
862 ensureThreadFailure: new Error("Pi RPC read failed: pi process exited with code 1."),
863 });
864
865 yield* harness.start;
866
867 expect(harness.startRootRun).not.toHaveBeenCalled();
868 const projection = harness.projection();
869 expect(projection.runs.at(-1)).toMatchObject({ status: "failed", startedAt: null });
870 expect(projection.attempts[0]).toMatchObject({ status: "failed", startedAt: null });
871 expect(projection.nodes[0]).toMatchObject({ status: "failed", startedAt: null });
872 expect(projection.turnItems).toMatchObject([
873 {
874 type: "error",
875 title: "Provider turn failed to start",
876 failure: {
877 class: "provider_error",
878 message: "Pi RPC read failed: pi process exited with code 1.",
879 },
880 },
881 ]);
882 }),
883);
884
885effectIt.effect("leaves the run starting when a thread-load failure will be retried", () =>
886 Effect.gen(function* () {
887 const harness = makeLocalCommandHarness({
888 text: "Continue",
889 ensureThreadFailure: new Error("pi process exited with code 1"),
890 });
891
892 const error = yield* harness.startWithRetry.pipe(Effect.flip);
893
894 expect(error._tag).toBe("ProviderTurnStartError");
895 expect(harness.writeIfRunCurrent).not.toHaveBeenCalled();
896 expect(harness.projection().runs.at(-1)?.status).toBe("starting");
897 }),
898);
899
900effectIt.effect("keeps a thread-load failure retryable when terminal persistence fails", () =>
901 Effect.gen(function* () {
902 const harness = makeLocalCommandHarness({
903 text: "Continue",
904 ensureThreadFailure: new Error("pi process exited with code 1"),
905 writeFailure: new Error("database unavailable"),
906 });
907
908 const error = yield* harness.start.pipe(Effect.flip);
909
910 expect(error._tag).toBe("ProviderTurnStartError");
911 expect(harness.projection().runs.at(-1)?.status).toBe("starting");
912 expect(harness.events).toEqual([]);
913 }),
914);
915
916effectIt.effect(
917 "keeps a store failure after the provider loaded the thread typed and retryable",
918 () =>
919 Effect.gen(function* () {
920 const harness = makeLocalCommandHarness({
921 text: "Continue",
922 historyReadFailureAfterFallback: new Error("database unavailable"),
923 });
924
925 const error = yield* harness.start.pipe(Effect.flip);
926
927 // The provider succeeded; the failing stage is the projection read, so
928 // the run is not failed as a provider error on the last attempt.
929 expect(error._tag).toBe("ProviderTurnStartError");
930 expect((error.cause as { _tag?: string } | undefined)?._tag).toBe("ProjectionStoreReadError");
931 expect(harness.writeIfRunCurrent).not.toHaveBeenCalled();
932 expect(harness.projection().runs.at(-1)?.status).toBe("starting");
933 expect(harness.events).toEqual([]);
934 }),
935);
936
937effectIt.effect("does not mistake a failed state read for a superseded run", () =>
938 Effect.gen(function* () {
939 const harness = makeLocalCommandHarness({ text: "Continue", failReadsAfterRunning: true });
940
941 yield* harness.start;
942
943 expect(harness.projection().runs.at(-1)?.status).toBe("running");
944 const controls = harness.startRootRun.mock.calls[0]?.[0];
945 expect(controls).toBeDefined();
946 if (controls === undefined) return;
947 // "false" would skip the provider turn or the terminal write and leave the
948 // run active. A read failure must reach the caller instead.
949 const startCheck = yield* Effect.flip(controls.shouldStartProviderTurn!());
950 const finalizeCheck = yield* Effect.flip(controls.shouldFinalizeRun!());
951 expect(startCheck._tag).toBe("ProjectionStoreReadError");
952 expect(finalizeCheck._tag).toBe("ProjectionStoreReadError");
953 }),
954);
955
956effectIt.effect("does not overwrite a run interrupted while its thread loads", () =>
957 Effect.gen(function* () {
958 const harness = makeLocalCommandHarness({
959 text: "Continue",
960 ensureThreadFailure: new Error("pi process exited with code 1"),
961 interruptRunBeforeOpenFailure: true,
962 });
963
964 yield* harness.start;
965
966 expect(harness.projection().runs.at(-1)?.status).toBe("interrupted");
967 expect(harness.projection().turnItems).toEqual([]);
968 expect(harness.events).toEqual([]);
969 }),
970);
971
972effectIt.effect(
973 "signs out the existing native provider before opening the newly selected provider",
974 () =>
975 Effect.gen(function* () {
976 const harness = makeLocalCommandHarness({ text: "/logout", previousNativeSession: true });
977
978 yield* harness.start;
979 yield* harness.start;
980
981 expect(harness.tryHandlePromptCommand).toHaveBeenCalledExactlyOnceWith({
982 instanceId: harness.oldInstanceId,
983 text: "/logout",
984 hasAttachments: false,
985 });
986 expect(harness.open).not.toHaveBeenCalled();
987 expect(harness.startRootRun).not.toHaveBeenCalled();
988 const projection = harness.projection();
989 expect(projection.runs.at(-1)?.status).toBe("completed");
990 expect(projection.attempts[0]?.status).toBe("completed");
991 expect(projection.nodes[0]?.status).toBe("completed");
992 expect(projection.turnItems).toMatchObject([
993 {
994 type: "command_execution",
995 title: "Provider signed out",
996 output: "Provider signed out",
997 status: "completed",
998 },
999 ]);
1000 expect(projection.providerTurns).toEqual([]);
1001 expect(projection.checkpoints).toEqual([]);
1002 }),
1003);
1004
1005effectIt.effect("persists a failed sign-out without starting a provider turn", () =>
1006 Effect.gen(function* () {
1007 const harness = makeLocalCommandHarness({
1008 text: "/logout",
1009 logoutFailure: "Could not stop all sessions for this provider. Try again.",
1010 });
1011
1012 yield* harness.start;
1013
1014 expect(harness.open).not.toHaveBeenCalled();
1015 expect(harness.startRootRun).not.toHaveBeenCalled();
1016 expect(harness.projection().runs.at(-1)?.status).toBe("failed");
1017 expect(harness.projection().turnItems).toMatchObject([
1018 {
1019 type: "error",
1020 title: "Provider sign-out failed",
1021 failure: {
1022 class: "permission_error",
1023 message: "Could not stop all sessions for this provider. Try again.",
1024 },
1025 },
1026 ]);
1027 }),
1028);
1029
1030for (const previousMessages of [[], ["/compact", " /COMPACT "]]) {
1031 effectIt.effect(
1032 `rejects compaction without conversation context after ${previousMessages.length} prior compactions`,
1033 () =>
1034 Effect.gen(function* () {
1035 const harness = makeLocalCommandHarness({ text: "/compact", previousMessages });
1036
1037 yield* harness.start;
1038
1039 expect(harness.open).not.toHaveBeenCalled();
1040 expect(harness.tryHandlePromptCommand).not.toHaveBeenCalled();
1041 expect(harness.startRootRun).not.toHaveBeenCalled();
1042 expect(harness.projection().runs.at(-1)?.status).toBe("failed");
1043 expect(harness.projection().turnItems).toMatchObject([
1044 {
1045 type: "error",
1046 failure: {
1047 class: "validation_error",
1048 message: "Start a conversation before compacting this thread.",
1049 },
1050 },
1051 ]);
1052 }),
1053 );
1054}