1import { assert, it } from "@effect/vitest";
2import { CheckpointScopeId, CommandId, GitCommandError, RunId, ThreadId } from "@t3tools/contracts";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Option from "effect/Option";
6
7import * as SqlitePersistence from "../persistence/Sqlite.ts";
8import * as ProjectEnrichment from "../project/ProjectEnrichmentService.ts";
9import * as ProjectService from "../project/ProjectService.ts";
10import * as PullRequestService from "../pullRequest/PullRequestService.ts";
11import * as ServerSettings from "../serverSettings.ts";
12import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts";
13import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts";
14import * as CheckpointCapture from "./CheckpointCaptureService.ts";
15import * as CheckpointRollback from "./CheckpointRollbackService.ts";
16import * as EffectOutbox from "./EffectOutbox.ts";
17import * as EffectWorker from "./EffectWorker.ts";
18import * as ProjectionStore from "./ProjectionStore.ts";
19import * as ProviderSessions from "./ProviderSessionManager.ts";
20import * as ProviderTurnControl from "./ProviderTurnControlService.ts";
21import * as ProviderTurnStart from "./ProviderTurnStartService.ts";
22import * as RunFinalization from "./RunFinalizationService.ts";
23import * as RuntimeRequests from "./RuntimeRequestService.ts";
24import * as ThreadManagement from "./ThreadManagementService.ts";
25import * as ThreadTitles from "./ThreadTitleRegenerationService.ts";
26
27it.effect.each([
28 ["starts the next turn after a post-capture Git status failure", false],
29 ["keeps the next turn queued when checkpoint capture fails", true],
30] as const)("%s", ([, captureFails]) => {
31 const threadId = ThreadId.make("thread:finalization-queue");
32 const runId = RunId.make("run:finalization-queue");
33 const scopeId = CheckpointScopeId.make("scope:finalization-queue");
34 const calls: string[] = [];
35 const checkpointFailure = new CheckpointCapture.CheckpointCaptureExecutionError({
36 threadId,
37 runId,
38 scopeId,
39 cause: "checkpoint storage unavailable",
40 });
41 const projections = Layer.mock(ProjectionStore.ProjectionStoreV2)({
42 getCheckpointContext: () =>
43 Effect.succeed({
44 runs: [],
45 checkpointScopes: [{ id: scopeId, runId, kind: "root_run" as const, cwd: "/repo" }],
46 checkpoints: [],
47 }),
48 getThreadShell: () => Effect.succeed(null),
49 });
50 const observer = RunFinalization.layerObserver.pipe(
51 Layer.provide(
52 Layer.mergeAll(
53 projections,
54 Layer.mock(ProjectService.ProjectService)({}),
55 Layer.mock(ProjectEnrichment.ProjectEnrichmentService)({}),
56 Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: () => Effect.void }),
57 Layer.mock(PullRequestService.PullRequestService)({ refreshAfterTurn: () => Effect.void }),
58 Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({
59 refreshLocalStatus: () =>
60 Effect.gen(function* () {
61 calls.push("refresh");
62 return yield* new GitCommandError({
63 operation: "statusDetailsLocal",
64 command: "git",
65 cwd: "/repo",
66 detail: "VCS process timed out after 30000ms",
67 });
68 }),
69 }),
70 ),
71 ),
72 );
73 const finalization = RunFinalization.layer.pipe(
74 Layer.provide(
75 Layer.mergeAll(
76 projections,
77 observer,
78 Layer.mock(CheckpointCapture.CheckpointCaptureServiceV2)({
79 execute: () =>
80 Effect.gen(function* () {
81 calls.push("capture");
82 if (captureFails) return yield* checkpointFailure;
83 }),
84 }),
85 ),
86 ),
87 );
88 const executor = EffectWorker.layerExecutor.pipe(
89 Layer.provide(
90 Layer.mergeAll(
91 finalization,
92 Layer.mock(ProviderSessions.ProviderSessionManagerV2)({}),
93 Layer.mock(ProviderTurnControl.ProviderTurnControlServiceV2)({}),
94 Layer.mock(CheckpointRollback.CheckpointRollbackServiceV2)({}),
95 Layer.mock(RuntimeRequests.RuntimeRequestServiceV2)({}),
96 Layer.mock(ThreadTitles.ThreadTitleRegenerationService)({}),
97 Layer.mock(ThreadManagement.ThreadManagementService)({
98 recoverDelegatedTask: () => Effect.void,
99 }),
100 ServerSettings.layerTest(),
101 Layer.mock(ProviderTurnStart.ProviderTurnStartServiceV2)({
102 start: () => Effect.sync(() => calls.push("start")).pipe(Effect.asVoid),
103 }),
104 ),
105 ),
106 );
107 const outbox = EffectOutbox.layer.pipe(Layer.provide(SqlitePersistence.layerMemory));
108 const worker = EffectWorker.layer.pipe(Layer.provideMerge(Layer.merge(executor, outbox)));
109
110 return Effect.gen(function* () {
111 const queue = yield* EffectOutbox.EffectOutboxV2;
112 const effectWorker = yield* EffectWorker.OrchestrationEffectWorkerV2;
113 yield* queue.enqueue([
114 {
115 id: "effect:finalization-queue:capture",
116 commandId: CommandId.make("command:finalization-queue:capture"),
117 threadId,
118 request: { type: "checkpoint.capture", runId, scopeId },
119 },
120 {
121 id: "effect:finalization-queue:start",
122 commandId: CommandId.make("command:finalization-queue:start"),
123 threadId,
124 request: { type: "provider-turn.start", runId: RunId.make("run:next") },
125 },
126 ]);
127
128 assert.equal(yield* effectWorker.drain(2), captureFails ? 1 : 2);
129 const captured = Option.getOrThrow(yield* queue.get("effect:finalization-queue:capture"));
130 const started = Option.getOrThrow(yield* queue.get("effect:finalization-queue:start"));
131 assert.equal(captured.status, captureFails ? "pending" : "succeeded");
132 assert.equal(captured.attemptCount, 1);
133 assert.equal(started.status, captureFails ? "pending" : "succeeded");
134 assert.equal(started.attemptCount, captureFails ? 0 : 1);
135 assert.deepEqual(calls, captureFails ? ["capture"] : ["capture", "refresh", "start"]);
136 if (captureFails) assert.include(captured.lastError ?? "", "checkpoint storage unavailable");
137 else assert.equal(captured.lastError, null);
138 }).pipe(Effect.provide(Layer.fresh(worker)));
139});