Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions apps/desktop/src/settings/DesktopClientSettings.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ const clientSettings: ClientSettings = {
sidebarProjectSortOrder: "manual",
sidebarThreadSortOrder: "created_at",
sidebarThreadPreviewCount: 6,
sidebarShowSubagentThreads: false,
timestampFormat: "24-hour",
wordWrap: true,
};
Expand Down
82 changes: 82 additions & 0 deletions apps/server/src/orchestration-v2/CheckpointCleanupService.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
import { assert, it } from "@effect/vitest";
import {
CheckpointId,
CheckpointRef,
CheckpointScopeId,
ThreadId,
type OrchestrationV2Checkpoint,
type OrchestrationV2CheckpointScope,
type OrchestrationV2ThreadProjection,
} from "@t3tools/contracts";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";

import { CheckpointCleanupServiceV2, layer } from "./CheckpointCleanupService.ts";
import { CheckpointServiceV2 } from "./CheckpointService.ts";
import { ProjectionStoreV2 } from "./ProjectionStore.ts";

it.effect("cleans every checkpoint scope through the latest known ordinal", () => {
const threadId = ThreadId.make("thread_checkpoint_cleanup");
const firstScope = {
id: CheckpointScopeId.make("scope_first"),
threadId,
} as OrchestrationV2CheckpointScope;
const secondScope = {
id: CheckpointScopeId.make("scope_second"),
threadId,
} as OrchestrationV2CheckpointScope;
const checkpoint = {
id: CheckpointId.make("checkpoint_first"),
scopeId: firstScope.id,
ordinalWithinScope: 2,
ref: CheckpointRef.make("refs/t3/test/checkpoint-first"),
} as OrchestrationV2Checkpoint;
const projection = {
thread: { id: threadId },
runs: [{ ordinal: 4 }],
checkpointScopes: [firstScope, secondScope],
checkpoints: [checkpoint],
} as unknown as OrchestrationV2ThreadProjection;
const calls: Array<{
readonly scopeId: CheckpointScopeId;
readonly checkpointIds: ReadonlyArray<CheckpointId>;
readonly throughOrdinal: number;
}> = [];
const testLayer = layer.pipe(
Layer.provide(
Layer.merge(
Layer.mock(ProjectionStoreV2)({
getThreadProjection: () => Effect.succeed(projection),
}),
Layer.mock(CheckpointServiceV2)({
deleteScopeRefs: (input) => {
calls.push({
scopeId: input.scope.id,
checkpointIds: input.checkpoints.map((item) => item.id),
throughOrdinal: input.throughOrdinal,
});
return Effect.void;
},
}),
),
),
);

return Effect.gen(function* () {
const cleanup = yield* CheckpointCleanupServiceV2;
yield* cleanup.cleanup(threadId);

assert.deepEqual(calls, [
{
scopeId: firstScope.id,
checkpointIds: [checkpoint.id],
throughOrdinal: 4,
},
{
scopeId: secondScope.id,
checkpointIds: [],
throughOrdinal: 4,
},
]);
}).pipe(Effect.provide(testLayer));
});
70 changes: 70 additions & 0 deletions apps/server/src/orchestration-v2/CheckpointCleanupService.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
import { 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 { CheckpointServiceV2 } from "./CheckpointService.ts";
import { ProjectionStoreV2 } from "./ProjectionStore.ts";

export class CheckpointCleanupError extends Schema.TaggedErrorClass<CheckpointCleanupError>()(
"CheckpointCleanupError",
{
threadId: ThreadId,
cause: Schema.Defect(),
},
) {
override get message(): string {
return `Failed to clean checkpoint refs for thread ${this.threadId}.`;
}
}

export class CheckpointCleanupServiceV2 extends Context.Service<
CheckpointCleanupServiceV2,
{
readonly cleanup: (threadId: ThreadId) => Effect.Effect<void, CheckpointCleanupError>;
}
>()("t3/orchestration-v2/CheckpointCleanupService/CheckpointCleanupServiceV2") {}

export const layer: Layer.Layer<
CheckpointCleanupServiceV2,
never,
CheckpointServiceV2 | ProjectionStoreV2
> = Layer.effect(
CheckpointCleanupServiceV2,
Effect.gen(function* () {
const checkpoints = yield* CheckpointServiceV2;
const projections = yield* ProjectionStoreV2;

const cleanup = Effect.fn("orchestrationV2.checkpointCleanup.cleanup")(function* (
threadId: ThreadId,
) {
const projection = yield* projections.getThreadProjection(threadId);
const throughOrdinal = Math.max(
0,
...projection.runs.map((run) => run.ordinal),
...projection.checkpoints.map((checkpoint) => checkpoint.ordinalWithinScope),
);

yield* Effect.forEach(
projection.checkpointScopes,
(scope) =>
checkpoints.deleteScopeRefs({
scope,
checkpoints: projection.checkpoints.filter(
(checkpoint) => checkpoint.scopeId === scope.id,
),
throughOrdinal,
}),
{ concurrency: 1, discard: true },
);
});

return CheckpointCleanupServiceV2.of({
cleanup: (threadId) =>
cleanup(threadId).pipe(
Effect.mapError((cause) => new CheckpointCleanupError({ threadId, cause })),
),
});
}),
);
58 changes: 58 additions & 0 deletions apps/server/src/orchestration-v2/CheckpointService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -103,13 +103,26 @@ export class CheckpointDeleteStaleRefsError extends Schema.TaggedErrorClass<Chec
}
}

