perf(workspaces): wake terminal cleanup from committed changes (#16282)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Agents use execution workspaces to retain their files and Git
branches.
> - A reaper closes eligible workspaces after their task tree is
terminal.
> - The reaper scanned every heartbeat tick and read Git before cheap
eligibility checks.
> - Permanent scans also query idle databases when no cleanup work
exists.
> - This pull request wakes cleanup after transaction settlement and
restores bounded work at startup.
> - The benefit is fewer scans while fresh cleanup guards continue to
protect retained work.

## Linked Issues or Issue Description

Related work: Refs #13619, Refs #11753. Those PRs bound scans or back
off unchanged candidates. This PR uses the transaction work coordinator
to remove the permanent reaper tick. It overlaps their early
terminal-state filtering but does not change their runtime restart
behavior or introduce a second Git scheduler cache.

**What existing behavior does this improve?**

Terminal workspace cleanup and Git inspection for workspace detail
reads.

**Subsystem affected**

Server orchestration, execution workspaces, and idle sleep admission.

**Current behavior**

The heartbeat tick scans workspace candidates every thirty seconds. It
runs Git before rejecting nonterminal trees, cooldowns, and active runs.
Display inspections expire after five seconds.

**Proposed behavior**

Recover workspace rows in bounded startup pages. Wake after relevant
issue, workspace, and pull-request work-product transactions settle.
Schedule the terminal tree cooldown or a five-minute eligibility retry.
Back off unchanged candidates across unrelated wakes, and invalidate
their backoff after relevant committed changes. Park empty and
nonterminal-only lanes. Reuse display inspections for thirty seconds.
Read fresh Git before destructive cleanup.

**Reason and benefit**

Avoid recurring database and Git work when no cleanup is possible.
Retain transaction admission holds, restart recovery, and fresh deletion
checks. Retention eligibility can wait for the next host wake; active
and failed recovery work still blocks sleep.

**Breaking changes**

No schema or API shape changes. Display Git state can lag by thirty
seconds. Cleanup eligibility retries can take five minutes.

## What Changed

- Add a terminal-workspace worker to the existing transaction work
coordinator and scheduler.
- Signal relevant writes before mutation on the owning transaction.
Outer transaction settlement wakes cleanup.
- Derive cooldown deadlines from the full company-scoped terminal task
tree. Continue bounded recovery pages without a permanent sweep.
- Bound unchanged candidate inspections to five minutes, retain at most
256 entries, and invalidate affected entries with scoped transaction
hints. Repeated wakes preserve the earliest inspection deadline.
- Collect deadlines from task trees already read by bounded pages.
Restore the cursor after a failed page and rescan commits received
during a rotation.
- Check terminal state, cooldown, and active runs before Git. Preserve
lifecycle locks, generation fences, and fresh final checks.
- Keep failed or incomplete recovery visible to idle admission. Preserve
the passive retention exception.
- Extend the bounded display cache to thirty seconds and document the
scheduling and wake limits.

## Verification

- Focused workspace, coordinator, cache, and local sleep suites: 111
tests passed with real local PostgreSQL and Git worktrees where
applicable.
- Final worker, display cache, and local sleep suites: 28 tests passed.
- The regressions prove three ineligible workspaces invoke zero Git
scans, cached clean display state cannot remove new untracked files, and
real producer notifications wait for outer transaction settlement.
- `pnpm -r typecheck` and `pnpm build` passed locally. Map and
module-boundary checks passed.
- Work-product and sleep-safety suites: 221 tests passed. Reopen and
issue-tree control suites: 46 tests passed. Final backoff, workspace,
worker, and work-product suites: 87 tests passed.
- Final workspace and worker suites: 81 tests passed. Deadline selection
performs zero database reads. A failed full page retries its unread
terminal workspace before recovery can park.
- Full CI coverage passed on head `2c397970f7`: general tests,
serialized server suites, runner verification, E2E shards, typecheck,
build, and the final verification gate. The duplicate serial local `pnpm
test:run` was stopped after full CI coverage passed; it is not reported
as a completed local pass.
- Greptile: 5/5 with both findings resolved. The security scan also
passed. No failing or pending PR checks remain.

## Risks

- Display state can be stale for thirty seconds. Destructive guards
bypass the display cache.
- Undelivered or otherwise ineligible terminal work retries after five
minutes instead of each heartbeat tick.
- Transaction signals cover the owning `createDb` scope. Direct SQL and
independent server processes require startup recovery.
- Local cooldown timers are maintenance hints. They do not establish an
external wake alarm or whole-instance sleep safety. Existing blockers
remain in force; failed bounded recovery keeps the host awake.

## Model Used

OpenAI GPT-6, model ID `gpt-6`, through Codex. Reasoning, repository
inspection, shell execution, and browser tools assisted the change. The
runtime does not expose the context-window size.

## 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
- [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
- [x] I have considered and documented any risks above
- [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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-14 08:28:45 -05:00
1 parent 1e731520bb
commit 7c016c049b
17 files changed
+580 -54

No files matched your search

+29 -1
View File
@@ -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
+7
View File
@@ -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
+8
View File
@@ -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.
@@ -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);
@@ -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);
+17 -1
View File
@@ -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"));
@@ -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<typeof createDeliveryWorkCoordinator>[] = [];
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<any>) {
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<number | null> => 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<void>(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);
});
});
@@ -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.
+20 -31
View File
@@ -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<typeof createDeliveryWorkCoordinator>;
let unregisterTerminalWorkspaceProbe: (() => void) | undefined;
const fastResponses = app.locals.fastResponses as ReturnType<typeof import("./services/fast-responses.js").fastResponseService>;
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);
@@ -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];
+104 -11
View File
@@ -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<void>;
@@ -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<string, { revision: string; issueIds: Set<string>; nextAt: number }>();
const terminalWorkspaceDeadlines = new Map<string, number>();
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<string, unknown> | 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<string, unknown> | 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<string, unknown> | 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<string>) => {
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<typeof executionWorkspaces.$inferInsert>) => {
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)
+10
View File
@@ -5,6 +5,11 @@ import { idleWorkSnapshot } from "./task-admission.js";
export type IdleLocalWork = "none" | "present" | "unknown";
const spoolDirectories = new Set<string>();
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),
@@ -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({
+13
View File
@@ -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)
@@ -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<typeof createDeliveryWorkCoordinator>,
service: {
sweepTerminalWorkspaces(limit?: number, signal?: AbortSignal): Promise<{ checked: number }>;
nextTerminalWorkspaceSweepAt(): Promise<number | null>;
hasTerminalWorkspaceSweepContinuation(): boolean;
terminalWorkspaceWorkSettled?(hints?: ReadonlySet<string>): 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 };
}
+14 -5
View File
@@ -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<typeof issueWorkProducts.$inferInsert, "issueId" | "companyId">) => {
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;
},
};
@@ -19,7 +19,7 @@ export function createWorkspaceGitInspectionCache<T>(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);