diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index ebc4f984b86..a17727f6723 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -78,6 +78,7 @@ import { VcsStatusBroadcaster } from "../src/vcs/VcsStatusBroadcaster.ts"; import { GitWorkflowService } from "../src/git/GitWorkflowService.ts"; import * as VcsProcess from "../src/vcs/VcsProcess.ts"; import * as AgentAwarenessRelay from "../src/relay/AgentAwarenessRelay.ts"; +import * as VcsMaintenanceReactor from "../src/vcs/VcsMaintenanceReactor.ts"; const decodeCodexSettings = Schema.decodeEffect(CodexSettings); @@ -372,6 +373,12 @@ export const makeOrchestrationIntegrationHarness = ( start: () => Effect.void, }), ), + Layer.provideMerge( + Layer.succeed(VcsMaintenanceReactor.VcsMaintenanceReactor, { + start: () => Effect.void, + sweep: () => Effect.void, + }), + ), ); const layer = Layer.empty.pipe( Layer.provideMerge(runtimeServicesLayer), diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts index c1dbc833718..bed8434c10f 100644 --- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts @@ -69,6 +69,8 @@ describe("CheckpointDiffQuery.layer", () => { return "full thread diff patch"; }), deleteCheckpointRefs: () => Effect.void, + pruneCheckpointRefs: () => + Effect.succeed({ scannedCount: 0, keptCount: 0, deletedCount: 0, threadCount: 0 }), }; const layer = CheckpointDiffQuery.layer.pipe( @@ -176,6 +178,8 @@ describe("CheckpointDiffQuery.layer", () => { return "diff patch"; }), deleteCheckpointRefs: () => Effect.void, + pruneCheckpointRefs: () => + Effect.succeed({ scannedCount: 0, keptCount: 0, deletedCount: 0, threadCount: 0 }), }; const layer = CheckpointDiffQuery.layer.pipe( @@ -258,6 +262,8 @@ describe("CheckpointDiffQuery.layer", () => { return "diff patch"; }), deleteCheckpointRefs: () => Effect.void, + pruneCheckpointRefs: () => + Effect.succeed({ scannedCount: 0, keptCount: 0, deletedCount: 0, threadCount: 0 }), }; const layer = CheckpointDiffQuery.layer.pipe( @@ -325,6 +331,8 @@ describe("CheckpointDiffQuery.layer", () => { restoreCheckpoint: () => Effect.succeed(true), diffCheckpoints: () => Effect.succeed("diff patch"), deleteCheckpointRefs: () => Effect.void, + pruneCheckpointRefs: () => + Effect.succeed({ scannedCount: 0, keptCount: 0, deletedCount: 0, threadCount: 0 }), }; const layer = CheckpointDiffQuery.layer.pipe( @@ -377,6 +385,8 @@ describe("CheckpointDiffQuery.layer", () => { restoreCheckpoint: () => Effect.succeed(true), diffCheckpoints: () => Effect.succeed(""), deleteCheckpointRefs: () => Effect.void, + pruneCheckpointRefs: () => + Effect.succeed({ scannedCount: 0, keptCount: 0, deletedCount: 0, threadCount: 0 }), }; const layer = CheckpointDiffQuery.layer.pipe( diff --git a/apps/server/src/checkpointing/CheckpointStore.test.ts b/apps/server/src/checkpointing/CheckpointStore.test.ts index bf332d20d0d..12b56b85ede 100644 --- a/apps/server/src/checkpointing/CheckpointStore.test.ts +++ b/apps/server/src/checkpointing/CheckpointStore.test.ts @@ -114,6 +114,98 @@ it.layer(TestLayer)("CheckpointStore.layer", (it) => { ); }); + describe("pruneCheckpointRefs", () => { + it.effect("keeps turn zero and the newest tail per thread", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir(); + yield* initRepoWithCommit(tmp); + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const threadId = ThreadId.make("thread-checkpoint-prune"); + const refs = [0, 1, 2, 3, 4].map((turn) => checkpointRefForThreadTurn(threadId, turn)); + + for (const [index, checkpointRef] of refs.entries()) { + yield* writeTextFile(NodePath.join(tmp, "README.md"), `# turn ${index}\n`); + yield* checkpointStore.captureCheckpoint({ + cwd: tmp, + checkpointRef, + }); + } + + const result = yield* checkpointStore.pruneCheckpointRefs({ + cwd: tmp, + keepPerThread: 3, + }); + + expect(result).toMatchObject({ + scannedCount: 5, + keptCount: 3, + deletedCount: 2, + threadCount: 1, + }); + expect(yield* checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef: refs[0]! })).toBe( + true, + ); + expect(yield* checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef: refs[1]! })).toBe( + false, + ); + expect(yield* checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef: refs[2]! })).toBe( + false, + ); + expect(yield* checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef: refs[3]! })).toBe( + true, + ); + expect(yield* checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef: refs[4]! })).toBe( + true, + ); + }), + ); + + it.effect("uses the full keep budget when a thread has no turn zero", () => + Effect.gen(function* () { + const tmp = yield* makeTmpDir(); + yield* initRepoWithCommit(tmp); + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const threadId = ThreadId.make("thread-checkpoint-prune-no-baseline"); + const refs = [1, 2, 3, 4, 5].map((turn) => checkpointRefForThreadTurn(threadId, turn)); + + for (const [index, checkpointRef] of refs.entries()) { + yield* writeTextFile(NodePath.join(tmp, "README.md"), `# turn ${index + 1}\n`); + yield* checkpointStore.captureCheckpoint({ + cwd: tmp, + checkpointRef, + }); + } + + const result = yield* checkpointStore.pruneCheckpointRefs({ + cwd: tmp, + keepPerThread: 3, + }); + + expect(result).toMatchObject({ + scannedCount: 5, + keptCount: 3, + deletedCount: 2, + threadCount: 1, + }); + expect(yield* checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef: refs[0]! })).toBe( + false, + ); + expect(yield* checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef: refs[1]! })).toBe( + false, + ); + expect(yield* checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef: refs[2]! })).toBe( + true, + ); + expect(yield* checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef: refs[3]! })).toBe( + true, + ); + expect(yield* checkpointStore.hasCheckpointRef({ cwd: tmp, checkpointRef: refs[4]! })).toBe( + true, + ); + }), + ); + }); + describe("diffCheckpoints", () => { it.effect("returns full oversized checkpoint diffs without truncation", () => Effect.gen(function* () { diff --git a/apps/server/src/checkpointing/CheckpointStore.ts b/apps/server/src/checkpointing/CheckpointStore.ts index f13aa4572c1..6429d20f370 100644 --- a/apps/server/src/checkpointing/CheckpointStore.ts +++ b/apps/server/src/checkpointing/CheckpointStore.ts @@ -46,6 +46,18 @@ export interface DeleteCheckpointRefsInput { readonly checkpointRefs: ReadonlyArray; } +export interface PruneCheckpointRefsInput { + readonly cwd: string; + readonly keepPerThread: number; +} + +export interface PruneCheckpointRefsResult { + readonly scannedCount: number; + readonly keptCount: number; + readonly deletedCount: number; + readonly threadCount: number; +} + /** Service tag for checkpoint persistence and restore operations. */ export class CheckpointStore extends Context.Service< CheckpointStore, @@ -93,6 +105,16 @@ export class CheckpointStore extends Context.Service< readonly deleteCheckpointRefs: ( input: DeleteCheckpointRefsInput, ) => Effect.Effect; + + /** + * Prune hidden checkpoint refs, retaining only the bounded tail per thread. + * + * Turn 0 is preserved as the thread baseline; the remaining budget is spent + * on the newest positive turn refs. + */ + readonly pruneCheckpointRefs: ( + input: PruneCheckpointRefsInput, + ) => Effect.Effect; } >()("t3/checkpointing/CheckpointStore") {} @@ -157,6 +179,13 @@ export const make = Effect.gen(function* () { return yield* checkpoints.deleteCheckpointRefs(input); }); + const pruneCheckpointRefs: CheckpointStore["Service"]["pruneCheckpointRefs"] = Effect.fn( + "pruneCheckpointRefs", + )(function* (input) { + const checkpoints = yield* resolveCheckpoints("CheckpointStore.pruneCheckpointRefs", input.cwd); + return yield* checkpoints.pruneCheckpointRefs(input); + }); + return CheckpointStore.of({ isGitRepository, captureCheckpoint, @@ -164,6 +193,7 @@ export const make = Effect.gen(function* () { restoreCheckpoint, diffCheckpoints, deleteCheckpointRefs, + pruneCheckpointRefs, }); }); diff --git a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts index 300d1526bb9..3e39728808c 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts @@ -12,6 +12,7 @@ import { ThreadDeletionReactor } from "../Services/ThreadDeletionReactor.ts"; import { OrchestrationReactor } from "../Services/OrchestrationReactor.ts"; import { makeOrchestrationReactor } from "./OrchestrationReactor.ts"; import * as AgentAwarenessRelay from "../../relay/AgentAwarenessRelay.ts"; +import * as VcsMaintenanceReactor from "../../vcs/VcsMaintenanceReactor.ts"; describe("OrchestrationReactor", () => { let runtime: ManagedRuntime.ManagedRuntime | null = null; @@ -23,7 +24,7 @@ describe("OrchestrationReactor", () => { runtime = null; }); - it("starts provider ingestion, provider command, checkpoint, and thread deletion reactors", async () => { + it("starts provider ingestion, provider command, checkpoint, thread deletion, and maintenance reactors", async () => { const started: string[] = []; runtime = ManagedRuntime.make( @@ -73,6 +74,15 @@ describe("OrchestrationReactor", () => { }, }), ), + Layer.provideMerge( + Layer.succeed(VcsMaintenanceReactor.VcsMaintenanceReactor, { + start: () => { + started.push("vcs-maintenance-reactor"); + return Effect.void; + }, + sweep: () => Effect.void, + }), + ), ), ); @@ -86,6 +96,7 @@ describe("OrchestrationReactor", () => { "checkpoint-reactor", "thread-deletion-reactor", "agent-awareness-relay", + "vcs-maintenance-reactor", ]); await Effect.runPromise(Scope.close(scope, Exit.void)); diff --git a/apps/server/src/orchestration/Layers/OrchestrationReactor.ts b/apps/server/src/orchestration/Layers/OrchestrationReactor.ts index fb7543e31af..24abfed17b7 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationReactor.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationReactor.ts @@ -10,6 +10,7 @@ import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; import { ThreadDeletionReactor } from "../Services/ThreadDeletionReactor.ts"; import * as AgentAwarenessRelay from "../../relay/AgentAwarenessRelay.ts"; +import * as VcsMaintenanceReactor from "../../vcs/VcsMaintenanceReactor.ts"; export const makeOrchestrationReactor = Effect.gen(function* () { const providerRuntimeIngestion = yield* ProviderRuntimeIngestionService; @@ -17,6 +18,7 @@ export const makeOrchestrationReactor = Effect.gen(function* () { const checkpointReactor = yield* CheckpointReactor; const threadDeletionReactor = yield* ThreadDeletionReactor; const agentAwarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay; + const vcsMaintenanceReactor = yield* VcsMaintenanceReactor.VcsMaintenanceReactor; const start: OrchestrationReactorShape["start"] = Effect.fn("start")(function* () { yield* providerRuntimeIngestion.start(); @@ -24,6 +26,7 @@ export const makeOrchestrationReactor = Effect.gen(function* () { yield* checkpointReactor.start(); yield* threadDeletionReactor.start(); yield* agentAwarenessRelay.start(); + yield* vcsMaintenanceReactor.start(); }); return { diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index c930ba315e8..8d233c77ff9 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -15,6 +15,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts"; +import { MAX_THREAD_CHECKPOINTS } from "../checkpointRetention.ts"; import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; @@ -860,6 +861,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { projectId: asProjectId("project-context"), workspaceRoot: "/tmp/context-workspace", worktreePath: "/tmp/context-worktree", + trackedCheckpointTurnIds: [asTurnId("turn-1"), asTurnId("turn-2")], checkpoints: [ { turnId: asTurnId("turn-1"), @@ -885,6 +887,165 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { }), ); + it.effect("exposes only checkpoint refs in the retained tail", () => + Effect.gen(function* () { + const snapshotQuery = yield* ProjectionSnapshotQuery; + const sql = yield* SqlClient.SqlClient; + + yield* sql`DELETE FROM projection_projects`; + yield* sql`DELETE FROM projection_threads`; + yield* sql`DELETE FROM projection_turns`; + + yield* sql` + INSERT INTO projection_projects ( + project_id, + title, + workspace_root, + default_model_selection_json, + scripts_json, + created_at, + updated_at, + deleted_at + ) + VALUES ( + 'project-retention', + 'Retention Project', + '/tmp/retention-workspace', + NULL, + '[]', + '2026-03-03T00:00:00.000Z', + '2026-03-03T00:00:01.000Z', + NULL + ) + `; + + yield* sql` + INSERT INTO projection_threads ( + thread_id, + project_id, + title, + model_selection_json, + runtime_mode, + interaction_mode, + branch, + worktree_path, + latest_turn_id, + created_at, + updated_at, + archived_at, + deleted_at + ) + VALUES ( + 'thread-retention', + 'project-retention', + 'Retention Thread', + '{"provider":"codex","model":"gpt-5-codex"}', + 'full-access', + 'default', + 'feature/retention', + '/tmp/retention-worktree', + NULL, + '2026-03-03T00:00:02.000Z', + '2026-03-03T00:00:03.000Z', + NULL, + NULL + ) + `; + + yield* sql` + WITH RECURSIVE checkpoint_counts(checkpoint_turn_count) AS ( + SELECT 1 + UNION ALL + SELECT checkpoint_turn_count + 1 + FROM checkpoint_counts + WHERE checkpoint_turn_count < ${MAX_THREAD_CHECKPOINTS + 5} + ) + INSERT INTO projection_turns ( + thread_id, + turn_id, + pending_message_id, + source_proposed_plan_thread_id, + source_proposed_plan_id, + assistant_message_id, + state, + requested_at, + started_at, + completed_at, + checkpoint_turn_count, + checkpoint_ref, + checkpoint_status, + checkpoint_files_json + ) + SELECT + 'thread-retention', + 'turn-' || checkpoint_turn_count, + NULL, + NULL, + NULL, + NULL, + 'completed', + '2026-03-03T00:00:04.000Z', + '2026-03-03T00:00:04.000Z', + '2026-03-03T00:00:04.000Z', + checkpoint_turn_count, + 'refs/t3/checkpoints/thread-retention/turn/' || checkpoint_turn_count, + 'ready', + '[]' + FROM checkpoint_counts + `; + + const context = yield* snapshotQuery.getThreadCheckpointContext( + ThreadId.make("thread-retention"), + ); + assert.equal(context._tag, "Some"); + if (context._tag === "Some") { + assert.equal(context.value.trackedCheckpointTurnIds?.length, MAX_THREAD_CHECKPOINTS + 5); + assert.equal(context.value.trackedCheckpointTurnIds?.[0], asTurnId("turn-1")); + assert.equal(context.value.trackedCheckpointTurnIds?.at(-1), asTurnId("turn-505")); + assert.equal(context.value.checkpoints.length, MAX_THREAD_CHECKPOINTS + 1); + assert.equal(context.value.checkpoints[0]?.checkpointTurnCount, 5); + assert.equal( + context.value.checkpoints[0]?.checkpointRef, + asCheckpointRef("refs/t3/checkpoints/thread-retention/turn/5"), + ); + assert.equal(context.value.checkpoints.at(-1)?.checkpointTurnCount, 505); + } + + const prunedDiffContext = yield* snapshotQuery.getFullThreadDiffContext( + ThreadId.make("thread-retention"), + 4, + ); + assert.equal(prunedDiffContext._tag, "Some"); + if (prunedDiffContext._tag === "Some") { + assert.equal(prunedDiffContext.value.toCheckpointRef, null); + } + + const boundaryDiffContext = yield* snapshotQuery.getFullThreadDiffContext( + ThreadId.make("thread-retention"), + 5, + ); + assert.equal(boundaryDiffContext._tag, "Some"); + if (boundaryDiffContext._tag === "Some") { + assert.equal( + boundaryDiffContext.value.toCheckpointRef, + asCheckpointRef("refs/t3/checkpoints/thread-retention/turn/5"), + ); + } + + const retainedDiffContext = yield* snapshotQuery.getFullThreadDiffContext( + ThreadId.make("thread-retention"), + 6, + ); + assert.equal(retainedDiffContext._tag, "Some"); + if (retainedDiffContext._tag === "Some") { + assert.equal( + retainedDiffContext.value.toCheckpointRef, + asCheckpointRef("refs/t3/checkpoints/thread-retention/turn/6"), + ); + } + }), + ); + it.effect("keeps thread detail activity ordering consistent with shell snapshot ordering", () => Effect.gen(function* () { const snapshotQuery = yield* ProjectionSnapshotQuery; diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index fb8a37ea53c..b7f21061ad6 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -40,6 +40,7 @@ import { toPersistenceSqlError, type ProjectionRepositoryError, } from "../../persistence/Errors.ts"; +import { MAX_THREAD_CHECKPOINTS } from "../checkpointRetention.ts"; import { ProjectionCheckpoint } from "../../persistence/Services/ProjectionCheckpoints.ts"; import { ProjectionProject } from "../../persistence/Services/ProjectionProjects.ts"; import { ProjectionState } from "../../persistence/Services/ProjectionState.ts"; @@ -58,6 +59,8 @@ import { type ProjectionSnapshotQueryShape, } from "../Services/ProjectionSnapshotQuery.ts"; +const CHECKPOINT_DIFF_CONTEXT_KEEP_PER_THREAD = MAX_THREAD_CHECKPOINTS + 1; + const decodeReadModel = Schema.decodeUnknownEffect(OrchestrationReadModel); const decodeShellSnapshot = Schema.decodeUnknownEffect(OrchestrationShellSnapshot); const decodeThread = Schema.decodeUnknownEffect(OrchestrationThread); @@ -126,6 +129,9 @@ const ProjectionThreadCheckpointContextThreadRowSchema = Schema.Struct({ workspaceRoot: Schema.String, worktreePath: Schema.NullOr(Schema.String), }); +const ProjectionCheckpointTurnIdRowSchema = Schema.Struct({ + turnId: TurnId, +}); const FullThreadDiffContextLookupInput = Schema.Struct({ threadId: ThreadId, checkpointTurnCount: NonNegativeInt, @@ -556,16 +562,32 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { execute: () => sql` SELECT - thread_id AS "threadId", - turn_id AS "turnId", - checkpoint_turn_count AS "checkpointTurnCount", - checkpoint_ref AS "checkpointRef", - checkpoint_status AS "status", - checkpoint_files_json AS "files", - assistant_message_id AS "assistantMessageId", - completed_at AS "completedAt" - FROM projection_turns - WHERE checkpoint_turn_count IS NOT NULL + checkpoint_rows.thread_id AS "threadId", + checkpoint_rows.turn_id AS "turnId", + checkpoint_rows.checkpoint_turn_count AS "checkpointTurnCount", + checkpoint_rows.checkpoint_ref AS "checkpointRef", + checkpoint_rows.checkpoint_status AS "status", + checkpoint_rows.checkpoint_files_json AS "files", + checkpoint_rows.assistant_message_id AS "assistantMessageId", + checkpoint_rows.completed_at AS "completedAt" + FROM ( + SELECT + thread_id, + turn_id, + checkpoint_turn_count, + checkpoint_ref, + checkpoint_status, + checkpoint_files_json, + assistant_message_id, + completed_at, + ROW_NUMBER() OVER ( + PARTITION BY thread_id + ORDER BY checkpoint_turn_count DESC + ) AS checkpoint_rank + FROM projection_turns + WHERE checkpoint_turn_count IS NOT NULL + ) checkpoint_rows + WHERE checkpoint_rows.checkpoint_rank <= ${MAX_THREAD_CHECKPOINTS} ORDER BY thread_id ASC, checkpoint_turn_count ASC `, }); @@ -903,14 +925,43 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { execute: ({ threadId }) => sql` SELECT - thread_id AS "threadId", - turn_id AS "turnId", - checkpoint_turn_count AS "checkpointTurnCount", - checkpoint_ref AS "checkpointRef", - checkpoint_status AS "status", - checkpoint_files_json AS "files", - assistant_message_id AS "assistantMessageId", - completed_at AS "completedAt" + checkpoint_rows.thread_id AS "threadId", + checkpoint_rows.turn_id AS "turnId", + checkpoint_rows.checkpoint_turn_count AS "checkpointTurnCount", + checkpoint_rows.checkpoint_ref AS "checkpointRef", + checkpoint_rows.checkpoint_status AS "status", + checkpoint_rows.checkpoint_files_json AS "files", + checkpoint_rows.assistant_message_id AS "assistantMessageId", + checkpoint_rows.completed_at AS "completedAt" + FROM ( + SELECT + thread_id, + turn_id, + checkpoint_turn_count, + checkpoint_ref, + checkpoint_status, + checkpoint_files_json, + assistant_message_id, + completed_at, + ROW_NUMBER() OVER ( + ORDER BY checkpoint_turn_count DESC + ) AS checkpoint_rank + FROM projection_turns + WHERE thread_id = ${threadId} + AND checkpoint_turn_count IS NOT NULL + ) checkpoint_rows + WHERE checkpoint_rows.checkpoint_rank <= ${CHECKPOINT_DIFF_CONTEXT_KEEP_PER_THREAD} + ORDER BY checkpoint_rows.checkpoint_turn_count ASC + `, + }); + + const listCheckpointTurnIdRowsByThread = SqlSchema.findAll({ + Request: ThreadIdLookupInput, + Result: ProjectionCheckpointTurnIdRowSchema, + execute: ({ threadId }) => + sql` + SELECT + turn_id AS "turnId" FROM projection_turns WHERE thread_id = ${threadId} AND checkpoint_turn_count IS NOT NULL @@ -935,10 +986,20 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { AND turns.checkpoint_turn_count IS NOT NULL ) AS "latestCheckpointTurnCount", ( - SELECT turns.checkpoint_ref - FROM projection_turns AS turns - WHERE turns.thread_id = threads.thread_id - AND turns.checkpoint_turn_count = ${checkpointTurnCount} + SELECT checkpoint_rows.checkpoint_ref + FROM ( + SELECT + turns.checkpoint_turn_count, + turns.checkpoint_ref, + ROW_NUMBER() OVER ( + ORDER BY turns.checkpoint_turn_count DESC + ) AS checkpoint_rank + FROM projection_turns AS turns + WHERE turns.thread_id = threads.thread_id + AND turns.checkpoint_turn_count IS NOT NULL + ) checkpoint_rows + WHERE checkpoint_rows.checkpoint_turn_count = ${checkpointTurnCount} + AND checkpoint_rows.checkpoint_rank <= ${CHECKPOINT_DIFF_CONTEXT_KEEP_PER_THREAD} LIMIT 1 ) AS "toCheckpointRef" FROM projection_threads AS threads @@ -1813,20 +1874,31 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { return Option.none(); } - const checkpointRows = yield* listCheckpointRowsByThread({ threadId }).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "ProjectionSnapshotQuery.getThreadCheckpointContext:listCheckpoints:query", - "ProjectionSnapshotQuery.getThreadCheckpointContext:listCheckpoints:decodeRows", + const [checkpointRows, checkpointTurnIdRows] = yield* Effect.all([ + listCheckpointRowsByThread({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getThreadCheckpointContext:listCheckpoints:query", + "ProjectionSnapshotQuery.getThreadCheckpointContext:listCheckpoints:decodeRows", + ), ), ), - ); + listCheckpointTurnIdRowsByThread({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getThreadCheckpointContext:listCheckpointTurnIds:query", + "ProjectionSnapshotQuery.getThreadCheckpointContext:listCheckpointTurnIds:decodeRows", + ), + ), + ), + ]); return Option.some({ threadId: threadRow.value.threadId, projectId: threadRow.value.projectId, workspaceRoot: threadRow.value.workspaceRoot, worktreePath: threadRow.value.worktreePath, + trackedCheckpointTurnIds: checkpointTurnIdRows.map((row) => row.turnId), checkpoints: checkpointRows.map( (row): OrchestrationCheckpointSummary => ({ turnId: row.turnId, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 2a6a28efcbe..69202e01e30 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -140,12 +140,9 @@ function findProposedPlanById( return undefined; } -function hasCheckpointForTurn( - checkpoints: ReadonlyArray, - turnId: TurnId, -): boolean { - for (let index = 0; index < checkpoints.length; index += 1) { - if (checkpoints[index]?.turnId === turnId) { +function hasTrackedCheckpointTurnId(turnIds: ReadonlyArray, turnId: TurnId): boolean { + for (let index = 0; index < turnIds.length; index += 1) { + if (turnIds[index] === turnId) { return true; } } @@ -1725,7 +1722,10 @@ const make = Effect.gen(function* () { // (non-placeholder) capture from CheckpointReactor should not // be clobbered, and dispatching a duplicate placeholder for the // same turnId would produce an unstable checkpointTurnCount. - if (hasCheckpointForTurn(checkpointContext.checkpoints, turnId)) { + const trackedCheckpointTurnIds = + checkpointContext.trackedCheckpointTurnIds ?? + checkpointContext.checkpoints.map((checkpoint) => checkpoint.turnId); + if (hasTrackedCheckpointTurnId(trackedCheckpointTurnIds, turnId)) { // Already tracked; no-op. } else { const assistantMessageId = MessageId.make( diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index 7d85f0240f7..007cce74c0d 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -17,6 +17,7 @@ import type { OrchestrationThreadShell, ProjectId, ThreadId, + TurnId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; import type * as Option from "effect/Option"; @@ -39,6 +40,7 @@ export interface ProjectionThreadCheckpointContext { readonly workspaceRoot: string; readonly worktreePath: string | null; readonly checkpoints: ReadonlyArray; + readonly trackedCheckpointTurnIds?: ReadonlyArray; } export interface ProjectionFullThreadDiffContext { diff --git a/apps/server/src/orchestration/checkpointRetention.ts b/apps/server/src/orchestration/checkpointRetention.ts new file mode 100644 index 00000000000..e032a8acacd --- /dev/null +++ b/apps/server/src/orchestration/checkpointRetention.ts @@ -0,0 +1 @@ +export const MAX_THREAD_CHECKPOINTS = 500; diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index ad7bd2aada5..756a1fdb9c9 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -14,6 +14,7 @@ import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts"; +import { MAX_THREAD_CHECKPOINTS } from "./checkpointRetention.ts"; import { MessageSentPayloadSchema, ProjectCreatedPayload, @@ -35,7 +36,6 @@ import { type ThreadPatch = Partial>; const MAX_THREAD_MESSAGES = 2_000; -const MAX_THREAD_CHECKPOINTS = 500; function checkpointStatusToLatestTurnState(status: "ready" | "missing" | "error") { if (status === "error") return "error" as const; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 0576dc36d0b..97098d80dcd 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -73,6 +73,7 @@ import * as VcsProjectConfig from "./vcs/VcsProjectConfig.ts"; import * as VcsProcess from "./vcs/VcsProcess.ts"; import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts"; +import * as VcsMaintenanceReactor from "./vcs/VcsMaintenanceReactor.ts"; import * as GitWorkflowService from "./git/GitWorkflowService.ts"; import * as ReviewService from "./review/ReviewService.ts"; import * as SourceControlProviderRegistry from "./sourceControl/SourceControlProviderRegistry.ts"; @@ -174,6 +175,7 @@ const ReactorLayerLive = Layer.empty.pipe( Layer.provideMerge(ProviderCommandReactorLive), Layer.provideMerge(CheckpointReactorLive), Layer.provideMerge(ThreadDeletionReactorLive), + Layer.provideMerge(VcsMaintenanceReactor.layer), Layer.provideMerge( ScheduledTasksReactorLive.pipe(Layer.provide(BootstrapTurnStartDispatcher.layer)), ), diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index 55aa8f38835..aace787d9ee 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -253,6 +253,7 @@ export class GitVcsDriver extends Context.Service< readonly removeWorktree: ( input: VcsRemoveWorktreeInput, ) => Effect.Effect; + readonly pruneWorktrees: (cwd: string) => Effect.Effect; readonly renameBranch: ( input: GitRenameBranchInput, ) => Effect.Effect; @@ -270,6 +271,7 @@ export class GitVcsDriver extends Context.Service< const WORKSPACE_FILES_MAX_OUTPUT_BYTES = 16 * 1024 * 1024; const GIT_CHECK_IGNORE_MAX_STDIN_BYTES = 256 * 1024; const CHECKPOINT_DIFF_MAX_OUTPUT_BYTES = 10_000_000; +const CHECKPOINT_REFS_PREFIX = "refs/t3/checkpoints"; const WORKSPACE_GIT_HARDENED_CONFIG_ARGS = [ "-c", "core.fsmonitor=false", @@ -327,6 +329,96 @@ function chunkPathsForGitCheckIgnore(relativePaths: ReadonlyArray): stri return chunks; } +interface CheckpointRefEntry { + readonly refName: string; + readonly threadSegment: string; + readonly turnCount: number; +} + +function parseCheckpointRefEntry(refName: string): CheckpointRefEntry | null { + const prefix = `${CHECKPOINT_REFS_PREFIX}/`; + if (!refName.startsWith(prefix)) { + return null; + } + + const rest = refName.slice(prefix.length); + const match = /^([^/]+)\/turn\/(\d+)$/.exec(rest); + if (!match) { + return null; + } + + const threadSegment = match[1]; + const turnCount = Number.parseInt(match[2] ?? "", 10); + if (!threadSegment || !Number.isSafeInteger(turnCount) || turnCount < 0) { + return null; + } + + return { refName, threadSegment, turnCount }; +} + +function selectCheckpointRefsToPrune( + entries: ReadonlyArray, + keepPerThread: number, +): { + readonly refsToDelete: ReadonlyArray; + readonly keptCount: number; + readonly threadCount: number; +} { + const boundedKeepPerThread = Math.max(1, Math.floor(keepPerThread)); + const keepPositiveTurnCount = Math.max(0, boundedKeepPerThread - 1); + const byThread = new Map(); + + for (const entry of entries) { + const threadEntries = byThread.get(entry.threadSegment); + if (threadEntries) { + threadEntries.push(entry); + } else { + byThread.set(entry.threadSegment, [entry]); + } + } + + const refsToDelete: string[] = []; + let keptCount = 0; + + for (const threadEntries of byThread.values()) { + const baseline = threadEntries + .filter((entry) => entry.turnCount === 0) + .toSorted((left, right) => left.refName.localeCompare(right.refName)); + const firstBaseline = baseline[0]; + const positiveTurnKeepCount = + firstBaseline === undefined ? boundedKeepPerThread : keepPositiveTurnCount; + const newestPositiveTurns = threadEntries + .filter((entry) => entry.turnCount > 0) + .toSorted((left, right) => { + if (left.turnCount !== right.turnCount) { + return right.turnCount - left.turnCount; + } + return left.refName.localeCompare(right.refName); + }); + + const keptRefs = new Set(); + if (firstBaseline) { + keptRefs.add(firstBaseline.refName); + } + for (const entry of newestPositiveTurns.slice(0, positiveTurnKeepCount)) { + keptRefs.add(entry.refName); + } + + keptCount += keptRefs.size; + for (const entry of threadEntries) { + if (!keptRefs.has(entry.refName)) { + refsToDelete.push(entry.refName); + } + } + } + + return { + refsToDelete, + keptCount, + threadCount: byThread.size, + }; +} + function parseGitRemoteVerboseOutput( output: string, ): Map { @@ -848,6 +940,58 @@ export const makeVcsDriverShape = Effect.fn("makeGitVcsDriverShape")(function* ( ); }, ), + + pruneCheckpointRefs: Effect.fn("GitVcsDriver.checkpoints.pruneCheckpointRefs")( + function* (input) { + const result = yield* execute({ + operation: "GitVcsDriver.checkpoints.pruneCheckpointRefs.list", + cwd: input.cwd, + args: ["for-each-ref", "--format=%(refname)", CHECKPOINT_REFS_PREFIX], + allowNonZeroExit: true, + timeoutMs: 10_000, + maxOutputBytes: 2_000_000, + }); + + if (result.exitCode !== 0) { + return yield* new VcsProcessExitError({ + operation: "GitVcsDriver.checkpoints.pruneCheckpointRefs.list", + command: "git for-each-ref", + cwd: input.cwd, + exitCode: result.exitCode, + detail: result.stderr.trim() || "Checkpoint ref listing failed.", + }); + } + + const checkpointRefs = result.stdout + .split("\n") + .map((line) => line.trim()) + .filter((line) => line.length > 0) + .flatMap((line) => { + const entry = parseCheckpointRefEntry(line); + return entry ? [entry] : []; + }); + const selected = selectCheckpointRefsToPrune(checkpointRefs, input.keepPerThread); + + yield* Effect.forEach( + selected.refsToDelete, + (checkpointRef) => + execute({ + operation: "GitVcsDriver.checkpoints.pruneCheckpointRefs.delete", + cwd: input.cwd, + args: ["update-ref", "-d", checkpointRef], + timeoutMs: 5_000, + }), + { discard: true }, + ); + + return { + scannedCount: checkpointRefs.length, + keptCount: selected.keptCount, + deletedCount: selected.refsToDelete.length, + threadCount: selected.threadCount, + }; + }, + ), }; return { diff --git a/apps/server/src/vcs/GitVcsDriverCore.test.ts b/apps/server/src/vcs/GitVcsDriverCore.test.ts index dc58fc2543c..68258b531a0 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.test.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.test.ts @@ -523,6 +523,53 @@ it.layer(TestLayer)("GitVcsDriver core integration", (it) => { }), ); + it.effect("keeps local refs available when a remote-tracking ref is broken", () => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + const { initialBranch } = yield* initRepoWithCommit(cwd); + const fileSystem = yield* FileSystem.FileSystem; + const pathService = yield* Path.Path; + const brokenRefPath = pathService.join(cwd, ".git", "refs", "remotes", "origin", "broken"); + yield* fileSystem.makeDirectory(pathService.dirname(brokenRefPath), { recursive: true }); + yield* fileSystem.writeFileString( + brokenRefPath, + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n", + ); + const driver = yield* GitVcsDriver.GitVcsDriver; + + const refs = yield* driver.listRefs({ cwd, includeMatchingRemoteRefs: true }); + + assert.equal(refs.isRepo, true); + assert.equal( + refs.refs.some((ref) => ref.name === initialBranch && !ref.isRemote), + true, + ); + }), + ); + + it.effect("keeps healthy local refs available when a local ref target is missing", () => + Effect.gen(function* () { + const cwd = yield* makeTmpDir(); + const { initialBranch } = yield* initRepoWithCommit(cwd); + const fileSystem = yield* FileSystem.FileSystem; + const pathService = yield* Path.Path; + const brokenRefPath = pathService.join(cwd, ".git", "refs", "heads", "broken"); + yield* fileSystem.writeFileString( + brokenRefPath, + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n", + ); + const driver = yield* GitVcsDriver.GitVcsDriver; + + const refs = yield* driver.listRefs({ cwd }); + + assert.equal(refs.isRepo, true); + assert.equal( + refs.refs.some((ref) => ref.name === initialBranch && !ref.isRemote), + true, + ); + }), + ); + it.effect("creates, checks out, renames, and lists refs", () => Effect.gen(function* () { const cwd = yield* makeTmpDir(); @@ -589,6 +636,11 @@ it.layer(TestLayer)("GitVcsDriver core integration", (it) => { assert.equal(created.worktree.path, worktreePath); assert.equal(created.worktree.refName, "feature/worktree"); assert.equal(yield* git(worktreePath, ["branch", "--show-current"]), "feature/worktree"); + const refs = yield* driver.listRefs({ cwd }); + assert.equal( + refs.refs.find((ref) => ref.name === "feature/worktree")?.worktreePath, + worktreePath, + ); yield* driver.removeWorktree({ cwd, path: worktreePath }); const fileSystem = yield* FileSystem.FileSystem; diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index a406cbce549..2c2057dda94 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -161,21 +161,6 @@ function parsePorcelainPath(line: string): string | null { return filePath.length > 0 ? filePath : null; } -function parseBranchLine(line: string): { name: string; current: boolean } | null { - const trimmed = line.trim(); - if (trimmed.length === 0) return null; - - const name = trimmed.replace(/^[*+]\s+/, ""); - // Exclude symbolic refs like: "origin/HEAD -> origin/main". - // Exclude detached HEAD pseudo-refs like: "(HEAD detached at origin/main)". - if (name.includes(" -> ") || name.startsWith("(")) return null; - - return { - name, - current: trimmed.startsWith("* "), - }; -} - function filterBranchesForListQuery( refs: ReadonlyArray, query?: string, @@ -210,6 +195,46 @@ function paginateBranches(input: { }; } +interface GitRefListEntry { + readonly fullName: string; + readonly lastCommit: number; +} + +function parseForEachRefLine(line: string): GitRefListEntry | null { + const [fullName, lastCommitRaw] = line.split("\t"); + if (!fullName) { + return null; + } + const lastCommit = Number.parseInt(lastCommitRaw ?? "0", 10); + return { + fullName, + lastCommit: Number.isFinite(lastCommit) ? lastCommit : 0, + }; +} + +function parseForEachRefEntries(stdout: string): ReadonlyArray { + return Arr.filterMap(stdout.split("\n"), (line) => { + const parsed = parseForEachRefLine(line); + return parsed === null ? Result.failVoid : Result.succeed(parsed); + }); +} + +function parseBranchListEntries(stdout: string): ReadonlyArray { + return Arr.filterMap(stdout.split("\n"), (line) => { + let branchName = line.trim(); + if (branchName.startsWith("*") || branchName.startsWith("+")) { + branchName = branchName.slice(1).trim(); + } + if (!branchName || branchName.startsWith("(")) { + return Result.failVoid; + } + return Result.succeed({ + fullName: `refs/heads/${branchName}`, + lastCommit: 0, + }); + }); +} + function splitNullSeparatedPaths(input: string, truncated: boolean): string[] { const parts = input.split("\0"); if (parts.length === 0) return []; @@ -1267,42 +1292,6 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* }; }); - const readBranchRecency = Effect.fn("readBranchRecency")(function* (cwd: string) { - const branchRecency = yield* executeGit( - "GitVcsDriver.readBranchRecency", - cwd, - [ - "for-each-ref", - "--format=%(refname:short)%09%(committerdate:unix)", - "refs/heads", - "refs/remotes", - ], - { - timeoutMs: 15_000, - allowNonZeroExit: true, - }, - ); - - const branchLastCommit = new Map(); - if (branchRecency.exitCode !== 0) { - return branchLastCommit; - } - - for (const line of branchRecency.stdout.split("\n")) { - if (line.length === 0) { - continue; - } - const [name, lastCommitRaw] = line.split("\t"); - if (!name) { - continue; - } - const lastCommit = Number.parseInt(lastCommitRaw ?? "0", 10); - branchLastCommit.set(name, Number.isFinite(lastCommit) ? lastCommit : 0); - } - - return branchLastCommit; - }); - const readStatusDetailsLocal = Effect.fn("readStatusDetailsLocal")(function* (cwd: string) { const statusResult = yield* executeGit( "GitVcsDriver.statusDetails.status", @@ -1982,13 +1971,11 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* const listRefs: GitVcsDriver.GitVcsDriver["Service"]["listRefs"] = Effect.fn("listRefs")( function* (input) { - const branchRecencyPromise = readBranchRecency(input.cwd).pipe( - Effect.orElseSucceed(() => new Map()), - ); - const localBranchResult = yield* executeGit( - "GitVcsDriver.listRefs.branchNoColor", + const refListFormat = "%(refname)%09%(committerdate:unix)"; + const localRefListResult = yield* executeGit( + "GitVcsDriver.listRefs.localForEachRef", input.cwd, - ["branch", "--no-color", "--no-column"], + ["for-each-ref", `--format=${refListFormat}`, "refs/heads"], { timeoutMs: 10_000, allowNonZeroExit: true, @@ -2008,8 +1995,9 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* }), ); - if (localBranchResult.exitCode !== 0) { - const stderr = localBranchResult.stderr.trim(); + let localRefEntries: ReadonlyArray; + if (localRefListResult.exitCode !== 0) { + const stderr = localRefListResult.stderr.trim(); if (isNonRepositoryGitStderr(stderr)) { return { refs: [], @@ -2019,23 +2007,52 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* totalCount: 0, }; } - return yield* new GitCommandError({ - ...gitCommandContext({ - operation: "GitVcsDriver.listRefs", - cwd: input.cwd, - args: ["branch", "--no-color", "--no-column"], - }), - detail: "Git branch listing failed.", - exitCode: localBranchResult.exitCode, - stdoutLength: localBranchResult.stdout.length, - stderrLength: localBranchResult.stderr.length, - }); + + yield* Effect.logWarning( + `GitVcsDriver.listRefs: local ref lookup returned code ${localRefListResult.exitCode} for ${input.cwd}: ${stderr}. Falling back to git branch.`, + ); + + const fallbackLocalBranchResult = yield* executeGit( + "GitVcsDriver.listRefs.localBranchFallback", + input.cwd, + ["branch", "--no-color", "--no-column"], + { + timeoutMs: 10_000, + allowNonZeroExit: true, + }, + ); + if (fallbackLocalBranchResult.exitCode !== 0) { + const fallbackStderr = fallbackLocalBranchResult.stderr.trim(); + if (isNonRepositoryGitStderr(fallbackStderr)) { + return { + refs: [], + isRepo: false, + hasPrimaryRemote: false, + nextCursor: null, + totalCount: 0, + }; + } + return yield* new GitCommandError({ + ...gitCommandContext({ + operation: "GitVcsDriver.listRefs", + cwd: input.cwd, + args: ["for-each-ref", `--format=${refListFormat}`, "refs/heads"], + }), + detail: "Git local ref listing failed.", + exitCode: localRefListResult.exitCode, + stdoutLength: localRefListResult.stdout.length, + stderrLength: localRefListResult.stderr.length, + }); + } + localRefEntries = parseBranchListEntries(fallbackLocalBranchResult.stdout); + } else { + localRefEntries = parseForEachRefEntries(localRefListResult.stdout); } - const remoteBranchResultEffect = executeGit( - "GitVcsDriver.listRefs.remoteBranches", + const remoteRefListResultEffect = executeGit( + "GitVcsDriver.listRefs.remoteForEachRef", input.cwd, - ["branch", "--no-color", "--no-column", "--remotes"], + ["for-each-ref", `--format=${refListFormat}`, "refs/remotes"], { timeoutMs: 10_000, allowNonZeroExit: true, @@ -2064,6 +2081,16 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* }), ); + const currentBranchResultEffect = executeGit( + "GitVcsDriver.listRefs.currentBranch", + input.cwd, + ["symbolic-ref", "--quiet", "--short", "HEAD"], + { + timeoutMs: 5_000, + allowNonZeroExit: true, + }, + ); + const remoteNamesResultEffect = executeGit( "GitVcsDriver.listRefs.remoteNames", input.cwd, @@ -2096,44 +2123,49 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* }), ); - const [defaultRef, worktreeList, remoteBranchResult, remoteNamesResult, branchLastCommit] = - yield* Effect.all( - [ - executeGit( - "GitVcsDriver.listRefs.defaultRef", - input.cwd, - ["symbolic-ref", "refs/remotes/origin/HEAD"], - { - timeoutMs: 5_000, - allowNonZeroExit: true, - }, - ), - executeGit( - "GitVcsDriver.listRefs.worktreeList", - input.cwd, - ["worktree", "list", "--porcelain"], - { - timeoutMs: 5_000, - allowNonZeroExit: true, - }, - ), - remoteBranchResultEffect, - remoteNamesResultEffect, - branchRecencyPromise, - ], - { concurrency: "unbounded" }, - ); + const [ + remoteRefListResult, + currentBranchResult, + defaultRef, + worktreeList, + remoteNamesResult, + ] = yield* Effect.all( + [ + remoteRefListResultEffect, + currentBranchResultEffect, + executeGit( + "GitVcsDriver.listRefs.defaultRef", + input.cwd, + ["symbolic-ref", "refs/remotes/origin/HEAD"], + { + timeoutMs: 5_000, + allowNonZeroExit: true, + }, + ), + executeGit( + "GitVcsDriver.listRefs.worktreeList", + input.cwd, + ["worktree", "list", "--porcelain"], + { + timeoutMs: 5_000, + allowNonZeroExit: true, + }, + ), + remoteNamesResultEffect, + ], + { concurrency: "unbounded" }, + ); const remoteNames = remoteNamesResult.exitCode === 0 ? parseRemoteNames(remoteNamesResult.stdout) : []; - if (remoteBranchResult.exitCode !== 0 && remoteBranchResult.stderr.trim().length > 0) { + if (remoteNamesResult.exitCode !== 0 && remoteNamesResult.stderr.trim().length > 0) { yield* Effect.logWarning( - `GitVcsDriver.listRefs: remote refName lookup returned code ${remoteBranchResult.exitCode} for ${input.cwd}: ${remoteBranchResult.stderr.trim()}. Falling back to an empty remote refName list.`, + `GitVcsDriver.listRefs: remote name lookup returned code ${remoteNamesResult.exitCode} for ${input.cwd}: ${remoteNamesResult.stderr.trim()}. Falling back to an empty remote name list.`, ); } - if (remoteNamesResult.exitCode !== 0 && remoteNamesResult.stderr.trim().length > 0) { + if (remoteRefListResult.exitCode !== 0 && remoteRefListResult.stderr.trim().length > 0) { yield* Effect.logWarning( - `GitVcsDriver.listRefs: remote name lookup returned code ${remoteNamesResult.exitCode} for ${input.cwd}: ${remoteNamesResult.stderr.trim()}. Falling back to an empty remote name list.`, + `GitVcsDriver.listRefs: remote ref lookup returned code ${remoteRefListResult.exitCode} for ${input.cwd}: ${remoteRefListResult.stderr.trim()}. Falling back to an empty remote ref list.`, ); } @@ -2142,81 +2174,112 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* ? defaultRef.stdout.trim().replace(/^refs\/remotes\/origin\//, "") : null; + const currentBranch = + currentBranchResult.exitCode === 0 ? currentBranchResult.stdout.trim() || null : null; + const worktreeMap = new Map(); if (worktreeList.exitCode === 0) { let currentPath: string | null = null; + let currentBranchName: string | null = null; + let currentPrunable = false; + const flushWorktree = () => { + if (currentPath && currentBranchName && !currentPrunable) { + worktreeMap.set(currentBranchName, currentPath); + } + currentPath = null; + currentBranchName = null; + currentPrunable = false; + }; for (const line of worktreeList.stdout.split("\n")) { if (line.startsWith("worktree ")) { - const candidatePath = line.slice("worktree ".length); - const exists = yield* fileSystem.stat(candidatePath).pipe( - Effect.map(() => true), - Effect.orElseSucceed(() => false), - ); - currentPath = exists ? candidatePath : null; + flushWorktree(); + currentPath = line.slice("worktree ".length); } else if (line.startsWith("branch refs/heads/") && currentPath) { - worktreeMap.set(line.slice("branch refs/heads/".length), currentPath); + currentBranchName = line.slice("branch refs/heads/".length); + } else if (line.startsWith("prunable ")) { + currentPrunable = true; } else if (line === "") { - currentPath = null; + flushWorktree(); } } + flushWorktree(); } - const localBranches = Arr.filterMap(localBranchResult.stdout.split("\n"), (line) => { - const refName = parseBranchLine(line); - return refName === null - ? Result.failVoid - : Result.succeed({ - name: refName.name, - current: refName.current, - isRemote: false, - isDefault: refName.name === defaultBranch, - worktreePath: worktreeMap.get(refName.name) ?? null, - }); - }).toSorted((a, b) => { - const aPriority = a.current ? 0 : a.isDefault ? 1 : 2; - const bPriority = b.current ? 0 : b.isDefault ? 1 : 2; - if (aPriority !== bPriority) return aPriority - bPriority; - - const aLastCommit = branchLastCommit.get(a.name) ?? 0; - const bLastCommit = branchLastCommit.get(b.name) ?? 0; - if (aLastCommit !== bLastCommit) return bLastCommit - aLastCommit; - return a.name.localeCompare(b.name); - }); - - const remoteBranches = - remoteBranchResult.exitCode === 0 - ? Arr.filterMap(remoteBranchResult.stdout.split("\n"), (line) => { - const refName = parseBranchLine(line); - if (refName === null) { - return Result.failVoid; - } - const parsedRemoteRef = parseRemoteRefWithRemoteNames(refName.name, remoteNames); - const remoteBranch: { - name: string; - current: boolean; - isRemote: boolean; - remoteName?: string; - isDefault: boolean; - worktreePath: string | null; - } = { - name: refName.name, - current: false, - isRemote: true, - isDefault: false, - worktreePath: null, - }; - if (parsedRemoteRef) { - remoteBranch.remoteName = parsedRemoteRef.remoteName; - } - return Result.succeed(remoteBranch); - }).toSorted((a, b) => { - const aLastCommit = branchLastCommit.get(a.name) ?? 0; - const bLastCommit = branchLastCommit.get(b.name) ?? 0; - if (aLastCommit !== bLastCommit) return bLastCommit - aLastCommit; - return a.name.localeCompare(b.name); - }) + const remoteRefEntries = + remoteRefListResult.exitCode === 0 + ? parseForEachRefEntries(remoteRefListResult.stdout) : []; + const localBranches = localRefEntries + .map((entry) => { + const name = entry.fullName.slice("refs/heads/".length); + return { + name, + current: name === currentBranch, + isRemote: false, + isDefault: name === defaultBranch, + worktreePath: worktreeMap.get(name) ?? null, + lastCommit: entry.lastCommit, + }; + }) + .toSorted((a, b) => { + const aPriority = a.current ? 0 : a.isDefault ? 1 : 2; + const bPriority = b.current ? 0 : b.isDefault ? 1 : 2; + if (aPriority !== bPriority) return aPriority - bPriority; + + if (a.lastCommit !== b.lastCommit) return b.lastCommit - a.lastCommit; + return a.name.localeCompare(b.name); + }) + .map((branch) => ({ + name: branch.name, + current: branch.current, + isRemote: branch.isRemote, + isDefault: branch.isDefault, + worktreePath: branch.worktreePath, + })); + + const remoteBranches = remoteRefEntries + .filter( + (entry) => + entry.fullName.startsWith("refs/remotes/") && !entry.fullName.endsWith("/HEAD"), + ) + .map((entry) => { + const name = entry.fullName.slice("refs/remotes/".length); + const parsedRemoteRef = parseRemoteRefWithRemoteNames(name, remoteNames); + const remoteBranch: { + name: string; + current: boolean; + isRemote: boolean; + remoteName?: string; + isDefault: boolean; + worktreePath: string | null; + lastCommit: number; + } = { + name, + current: false, + isRemote: true, + isDefault: false, + worktreePath: null, + lastCommit: entry.lastCommit, + }; + if (parsedRemoteRef) { + remoteBranch.remoteName = parsedRemoteRef.remoteName; + } + return remoteBranch; + }) + .toSorted((a, b) => { + if (a.lastCommit !== b.lastCommit) return b.lastCommit - a.lastCommit; + return a.name.localeCompare(b.name); + }) + .map((branch) => ({ + name: branch.name, + current: branch.current, + isRemote: branch.isRemote, + ...(branch.remoteName ? { remoteName: branch.remoteName } : {}), + isDefault: branch.isDefault, + worktreePath: branch.worktreePath, + })); + const allBranches = input.includeMatchingRemoteRefs ? [...localBranches, ...remoteBranches] : dedupeRemoteBranchesWithLocalMatches([...localBranches, ...remoteBranches]); @@ -2385,6 +2448,12 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* }); }); + const pruneWorktrees: GitVcsDriver.GitVcsDriver["Service"]["pruneWorktrees"] = (cwd) => + executeGit("GitVcsDriver.pruneWorktrees", cwd, ["worktree", "prune"], { + timeoutMs: 15_000, + fallbackErrorDetail: "git worktree prune failed", + }).pipe(Effect.asVoid); + const renameBranch: GitVcsDriver.GitVcsDriver["Service"]["renameBranch"] = Effect.fn( "renameBranch", )(function* (input) { @@ -2553,6 +2622,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* fetchRemoteTrackingBranch, setBranchUpstream, removeWorktree, + pruneWorktrees, renameBranch, createRef, switchRef, diff --git a/apps/server/src/vcs/VcsDriver.ts b/apps/server/src/vcs/VcsDriver.ts index f2daf793502..cf69cd805c9 100644 --- a/apps/server/src/vcs/VcsDriver.ts +++ b/apps/server/src/vcs/VcsDriver.ts @@ -38,6 +38,18 @@ export interface VcsDeleteCheckpointRefsInput { readonly checkpointRefs: ReadonlyArray; } +export interface VcsPruneCheckpointRefsInput { + readonly cwd: string; + readonly keepPerThread: number; +} + +export interface VcsPruneCheckpointRefsResult { + readonly scannedCount: number; + readonly keptCount: number; + readonly deletedCount: number; + readonly threadCount: number; +} + export interface VcsCheckpointOps { readonly captureCheckpoint: (input: VcsCaptureCheckpointInput) => Effect.Effect; readonly hasCheckpointRef: ( @@ -50,6 +62,9 @@ export interface VcsCheckpointOps { readonly deleteCheckpointRefs: ( input: VcsDeleteCheckpointRefsInput, ) => Effect.Effect; + readonly pruneCheckpointRefs: ( + input: VcsPruneCheckpointRefsInput, + ) => Effect.Effect; } export class VcsDriver extends Context.Service< diff --git a/apps/server/src/vcs/VcsMaintenanceReactor.test.ts b/apps/server/src/vcs/VcsMaintenanceReactor.test.ts new file mode 100644 index 00000000000..9712b5cfc17 --- /dev/null +++ b/apps/server/src/vcs/VcsMaintenanceReactor.test.ts @@ -0,0 +1,318 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { MAX_THREAD_CHECKPOINTS } from "../orchestration/checkpointRetention.ts"; +import { + CHECKPOINT_REFS_KEEP_PER_THREAD, + isWorktreePathListed, + selectStaleWorktreeReapCandidates, + shouldRetainWorktreeMetadataAfterListFailure, + type WorktreeMaintenanceRow, +} from "./VcsMaintenanceReactor.ts"; + +const NOW = Date.parse("2026-07-06T12:00:00.000Z"); + +function row( + overrides: Partial & Pick, +): WorktreeMaintenanceRow { + return { + threadId: overrides.threadId, + projectCwd: overrides.projectCwd ?? "/repo", + worktreePath: overrides.worktreePath ?? `/worktrees/${overrides.threadId}`, + worktreeRemovable: overrides.worktreeRemovable ?? true, + worktreeRemovalPath: overrides.worktreeRemovalPath ?? null, + updatedAt: overrides.updatedAt ?? "2026-07-05T00:00:00.000Z", + archivedAt: overrides.archivedAt ?? null, + deletedAt: overrides.deletedAt ?? null, + sessionStatus: overrides.sessionStatus ?? "stopped", + runtimeStatus: overrides.runtimeStatus ?? null, + pendingApprovalCount: overrides.pendingApprovalCount ?? 0, + pendingUserInputCount: overrides.pendingUserInputCount ?? 0, + }; +} + +describe("selectStaleWorktreeReapCandidates", () => { + it("keeps physical refs for the turn zero baseline and boundary predecessor", () => { + expect(CHECKPOINT_REFS_KEEP_PER_THREAD).toBe(MAX_THREAD_CHECKPOINTS + 2); + }); + + it("selects old removable stopped worktrees", () => { + expect( + selectStaleWorktreeReapCandidates([row({ threadId: "thread-old" })], ["/repo"], NOW, { + stoppedAgeMs: 60_000, + }), + ).toEqual([ + { + threadId: "thread-old", + threadIds: ["thread-old"], + projectCwd: "/repo", + path: "/worktrees/thread-old", + }, + ]); + }); + + it("coalesces shared stale archived worktree paths", () => { + expect( + selectStaleWorktreeReapCandidates( + [ + row({ + threadId: "archived-a", + worktreePath: "/worktrees/shared-archived", + archivedAt: "2026-07-06T11:30:00.000Z", + updatedAt: "2026-07-06T11:30:00.000Z", + }), + row({ + threadId: "archived-b", + worktreePath: "/worktrees/shared-archived", + archivedAt: "2026-07-06T11:30:00.000Z", + updatedAt: "2026-07-06T11:30:00.000Z", + }), + ], + ["/repo"], + NOW, + { + archivedAgeMs: 20 * 60_000, + stoppedAgeMs: 60 * 60_000, + }, + ), + ).toEqual([ + { + threadId: "archived-a", + threadIds: ["archived-a", "archived-b"], + projectCwd: "/repo", + path: "/worktrees/shared-archived", + }, + ]); + }); + + it("does not let deleted metadata rows retain a stale worktree path", () => { + expect( + selectStaleWorktreeReapCandidates( + [ + row({ + threadId: "stale", + worktreePath: "/worktrees/deleted-row-overlap", + }), + row({ + threadId: "deleted-metadata", + worktreePath: "/worktrees/deleted-row-overlap", + worktreeRemovable: false, + deletedAt: "2026-07-06T11:30:00.000Z", + }), + ], + ["/repo"], + NOW, + { + stoppedAgeMs: 60_000, + }, + ), + ).toEqual([ + { + threadId: "stale", + threadIds: ["stale"], + projectCwd: "/repo", + path: "/worktrees/deleted-row-overlap", + }, + ]); + }); + + it("ignores stale pending counters on deleted threads", () => { + expect( + selectStaleWorktreeReapCandidates( + [ + row({ + threadId: "deleted-pending", + deletedAt: "2026-07-06T11:30:00.000Z", + updatedAt: "2026-07-06T11:30:00.000Z", + pendingApprovalCount: 1, + pendingUserInputCount: 1, + }), + ], + ["/repo"], + NOW, + { + archivedAgeMs: 20 * 60_000, + }, + ), + ).toEqual([ + { + threadId: "deleted-pending", + threadIds: ["deleted-pending"], + projectCwd: "/repo", + path: "/worktrees/deleted-pending", + }, + ]); + }); + + it("ignores stale pending counters on archived threads", () => { + expect( + selectStaleWorktreeReapCandidates( + [ + row({ + threadId: "archived-pending", + archivedAt: "2026-07-06T11:30:00.000Z", + updatedAt: "2026-07-06T11:30:00.000Z", + pendingApprovalCount: 1, + pendingUserInputCount: 1, + }), + ], + ["/repo"], + NOW, + { + archivedAgeMs: 20 * 60_000, + }, + ), + ).toEqual([ + { + threadId: "archived-pending", + threadIds: ["archived-pending"], + projectCwd: "/repo", + path: "/worktrees/archived-pending", + }, + ]); + }); + + it("keeps active, pending, young, shared, and project-root paths", () => { + const rows = [ + row({ threadId: "running", sessionStatus: "running" }), + row({ threadId: "starting", sessionStatus: "starting" }), + row({ threadId: "waiting", sessionStatus: "waiting" }), + row({ threadId: "idle", sessionStatus: "idle" }), + row({ threadId: "interrupted", sessionStatus: "interrupted" }), + row({ threadId: "ready-runtime", runtimeStatus: "ready" }), + row({ threadId: "waiting-runtime", runtimeStatus: "waiting" }), + row({ threadId: "starting-runtime", runtimeStatus: "starting" }), + row({ threadId: "approval", pendingApprovalCount: 1 }), + row({ threadId: "input", pendingUserInputCount: 1 }), + row({ threadId: "young", updatedAt: "2026-07-06T11:59:30.000Z" }), + row({ threadId: "not-removable", worktreeRemovable: false }), + row({ threadId: "project-root", worktreePath: "/repo" }), + row({ threadId: "nested-project-path", worktreePath: "/repo/worktrees/old" }), + row({ threadId: "shared-a", worktreePath: "/worktrees/shared" }), + row({ + threadId: "shared-b", + worktreePath: "/worktrees/shared/nested", + worktreeRemovable: false, + }), + ]; + + expect( + selectStaleWorktreeReapCandidates(rows, ["/repo"], NOW, { + stoppedAgeMs: 60_000, + }), + ).toEqual([]); + }); + + it("keeps stale paths that overlap a live thread path", () => { + const rows = [ + row({ + threadId: "stale-parent", + worktreePath: "/worktrees/shared", + }), + row({ + threadId: "live-child", + worktreePath: "/worktrees/shared/nested", + worktreeRemovable: false, + }), + row({ + threadId: "stale-child", + worktreePath: "/worktrees/live/nested", + }), + row({ + threadId: "live-parent", + worktreePath: "/worktrees/live", + worktreeRemovable: false, + }), + ]; + + expect( + selectStaleWorktreeReapCandidates(rows, ["/repo"], NOW, { + stoppedAgeMs: 60_000, + }), + ).toEqual([]); + }); + + it("uses the shorter archived age for archived and deleted threads", () => { + const rows = [ + row({ + threadId: "archived", + archivedAt: "2026-07-06T11:30:00.000Z", + updatedAt: "2026-07-06T11:30:00.000Z", + }), + row({ + threadId: "deleted", + deletedAt: "2026-07-06T11:30:00.000Z", + updatedAt: "2026-07-06T11:30:00.000Z", + }), + row({ + threadId: "stopped", + updatedAt: "2026-07-06T11:30:00.000Z", + }), + ]; + + expect( + selectStaleWorktreeReapCandidates(rows, ["/repo"], NOW, { + archivedAgeMs: 20 * 60_000, + stoppedAgeMs: 60 * 60_000, + }).map((candidate) => candidate.threadId), + ).toEqual(["archived", "deleted"]); + }); +}); + +describe("worktree registration helpers", () => { + it("matches exact normalized paths in git worktree porcelain output", () => { + expect( + isWorktreePathListed( + [ + "worktree /repo", + "HEAD abc", + "branch refs/heads/main", + "", + "worktree /tmp/t3-worktree", + "HEAD def", + "branch refs/heads/feature", + "", + ].join("\n"), + "/tmp/t3-worktree", + ), + ).toBe(true); + + expect(isWorktreePathListed("worktree /tmp/t3-worktree-child\n", "/tmp/t3-worktree")).toBe( + false, + ); + }); + + it("only clears metadata after a failed worktree listing when both paths are gone", () => { + expect( + shouldRetainWorktreeMetadataAfterListFailure({ + projectRootExists: false, + worktreePathExists: false, + }), + ).toBe(false); + expect( + shouldRetainWorktreeMetadataAfterListFailure({ + projectRootExists: true, + worktreePathExists: false, + }), + ).toBe(true); + expect( + shouldRetainWorktreeMetadataAfterListFailure({ + projectRootExists: false, + worktreePathExists: true, + }), + ).toBe(true); + expect( + shouldRetainWorktreeMetadataAfterListFailure({ + projectRootExists: true, + worktreePathExists: false, + detail: "fatal: not a git repository", + }), + ).toBe(false); + expect( + shouldRetainWorktreeMetadataAfterListFailure({ + projectRootExists: true, + worktreePathExists: false, + detail: "permission denied", + }), + ).toBe(true); + }); +}); diff --git a/apps/server/src/vcs/VcsMaintenanceReactor.ts b/apps/server/src/vcs/VcsMaintenanceReactor.ts new file mode 100644 index 00000000000..365ca5b43c7 --- /dev/null +++ b/apps/server/src/vcs/VcsMaintenanceReactor.ts @@ -0,0 +1,581 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodePath from "node:path"; + +import { CommandId, ThreadId } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as PlatformError from "effect/PlatformError"; +import * as Result from "effect/Result"; +import * as Schedule from "effect/Schedule"; +import type * as Scope from "effect/Scope"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; +import * as GitWorkflowService from "../git/GitWorkflowService.ts"; +import { MAX_THREAD_CHECKPOINTS } from "../orchestration/checkpointRetention.ts"; +import { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts"; +import * as GitVcsDriver from "./GitVcsDriver.ts"; + +// The projection exposes the newest positive-turn checkpoints. Pruning also +// preserves turn 0 as the thread baseline and one predecessor for boundary +// per-turn diffs when those refs exist. +export const CHECKPOINT_REFS_KEEP_PER_THREAD = MAX_THREAD_CHECKPOINTS + 2; +const MAINTENANCE_BOOT_DELAY = Duration.seconds(10); +const MAINTENANCE_SWEEP_INTERVAL = Duration.minutes(30); +const STOPPED_WORKTREE_REAP_AGE_MS = Duration.toMillis(Duration.hours(12)); +const ARCHIVED_WORKTREE_REAP_AGE_MS = Duration.toMillis(Duration.hours(1)); + +const REAPABLE_SESSION_STATUSES = new Set(["stopped", "error"]); + +export interface WorktreeMaintenanceRow { + readonly threadId: string; + readonly projectCwd: string; + readonly worktreePath: string | null; + readonly worktreeRemovable: boolean; + readonly worktreeRemovalPath: string | null; + readonly updatedAt: string; + readonly archivedAt: string | null; + readonly deletedAt: string | null; + readonly sessionStatus: string | null; + readonly runtimeStatus: string | null; + readonly pendingApprovalCount: number; + readonly pendingUserInputCount: number; +} + +export interface WorktreeReapCandidate { + readonly threadId: string; + readonly threadIds: ReadonlyArray; + readonly projectCwd: string; + readonly path: string; +} + +export interface WorktreeReapOptions { + readonly stoppedAgeMs?: number; + readonly archivedAgeMs?: number; +} + +function normalizePath(value: string | null | undefined): string | null { + const trimmed = value?.trim(); + if (!trimmed || !NodePath.isAbsolute(trimmed)) { + return null; + } + const normalized = NodePath.resolve(trimmed); + return normalized === NodePath.parse(normalized).root ? null : normalized; +} + +function isSameOrNestedPath(candidate: string | null, root: string): boolean { + if (!candidate) { + return false; + } + const normalizedCandidate = candidate.replace(/\\/g, "/").replace(/\/+$/, ""); + const normalizedRoot = root.replace(/\\/g, "/").replace(/\/+$/, ""); + return ( + normalizedCandidate === normalizedRoot || + normalizedCandidate.startsWith( + normalizedRoot.endsWith("/") ? normalizedRoot : `${normalizedRoot}/`, + ) + ); +} + +function pathsOverlap(left: string | null, right: string): boolean { + return left !== null && (isSameOrNestedPath(left, right) || isSameOrNestedPath(right, left)); +} + +function rowRemovalPath(row: WorktreeMaintenanceRow): string | null { + return normalizePath(row.worktreeRemovalPath ?? row.worktreePath); +} + +function rowWorktreePath(row: WorktreeMaintenanceRow): string | null { + return normalizePath(row.worktreePath); +} + +function isRowActive(row: WorktreeMaintenanceRow): boolean { + if (row.deletedAt !== null) { + return false; + } + + const hasLiveSessionStatus = + (row.sessionStatus !== null && !REAPABLE_SESSION_STATUSES.has(row.sessionStatus)) || + (row.runtimeStatus !== null && !REAPABLE_SESSION_STATUSES.has(row.runtimeStatus)); + + if (row.archivedAt !== null) { + return hasLiveSessionStatus; + } + + return hasLiveSessionStatus || row.pendingApprovalCount > 0 || row.pendingUserInputCount > 0; +} + +function isRowOldEnough( + row: WorktreeMaintenanceRow, + nowMs: number, + options: Required, +): boolean { + const updatedAtMs = Date.parse(row.updatedAt); + if (!Number.isFinite(updatedAtMs)) { + return false; + } + const ageMs = nowMs - updatedAtMs; + if (ageMs < 0) { + return false; + } + if (row.deletedAt !== null || row.archivedAt !== null) { + return ageMs >= options.archivedAgeMs; + } + return ageMs >= options.stoppedAgeMs; +} + +function isProtectedProjectPath(path: string, projectRoots: ReadonlyArray): boolean { + return ( + projectRoots.some((projectRoot) => isSameOrNestedPath(projectRoot, path)) || + projectRoots.some((projectRoot) => isSameOrNestedPath(path, projectRoot)) + ); +} + +function reapEligibility( + row: WorktreeMaintenanceRow, + normalizedProjectRoots: ReadonlyArray, + nowMs: number, + options: Required, +): { readonly path: string; readonly key: string } | null { + const removalPath = rowRemovalPath(row); + if ( + !removalPath || + !row.worktreeRemovable || + isRowActive(row) || + !isRowOldEnough(row, nowMs, options) || + isProtectedProjectPath(removalPath, normalizedProjectRoots) + ) { + return null; + } + + return { + path: removalPath, + key: `${row.projectCwd}\0${removalPath}`, + }; +} + +export function selectStaleWorktreeReapCandidates( + rows: ReadonlyArray, + projectRoots: ReadonlyArray, + nowMs: number, + options: WorktreeReapOptions = {}, +): ReadonlyArray { + const resolvedOptions = { + stoppedAgeMs: options.stoppedAgeMs ?? STOPPED_WORKTREE_REAP_AGE_MS, + archivedAgeMs: options.archivedAgeMs ?? ARCHIVED_WORKTREE_REAP_AGE_MS, + }; + const normalizedProjectRoots = projectRoots.flatMap((projectRoot) => { + const normalized = normalizePath(projectRoot); + return normalized ? [normalized] : []; + }); + const candidates: WorktreeReapCandidate[] = []; + const selectedPaths = new Set(); + const eligibilityByThreadId = new Map(); + + for (const row of rows) { + const eligibility = reapEligibility(row, normalizedProjectRoots, nowMs, resolvedOptions); + if (eligibility) { + eligibilityByThreadId.set(row.threadId, eligibility); + } + } + + for (const row of rows) { + const eligibility = eligibilityByThreadId.get(row.threadId); + if (!eligibility || selectedPaths.has(eligibility.key)) { + continue; + } + + const sharedByRetainedThread = rows.some((other) => { + if (other.threadId === row.threadId || other.deletedAt !== null) { + return false; + } + if ( + !pathsOverlap(rowWorktreePath(other), eligibility.path) && + !pathsOverlap(rowRemovalPath(other), eligibility.path) + ) { + return false; + } + + const otherEligibility = eligibilityByThreadId.get(other.threadId); + return otherEligibility?.key !== eligibility.key; + }); + if (sharedByRetainedThread) { + continue; + } + + const threadIds = Array.from( + new Set( + rows.flatMap((other) => { + const otherEligibility = eligibilityByThreadId.get(other.threadId); + return otherEligibility?.key === eligibility.key ? [other.threadId] : []; + }), + ), + ); + + selectedPaths.add(eligibility.key); + candidates.push({ + threadId: row.threadId, + threadIds, + projectCwd: row.projectCwd, + path: eligibility.path, + }); + } + + return candidates; +} + +export function isWorktreePathListed(porcelainOutput: string, worktreePath: string): boolean { + return porcelainOutput.split("\n").some((line) => { + if (!line.startsWith("worktree ")) { + return false; + } + return normalizePath(line.slice("worktree ".length)) === worktreePath; + }); +} + +export function shouldRetainWorktreeMetadataAfterListFailure(input: { + readonly projectRootExists: boolean; + readonly worktreePathExists: boolean; + readonly detail?: string; +}): boolean { + if (input.worktreePathExists) { + return true; + } + if (!input.projectRootExists) { + return false; + } + return !isNonRepositoryGitDetail(input.detail ?? ""); +} + +function isNonRepositoryGitDetail(detail: string): boolean { + return detail.toLowerCase().includes("not a git repository"); +} + +export interface VcsMaintenanceReactorShape { + readonly start: () => Effect.Effect; + readonly sweep: () => Effect.Effect; +} + +export class VcsMaintenanceReactor extends Context.Service< + VcsMaintenanceReactor, + VcsMaintenanceReactorShape +>()("t3/vcs/VcsMaintenanceReactor") {} + +type ProjectRootRow = { + readonly workspaceRoot: string; +}; + +type WorktreeRow = { + readonly threadId: string; + readonly projectCwd: string; + readonly worktreePath: string | null; + readonly worktreeRemovable: number; + readonly worktreeRemovalPath: string | null; + readonly updatedAt: string; + readonly archivedAt: string | null; + readonly deletedAt: string | null; + readonly sessionStatus: string | null; + readonly runtimeStatus: string | null; + readonly pendingApprovalCount: number; + readonly pendingUserInputCount: number; +}; + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const checkpointStore = yield* CheckpointStore.CheckpointStore; + const gitWorkflow = yield* GitWorkflowService.GitWorkflowService; + const git = yield* GitVcsDriver.GitVcsDriver; + const orchestrationEngine = yield* OrchestrationEngineService; + const crypto = yield* Crypto.Crypto; + const fileSystem = yield* FileSystem.FileSystem; + + const serverCommandId = (tag: string) => + crypto.randomUUIDv4.pipe(Effect.map((uuid) => CommandId.make(`server:${tag}:${uuid}`))); + + const errorDetail = (error: unknown): string => + error instanceof Error ? error.message : String(error); + + const listProjectRoots = Effect.fn("VcsMaintenanceReactor.listProjectRoots")(function* () { + const rows = yield* sql` + SELECT workspace_root AS "workspaceRoot" + FROM projection_projects + `; + return rows.map((row) => row.workspaceRoot); + }); + + const listActiveProjectRoots = Effect.fn("VcsMaintenanceReactor.listActiveProjectRoots")( + function* () { + const rows = yield* sql` + SELECT workspace_root AS "workspaceRoot" + FROM projection_projects + WHERE deleted_at IS NULL + `; + return rows.map((row) => row.workspaceRoot); + }, + ); + + const listWorktreeRows = Effect.fn("VcsMaintenanceReactor.listWorktreeRows")(function* () { + const rows = yield* sql` + SELECT + t.thread_id AS "threadId", + p.workspace_root AS "projectCwd", + t.worktree_path AS "worktreePath", + t.worktree_removable AS "worktreeRemovable", + t.worktree_removal_path AS "worktreeRemovalPath", + t.updated_at AS "updatedAt", + t.archived_at AS "archivedAt", + t.deleted_at AS "deletedAt", + s.status AS "sessionStatus", + r.status AS "runtimeStatus", + t.pending_approval_count AS "pendingApprovalCount", + t.pending_user_input_count AS "pendingUserInputCount" + FROM projection_threads t + INNER JOIN projection_projects p ON p.project_id = t.project_id + LEFT JOIN projection_thread_sessions s ON s.thread_id = t.thread_id + LEFT JOIN provider_session_runtime r ON r.thread_id = t.thread_id + WHERE t.worktree_path IS NOT NULL + `; + return rows.map( + (row): WorktreeMaintenanceRow => ({ + ...row, + worktreeRemovable: row.worktreeRemovable > 0, + }), + ); + }); + + const isWorktreeRegistered = Effect.fn("VcsMaintenanceReactor.isWorktreeRegistered")(function* ( + candidate: WorktreeReapCandidate, + ) { + const pathExists = (path: string) => + fileSystem.stat(path).pipe( + Effect.as(true), + Effect.catchTags({ + PlatformError: (error: PlatformError.PlatformError) => + Effect.succeed(error.reason._tag !== "NotFound"), + }), + ); + + const failClosedUnlessBothPathsAreGone = Effect.fn( + "VcsMaintenanceReactor.failClosedUnlessBothPathsAreGone", + )(function* (detail: string) { + const [projectRootExists, worktreePathExists] = yield* Effect.all([ + pathExists(candidate.projectCwd), + pathExists(candidate.path), + ]); + if (!projectRootExists && !worktreePathExists) { + yield* Effect.logWarning("vcs.maintenance.worktree-list-root-missing", { + threadIds: candidate.threadIds, + projectRoot: candidate.projectCwd, + path: candidate.path, + detail, + }); + } + return shouldRetainWorktreeMetadataAfterListFailure({ + projectRootExists, + worktreePathExists, + detail, + }); + }); + + const result = yield* git + .execute({ + operation: "VcsMaintenanceReactor.isWorktreeRegistered", + cwd: candidate.projectCwd, + args: ["worktree", "list", "--porcelain"], + allowNonZeroExit: true, + timeoutMs: 5_000, + }) + .pipe(Effect.result); + + if (Result.isFailure(result)) { + yield* Effect.logWarning("vcs.maintenance.worktree-list-failed", { + threadIds: candidate.threadIds, + projectRoot: candidate.projectCwd, + path: candidate.path, + detail: errorDetail(result.failure), + }); + return yield* failClosedUnlessBothPathsAreGone(errorDetail(result.failure)); + } + + if (result.success.exitCode !== 0) { + const detail = result.success.stderr.trim() || "git worktree list failed"; + yield* Effect.logWarning("vcs.maintenance.worktree-list-failed", { + threadIds: candidate.threadIds, + projectRoot: candidate.projectCwd, + path: candidate.path, + detail, + }); + return yield* failClosedUnlessBothPathsAreGone(detail); + } + + return isWorktreePathListed(result.success.stdout, candidate.path); + }); + + const pruneProjectRepository = Effect.fn("VcsMaintenanceReactor.pruneProjectRepository")( + function* (projectRoot: string) { + const isGitRepository = yield* checkpointStore + .isGitRepository(projectRoot) + .pipe(Effect.orElseSucceed(() => false)); + if (!isGitRepository) { + return; + } + + yield* checkpointStore + .pruneCheckpointRefs({ + cwd: projectRoot, + keepPerThread: CHECKPOINT_REFS_KEEP_PER_THREAD, + }) + .pipe( + Effect.tap((checkpointResult) => + checkpointResult.deletedCount > 0 + ? Effect.logInfo("vcs.maintenance.checkpoint-refs-pruned", { + projectRoot, + ...checkpointResult, + }) + : Effect.void, + ), + Effect.catch((error) => + Effect.logWarning("vcs.maintenance.checkpoint-refs-prune-failed", { + projectRoot, + detail: errorDetail(error), + }), + ), + ); + + yield* git.pruneWorktrees(projectRoot).pipe( + Effect.catch((error) => + Effect.logWarning("vcs.maintenance.worktree-prune-failed", { + projectRoot, + detail: errorDetail(error), + }), + ), + ); + }, + ); + + const reapWorktrees = Effect.fn("VcsMaintenanceReactor.reapWorktrees")(function* ( + projectRoots: ReadonlyArray, + ) { + const rows = yield* listWorktreeRows(); + const now = yield* DateTime.now; + const candidates = selectStaleWorktreeReapCandidates( + rows, + projectRoots, + DateTime.toEpochMillis(now), + ); + + const clearCandidateMetadata = Effect.fn("VcsMaintenanceReactor.clearCandidateMetadata")( + function* (candidate: WorktreeReapCandidate) { + for (const threadId of candidate.threadIds) { + const commandId = yield* serverCommandId("vcs-maintenance-worktree-reaped").pipe( + Effect.orDie, + ); + yield* orchestrationEngine + .dispatch({ + type: "thread.meta.update", + commandId, + threadId: ThreadId.make(threadId), + worktreePath: null, + worktreeRemovable: false, + worktreeRemovalPath: null, + }) + .pipe( + Effect.tap(() => + Effect.logInfo("vcs.maintenance.worktree-reaped", { + threadId, + projectRoot: candidate.projectCwd, + path: candidate.path, + }), + ), + Effect.catchCause((cause) => + Effect.logWarning("vcs.maintenance.worktree-metadata-clear-failed", { + threadId, + projectRoot: candidate.projectCwd, + path: candidate.path, + cause: Cause.pretty(cause), + }), + ), + ); + } + }, + ); + + for (const candidate of candidates) { + if (yield* isWorktreeRegistered(candidate)) { + const removeResult = yield* gitWorkflow + .removeWorktree({ + cwd: candidate.projectCwd, + path: candidate.path, + }) + .pipe(Effect.result); + if (Result.isFailure(removeResult)) { + yield* Effect.logWarning("vcs.maintenance.worktree-remove-failed", { + threadIds: candidate.threadIds, + projectRoot: candidate.projectCwd, + path: candidate.path, + detail: errorDetail(removeResult.failure), + }); + yield* git.pruneWorktrees(candidate.projectCwd).pipe( + Effect.catch((error) => + Effect.logWarning("vcs.maintenance.worktree-prune-failed", { + projectRoot: candidate.projectCwd, + detail: errorDetail(error), + }), + ), + ); + if (yield* isWorktreeRegistered(candidate)) { + continue; + } + } + } + yield* clearCandidateMetadata(candidate); + } + }); + + const runSweep = Effect.fn("VcsMaintenanceReactor.runSweep")(function* () { + const protectionProjectRoots = yield* listProjectRoots(); + yield* reapWorktrees(protectionProjectRoots); + const activeProjectRoots = yield* listActiveProjectRoots(); + for (const projectRoot of activeProjectRoots) { + yield* pruneProjectRepository(projectRoot); + } + }); + + const sweep: VcsMaintenanceReactorShape["sweep"] = () => + runSweep().pipe( + Effect.catchCause((cause) => + Effect.logWarning("vcs.maintenance.sweep-failed", { + cause: Cause.pretty(cause), + }), + ), + ); + + const start: VcsMaintenanceReactorShape["start"] = () => + Effect.gen(function* () { + yield* Effect.forkScoped( + Effect.sleep(MAINTENANCE_BOOT_DELAY).pipe( + Effect.andThen(sweep()), + Effect.repeat(Schedule.spaced(MAINTENANCE_SWEEP_INTERVAL)), + ), + ); + yield* Effect.logInfo("vcs.maintenance.reactor.started", { + checkpointRefsKeepPerThread: CHECKPOINT_REFS_KEEP_PER_THREAD, + sweepIntervalMs: Duration.toMillis(MAINTENANCE_SWEEP_INTERVAL), + stoppedWorktreeReapAgeMs: STOPPED_WORKTREE_REAP_AGE_MS, + archivedWorktreeReapAgeMs: ARCHIVED_WORKTREE_REAP_AGE_MS, + }); + }); + + return { + start, + sweep, + } satisfies VcsMaintenanceReactorShape; +}); + +export const layer = Layer.effect(VcsMaintenanceReactor, make);