export class CheckpointDeleteScopeRefsError extends Schema.TaggedErrorClass<CheckpointDeleteScopeRefsError>()(
"CheckpointDeleteScopeRefsError",
{
scopeId: CheckpointScopeId,
cause: Schema.Defect(),
},
) {
override get message(): string {
return `Failed to delete checkpoint refs for scope ${this.scopeId}.`;
}
}

export const CheckpointServiceV2Error = Schema.Union([
CheckpointRootScopePrepareError,
CheckpointScopeEnsureError,
CheckpointBaselineCaptureError,
CheckpointCaptureError,
CheckpointRestoreError,
CheckpointDeleteStaleRefsError,
CheckpointDeleteScopeRefsError,
]);
export type CheckpointServiceV2Error = typeof CheckpointServiceV2Error.Type;

Expand Down Expand Up @@ -150,6 +163,16 @@ export interface CheckpointServiceV2Shape {
readonly scope: OrchestrationV2CheckpointScope;
readonly checkpoints: ReadonlyArray<OrchestrationV2Checkpoint>;
}) => Effect.Effect<void, CheckpointServiceV2Error>;
readonly deleteScopeRefs: (input: {
readonly scope: OrchestrationV2CheckpointScope;
readonly checkpoints: ReadonlyArray<OrchestrationV2Checkpoint>;
/**
* Baselines can be captured before their projection event is committed.
* Include every deterministic ordinal through this value so deletion also
* cleans refs left by an interrupted capture.
*/
readonly throughOrdinal: number;
}) => Effect.Effect<void, CheckpointServiceV2Error>;
}

export class CheckpointServiceV2 extends Context.Service<
Expand Down Expand Up @@ -541,6 +564,40 @@ export const layer: Layer.Layer<
),
);

const deleteScopeRefs: CheckpointServiceV2Shape["deleteScopeRefs"] = (input) => {
const checkpointRefs = new Set<CheckpointRef>(
input.checkpoints.map((checkpoint) => checkpoint.ref),
);
for (
let ordinalWithinScope = 0;
ordinalWithinScope <= input.throughOrdinal;
ordinalWithinScope += 1
) {
checkpointRefs.add(
checkpointRefForScopeOrdinal({
scopeId: input.scope.id,
ordinalWithinScope,
}),
);
}

return withWorkspaceLock(
input.scope.cwd,
checkpointStore.deleteCheckpointRefs({
cwd: input.scope.cwd,
checkpointRefs: Array.from(checkpointRefs),
}),
).pipe(
Effect.mapError(
(cause) =>
new CheckpointDeleteScopeRefsError({
scopeId: input.scope.id,
cause,
}),
),
);
};

