1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 MessageId,
5 ProjectId,
6 ProviderDriverKind,
7 ProviderInstanceId,
8 ThreadId,
9 TurnItemId,
10} from "@t3tools/contracts";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Exit from "effect/Exit";
14import * as SqlClient from "effect/sql/SqlClient";
15import * as SqlitePersistence from "../persistence/Sqlite.ts";
16import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
17import * as Orchestrator from "./Orchestrator.ts";
18import * as ProjectionStore from "./ProjectionStore.ts";
19import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
20import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
21import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
22
23const instanceId = ProviderInstanceId.make("codex");
24const modelSelection = { instanceId, model: "gpt-5.1-codex" };
25const adapter = {
26 instanceId,
27 driver: ProviderDriverKind.make("codex"),
28 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
29 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
30 openSession: () => Effect.die("A CI failure must not start a provider"),
31} as ProviderAdapter.ProviderAdapterV2["Service"];
32const databaseLayer = SqlitePersistence.layerMemory;
33const testLayer = Layer.mergeAll(
34 databaseLayer,
35 ProjectionStore.layer.pipe(Layer.provide(databaseLayer)),
36 ProviderReplayHarness.layerWithRegistry(
37 { name: "auto-archive" },
38 ProviderAdapterRegistry.layerFromAdapters([adapter]),
39 { databaseLayer, runEffectWorker: false },
40 ),
41);
42
43it.effect(
44 "CI failure wakes atomically with a durable local notification, deduplicates across settlement, and never requests provider work",
45 () =>
46 Effect.gen(function* () {
47 const engine = yield* Orchestrator.OrchestratorV2;
48 const projections = yield* ProjectionStore.ProjectionStoreV2;
49 const sql = yield* SqlClient.SqlClient;
50 const threadId = ThreadId.make("ci-notification");
51 yield* engine.dispatch({
52 type: "thread.create",
53 commandId: CommandId.make("create-ci"),
54 threadId,
55 projectId: ProjectId.make("project-ci"),
56 title: "CI notification",
57 modelSelection,
58 runtimeMode: "full-access",
59 interactionMode: "default",
60 branch: null,
61 worktreePath: null,
62 createdBy: "user",
63 creationSource: "web",
64 });
65 yield* engine.dispatch({
66 type: "thread.settle",
67 commandId: CommandId.make("settle-ci"),
68 threadId,
69 });
70 assert.deepStrictEqual(
71 (yield* projections.getShellSnapshot({
72 location: "active",
73 settledOnly: true,
74 })).threads.map((thread) => thread.id),
75 [threadId],
76 );
77 const itemId = TurnItemId.make("ci-head-notice");
78 const dispatchFailure = (commandId: string, snapshotSequence: number) =>
79 engine.dispatch({
80 type: "thread.ci-failed",
81 commandId: CommandId.make(commandId),
82 threadId,
83 itemId,
84 snapshotSequence,
85 pullRequestNumber: 42,
86 url: "https://mergebot.example.test/owner/repository/pull/42",
87 });
88 yield* dispatchFailure("wake-ci", (yield* projections.getShellSnapshot()).snapshotSequence);
89 const thread = yield* projections.getThread(threadId);
90 assert.equal(thread.settledOverride, "active");
91 assert.equal(thread.settledAt, null);
92 assert.equal(
93 (yield* projections.getShellSnapshot({ location: "active", settledOnly: true })).threads
94 .length,
95 0,
96 );
97 const item = yield* projections.getTurnItem({ threadId, itemId });
98 assert.ok(item?.type === "notification");
99 assert.deepStrictEqual(item.source, { kind: "system" });
100 assert.equal(item.summary, "CI failed for PR #42");
101 assert.equal(
102 item.detail,
103 "[Open PR #42 in Mergebot](<https://mergebot.example.test/owner/repository/pull/42>)",
104 );
105 assert.equal(item.runId, null);
106 assert.equal(
107 (yield* projections.getThreadRecords(threadId, ["runs", "messages"])).runs.length,
108 0,
109 );
110 assert.equal(
111 (yield* projections.getThreadRecords(threadId, ["messages"])).messages.length,
112 0,
113 );
114 const effects = yield* sql<{
115 type: string;
116 }>`SELECT effect_type AS type FROM orchestration_v2_effect_outbox`;
117 assert.ok(
118 effects.every(
119 (effect) =>
120 ![
121 "provider-runtime.continue",
122 "provider-turn.start",
123 "provider-turn.steer",
124 "provider-turn.restart",
125 ].includes(effect.type),
126 ),
127 );
128 yield* engine.dispatch({
129 type: "thread.settle",
130 commandId: CommandId.make("resettle-ci"),
131 threadId,
132 });
133 yield* dispatchFailure(
134 "duplicate-ci",
135 (yield* projections.getShellSnapshot()).snapshotSequence,
136 );
137 assert.equal((yield* projections.getThread(threadId)).settledOverride, "settled");
138 assert.equal(
139 (yield* projections.getThreadRecords(threadId, ["turnItems"])).turnItems.length,
140 1,
141 );
142 }).pipe(Effect.provide(testLayer)),
143);
144
145it.effect(
146 "a concurrent activation or edit invalidates pending automatic archival and CI wakeup",
147 () =>
148 Effect.gen(function* () {
149 const engine = yield* Orchestrator.OrchestratorV2;
150 const projections = yield* ProjectionStore.ProjectionStoreV2;
151 const threadId = ThreadId.make("archive-race");
152 yield* engine.dispatch({
153 type: "thread.create",
154 commandId: CommandId.make("create-race"),
155 threadId,
156 projectId: ProjectId.make("project-ci"),
157 title: "Archive race",
158 modelSelection,
159 runtimeMode: "full-access",
160 interactionMode: "default",
161 branch: null,
162 worktreePath: null,
163 createdBy: "user",
164 creationSource: "web",
165 });
166 yield* engine.dispatch({
167 type: "thread.settle",
168 commandId: CommandId.make("settle-race"),
169 threadId,
170 });
171 const before = (yield* projections.getShellSnapshot()).snapshotSequence;
172 yield* engine.dispatch({
173 type: "thread.unsettle",
174 commandId: CommandId.make("activate-race"),
175 threadId,
176 reason: "user",
177 });
178 yield* engine.dispatch({
179 type: "thread.settle",
180 commandId: CommandId.make("resettle-race"),
181 threadId,
182 });
183 for (const command of [
184 {
185 type: "thread.archive" as const,
186 commandId: CommandId.make("stale-archive"),
187 threadId,
188 snapshotSequence: before,
189 },
190 {
191 type: "thread.ci-failed" as const,
192 commandId: CommandId.make("stale-ci"),
193 threadId,
194 snapshotSequence: before,
195 itemId: TurnItemId.make("stale-notice"),
196 pullRequestNumber: 42,
197 url: "https://mergebot.example.test/owner/repository/pull/42",
198 },
199 ])
200 assert.ok(Exit.isFailure(yield* Effect.exit(engine.dispatch(command))));
201 assert.equal((yield* projections.getThread(threadId)).archivedAt, null);
202 assert.equal(
203 (yield* projections.getThreadRecords(threadId, ["turnItems"])).turnItems.length,
204 0,
205 );
206 yield* engine.dispatch({
207 type: "thread.archive",
208 commandId: CommandId.make("fresh-archive"),
209 threadId,
210 snapshotSequence: (yield* projections.getShellSnapshot()).snapshotSequence,
211 });
212 assert.notEqual((yield* projections.getThread(threadId)).archivedAt, null);
213 }).pipe(Effect.provide(testLayer)),
214);
215
216it.effect(
217 "a new user turn prevents a pending archive or CI observation from changing the thread",
218 () =>
219 Effect.gen(function* () {
220 const engine = yield* Orchestrator.OrchestratorV2;
221 const projections = yield* ProjectionStore.ProjectionStoreV2;
222 const threadId = ThreadId.make("new-run-race");
223 yield* engine.dispatch({
224 type: "thread.create",
225 commandId: CommandId.make("create-new-run"),
226 threadId,
227 projectId: ProjectId.make("project-ci"),
228 title: "New run race",
229 modelSelection,
230 runtimeMode: "full-access",
231 interactionMode: "default",
232 branch: null,
233 worktreePath: null,
234 createdBy: "user",
235 creationSource: "web",
236 });
237 yield* engine.dispatch({
238 type: "thread.settle",
239 commandId: CommandId.make("settle-new-run"),
240 threadId,
241 });
242 const snapshotSequence = (yield* projections.getShellSnapshot()).snapshotSequence;
243 yield* engine.dispatch({
244 type: "message.dispatch",
245 commandId: CommandId.make("new-user-turn"),
246 threadId,
247 messageId: MessageId.make("new-message"),
248 text: "Continue this work",
249 attachments: [],
250 dispatchMode: { type: "queue_after_active" },
251 createdBy: "user",
252 creationSource: "web",
253 });
254 for (const command of [
255 {
256 type: "thread.archive" as const,
257 commandId: CommandId.make("archive-after-new-run"),
258 threadId,
259 snapshotSequence,
260 },
261 {
262 type: "thread.ci-failed" as const,
263 commandId: CommandId.make("ci-after-new-run"),
264 threadId,
265 snapshotSequence,
266 itemId: TurnItemId.make("new-run-notice"),
267 pullRequestNumber: 42,
268 url: "https://mergebot.example.test/owner/repository/pull/42",
269 },
270 ])
271 assert.ok(Exit.isFailure(yield* Effect.exit(engine.dispatch(command))));
272 assert.equal((yield* projections.getThread(threadId)).archivedAt, null);
273 assert.equal(
274 yield* projections.getTurnItem({ threadId, itemId: TurnItemId.make("new-run-notice") }),
275 null,
276 );
277 assert.equal((yield* projections.getThreadRecords(threadId, ["runs"])).runs.length, 1);
278 }).pipe(Effect.provide(testLayer)),
279);