1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import { assert, it } from "@effect/vitest";
3import {
4 CommandId,
5 ProjectId,
6 ProviderDriverKind,
7 ProviderInstanceId,
8 SourceControlProviderKind,
9 ThreadId,
10 type PullRequestDetail,
11 type MergebotPullRequestDetails,
12 type ThreadPullRequestSnapshot,
13} from "@t3tools/contracts";
14import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Ref from "effect/Ref";
18import * as Stream from "effect/Stream";
19import * as MergebotQueue from "../mergebot/MergebotQueue.ts";
20import * as TestClock from "effect/testing/TestClock";
21
22import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
23import * as SqlitePersistence from "../persistence/Sqlite.ts";
24import * as PullRequestService from "../pullRequest/PullRequestService.ts";
25import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
26import * as Orchestrator from "./Orchestrator.ts";
27import * as ProjectionStore from "./ProjectionStore.ts";
28import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
29import * as PullRequestWatchReactor from "./PullRequestWatchReactor.ts";
30import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
31
32const threadId = ThreadId.make("watch-merge");
33const projectId = ProjectId.make("watch-project");
34const instanceId = ProviderInstanceId.make("codex");
35const key = { host: "github.com", repository: "owner/repository", number: 42 };
36const url = "https://github.com/owner/repository/pull/42";
37const timestamp = "2026-10-14T12:00:00.000Z";
38const adapter = {
39 instanceId,
40 driver: ProviderDriverKind.make("codex"),
41 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
42 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
43 openSession: () => Effect.die("Merge-watch tests do not start providers"),
44} as ProviderAdapter.ProviderAdapterV2["Service"];
45const database = SqlitePersistence.layerMemory;
46const layerTest = Layer.mergeAll(
47 NodeCrypto.layer,
48 ProjectionStore.layer.pipe(Layer.provide(database)),
49 OrchestrationEventStore.layer.pipe(Layer.provide(database)),
50 ProviderReplayHarness.layerWithRegistry(
51 { name: "pull-request-watch" },
52 ProviderAdapterRegistry.layerFromAdapters([adapter]),
53 { databaseLayer: database, runEffectWorker: false },
54 ),
55);
56
57const detail: PullRequestDetail = {
58 provider: SourceControlProviderKind.make("github"),
59 capabilities: {
60 diff: false,
61 comment: false,
62 actions: [],
63 mergeMethods: [],
64 search: false,
65 review: { inlineComment: false, reply: false, resolve: false, verdicts: [] },
66 reviewers: { request: false, listCandidates: false },
67 },
68 viewerPermissions: {
69 actions: [],
70 comment: false,
71 resolve: false,
72 verdicts: [],
73 requestReviewers: false,
74 },
75 projectId,
76 projectTitle: "Project",
77 workspaceRoot: "/workspace",
78 repository: key.repository,
79 number: key.number,
80 title: "Change",
81 body: "",
82 url,
83 author: null,
84 state: "merged",
85 isDraft: false,
86 mergeability: "mergeable",
87 additions: 0,
88 deletions: 0,
89 changedFiles: 0,
90 headBranch: "feature",
91 baseBranch: "main",
92 createdAt: timestamp,
93 updatedAt: timestamp,
94 mergedAt: timestamp,
95 closedAt: null,
96 reviewers: [],
97 labels: [],
98 checks: [],
99 mergeCapabilities: { merge: false, squash: false, rebase: false },
100};
101
102const makeHarness = Effect.fn("makeMergeWatchHarness")(function* (options: {
103 readonly mergebot?: MergebotQueue.MergebotQueue["Service"];
104 readonly snapshotState: ThreadPullRequestSnapshot["state"];
105 readonly beforeRead?: (engine: Orchestrator.OrchestratorV2["Service"]) => Effect.Effect<void>;
106}) {
107 const engine = yield* Orchestrator.OrchestratorV2;
108 yield* engine.dispatch({
109 type: "thread.create",
110 commandId: CommandId.make("create"),
111 threadId,
112 projectId,
113 title: "Wait for merge",
114 modelSelection: { instanceId, model: "gpt-5.1-codex" },
115 runtimeMode: "full-access",
116 interactionMode: "default",
117 branch: null,
118 worktreePath: null,
119 createdBy: "user",
120 creationSource: "web",
121 });
122 yield* engine.dispatch({
123 type: "thread.pull-request.watch",
124 commandId: CommandId.make("watch"),
125 threadId,
126 ...key,
127 watching: true,
128 link: { url, source: "agent" },
129 });
130 yield* engine.dispatch({
131 type: "thread.pull-request-link.sync",
132 commandId: CommandId.make("snapshot"),
133 threadId,
134 ...key,
135 snapshot: {
136 state: options.snapshotState,
137 title: "Change",
138 headBranch: "feature",
139 baseBranch: "main",
140 isDraft: false,
141 updatedAt: timestamp,
142 syncedAt: timestamp,
143 },
144 stack: null,
145 });
146 const reads = yield* Ref.make(0);
147 const reactor = yield* PullRequestWatchReactor.make.pipe(
148 Effect.provide(
149 Layer.merge(
150 options.mergebot
151 ? Layer.succeed(MergebotQueue.MergebotQueue, options.mergebot)
152 : Layer.mock(MergebotQueue.MergebotQueue)({
153 configuration: null,
154 streamChanges: Stream.empty,
155 }),
156 Layer.mock(PullRequestService.PullRequestService)({
157 watchFingerprint: () => Effect.succeed(null),
158 detail: () =>
159 Ref.update(reads, (count) => count + 1).pipe(
160 Effect.andThen(options.beforeRead?.(engine) ?? Effect.void),
161 Effect.as(detail),
162 ),
163 activity: () =>
164 Effect.succeed({
165 comments: [],
166 commentCount: 0,
167 commentsTruncated: false,
168 reviewThreads: [],
169 commits: [],
170 }),
171 }),
172 ),
173 ),
174 );
175 return { engine, reactor, reads };
176});
177
178it.effect.each(["merged", "open"] as const)(
179 "wakes once and ends the watch when the %s snapshot leads to a merged PR",
180 (snapshotState) =>
181 Effect.gen(function* () {
182 const { engine, reactor, reads } = yield* makeHarness({ snapshotState });
183 yield* reactor.sweep;
184 const first = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]);
185 assert.isUndefined(first.thread.pullRequests?.[0]?.watch);
186 assert.equal(first.runs.length, 1);
187 const notice = first.turnItems.find((item) => item.type === "notification");
188 assert.ok(notice?.type === "notification");
189 assert.equal(notice.summary, "Pull request #42 merged");
190 assert.equal(notice.outcome, "completed");
191 assert.deepEqual(notice.source, { kind: "monitor" });
192 assert.equal(yield* Ref.get(reads), snapshotState === "merged" ? 0 : 1);
193 yield* reactor.sweep;
194 const second = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]);
195 assert.equal(second.runs.length, 1);
196 assert.equal(second.turnItems.filter((item) => item.type === "notification").length, 1);
197 }).pipe(Effect.provide(layerTest)),
198);
199
200it.effect.each(["stop", "settle", "archive", "restart"] as const)(
201 "a %s during the host read prevents a stale merged wake",
202 (action) =>
203 Effect.gen(function* () {
204 const { engine, reactor } = yield* makeHarness({
205 snapshotState: "open",
206 beforeRead: (engine) =>
207 Effect.gen(function* () {
208 if (action === "settle" || action === "archive") {
209 yield* engine.dispatch({
210 type: action === "settle" ? "thread.settle" : "thread.archive",
211 commandId: CommandId.make(action),
212 threadId,
213 });
214 return;
215 }
216 yield* engine.dispatch({
217 type: "thread.pull-request.watch",
218 commandId: CommandId.make("stop"),
219 threadId,
220 ...key,
221 watching: false,
222 });
223 if (action === "restart") {
224 yield* TestClock.adjust(1);
225 yield* engine.dispatch({
226 type: "thread.pull-request.watch",
227 commandId: CommandId.make("restart"),
228 threadId,
229 ...key,
230 watching: true,
231 });
232 }
233 }).pipe(Effect.orDie),
234 });
235 yield* reactor.sweep;
236 const records = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]);
237 assert.equal(records.runs.length, 0);
238 assert.equal(records.turnItems.filter((item) => item.type === "notification").length, 0);
239 if (action === "restart") assert.isDefined(records.thread.pullRequests?.[0]?.watch);
240 else assert.isUndefined(records.thread.pullRequests?.[0]?.watch);
241 if (action === "settle") assert.equal(records.thread.settledOverride, "settled");
242 if (action === "archive") assert.isNotNull(records.thread.archivedAt);
243 }).pipe(Effect.provide(layerTest)),
244);
245
246it.effect.each(["settle", "archive"] as const)(
247 "a thread already %s does not wake on a merged snapshot",
248 (action) =>
249 Effect.gen(function* () {
250 const { engine, reactor, reads } = yield* makeHarness({ snapshotState: "merged" });
251 yield* engine.dispatch({
252 type: action === "settle" ? "thread.settle" : "thread.archive",
253 commandId: CommandId.make(action),
254 threadId,
255 });
256 yield* reactor.sweep;
257 const records = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]);
258 assert.equal(records.runs.length, 0);
259 assert.equal(records.turnItems.filter((item) => item.type === "notification").length, 0);
260 assert.equal(yield* Ref.get(reads), 0);
261 }).pipe(Effect.provide(layerTest)),
262);
263
264it.effect(
265 "work PR watches verify Mergebot and never call GitHub detail, activity or fingerprint reads",
266 () =>
267 Effect.gen(function* () {
268 let nativeReads = 0;
269 const value: MergebotPullRequestDetails = {
270 number: key.number,
271 headSha: "head",
272 lifecycle: "merged",
273 queueStatus: "not_queued",
274 position: null,
275 mergeEtaAt: null,
276 title: "Change",
277 body: null,
278 url: "https://mergebot.example/owner/repository/pull/42",
279 createdAt: timestamp,
280 updatedAt: timestamp,
281 mergedAt: timestamp,
282 closedAt: timestamp,
283 author: null,
284 additions: 3,
285 deletions: 1,
286 changedFilesCount: 1,
287 labels: [],
288 comments: [],
289 observedAt: Date.parse(timestamp),
290 draft: false,
291 baseBranch: "main",
292 headBranch: "feature",
293 ciStatus: "passed",
294 hasConflicts: false,
295 checks: [],
296 };
297 const { engine, reactor, reads } = yield* makeHarness({
298 snapshotState: "merged",
299 mergebot: {
300 configuration: { origin: "https://mergebot.example", repositories: [key.repository] },
301 read: () => Effect.succeed(null),
302 watchPullRequestStatus: () => Stream.empty,
303 watchDeployments: () => Stream.empty,
304 openPullRequest: () => Effect.die("unused"),
305 setAutoMerge: () => Effect.die("unused"),
306 updatePullRequest: () => Effect.die("unused"),
307 deployProduction: () => Effect.die("unused"),
308 watchQueue: () => Stream.empty,
309 findPullRequest: () => Effect.succeed(null),
310 readPullRequest: () =>
311 Effect.sync(() => {
312 nativeReads++;
313 return value;
314 }),
315 streamChanges: Stream.empty,
316 },
317 });
318 yield* reactor.sweep;
319 const records = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]);
320 assert.equal(records.runs.length, 1);
321 assert.isUndefined(records.thread.pullRequests?.[0]?.watch);
322 assert.equal(yield* Ref.get(reads), 0);
323 assert.isAbove(nativeReads, 0);
324 }).pipe(Effect.provide(layerTest)),
325);