return CheckpointServiceV2.of({
prepareRootRunScope: (input) =>
makeRootRunScope({ ...input, idAllocator }).pipe(
Expand All @@ -559,6 +616,7 @@ export const layer: Layer.Layer<
capture,
restore,
deleteStaleRefs,
deleteScopeRefs,
} satisfies CheckpointServiceV2Shape);
}),
);
4 changes: 4 additions & 0 deletions apps/server/src/orchestration-v2/EffectOutbox.ts
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,9 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
runId: RunId,
scopeId: CheckpointScopeId,
}),
Schema.Struct({
type: Schema.Literal("checkpoint.cleanup"),
}),
Schema.Struct({
type: Schema.Literal("terminal.cleanup"),
}),
Expand All @@ -95,6 +98,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [
"provider-session.detach",
"provider-thread.rollback",
"checkpoint.capture",
"checkpoint.cleanup",
"terminal.cleanup",
"attachment.cleanup",
] as const satisfies ReadonlyArray<OrchestrationEffectRequestV2["type"]>;
Expand Down
5 changes: 5 additions & 0 deletions apps/server/src/orchestration-v2/EffectWorker.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Ref from "effect/Ref";

import { CheckpointCleanupServiceV2 } from "./CheckpointCleanupService.ts";
import { CheckpointRollbackServiceV2 } from "./CheckpointRollbackService.ts";
import type { OrchestrationEffectV2 } from "./EffectOutbox.ts";
import { executorLayer, OrchestrationEffectExecutorV2 } from "./EffectWorker.ts";
Expand Down Expand Up @@ -125,6 +126,10 @@ function makeExecutorLayer(input: {
CheckpointRollbackServiceV2,
CheckpointRollbackServiceV2.of({ execute: () => Effect.void }),
),
Layer.succeed(
CheckpointCleanupServiceV2,
CheckpointCleanupServiceV2.of({ cleanup: () => Effect.void }),
),
Layer.succeed(
RuntimeRequestServiceV2,
RuntimeRequestServiceV2.of({ respond: () => Effect.void }),
Expand Down
18 changes: 16 additions & 2 deletions apps/server/src/orchestration-v2/EffectWorker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,11 @@ import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Schema from "effect/Schema";

import { CheckpointCleanupServiceV2 } from "./CheckpointCleanupService.ts";
import { CheckpointRollbackServiceV2 } from "./CheckpointRollbackService.ts";
import { EffectOutboxV2, type OrchestrationEffectV2 } from "./EffectOutbox.ts";
import { RunFinalizationService } from "./RunFinalizationService.ts";
import { ResourceCleanupService } from "./ResourceCleanupService.ts";
import { EffectOutboxV2, type OrchestrationEffectV2 } from "./EffectOutbox.ts";
import { CheckpointRollbackServiceV2 } from "./CheckpointRollbackService.ts";
import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts";
import { ProviderTurnControlServiceV2 } from "./ProviderTurnControlService.ts";
import { ProviderTurnStartServiceV2 } from "./ProviderTurnStartService.ts";
Expand Down Expand Up @@ -41,6 +42,7 @@ export const executorLayer: Layer.Layer<
never,
| ProviderSessionManagerV2
| RunFinalizationService
| CheckpointCleanupServiceV2
| CheckpointRollbackServiceV2
| ProviderTurnControlServiceV2
| ProviderTurnStartServiceV2
Expand All @@ -50,6 +52,7 @@ export const executorLayer: Layer.Layer<
Effect.gen(function* () {
const runFinalization = yield* RunFinalizationService;
const resourceCleanup = yield* ResourceCleanupService;
const checkpointCleanup = yield* CheckpointCleanupServiceV2;
const checkpointRollback = yield* CheckpointRollbackServiceV2;
const providerSessions = yield* ProviderSessionManagerV2;
const providerTurnControl = yield* ProviderTurnControlServiceV2;
Expand Down Expand Up @@ -229,6 +232,17 @@ export const executorLayer: Layer.Layer<
}),
),
);
case "checkpoint.cleanup":
return checkpointCleanup.cleanup(effect.threadId).pipe(
Effect.mapError(
(cause) =>
new OrchestrationEffectExecutionError({
effectId: effect.id,
effectType: effect.request.type,
cause,
}),
),
);
case "terminal.cleanup":
return resourceCleanup.cleanupTerminals(effect.threadId).pipe(
Effect.mapError(
Expand Down
Loading
Loading