import { CheckpointScopeId, RunId, ThreadId } from "@t3tools/contracts";import * as Context from "effect/Context";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Schema from "effect/Schema"; import * as PullRequestService from "../pullRequest/PullRequestService.ts";import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts";import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts";import * as CheckpointCapture from "./CheckpointCaptureService.ts";import * as ProjectionStore from "./ProjectionStore.ts"; export class RunFinalizationError extends Schema.TaggedErrorClass<RunFinalizationError>()( "RunFinalizationError", { threadId: ThreadId, runId: RunId, scopeId: CheckpointScopeId, operation: Schema.Literals(["capture-checkpoint", "refresh-workspace"]), cause: Schema.Defect(), },) {} export class RunFinalizationRefreshError extends Schema.TaggedErrorClass<RunFinalizationRefreshError>()( "RunFinalizationRefreshError", { cwd: Schema.String, cause: Schema.Defect() },) {} export class RunFinalizationObserver extends Context.Reference<{ readonly refreshAfterTurn: Effect.Effect<void>; readonly refresh: (input: { readonly cwd: string; readonly threadId: ThreadId; readonly runId: RunId; }) => Effect.Effect<void, RunFinalizationRefreshError>;}>("t3/orchestration-v2/RunFinalizationObserver", { defaultValue: () => ({ refresh: () => Effect.void, refreshAfterTurn: Effect.void }),}) {} export class RunFinalizationService extends Context.Service< RunFinalizationService, { readonly finalize: (input: { readonly threadId: ThreadId; readonly runId: RunId; readonly scopeId: CheckpointScopeId; }) => Effect.Effect<void, RunFinalizationError>; }>()("t3/orchestration-v2/RunFinalizationService") {} export const make = Effect.gen(function* () { const checkpointCapture = yield* CheckpointCapture.CheckpointCaptureServiceV2; const projections = yield* ProjectionStore.ProjectionStoreV2; const observer = yield* RunFinalizationObserver; const finalize: RunFinalizationService["Service"]["finalize"] = Effect.fn( "RunFinalizationService.finalize", )(function* (input) { yield* checkpointCapture .execute(input) .pipe( Effect.mapError( (cause) => new RunFinalizationError({ ...input, operation: "capture-checkpoint", cause }), ), ); const projection = yield* projections .getThreadProjection(input.threadId) .pipe( Effect.mapError( (cause) => new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }), ), ); const cwd = projection.checkpointScopes.find((scope) => scope.id === input.scopeId)?.cwd; if (cwd !== undefined) { yield* observer .refresh({ cwd, threadId: input.threadId, runId: input.runId }) .pipe( Effect.mapError( (cause) => new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }), ), ); } }); return RunFinalizationService.of({ finalize });}); export const layer = Layer.effect(RunFinalizationService, make); export const observerLive = Layer.effect( RunFinalizationObserver, Effect.gen(function* () { const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; const vcsStatus = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; const projections = yield* ProjectionStore.ProjectionStoreV2; const pullRequests = yield* PullRequestService.PullRequestService; return { refreshAfterTurn: pullRequests.refreshAfterTurn, refresh: ({ cwd, threadId, runId }) => Effect.gen(function* () { const [, local] = yield* Effect.all( [workspaceEntries.refresh(cwd), vcsStatus.refreshLocalStatus(cwd)], { concurrency: "unbounded" }, ); if (local.refName === null || local.isDefaultRef) return; const thread = yield* projections.getThreadShell(threadId); if (!thread || thread.branch !== local.refName) return; if (thread.activeRunId !== null && thread.activeRunId !== runId) return; yield* vcsStatus.refreshPullRequestStatus(cwd).pipe( Effect.catch((error) => Effect.logWarning("failed to refresh pull request status after run completion", { threadId, cwd, detail: error.message, }), ), ); }).pipe(Effect.mapError((cause) => new RunFinalizationRefreshError({ cwd, cause }))), }; }),);
import { CheckpointScopeId, RunId, ThreadId } from "@t3tools/contracts";
import * as Context from "effect/Context";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Schema from "effect/Schema";
import * as PullRequestService from "../pullRequest/PullRequestService.ts";
import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts";
import * as WorkspaceEntries from "../workspace/WorkspaceEntries.ts";
import * as CheckpointCapture from "./CheckpointCaptureService.ts";
import * as ProjectionStore from "./ProjectionStore.ts";
export class RunFinalizationError extends Schema.TaggedErrorClass<RunFinalizationError>()(
"RunFinalizationError",
{
threadId: ThreadId,
runId: RunId,
scopeId: CheckpointScopeId,
operation: Schema.Literals(["capture-checkpoint", "refresh-workspace"]),
cause: Schema.Defect(),
},
) {}
export class RunFinalizationRefreshError extends Schema.TaggedErrorClass<RunFinalizationRefreshError>()(
"RunFinalizationRefreshError",
{ cwd: Schema.String, cause: Schema.Defect() },
) {}
export class RunFinalizationObserver extends Context.Reference<{
readonly refreshAfterTurn: Effect.Effect<void>;
readonly refresh: (input: {
readonly cwd: string;
readonly threadId: ThreadId;
readonly runId: RunId;
}) => Effect.Effect<void, RunFinalizationRefreshError>;
}>("t3/orchestration-v2/RunFinalizationObserver", {
defaultValue: () => ({ refresh: () => Effect.void, refreshAfterTurn: Effect.void }),
}) {}
export class RunFinalizationService extends Context.Service<
RunFinalizationService,
{
readonly finalize: (input: {
readonly threadId: ThreadId;
readonly runId: RunId;
readonly scopeId: CheckpointScopeId;
}) => Effect.Effect<void, RunFinalizationError>;
}
>()("t3/orchestration-v2/RunFinalizationService") {}
export const make = Effect.gen(function* () {
const checkpointCapture = yield* CheckpointCapture.CheckpointCaptureServiceV2;
const projections = yield* ProjectionStore.ProjectionStoreV2;
const observer = yield* RunFinalizationObserver;
const finalize: RunFinalizationService["Service"]["finalize"] = Effect.fn(
"RunFinalizationService.finalize",
)(function* (input) {
yield* checkpointCapture
.execute(input)
.pipe(
Effect.mapError(
(cause) => new RunFinalizationError({ ...input, operation: "capture-checkpoint", cause }),
),
);
const projection = yield* projections
.getThreadProjection(input.threadId)
.pipe(
Effect.mapError(
(cause) => new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }),
),
);
const cwd = projection.checkpointScopes.find((scope) => scope.id === input.scopeId)?.cwd;
if (cwd !== undefined) {
yield* observer
.refresh({ cwd, threadId: input.threadId, runId: input.runId })
.pipe(
Effect.mapError(
(cause) =>
new RunFinalizationError({ ...input, operation: "refresh-workspace", cause }),
),
);
}
});
return RunFinalizationService.of({ finalize });
});
export const layer = Layer.effect(RunFinalizationService, make);
export const observerLive = Layer.effect(
RunFinalizationObserver,
Effect.gen(function* () {
const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
const vcsStatus = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
const projections = yield* ProjectionStore.ProjectionStoreV2;
const pullRequests = yield* PullRequestService.PullRequestService;
return {
refreshAfterTurn: pullRequests.refreshAfterTurn,
refresh: ({ cwd, threadId, runId }) =>
Effect.gen(function* () {
const [, local] = yield* Effect.all(
[workspaceEntries.refresh(cwd), vcsStatus.refreshLocalStatus(cwd)],
{ concurrency: "unbounded" },
);
if (local.refName === null || local.isDefaultRef) return;
const thread = yield* projections.getThreadShell(threadId);
if (!thread || thread.branch !== local.refName) return;
if (thread.activeRunId !== null && thread.activeRunId !== runId) return;
yield* vcsStatus.refreshPullRequestStatus(cwd).pipe(
Effect.catch((error) =>
Effect.logWarning("failed to refresh pull request status after run completion", {
threadId,
cwd,
detail: error.message,
}),
),
);
}).pipe(Effect.mapError((cause) => new RunFinalizationRefreshError({ cwd, cause }))),
};
}),
);