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