clover's git

the-knife

1	import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
2	import {
3	threadPullRequestKeyOf,
4	normalizeThreadPullRequestKey,
5	legacyThreadPullRequestKey,
6	visibleThreadPullRequests,
7	} from "@t3tools/shared/threadPullRequests";
8	import {
9	CommandId,
10	TurnItemId,
11	MergebotPullRequestCheck,
12	isMergebotRepository,
13	type ThreadId,
14	type PullRequestRef,
15	type OrchestrationV2AppThread,
16	type OrchestrationV2StoredEvent,
17	type OrchestrationV2ThreadShell,
18	} from "@t3tools/contracts";
19	import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
20	import * as Cause from "effect/Cause";
21	import * as Context from "effect/Context";
22	import * as Crypto from "effect/Crypto";
23	import * as DateTime from "effect/DateTime";
24	import * as Effect from "effect/Effect";
25	import * as FileSystem from "effect/FileSystem";
26	import * as Layer from "effect/Layer";
27	import * as Schema from "effect/Schema";
28	import * as Hex from "effect/encoding/Hex";
29	import * as Schedule from "effect/Schedule";
30	import type * as Scope from "effect/Scope";
31	import * as Stream from "effect/Stream";
32
33	import * as GitManager from "../git/GitManager.ts";
34	import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts";
35	import * as PullRequestService from "../pullRequest/PullRequestService.ts";
36	import * as ServerSettings from "../serverSettings.ts";
37	import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
38	import { forkParked } from "../serverActivation.ts";
39	import * as TerminalManager from "../terminal/Manager.ts";
40	import * as PreviewManager from "../preview/Manager.ts";
41	import * as ProjectStore from "./ProjectStore.ts";
42	import * as Orchestrator from "./Orchestrator.ts";
43	import * as ProjectionStore from "./ProjectionStore.ts";
44	import * as ManagedWorkspaces from "../workspace/ManagedWorkspaces.ts";
45	import * as DevServer from "../workspace/DevServer.ts";
46	import * as MergebotQueue from "../mergebot/MergebotQueue.ts";
47	import { isManagedWorkspaceArchiveCandidate } from "../workspace/managedWorkspaceArchive.ts";
48
49	const encodeCiFailureIdentity = Schema.encodeSync(
50	Schema.fromJsonString(
51	Schema.Array(
52	MergebotPullRequestCheck.mapFields(({ id, status, conclusion, completedAt }) => ({
53	id,
54	status,
55	conclusion,
56	completedAt,
57	})),
58	),
59	),
60	);
61
62	export interface SettlementPullRequest {
63	readonly state: "open" | "closed" | "merged";
64	readonly closedAt?: string | null;
65	readonly mergedAt?: string | null;
66	}
67
68	export function lastSaturdayArchiveBoundary(nowMs: number): number {
69	const now = DateTime.makeZonedUnsafe(nowMs, { timeZone: DateTime.zoneMakeLocal() });
70	const days = (DateTime.toParts(now).weekDay + 1) % 7;
71	const boundary = DateTime.setParts(DateTime.subtract(now, { days }), {
72	hour: 9,
73	minute: 0,
74	second: 0,
75	millisecond: 0,
76	});
77	return DateTime.toEpochMillis(
78	DateTime.toEpochMillis(boundary) > nowMs ? DateTime.subtract(boundary, { weeks: 1 }) : boundary,
79	);
80	}
81
82	const DAY_MS = 24 * 60 * 60 * 1_000;
83	export const QUEUED_TURN_START_GRACE_MS = 2 * 60 * 1_000;
84
85	function toMillis(value: DateTime.Utc | null | undefined): number | null {
86	return value == null ? null : DateTime.toEpochMillis(value);
87	}
88
89	function latestMillis(values: ReadonlyArray<number | null>): number | null {
90	let latest: number | null = null;
91	for (const value of values) {
92	if (value === null) continue;
93	if (latest === null || value > latest) latest = value;
94	}
95	return latest;
96	}
97
98	function canonicalRepositoryKey(key: string): string {
99	return key
100	.replace(
101	/^(?:ssh\.dev\.azure\.com|vs-ssh\.visualstudio\.com)\/v3\/([^/]+)\/([^/]+)\/([^/]+)$/u,
102	"dev.azure.com/$1/$2/_git/$3",
103	)
104	.replace(
105	/^([^.]+)\.visualstudio\.com\/(?:defaultcollection\/)?([^/]+)\/_git\/([^/]+)$/u,
106	"dev.azure.com/$1/$2/_git/$3",
107	);
108	}
109
110	function pullRequestMatchesProject(
111	pullRequest: GitManager.GitBranchPullRequest,
112	project: {
113	readonly repositoryIdentity?: { readonly canonicalKey: string } | null | undefined;
114	},
115	): boolean {
116	return (
117	pullRequest.repositoryKey !== null &&
118	project.repositoryIdentity != null &&
119	canonicalRepositoryKey(pullRequest.repositoryKey) ===
120	canonicalRepositoryKey(project.repositoryIdentity.canonicalKey)
121	);
122	}
123
124	/**
125	* A recent user message stays queued until a run adopts its timestamp.
126	* Absolute age bounds client clock skew in both directions and stops stale
127	* pre-adoption data from blocking the thread forever. A failed run start
128	* clears the block immediately (mirrors the v1 session "error" rule).
129	*/
130	export function threadHasQueuedTurnStart(
131	thread: Pick<
132	OrchestrationV2ThreadShell,
133	| "latestUserMessageAt"
134	| "latestRunRequestedAt"
135	| "latestRunStartedAt"
136	| "latestRunCompletedAt"
137	| "latestRunId"
138	| "status"
139	>,
140	nowMs: number,
141	): boolean {
142	const messageAtMs = toMillis(thread.latestUserMessageAt);
143	if (messageAtMs === null || thread.status === "failed") return false;
144	const age = nowMs - messageAtMs;
145	if (Number.isNaN(age) || Math.abs(age) > QUEUED_TURN_START_GRACE_MS) return false;
146	if (thread.latestRunId === null) return true;
147	return [
148	toMillis(thread.latestRunRequestedAt),
149	toMillis(thread.latestRunStartedAt),
150	toMillis(thread.latestRunCompletedAt),
151	].every((value) => value === null || value < messageAtMs);
152	}
153
154	/**
155	* A merged or closed pull request settles the thread unless the user wrote to
156	* it afterwards. Runs that background work, a PR watch, or another agent
157	* started do not count, so they cannot hold a merged thread open.
158	*/
159	function pullRequestSettles(
160	thread: Pick<
161	ProjectionStore.ProjectionSettlementCandidate,
162	"createdAt" | "latestUserAuthoredMessageAt"
163	>,
164	pullRequest: SettlementPullRequest,
165	autoSettleOnMerge: boolean,
166	): boolean {
167	if (pullRequest.state !== "closed" && (pullRequest.state !== "merged" || !autoSettleOnMerge)) {
168	return false;
169	}
170	const terminalAt = pullRequest.state === "merged" ? pullRequest.mergedAt : pullRequest.closedAt;
171	if (terminalAt == null) return false;
172	const userAnchorMs = latestMillis([
173	toMillis(thread.createdAt),
174	toMillis(thread.latestUserAuthoredMessageAt),
175	]);
176	if (userAnchorMs === null) return false;
177	const pullRequestAtMs = Date.parse(terminalAt);
178	if (Number.isNaN(pullRequestAtMs)) return false;
179	return pullRequestAtMs >= userAnchorMs;
180	}
181
182	/** Cheap checks that run before any source control lookup. */
183	export function isAutoSettlementCandidate(
184	thread: Omit<ProjectionStore.ProjectionSettlementCandidate, "latestUserAuthoredMessageAt">,
185	nowMs: number,
186	): boolean {
187	if (thread.archivedAt !== null || thread.settledOverride !== null) return false;
188	if (thread.pinnedAt != null || thread.autoSettleDisabledAt != null) return false;
189	// Blocked-on-you work must never park behind a settled override.
190	if (thread.pendingRuntimeRequest !== null) return false;
191	if (thread.activityRunStatus != null) return false;
192	if ((thread.pendingBackgroundTasks?.length ?? 0) > 0) return false;
193	if (threadHasQueuedTurnStart(thread, nowMs)) return false;
194	const snoozedUntilMs = toMillis(thread.snoozedUntil);
195	if (snoozedUntilMs === null || snoozedUntilMs <= nowMs) return true;
196	// A snoozed thread that woke early (error or completed work) can settle;
197	// one still parked on its wake time keeps its stronger statement.
198	const snoozedAtMs = toMillis(thread.snoozedAt);
199	const completedAtMs = toMillis(thread.latestRunCompletedAt);
200	const wokeOnError =
201	thread.status === "failed" &&
202	(snoozedAtMs === null || (completedAtMs !== null && completedAtMs > snoozedAtMs));
203	const wokeOnCompletion =
204	snoozedAtMs !== null && completedAtMs !== null && completedAtMs > snoozedAtMs;
205	return wokeOnError || wokeOnCompletion;
206	}
207
208	/**
209	* Whether a thread is parked on its snooze: its wake time is in the future and
210	* it has not raised its hand with a pending request, a fresh failure, or work
211	* that completed after the snooze. Server twin of the client's
212	* `effectiveSnoozed`, so agents and the sidebar agree on what is snoozed. One
213	* difference: a failure counts as fresh when its run completed after the
214	* snooze, like `isAutoSettlementCandidate`. The client compares the shell's
215	* update time, so a rename can wake a failed thread there but not here.
216	*/
217	export function isSnoozed(
218	thread: Pick<
219	ProjectionStore.ProjectionSettlementCandidate,
220	"snoozedUntil" | "snoozedAt" | "latestRunCompletedAt" | "status" | "pendingRuntimeRequest"
221	>,
222	nowMs: number,
223	): boolean {
224	const snoozedUntilMs = toMillis(thread.snoozedUntil);
225	if (snoozedUntilMs === null || snoozedUntilMs <= nowMs) return false;
226	if (thread.pendingRuntimeRequest !== null) return false;
227	const snoozedAtMs = toMillis(thread.snoozedAt);
228	const completedAtMs = toMillis(thread.latestRunCompletedAt);
229	const wokeOnError =
230	thread.status === "failed" &&
231	(snoozedAtMs === null || (completedAtMs !== null && completedAtMs > snoozedAtMs));
232	// Like the client, only a run that completed wakes it; an interrupt or cancel does not.
233	const wokeOnCompletion =
234	thread.status === "completed" &&
235	snoozedAtMs !== null &&
236	completedAtMs !== null &&
237	completedAtMs > snoozedAtMs;
238	return !wokeOnError && !wokeOnCompletion;
239	}
240
241	export function resolveAutoSettlementAt(input: {
242	readonly thread: ProjectionStore.ProjectionSettlementCandidate;
243	readonly pullRequest: SettlementPullRequest | null;
244	readonly nowMs: number;
245	readonly autoSettleAfterDays: number | null;
246	readonly autoSettleOnMerge: boolean;
247	readonly autoSettleKeepPullRequests?: boolean;
248	readonly usePullRequestSnapshots?: boolean;
249	}): DateTime.Utc | null {
250	const { thread } = input;
251	let pullRequest = input.pullRequest;
252	const links = visibleThreadPullRequests(thread.pullRequests ?? []);
253	if (
254	input.usePullRequestSnapshots !== false &&
255	links.some((link) => link.snapshot === null || link.snapshot.state === "open")
256	)
257	return null;
258	if (input.usePullRequestSnapshots !== false && links.length > 0) {
259	const terminalAt = (link: (typeof links)[number]) => {
260	const snapshot = link.snapshot;
261	const value = snapshot?.state === "merged" ? snapshot.mergedAt : snapshot?.closedAt;
262	const timestamp = Date.parse(value ?? "");
263	return Number.isNaN(timestamp) ? Number.NEGATIVE_INFINITY : timestamp;
264	};
265	const latest = links.reduce((current, candidate) =>
266	terminalAt(candidate) > terminalAt(current) ? candidate : current,
267	);
268	pullRequest =
269	latest.snapshot === null
270	? null
271	: {
272	state: latest.snapshot.state,
273	mergedAt: latest.snapshot.mergedAt ?? null,
274	closedAt: latest.snapshot.closedAt ?? null,
275	};
276	}
277	if (!isAutoSettlementCandidate(thread, input.nowMs)) return null;
278	const activityAtMs = latestMillis([
279	toMillis(thread.latestUserMessageAt),
280	toMillis(thread.latestRunRequestedAt),
281	toMillis(thread.latestRunStartedAt),
282	toMillis(thread.latestRunCompletedAt),
283	]);
284	if (pullRequest !== null && pullRequestSettles(thread, pullRequest, input.autoSettleOnMerge)) {
285	return activityAtMs === null ? thread.createdAt : DateTime.makeUnsafe(activityAtMs);
286	}
287	const hasPullRequest =
288	links.length > 0 ||
289	pullRequest !== null ||
290	thread.linkedPullRequest != null ||
291	thread.branchPullRequest != null;
292	if (input.autoSettleKeepPullRequests === true && hasPullRequest) return null;
293	if (input.autoSettleAfterDays === null || activityAtMs === null) return null;
294	// A passed wake timer restarts inactivity. An earlier wake can only delay settlement.
295	const snoozedUntilMs = toMillis(thread.snoozedUntil);
296	const timerWakeAtMs =
297	snoozedUntilMs !== null && snoozedUntilMs <= input.nowMs ? snoozedUntilMs : null;
298	const inactivityAtMs =
299	latestMillis([activityAtMs, toMillis(thread.lastSnoozeWakeAt), timerWakeAtMs]) ?? activityAtMs;
300	return inactivityAtMs < input.nowMs - input.autoSettleAfterDays * DAY_MS
301	? DateTime.makeUnsafe(inactivityAtMs)
302	: null;
303	}
304
305	export class ThreadSettlementServiceV2 extends Context.Service<
306	ThreadSettlementServiceV2,
307	{
308	readonly start: () => Effect.Effect<void, never, Scope.Scope>;
309	readonly drain: Effect.Effect<void>;
310	}
311	>()("t3/orchestration-v2/ThreadSettlementService/ThreadSettlementServiceV2") {}
312
313	function autoSettlementConfigured(settings: import("@t3tools/contracts").ServerSettings): boolean {
314	if (settings.sidebarAutoSettleOnMerge || settings.sidebarAutoSettleAfterDays !== null) {
315	return true;
316	}
317	return Object.values(settings.projectSettingsOverrides).some(
318	(entry) =>
319	entry.sidebarAutoSettleOnMerge === true ||
320	(entry.sidebarAutoSettleAfterDays !== undefined && entry.sidebarAutoSettleAfterDays !== null),
321	);
322	}
323
324	/** Identity of every settlement input, so unrelated settings edits do not trigger a sweep. */
325	/** @internal Exported for tests. */
326	export function autoSettlementSettingsKey(
327	settings: import("@t3tools/contracts").ServerSettings,
328	): string {
329	return JSON.stringify([
330	settings.sidebarAutoSettleOnMerge,
331	settings.sidebarAutoSettleAfterDays,
332	settings.sidebarAutoSettleKeepPullRequests,
333	// Only entries that touch settlement, in a stable order, so a project
334	// override on an unrelated key does not queue a sweep. JSON drops
335	// undefined, so inherit (absent) and never (null) need distinct marks.
336	Object.entries(settings.projectSettingsOverrides)
337	.filter(
338	([, entry]) =>
339	entry.sidebarAutoSettleOnMerge !== undefined ||
340	entry.sidebarAutoSettleAfterDays !== undefined ||
341	entry.sidebarAutoSettleKeepPullRequests !== undefined,
342	)
343	.sort(([left], [right]) => left.localeCompare(right))
344	.map(([projectId, entry]) => [
345	projectId,
346	entry.sidebarAutoSettleOnMerge ?? "inherit",
347	entry.sidebarAutoSettleAfterDays === undefined
348	? "inherit"
349	: entry.sidebarAutoSettleAfterDays,
350	entry.sidebarAutoSettleKeepPullRequests ?? "inherit",
351	]),
352	]);
353	}
354
355	export const make = Effect.gen(function* () {
356	const orchestrator = yield* Orchestrator.OrchestratorV2;
357	const projections = yield* ProjectionStore.ProjectionStoreV2;
358	const projectStore = yield* ProjectStore.ProjectStoreV2;
359	const settingsService = yield* ServerSettings.ServerSettingsService;
360	const git = yield* GitManager.GitManager;
361	const pullRequests = yield* PullRequestService.PullRequestService;
362	const crypto = yield* Crypto.Crypto;
363	const fileSystem = yield* FileSystem.FileSystem;
364	const terminals = yield* TerminalManager.TerminalManager;
365	const previews = yield* PreviewManager.PreviewManager;
366	const projectScripts = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner;
367	const managedWorkspaces = yield* ManagedWorkspaces.ManagedWorkspaces;
368	const devServer = yield* DevServer.DevServer;
369	const mergebot = yield* MergebotQueue.MergebotQueue;
370	const eventStore = yield* OrchestrationEventStore.OrchestrationEventStore;
371	// Settling a settled thread re-emits thread.settled with the same settledAt,
372	// so this keeps the settle action to one run per settlement.
373	const settleActionRunAt = new Map<ThreadId, number>();
374
375	const configuredRepository = (reference: {
376	readonly host?: string | undefined;
377	readonly repository: string;
378	}) =>
379	reference.host?.toLowerCase() === "github.com"
380	? mergebot.configuration?.repositories.find(
381	(repository) => repository.toLowerCase() === reference.repository.toLowerCase(),
382	)
383	: undefined;
384
385	const hasOpenResources = Effect.fn("ThreadSettlementServiceV2.hasOpenResources")(function* (
386	threadId: ThreadId,
387	) {
388	if ((yield* terminals.list({ threadId })).length > 0) return true;
389	return (yield* previews.list({ threadId })).sessions.length > 0;
390	});
391
392	const stopUnusedDevServer = Effect.fn("ThreadSettlementServiceV2.stopUnusedDevServer")(
393	(threadId: ThreadId) => devServer.stopUnused(threadId),
394	(effect, threadId) =>
395	effect.pipe(
396	Effect.catchCause((cause) =>
397	Cause.hasInterruptsOnly(cause)
398	? Effect.failCause(cause)
399	: Effect.logWarning("stopping an unused development server failed", {
400	threadId,
401	cause: Cause.pretty(cause),
402	}),
403	),
404	),
405	);
406
407	const archiveSettledThreads = Effect.fn("ThreadSettlementServiceV2.archiveSettledThreads")(
408	function* (threadId?: ThreadId) {
409	const snapshot =
410	threadId === undefined
411	? yield* projections.getShellSnapshot()
412	: yield* Effect.gen(function* () {
413	const snapshotSequence = yield* eventStore.latestAgentSequence(threadId);
414	const thread = yield* projections.getThreadShell(threadId);
415	return {
416	snapshotSequence,
417	threads: thread === null ? [] : [thread],
418	archivedThreads: [],
419	};
420	});
421	const parked = [...snapshot.threads, ...snapshot.archivedThreads].filter(
422	(thread) => thread.settledOverride === "settled" || thread.archivedAt !== null,
423	);
424	yield* Effect.forEach(parked, (thread) => stopUnusedDevServer(thread.id), {
425	concurrency: 8,
426	discard: true,
427	});
428	const nowMs = DateTime.toEpochMillis(yield* DateTime.now);
429	const boundary = lastSaturdayArchiveBoundary(nowMs);
430	const candidates = snapshot.threads
431	.filter(isManagedWorkspaceArchiveCandidate)
432	.filter(
433	(thread) =>
434	(threadId === undefined || thread.id === threadId) &&
435	!threadHasQueuedTurnStart(thread, nowMs),
436	);
437	const readPullRequest = Effect.fn("ThreadSettlementServiceV2.observePullRequest")(function* (
438	reference: PullRequestRef,
439	) {
440	const repository = configuredRepository(reference);
441	if (repository === undefined) {
442	yield* pullRequests.invalidate({ reference }, { notifyReaders: false });
443	const detail = yield* pullRequests.detail({ ...reference, allowStale: false });
444	return { ...detail, failureIdentity: null };
445	}
446	const detail = yield* mergebot.readPullRequest({ repository, number: reference.number });
447	if (detail === null) return null;
448	const failedChecks = detail.checks.filter((check) =>
449	["failure", "action_required", "timed_out", "startup_failure"].includes(
450	check.conclusion ?? "",
451	),
452	);
453	const failed =
454	detail.lifecycle === "open" &&
455	(failedChecks.length > 0 ||
456	detail.ciStatus === "failed" ||
457	detail.queueStatus === "ci_failed");
458	const identity = encodeCiFailureIdentity(
459	failedChecks.toSorted((left, right) => left.id.localeCompare(right.id)),
460	);
461	return {
462	...detail,
463	repository,
464	state: detail.lifecycle,
465	failureIdentity: failed
466	? Hex.encode(yield* crypto.digest("SHA-256", new TextEncoder().encode(identity)))
467	: null,
468	};
469	});
470	const reads = new Map<string, ReturnType<typeof readPullRequest>>();
471	const targets = (yield* Effect.forEach(
472	candidates,
473	(thread) =>
474	Effect.gen(function* () {
475	const links = visibleThreadPullRequests(thread.pullRequests ?? []);
476	const workPullRequest = links.some(
477	(link) =>
478	normalizeThreadPullRequestKey(link).host === "github.com" &&
479	isMergebotRepository(mergebot.configuration ?? undefined, link.repository),
480	);
481	const work =
482	links.length > 0 &&
483	(workPullRequest || (yield* managedWorkspaces.isManagedThread(thread.id)));
484	return { thread, links, work };
485	}).pipe(
486	Effect.catchTags({
487	ManagedWorkspaceError: () =>
488	Effect.logWarning("automatic thread archival workspace lookup skipped", {
489	threadId: thread.id,
490	}).pipe(Effect.as(null)),
491	}),
492	),
493	{ concurrency: 8 },
494	)).filter((target) => target !== null);
495	for (const { thread, links, work } of targets) {
496	if (!work) continue;
497	for (const link of links) {
498	const reference = {
499	projectId: thread.projectId,
500	host: link.host,
501	repository: link.repository,
502	number: link.number,
503	};
504	const key = `${reference.projectId}:${threadPullRequestKeyOf(reference)}`;
505	if (!reads.has(key)) reads.set(key, yield* Effect.cached(readPullRequest(reference)));
506	}
507	}
508	yield* Effect.forEach(
509	targets,
510	({ thread, links, work }) =>
511	Effect.gen(function* () {
512	let merged = false;
513	if (work) {
514	const observed = yield* Effect.forEach(
515	links,
516	(link) =>
517	reads.get(`${thread.projectId}:${threadPullRequestKeyOf(link)}`) ??
518	Effect.die("Pull request observation was not prepared"),
519	);
520	for (const failed of observed) {
521	if (failed === null || failed.failureIdentity === null) continue;
522	const itemId = TurnItemId.make(
523	`server:ci-failed:${thread.id}:${failed.repository}:${failed.number}:${failed.headSha}:${failed.failureIdentity}`,
524	);
525	const previous = yield* projections.getTurnItem({ threadId: thread.id, itemId });
526	if (previous === null) {
527	const uuid = yield* crypto.randomUUIDv4();
528	yield* orchestrator.dispatch({
529	type: "thread.ci-failed",
530	commandId: CommandId.make(`server:ci-failed:${thread.id}:${uuid}`),
531	threadId: thread.id,
532	snapshotSequence: snapshot.snapshotSequence,
533	itemId,
534	pullRequestNumber: failed.number,
535	url: failed.url,
536	});
537	return;
538	}
539	}
540	merged =
541	observed.length > 0 && observed.every((detail) => detail?.state === "merged");
542	}
543	if (!merged) {
544	const settledAt = yield* eventStore.getThreadSettlementTime(thread.id);
545	if (settledAt === null || DateTime.toEpochMillis(settledAt) > boundary) return;
546	}
547	const uuid = yield* crypto.randomUUIDv4();
548	yield* orchestrator.dispatch({
549	type: "thread.archive",
550	commandId: CommandId.make(`server:auto-archive:${thread.id}:${uuid}`),
551	threadId: thread.id,
552	snapshotSequence: snapshot.snapshotSequence,
553	});
554	}).pipe(
555	Effect.catchCause((cause) =>
556	Cause.hasInterruptsOnly(cause)
557	? Effect.failCause(cause)
558	: Effect.logWarning("automatic thread archival skipped", {
559	threadId: thread.id,
560	cause: Cause.pretty(cause),
561	}),
562	),
563	),
564	{ concurrency: 8, discard: true },
565	);
566	},
567	);
568
569	const sweep = Effect.fn("ThreadSettlementServiceV2.sweep")(function* (
570	mergedPullRequest: PullRequestService.PullRequestMergeEvent | null,
571	threadId?: ThreadId,
572	) {
573	yield* archiveSettledThreads(threadId);
574	const settings = yield* settingsService.getSettings;
575	if (!autoSettlementConfigured(settings)) {
576	return;
577	}
578	// A sweep for one thread reads only that thread's candidate row.
579	const threads = yield* projections.getSettlementCandidates(threadId);
580	if (threads.length === 0) return;
581	const projectShells = yield* projectStore.listShells();
582	const nowMs = DateTime.toEpochMillis(yield* DateTime.now);
583	const projects = new Map(projectShells.map((project) => [project.id, project]));
584	const workRepositoryFor = (thread: (typeof threads)[number]) => {
585	for (const link of visibleThreadPullRequests(thread.pullRequests ?? [])) {
586	const repository = configuredRepository(normalizeThreadPullRequestKey(link));
587	if (repository !== undefined) return repository;
588	}
589	const reference = thread.linkedPullRequest ?? thread.branchPullRequest;
590	if (reference != null) {
591	const repository = configuredRepository(legacyThreadPullRequestKey(reference));
592	if (repository !== undefined) return repository;
593	}
594	const canonical = projects
595	.get(thread.projectId)
596	?.repositoryIdentity?.canonicalKey.toLowerCase();
597	return mergebot.configuration?.repositories.find(
598	(repository) => canonical === `github.com/${repository.toLowerCase()}`,
599	);
600	};
601	// A merge event re-sweeps every candidate, not just the threads linked to
602	// the merged pull request: most threads carry no link and settle from
603	// their branch lookup, which would otherwise wait for the next minute's
604	// sweep on a possibly stale cached answer.
605	const candidates = threads.filter((thread) => isAutoSettlementCandidate(thread, nowMs));
606
607	const settleThread = Effect.fn("ThreadSettlementServiceV2.settleThread")(
608	function* (thread: (typeof candidates)[number], pullRequest: SettlementPullRequest | null) {
609	if (yield* hasOpenResources(thread.id)) return null;
610	const currentSettings = resolveProjectSettings(
611	yield* settingsService.getSettings,
612	thread.projectId,
613	).settings;
614	const decisionNow = yield* DateTime.now;
615	const settledAt = resolveAutoSettlementAt({
616	thread,
617	pullRequest,
618	usePullRequestSnapshots: workRepositoryFor(thread) === undefined,
619	nowMs: DateTime.toEpochMillis(decisionNow),
620	autoSettleAfterDays: currentSettings.sidebarAutoSettleAfterDays,
621	autoSettleOnMerge: currentSettings.sidebarAutoSettleOnMerge,
622	autoSettleKeepPullRequests: currentSettings.sidebarAutoSettleKeepPullRequests,
623	});
624	if (settledAt === null) return thread;
625	const uuid = yield* crypto.randomUUIDv4();
626	yield* orchestrator.dispatch({
627	type: "thread.auto-settle",
628	commandId: CommandId.make(`server:auto-settle:${thread.id}:${uuid}`),
629	threadId: thread.id,
630	snapshotAt: thread.updatedAt,
631	settledAt,
632	});
633	return null;
634	},
635	(effect, thread) =>
636	effect.pipe(
637	Effect.catchCause((cause) =>
638	Cause.hasInterruptsOnly(cause)
639	? Effect.failCause(cause)
640	: Effect.logWarning("automatic thread settlement skipped", {
641	threadId: thread.id,
642	cause: Cause.pretty(cause),
643	}).pipe(Effect.as(null)),
644	),
645	),
646	);
647
648	// Inactivity is entirely projection-backed. Complete those decisions before
649	// a source-control lookup can delay or fail an otherwise eligible thread.
650	const lookupCandidates = (yield* Effect.forEach(
651	candidates,
652	(thread) => settleThread(thread, null),
653	{ concurrency: 8 },
654	))
655	.filter((thread) => thread !== null)
656	.filter(
657	(thread) =>
658	visibleThreadPullRequests(thread.pullRequests ?? []).length === 0 ||
659	workRepositoryFor(thread) !== undefined,
660	);
661	// Use the same cwd as the sidebar so both paths share GitManager's PR cache.
662	const lookupCwdByThreadId = new Map<string, string>();
663	yield* Effect.forEach(
664	lookupCandidates,
665	(thread) =>
666	Effect.gen(function* () {
667	const project = projects.get(thread.projectId);
668	if (project === undefined || thread.branch === null) return;
669	const worktreeExists =
670	thread.worktreePath !== null &&
671	(yield* fileSystem.exists(thread.worktreePath).pipe(Effect.orElseSucceed(() => false)));
672	lookupCwdByThreadId.set(
673	thread.id,
674	worktreeExists && thread.worktreePath !== null
675	? thread.worktreePath
676	: project.workspaceRoot,
677	);
678	}),
679	{ concurrency: 8, discard: true },
680	);
681	if (mergedPullRequest !== null) {
682	// The merge just confirmed a terminal state the lookup caches can still
683	// call open (branch answers live two minutes, the sweep runs every
684	// minute). Drop the swept checkouts' cached answers so the merge settles
685	// its branch threads now instead of on a later sweep. Threads linked to
686	// the merged pull request settle from the event itself below and need no
687	// lookup, so they are absent from this map by construction.
688	const cwds = [
689	...new Set(
690	lookupCandidates
691	.filter((thread) => workRepositoryFor(thread) === undefined)
692	.flatMap((thread) => {
693	const cwd = lookupCwdByThreadId.get(thread.id);
694	return cwd === undefined ? [] : [cwd];
695	}),
696	),
697	];
698	yield* Effect.forEach(cwds, (cwd) => git.invalidateStatus(cwd), {
699	concurrency: 8,
700	discard: true,
701	});
702	}
703	const lookupKey = (thread: (typeof lookupCandidates)[number]) => {
704	const reference = thread.linkedPullRequest ?? thread.branchPullRequest;
705	if (reference != null) {
706	return JSON.stringify([
707	"linked",
708	reference.projectId,
709	reference.repository,
710	reference.number,
711	lookupCwdByThreadId.get(thread.id),
712	thread.branch,
713	]);
714	}
715	if (thread.branch === null) return JSON.stringify(["none", thread.id]);
716	const cwd = lookupCwdByThreadId.get(thread.id);
717	return JSON.stringify(
718	cwd === undefined ? ["missing-project", thread.id] : ["branch", cwd, thread.branch],
719	);
720	};
721	const groups = Map.groupBy(lookupCandidates, lookupKey);
722
723	const wouldSettle = Effect.fn("ThreadSettlementServiceV2.wouldSettle")(function* (
724	group: ReadonlyArray<(typeof lookupCandidates)[number]>,
725	pullRequest: SettlementPullRequest,
726	) {
727	const currentSettings = yield* settingsService.getSettings;
728	const decisionNow = yield* DateTime.now;
729	return group.some((thread) => {
730	const { settings } = resolveProjectSettings(currentSettings, thread.projectId);
731	return (
732	resolveAutoSettlementAt({
733	thread,
734	pullRequest,
735	usePullRequestSnapshots: workRepositoryFor(thread) === undefined,
736	nowMs: DateTime.toEpochMillis(decisionNow),
737	autoSettleAfterDays: settings.sidebarAutoSettleAfterDays,
738	autoSettleOnMerge: settings.sidebarAutoSettleOnMerge,
739	autoSettleKeepPullRequests: settings.sidebarAutoSettleKeepPullRequests,
740	}) !== null
741	);
742	});
743	});
744
745	const pullRequestFor = Effect.fn("ThreadSettlementServiceV2.pullRequestFor")(function* (
746	group: ReadonlyArray<(typeof lookupCandidates)[number]>,
747	) {
748	const thread = group[0]!;
749	const reference = thread.linkedPullRequest ?? thread.branchPullRequest;
750	const workRepository = workRepositoryFor(thread);
751	if (workRepository !== undefined) {
752	const links = visibleThreadPullRequests(thread.pullRequests ?? []);
753	const references =
754	links.length > 0
755	? links.map(normalizeThreadPullRequestKey)
756	: reference === null || reference === undefined
757	? []
758	: [legacyThreadPullRequestKey(reference)];
759	const observed =
760	references.length > 0
761	? yield* Effect.forEach(references, (link) =>
762	Effect.gen(function* () {
763	const repository = configuredRepository(link);
764	if (repository !== undefined) {
765	const detail = yield* mergebot.readPullRequest({
766	repository,
767	number: link.number,
768	});
769	return detail === null
770	? null
771	: {
772	state: detail.lifecycle,
773	mergedAt: detail.mergedAt ?? null,
774	closedAt: detail.closedAt ?? null,
775	};
776	}
777	const detail = yield* pullRequests.summary(
778	{ projectId: thread.projectId, ...link },
779	{ recoverTransientFailure: false },
780	);
781	return {
782	state: detail.state,
783	mergedAt: detail.mergedAt ?? null,
784	closedAt: detail.closedAt ?? null,
785	};
786	}),
787	)
788	: thread.branch === null
789	? []
790	: [
791	yield* mergebot
792	.findPullRequest({ repository: workRepository, headBranch: thread.branch })
793	.pipe(
794	Effect.map((detail) =>
795	detail === null
796	? null
797	: {
798	state: detail.lifecycle,
799	mergedAt: detail.mergedAt ?? null,
800	closedAt: detail.closedAt ?? null,
801	},
802	),
803	),
804	];
805	const terminal = observed.filter((detail) => detail !== null);
806	if (
807	terminal.length === 0 ||
808	terminal.length !== observed.length ||
809	terminal.some(
810	(detail) =>
811	detail.state === "open" ||
812	!Number.isFinite(
813	Date.parse((detail.state === "merged" ? detail.mergedAt : detail.closedAt) ?? ""),
814	),
815	)
816	)
817	return null;
818	if (references.length > 0 && thread.branch !== null) {
819	const current = yield* mergebot.findPullRequest({
820	repository: workRepository,
821	headBranch: thread.branch,
822	});
823	if (current?.lifecycle === "open") return null;
824	}
825	return terminal.reduce((latest, detail) => {
826	const latestAt = Date.parse(
827	(latest.state === "merged" ? latest.mergedAt : latest.closedAt) ?? "",
828	);
829	const detailAt = Date.parse(
830	(detail.state === "merged" ? detail.mergedAt : detail.closedAt) ?? "",
831	);
832	return detailAt > latestAt ? detail : latest;
833	});
834	}
835	if (reference != null) {
836	// The event carries the merged state, so only the threads linked to
837	// that exact pull request settle from it. Every other linked thread
838	// falls through to a fresh summary lookup below: the merge sweep
839	// covers all candidates, and an unrelated merge must never settle
840	// them.
841	const matchesMerge =
842	mergedPullRequest !== null &&
843	reference.projectId === mergedPullRequest.projectId &&
844	reference.repository.toLowerCase() === mergedPullRequest.repository.toLowerCase() &&
845	reference.number === mergedPullRequest.number;
846	if (!matchesMerge && !projects.has(reference.projectId)) {
847	return yield* Effect.die(new Error("linked pull request project not found"));
848	}
849	const summary = matchesMerge
850	? ({
851	state: "merged",
852	closedAt: null,
853	mergedAt: mergedPullRequest.mergedAt,
854	} satisfies SettlementPullRequest)
855	: yield* pullRequests.summary(
856	{
857	projectId: reference.projectId,
858	repository: reference.repository,
859	number: reference.number,
860	},
861	{ recoverTransientFailure: false },
862	);
863	const terminal = {
864	state: summary.state,
865	closedAt: summary.closedAt ?? null,
866	mergedAt: summary.mergedAt ?? null,
867	} satisfies SettlementPullRequest;
868	const cwd = lookupCwdByThreadId.get(thread.id);
869	if (summary.state !== "open" && thread.branch !== null && cwd !== undefined) {
870	// Recheck reused branches only when this sweep would settle a thread.
871	// Eligibility that changes after this check waits for the next sweep.
872	if (!(yield* wouldSettle(group, terminal))) return undefined;
873	const current = yield* git.branchPullRequest(
874	{ cwd, branch: thread.branch },
875	{ refresh: true },
876	);
877	const project = projects.get(thread.projectId);
878	if (
879	current?.state === "open" &&
880	project !== undefined &&
881	pullRequestMatchesProject(current, project)
882	) {
883	return current;
884	}
885	}
886	return terminal;
887	}
888	if (thread.branch === null) return null;
889	const cwd = lookupCwdByThreadId.get(thread.id);
890	if (cwd === undefined) {
891	return yield* Effect.die(new Error("thread project not found"));
892	}
893	return yield* git.branchPullRequest({ cwd, branch: thread.branch });
894	});
895
896	yield* Effect.forEach(
897	groups.values(),
898	(group) =>
899	Effect.gen(function* () {
900	const pullRequest = yield* pullRequestFor(group);
901	if (pullRequest === undefined) return;
902	yield* Effect.forEach(group, (thread) => settleThread(thread, pullRequest), {
903	discard: true,
904	});
905	}).pipe(
906	Effect.catchCause((cause) =>
907	Cause.hasInterruptsOnly(cause)
908	? Effect.failCause(cause)
909	: Effect.logWarning("automatic thread settlement skipped", {
910	threadIds: group.map((thread) => thread.id),
911	cause: Cause.pretty(cause),
912	}),
913	),
914	),
915	{ concurrency: 8, discard: true },
916	);
917	});
918
919	const runSweep = (
920	mergedPullRequest: PullRequestService.PullRequestMergeEvent | null,
921	threadId?: ThreadId,
922	) =>
923	sweep(mergedPullRequest, threadId).pipe(
924	Effect.catchCause((cause) =>
925	Cause.hasInterruptsOnly(cause)
926	? Effect.failCause(cause)
927	: Effect.logWarning("automatic thread settlement sweep failed", {
928	cause: Cause.pretty(cause),
929	}),
930	),
931	);
932	const worker = yield* makeDrainableWorker((threadId: ThreadId | undefined) =>
933	runSweep(null, threadId),
934	);
935
936	const cleanUpSettledThread = Effect.fn("ThreadSettlementServiceV2.cleanUpSettledThread")(
937	function* (settlement: OrchestrationV2AppThread, automaticSequence: number | undefined) {
938	const threadId = settlement.id;
939	const settledAtMs = toMillis(settlement.settledAt);
940	// A thread re-engaged before this event ran keeps its shells.
941	const settled = yield* projections.getThread(threadId);
942	if (settled.settledOverride !== "settled" || toMillis(settled.settledAt) !== settledAtMs)
943	return;
944	if (automaticSequence !== undefined) {
945	if (yield* hasOpenResources(threadId)) {
946	const uuid = yield* crypto.randomUUIDv4();
947	yield* orchestrator.dispatch({
948	type: "thread.unsettle",
949	commandId: CommandId.make(`server:auto-settle-resources:${threadId}:${uuid}`),
950	threadId,
951	reason: "user",
952	snapshotSequence: automaticSequence,
953	});
954	return;
955	}
956	} else {
957	yield* terminals.close({ threadId });
958	const current = yield* projections.getThread(threadId);
959	if (current.settledOverride !== "settled" || toMillis(current.settledAt) !== settledAtMs)
960	return;
961	yield* previews.close({ threadId });
962	}
963	if (yield* managedWorkspaces.isManagedThread(threadId)) return;
964	const worktreePath = settled.worktreePath;
965	if (worktreePath === null || !(yield* fileSystem.exists(worktreePath))) return;
966	// Closing and the worktree check wait on I/O. A thread re-engaged
967	// meanwhile is working again, so its worktree is no place for cleanup.
968	const thread = yield* projections.getThread(threadId);
969	if (thread.settledOverride !== "settled" || toMillis(thread.settledAt) !== settledAtMs)
970	return;
971	if (settledAtMs === null || settleActionRunAt.get(threadId) === settledAtMs) return;
972	const run = yield* projectScripts.runForThread({
973	threadId,
974	projectId: thread.projectId,
975	worktreePath,
976	trigger: "settle",
977	// A clean exit closes the script's shell so it does not hold the worktree.
978	observeCompletion: {},
979	});
980	// Recorded after a successful start, so a failed start retries on the next event.
981	settleActionRunAt.set(threadId, settledAtMs);
982	if (run.status === "started" && run.completion) {
983	yield* run.completion.pipe(Effect.forkDetach);
984	}
985	},
986	(effect, settlement) =>
987	effect.pipe(
988	Effect.catchCause((cause) =>
989	Cause.hasInterruptsOnly(cause)
990	? Effect.failCause(cause)
991	: Effect.logWarning("cleaning up a settled thread failed", {
992	threadId: settlement.id,
993	cause: Cause.pretty(cause),
994	}),
995	),
996	),
997	);
998
999	const processEvent = Effect.fn("ThreadSettlementServiceV2.processEvent")(function* (
1000	stored: OrchestrationV2StoredEvent,
1001	) {
1002	const event = stored.event;
1003	switch (event.type) {
1004	case "thread.settled":
1005	return yield* cleanUpSettledThread(
1006	event.payload,
1007	stored.commandId?.startsWith("server:auto-settle:") ? stored.sequence : undefined,
1008	).pipe(Effect.andThen(worker.enqueue(event.threadId)));
1009	case "thread.archived":
1010	case "thread.pull-request-link-updated":
1011	case "thread.pull-request-synced":
1012	case "provider-session.detached":
1013	return yield* worker.enqueue(event.threadId);
1014	case "provider-session.updated":
1015	if (event.payload.status !== "starting" && event.payload.status !== "running") {
1016	yield* worker.enqueue(event.threadId);
1017	}
1018	return;
1019	case "run.updated":
1020	if (
1021	["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes(
1022	event.payload.status,
1023	)
1024	) {
1025	yield* worker.enqueue(event.threadId);
1026	}
1027	return;
1028	default:
1029	return;
1030	}
1031	});
1032
1033	const start: ThreadSettlementServiceV2["Service"]["start"] = Effect.fn(
1034	"ThreadSettlementServiceV2.start",
1035	)(function* () {
1036	const settingsChanges = yield* settingsService.subscribeChanges;
1037	const mergedPullRequests = yield* pullRequests.subscribeMerges;
1038	const afterSequence = yield* eventStore.latestAgentSequence().pipe(Effect.orDie);
1039	const events = orchestrator.streamStoredEventsFrom({ afterSequence });
1040	const initialSettings = yield* settingsService.getSettings.pipe(Effect.orDie);
1041	let lastSettlementSettings = autoSettlementSettingsKey(initialSettings);
1042	yield* forkParked(
1043	Effect.gen(function* () {
1044	yield* worker.enqueue(undefined);
1045	yield* worker.drain;
1046	}).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid),
1047	);
1048	yield* forkParked(
1049	Stream.runForEach(settingsChanges, (settings) => {
1050	const key = autoSettlementSettingsKey(settings);
1051	if (key === lastSettlementSettings) {
1052	return Effect.void;
1053	}
1054	lastSettlementSettings = key;
1055	return worker.enqueue(undefined);
1056	}),
1057	);
1058	yield* forkParked(Stream.runForEach(mergedPullRequests, (event) => runSweep(event)));
1059	yield* forkParked(
1060	Stream.runForEach(mergebot.streamChanges.pipe(Stream.debounce("100 millis")), () =>
1061	worker.enqueue(undefined),
1062	),
1063	);
1064	yield* forkParked(
1065	Stream.runForEach(events, processEvent).pipe(
1066	Effect.catchCause((cause) =>
1067	Effect.logWarning("Thread settlement event stream failed", { cause }),
1068	),
1069	),
1070	);
1071	});
1072
1073	return { start, drain: worker.drain } satisfies ThreadSettlementServiceV2["Service"];
1074	});
1075
1076	export const layer = Layer.effect(ThreadSettlementServiceV2, make);