1import { CheckpointScopeId, ProjectId, RunId, ThreadId } from "@t3tools/contracts";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Schema from "effect/Schema";
6
7import * as PullRequestService from "../pullRequest/PullRequestService.ts";
8import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts";
9import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts";
10import * as CheckpointCapture from "./CheckpointCaptureService.ts";
11import * as ProjectionStore from "./ProjectionStore.ts";
12
13export class RunFinalizationError extends Schema.TaggedError<RunFinalizationError>()(
14 "RunFinalizationError",
15 {
16 threadId: ThreadId,
17 runId: RunId,
18 scopeId: CheckpointScopeId,
19 operation: Schema.Literals(["capture-checkpoint", "refresh-workspace"]),
20 cause: Schema.Defect(),
21 },
22) {}
23
24export class RunFinalizationRefreshError extends Schema.TaggedError<RunFinalizationRefreshError>()(
25 "RunFinalizationRefreshError",
26 { cwd: Schema.String, cause: Schema.Defect() },
27) {}
28
29export class RunFinalizationObserver extends Context.Reference<{
30 readonly refreshAfterTurn: (projectId: ProjectId) => Effect.Effect<void>;
31 readonly refresh: (input: {
32 readonly cwd: string;
33 readonly threadId: ThreadId;
34 readonly runId: RunId;
35 }) => Effect.Effect<void, RunFinalizationRefreshError>;
36}>("t3/orchestration-v2/RunFinalizationObserver", {
37 defaultValue: () => ({ refresh: () => Effect.void, refreshAfterTurn: () => Effect.void }),
38}) {}
39
40export class RunFinalizationService extends Context.Service<
41 RunFinalizationService,
42 {
43 readonly finalize: (input: {
44 readonly threadId: ThreadId;
45 readonly runId: RunId;
46 readonly scopeId: CheckpointScopeId;
47 }) => Effect.Effect<void, RunFinalizationError>;
48 }
49>()("t3/orchestration-v2/RunFinalizationService") {}
50
51const make = Effect.gen(function* () {
52 const checkpointCapture = yield* CheckpointCapture.CheckpointCaptureServiceV2;
53 const projections = yield* ProjectionStore.ProjectionStoreV2;
54 const observer = yield* RunFinalizationObserver;
55
56 const finalize: RunFinalizationService["Service"]["finalize"] = Effect.fn(
57 "RunFinalizationService.finalize",
58 )(function* (input) {
59 yield* checkpointCapture
60 .execute(input)
61 .pipe(
62 Effect.mapError(
63 (cause) => new RunFinalizationError({ ...input, operation: "capture-checkpoint", cause }),
64 ),
65 );
66 const projection = yield* projections
67 .getCheckpointContext(input.threadId)
68 .pipe(
69 Effect.mapError(
70 (cause) => new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }),
71 ),
72 );
73 const cwd = projection.checkpointScopes.find((scope) => scope.id === input.scopeId)?.cwd;
74 if (cwd !== undefined) {
75 yield* observer
76 .refresh({ cwd, threadId: input.threadId, runId: input.runId })
77 .pipe(
78 Effect.mapError(
79 (cause) =>
80 new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }),
81 ),
82 );
83 }
84 });
85 return RunFinalizationService.of({ finalize });
86});
87
88export const layer = Layer.effect(RunFinalizationService, make);
89
90export const layerObserver = Layer.effect(
91 RunFinalizationObserver,
92 Effect.gen(function* () {
93 const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
94 const vcsStatus = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
95 const projections = yield* ProjectionStore.ProjectionStoreV2;
96 const pullRequests = yield* PullRequestService.PullRequestService;
97 return {
98 refreshAfterTurn: pullRequests.refreshAfterTurn,
99 refresh: ({ cwd, threadId, runId }) =>
100 Effect.gen(function* () {
101 const [, local] = yield* Effect.all(
102 [workspaceEntries.refresh(cwd), vcsStatus.refreshLocalStatus(cwd)],
103 { concurrency: "unbounded" },
104 );
105 if (local.refName === null || local.isDefaultRef) return;
106 const thread = yield* projections.getThreadShell(threadId);
107 if (!thread || thread.branch !== local.refName) return;
108 if (thread.activeRunId !== null && thread.activeRunId !== runId) return;
109 yield* vcsStatus.refreshPullRequestStatus(cwd).pipe(
110 Effect.catch((error) =>
111 Effect.logWarning("failed to refresh pull request status after run completion", {
112 threadId,
113 cwd,
114 detail: error.message,
115 }),
116 ),
117 );
118 }).pipe(Effect.mapError((cause) => new RunFinalizationRefreshError({ cwd, cause }))),
119 };
120 }),
121);