1import {
2 backgroundWorkHoldsCompletion,
3 isPullRequestWatchTask,
4 turnItemUpdateCanEndBackgroundWork,
5} from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
6import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
7import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests";
8import {
9 CommandId,
10 isOrchestrationV2WorkActive,
11 type ThreadId,
12 type OrchestrationV2DomainEvent,
13 type OrchestrationV2ThreadShell,
14} from "@t3tools/contracts";
15import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
16import * as Cause from "effect/Cause";
17import * as Context from "effect/Context";
18import * as SecureRandom from "effect/crypto/SecureRandom";
19import * as DateTime from "effect/DateTime";
20import * as Effect from "effect/Effect";
21import * as FiberSet from "effect/FiberSet";
22import * as FileSystem from "effect/FileSystem";
23import * as Path from "effect/Path";
24import * as Schedule from "effect/Schedule";
25import * as Semaphore from "effect/Semaphore";
26import type * as Scope from "effect/Scope";
27import * as Stream from "effect/Stream";
28
29import * as GitManager from "../git/GitManager.ts";
30import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts";
31import * as PullRequestService from "../pullRequest/PullRequestService.ts";
32import * as ServerSettings from "../serverSettings.ts";
33import { forkParked } from "../serverActivation.ts";
34import * as TerminalManager from "../terminal/Manager.ts";
35import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
36import * as ProjectStore from "./ProjectStore.ts";
37import { sourceControlDefinitions } from "../sourceControl/sourceControlDefinitions.ts";
38import * as Orchestrator from "./Orchestrator.ts";
39import * as ProjectionStore from "./ProjectionStore.ts";
40
41export interface SettlementPullRequest {
42 readonly state: "open" | "closed" | "merged";
43 readonly closedAt?: string | null;
44 readonly mergedAt?: string | null;
45}
46
47const DAY_MS = 24 * 60 * 60 * 1_000;
48const QUEUED_TURN_START_GRACE_MS = 2 * 60 * 1_000;
49
50function toMillis(value: DateTime.Utc | null | undefined): number | null {
51 return value == null ? null : DateTime.toEpochMillis(value);
52}
53
54function latestMillis(values: ReadonlyArray<number | null>): number | null {
55 let latest: number | null = null;
56 for (const value of values) {
57 if (value === null) continue;
58 if (latest === null || value > latest) latest = value;
59 }
60 return latest;
61}
62
63function pullRequestMatchesProject(
64 pullRequest: GitManager.GitBranchPullRequest,
65 project: {
66 readonly repositoryIdentity?: { readonly canonicalKey: string } | null | undefined;
67 },
68): boolean {
69 return (
70 pullRequest.repositoryKey !== null &&
71 project.repositoryIdentity != null &&
72 sourceControlDefinitions.canonicalRepositoryKey(pullRequest.repositoryKey) ===
73 sourceControlDefinitions.canonicalRepositoryKey(project.repositoryIdentity.canonicalKey)
74 );
75}
76
77type SettlementPullRequestLink = NonNullable<
78 ProjectionStore.ProjectionSettlementCandidate["pullRequests"]
79>[number];
80
81/** When the linked pull request merged or closed, or -Infinity if it has not. */
82function linkTerminalAtMs(link: SettlementPullRequestLink): number {
83 const snapshot = link.snapshot;
84 const value = snapshot?.state === "merged" ? snapshot.mergedAt : snapshot?.closedAt;
85 const timestamp = Date.parse(value ?? "");
86 return Number.isNaN(timestamp) ? Number.NEGATIVE_INFINITY : timestamp;
87}
88
89/** Exclude links added after merge or close; they record completed work rather than outcomes this thread awaits. */
90function settlementPullRequestLinks(
91 links: ReadonlyArray<SettlementPullRequestLink>,
92): ReadonlyArray<SettlementPullRequestLink> {
93 return visibleThreadPullRequests(links).filter((link) => {
94 if (link.snapshot === null || link.snapshot.state === "open") return true;
95 const linkedAtMs = Date.parse(link.linkedAt);
96 return Number.isNaN(linkedAtMs) || linkTerminalAtMs(link) >= linkedAtMs;
97 });
98}
99
100/** Absolute age bounds clock skew in both directions and expires stale pre-adoption data. */
101export function threadHasQueuedTurnStart(
102 thread: Pick<
103 OrchestrationV2ThreadShell,
104 | "latestUserMessageAt"
105 | "latestRunRequestedAt"
106 | "latestRunStartedAt"
107 | "latestRunCompletedAt"
108 | "latestRunId"
109 | "status"
110 >,
111 nowMs: number,
112): boolean {
113 const messageAtMs = toMillis(thread.latestUserMessageAt);
114 if (messageAtMs === null || thread.status === "failed") return false;
115 const age = nowMs - messageAtMs;
116 if (Number.isNaN(age) || Math.abs(age) > QUEUED_TURN_START_GRACE_MS) return false;
117 if (thread.latestRunId === null) return true;
118 return [
119 toMillis(thread.latestRunRequestedAt),
120 toMillis(thread.latestRunStartedAt),
121 toMillis(thread.latestRunCompletedAt),
122 ].every((value) => value === null || value < messageAtMs);
123}
124
125function pullRequestSettles(
126 thread: Pick<
127 ProjectionStore.ProjectionSettlementCandidate,
128 "createdAt" | "latestUserAuthoredMessageAt"
129 >,
130 pullRequest: SettlementPullRequest,
131 autoSettleOnMerge: boolean,
132): boolean {
133 if (pullRequest.state !== "closed" && (pullRequest.state !== "merged" || !autoSettleOnMerge)) {
134 return false;
135 }
136 const terminalAt = pullRequest.state === "merged" ? pullRequest.mergedAt : pullRequest.closedAt;
137 if (terminalAt == null) return false;
138 const userAnchorMs = latestMillis([
139 toMillis(thread.createdAt),
140 toMillis(thread.latestUserAuthoredMessageAt),
141 ]);
142 if (userAnchorMs === null) return false;
143 const pullRequestAtMs = Date.parse(terminalAt);
144 if (Number.isNaN(pullRequestAtMs)) return false;
145 return pullRequestAtMs >= userAnchorMs;
146}
147
148export function isAutoSettlementCandidate(
149 thread: Omit<ProjectionStore.ProjectionSettlementCandidate, "latestUserAuthoredMessageAt">,
150 nowMs: number,
151): boolean {
152 if (thread.archivedAt !== null || thread.settledOverride !== null) return false;
153 if (thread.pinnedAt != null || thread.autoSettleDisabledAt != null) return false;
154 if (thread.pendingRuntimeRequest !== null) return false;
155 // A guest is waiting on the owner; settling would hide the Request.
156 if ((thread.pendingGuestMessageCount ?? 0) > 0) return false;
157 if (thread.activityRunStatus != null) return false;
158 const pendingTasks = thread.pendingBackgroundTasks ?? [];
159 if (backgroundWorkHoldsCompletion(pendingTasks)) return false;
160 // Settling ends a watch, which would silently drop a reviewer's later comments.
161 if (pendingTasks.some(isPullRequestWatchTask)) return false;
162 if (threadHasQueuedTurnStart(thread, nowMs)) return false;
163 // Interrupted runs must keep the same snooze protection the sidebar shows.
164 return !isSnoozed(thread, nowMs);
165}
166
167/** Match client effectiveSnoozed so agents and sidebar agree. */
168export function isSnoozed(
169 thread: Pick<
170 ProjectionStore.ProjectionSettlementCandidate,
171 | "snoozedUntil"
172 | "snoozedAt"
173 | "latestRunCompletedAt"
174 | "status"
175 | "pendingRuntimeRequest"
176 | "pendingGuestMessageCount"
177 >,
178 nowMs: number,
179): boolean {
180 const snoozedUntilMs = toMillis(thread.snoozedUntil);
181 if (snoozedUntilMs === null || snoozedUntilMs <= nowMs) return false;
182 // Same as the client: a waiting guest message wakes the thread, and the snooze stays stored.
183 if ((thread.pendingGuestMessageCount ?? 0) > 0) return false;
184 const snoozedAtMs = toMillis(thread.snoozedAt);
185 if (
186 thread.pendingRuntimeRequest !== null &&
187 (thread.pendingRuntimeRequest.kind !== "user_input" ||
188 snoozedAtMs === null ||
189 DateTime.toEpochMillis(thread.pendingRuntimeRequest.createdAt) > snoozedAtMs)
190 ) {
191 return false;
192 }
193 const completedAtMs = toMillis(thread.latestRunCompletedAt);
194 const wokeOnError =
195 thread.status === "failed" &&
196 (snoozedAtMs === null || (completedAtMs !== null && completedAtMs > snoozedAtMs));
197 return !wokeOnError;
198}
199
200export function resolveAutoSettlementAt(input: {
201 readonly thread: ProjectionStore.ProjectionSettlementCandidate;
202 readonly pullRequest: SettlementPullRequest | null;
203 readonly nowMs: number;
204 readonly autoSettleAfterDays: number | null;
205 readonly autoSettleOnMerge: boolean;
206 readonly autoSettleKeepPullRequests?: boolean;
207}): DateTime.Utc | null {
208 const { thread } = input;
209 let pullRequest = input.pullRequest;
210 const links = settlementPullRequestLinks(thread.pullRequests ?? []);
211 if (links.some((link) => link.snapshot === null)) return null;
212 // An open link means the pull request has not finished, so only inactivity
213 // can settle the thread, never a merged or closed lookup.
214 const hasOpenLink = links.some((link) => link.snapshot?.state === "open");
215 if (hasOpenLink) {
216 pullRequest = null;
217 } else if (links.length > 0) {
218 const latest = links.reduce((current, candidate) =>
219 linkTerminalAtMs(candidate) > linkTerminalAtMs(current) ? candidate : current,
220 );
221 pullRequest =
222 latest.snapshot === null
223 ? null
224 : {
225 state: latest.snapshot.state,
226 mergedAt: latest.snapshot.mergedAt ?? null,
227 closedAt: latest.snapshot.closedAt ?? null,
228 };
229 }
230 if (!isAutoSettlementCandidate(thread, input.nowMs)) return null;
231 const activityAtMs =
232 latestMillis([
233 toMillis(thread.latestUserMessageAt),
234 toMillis(thread.latestRunRequestedAt),
235 toMillis(thread.latestRunStartedAt),
236 toMillis(thread.latestRunCompletedAt),
237 ]) ?? DateTime.toEpochMillis(thread.createdAt);
238 if (pullRequest !== null && pullRequestSettles(thread, pullRequest, input.autoSettleOnMerge)) {
239 return DateTime.makeUnsafe(activityAtMs);
240 }
241 const hasPullRequest =
242 hasOpenLink ||
243 pullRequest !== null ||
244 thread.linkedPullRequest != null ||
245 thread.branchPullRequest != null;
246 if (input.autoSettleKeepPullRequests === true && hasPullRequest) return null;
247 if (input.autoSettleAfterDays === null) return null;
248 const snoozedUntilMs = toMillis(thread.snoozedUntil);
249 const timerWakeAtMs =
250 snoozedUntilMs !== null && snoozedUntilMs <= input.nowMs ? snoozedUntilMs : null;
251 const inactivityAtMs =
252 latestMillis([activityAtMs, toMillis(thread.lastSnoozeWakeAt), timerWakeAtMs]) ?? activityAtMs;
253 return inactivityAtMs < input.nowMs - input.autoSettleAfterDays * DAY_MS
254 ? DateTime.makeUnsafe(inactivityAtMs)
255 : null;
256}
257
258export class ThreadSettlementServiceV2 extends Context.Service<
259 ThreadSettlementServiceV2,
260 {
261 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
262 readonly drain: Effect.Effect<void>;
263 }
264>()("t3/orchestration-v2/ThreadSettlementService/ThreadSettlementServiceV2") {}
265
266function autoSettlementConfigured(settings: import("@t3tools/contracts").ServerSettings): boolean {
267 if (settings.sidebarAutoSettleOnMerge || settings.sidebarAutoSettleAfterDays !== null) {
268 return true;
269 }
270 return Object.values(settings.projectSettingsOverrides).some(
271 (entry) =>
272 entry.sidebarAutoSettleOnMerge === true ||
273 (entry.sidebarAutoSettleAfterDays !== undefined && entry.sidebarAutoSettleAfterDays !== null),
274 );
275}
276
277/** @internal Exported for tests. */
278export function autoSettlementSettingsKey(
279 settings: import("@t3tools/contracts").ServerSettings,
280): string {
281 return JSON.stringify([
282 settings.sidebarAutoSettleOnMerge,
283 settings.sidebarAutoSettleAfterDays,
284 settings.sidebarAutoSettleKeepPullRequests,
285 Object.entries(settings.projectSettingsOverrides)
286 .filter(
287 ([, entry]) =>
288 entry.sidebarAutoSettleOnMerge !== undefined ||
289 entry.sidebarAutoSettleAfterDays !== undefined ||
290 entry.sidebarAutoSettleKeepPullRequests !== undefined,
291 )
292 .sort(([left], [right]) => left.localeCompare(right))
293 .map(([projectId, entry]) => [
294 projectId,
295 entry.sidebarAutoSettleOnMerge ?? "inherit",
296 entry.sidebarAutoSettleAfterDays === undefined
297 ? "inherit"
298 : entry.sidebarAutoSettleAfterDays,
299 entry.sidebarAutoSettleKeepPullRequests ?? "inherit",
300 ]),
301 ]);
302}
303
304export const make = Effect.gen(function* () {
305 const orchestrator = yield* Orchestrator.OrchestratorV2;
306 const projections = yield* ProjectionStore.ProjectionStoreV2;
307 const projectStore = yield* ProjectStore.ProjectStoreV2;
308 const settingsService = yield* ServerSettings.ServerSettingsService;
309 const git = yield* GitManager.GitManager;
310 const pullRequests = yield* PullRequestService.PullRequestService;
311 const random = yield* SecureRandom.SecureRandom;
312 const fileSystem = yield* FileSystem.FileSystem;
313 const path = yield* Path.Path;
314 // A shell merge or close sends no event, and the branch lookup caches an open pull request for
315 // a minute, so these threads' next lookup reads the host fresh.
316 const shellClosedThreads = new Set<ThreadId>();
317 const terminals = yield* TerminalManager.TerminalManager;
318 const projectScripts = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner;
319 // Repeated thread.settled retains settledAt; run the action only once per settlement.
320 const settleActionRunAt = new Map<ThreadId, number>();
321
322 const sweep = Effect.fn("ThreadSettlementServiceV2.sweep")(function* (
323 mergedPullRequest: PullRequestService.PullRequestMergeEvent | null,
324 threadId?: ThreadId,
325 ) {
326 const settings = yield* settingsService.getSettings;
327 const automaticSettlementEnabled = autoSettlementConfigured(settings);
328 const threads = yield* projections.getSettlementCandidates(threadId);
329 if (threads.length === 0) return;
330 const projectShells = yield* projectStore.listShells();
331 const nowMs = DateTime.toEpochMillis(yield* DateTime.now);
332 const projects = new Map(projectShells.map((project) => [project.id, project]));
333 // Merge events resweep all candidates because unlinked threads settle through branch lookup.
334 const candidates = threads.filter(
335 (thread) =>
336 (automaticSettlementEnabled || thread.lineage.relationshipToParent === "subagent") &&
337 isAutoSettlementCandidate(thread, nowMs),
338 );
339
340 // Runs per candidate thread, thousands per sweep on busy servers; the sweep owns the span.
341 const settleThread = Effect.fnUntraced(
342 function* (thread: (typeof candidates)[number], pullRequest: SettlementPullRequest | null) {
343 if (thread.lineage.relationshipToParent === "subagent") {
344 // The store returns a subagent only once its parent is done, so it
345 // settles with the parent and never runs its own lookup.
346 const uuid = yield* random.uuidV4();
347 yield* orchestrator.dispatch({
348 type: "thread.auto-settle",
349 commandId: CommandId.make(`server:auto-settle:${thread.id}:${uuid}`),
350 threadId: thread.id,
351 snapshotAt: thread.updatedAt,
352 });
353 return null;
354 }
355 // thread.auto-settle does not recheck settings, so read them per decision; reads are cached.
356 const currentSettings = resolveProjectSettings(
357 yield* settingsService.getSettings,
358 thread.projectId,
359 ).settings;
360 const decisionNow = yield* DateTime.now;
361 const settledAt = resolveAutoSettlementAt({
362 thread,
363 pullRequest,
364 nowMs: DateTime.toEpochMillis(decisionNow),
365 autoSettleAfterDays: currentSettings.sidebarAutoSettleAfterDays,
366 autoSettleOnMerge: currentSettings.sidebarAutoSettleOnMerge,
367 autoSettleKeepPullRequests: currentSettings.sidebarAutoSettleKeepPullRequests,
368 });
369 if (settledAt === null) return thread;
370 const uuid = yield* random.uuidV4();
371 yield* orchestrator.dispatch({
372 type: "thread.auto-settle",
373 commandId: CommandId.make(`server:auto-settle:${thread.id}:${uuid}`),
374 threadId: thread.id,
375 snapshotAt: thread.updatedAt,
376 settledAt,
377 });
378 return null;
379 },
380 (effect, thread) =>
381 effect.pipe(
382 Effect.catchCauseIf(
383 (cause) => !Cause.hasInterruptsOnly(cause),
384 (cause) =>
385 Effect.logWarning("automatic thread settlement skipped", {
386 threadId: thread.id,
387 cause: Cause.pretty(cause),
388 }).pipe(Effect.as(null)),
389 ),
390 ),
391 );
392
393 const lookupCandidates = (yield* Effect.forEach(
394 candidates,
395 (thread) => {
396 if (thread.lineage.relationshipToParent === "subagent") return settleThread(thread, null);
397 const projectSettings = resolveProjectSettings(settings, thread.projectId).settings;
398 // Skip no-op decisions; actual settlement still reads fresh settings before dispatch.
399 const settledAt = resolveAutoSettlementAt({
400 thread,
401 pullRequest: null,
402 nowMs,
403 autoSettleAfterDays: projectSettings.sidebarAutoSettleAfterDays,
404 autoSettleOnMerge: projectSettings.sidebarAutoSettleOnMerge,
405 autoSettleKeepPullRequests: projectSettings.sidebarAutoSettleKeepPullRequests,
406 });
407 return settledAt === null ? Effect.succeed(thread) : settleThread(thread, null);
408 },
409 { concurrency: 8 },
410 ))
411 .filter((thread) => thread !== null)
412 .filter((thread) => visibleThreadPullRequests(thread.pullRequests ?? []).length === 0)
413 .filter(
414 (thread) =>
415 thread.branch !== null ||
416 thread.linkedPullRequest != null ||
417 thread.branchPullRequest != null,
418 );
419 const lookupCwdByThreadId = new Map<string, string>();
420 yield* Effect.forEach(
421 lookupCandidates,
422 (thread) =>
423 Effect.gen(function* () {
424 const project = projects.get(thread.projectId);
425 if (project === undefined || thread.branch === null) return;
426 // A worktree folder can outlive its .git, and git there fails with not_a_repository.
427 // An unreadable .git keeps the worktree, so git fails there and the thread is skipped.
428 lookupCwdByThreadId.set(
429 thread.id,
430 thread.worktreePath !== null &&
431 (yield* fileSystem
432 .exists(path.join(thread.worktreePath, ".git"))
433 .pipe(Effect.orElseSucceed(() => true)))
434 ? thread.worktreePath
435 : project.workspaceRoot,
436 );
437 }),
438 { concurrency: 8, discard: true },
439 );
440 if (mergedPullRequest !== null) {
441 // Invalidate checkout caches on merge; cached open branches would delay settlement beyond this sweep.
442 const cwds = [...new Set(lookupCwdByThreadId.values())];
443 yield* Effect.forEach(cwds, (cwd) => git.invalidateStatus(cwd), {
444 concurrency: 8,
445 discard: true,
446 });
447 }
448 const lookupKey = (thread: (typeof lookupCandidates)[number]) => {
449 const reference = thread.linkedPullRequest ?? thread.branchPullRequest;
450 if (reference != null) {
451 return JSON.stringify([
452 "linked",
453 reference.projectId,
454 reference.repository,
455 reference.number,
456 lookupCwdByThreadId.get(thread.id),
457 thread.branch,
458 ]);
459 }
460 if (thread.branch === null) return JSON.stringify(["none", thread.id]);
461 const cwd = lookupCwdByThreadId.get(thread.id);
462 return JSON.stringify(
463 cwd === undefined ? ["missing-project", thread.id] : ["branch", cwd, thread.branch],
464 );
465 };
466 const groups = Map.groupBy(lookupCandidates, lookupKey);
467
468 // wouldSettle and pullRequestFor run per lookup group; the git and pull request reads they make carry the spans.
469 const wouldSettle = Effect.fnUntraced(function* (
470 group: ReadonlyArray<(typeof lookupCandidates)[number]>,
471 pullRequest: SettlementPullRequest,
472 ) {
473 const currentSettings = yield* settingsService.getSettings;
474 const decisionNow = yield* DateTime.now;
475 return group.some((thread) => {
476 const { settings } = resolveProjectSettings(currentSettings, thread.projectId);
477 return (
478 resolveAutoSettlementAt({
479 thread,
480 pullRequest,
481 nowMs: DateTime.toEpochMillis(decisionNow),
482 autoSettleAfterDays: settings.sidebarAutoSettleAfterDays,
483 autoSettleOnMerge: settings.sidebarAutoSettleOnMerge,
484 autoSettleKeepPullRequests: settings.sidebarAutoSettleKeepPullRequests,
485 }) !== null
486 );
487 });
488 });
489
490 const pullRequestFor = Effect.fnUntraced(function* (
491 group: ReadonlyArray<(typeof lookupCandidates)[number]>,
492 ) {
493 const thread = group[0]!;
494 const reference = thread.linkedPullRequest ?? thread.branchPullRequest;
495 if (reference != null) {
496 // Only matching PR links settle from this merge event; unrelated links require fresh lookup.
497 const matchesMerge =
498 mergedPullRequest !== null &&
499 reference.projectId === mergedPullRequest.projectId &&
500 reference.repository.toLowerCase() === mergedPullRequest.repository.toLowerCase() &&
501 reference.number === mergedPullRequest.number;
502 if (!matchesMerge && !projects.has(reference.projectId)) {
503 return yield* Effect.die(new Error("linked pull request project not found"));
504 }
505 const summary = matchesMerge
506 ? ({
507 state: "merged",
508 closedAt: null,
509 mergedAt: mergedPullRequest.mergedAt,
510 } satisfies SettlementPullRequest)
511 : yield* pullRequests.summary(
512 {
513 projectId: reference.projectId,
514 repository: reference.repository,
515 number: reference.number,
516 },
517 { recoverTransientFailure: false },
518 );
519 const terminal = {
520 state: summary.state,
521 closedAt: summary.closedAt ?? null,
522 mergedAt: summary.mergedAt ?? null,
523 } satisfies SettlementPullRequest;
524 const cwd = lookupCwdByThreadId.get(thread.id);
525 if (summary.state !== "open" && thread.branch !== null && cwd !== undefined) {
526 if (!(yield* wouldSettle(group, terminal))) return undefined;
527 const current = yield* git.branchPullRequest(
528 { cwd, branch: thread.branch },
529 { refresh: true },
530 );
531 const project = projects.get(thread.projectId);
532 if (
533 current?.state === "open" &&
534 project !== undefined &&
535 pullRequestMatchesProject(current, project)
536 ) {
537 return current;
538 }
539 }
540 return terminal;
541 }
542 if (thread.branch === null) return null;
543 const cwd = lookupCwdByThreadId.get(thread.id);
544 if (cwd === undefined) {
545 return yield* Effect.die(new Error("thread project not found"));
546 }
547 const refresh = group.filter((member) => shellClosedThreads.delete(member.id)).length > 0;
548 // Remembered git state is enough to rule a thread out; only an answer
549 // that would settle it is confirmed against fresh git state.
550 const remembered = yield* git.branchPullRequest(
551 { cwd, branch: thread.branch },
552 { refresh, confirmTerminal: false },
553 );
554 if (refresh || remembered === null || remembered.state === "open") return remembered;
555 if (!(yield* wouldSettle(group, remembered))) return undefined;
556 return yield* git.branchPullRequest({ cwd, branch: thread.branch }, { freshGitState: true });
557 });
558
559 yield* Effect.forEach(
560 groups.values(),
561 (group) =>
562 Effect.gen(function* () {
563 const pullRequest = yield* pullRequestFor(group);
564 if (pullRequest === undefined) return;
565 yield* Effect.forEach(group, (thread) => settleThread(thread, pullRequest), {
566 discard: true,
567 });
568 }).pipe(
569 Effect.catchCauseIf(
570 (cause) => !Cause.hasInterruptsOnly(cause),
571 (cause) =>
572 Effect.logWarning("automatic thread settlement skipped", {
573 threadIds: group.map((thread) => thread.id),
574 cause: Cause.pretty(cause),
575 }),
576 ),
577 ),
578 { concurrency: 8, discard: true },
579 );
580 });
581
582 const runSweep = (
583 mergedPullRequest: PullRequestService.PullRequestMergeEvent | null,
584 threadId?: ThreadId,
585 ) =>
586 sweep(mergedPullRequest, threadId).pipe(
587 Effect.provideService(GitVcsDriver.BackgroundGitWork, true),
588 Effect.catchCauseIf(
589 (cause) => !Cause.hasInterruptsOnly(cause),
590 (cause) =>
591 Effect.logWarning("automatic thread settlement sweep failed", {
592 cause: Cause.pretty(cause),
593 }),
594 ),
595 );
596 let fullSweepPending = false;
597 // A thread that just stopped must not wait behind a full sweep or another thread's host
598 // reads: clients hold it in Working only briefly, so a late settle makes it appear and then vanish.
599 const fullSweeps = yield* makeDrainableWorker(() => {
600 // Clear at the start so events during this sweep can request one follow-up.
601 fullSweepPending = false;
602 return runSweep(null);
603 });
604 const enqueueFullSweep = () =>
605 Effect.suspend(() => {
606 if (fullSweepPending) return Effect.void;
607 fullSweepPending = true;
608 return fullSweeps.enqueue(undefined);
609 });
610 const threadSweeps = yield* FiberSet.make<void>();
611 const threadSweepSlots = yield* Semaphore.make(8);
612 // One sweep per thread at a time; events during a running sweep coalesce into one more pass.
613 const threadSweepStates = new Map<ThreadId, "queued" | "running" | "rerun">();
614 const sweepThread = (
615 threadId: ThreadId,
616 ): Effect.Effect<void, Effect.Error<ReturnType<typeof runSweep>>> =>
617 Effect.sync(() => threadSweepStates.set(threadId, "running")).pipe(
618 Effect.andThen(runSweep(null, threadId)),
619 threadSweepSlots.withPermits(1),
620 Effect.andThen(
621 Effect.suspend(() => {
622 if (threadSweepStates.get(threadId) === "rerun") return sweepThread(threadId);
623 threadSweepStates.delete(threadId);
624 return Effect.void;
625 }),
626 ),
627 );
628 const enqueueThread = (threadId: ThreadId) =>
629 Effect.suspend(() => {
630 const state = threadSweepStates.get(threadId);
631 if (state === "running") threadSweepStates.set(threadId, "rerun");
632 if (state !== undefined) return Effect.void;
633 threadSweepStates.set(threadId, "queued");
634 return sweepThread(threadId).pipe(FiberSet.run(threadSweeps), Effect.asVoid);
635 });
636
637 // Skip settle scripts in shared checkouts where other threads may still work.
638 const cleanUpSettledThread = Effect.fn("ThreadSettlementServiceV2.cleanUpSettledThread")(
639 function* (threadId: ThreadId) {
640 const settled = yield* projections.getThread(threadId);
641 if (settled.settledOverride !== "settled") return;
642 yield* terminals.closeOnSettle({ threadId });
643 const worktreePath = settled.worktreePath;
644 if (worktreePath === null || !(yield* fileSystem.exists(worktreePath))) return;
645 // Recheck after closing and worktree I/O; re-engaged threads must not be cleaned up.
646 const thread = yield* projections.getThread(threadId);
647 if (thread.settledOverride !== "settled") return;
648 const settledAtMs = toMillis(thread.settledAt);
649 if (settledAtMs === null || settleActionRunAt.get(threadId) === settledAtMs) return;
650 // The parent's worktree belongs to the parent; shared-worktree children
651 // skip duplicate cleanup there.
652 if (
653 thread.lineage.relationshipToParent === "subagent" &&
654 thread.lineage.parentThreadId !== null
655 ) {
656 const parent = yield* projections.getThread(thread.lineage.parentThreadId).pipe(
657 Effect.catchTags({
658 ProjectionStoreThreadNotFoundError: () => Effect.succeed(null),
659 }),
660 );
661 if (parent?.worktreePath === worktreePath) return;
662 }
663 const run = yield* projectScripts.runForThread({
664 threadId,
665 projectId: thread.projectId,
666 worktreePath,
667 trigger: "settle",
668 observeCompletion: {},
669 });
670 settleActionRunAt.set(threadId, settledAtMs);
671 if (run.status === "started" && run.completion) {
672 yield* run.completion.pipe(Effect.forkDetach);
673 }
674 },
675 (effect, threadId) =>
676 effect.pipe(
677 Effect.catchCauseIf(
678 (cause) => !Cause.hasInterruptsOnly(cause),
679 (cause) =>
680 Effect.logWarning("cleaning up a settled thread failed", {
681 threadId,
682 cause: Cause.pretty(cause),
683 }),
684 ),
685 ),
686 );
687
688 const processEvent = (event: OrchestrationV2DomainEvent) => {
689 switch (event.type) {
690 case "thread.settled":
691 return cleanUpSettledThread(event.threadId).pipe(Effect.andThen(enqueueFullSweep()));
692 case "thread.archived":
693 case "thread.deleted":
694 return enqueueFullSweep();
695 case "thread.pull-request-link-updated":
696 case "thread.pull-request-synced":
697 case "provider-session.detached":
698 case "run.background-work-cancelled":
699 return enqueueThread(event.threadId);
700 // Sweep only when the work can have ended: progress updates cannot settle anything.
701 case "subagent.updated":
702 return isOrchestrationV2WorkActive(event.payload.status)
703 ? Effect.void
704 : enqueueThread(event.threadId);
705 case "provider-thread.updated":
706 return backgroundWorkHoldsCompletion(event.payload.pendingBackgroundTasks ?? [])
707 ? Effect.void
708 : enqueueThread(event.threadId);
709 case "turn-item.updated": {
710 const item = event.payload;
711 if (
712 item.type === "command_execution" &&
713 !isOrchestrationV2WorkActive(item.status) &&
714 sourceControlDefinitions.definitions.some(
715 (definition) => definition.changeRequestCloseCommand?.test(item.input) === true,
716 )
717 ) {
718 shellClosedThreads.add(event.threadId);
719 }
720 return turnItemUpdateCanEndBackgroundWork(item)
721 ? enqueueThread(event.threadId)
722 : Effect.void;
723 }
724 case "provider-session.updated":
725 return event.payload.status !== "starting" && event.payload.status !== "running"
726 ? enqueueThread(event.threadId)
727 : Effect.void;
728 case "run.updated":
729 return ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes(
730 event.payload.status,
731 )
732 ? enqueueThread(event.threadId)
733 : Effect.void;
734 default:
735 return Effect.void;
736 }
737 };
738
739 const start: ThreadSettlementServiceV2["Service"]["start"] = Effect.fn(
740 "ThreadSettlementServiceV2.start",
741 )(function* () {
742 const settingsChanges = yield* settingsService.subscribeChanges;
743 const mergedPullRequests = yield* pullRequests.subscribeMerges;
744 const events = orchestrator.streamDomainEvents;
745 const initialSettings = yield* settingsService.getSettings.pipe(Effect.orDie);
746 let lastSettlementSettings = autoSettlementSettingsKey(initialSettings);
747 yield* forkParked(
748 Effect.gen(function* () {
749 yield* enqueueFullSweep();
750 yield* fullSweeps.drain;
751 }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid),
752 );
753 yield* forkParked(
754 Stream.runForEach(settingsChanges, (settings) => {
755 const key = autoSettlementSettingsKey(settings);
756 if (key === lastSettlementSettings) {
757 return Effect.void;
758 }
759 lastSettlementSettings = key;
760 return enqueueFullSweep();
761 }),
762 );
763 yield* forkParked(Stream.runForEach(mergedPullRequests, (event) => runSweep(event)));
764 yield* forkParked(
765 Stream.runForEach(events, processEvent).pipe(
766 Effect.catchCause((cause) =>
767 Effect.logWarning("Thread settlement event stream failed", { cause }),
768 ),
769 ),
770 );
771 });
772
773 return {
774 start,
775 drain: Effect.andThen(fullSweeps.drain, FiberSet.awaitEmpty(threadSweeps)),
776 } satisfies ThreadSettlementServiceV2["Service"];
777});