mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-03 17:51:58 +01:00
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Completed tasks leave workspaces that need safe cleanup. > - Providers without ancestry inspection currently need a recorded merged PR before cleanup can start. > - A squash merge changes commit identity, so direct ancestry cannot prove that delivery. > - This change proves delivery from the finalized HEAD and captured destination before storage inspection. > - Fresh storage checks and a second delivery check still protect removal. > - This reuses the existing worker and receipts. It adds no table or timer. ## Linked Issues or Issue Description Refs #18410. Related: #13400 addresses workspace identity during inspection; this change adds destination-aware delivery discovery. ## What Changed - Resolve direct ancestry or discover an associated merged GitHub PR without a recorded work product. Require exact HEAD, destination repository and branch, and a reachable integration commit. - Use the latest settled finalization receipt as a candidate. Reject missing, failed, pending, conflicting, or changed evidence. - Inspect storage only after positive remote proof for providers that disable ancestry inspection. Recheck HEAD, branch, source identity and delivery before removal intent. - Add an optional provider capability for a specific PR recovery hint and an optional prepared destination receipt. Older providers keep their existing protocol. - Wake cleanup when finalization writes settle or gain a workspace binding. Keep remote-only retries for concrete unresolved delivery. Park missing or unsupported candidate identity. - Reuse the source project's explicit GitHub consent for delivery checks through the existing authorization service. Revalidate the bound repository and consent before removal; missing or revoked consent retains storage without switching identities. Public sources remain anonymous. - Update the contracts, documentation, access map, and targeted regressions. ## Verification - Initial-head GitHub CI passed at `5acd4c818be6f1764f8319f42891f9956c357c26`, including full build and typecheck. Standard Greptile is 5/5 with no open threads. The first Apex review scored 3/5 and found startup latency from remote proof lookups and removal retry identity changes across provider capability changes. Both corrections are pushed in `9d1a38df3d`, with the qualified destination-ref length fix. Red regressions reproduced each issue. Latest targeted checks passed: cleanup/worker 26, provider integration 41, SDK contract 8. Head `9d1a38df3d` passed standard and Apex reviews at 5/5 with no open threads, plus all CI including full build/typecheck. Canary Dry Run initially failed on an npm 404 fetching `rollup-4.67.3.tgz`; the exact tarball subsequently returned HTTP 200, and one targeted retry passed without source changes. Aggregate verification also passed. After #18403 merged, master introduced conflicts in shared workspace cleanup and metadata handling. Resolved them in `7700832536`, preserving remote snapshot cleanup and Git delivery-proof cleanup. Targeted merge verification: 93 tests passed in the first run (59 opt-in skips), then all 16 remote lifecycle tests passed with explicit database opt-in; map checks passed. The isolated database is stopped. The merge head received standard review 5/5. Its CI found two failures in unchanged test cleanup paths: native runtime temporary-directory ENOTEMPTY and agent-chat cancellation lock timeout. Those failures remain recorded and require successful final-head verification. The subsequent project-consent change has targeted coverage: authorization36, resolver33 and cleanup19 tests passed, including revoked consent, anonymous public access and source-binding races. The positive consent test failed before the fix. Current head `c772679e05` has green GitHub CI, including full build/typecheck and both previously failing test jobs, plus standard Greptile 5/5 with no open threads. Neither cleanup failure required changes to its test path. Apex also completed at 5/5 on this exact head (check `124651548532`, summary updated 2026-10-31 20:45 UTC), with no open review threads. The PR remains draft. Earlier failures remain recorded above. - Final focused run on the rebased branch: **182 tests passed across six files**, covering the resolver, delivery cleanup, provider bridge, finalization notifications, existing workspace service and SDK contracts. The disposable test database was stopped afterward. - Targeted resolver tests: 33 passed. These cover ancestry, discovered squash PRs, wrong HEAD/base/repository, removed integration commits, bounded responses, cancellation and credential isolation. - Targeted cleanup integration tests: 17 passed. Provider bridge integration: 36 passed. SDK contract tests: 7 passed. Finalization notification tests: 9 passed. - A negative control that removed the final delivery check failed the race regression as expected. The check was restored. - Read-only live GitHub probe: the new resolver discovered PR #18410 from HEAD `3f5ade4cd42b707040e9a82fdaed343e3d736a09`, without PR hints. It proved integration commit `6484f96c28bfb11a793ce4be6cc5ac15643a5a02` reachable from observed master `5bc029bf6c1db6647e96acea323f3723fd1d3eba`. GitHub also advertised the exact original HEAD at `refs/pull/18410/head`. - `node scripts/check-maps.mjs` passed. Structural checks do not prove deployment behavior. - Local tests are targeted by user request. Full build, typecheck and test suite run in GitHub CI. A standalone server typecheck could not complete because local Runner build declarations were absent; no changed-file diagnostics were reported. - Initial local execution hit a stalled pnpm wrapper, sandbox socket denial and stale generated SDK declarations. Direct installed binaries, the authorized isolated test database and rebuilt SDK resolved those setup failures. No deployment was performed. ## Risks - Destructive cleanup still requires fresh storage safety and lifecycle checks. Remote proof alone never authorizes removal. A persisted removal intent preserves existing recovery semantics. - Legacy workspaces without an unambiguous destination or completion HEAD remain retained. This does not guess the original default branch. - Private repositories use the explicit standing GitHub consent on the bound source project, with current membership, grant, repository access and trust checks. No separate company token is required when that consent is valid. Missing or revoked consent and unavailable or rate-limited access retain storage; the worker does not borrow a task or agent identity. - A known undelivered commit keeps the existing five-minute retry because external GitHub changes have no guaranteed ingress to this lane. This can keep cleanup scheduled. Missing evidence parks instead. - Providers need the new recovery capability to clean a squash merge whose original HEAD is reachable only through its PR ref. Providers without it remain conservative. - The resolver currently supports GitHub.com. Existing local and other-provider ancestry paths stay in place. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, repository tools and parallel code agents. The exact runtime model ID and context-window size were not exposed to this session. ## 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>
1192 lines
51 KiB
TypeScript
1192 lines
51 KiB
TypeScript
import { createPluginWorkspaceFixture } from "./helpers/plugin-workspaces.js";
|
|
import { randomUUID } from "node:crypto";
|
|
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
|
import { and, eq, sql } from "drizzle-orm";
|
|
import {
|
|
createDb,
|
|
subscribeDatabaseWork,
|
|
executionWorkspaces,
|
|
heartbeatRuns,
|
|
issues,
|
|
issueWorkProducts,
|
|
plugins,
|
|
pluginCompanySettings,
|
|
} from "@paperclipai/db";
|
|
import {
|
|
preparePluginWorkspace,
|
|
inspectPluginWorkspace,
|
|
finalizePluginWorkspace,
|
|
removePluginWorkspace,
|
|
type WorkspaceWorkers,
|
|
} from "../services/plugin-workspaces.js";
|
|
import { executionWorkspaceService } from "../services/execution-workspaces.js";
|
|
import { pluginRegistryService } from "../services/plugin-registry.js";
|
|
import { projectService } from "../services/projects.js";
|
|
import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
|
|
import {
|
|
getEmbeddedPostgresTestSupport,
|
|
startEmbeddedPostgresTestDatabase,
|
|
} from "./helpers/embedded-postgres.js";
|
|
const support = await getEmbeddedPostgresTestSupport();
|
|
(support.supported ? describe : describe.skip)(
|
|
"plugin workspace ownership",
|
|
() => {
|
|
let db: ReturnType<typeof createDb>,
|
|
stopDb: (() => Promise<void>) | undefined;
|
|
beforeAll(async () => {
|
|
const started =
|
|
await startEmbeddedPostgresTestDatabase("plugin-workspaces");
|
|
db = createDb(started.connectionString, { maxConnections: 1 });
|
|
stopDb = started.stop;
|
|
});
|
|
afterAll(async () => {
|
|
await db?.$client.end();
|
|
await stopDb?.();
|
|
});
|
|
const fixture = (companyId?: string) =>
|
|
createPluginWorkspaceFixture(db, companyId);
|
|
it.each([1, undefined])("gates recovery hints by provider capability (%s)", async (version) => {
|
|
const f = await fixture();
|
|
const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId));
|
|
await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson,
|
|
workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, recoveryPullRequestVersion: version as 1 | undefined } } }).where(eq(plugins.id, f.pluginId));
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
const recoveryPullRequest = { number: 42, headSha: "a".repeat(40) };
|
|
await inspectPluginWorkspace(db, f.workers, workspace!, recoveryPullRequest);
|
|
expect(f.calls.at(-1).operation.recoveryPullRequest).toEqual(version === 1 ? recoveryPullRequest : undefined);
|
|
let freshInspectionCompleted = false;
|
|
await expect(removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40), recoveryPullRequest, async () => {
|
|
freshInspectionCompleted = f.calls.at(-1).operation.kind === "inspect";
|
|
throw new Error("delivery changed");
|
|
})).rejects.toThrow("delivery changed");
|
|
expect(freshInspectionCompleted).toBe(true);
|
|
const [unchanged] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id));
|
|
expect(unchanged.metadata?.workspaceProviderRemoval).toBeUndefined();
|
|
f.loseRemoval();
|
|
await expect(removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40), recoveryPullRequest)).rejects.toThrow();
|
|
const [pending] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id));
|
|
expect((pending.metadata?.workspaceProviderRemoval as { operation: { recoveryPullRequest?: unknown } }).operation.recoveryPullRequest)
|
|
.toEqual(version === 1 ? recoveryPullRequest : undefined);
|
|
await removePluginWorkspace(db, f.workers, pending, "a".repeat(40), recoveryPullRequest, async () => { throw new Error("must not reauthorize pending removal"); });
|
|
expect(f.calls.at(-1).operation.recoveryPullRequest).toEqual(version === 1 ? recoveryPullRequest : undefined);
|
|
});
|
|
it.each([1, undefined])("replays a lost removal acknowledgement unchanged after capability flip (%s)", async (version) => {
|
|
const f = await fixture();
|
|
const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId));
|
|
const setCapability = (recoveryPullRequestVersion: 1 | undefined) => db.update(plugins).set({ manifestJson: {
|
|
...plugin.manifestJson, workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, recoveryPullRequestVersion },
|
|
} }).where(eq(plugins.id, f.pluginId));
|
|
await setCapability(version as 1 | undefined);
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
f.loseRemoval();
|
|
await expect(removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40), { number: 42, headSha: "a".repeat(40) })).rejects.toThrow();
|
|
const first = f.calls.at(-1);
|
|
await db.update(plugins).set({ status: "disabled" }).where(eq(plugins.id, f.pluginId));
|
|
const count = f.calls.length;
|
|
await expect(removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40))).rejects.toThrow("unavailable");
|
|
expect(f.calls).toHaveLength(count);
|
|
await db.update(plugins).set({ status: "ready" }).where(eq(plugins.id, f.pluginId));
|
|
await setCapability(version === 1 ? undefined : 1);
|
|
const [pending] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id));
|
|
await removePluginWorkspace(db, f.workers, pending, "a".repeat(40));
|
|
expect(f.calls.at(-1).operation).toEqual(first.operation);
|
|
expect(f.calls.at(-1).operationId).toBe(first.operationId);
|
|
expect(pending.metadata?.workspaceProviderRemoval).toMatchObject({ operation: first.operation, operationId: first.operationId });
|
|
const [removed] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id));
|
|
expect(removed.metadata?.workspaceProviderRemoved).toBe(true);
|
|
});
|
|
it.each(["legacy", "ambiguous", "mismatched"] as const)("fails closed for uncertain removal identities (%s)", async (identity) => {
|
|
const hint = identity !== "legacy";
|
|
const f = await fixture();
|
|
const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId));
|
|
await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson,
|
|
workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, recoveryPullRequestVersion: 1 } } }).where(eq(plugins.id, f.pluginId));
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
f.loseRemoval();
|
|
await expect(removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40), hint ? { number: 42, headSha: "a".repeat(40) } : undefined)).rejects.toThrow();
|
|
const first = f.calls.at(-1);
|
|
const [pending] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id));
|
|
const intent = pending.metadata!.workspaceProviderRemoval as Record<string, unknown>;
|
|
if (identity === "mismatched") intent.operationId = "0".repeat(64);
|
|
else delete intent.operationId;
|
|
await db.update(executionWorkspaces).set({ metadata: { ...pending.metadata, workspaceProviderRemoval: intent } }).where(eq(executionWorkspaces.id, pending.id));
|
|
if (hint) {
|
|
const count = f.calls.length;
|
|
await expect(removePluginWorkspace(db, f.workers, pending, "a".repeat(40))).rejects.toThrow(identity === "ambiguous" ? "identity is ambiguous" : "identity does not match");
|
|
expect(f.calls).toHaveLength(count);
|
|
} else {
|
|
await removePluginWorkspace(db, f.workers, pending, "a".repeat(40));
|
|
expect(f.calls.at(-1).operationId).toBe(first.operationId);
|
|
expect(f.calls.at(-1).operation).toEqual(first.operation);
|
|
}
|
|
});
|
|
it.each(["refs/heads/main", null, undefined])("persists the actual prepared destination (%s)", async (destinationRef) => {
|
|
const f = await fixture();
|
|
const workers: WorkspaceWorkers = { ...f.workers, call: async (...args: Parameters<typeof f.workers.call>) => {
|
|
const result = await f.workers.call(...args);
|
|
return result.kind === "prepared" ? { ...result, ...(destinationRef !== undefined ? { destinationRef } : {}) } : result;
|
|
} };
|
|
const workspace = await preparePluginWorkspace(db, workers, f.input);
|
|
expect(workspace!.metadata?.workspaceProviderDestinationRef).toBe(destinationRef);
|
|
});
|
|
it.each([true, undefined])("wakes parked cleanup when a provider upgrade enables ancestry (%s)", async (supportsMergeAncestry) => {
|
|
const f = await fixture();
|
|
await preparePluginWorkspace(db, f.workers, f.input);
|
|
const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId));
|
|
await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson,
|
|
workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry: false } } }).where(eq(plugins.id, f.pluginId));
|
|
await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId));
|
|
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
|
|
const svc = executionWorkspaceService(db, { pluginWorkerManager: f.workers, workspaceReaperCooldownDays: 0 });
|
|
await svc.sweepTerminalWorkspaces(100);
|
|
expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull();
|
|
expect(f.calls.some(call => call.operation.kind === "inspect")).toBe(false);
|
|
const events: string[] = [];
|
|
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.terminalWorkspaces, (event, hints) => {
|
|
events.push(event);
|
|
if (event === "settled") svc.terminalWorkspaceWorkSettled(hints);
|
|
});
|
|
try {
|
|
await pluginRegistryService(db).update(f.pluginId, { manifest: { ...plugin.manifestJson,
|
|
workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry } } });
|
|
expect(events).toEqual(["intent", "settled"]);
|
|
} finally { unsubscribe(); }
|
|
await svc.sweepTerminalWorkspaces(100);
|
|
expect(f.calls.some(call => call.operation.kind === "inspect")).toBe(true);
|
|
expect(f.calls.some(call => call.operation.kind === "remove")).toBe(true);
|
|
});
|
|
it("wakes cleanup only for provider contract changes and activation", async () => {
|
|
const f = await fixture();
|
|
const registry = pluginRegistryService(db);
|
|
const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId));
|
|
const events: string[] = [];
|
|
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.terminalWorkspaces, event => events.push(event));
|
|
try {
|
|
await registry.update(f.pluginId, { manifest: { ...plugin.manifestJson, description: "Updated description" } });
|
|
await registry.updateStatus(f.pluginId, { status: "disabled" });
|
|
expect(events).toEqual([]);
|
|
await registry.updateStatus(f.pluginId, { status: "ready" });
|
|
expect(events.splice(0)).toEqual(["intent", "settled"]);
|
|
await registry.updateStatus(f.pluginId, { status: "ready" });
|
|
await registry.upsertCompanySettings(f.pluginId, f.companyId, { enabled: false, settingsJson: {} });
|
|
expect(events).toEqual([]);
|
|
await registry.upsertCompanySettings(f.pluginId, f.companyId, { enabled: true, settingsJson: {} });
|
|
expect(events.splice(0)).toEqual(["intent", "settled"]);
|
|
await registry.upsertCompanySettings(f.pluginId, f.companyId, { enabled: true, settingsJson: {} });
|
|
expect(events).toEqual([]);
|
|
await db.transaction(async tx => {
|
|
await pluginRegistryService(tx as unknown as typeof db).update(f.pluginId, {
|
|
manifest: { ...plugin.manifestJson, workspaceProvider: undefined },
|
|
});
|
|
expect(events).toEqual(["intent"]);
|
|
});
|
|
expect(events.splice(0)).toEqual(["intent", "settled"]);
|
|
await registry.updateStatus(f.pluginId, { status: "disabled" });
|
|
await registry.updateStatus(f.pluginId, { status: "ready" });
|
|
await registry.update(f.pluginId, { manifest: { ...plugin.manifestJson, workspaceProvider: undefined, description: "Other plugin" } });
|
|
expect(events).toEqual([]);
|
|
} finally { unsubscribe(); }
|
|
});
|
|
it("keeps storage idle without a settled completion HEAD, even with a recorded PR", async () => {
|
|
const f = await fixture();
|
|
await preparePluginWorkspace(db, f.workers, f.input);
|
|
const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId));
|
|
await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson,
|
|
workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry: false } } }).where(eq(plugins.id, f.pluginId));
|
|
await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId));
|
|
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
|
|
await db.insert(issueWorkProducts).values({ companyId: f.companyId, issueId: f.issueId,
|
|
type: "pull_request", provider: "github", title: "Work", status: "active",
|
|
url: "https://github.com/example/repo/pull/1" });
|
|
const options = { pluginWorkerManager: f.workers, workspaceReaperCooldownDays: 0,
|
|
resolveWorkspaceDelivery: async () => { throw new Error("No delivery lookup without a completion HEAD"); } };
|
|
const svc = executionWorkspaceService(db, options);
|
|
await svc.sweepTerminalWorkspaces(100);
|
|
expect(f.calls.some(call => call.operation.kind === "inspect")).toBe(false);
|
|
expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull();
|
|
svc.terminalWorkspaceWorkSettled(new Set([`issue:${randomUUID()}`]));
|
|
await svc.sweepTerminalWorkspaces(100);
|
|
expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull();
|
|
await executionWorkspaceService(db, options).sweepTerminalWorkspaces(100);
|
|
expect(f.calls.some(call => call.operation.kind === "inspect")).toBe(false);
|
|
});
|
|
it("uses bounded finalization only for opted-in providers, preserving maintenance inspection", async () => {
|
|
const f = await fixture();
|
|
const workspace = (await preparePluginWorkspace(db, f.workers, f.input))!;
|
|
const context = { runId: f.runId, agentId: f.input.agentId };
|
|
expect((await finalizePluginWorkspace(db, f.workers, workspace, context)).kind).toBe("inspection");
|
|
const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId));
|
|
await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson,
|
|
workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, finalizationVersion: 1 } } }).where(eq(plugins.id, f.pluginId));
|
|
const completionIds: string[] = [];
|
|
const call = f.workers.call.bind(f.workers);
|
|
f.workers.call = async (...args) => {
|
|
const p = args[2] as any;
|
|
if (p.operation.kind !== "finalize") return call(...args);
|
|
expect(p.agentId).toBe(context.agentId);
|
|
completionIds.push(p.operationId);
|
|
expect(Date.parse(p.deadline) - Date.now()).toBeLessThanOrEqual(30_000);
|
|
expect(args[3]).toBe(40_000);
|
|
return { kind: "finalized", branchName: workspace.branchName, headSha: "a".repeat(40) };
|
|
};
|
|
expect((await finalizePluginWorkspace(db, f.workers, workspace, context)).kind).toBe("finalized");
|
|
await finalizePluginWorkspace(db, f.workers, workspace, context);
|
|
await finalizePluginWorkspace(db, f.workers, workspace, { ...context, runId: randomUUID() });
|
|
expect(completionIds[1]).toBe(completionIds[0]);
|
|
expect(completionIds[2]).not.toBe(completionIds[0]);
|
|
expect((await inspectPluginWorkspace(db, f.workers, workspace)).kind).toBe("inspection");
|
|
f.workers.call = async () => { throw new Error("sync failed"); };
|
|
await expect(finalizePluginWorkspace(db, f.workers, workspace, context)).rejects.toThrow("incomplete");
|
|
});
|
|
it("keeps the single-connection pool responsive while provider preparation holds its lifecycle lock", async () => {
|
|
const f = await fixture();
|
|
let entered!: () => void, release!: () => void;
|
|
const started = new Promise<void>((resolve) => {
|
|
entered = resolve;
|
|
});
|
|
const resume = new Promise<void>((resolve) => {
|
|
release = resolve;
|
|
});
|
|
const call = f.workers.call.bind(f.workers);
|
|
f.workers.call = async (...args) => {
|
|
if ((args[2] as any).operation.kind === "prepare") {
|
|
entered();
|
|
await resume;
|
|
}
|
|
return call(...args);
|
|
};
|
|
const preparing = preparePluginWorkspace(db, f.workers, f.input);
|
|
await started;
|
|
try {
|
|
await expect(
|
|
projectService(db).updateWorkspace(
|
|
f.input.projectId,
|
|
f.input.projectWorkspaceId,
|
|
{ repoRef: "next" },
|
|
),
|
|
).rejects.toThrow("Workspace lifecycle is busy");
|
|
expect(await db.execute(sql`select 1 as usable`)).toMatchObject([
|
|
{ usable: 1 },
|
|
]);
|
|
} finally {
|
|
release();
|
|
}
|
|
const workspace = await preparing;
|
|
await expect(
|
|
inspectPluginWorkspace(db, f.workers, workspace!),
|
|
).resolves.toMatchObject({ kind: "inspection" });
|
|
});
|
|
it("preserves a scheduled selection through the first provider binding and next admission", async () => {
|
|
const f = await fixture();
|
|
const svc = executionWorkspaceService(db, {
|
|
pluginWorkerManager: f.workers,
|
|
});
|
|
const actor = {
|
|
type: "board" as const,
|
|
source: "local_implicit" as const,
|
|
isInstanceAdmin: true,
|
|
};
|
|
const request = {
|
|
companyId: f.companyId,
|
|
issueId: f.issueId,
|
|
actor,
|
|
selection: { kind: "task_directory" as const },
|
|
expectedBindingRevision: 0,
|
|
requestKey: "during-prepare",
|
|
};
|
|
await svc.selectTaskWorkspace(request);
|
|
await preparePluginWorkspace(db, f.workers, f.input);
|
|
expect(
|
|
await svc.inspectTaskWorkspace(f.companyId, f.issueId, actor),
|
|
).toMatchObject({
|
|
bindingRevision: 2,
|
|
pendingSelection: {
|
|
expectedBindingRevision: 2,
|
|
intent: { request: { expectedBindingRevision: 0 } },
|
|
},
|
|
});
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "succeeded" })
|
|
.where(eq(heartbeatRuns.id, f.runId));
|
|
expect(
|
|
await svc.applyPendingTaskWorkspaceSelection({
|
|
companyId: f.companyId,
|
|
issueId: f.issueId,
|
|
runId: randomUUID(),
|
|
actor,
|
|
}),
|
|
).toBe(true);
|
|
expect(await svc.selectTaskWorkspace(request)).toMatchObject({
|
|
kind: "applied",
|
|
});
|
|
});
|
|
it("does not retire reusable storage for rejected or normalized unchanged source edits", async () => {
|
|
const f = await fixture(),
|
|
workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
await db
|
|
.update(executionWorkspaces)
|
|
.set({ status: "archived", closedAt: new Date() })
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
await removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40));
|
|
expect(
|
|
await projectService(db).updateWorkspace(
|
|
f.input.projectId,
|
|
f.input.projectWorkspaceId,
|
|
{ repoRef: " " },
|
|
),
|
|
).toMatchObject({ repoRef: null });
|
|
expect(
|
|
await projectService(db).updateWorkspace(
|
|
f.input.projectId,
|
|
f.input.projectWorkspaceId,
|
|
{ repoUrl: null },
|
|
),
|
|
).toBeNull();
|
|
const [row] = await db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
expect(row.metadata?.workspaceProviderRetired).not.toBe(true);
|
|
expect(
|
|
await executionWorkspaceService(db, {
|
|
pluginWorkerManager: f.workers,
|
|
}).reopenClosedIsolatedExecutionWorkspaceForIssue({
|
|
workspaceId: workspace!.id,
|
|
issue: {
|
|
id: f.issueId,
|
|
companyId: f.companyId,
|
|
projectId: f.input.projectId,
|
|
},
|
|
actor: { actorType: "user", agentId: null },
|
|
authorizationActor: {
|
|
type: "board",
|
|
source: "local_implicit",
|
|
isInstanceAdmin: true,
|
|
},
|
|
}),
|
|
).toMatchObject({ ok: true, reopened: true });
|
|
});
|
|
it("uses new inspection identities even when the provider caches operation results", async () => {
|
|
const f = await fixture(),
|
|
workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
const call = f.workers.call.bind(f.workers),
|
|
cache = new Map<string, unknown>();
|
|
f.workers.call = async (...args) => {
|
|
const params = args[2] as any;
|
|
if (cache.has(params.operationId))
|
|
return cache.get(params.operationId) as any;
|
|
const result = await call(...args);
|
|
cache.set(params.operationId, result);
|
|
return result;
|
|
};
|
|
expect(
|
|
await inspectPluginWorkspace(db, f.workers, workspace!),
|
|
).toMatchObject({ hasDirtyTrackedFiles: false });
|
|
f.dirty();
|
|
expect(
|
|
await inspectPluginWorkspace(db, f.workers, workspace!),
|
|
).toMatchObject({ hasDirtyTrackedFiles: true });
|
|
expect(cache.size).toBe(2);
|
|
});
|
|
it("retries closed storage cleanup when the pre-delete inspection failed before recording removal", async () => {
|
|
const f = await fixture(),
|
|
workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
await db
|
|
.update(issues)
|
|
.set({ status: "done", completedAt: new Date() })
|
|
.where(eq(issues.id, f.issueId));
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "succeeded" })
|
|
.where(eq(heartbeatRuns.id, f.runId));
|
|
const call = f.workers.call.bind(f.workers);
|
|
let inspections = 0;
|
|
const workers = {
|
|
getWorker: (id: string) =>
|
|
id === f.pluginId
|
|
? { supportedMethods: ["workspaceOperation"] }
|
|
: undefined,
|
|
call: async (...args: Parameters<WorkspaceWorkers["call"]>) => {
|
|
if (
|
|
(args[2] as any).operation.kind === "inspect" &&
|
|
++inspections === 2
|
|
)
|
|
throw new Error("temporary inspection outage");
|
|
return call(...args);
|
|
},
|
|
} as unknown as WorkspaceWorkers;
|
|
const svc = executionWorkspaceService(db, {
|
|
pluginWorkerManager: workers,
|
|
workspaceReaperCooldownDays: 0,
|
|
});
|
|
const sweep = () => svc.sweepTerminalWorkspaces(100);
|
|
expect(await sweep()).toMatchObject({ cleanupFailed: 1 });
|
|
const [failed] = await db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
expect(failed.status).toBe("cleanup_failed");
|
|
expect(failed.metadata?.workspaceProviderRemoval).toBeUndefined();
|
|
expect(await svc.nextTerminalWorkspaceSweepAt()).not.toBeNull();
|
|
await sweep();
|
|
const [removed] = await db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
expect(removed.metadata?.workspaceProviderRemoved).toBe(true);
|
|
expect(removed.status).toBe("archived");
|
|
});
|
|
it("prepares a durable reference without a local checkout or agent allocation", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
expect(workspace).toMatchObject({
|
|
providerType: "plugin",
|
|
cwd: null,
|
|
status: "active",
|
|
});
|
|
expect(f.calls[0].operation.source).not.toHaveProperty("cwd");
|
|
const [task] = await db
|
|
.select()
|
|
.from(issues)
|
|
.where(eq(issues.id, f.issueId));
|
|
expect(task.executionWorkspaceId).toBe(workspace!.id);
|
|
expect(
|
|
await inspectPluginWorkspace(db, f.workers, workspace!),
|
|
).toMatchObject({ remotelyRecoverable: true });
|
|
});
|
|
it("retains the reservation and retries the same operation after a lost response", async () => {
|
|
const f = await fixture();
|
|
f.lose();
|
|
await expect(
|
|
preparePluginWorkspace(db, f.workers, f.input),
|
|
).rejects.toThrow("incomplete");
|
|
const [task] = await db
|
|
.select()
|
|
.from(issues)
|
|
.where(eq(issues.id, f.issueId));
|
|
const result = await preparePluginWorkspace(db, f.workers, {
|
|
...f.input,
|
|
existingWorkspaceId: task.executionWorkspaceId,
|
|
});
|
|
expect(result!.id).toBe(task.executionWorkspaceId);
|
|
expect(f.calls[1].operationId).toBe(f.calls[0].operationId);
|
|
});
|
|
it("reconciles a lost preparation result after the task run stops", async () => {
|
|
const f = await fixture();
|
|
f.lose();
|
|
await expect(
|
|
preparePluginWorkspace(db, f.workers, f.input),
|
|
).rejects.toThrow("incomplete");
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "cancelled" })
|
|
.where(eq(heartbeatRuns.id, f.runId));
|
|
const [task] = await db
|
|
.select()
|
|
.from(issues)
|
|
.where(eq(issues.id, f.issueId));
|
|
const [reservation] = await db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, task.executionWorkspaceId!));
|
|
expect(reservation.providerRef).toBeNull();
|
|
await expect(
|
|
inspectPluginWorkspace(db, f.workers, reservation),
|
|
).resolves.toMatchObject({ remotelyRecoverable: true });
|
|
expect(f.calls[1]).toMatchObject({
|
|
runId: null,
|
|
agentId: null,
|
|
operation: { kind: "prepare" },
|
|
});
|
|
expect(f.calls[1].operationId).toBe(f.calls[0].operationId);
|
|
const [recorded] = await db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, reservation.id));
|
|
expect(recorded.providerRef).not.toBeNull();
|
|
});
|
|
it("reconciles an unfinished reservation before applying PR-only cleanup eligibility", async () => {
|
|
const f = await fixture();
|
|
const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId));
|
|
await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson,
|
|
workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry: false } } }).where(eq(plugins.id, f.pluginId));
|
|
f.lose();
|
|
await expect(preparePluginWorkspace(db, f.workers, f.input)).rejects.toThrow("incomplete");
|
|
await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, f.runId));
|
|
await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId));
|
|
f.dirty();
|
|
await executionWorkspaceService(db, { pluginWorkerManager: f.workers, workspaceReaperCooldownDays: 0 }).sweepTerminalWorkspaces(100);
|
|
expect(f.calls.filter(call => call.operation.kind === "prepare")).toHaveLength(2);
|
|
expect(f.calls.filter(call => call.operation.kind === "inspect")).toHaveLength(1);
|
|
expect(f.calls.some(call => call.operation.kind === "remove")).toBe(false);
|
|
const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId));
|
|
const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, task.executionWorkspaceId!));
|
|
expect(workspace.providerRef).not.toBeNull();
|
|
});
|
|
it("fails closed for a disabled plugin, stopped run, and cross-company source", async () => {
|
|
const f = await fixture();
|
|
await db.insert(pluginCompanySettings).values({
|
|
pluginId: f.pluginId,
|
|
companyId: f.companyId,
|
|
enabled: false,
|
|
});
|
|
await expect(
|
|
preparePluginWorkspace(db, f.workers, f.input),
|
|
).rejects.toThrow("unavailable");
|
|
await db
|
|
.update(pluginCompanySettings)
|
|
.set({ enabled: true })
|
|
.where(eq(pluginCompanySettings.pluginId, f.pluginId));
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "cancelled" })
|
|
.where(eq(heartbeatRuns.id, f.runId));
|
|
await expect(
|
|
preparePluginWorkspace(db, f.workers, f.input),
|
|
).rejects.toThrow();
|
|
expect(f.calls).toHaveLength(0);
|
|
await expect(
|
|
preparePluginWorkspace(db, f.workers, {
|
|
...f.input,
|
|
companyId: randomUUID(),
|
|
}),
|
|
).resolves.toBeNull();
|
|
});
|
|
it("keeps source ownership immutable and preserves bindings through metadata edits", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
const svc = projectService(db);
|
|
await expect(
|
|
svc.update(f.input.projectId, { executionWorkspacePolicy: {} }),
|
|
).rejects.toThrow("storage owner");
|
|
await expect(
|
|
svc.updateWorkspace(f.input.projectId, f.input.projectWorkspaceId, {
|
|
repoUrl: "https://github.com/example/other.git",
|
|
}),
|
|
).rejects.toThrow("repository source");
|
|
await expect(
|
|
svc.removeWorkspace(f.input.projectId, f.input.projectWorkspaceId),
|
|
).rejects.toThrow("repository source");
|
|
const execution = executionWorkspaceService(db);
|
|
const updated = await execution.update(workspace!.id, {
|
|
metadata: {
|
|
note: "caller edit",
|
|
workspaceProvider: { workspaceRef: "foreign" },
|
|
},
|
|
});
|
|
expect(updated!.metadata?.workspaceProvider).toEqual(
|
|
workspace!.metadata?.workspaceProvider,
|
|
);
|
|
await expect(
|
|
execution.update(workspace!.id, { providerType: "local_fs" }),
|
|
).rejects.toThrow("storage plugin");
|
|
});
|
|
it("describes provider removal without promising host cleanup commands", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
await db
|
|
.update(issues)
|
|
.set({ status: "done" })
|
|
.where(eq(issues.id, f.issueId));
|
|
const readiness = await executionWorkspaceService(db, {
|
|
pluginWorkerManager: f.workers,
|
|
}).getCloseReadiness(workspace!.id);
|
|
expect(readiness!.plannedActions).toContainEqual(
|
|
expect.objectContaining({
|
|
kind: "remove_provider_workspace",
|
|
command: null,
|
|
}),
|
|
);
|
|
expect(
|
|
readiness!.plannedActions.some(
|
|
(action) => action.kind === "git_branch_delete",
|
|
),
|
|
).toBe(false);
|
|
});
|
|
it("reopens through the same provider without requiring a live run or a host checkout", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "succeeded" })
|
|
.where(eq(heartbeatRuns.id, f.runId));
|
|
await db
|
|
.update(executionWorkspaces)
|
|
.set({ status: "archived", closedAt: new Date() })
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
const result = await executionWorkspaceService(db, {
|
|
pluginWorkerManager: f.workers,
|
|
}).reopenClosedIsolatedExecutionWorkspaceForIssue({
|
|
workspaceId: workspace!.id,
|
|
issue: {
|
|
id: f.issueId,
|
|
companyId: f.companyId,
|
|
projectId: f.input.projectId,
|
|
},
|
|
actor: { actorType: "user", agentId: null },
|
|
authorizationActor: {
|
|
type: "board",
|
|
source: "local_implicit",
|
|
isInstanceAdmin: true,
|
|
},
|
|
});
|
|
expect(result).toMatchObject({
|
|
ok: true,
|
|
reopened: true,
|
|
generation: 1,
|
|
workspace: { cwd: null, providerType: "plugin" },
|
|
});
|
|
expect(f.calls.at(-1)).toMatchObject({
|
|
runId: null,
|
|
agentId: null,
|
|
operation: { kind: "prepare", branchMode: "existing" },
|
|
});
|
|
expect(f.calls.at(-1).operation.existing.workspaceRef).toBe(
|
|
workspace!.providerRef,
|
|
);
|
|
});
|
|
it("keeps a relocated restoration reference consistent through subsequent removal", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "succeeded" })
|
|
.where(eq(heartbeatRuns.id, f.runId));
|
|
await db
|
|
.update(executionWorkspaces)
|
|
.set({ status: "archived", closedAt: new Date() })
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
await removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40));
|
|
f.relocate();
|
|
const svc = executionWorkspaceService(db, {
|
|
pluginWorkerManager: f.workers,
|
|
});
|
|
const result = await svc.reopenClosedIsolatedExecutionWorkspaceForIssue({
|
|
workspaceId: workspace!.id,
|
|
issue: {
|
|
id: f.issueId,
|
|
companyId: f.companyId,
|
|
projectId: f.input.projectId,
|
|
},
|
|
actor: { actorType: "user", agentId: null },
|
|
authorizationActor: {
|
|
type: "board",
|
|
source: "local_implicit",
|
|
isInstanceAdmin: true,
|
|
},
|
|
});
|
|
expect(result).toMatchObject({ ok: true, reopened: true });
|
|
const [restored] = await db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
expect(restored.providerRef).toBe(`${workspace!.providerRef}:restored`);
|
|
expect(restored.metadata?.workspaceProvider).toMatchObject({
|
|
workspaceRef: restored.providerRef,
|
|
});
|
|
await removePluginWorkspace(db, f.workers, restored, "a".repeat(40));
|
|
expect(f.calls.at(-1).operation).toMatchObject({
|
|
kind: "remove",
|
|
binding: { workspaceRef: restored.providerRef },
|
|
});
|
|
});
|
|
it("reads workspace metadata without provider inspection", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
await db
|
|
.update(issues)
|
|
.set({ status: "done" })
|
|
.where(eq(issues.id, f.issueId));
|
|
f.workers.call = async () => { throw new Error("Storage unavailable"); };
|
|
const svc = executionWorkspaceService(db, {
|
|
pluginWorkerManager: f.workers,
|
|
inspectGitCloseReadiness: async () => {
|
|
throw new Error("Host inspection must not run for provider storage");
|
|
},
|
|
});
|
|
expect(await svc.getById(workspace!.id)).toMatchObject({
|
|
deliveryState: "unknown",
|
|
});
|
|
expect(await svc.getById(workspace!.id)).toMatchObject({
|
|
deliveryState: "unknown",
|
|
});
|
|
expect(
|
|
f.calls.filter((call) => call.operation.kind === "inspect"),
|
|
).toHaveLength(0);
|
|
});
|
|
it("replays a durable removal after a lost result without inspecting the missing checkout", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
f.loseRemoval();
|
|
await expect(
|
|
removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40)),
|
|
).rejects.toThrow("incomplete");
|
|
const inspections = f.calls.filter(
|
|
(call) => call.operation.kind === "inspect",
|
|
).length;
|
|
await removePluginWorkspace(db, f.workers, workspace!, null);
|
|
expect(
|
|
f.calls.filter((call) => call.operation.kind === "inspect"),
|
|
).toHaveLength(inspections);
|
|
const removals = f.calls.filter(
|
|
(call) => call.operation.kind === "remove",
|
|
);
|
|
expect(removals).toHaveLength(2);
|
|
expect(removals[0].operationId).toBe(removals[1].operationId);
|
|
const [row] = await db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
expect(row.metadata).toMatchObject({
|
|
workspaceProviderRemoved: true,
|
|
workspaceProviderRemoval: null,
|
|
});
|
|
});
|
|
it("retires cleaned historical bindings when the storage source changes and refuses later reopen", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
await db
|
|
.update(executionWorkspaces)
|
|
.set({ status: "archived", closedAt: new Date() })
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
await removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40));
|
|
await expect(
|
|
projectService(db).updateWorkspace(
|
|
f.input.projectId,
|
|
f.input.projectWorkspaceId,
|
|
{ repoUrl: "https://github.com/example/new.git" },
|
|
),
|
|
).resolves.not.toBeNull();
|
|
const before = f.calls.length;
|
|
const result = await executionWorkspaceService(db, {
|
|
pluginWorkerManager: f.workers,
|
|
}).reopenClosedIsolatedExecutionWorkspaceForIssue({
|
|
workspaceId: workspace!.id,
|
|
issue: {
|
|
id: f.issueId,
|
|
companyId: f.companyId,
|
|
projectId: f.input.projectId,
|
|
},
|
|
actor: { actorType: "user", agentId: null },
|
|
authorizationActor: {
|
|
type: "board",
|
|
source: "local_implicit",
|
|
isInstanceAdmin: true,
|
|
},
|
|
});
|
|
expect(result).toMatchObject({ ok: false, code: "not_reopenable" });
|
|
expect(f.calls).toHaveLength(before);
|
|
await expect(
|
|
projectService(db).update(f.input.projectId, {
|
|
executionWorkspacePolicy: {},
|
|
}),
|
|
).resolves.not.toBeNull();
|
|
await expect(
|
|
projectService(db).removeWorkspace(
|
|
f.input.projectId,
|
|
f.input.projectWorkspaceId,
|
|
),
|
|
).resolves.not.toBeNull();
|
|
});
|
|
it("keeps sweeping later workspaces when one storage plugin is unavailable", async () => {
|
|
const first = await fixture(),
|
|
second = await fixture();
|
|
await preparePluginWorkspace(db, first.workers, first.input);
|
|
await preparePluginWorkspace(db, second.workers, second.input);
|
|
await db.insert(pluginCompanySettings).values({
|
|
pluginId: first.pluginId,
|
|
companyId: first.companyId,
|
|
enabled: false,
|
|
});
|
|
for (const f of [first, second])
|
|
await db
|
|
.update(issues)
|
|
.set({ status: "done", completedAt: new Date() })
|
|
.where(eq(issues.id, f.issueId));
|
|
const workers = {
|
|
getWorker: (id: string) =>
|
|
[first.pluginId, second.pluginId].includes(id)
|
|
? { supportedMethods: ["workspaceOperation"] }
|
|
: undefined,
|
|
call: (id: string, ...args: unknown[]) =>
|
|
(id === first.pluginId ? first.workers : second.workers).call(
|
|
id,
|
|
...(args as [any, any, any]),
|
|
),
|
|
} as unknown as WorkspaceWorkers;
|
|
await expect(
|
|
executionWorkspaceService(db, {
|
|
pluginWorkerManager: workers,
|
|
workspaceReaperCooldownDays: 0,
|
|
}).sweepTerminalWorkspaces(100),
|
|
).resolves.toBeDefined();
|
|
expect(
|
|
second.calls.some((call) => call.operation.kind === "inspect"),
|
|
).toBe(true);
|
|
});
|
|
it("retries an ambiguous deletion through the existing terminal sweep", async () => {
|
|
const f = await fixture();
|
|
const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId));
|
|
await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson,
|
|
workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry: false } } }).where(eq(plugins.id, f.pluginId));
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
f.loseRemoval();
|
|
await db
|
|
.update(executionWorkspaces)
|
|
.set({ status: "cleanup_failed", closedAt: new Date() })
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
await db
|
|
.update(issues)
|
|
.set({ status: "done", completedAt: new Date() })
|
|
.where(eq(issues.id, f.issueId));
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "cancelled" })
|
|
.where(eq(heartbeatRuns.id, f.runId));
|
|
await expect(
|
|
removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40)),
|
|
).rejects.toThrow("incomplete");
|
|
const workers = {
|
|
getWorker: (id: string) =>
|
|
id === f.pluginId
|
|
? { supportedMethods: ["workspaceOperation"] }
|
|
: undefined,
|
|
call: f.workers.call.bind(f.workers),
|
|
} as unknown as WorkspaceWorkers;
|
|
await executionWorkspaceService(db, {
|
|
pluginWorkerManager: workers,
|
|
workspaceReaperCooldownDays: 0,
|
|
}).sweepTerminalWorkspaces(100);
|
|
const [row] = await db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
expect(row).toMatchObject({
|
|
status: "archived",
|
|
metadata: {
|
|
workspaceProviderRemoved: true,
|
|
workspaceProviderRemoval: null,
|
|
},
|
|
});
|
|
expect(
|
|
f.calls.filter((call) => call.operation.kind === "remove"),
|
|
).toHaveLength(2);
|
|
});
|
|
it("keeps restoration intent durable when reopening loses the provider response", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
await db
|
|
.update(executionWorkspaces)
|
|
.set({ status: "archived", closedAt: new Date() })
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
await removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40));
|
|
f.lose();
|
|
const request = {
|
|
workspaceId: workspace!.id,
|
|
issue: {
|
|
id: f.issueId,
|
|
companyId: f.companyId,
|
|
projectId: f.input.projectId,
|
|
},
|
|
actor: { actorType: "user", agentId: null },
|
|
authorizationActor: {
|
|
type: "board" as const,
|
|
source: "local_implicit" as const,
|
|
isInstanceAdmin: true,
|
|
},
|
|
};
|
|
const svc = executionWorkspaceService(db, {
|
|
pluginWorkerManager: f.workers,
|
|
});
|
|
expect(
|
|
await svc.reopenClosedIsolatedExecutionWorkspaceForIssue(request),
|
|
).toMatchObject({ ok: false, code: "rebuild_failed" });
|
|
const [pending] = await db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, workspace!.id));
|
|
expect(pending.metadata).toMatchObject({
|
|
workspaceProviderRemoved: false,
|
|
workspaceProviderRestoration: { generation: 1 },
|
|
});
|
|
await expect(
|
|
projectService(db).updateWorkspace(
|
|
f.input.projectId,
|
|
f.input.projectWorkspaceId,
|
|
{ repoRef: "new" },
|
|
),
|
|
).rejects.toThrow("storage owner");
|
|
expect(
|
|
await svc.reopenClosedIsolatedExecutionWorkspaceForIssue(request),
|
|
).toMatchObject({ ok: true, generation: 1 });
|
|
});
|
|
it("reconciles a remotely completed restoration before reaping and never reuses an older removal receipt", async () => {
|
|
const f = await fixture();
|
|
const call = f.workers.call.bind(f.workers),
|
|
cache = new Map<string, unknown>();
|
|
let loseRestoreReply = false;
|
|
f.workers.call = async (...args) => {
|
|
const params = args[2] as any;
|
|
if (params.operation.kind === "inspect") return call(...args);
|
|
if (cache.has(params.operationId))
|
|
return cache.get(params.operationId) as any;
|
|
const result = await call(...args);
|
|
cache.set(params.operationId, result);
|
|
if (
|
|
loseRestoreReply &&
|
|
params.operation.kind === "prepare" &&
|
|
params.operation.existing
|
|
) {
|
|
loseRestoreReply = false;
|
|
throw new Error("lost completed restoration reply");
|
|
}
|
|
return result;
|
|
};
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "succeeded" })
|
|
.where(eq(heartbeatRuns.id, f.runId));
|
|
await db
|
|
.update(issues)
|
|
.set({ status: "done", completedAt: new Date() })
|
|
.where(eq(issues.id, f.issueId));
|
|
const svc = executionWorkspaceService(db, {
|
|
pluginWorkerManager: f.workers,
|
|
workspaceReaperCooldownDays: 0,
|
|
});
|
|
await svc.archiveWorkspaceUnderLifecycleLock({
|
|
id: workspace!.id,
|
|
patch: {},
|
|
closedAt: new Date(),
|
|
});
|
|
const read = () =>
|
|
db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, workspace!.id))
|
|
.then((rows) => rows[0]!);
|
|
await removePluginWorkspace(db, f.workers, await read(), "a".repeat(40));
|
|
const firstRemoval = f.calls.find(
|
|
(call) => call.operation.kind === "remove",
|
|
);
|
|
loseRestoreReply = true;
|
|
const restoreEvents: string[] = [];
|
|
const stopRestoreEvents = subscribeDatabaseWork(
|
|
db,
|
|
DELIVERY_QUEUES.terminalWorkspaces,
|
|
(event, hints) => {
|
|
if (hints?.has(`workspace:${workspace!.id}`))
|
|
restoreEvents.push(event);
|
|
},
|
|
);
|
|
try {
|
|
expect(
|
|
await svc.reopenClosedIsolatedExecutionWorkspaceForIssue({
|
|
workspaceId: workspace!.id,
|
|
issue: {
|
|
id: f.issueId,
|
|
companyId: f.companyId,
|
|
projectId: f.input.projectId,
|
|
},
|
|
actor: { actorType: "user", agentId: null },
|
|
authorizationActor: {
|
|
type: "board",
|
|
source: "local_implicit",
|
|
isInstanceAdmin: true,
|
|
},
|
|
}),
|
|
).toMatchObject({ ok: false, code: "rebuild_failed" });
|
|
} finally {
|
|
stopRestoreEvents();
|
|
}
|
|
expect(restoreEvents).toEqual(["intent", "settled"]);
|
|
const pending = await read();
|
|
expect(pending.metadata?.workspaceProviderRestoration).toMatchObject({
|
|
generation: firstRemoval.generation + 1,
|
|
});
|
|
await expect(
|
|
removePluginWorkspace(db, f.workers, pending, "a".repeat(40)),
|
|
).rejects.toThrow("restoration");
|
|
const settled: Array<ReturnType<typeof read>> = [];
|
|
const events: string[] = [];
|
|
const unsubscribe = subscribeDatabaseWork(
|
|
db,
|
|
DELIVERY_QUEUES.terminalWorkspaces,
|
|
(event, hints) => {
|
|
if (!hints?.has(`workspace:${workspace!.id}`)) return;
|
|
events.push(event);
|
|
if (event === "settled") settled.push(read());
|
|
},
|
|
);
|
|
try {
|
|
await svc.sweepTerminalWorkspaces(100);
|
|
} finally {
|
|
unsubscribe();
|
|
}
|
|
expect(events).toEqual(["intent", "settled"]);
|
|
expect(await settled[0]).toMatchObject({
|
|
metadata: {
|
|
workspaceProviderRestoration: null,
|
|
lifecycleGeneration: firstRemoval.generation + 1,
|
|
workspaceProviderRemoved: false,
|
|
},
|
|
});
|
|
const recovered = await read();
|
|
expect(recovered.metadata).toMatchObject({
|
|
workspaceProviderRemoved: false,
|
|
workspaceProviderRestoration: null,
|
|
lifecycleGeneration: firstRemoval.generation + 1,
|
|
});
|
|
await expect(
|
|
projectService(db).updateWorkspace(
|
|
f.input.projectId,
|
|
f.input.projectWorkspaceId,
|
|
{ repoRef: "after-recovery" },
|
|
),
|
|
).rejects.toThrow("storage owner");
|
|
await svc.sweepTerminalWorkspaces(100);
|
|
expect((await read()).metadata?.workspaceProviderRemoved).toBe(true);
|
|
const removals = f.calls.filter(
|
|
(call) => call.operation.kind === "remove",
|
|
);
|
|
expect(removals).toHaveLength(2);
|
|
expect(removals[1].generation).toBe(firstRemoval.generation + 1);
|
|
expect(removals[1].operationId).not.toBe(firstRemoval.operationId);
|
|
});
|
|
it("rejects unsupported provider modes before making a reservation", async () => {
|
|
const f = await fixture();
|
|
await expect(
|
|
preparePluginWorkspace(db, f.workers, {
|
|
...f.input,
|
|
mode: "shared_workspace",
|
|
strategy: "project_primary",
|
|
}),
|
|
).rejects.toThrow("does not support");
|
|
const [task] = await db
|
|
.select()
|
|
.from(issues)
|
|
.where(eq(issues.id, f.issueId));
|
|
expect(task.executionWorkspaceId).toBeNull();
|
|
expect(f.calls).toHaveLength(0);
|
|
});
|
|
it("checks the saved mode when project defaults change on a reused workspace", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
const reused = await preparePluginWorkspace(db, f.workers, {
|
|
...f.input,
|
|
existingWorkspaceId: workspace!.id,
|
|
mode: "shared_workspace",
|
|
strategy: "project_primary",
|
|
});
|
|
expect(reused!.id).toBe(workspace!.id);
|
|
expect(f.calls.at(-1).operation).toMatchObject({
|
|
mode: "isolated_workspace",
|
|
strategy: "git_worktree",
|
|
});
|
|
});
|
|
it("preserves a saved null branch when reuse defaults change", async () => {
|
|
const f = await fixture();
|
|
const [plugin] = await db
|
|
.select()
|
|
.from(plugins)
|
|
.where(eq(plugins.id, f.pluginId));
|
|
await db
|
|
.update(plugins)
|
|
.set({
|
|
manifestJson: {
|
|
...plugin.manifestJson,
|
|
workspaceProvider: {
|
|
modes: ["shared_workspace", "isolated_workspace"],
|
|
strategies: ["project_primary", "git_worktree"],
|
|
},
|
|
},
|
|
})
|
|
.where(eq(plugins.id, f.pluginId));
|
|
const workspace = await preparePluginWorkspace(db, f.workers, {
|
|
...f.input,
|
|
mode: "shared_workspace",
|
|
strategy: "project_primary",
|
|
branchName: null,
|
|
});
|
|
expect(workspace!.branchName).toBeNull();
|
|
await preparePluginWorkspace(db, f.workers, {
|
|
...f.input,
|
|
existingWorkspaceId: workspace!.id,
|
|
branchName: "new-default",
|
|
});
|
|
expect(f.calls.at(-1).operation).toMatchObject({
|
|
mode: "shared_workspace",
|
|
strategy: "project_primary",
|
|
branchName: null,
|
|
});
|
|
});
|
|
it("does not share inspection results across opaque references owned by different providers", async () => {
|
|
const first = await fixture(),
|
|
second = await fixture(first.companyId);
|
|
for (const f of [first, second]) {
|
|
const workers = {
|
|
getWorker: f.workers.getWorker.bind(f.workers),
|
|
call: async (...args: any[]) => {
|
|
const result = await (f.workers.call as any)(...args);
|
|
return result.kind === "prepared"
|
|
? {
|
|
...result,
|
|
binding: { ...result.binding, workspaceRef: "opaque" },
|
|
}
|
|
: result;
|
|
},
|
|
} as unknown as WorkspaceWorkers;
|
|
await preparePluginWorkspace(db, workers, f.input);
|
|
await db
|
|
.update(issues)
|
|
.set({ status: "done", completedAt: new Date() })
|
|
.where(eq(issues.id, f.issueId));
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "cancelled" })
|
|
.where(eq(heartbeatRuns.id, f.runId));
|
|
}
|
|
second.dirty();
|
|
const workers = {
|
|
getWorker: (id: string) =>
|
|
[first.pluginId, second.pluginId].includes(id)
|
|
? { supportedMethods: ["workspaceOperation"] }
|
|
: undefined,
|
|
call: (id: string, ...args: any[]) =>
|
|
(
|
|
(id === first.pluginId ? first.workers : second.workers).call as any
|
|
)(id, ...args),
|
|
} as unknown as WorkspaceWorkers;
|
|
await executionWorkspaceService(db, {
|
|
pluginWorkerManager: workers,
|
|
workspaceReaperCooldownDays: 0,
|
|
}).sweepTerminalWorkspaces(100);
|
|
const [task] = await db
|
|
.select()
|
|
.from(issues)
|
|
.where(eq(issues.id, second.issueId));
|
|
const [retained] = await db
|
|
.select()
|
|
.from(executionWorkspaces)
|
|
.where(eq(executionWorkspaces.id, task.executionWorkspaceId!));
|
|
expect(retained.status).toBe("active");
|
|
expect(
|
|
second.calls.some((call) => call.operation.kind === "inspect"),
|
|
).toBe(true);
|
|
});
|
|
it("never removes a checkout with local-only changes", async () => {
|
|
const f = await fixture();
|
|
const workspace = await preparePluginWorkspace(db, f.workers, f.input);
|
|
f.dirty();
|
|
await expect(
|
|
removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40)),
|
|
).rejects.toThrow("local changes");
|
|
expect(f.calls.some((call) => call.operation.kind === "remove")).toBe(
|
|
false,
|
|
);
|
|
});
|
|
},
|
|
);
|