diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index da60b3bfd61..dc4c1d3f7d9 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2271,11 +2271,39 @@ that flushes after the tab hides marks data stale without starting a refetch. The developer-server health poll also stops in hidden tabs. Workspace detail responses share concurrent Git inspections and reuse their -results for up to five seconds after completion. The cache holds at most 256 +results for up to thirty seconds after completion. Workspace identity and metadata +changes invalidate the cached result. The cache holds at most 256 entries. Close-readiness checks, the terminal-workspace reaper, and the final cleanup validation still inspect Git afresh. A display result never authorizes worktree removal. +The terminal-workspace reaper uses the database work coordinator. Startup scans +recover existing workspace rows in bounded pages. Issue status and tree changes, +workspace lifecycle writes, and pull-request work-product writes signal work on +their transaction before writing. The coordinator wakes after the outer +transaction settles and holds idle admission while writes or cleanup are active. +Empty and nonterminal-only lanes have no recurring timer or query. Terminal trees +schedule their actual cooldown deadline; unresolved delivery or cleanup eligibility +is checked again after five minutes. Nonterminal trees, cooldowns, and active runs +are checked before Git. Unchanged terminal candidates also back off for five +minutes during unrelated commit wakes. Scoped task/workspace hints invalidate +affected candidates; changed tree or workspace revisions invalidate them too. +Deadline selection reuses task trees read by those pages and performs no separate +full-inventory query. Commits during a paged rotation request another bounded +rotation so changes behind its cursor are recovered. +The backoff retains at most 256 candidates and only applies to the background +worker. Final cleanup still checks the current task tree, +delivery evidence, ownership, lifecycle generation, and Git state under the +existing cleanup locks. + +These wakes cover the owning `createDb` scope in this server process. Direct SQL +and other server processes require startup recovery. Local deadlines do not +establish external wake-alarm coverage or whole-instance sleep safety. The local +sleep inventory blocks host suspension until bounded recovery succeeds. Cooldown +eligibility is an earliest allowed cleanup time; passive retained directories can +wait for the next wake, when startup restores their rows. Existing idle-sleep +blockers and cleanup admission holds remain authoritative. + The connection-health sweep selects only due IDs in SQL before applying its limit. Legacy `paperclip_plugin` placeholder connections are excluded: their tools run in plugin workers and do not have remote MCP endpoints. These rows diff --git a/doc/idle-sleep-safety.md b/doc/idle-sleep-safety.md index 10e82a56879..ec87d70985e 100644 --- a/doc/idle-sleep-safety.md +++ b/doc/idle-sleep-safety.md @@ -123,6 +123,13 @@ running the timer while asleep. A generation counter invalidates a scan if tracked work both starts and finishes during inspection. Reports remain unknown until startup recovery and HTTP tracking are installed. +Terminal workspace recovery also registers a local sleep probe. Incomplete +startup pages and failed recovery remain blockers until a successful pass. +Active cleanup and unresolved producer transactions retain their admission holds. +Cooldown and inspection-backoff timers are maintenance hints over retained domain +rows. They can wait for the next wake and are restored at startup; they do not +create external alarm coverage or exempt other persisted work from this report. + ### Backup checkpoints for idle instances `PAPERCLIP_DB_BACKUP_IDLE_CHECKPOINT_ENABLED=1` opts an instance with automatic diff --git a/maps/feature-map/workspaces.md b/maps/feature-map/workspaces.md index 8d5ee64f22d..35312f6fdfb 100644 --- a/maps/feature-map/workspaces.md +++ b/maps/feature-map/workspaces.md @@ -107,6 +107,14 @@ Manual: Provision a disposable workspace, wait for readiness, start an available - A link or running badge does not prove the service is reachable from the user’s browser. - Workspace IDs, runtime leases, and repository paths are not interchangeable. +- Terminal workspace cleanup wakes after committed task/workspace/PR changes and + restores cooldown deadlines at startup. Empty lanes stop scanning. Display Git + results can lag by thirty seconds; destructive cleanup still reads fresh Git. + The [worker tests](../../server/src/__tests__/terminal-workspace-work.test.ts) + cover quiet lanes and deadlines. The + [service tests](../../server/src/__tests__/execution-workspaces-service.test.ts) + cover real PostgreSQL signals and retained worktrees. These tests do not + qualify an external wake alarm or a live provider deployment. - Organizational project, source project, task files, personal `AGENT_HOME`, and provider home have separate meanings. Private source ownership remains attached to copied files after source configuration changes or workspace closure. diff --git a/server/src/__tests__/execution-workspace-reopen.test.ts b/server/src/__tests__/execution-workspace-reopen.test.ts index 98d0dee41b4..0e6daad0bc8 100644 --- a/server/src/__tests__/execution-workspace-reopen.test.ts +++ b/server/src/__tests__/execution-workspace-reopen.test.ts @@ -1,4 +1,5 @@ -import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, subscribeDatabaseWork } from "@paperclipai/db"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { execFileSync } from "node:child_process"; import { mkdirSync, writeFileSync } from "node:fs"; import { mkdtemp, rm, stat, readFile, writeFile } from "node:fs/promises"; @@ -345,13 +346,16 @@ describeEmbeddedPostgres("reopen archived isolated execution workspace", () => { const issueId = await seedIssue({ companyId, projectId, workspaceId, issueNumber: 4100 }); const svc = executionWorkspaceService(db); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.terminalWorkspaces, event => events.push(event)); const result = await svc.reopenClosedIsolatedExecutionWorkspaceForIssue({ workspaceId, issue: { id: issueId, companyId, projectId }, actor: { agentId: null, actorType: "user" }, authorizationActor: { type: "board", source: "local_implicit", isInstanceAdmin: true }, - }); + }).finally(unsubscribe); + expect(events).toEqual(["intent", "settled"]); expect(result.ok).toBe(true); if (!result.ok) return; expect(result.reopened).toBe(true); diff --git a/server/src/__tests__/execution-workspaces-service.test.ts b/server/src/__tests__/execution-workspaces-service.test.ts index fee0730d7b9..645dcf51077 100644 --- a/server/src/__tests__/execution-workspaces-service.test.ts +++ b/server/src/__tests__/execution-workspaces-service.test.ts @@ -14,6 +14,7 @@ import { agents, companies, createDb, + databaseWorkPending, executionWorkspaces, heartbeatRuns, issueComments, @@ -23,6 +24,7 @@ import { issues, projectWorkspaces, projects, + subscribeDatabaseWork, workspaceRuntimeServices, } from "@paperclipai/db"; import { @@ -41,6 +43,8 @@ import { readMetadataReopenPendingConsumptionSince, } from "../services/execution-workspaces.ts"; import { issueService } from "../services/issues.ts"; +import { workProductService } from "../services/work-products.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { startRuntimeServicesForWorkspaceControl, stopRuntimeServicesForExecutionWorkspace, @@ -519,6 +523,171 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { return { companyId, projectId, executionWorkspaceId, sourceIssueId, identifier, repoRoot, worktreePath, headSha }; } + it("skips nonterminal trees, cooldowns, and active runs without a Git scan", async () => { + await seedTerminalWorkspace({ childStatus: "todo" }); + const recent = await seedTerminalWorkspace({ mergedPr: true }); + const running = await seedTerminalWorkspace({ activeRun: true }); + await db.update(issues).set({ completedAt: new Date("2020-01-01") }).where(eq(issues.id, running.sourceIssueId)); + await db.update(issues).set({ completedAt: new Date() }).where(eq(issues.id, recent.sourceIssueId)); + const service = executionWorkspaceService(db, { workspaceReaperCooldownDays: 7 }); + const scan = vi.spyOn(workspaceGitOperationScheduler, "run"); + try { + expect(await service.sweepTerminalWorkspaces()).toMatchObject({ + archived: 0, skippedNonTerminalTree: 1, skippedCooldown: 1, skippedActiveRun: 1, + }); + expect(scan).not.toHaveBeenCalled(); + } finally { scan.mockRestore(); } + }); + + it("derives cooldown deadlines from the full company-scoped terminal tree", async () => { + const seed = await seedTerminalWorkspace({ childStatus: "done" }); + const [child] = await db.select().from(issues).where(eq(issues.parentId, seed.sourceIssueId)); + const sourceDone = new Date("2026-01-01T00:00:00Z"), childCancelled = new Date("2026-01-02T00:00:00Z"); + await db.update(issues).set({ completedAt: sourceDone }).where(eq(issues.id, seed.sourceIssueId)); + await db.update(issues).set({ status: "cancelled", cancelledAt: childCancelled }).where(eq(issues.id, child!.id)); + const unrelated = await seedTerminalWorkspace(); + await db.update(issues).set({ parentId: seed.sourceIssueId, status: "todo" }).where(eq(issues.id, unrelated.sourceIssueId)); + const service = executionWorkspaceService(db, { + workspaceReaperCooldownDays: 7, now: () => new Date("2026-01-03T00:00:00Z"), + }); + await service.sweepTerminalWorkspaces(); + const select = vi.spyOn(db, "select"); + try { + expect(await service.nextTerminalWorkspaceSweepAt()).toBe(childCancelled.getTime() + 7 * 86_400_000); + expect(await service.nextTerminalWorkspaceSweepAt()).toBe(childCancelled.getTime() + 7 * 86_400_000); + expect(select).not.toHaveBeenCalled(); + } finally { select.mockRestore(); } + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, child!.id)); + await service.sweepTerminalWorkspaces(); + expect(await service.nextTerminalWorkspaceSweepAt()).toBeNull(); + await db.update(issues).set({ status: "done", completedAt: null, updatedAt: sourceDone }).where(eq(issues.id, child!.id)); + await service.sweepTerminalWorkspaces(); + expect(await service.nextTerminalWorkspaceSweepAt()).toBe(sourceDone.getTime() + 7 * 86_400_000); + }); + + it("does not use cached display Git state to remove a worktree", async () => { + const seed = await seedTerminalWorkspace({ mergedPr: true }); + await svc.getById(seed.executionWorkspaceId); + await fs.writeFile(path.join(seed.worktreePath, "unsaved.txt"), "keep this work\n"); + expect(await svc.sweepTerminalWorkspaces()).toMatchObject({ archived: 0, skippedUndelivered: 1 }); + expect(await fs.readFile(path.join(seed.worktreePath, "unsaved.txt"), "utf8")).toBe("keep this work\n"); + }); + + it("backs off unchanged terminal candidates while unrelated commits wake the worker", async () => { + await seedTerminalWorkspace(); + let clock = Date.now(); + const service = executionWorkspaceService(db, { + workspaceReaperCooldownDays: 0, terminalWorkspaceInspectionBackoffMs: 300_000, + now: () => new Date(clock), + }); + const scan = vi.spyOn(workspaceGitOperationScheduler, "run"); + try { + expect(await service.sweepTerminalWorkspaces()).toMatchObject({ archived: 0, skippedUndelivered: 1 }); + const initial = scan.mock.calls.length; + expect(initial).toBeGreaterThan(0); + service.terminalWorkspaceWorkSettled(new Set(["issue:unrelated"])); + expect(await service.sweepTerminalWorkspaces()).toMatchObject({ skippedBackoff: 1 }); + expect(scan).toHaveBeenCalledTimes(initial); + clock += 240_000; + expect(await service.nextTerminalWorkspaceSweepAt()).toBe(clock + 60_000); + clock += 60_000; + await service.sweepTerminalWorkspaces(); + expect(scan).toHaveBeenCalledTimes(initial * 2); + } finally { scan.mockRestore(); } + }); + + it("invalidates a candidate backoff after its real PR producer commits", async () => { + const seed = await seedTerminalWorkspace(); + const service = executionWorkspaceService(db, { + workspaceReaperCooldownDays: 0, terminalWorkspaceInspectionBackoffMs: 300_000, + resolvePullRequestDetails: async () => ({ state: "merged", headRef: "PAP-16015-delivery", headSha: seed.headSha }), + }); + expect(await service.sweepTerminalWorkspaces()).toMatchObject({ archived: 0, skippedUndelivered: 1 }); + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.terminalWorkspaces, (event, hints) => { + if (event === "settled") service.terminalWorkspaceWorkSettled(hints); + }); + try { + await workProductService(db).createForIssue(seed.sourceIssueId, seed.companyId, { + type: "pull_request", provider: "github", title: "PR", status: "merged", + url: "https://github.com/paperclipai/paperclip/pull/10623", executionWorkspaceId: seed.executionWorkspaceId, + }); + expect(await service.sweepTerminalWorkspaces()).toMatchObject({ archived: 1 }); + await expect(fs.access(seed.worktreePath)).rejects.toMatchObject({ code: "ENOENT" }); + } finally { unsubscribe(); } + }); + + it("rescans committed changes that arrive behind a bounded recovery cursor", async () => { + const seed = await seedTerminalWorkspace(); + const service = executionWorkspaceService(db, { + workspaceReaperCooldownDays: 0, + resolvePullRequestDetails: async () => ({ state: "merged", headRef: "PAP-16015-delivery", headSha: seed.headSha }), + }); + expect(await service.sweepTerminalWorkspaces(1)).toMatchObject({ checked: 1, skippedUndelivered: 1 }); + await workProductService(db).createForIssue(seed.sourceIssueId, seed.companyId, { + type: "pull_request", provider: "github", title: "PR", status: "merged", + url: "https://github.com/paperclipai/paperclip/pull/10623", + }); + service.terminalWorkspaceWorkSettled(new Set([`issue:${seed.sourceIssueId}`])); + expect(await service.sweepTerminalWorkspaces(1)).toMatchObject({ checked: 0 }); + expect(service.hasTerminalWorkspaceSweepContinuation()).toBe(true); + expect(await service.sweepTerminalWorkspaces(2)).toMatchObject({ checked: 1, archived: 1 }); + expect(service.hasTerminalWorkspaceSweepContinuation()).toBe(false); + expect(await service.nextTerminalWorkspaceSweepAt()).toBeNull(); + }); + + it("retries unread rows from a failed full page before it can park recovery", async () => { + const seed = await seedTerminalWorkspace({ mergedPr: true }); + await seedTerminalWorkspace({ childStatus: "todo" }); + await db.update(executionWorkspaces).set({ updatedAt: new Date("2019-01-01") }) + .where(eq(executionWorkspaces.id, seed.executionWorkspaceId)); + const service = executionWorkspaceService(db, { + workspaceReaperCooldownDays: 0, + resolvePullRequestDetails: async () => ({ state: "merged", headRef: "PAP-16015-delivery", headSha: seed.headSha }), + }); + const original = db.select.bind(db); + let fail = true; + const select = vi.spyOn(db, "select").mockImplementation(((fields: any) => { + if (fail && fields?.completedAt && fields?.cancelledAt) { + fail = false; + throw new Error("task tree read failed"); + } + return original(fields); + }) as typeof db.select); + try { + await expect(service.sweepTerminalWorkspaces(1)).rejects.toThrow("task tree read failed"); + expect(service.hasTerminalWorkspaceSweepContinuation()).toBe(true); + expect(await service.sweepTerminalWorkspaces(1)).toMatchObject({ checked: 1, archived: 1 }); + expect(await service.sweepTerminalWorkspaces(1)).toMatchObject({ checked: 1, skippedNonTerminalTree: 1 }); + expect(await service.sweepTerminalWorkspaces(1)).toMatchObject({ checked: 0 }); + expect(service.hasTerminalWorkspaceSweepContinuation()).toBe(false); + expect(await service.nextTerminalWorkspaceSweepAt()).toBeNull(); + await expect(fs.access(seed.worktreePath)).rejects.toMatchObject({ code: "ENOENT" }); + } finally { select.mockRestore(); } + }); + + it("signals real issue producers before writes and waits for the outer transaction", async () => { + const seed = await seedTerminalWorkspace(); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.terminalWorkspaces, event => events.push(event)); + try { + await db.transaction(async tx => { + await issueService(db).update(seed.sourceIssueId, { status: "cancelled" }, tx); + expect(events).toEqual(["intent"]); + expect(databaseWorkPending(db, DELIVERY_QUEUES.terminalWorkspaces)).toBe(true); + const [outside] = await db.select().from(issues).where(eq(issues.id, seed.sourceIssueId)); + expect(outside?.status).toBe("done"); + }); + expect(events).toEqual(["intent", "settled"]); + expect(databaseWorkPending(db, DELIVERY_QUEUES.terminalWorkspaces)).toBe(false); + await workProductService(db).createForIssue(seed.sourceIssueId, seed.companyId, { + type: "pull_request", provider: "github", title: "PR", status: "merged", url: "https://github.com/paperclipai/paperclip/pull/10623", + }); + expect(events.slice(-2)).toEqual(["intent", "settled"]); + await svc.update(seed.executionWorkspaceId, { status: "idle" }); + expect(events).toHaveLength(6); + } finally { unsubscribe(); } + }); + it("reports a squash cross-branch delivery as merged_via_pr and suppresses the ancestry warning", async () => { const repoRoot = await createTempRepo(); tempDirs.add(repoRoot); diff --git a/server/src/__tests__/idle-local-work.test.ts b/server/src/__tests__/idle-local-work.test.ts index 9728c97562b..4d8426796f8 100644 --- a/server/src/__tests__/idle-local-work.test.ts +++ b/server/src/__tests__/idle-local-work.test.ts @@ -2,7 +2,7 @@ import { promises as fs } from "node:fs"; import os from "node:os"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { inspectIdleSpool, inspectIdleQuotaRecovery, idleQuotaRecoveryPath, readIdleLocalWork, markIdleIngressTracked, markIdleStartupComplete, registerIdleSpoolDirectory } from "../services/idle-local-work.js"; +import { inspectIdleSpool, inspectIdleQuotaRecovery, idleQuotaRecoveryPath, readIdleLocalWork, markIdleIngressTracked, markIdleStartupComplete, registerIdleSpoolDirectory, registerIdleLocalWorkProbe } from "../services/idle-local-work.js"; let root: string; beforeEach(async () => { root = await fs.mkdtemp(path.join(os.tmpdir(), "idle-spool-")); }); @@ -31,6 +31,22 @@ describe("idle cleanup inspection", () => { vi.spyOn(fs, "opendir").mockRejectedValue(Object.assign(new Error("private path"), { code: "EACCES" })); expect(await inspectIdleSpool(root)).toBe("unknown"); }); + it("retains failed local recovery and fails closed on probe errors", async () => { + vi.stubEnv("PAPERCLIP_HOME", root); + markIdleIngressTracked(); markIdleStartupComplete({ scheduledBackups: false }); + let pending = true; + const unregister = registerIdleLocalWorkProbe(() => pending ? "present" : "none"); + try { + expect(await readIdleLocalWork()).toBe("present"); + pending = false; + expect(await readIdleLocalWork()).toBe("none"); + } finally { unregister(); } + const removeBroken = registerIdleLocalWorkProbe(() => { throw new Error("unknown deadline"); }); + try { expect(await readIdleLocalWork()).toBe("unknown"); } + finally { removeBroken(); } + expect(await readIdleLocalWork()).toBe("none"); + }); + it("checks custom spool directories as well as the default paths", async () => { vi.stubEnv("PAPERCLIP_HOME", root); vi.stubEnv("SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR", path.join(root, "orphans")); diff --git a/server/src/__tests__/terminal-workspace-work.test.ts b/server/src/__tests__/terminal-workspace-work.test.ts new file mode 100644 index 00000000000..a47b262ff5c --- /dev/null +++ b/server/src/__tests__/terminal-workspace-work.test.ts @@ -0,0 +1,130 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { notifyTerminalWorkspaceWork, registerTerminalWorkspaceWork, TERMINAL_WORKSPACE_RETRY_MS } from "../services/terminal-workspace-work.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const coordinators: ReturnType[] = []; +afterEach(async () => { + for (const coordinator of coordinators.splice(0)) await coordinator.stop(); + vi.useRealTimers(); +}); + +function setup() { + vi.useFakeTimers(); + const execute = vi.fn(async () => [{ xid: "42", status: "committed" }]); + const owner = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise) { + return callback({ execute }); + } }); + const enabled = vi.fn(() => true), primary = vi.fn(() => true), onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner, + canRun: () => enabled() && primary(), canReconcile: primary, onError, + }); + coordinators.push(coordinator); + const service = { + sweepTerminalWorkspaces: vi.fn(async (_limit?: number, _signal?: AbortSignal) => ({ checked: 0 })), + hasTerminalWorkspaceSweepContinuation: vi.fn(() => false), + nextTerminalWorkspaceSweepAt: vi.fn(async (): Promise => null), + }; + return { owner, coordinator, service, enabled, primary, onError, + start: () => registerTerminalWorkspaceWork(coordinator, service, enabled) }; +} + +describe("terminal workspace work", () => { + it("recovers once at startup and makes no scans or SQL for an empty lane", async () => { + const s = setup(); + const worker = s.start(); + await worker.ready; + await vi.advanceTimersByTimeAsync(3_600_000); + expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); + expect(s.service.nextTerminalWorkspaceSweepAt).toHaveBeenCalledTimes(1); + expect(s.coordinator.nextWakeAt()).toBeNull(); + expect(vi.getTimerCount()).toBe(0); + expect(idleWorkSnapshot().active).toBe(0); + expect(worker.hasUnrecoveredWork()).toBe(false); + }); + + it("restores the cooldown deadline, lets commits preempt it, and parks after cleanup", async () => { + const s = setup(), due = Date.now() + 86_400_000; + s.service.nextTerminalWorkspaceSweepAt.mockResolvedValue(due); + const worker = s.start(); + await worker.ready; + expect(worker.hasUnrecoveredWork()).toBe(false); + await vi.advanceTimersByTimeAsync(3_600_000); + expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); + await s.owner.transaction(tx => notifyTerminalWorkspaceWork(tx)); + await s.owner.transaction(tx => notifyTerminalWorkspaceWork(tx)); + await vi.advanceTimersByTimeAsync(1); + expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(2); + expect(s.coordinator.nextWakeAt()).toBe(due); + s.service.nextTerminalWorkspaceSweepAt.mockResolvedValue(null); + await vi.advanceTimersByTimeAsync(due - Date.now()); + expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(3); + expect(vi.getTimerCount()).toBe(0); + }); + + it("finishes bounded recovery pages then retries only existing overdue work", async () => { + const s = setup(); + s.service.hasTerminalWorkspaceSweepContinuation.mockReturnValueOnce(true); + s.service.nextTerminalWorkspaceSweepAt.mockResolvedValue(Date.now() - 1); + await s.start().ready; + expect(s.coordinator.nextWakeAt()).toBe(Date.now() + 1000); + await vi.advanceTimersByTimeAsync(1000); + expect(s.coordinator.nextWakeAt()).toBe(Date.now() + TERMINAL_WORKSPACE_RETRY_MS); + s.service.nextTerminalWorkspaceSweepAt.mockResolvedValue(null); + await vi.advanceTimersByTimeAsync(TERMINAL_WORKSPACE_RETRY_MS); + expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(3); + expect(vi.getTimerCount()).toBe(0); + }); + + it("keeps failed recovery visible to sleep admission until a successful empty pass", async () => { + const s = setup(); + s.service.sweepTerminalWorkspaces.mockRejectedValueOnce(new Error("database unavailable")); + const worker = s.start(); + await worker.ready; + expect(s.onError).toHaveBeenCalledTimes(1); + expect(worker.hasUnrecoveredWork()).toBe(true); + await vi.advanceTimersByTimeAsync(TERMINAL_WORKSPACE_RETRY_MS); + expect(worker.hasUnrecoveredWork()).toBe(false); + expect(vi.getTimerCount()).toBe(0); + }); + + it("defers scans and deadline SQL during standby and drain", async () => { + const s = setup(); + s.primary.mockReturnValue(false); + await s.start().ready; + s.primary.mockReturnValue(true); + s.enabled.mockReturnValue(false); + await s.owner.transaction(tx => notifyTerminalWorkspaceWork(tx)); + await vi.advanceTimersByTimeAsync(1); + expect(s.service.sweepTerminalWorkspaces).not.toHaveBeenCalled(); + expect(s.service.nextTerminalWorkspaceSweepAt).not.toHaveBeenCalled(); + s.enabled.mockReturnValue(true); + await vi.advanceTimersByTimeAsync(TERMINAL_WORKSPACE_RETRY_MS); + expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); + + it("holds admission until accepted cleanup finishes and waits for it during stop", async () => { + const s = setup(); + let release!: () => void; + s.service.sweepTerminalWorkspaces.mockImplementationOnce(async (_limit, signal) => { + await new Promise(resolve => { release = resolve; }); + expect(signal?.aborted).toBe(true); + return { checked: 1 }; + }); + const worker = s.start(); + await vi.advanceTimersByTimeAsync(1); + expect(idleWorkSnapshot().active).toBe(1); + let stopped = false; + const stopping = s.coordinator.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1); + expect(stopped).toBe(false); + release(); + await stopping; + await worker.ready; + expect(idleWorkSnapshot().active).toBe(0); + expect(s.service.nextTerminalWorkspaceSweepAt).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); +}); diff --git a/server/src/__tests__/workspace-git-inspection-cache.test.ts b/server/src/__tests__/workspace-git-inspection-cache.test.ts index f4703cc225e..c31b1b57be5 100644 --- a/server/src/__tests__/workspace-git-inspection-cache.test.ts +++ b/server/src/__tests__/workspace-git-inspection-cache.test.ts @@ -5,7 +5,7 @@ import { createWorkspaceGitInspectionCache } from "../services/workspace-git-ins const workspace = { id: "workspace", companyId: "company", cwd: "/repo", baseRef: "master" } as ExecutionWorkspace; afterEach(() => vi.useRealTimers()); -it("coalesces concurrent display reads and expires after five seconds", async () => { +it("coalesces concurrent display reads and expires after thirty seconds", async () => { vi.useFakeTimers(); const inspect = vi.fn(async () => ({ dirty: false })); const read = createWorkspaceGitInspectionCache(inspect); @@ -13,7 +13,10 @@ it("coalesces concurrent display reads and expires after five seconds", async () expect(inspect).toHaveBeenCalledTimes(1); await read(workspace); expect(inspect).toHaveBeenCalledTimes(1); - vi.advanceTimersByTime(5_000); + vi.advanceTimersByTime(29_999); + await read(workspace); + expect(inspect).toHaveBeenCalledTimes(1); + vi.advanceTimersByTime(1); await read(workspace); expect(inspect).toHaveBeenCalledTimes(2); // Callers that authorize cleanup retain the uncached inspector. diff --git a/server/src/index.ts b/server/src/index.ts index be6b6bd701f..9b5f85562b6 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -2,8 +2,9 @@ import { skillSourceSyncService } from "./services/skill-source-sync.js"; import { subscribeAllCompanyLiveEvents } from "./services/live-events.js"; import { DELIVERY_QUEUES } from "./services/delivery-work-notifications.js"; import type { createDeliveryWorkCoordinator } from "./services/delivery-work-coordinator.js"; +import { registerTerminalWorkspaceWork, TERMINAL_WORKSPACE_RETRY_MS } from "./services/terminal-workspace-work.js"; import { isIdleTaskDrainActive, beginIdleTrackedWork, trackIdleWork } from "./services/task-admission.js"; -import { markIdleStartupComplete } from "./services/idle-local-work.js"; +import { markIdleStartupComplete, registerIdleLocalWorkProbe } from "./services/idle-local-work.js"; import { clearIdleBackupWakeRequired, idleBackupWakeMarker, markIdleBackupWakeRequired, verifyIdleDatabaseBackup } from "./services/idle-database-backup.js"; import { cloudWarmStandbyServerOptions } from "./middleware/cloud-warm-standby.js"; import { createCloudWarmStandby } from "./services/cloud-warm-standby.js"; @@ -1272,6 +1273,7 @@ async function startServerWithDatabaseTeardown( resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction), }); const deliveryWork = app.locals.deliveryWork as ReturnType; + let unregisterTerminalWorkspaceProbe: (() => void) | undefined; const fastResponses = app.locals.fastResponses as ReturnType; const fastResponseWorker = deliveryWork.register(DELIVERY_QUEUES.fastResponse, { retryMs: 100, run: (signal) => fastResponses.sweepPending(signal), hasPending: () => fastResponses.hasPending(), @@ -1372,6 +1374,7 @@ async function startServerWithDatabaseTeardown( }); const terminalWorkspaces = executionWorkspaceService(db as any, { workspaceReaperCooldownDays: config.workspaceReaperCooldownDays, + terminalWorkspaceInspectionBackoffMs: TERMINAL_WORKSPACE_RETRY_MS, }); const scheduleMergedPullRequestConfirmationSweep = () => { if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; @@ -1386,36 +1389,22 @@ async function startServerWithDatabaseTeardown( 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 (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - trackHeartbeatSchedulerWork(terminalWorkspaces - .sweepTerminalWorkspaces() - .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 - + result.skippedCooldown; - const nowMs = Date.now(); - if (skipped > 0 && nowMs - lastTerminalWorkspaceSkipLogAt >= terminalWorkspaceSkipLogIntervalMs) { - lastTerminalWorkspaceSkipLogAt = nowMs; - logger.info(result, "terminal issue workspace reaper skipped all candidates"); - } - }) - .catch((err) => { - logger.error({ err }, "terminal issue workspace reaper failed"); - })); - }; + const terminalWorkspaceWorker = registerTerminalWorkspaceWork(deliveryWork, { + ...terminalWorkspaces, + sweepTerminalWorkspaces: async (limit, signal) => { + const result = await terminalWorkspaces.sweepTerminalWorkspaces(limit, signal); + if (result.archived > 0 || result.cleanupFailed > 0) { + logger.info(result, "terminal issue workspace reaper changed workspace state"); + } else if (result.checked > 0 && Date.now() - lastTerminalWorkspaceSkipLogAt >= 10 * 60 * 1000) { + lastTerminalWorkspaceSkipLogAt = Date.now(); + logger.info(result, "terminal issue workspace reaper skipped all candidates"); + } + return result; + }, + }, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped); + unregisterTerminalWorkspaceProbe = registerIdleLocalWorkProbe(() => terminalWorkspaceWorker.hasUnrecoveredWork() ? "present" : "none"); + await terminalWorkspaceWorker.ready; // The restart-safe cleanup backstop for adapter login sessions. The // in-process five-minute timer stays the primary control. This reaper runs @@ -1723,7 +1712,6 @@ async function startServerWithDatabaseTeardown( scheduleMergedPullRequestConfirmationSweep(); scheduleGitHubConnectionEventPoll(); scheduleGitHubConnectionContinuitySweep(); - scheduleTerminalWorkspaceSweep(); scheduleAdapterLoginReaperSweep(); scheduleSetupTokenReaperSweep(); scheduleEnvironmentLeaseCleanupSweep(); @@ -1984,6 +1972,7 @@ async function startServerWithDatabaseTeardown( await skillSourceSync.stop(); unsubscribeChatCompletions(); await deliveryWork.stop(); + unregisterTerminalWorkspaceProbe?.(); clearInterval(executionControlInterval); if (heartbeatSchedulerInterval) { clearInterval(heartbeatSchedulerInterval); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 315b548b40c..658c5880dab 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -25,6 +25,7 @@ export const DELIVERY_QUEUES = { question: "question-responses", pluginLifecycle: "plugin-resource-lifecycle", toolAction: "tool-action-receipts", + terminalWorkspaces: "terminal-workspaces", } as const; export type DeliveryQueue = typeof DELIVERY_QUEUES[keyof typeof DELIVERY_QUEUES]; diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index d8ff62f5561..70227a2faaf 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -11,6 +11,7 @@ import { taskWorkspaceSelectionSchema, type TaskWorkspaceSelection, type TaskWor import { authorizationService, executionWorkspaceReadSqlCondition, projectReadSqlCondition, issueReadSqlCondition, type AuthorizationActor } from "./authorization.js"; import type { SQL } from "drizzle-orm"; import { createWorkspaceGitInspectionCache } from "./workspace-git-inspection-cache.js"; +import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; import { execFile } from "node:child_process"; import { createHash } from "node:crypto"; import fs from "node:fs/promises"; @@ -275,6 +276,7 @@ export type ReopenClosedIsolatedExecutionWorkspaceResult = | { ok: false; code: "not_reopenable" | "rebuild_failed"; message: string }; export type ExecutionWorkspaceServiceOptions = { + terminalWorkspaceInspectionBackoffMs?: number; resolvePullRequestDetails?: PullRequestMergeDetailsResolver; now?: () => Date; beforeTerminalWorkspaceCleanup?: (workspace: ExecutionWorkspaceRow) => Promise; @@ -1617,6 +1619,9 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic // 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; + const terminalInspections = new Map; nextAt: number }>(); + const terminalWorkspaceDeadlines = new Map(); + let terminalSweepRescanRequested = false; // 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 @@ -1897,6 +1902,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic } return input.onSkip(); } + await notifyTerminalWorkspaceWork(tx, `workspace:${input.workspaceId}`); return input.write({ tx, fresh }); }); } @@ -2859,7 +2865,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic }; }, - sweepTerminalWorkspaces: async (limit = 50) => { + sweepTerminalWorkspaces: async (limit = 50, signal?: AbortSignal) => { // 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. @@ -2875,9 +2881,12 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic skippedRace: 0, skippedReopened: 0, skippedCooldown: 0, + skippedBackoff: 0, clearedStaleReopenPending: 0, }; } + const previousCursor = terminalSweepCursor; + const previousBoundary = terminalSweepBoundary; terminalSweepInProgress = true; try { const baseCandidateFilter = and( @@ -2897,6 +2906,8 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic // rotation cover a finite set, so the cursor always reaches a short page. if (!cursor) { terminalSweepBoundary = now(); + terminalWorkspaceDeadlines.clear(); + terminalSweepRescanRequested = false; } const boundary = terminalSweepBoundary; const boundaryFilter = boundary @@ -2939,10 +2950,57 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic skippedRace: 0, skippedReopened: 0, skippedCooldown: 0, + skippedBackoff: 0, clearedStaleReopenPending: 0, }; for (const workspace of candidates) { + signal?.throwIfAborted(); + const issueTree = await listWorkspaceIssueTree(workspace); + const sourceIssue = issueTree.find((issue) => issue.id === workspace.sourceIssueId); + const reopenPending = metadataHasReopenPendingConsumption( + workspace.metadata as Record | null, + ); + if (reopenPending) terminalWorkspaceDeadlines.set(workspace.id, now().getTime()); + if (!sourceIssue || issueTree.some(issue => !TERMINAL_ISSUE_STATUSES.has(issue.status))) { + if (reopenPending) { + await clearReopenPendingConsumptionUnderLock(workspace.id, { + expectedGeneration: readExecutionWorkspaceLifecycleGeneration( + workspace.metadata as Record | null, + ), + }); + } + result.skippedNonTerminalTree += 1; + continue; + } + const latestTerminalAt = Math.max(...issueTree.map(issue => issueTerminalTimestamp(issue)!.getTime())); + terminalWorkspaceDeadlines.set(workspace.id, reopenPending ? now().getTime() + : latestTerminalAt + workspaceReaperCooldownMs); + if (workspaceReaperCooldownMs > 0 && latestTerminalAt + workspaceReaperCooldownMs > now().getTime()) { + result.skippedCooldown += 1; + continue; + } + if (await workspaceHasActiveRun(workspace)) { + if (reopenPending) result.skippedReopened += 1; + else result.skippedActiveRun += 1; + continue; + } + if (opts.terminalWorkspaceInspectionBackoffMs) { + const revision = createHash("sha256").update(JSON.stringify([ + workspace.updatedAt, workspace.metadata, + [...issueTree].sort((a, b) => a.id.localeCompare(b.id)), + ])).digest("hex"); + const previous = terminalInspections.get(workspace.id); + if (previous?.revision === revision && previous.nextAt > now().getTime()) { + result.skippedBackoff += 1; + continue; + } + if (terminalInspections.size >= 256) terminalInspections.delete(terminalInspections.keys().next().value!); + terminalInspections.set(workspace.id, { + revision, issueIds: new Set(issueTree.map(issue => issue.id)), + nextAt: now().getTime() + opts.terminalWorkspaceInspectionBackoffMs, + }); + } const executionWorkspace = toExecutionWorkspace(workspace); const { git, statusInspectionSucceeded } = await inspectGitCloseReadiness(executionWorkspace); if (!statusInspectionSucceeded) { @@ -2950,9 +3008,6 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic continue; } const assessment = await assessDelivery(workspace, git); - const reopenPending = metadataHasReopenPendingConsumption( - workspace.metadata as Record | null, - ); if (!assessment.sourceIssueTerminal || !assessment.subtreeTerminal) { if (reopenPending) { // The source issue left the terminal state, so the reopen transition @@ -3158,6 +3213,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic result.skippedRace += 1; continue; } + terminalWorkspaceDeadlines.delete(workspace.id); await logActivity(db, { companyId: archived.companyId, @@ -3207,24 +3263,56 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic } } return result; + } catch (error) { + terminalSweepCursor = previousCursor; + terminalSweepBoundary = previousBoundary; + terminalSweepRescanRequested = true; + throw error; } finally { terminalSweepInProgress = false; } }, + hasTerminalWorkspaceSweepContinuation: () => terminalSweepCursor !== null || terminalSweepRescanRequested, + + terminalWorkspaceWorkSettled: (hints?: ReadonlySet) => { + if (terminalSweepInProgress || terminalSweepCursor !== null) terminalSweepRescanRequested = true; + pullRequestStateCache.clear(); + if (!hints) { terminalInspections.clear(); return; } + for (const [id, inspection] of terminalInspections) { + if (hints.has(`workspace:${id}`) || [...inspection.issueIds].some(issueId => hints.has(`issue:${issueId}`))) { + terminalInspections.delete(id); + } + } + }, + + nextTerminalWorkspaceSweepAt: async () => { + if (terminalWorkspaceDeadlines.size === 0) return null; + const next = [...terminalWorkspaceDeadlines.values()].reduce((earliest, deadline) => Math.min(earliest, deadline), Infinity); + const nowMs = now().getTime(); + if (next <= nowMs && opts.terminalWorkspaceInspectionBackoffMs) { + return Math.min(nowMs + opts.terminalWorkspaceInspectionBackoffMs, + ...[...terminalInspections.values()].filter(inspection => inspection.nextAt > nowMs).map(inspection => inspection.nextAt)); + } + return next; + }, + create: async (data: typeof executionWorkspaces.$inferInsert) => { - const row = await db - .insert(executionWorkspaces) - .values(data) - .returning() - .then((rows) => rows[0] ?? null); + const row = await db.transaction(async tx => { + await notifyTerminalWorkspaceWork(tx, data.id ? `workspace:${data.id}` + : data.sourceIssueId ? `issue:${data.sourceIssueId}` : undefined); + return tx.insert(executionWorkspaces).values(data).returning().then(rows => rows[0] ?? null); + }); return row ? toExecutionWorkspace(row) : null; }, update: async (id: string, patch: Partial) => { if (patch.metadata === undefined && !workspaceSourceKeys.some(key => patch[key] !== undefined)) { - const row = await db.update(executionWorkspaces).set({ ...patch, updatedAt: new Date() }) - .where(eq(executionWorkspaces.id, id)).returning().then(rows => rows[0] ?? null); + const row = await db.transaction(async tx => { + await notifyTerminalWorkspaceWork(tx, `workspace:${id}`); + return tx.update(executionWorkspaces).set({ ...patch, updatedAt: new Date() }) + .where(eq(executionWorkspaces.id, id)).returning().then(rows => rows[0] ?? null); + }); return row ? toExecutionWorkspace(row) : null; } const [identity] = await db.select({ companyId: executionWorkspaces.companyId }).from(executionWorkspaces) @@ -3237,6 +3325,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic )).for("no key update"); if (!current) return null; const nextPatch = await preserveWorkspaceOwnershipForPatch(tx, current, patch); + await notifyTerminalWorkspaceWork(tx, `workspace:${id}`); const [row] = await tx.update(executionWorkspaces).set({ ...nextPatch, updatedAt: new Date() }) .where(and(eq(executionWorkspaces.id, id), eq(executionWorkspaces.companyId, current.companyId))).returning(); return row ? toExecutionWorkspace(row) : null; @@ -3493,6 +3582,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic // generation so a queued cleanup that captured the old generation does // nothing. Clear cleanupEligibleAt so the reaper does not destroy the // half-built worktree while a later reopen retries. + await notifyTerminalWorkspaceWork(tx, `workspace:${row.id}`); await tx .update(executionWorkspaces) .set({ @@ -3543,6 +3633,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic // flag stops the terminal reaper and the archive route from archiving and // destroying the rebuilt worktree in that window. const activeMetadata = setMetadataReopenPendingConsumption(nextMetadata, now()); + await notifyTerminalWorkspaceWork(tx, `workspace:${row.id}`); const activeRow = await tx .update(executionWorkspaces) .set({ @@ -3735,6 +3826,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic const archiveMetadata = clearMetadataReopenPendingConsumption( bumpExecutionWorkspaceLifecycleGeneration(baseMetadata), ); + await notifyTerminalWorkspaceWork(tx, `workspace:${input.id}`); const archived = await tx .update(executionWorkspaces) .set({ @@ -3999,6 +4091,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic updatePatch.name = inspection.toBranch; } + await notifyTerminalWorkspaceWork(tx, `workspace:${lockedWorkspace.id}`); const [branchUpdatedRow] = await tx .update(executionWorkspaces) .set(updatePatch) diff --git a/server/src/services/idle-local-work.ts b/server/src/services/idle-local-work.ts index 0338a66330c..95b6549ebbc 100644 --- a/server/src/services/idle-local-work.ts +++ b/server/src/services/idle-local-work.ts @@ -5,6 +5,11 @@ import { idleWorkSnapshot } from "./task-admission.js"; export type IdleLocalWork = "none" | "present" | "unknown"; const spoolDirectories = new Set(); +const localWorkProbes = new Set<() => IdleLocalWork>(); +export function registerIdleLocalWorkProbe(probe: () => IdleLocalWork): () => void { + localWorkProbes.add(probe); + return () => { localWorkProbes.delete(probe); }; +} export function registerIdleSpoolDirectory(directory: string): void { spoolDirectories.add(path.resolve(directory)); } export const idleAccountingSpoolPath = () => path.join(resolvePaperclipInstanceRoot(), "accounting-receipts"); export const idleOrphanSpoolPath = () => process.env.SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR ?? @@ -61,6 +66,11 @@ export async function readIdleLocalWork(options: { backupCheckpoint?: boolean } // This exemption belongs only to a scan which will finish a new backup // under the same admission hold before it can return an empty report. if ((scheduledBackups && !options.backupCheckpoint) || idleWorkSnapshot().active !== 0) return "present"; + try { + const local = [...localWorkProbes].map(probe => probe()); + if (local.includes("unknown")) return "unknown"; + if (local.includes("present")) return "present"; + } catch { return "unknown"; } const directories = new Set([...spoolDirectories, idleAccountingSpoolPath(), idleOrphanSpoolPath()]); const results = await Promise.all([ ...[...directories].map(inspectIdleSpool), diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index c0b748b2a67..ecf8a07c6fd 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -23,6 +23,7 @@ import { } from "@paperclipai/shared"; import { conflict, notFound, unprocessable } from "../errors.js"; import type { IssuePostCommitAction } from "./issues.js"; +import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; type IssueRow = typeof issues.$inferSelect; type HoldRow = typeof issueTreeHolds.$inferSelect; @@ -977,6 +978,7 @@ export function issueTreeControlService(db: Db) { const restored: TreeStatusUpdateResult["updatedIssues"] = []; for (const [status, issueIdsForStatus] of issueIdsByStatus) { if (issueIdsForStatus.length === 0) continue; + for (const issueId of issueIdsForStatus) await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`); const rows = await tx .update(issues) .set({ diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 3a167a7865a..b8eb1e51c1f 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -6,6 +6,7 @@ import { assertTaskWorkspaceAccess, assertTaskWorkspaceSourceProjectAccess } fro import { type TaskWorkspaceSelection, type TaskWorkspaceIntent } from "@paperclipai/shared"; import { executionWorkspaceService } from "./execution-workspaces.js"; import { notifyChatPublicationWork } from "./chat-work-notifications.js"; +import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; import { recordChatHandoff, recordChatCompletion, existingChatCompletionReply, acknowledgeChatCompletionReply } from "./chat-completion-delivery.js"; import { mirrorSlackBoardComment, slackBoardReplyBindings } from "./slack-board-messages.js"; import { assertAgentRunWriteAllowed } from "../agent-run-cancellation.js"; @@ -10375,6 +10376,10 @@ export function issueService(db: Db) { }), ); + if (values.parentId || values.executionWorkspaceId || values.status === "done" || values.status === "cancelled") { + await notifyTerminalWorkspaceWork(tx, values.executionWorkspaceId ? `workspace:${values.executionWorkspaceId}` + : values.parentId ? `issue:${values.parentId}` : undefined); + } const [issue] = await tx.insert(issues).values(values).returning(); await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId }); if (issue.visibility === "private" && !inheritedPrivateSubtree && issueData.createdByAgentId) { @@ -11192,6 +11197,11 @@ export function issueService(db: Db) { (blockedByIssueIds !== undefined || issueData.unblockDescriptor !== undefined)))) { patch.statusVersion = sql`${issues.statusVersion} + 1` as unknown as number; } + if ((patch.status !== undefined && patch.status !== receiptExisting.status) + || patch.completedAt !== undefined || patch.cancelledAt !== undefined + || issueData.parentId !== undefined || issueData.executionWorkspaceId !== undefined) { + await notifyTerminalWorkspaceWork(tx, `issue:${id}`); + } const updated = await tx .update(issues) .set(patch) @@ -11593,6 +11603,7 @@ export function issueService(db: Db) { let removedIssue; try { + await notifyTerminalWorkspaceWork(tx, `issue:${id}`); removedIssue = await tx .delete(issues) .where(eq(issues.id, id)) @@ -12092,6 +12103,7 @@ export function issueService(db: Db) { const isTerminal = existing.status === "done" || existing.status === "cancelled"; const releaseStatus = existing.status === "in_progress" ? "todo" : existing.status; + await notifyTerminalWorkspaceWork(tx, `issue:${id}`); const updated = await tx .update(issues) .set({ @@ -12141,6 +12153,7 @@ export function issueService(db: Db) { patch.assigneeAgentId = null; } + await notifyTerminalWorkspaceWork(tx, `issue:${id}`); const updated = await tx .update(issues) .set(patch) diff --git a/server/src/services/terminal-workspace-work.ts b/server/src/services/terminal-workspace-work.ts new file mode 100644 index 00000000000..71d3bf637aa --- /dev/null +++ b/server/src/services/terminal-workspace-work.ts @@ -0,0 +1,44 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export const TERMINAL_WORKSPACE_RETRY_MS = 5 * 60 * 1000; + +export function notifyTerminalWorkspaceWork(transaction: object, hint?: string) { + return notifyDeliveryWork(transaction, DELIVERY_QUEUES.terminalWorkspaces, hint); +} + +export function registerTerminalWorkspaceWork( + coordinator: ReturnType, + service: { + sweepTerminalWorkspaces(limit?: number, signal?: AbortSignal): Promise<{ checked: number }>; + nextTerminalWorkspaceSweepAt(): Promise; + hasTerminalWorkspaceSweepContinuation(): boolean; + terminalWorkspaceWorkSettled?(hints?: ReadonlySet): void; + }, + canRun: () => boolean, +) { + let next: number | null = Date.now(); + let recovered = false; + const worker = coordinator.register(DELIVERY_QUEUES.terminalWorkspaces, { + retryMs: TERMINAL_WORKSPACE_RETRY_MS, + onSettled: service.terminalWorkspaceWorkSettled, + run: signal => { + recovered = false; + next = Date.now() + TERMINAL_WORKSPACE_RETRY_MS; + return service.sweepTerminalWorkspaces(50, signal); + }, + hasPending: async () => { + if (!canRun()) next = Date.now() + TERMINAL_WORKSPACE_RETRY_MS; + else if (service.hasTerminalWorkspaceSweepContinuation()) next = Date.now() + 1000; + else { + const deadline = await service.nextTerminalWorkspaceSweepAt(); + next = deadline === null ? null : deadline > Date.now() + ? deadline : Date.now() + TERMINAL_WORKSPACE_RETRY_MS; + recovered = true; + } + return false; + }, + nextRunAt: () => next, + }); + return { ...worker, hasUnrecoveredWork: () => !recovered }; +} diff --git a/server/src/services/work-products.ts b/server/src/services/work-products.ts index 0f3dc35ae2e..e4a8cbb5a10 100644 --- a/server/src/services/work-products.ts +++ b/server/src/services/work-products.ts @@ -4,6 +4,7 @@ import { executionWorkspaces, heartbeatRunEvents, issueWorkProducts, workspaceRu import type { IssueWorkProduct } from "@paperclipai/shared"; import { unprocessable } from "../errors.js"; import { insertRowsInChunks } from "./batch-insert.js"; +import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; import { createPullRequestMergeDetailsResolver, extractGitHubPullRequestReferences, @@ -272,6 +273,7 @@ export function workProductService( createForIssue: async (issueId: string, companyId: string, data: Omit) => { const row = await db.transaction(async (tx) => { + if (data.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`); await validateExecutionWorkspace(tx, companyId, data.executionWorkspaceId); if (data.isPrimary) { await tx @@ -307,6 +309,8 @@ export function workProductService( .then((rows) => rows[0] ?? null); if (!existing) return null; + if (existing.type === "pull_request" || patch.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`); + await validateExecutionWorkspace(tx, existing.companyId, patch.executionWorkspaceId); if (patch.isPrimary === true) { @@ -374,16 +378,21 @@ export function workProductService( // prefix behind (which a retry would then duplicate). Mirrors the // per-writer transaction the batched issue/document writers use. await db.transaction(async (tx) => { + for (const value of values) { + if (value.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${value.issueId}`); + } await insertRowsInChunks(tx, issueWorkProducts, values); }); }, remove: async (id: string) => { - const row = await db - .delete(issueWorkProducts) - .where(eq(issueWorkProducts.id, id)) - .returning() - .then((rows) => rows[0] ?? null); + const row = await db.transaction(async tx => { + const [existing] = await tx.select({ type: issueWorkProducts.type, issueId: issueWorkProducts.issueId }) + .from(issueWorkProducts).where(eq(issueWorkProducts.id, id)); + if (existing?.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`); + return tx.delete(issueWorkProducts).where(eq(issueWorkProducts.id, id)) + .returning().then(rows => rows[0] ?? null); + }); return row ? toIssueWorkProduct(row) : null; }, }; diff --git a/server/src/services/workspace-git-inspection-cache.ts b/server/src/services/workspace-git-inspection-cache.ts index 89ed9403c75..6fad32e4766 100644 --- a/server/src/services/workspace-git-inspection-cache.ts +++ b/server/src/services/workspace-git-inspection-cache.ts @@ -19,7 +19,7 @@ export function createWorkspaceGitInspectionCache(inspect: (workspace: Execut const entry = { expiresAt: Number.POSITIVE_INFINITY, promise: Promise.resolve().then(() => inspect(workspace)) }; entries.set(key, entry); entry.promise = entry.promise.then((result) => { - entry.expiresAt = Date.now() + 5_000; + entry.expiresAt = Date.now() + 30_000; return result; }, (error) => { if (entries.get(key) === entry) entries.delete(key);