1import { CheckpointScopeId, ProjectId, RunId, ThreadId } from "@t3tools/contracts";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as FiberMap from "effect/FiberMap";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7import * as Schema from "effect/Schema";
8
9import * as ProjectEnrichment from "../project/ProjectEnrichmentService.ts";
10import * as ProjectService from "../project/ProjectService.ts";
11import * as PullRequestService from "../pullRequest/PullRequestService.ts";
12import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts";
13import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts";
14import * as CheckpointCapture from "./CheckpointCaptureService.ts";
15import * as ProjectionStore from "./ProjectionStore.ts";
16
17export class RunFinalizationError extends Schema.TaggedError<RunFinalizationError>()(
18 "RunFinalizationError",
19 {
20 threadId: ThreadId,
21 runId: RunId,
22 scopeId: CheckpointScopeId,
23 operation: Schema.Literals(["capture-checkpoint", "refresh-workspace"]),
24 cause: Schema.Defect(),
25 },
26) {}
27
28export class RunFinalizationRefreshError extends Schema.TaggedError<RunFinalizationRefreshError>()(
29 "RunFinalizationRefreshError",
30 { cwd: Schema.String, cause: Schema.Defect() },
31) {}
32
33export class RunFinalizationObserver extends Context.Reference<{
34 readonly refreshAfterTurn: (projectId: ProjectId) => Effect.Effect<void>;
35 readonly refresh: (input: {
36 readonly cwd: string;
37 readonly threadId: ThreadId;
38 readonly runId: RunId;
39 }) => Effect.Effect<void, RunFinalizationRefreshError>;
40}>("t3/orchestration-v2/RunFinalizationObserver", {
41 defaultValue: () => ({ refresh: () => Effect.void, refreshAfterTurn: () => Effect.void }),
42}) {}
43
44export class RunFinalizationService extends Context.Service<
45 RunFinalizationService,
46 {
47 readonly finalize: (input: {
48 readonly threadId: ThreadId;
49 readonly runId: RunId;
50 readonly scopeId: CheckpointScopeId;
51 }) => Effect.Effect<void, RunFinalizationError>;
52 }
53>()("t3/orchestration-v2/RunFinalizationService") {}
54
55const make = Effect.gen(function* () {
56 const checkpointCapture = yield* CheckpointCapture.CheckpointCaptureServiceV2;
57 const projections = yield* ProjectionStore.ProjectionStoreV2;
58 const observer = yield* RunFinalizationObserver;
59
60 const finalize: RunFinalizationService["Service"]["finalize"] = Effect.fn(
61 "RunFinalizationService.finalize",
62 )(function* (input) {
63 yield* checkpointCapture
64 .execute(input)
65 .pipe(
66 Effect.mapError(
67 (cause) => new RunFinalizationError({ ...input, operation: "capture-checkpoint", cause }),
68 ),
69 );
70 const projection = yield* projections
71 .getCheckpointContext(input.threadId)
72 .pipe(
73 Effect.mapError(
74 (cause) => new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }),
75 ),
76 );
77 const cwd = projection.checkpointScopes.find((scope) => scope.id === input.scopeId)?.cwd;
78 if (cwd !== undefined) {
79 // Retrying a display refresh after the checkpoint is durable blocks the next turn.
80 yield* observer.refresh({ cwd, threadId: input.threadId, runId: input.runId }).pipe(
81 Effect.catchTags({
82 RunFinalizationRefreshError: (cause) =>
83 Effect.logWarning("Failed to refresh workspace after checkpoint capture", {
84 ...input,
85 cause,
86 }),
87 }),
88 );
89 }
90 });
91 return RunFinalizationService.of({ finalize });
92});
93
94export const layer = Layer.effect(RunFinalizationService, make);
95
96export const layerObserver = Layer.effect(
97 RunFinalizationObserver,
98 Effect.gen(function* () {
99 const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
100 const vcsStatus = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
101 const projections = yield* ProjectionStore.ProjectionStoreV2;
102 const pullRequests = yield* PullRequestService.PullRequestService;
103 const projects = yield* ProjectService.ProjectService;
104 const enrichment = yield* ProjectEnrichment.ProjectEnrichmentService;
105 const pullRequestLookups = yield* FiberMap.make<string>();
106 const identityRefreshes = yield* FiberMap.make<ProjectId>();
107 // A turn can add the first remote (`git remote add`, `gh repo create --push`), which the
108 // enrichment and resolver caches would otherwise keep reporting as missing for a minute.
109 const refreshMissingRepositoryIdentity = Effect.fn(
110 "RunFinalizationService.refreshMissingRepositoryIdentity",
111 )(
112 function* (projectId: ProjectId) {
113 const project = yield* projects.getById(projectId);
114 if (Option.isNone(project)) return;
115 const { workspaceRoot } = project.value;
116 if ((yield* enrichment.readRepositoryIdentity(workspaceRoot)) !== null) return;
117 yield* enrichment.refreshRepositoryIdentity(workspaceRoot);
118 },
119 (effect, projectId) =>
120 effect.pipe(
121 Effect.catch((error) =>
122 Effect.logWarning("failed to refresh repository identity after run completion", {
123 projectId,
124 detail: error.message,
125 }),
126 ),
127 ),
128 );
129 return {
130 refreshAfterTurn: pullRequests.refreshAfterTurn,
131 refresh: Effect.fn("RunFinalizationService.refresh")(
132 function* ({ cwd, threadId, runId }) {
133 const [, local] = yield* Effect.all(
134 [workspaceEntries.refresh(cwd), vcsStatus.refreshLocalStatus(cwd)],
135 { concurrency: "unbounded" },
136 );
137 const thread = yield* projections.getThreadShell(threadId);
138 if (!thread) return;
139 // Forge discovery probes can take seconds; like the PR lookup, the next turn must not wait.
140 yield* refreshMissingRepositoryIdentity(thread.projectId).pipe(
141 FiberMap.run(identityRefreshes, thread.projectId),
142 );
143 if (local.refName === null || local.isDefaultRef) return;
144 if (thread.branch !== local.refName) return;
145 if (thread.activeRunId !== null && thread.activeRunId !== runId) return;
146 // The forge CLI can take minutes, so the next turn must not wait for it. A newer
147 // turn's lookup replaces a running one instead of queueing behind the cwd's lock.
148 yield* vcsStatus.refreshPullRequestStatus(cwd).pipe(
149 Effect.catch((error) =>
150 Effect.logWarning("failed to refresh pull request status after run completion", {
151 threadId,
152 cwd,
153 detail: error.message,
154 }),
155 ),
156 FiberMap.run(pullRequestLookups, cwd),
157 );
158 },
159 (effect, { cwd }) =>
160 effect.pipe(Effect.mapError((cause) => new RunFinalizationRefreshError({ cwd, cause }))),
161 ),
162 };
163 }),
164);