From f9bd0438e1b17c4d4f0a604da8ace949dfcc37cd Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Wed, 12 Aug 2026 10:21:00 -0700 Subject: [PATCH] fix(server): stop terminal workspace reaper starving on oldest candidates (#11238) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The server runs a scheduled reaper that archives terminal workspaces after it checks their state. > - The reaper reads candidates in `updatedAt` order and skips candidates that do not qualify for archive. > - The fixed page kept the same skipped candidates at the front, so the reaper did not inspect later eligible workspaces. > - This pull request adds a keyset cursor and a throttled log for sweeps that archive no workspace. > - The benefit is that the reaper inspects all candidates over time and reports an inert sweep. ## Linked Issues or Issue Description **What happened?** The terminal workspace reaper inspected a fixed page of old candidates. Ineligible candidates stayed in that page, so the reaper skipped later eligible workspaces on every sweep. **Expected behavior** The reaper must inspect each candidate over time and archive every eligible terminal workspace. **Steps to reproduce** 1. Create more than 50 terminal workspace candidates. 2. Keep the oldest page ineligible for archive. 3. Place an eligible workspace after that page. 4. Run repeated reaper sweeps. 5. Observe that the later eligible workspace remains unarchived. **Paperclip version or commit** Commit `3efdf555e6e14a46747c796c3c554438bfc03261`. **Deployment mode** Built from source with the server test suite. **Installation method** Built from source. **Agent adapter(s) involved** Not adapter-specific (core bug). **Database mode** Not database-related. ## What Changed - Add a keyset cursor that uses `(updatedAt, id)` order across reaper pages. - Reset the cursor at the end of the candidate set so the next sweep starts at the beginning. - Add a throttled log when a sweep inspects candidates but archives none. - Add regression tests for archive delivery and starvation. ## Verification - `pnpm --filter @paperclipai/server exec vitest run src/__tests__/execution-workspaces-service.test.ts` — 45 tests pass. - `pnpm --filter @paperclipai/server exec vitest run src/__tests__/server-startup-feedback-export.test.ts` — 16 tests pass. - `pnpm --filter @paperclipai/server typecheck` — clean. ## Risks Low risk. The change affects only candidate paging and the related reaper log. The cursor resets after the candidate set, so the sweep remains periodic. ## Model Used Codex, OpenAI GPT-5, extended reasoning, tool use, and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above (no exact duplicate found; related scheduler PR #10911 is distinct) - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes (no documentation applies; this is an internal reaper behavior change) - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../execution-workspaces-service.test.ts | 309 ++++++++++++++++++ server/src/index.ts | 16 + server/src/services/execution-workspaces.ts | 93 +++++- 3 files changed, 412 insertions(+), 6 deletions(-) diff --git a/server/src/__tests__/execution-workspaces-service.test.ts b/server/src/__tests__/execution-workspaces-service.test.ts index b2765817c9..1bb8db49f3 100644 --- a/server/src/__tests__/execution-workspaces-service.test.ts +++ b/server/src/__tests__/execution-workspaces-service.test.ts @@ -466,6 +466,315 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { expect(archived?.cleanupEligibleAt).toBeInstanceOf(Date); }, 20_000); + async function seedAncestryTerminalWorkspace(overrides: { updatedAt?: Date } = {}) { + // Build a worktree whose HEAD equals the base ref, so HEAD is an ancestor + // of the base. This workspace landed by ancestry and carries no tracked + // pull request, so delivery derives to merged_by_ancestry. + const repoRoot = await createTempRepo(); + tempDirs.add(repoRoot); + const worktreePath = path.join(path.dirname(repoRoot), `paperclip-ancestry-${randomUUID()}`); + tempDirs.add(worktreePath); + const branchName = `ancestry-${randomUUID().slice(0, 8)}`; + await runGit(repoRoot, ["branch", branchName]); + await runGit(repoRoot, ["worktree", "add", worktreePath, branchName]); + + const companyId = randomUUID(); + const projectId = randomUUID(); + const executionWorkspaceId = randomUUID(); + const sourceIssueId = randomUUID(); + const issuePrefix = `P${companyId.slice(0, 8).toUpperCase()}`; + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Ancestry delivery", + status: "in_progress", + }); + await db.insert(executionWorkspaces).values({ + id: executionWorkspaceId, + companyId, + projectId, + mode: "isolated_workspace", + strategyType: "git_worktree", + name: `${issuePrefix}-1`, + status: "active", + cwd: worktreePath, + providerRef: worktreePath, + providerType: "git_worktree", + repoUrl: "https://github.com/paperclipai/paperclip.git", + baseRef: "main", + branchName, + }); + await db.insert(issues).values({ + id: sourceIssueId, + companyId, + projectId, + identifier: `${issuePrefix}-1`, + title: "Delivered by ancestry", + status: "done", + priority: "medium", + executionWorkspaceId, + }); + await db + .update(executionWorkspaces) + .set({ sourceIssueId, ...(overrides.updatedAt ? { updatedAt: overrides.updatedAt } : {}) }) + .where(eq(executionWorkspaces.id, executionWorkspaceId)); + return { companyId, projectId, executionWorkspaceId, sourceIssueId, worktreePath }; + } + + it("archives a terminal workspace delivered by ancestry with no pull request", async () => { + const seeded = await seedAncestryTerminalWorkspace(); + + const readiness = await svc.getCloseReadiness(seeded.executionWorkspaceId); + expect(readiness?.deliveryState).toBe("merged_by_ancestry"); + expect(readiness?.blockingReasons).toEqual([]); + + const sweep = await svc.sweepTerminalWorkspaces(); + const [workspace] = await db + .select({ status: executionWorkspaces.status, cleanupReason: executionWorkspaces.cleanupReason }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + + expect(sweep).toMatchObject({ archived: 1, cleanupFailed: 0 }); + expect(workspace).toMatchObject({ status: "archived", cleanupReason: "issue_terminal" }); + }, 20_000); + + it("skips a sweep that starts while another sweep runs", async () => { + // The scheduler can start a second sweep before the first one finishes. The + // sweeps share the cursor and the boundary. A concurrent sweep must skip + // instead of running, so it cannot corrupt the shared rotation state. + const seeded = await seedAncestryTerminalWorkspace(); + + // Start the first sweep and do not wait. An async function runs its body up + // to the first await, so the in-progress flag is set before the second call + // starts. The second call sees the flag and returns without a scan. + const firstSweepPromise = svc.sweepTerminalWorkspaces(); + const concurrentSweep = await svc.sweepTerminalWorkspaces(); + const firstSweep = await firstSweepPromise; + + // The concurrent sweep inspected no candidate and changed no state. + expect(concurrentSweep).toMatchObject({ checked: 0, archived: 0, eligible: 0 }); + // The first sweep archived the eligible workspace. + expect(firstSweep.archived).toBe(1); + + const [workspace] = await db + .select({ status: executionWorkspaces.status }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, seeded.executionWorkspaceId)); + expect(workspace?.status).toBe("archived"); + + // The flag resets after the first sweep, so a later sweep runs its scan. + const laterSweep = await svc.sweepTerminalWorkspaces(); + expect(laterSweep).toMatchObject({ archived: 0 }); + }, 20_000); + + it("archives an eligible workspace behind a full page of skipped candidates", async () => { + // Seed more skipped candidates than the sweep page holds, each older than + // the eligible workspace. A skipped candidate keeps its updatedAt, so a + // sweep that always reads the oldest page never reaches the eligible + // workspace. The reaper must rotate its scan window across sweeps. + const companyId = randomUUID(); + const projectId = randomUUID(); + const issuePrefix = `P${companyId.slice(0, 8).toUpperCase()}`; + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Starved reaper", + status: "in_progress", + }); + // Two open-issue workspaces that the reaper always skips. Their older + // updatedAt keeps them at the front of the ordered candidate set. + const skippedWorkspaceIds: string[] = []; + for (let index = 0; index < 2; index += 1) { + const workspaceId = randomUUID(); + const openIssueId = randomUUID(); + skippedWorkspaceIds.push(workspaceId); + await db.insert(executionWorkspaces).values({ + id: workspaceId, + companyId, + projectId, + mode: "isolated_workspace", + strategyType: "local_fs", + name: `${issuePrefix}-skip-${index}`, + status: "active", + providerType: "local_fs", + updatedAt: new Date(Date.UTC(2020, 0, index + 1)), + }); + await db.insert(issues).values({ + id: openIssueId, + companyId, + projectId, + title: `Open ${index}`, + status: "in_progress", + priority: "medium", + executionWorkspaceId: workspaceId, + }); + await db + .update(executionWorkspaces) + .set({ sourceIssueId: openIssueId, updatedAt: new Date(Date.UTC(2020, 0, index + 1)) }) + .where(eq(executionWorkspaces.id, workspaceId)); + } + // The eligible workspace is newest, so it sorts after the whole skipped page. + const eligible = await seedAncestryTerminalWorkspace({ updatedAt: new Date(Date.UTC(2020, 0, 9)) }); + + // A fresh service starts with an empty scan cursor, so each call inspects + // one row and advances. A single-row page never lands on the eligible + // workspace first. + const service = executionWorkspaceService(db, { + resolvePullRequestDetails: async () => ({ state: "unknown", headRef: null, headSha: null }), + }); + + const firstSweep = await service.sweepTerminalWorkspaces(1); + expect(firstSweep).toMatchObject({ checked: 1, archived: 0, skippedNonTerminalTree: 1 }); + const [afterFirst] = await db + .select({ status: executionWorkspaces.status }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, eligible.executionWorkspaceId)); + expect(afterFirst?.status).toBe("active"); + + let archivedSweep: Awaited> | null = null; + for (let attempt = 0; attempt < 4 && !archivedSweep; attempt += 1) { + const sweep = await service.sweepTerminalWorkspaces(1); + if (sweep.archived > 0) archivedSweep = sweep; + } + + expect(archivedSweep).toMatchObject({ archived: 1 }); + const workspaces = await db + .select({ id: executionWorkspaces.id, status: executionWorkspaces.status }) + .from(executionWorkspaces) + .where(inArray(executionWorkspaces.id, [eligible.executionWorkspaceId, ...skippedWorkspaceIds])); + const byId = new Map(workspaces.map((row) => [row.id, row.status])); + expect(byId.get(eligible.executionWorkspaceId)).toBe("archived"); + for (const skippedId of skippedWorkspaceIds) { + expect(byId.get(skippedId)).toBe("active"); + } + }, 20_000); + + it("revisits an eligible workspace behind the cursor despite continuous newer churn", async () => { + // Reproduce the rotation-starvation case. The scan cursor advances past a + // workspace while it is not eligible. The workspace then becomes eligible + // but keeps its old updatedAt, so it stays behind the cursor. Meanwhile a + // steady stream of newer candidates keeps every page full. Without a frozen + // per-rotation upper bound, the cursor never reaches the end, never resets, + // and never revisits the eligible workspace. The bound makes each rotation + // cover a finite set, so the cursor resets and the workspace is archived. + let clockMs = Date.UTC(2021, 6, 1); + const service = executionWorkspaceService(db, { + resolvePullRequestDetails: async () => ({ state: "unknown", headRef: null, headSha: null }), + now: () => new Date(clockMs), + }); + + // An eligible ancestry workspace with an old updatedAt. Its source issue + // starts non-terminal, so the first sweeps skip it and pass the cursor. + const eligible = await seedAncestryTerminalWorkspace({ updatedAt: new Date(Date.UTC(2021, 0, 2)) }); + await db + .update(issues) + .set({ status: "in_progress" }) + .where(eq(issues.id, eligible.sourceIssueId)); + + // One older skipped candidate. With a single-row page it sorts before the + // eligible workspace, so the first sweep advances the cursor onto it. + const olderSkippedId = randomUUID(); + const olderOpenIssueId = randomUUID(); + await db.insert(executionWorkspaces).values({ + id: olderSkippedId, + companyId: eligible.companyId, + projectId: eligible.projectId, + mode: "isolated_workspace", + strategyType: "local_fs", + name: "skip-older", + status: "active", + providerType: "local_fs", + updatedAt: new Date(Date.UTC(2021, 0, 1)), + }); + await db.insert(issues).values({ + id: olderOpenIssueId, + companyId: eligible.companyId, + projectId: eligible.projectId, + title: "Older open", + status: "in_progress", + priority: "medium", + executionWorkspaceId: olderSkippedId, + }); + await db + .update(executionWorkspaces) + .set({ sourceIssueId: olderOpenIssueId, updatedAt: new Date(Date.UTC(2021, 0, 1)) }) + .where(eq(executionWorkspaces.id, olderSkippedId)); + + // Sweep once per row so the cursor lands on the eligible workspace while it + // is still non-terminal. + await service.sweepTerminalWorkspaces(1); // reads olderSkipped + await service.sweepTerminalWorkspaces(1); // reads eligible, still non-terminal + + const [afterSkip] = await db + .select({ status: executionWorkspaces.status }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, eligible.executionWorkspaceId)); + expect(afterSkip?.status).toBe("active"); + + // The workspace becomes eligible now. Its updatedAt stays old, so it is + // behind the cursor. + await db + .update(issues) + .set({ status: "done" }) + .where(eq(issues.id, eligible.sourceIssueId)); + + // Drive continuous churn. Each sweep, advance the clock and add a newer + // skipped candidate ahead of the cursor. Without the frozen bound the cursor + // would chase this churn forever and never revisit the eligible workspace. + let archived = false; + for (let attempt = 0; attempt < 8 && !archived; attempt += 1) { + clockMs += 24 * 60 * 60 * 1000; + const churnId = randomUUID(); + const churnIssueId = randomUUID(); + await db.insert(executionWorkspaces).values({ + id: churnId, + companyId: eligible.companyId, + projectId: eligible.projectId, + mode: "isolated_workspace", + strategyType: "local_fs", + name: `churn-${attempt}`, + status: "active", + providerType: "local_fs", + updatedAt: new Date(clockMs), + }); + await db.insert(issues).values({ + id: churnIssueId, + companyId: eligible.companyId, + projectId: eligible.projectId, + title: `Churn ${attempt}`, + status: "in_progress", + priority: "medium", + executionWorkspaceId: churnId, + }); + await db + .update(executionWorkspaces) + .set({ sourceIssueId: churnIssueId, updatedAt: new Date(clockMs) }) + .where(eq(executionWorkspaces.id, churnId)); + + const sweep = await service.sweepTerminalWorkspaces(1); + if (sweep.archived > 0) archived = true; + } + + expect(archived).toBe(true); + const [finalState] = await db + .select({ status: executionWorkspaces.status }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, eligible.executionWorkspaceId)); + expect(finalState?.status).toBe("archived"); + }, 30_000); + it("does not treat an unrelated inbound issue mention as delivery evidence", async () => { const seeded = await seedTerminalWorkspace(); const unrelatedIssueId = randomUUID(); diff --git a/server/src/index.ts b/server/src/index.ts index 0a4bccf87f..b0dc51bf9d 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -995,6 +995,11 @@ export async function startServer(): Promise { logger.error({ err }, "merged pull-request confirmation sweep failed"); })); }; + // Emit a periodic signal when the reaper inspects candidates but archives + // none, so an inert reaper that skips every candidate is never fully silent. + // The throttle keeps the 30s cadence from flooding the log. + let lastTerminalWorkspaceSkipLogAt = 0; + const terminalWorkspaceSkipLogIntervalMs = 10 * 60 * 1000; const scheduleTerminalWorkspaceSweep = () => { if (heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(terminalWorkspaces @@ -1002,6 +1007,17 @@ export async function startServer(): Promise { .then((result) => { if (result.archived > 0 || result.cleanupFailed > 0) { logger.info(result, "terminal issue workspace reaper changed workspace state"); + return; + } + const skipped = + result.skippedActiveRun + + result.skippedNonTerminalTree + + result.skippedUndelivered + + result.skippedRace; + const nowMs = Date.now(); + if (skipped > 0 && nowMs - lastTerminalWorkspaceSkipLogAt >= terminalWorkspaceSkipLogIntervalMs) { + lastTerminalWorkspaceSkipLogAt = nowMs; + logger.info(result, "terminal issue workspace reaper skipped all candidates"); } }) .catch((err) => { diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index e9bb355a0c..37b5d438b4 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -3,7 +3,7 @@ import { createHash } from "node:crypto"; import fs from "node:fs/promises"; import path from "node:path"; import { promisify } from "node:util"; -import { and, asc, desc, eq, inArray, isNull, ne, or, sql } from "drizzle-orm"; +import { and, asc, desc, eq, gt, inArray, isNull, lte, ne, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { executionWorkspaces, @@ -1063,6 +1063,29 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic >(); const pullRequestStateCacheTtlMs = 5 * 60 * 1000; + // The terminal-workspace reaper scans the candidate set in fixed-size pages. + // It keeps this keyset cursor between sweeps so each sweep continues after the + // previous page. A skipped candidate keeps its updatedAt, so a cursor is + // required: without it the query re-selects the oldest ineligible rows every + // sweep and starves eligible rows behind them. The cursor resets to the start + // when a sweep reaches the end of the candidate set. + let terminalSweepCursor: { updatedAt: Date; id: string } | null = null; + // The reaper freezes an upper bound on updatedAt at the start of each + // rotation. The scan only reads candidates at or below the bound, so a steady + // stream of newer candidates cannot keep every page full and stop the cursor + // from ever reaching the end. A frozen set is finite, so the cursor always + // reaches a short page and resets, and older candidates that became eligible + // are revisited on the next rotation. The next rotation captures a new bound, + // so candidates updated after the previous bound enter the scan then. + let terminalSweepBoundary: Date | null = null; + // The scheduler starts a sweep on each tick and does not wait for the previous + // sweep to finish. A sweep that outlasts the tick interval overlaps the next + // sweep. Both sweeps share the cursor and the boundary above. Interleaved + // reads and writes can leave a non-null cursor with a null boundary, which + // removes the upper bound and makes the scan chase newer churn again. This + // flag lets only one sweep run at a time, so one sweep owns the shared state. + let terminalSweepInProgress = false; + async function listWorkspaceIssueTree(workspace: Pick) { if (!workspace.sourceIssueId) return []; return db @@ -2033,16 +2056,71 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic }, sweepTerminalWorkspaces: async (limit = 50) => { + // Skip this sweep while another sweep runs. A concurrent sweep would share + // the cursor and the boundary and could corrupt the rotation state. A + // skipped tick is safe: the next tick runs the sweep with intact state. + if (terminalSweepInProgress) { + return { + checked: 0, + eligible: 0, + archived: 0, + cleanupFailed: 0, + skippedActiveRun: 0, + skippedNonTerminalTree: 0, + skippedUndelivered: 0, + skippedRace: 0, + }; + } + terminalSweepInProgress = true; + try { + const baseCandidateFilter = and( + inArray(executionWorkspaces.status, ["active", "idle", "in_review"]), + isNull(executionWorkspaces.closedAt), + sql`${executionWorkspaces.sourceIssueId} IS NOT NULL`, + ); + // Continue the scan after the previous sweep's last row. The keyset + // predicate uses the same (updatedAt, id) order as the query, so each + // sweep advances past the rows it already inspected instead of re-reading + // the oldest ineligible candidates. + const cursor = terminalSweepCursor; + // Freeze an upper bound on updatedAt at the start of each rotation. Without + // a bound, a steady stream of newer candidates keeps every page full, so + // the cursor never reaches the end and never resets, and older candidates + // that became eligible are never revisited. The frozen bound makes the + // rotation cover a finite set, so the cursor always reaches a short page. + if (!cursor) { + terminalSweepBoundary = now(); + } + const boundary = terminalSweepBoundary; + const boundaryFilter = boundary + ? lte(executionWorkspaces.updatedAt, boundary) + : undefined; + const cursorFilter = cursor + ? or( + gt(executionWorkspaces.updatedAt, cursor.updatedAt), + and( + eq(executionWorkspaces.updatedAt, cursor.updatedAt), + gt(executionWorkspaces.id, cursor.id), + ), + ) + : undefined; + const scanFilter = and(baseCandidateFilter, boundaryFilter, cursorFilter); const candidates = await db .select() .from(executionWorkspaces) - .where(and( - inArray(executionWorkspaces.status, ["active", "idle", "in_review"]), - isNull(executionWorkspaces.closedAt), - sql`${executionWorkspaces.sourceIssueId} IS NOT NULL`, - )) + .where(scanFilter) .orderBy(asc(executionWorkspaces.updatedAt), asc(executionWorkspaces.id)) .limit(limit); + // Advance the cursor to this page's last row. A short page means the scan + // reached the end of the bounded candidate set, so reset the cursor and + // the bound to start a new rotation on the next sweep. + if (candidates.length < limit) { + terminalSweepCursor = null; + terminalSweepBoundary = null; + } else { + const lastCandidate = candidates[candidates.length - 1]!; + terminalSweepCursor = { updatedAt: lastCandidate.updatedAt, id: lastCandidate.id }; + } const result = { checked: candidates.length, eligible: 0, @@ -2178,6 +2256,9 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic } } return result; + } finally { + terminalSweepInProgress = false; + } }, create: async (data: typeof executionWorkspaces.$inferInsert) => {