Merge qualified workspace storage and external owner integration into PR 18380

Co-Authored-By: Paperclip <noreply@paperclip.ing>

* commit '3e58e12f877d70e5acab272d817ac65b273767c0':
  fix(server): schedule only outstanding merge confirmations (#18382)
  fix(workspaces): inspect storage only when needed (#18410)
This commit is contained in:
DottaandPaperclip committed 2026-10-31 13:50:23 -05:00
commit 56f78fa75e
14 files changed
+317 -183

No files matched your search

@@ -575,7 +575,7 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => {
expect(await service.nextTerminalWorkspaceSweepAt()).toBe(sourceDone.getTime() + 7 * 86_400_000);
});
it("does not use cached display Git state to remove a worktree", async () => {
it("checks current Git state before removal after a metadata read", 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");
@@ -945,7 +945,7 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => {
expect(workspace).toMatchObject({ status: "archived", cleanupReason: "issue_terminal" });
}, 20_000);
it("reads company-scoped execution configuration without live Git inspection", async () => {
it("reads workspace details and company-scoped execution configuration without live Git inspection", async () => {
const seed = await seedAncestryTerminalWorkspace();
const service = executionWorkspaceService(db, {
inspectGitCloseReadiness: async () => { throw new Error("Git inspection unavailable"); },
@@ -970,7 +970,11 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => {
await expect(service.getStoredById(randomUUID(), seed.executionWorkspaceId))
.rejects.toThrow("Task workspace access is no longer available");
expect(await service.getStoredById(seed.companyId, randomUUID())).toBeNull();
await expect(service.getById(seed.executionWorkspaceId)).rejects.toThrow("Git inspection unavailable");
await expect(service.getById(seed.executionWorkspaceId)).resolves.toMatchObject({
id: seed.executionWorkspaceId, deliveryState: "unknown", config,
runtimeServices: [{ id: runtimeServiceId, status: "running" }],
});
await expect(service.getCloseReadiness(seed.executionWorkspaceId)).rejects.toThrow("Git inspection unavailable");
});
it("derives ancestry from the existing comparison and stays unknown for an invalid base", async () => {
@@ -8,6 +8,7 @@ import {
executionWorkspaces,
heartbeatRuns,
issues,
issueWorkProducts,
plugins,
pluginCompanySettings,
} from "@paperclipai/db";
@@ -19,6 +20,7 @@ import {
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 {
@@ -43,6 +45,127 @@ const support = await getEmbeddedPostgresTestSupport();
});
const fixture = (companyId?: string) =>
createPluginWorkspaceFixture(db, companyId);
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 PR-only storage idle until delivery can qualify for fresh inspection", async () => {
const f = await fixture();
const workspace = (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 originalCall = f.workers.call.bind(f.workers);
f.workers.call = async (...args) => {
const value = await originalCall(...args);
return (args[2] as any).operation.kind === "inspect" ? { ...value as object, isMergedIntoBase: null } : value;
};
let clock = new Date();
let state: "open" | "merged" = "open";
let headSha = "b".repeat(40);
let pullRequestReads = 0;
const options = { pluginWorkerManager: f.workers, workspaceReaperCooldownDays: 0,
terminalWorkspaceInspectionBackoffMs: 300_000, now: () => clock,
resolvePullRequestDetails: async () => { pullRequestReads += 1; return { state, headRef: workspace.branchName, headSha }; } };
const svc = executionWorkspaceService(db, options);
const inspections = () => f.calls.filter(call => call.operation.kind === "inspect").length;
await svc.sweepTerminalWorkspaces(100);
expect(inspections()).toBe(0);
expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull();
svc.terminalWorkspaceWorkSettled(new Set([`issue:${randomUUID()}`]));
await svc.sweepTerminalWorkspaces(100);
expect(inspections()).toBe(0);
expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull();
await executionWorkspaceService(db, options).sweepTerminalWorkspaces(100);
expect(inspections()).toBe(0);
await db.insert(issueWorkProducts).values({ companyId: f.companyId, issueId: f.issueId,
type: "pull_request", provider: "github", title: "Unrelated work", status: "active",
url: "https://github.com/other/repository/pull/1" });
svc.terminalWorkspaceWorkSettled(new Set([`issue:${f.issueId}`]));
await svc.sweepTerminalWorkspaces(100);
expect(inspections()).toBe(0);
expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull();
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" });
svc.terminalWorkspaceWorkSettled(new Set([`issue:${f.issueId}`]));
await svc.sweepTerminalWorkspaces(100);
expect(inspections()).toBe(0);
expect(await svc.nextTerminalWorkspaceSweepAt()).not.toBeNull();
const readsBeforeUnrelatedWake = pullRequestReads;
svc.terminalWorkspaceWorkSettled(new Set([`issue:${randomUUID()}`]));
await svc.sweepTerminalWorkspaces(100);
expect(pullRequestReads).toBe(readsBeforeUnrelatedWake);
expect(inspections()).toBe(0);
state = "merged";
clock = new Date(clock.getTime() + 300_001);
await svc.sweepTerminalWorkspaces(100);
expect(inspections()).toBe(1);
expect(f.calls.some(call => call.operation.kind === "remove")).toBe(false);
headSha = "a".repeat(40);
clock = new Date(clock.getTime() + 300_001);
await svc.sweepTerminalWorkspaces(100);
expect(f.calls.some(call => call.operation.kind === "remove")).toBe(true);
});
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))!;
@@ -339,6 +462,24 @@ const support = await getEmbeddedPostgresTestSupport();
.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({
@@ -506,13 +647,14 @@ const support = await getEmbeddedPostgresTestSupport();
binding: { workspaceRef: restored.providerRef },
});
});
it("uses provider inspection for remote workspace display and shares repeated reads", async () => {
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 () => {
@@ -520,14 +662,14 @@ const support = await getEmbeddedPostgresTestSupport();
},
});
expect(await svc.getById(workspace!.id)).toMatchObject({
deliveryState: "merged_by_ancestry",
deliveryState: "unknown",
});
expect(await svc.getById(workspace!.id)).toMatchObject({
deliveryState: "merged_by_ancestry",
deliveryState: "unknown",
});
expect(
f.calls.filter((call) => call.operation.kind === "inspect"),
).toHaveLength(1);
).toHaveLength(0);
});
it("replays a durable removal after a lost result without inspecting the missing checkout", async () => {
const f = await fixture();
@@ -641,6 +783,9 @@ const support = await getEmbeddedPostgresTestSupport();
});
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
@@ -1,47 +0,0 @@
import type { ExecutionWorkspace } from "@paperclipai/shared";
import { afterEach, expect, it, vi } from "vitest";
import { createWorkspaceGitInspectionCache } from "../services/workspace-git-inspection-cache.js";
const workspace = { id: "workspace", companyId: "company", cwd: "/repo", baseRef: "master" } as ExecutionWorkspace;
afterEach(() => vi.useRealTimers());
it("coalesces concurrent display reads and expires after thirty seconds", async () => {
vi.useFakeTimers();
const inspect = vi.fn(async () => ({ dirty: false }));
const read = createWorkspaceGitInspectionCache(inspect);
await Promise.all(Array.from({ length: 100 }, () => read(workspace)));
expect(inspect).toHaveBeenCalledTimes(1);
await read(workspace);
expect(inspect).toHaveBeenCalledTimes(1);
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.
await inspect();
expect(inspect).toHaveBeenCalledTimes(3);
});
it("does not share results across companies, paths, base refs or workspace revisions", async () => {
const inspect = vi.fn(async () => null);
const read = createWorkspaceGitInspectionCache(inspect);
await read(workspace);
await read({ ...workspace, companyId: "other" });
await read({ ...workspace, cwd: "/other" });
await read({ ...workspace, baseRef: "other" });
await read({ ...workspace, updatedAt: new Date() });
expect(inspect).toHaveBeenCalledTimes(5);
});
it("retries failed inspections and bounds retained entries", async () => {
const inspect = vi.fn(async () => null).mockRejectedValueOnce(new Error("failed"));
const read = createWorkspaceGitInspectionCache(inspect);
await expect(read(workspace)).rejects.toThrow("failed");
await read(workspace);
expect(inspect).toHaveBeenCalledTimes(2);
for (let i = 0; i < 256; i++) await read({ ...workspace, id: String(i) });
await read(workspace);
expect(inspect).toHaveBeenCalledTimes(259);
});
+74 -68
View File
@@ -1,6 +1,6 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { effectiveProjectWorkspacePolicy } from "./project-workspace-provider.js";
import { inspectPluginWorkspace, removePluginWorkspace, restorePluginWorkspace, type WorkspaceWorkers } from "./plugin-workspaces.js";
import { inspectPluginWorkspace, removePluginWorkspace, restorePluginWorkspace, pluginWorkspaceSupportsMergeAncestry, type WorkspaceWorkers } from "./plugin-workspaces.js";
import { acquireExecutionWorkspaceLifecycleLock, withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js";
import { measureRunTransaction } from "./execution-transaction-timing.js";
import { measureRunPhase } from "./execution-preparation-timing.js";
@@ -31,7 +31,6 @@ import { retryIdempotentDatabaseOperation } from "../database-retry.js";
import { taskWorkspaceSelectionSchema, type TaskWorkspaceSelection, type TaskWorkspaceIntent } from "@paperclipai/shared";
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 { nativeSessionProviderEvidence } from "./native-runtime/native-session-resume.js";
import { hasWorkspaceRestoreFailure } from "@paperclipai/shared";
import { notifyTerminalWorkspaceWork, TERMINAL_WORKSPACE_RETRY_MS } from "./terminal-workspace-work.js";
@@ -340,10 +339,7 @@ export type ExecutionWorkspaceServiceOptions = {
// Test hook. Runs after compaction moves directories out of the worktree and
// before it re-checks the guard under the lifecycle lock.
afterWorkspaceCompactionMove?: (workspace: ExecutionWorkspaceRow) => Promise<void>;
inspectGitCloseReadiness?: (workspace: ExecutionWorkspace) => Promise<{
git: ExecutionWorkspaceCloseGitReadiness | null;
warnings: string[];
}>;
inspectGitCloseReadiness?: typeof inspectGitCloseReadiness;
};
function parseGitHubRepository(repoUrl: string | null) {
@@ -1370,8 +1366,6 @@ type WorkspaceOverviewIssueRow = WorkspaceOverviewLinkedIssue & {
executionWorkspaceId: string;
};
const inspectGitForDisplay = createWorkspaceGitInspectionCache(inspectGitCloseReadiness);
async function hasPendingTaskWorkspaceRecovery(tx: DbTransaction, input: {
companyId: string; issueId: string; excludeRunId?: string;
}) {
@@ -1431,7 +1425,7 @@ const WORKSPACE_COMPACTION_TRASH_PURGE_INTERVAL_MS = 30 * 60 * 1000;
export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServiceOptions = {}) {
async function inspectWorkspace(workspace: ExecutionWorkspace) {
if (workspace.providerType !== "plugin") return inspectGitCloseReadiness(workspace);
if (workspace.providerType !== "plugin") return (opts.inspectGitCloseReadiness ?? inspectGitCloseReadiness)(workspace);
const [row] = await db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.id, workspace.id), eq(executionWorkspaces.companyId, workspace.companyId)));
if (!row) throw notFound("Workspace is unavailable");
let state;
@@ -1729,13 +1723,6 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
...(selectedWorkspace ? { projectWorkspaceId: selectedWorkspace.projectWorkspaceId } : {}) };
}
const inspectLocalDisplay = opts.inspectGitCloseReadiness
? createWorkspaceGitInspectionCache(opts.inspectGitCloseReadiness)
: inspectGitForDisplay;
const inspectProviderDisplay = createWorkspaceGitInspectionCache(inspectWorkspace);
const inspectDisplay = (workspace: ExecutionWorkspace) => workspace.providerType === "plugin"
? inspectProviderDisplay(workspace)
: inspectLocalDisplay(workspace);
const recoveryActionsSvc = issueRecoveryActionService(db);
const resolvePullRequestDetails = opts.resolvePullRequestDetails ?? createPullRequestMergeDetailsResolver(db);
const now = opts.now ?? (() => new Date());
@@ -1915,14 +1902,14 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
)));
}
async function listDeliveryPullRequestProducts(
workspace: Pick<ExecutionWorkspaceRow, "companyId" | "sourceIssueId">,
async function listDeliveryPullRequestReferences(
workspace: ExecutionWorkspaceRow,
) {
const sourceIssueId = workspace.sourceIssueId;
if (!sourceIssueId) return [];
// A read-only lookup: rebuild it after a transient pooled disconnect.
return retryIdempotentDatabaseOperation(() => db
const products = await retryIdempotentDatabaseOperation(() => db
.select({
id: issueWorkProducts.id,
url: issueWorkProducts.url,
@@ -1939,6 +1926,27 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
))
.orderBy(desc(issueWorkProducts.updatedAt))
.limit(100));
return products.flatMap(product => extractGitHubPullRequestReferences([
product.url, product.externalId, product.title, product.summary,
product.metadata ? JSON.stringify(product.metadata) : null,
])).filter(reference => pullRequestMatchesWorkspaceRepository(reference, workspace));
}
async function cachedDeliveryPullRequestDetails(workspace: ExecutionWorkspaceRow, reference: GitHubPullRequestReference) {
const key = `${workspace.companyId}:${reference.owner.toLowerCase()}/${reference.repo.toLowerCase()}#${reference.number}`;
const cached = pullRequestStateCache.get(key);
if (cached && now().getTime() - cached.checkedAtMs < pullRequestStateCacheTtlMs) return cached.details;
const details = await resolvePullRequestDetails(workspace.companyId, reference);
setBoundedPullRequestCacheEntry(pullRequestStateCache, key, { details, checkedAtMs: now().getTime() });
return details;
}
async function hasMergedDeliveryPullRequest(workspace: ExecutionWorkspaceRow, references: GitHubPullRequestReference[]) {
for (const reference of references) {
const details = await cachedDeliveryPullRequestDetails(workspace, reference);
if (details.state === "merged" && details.headRef === workspace.branchName && details.headSha) return true;
}
return false;
}
async function assessDelivery(
@@ -1970,48 +1978,23 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
: null;
if (sourceIssueTerminal) {
const products = await listDeliveryPullRequestProducts(workspace);
for (const product of products) {
const references = extractGitHubPullRequestReferences([
product.url,
product.externalId,
product.title,
product.summary,
product.metadata ? JSON.stringify(product.metadata) : null,
]);
if (references.length === 0) continue;
for (const reference of references) {
if (!pullRequestMatchesWorkspaceRepository(reference, workspace)) continue;
const key = `${workspace.companyId}:${reference.owner.toLowerCase()}/${reference.repo.toLowerCase()}#${reference.number}`;
const cached = pullRequestStateCache.get(key);
let details;
if (cached && now().getTime() - cached.checkedAtMs < pullRequestStateCacheTtlMs) {
details = cached.details;
} else {
details = await resolvePullRequestDetails(workspace.companyId, reference);
setBoundedPullRequestCacheEntry(
pullRequestStateCache,
key,
{ details, checkedAtMs: now().getTime() },
);
}
if (
details.state === "merged"
&& details.headRef === workspace.branchName
&& details.headSha === workspaceHeadSha
&& workspaceHeadSha !== null
) {
mergedPullRequest = true;
break;
}
if (
details.state === "unknown"
|| (details.state === "merged" && (!details.headRef || !details.headSha || !workspaceHeadSha))
) {
pullRequestStateUnknown = true;
}
for (const reference of await listDeliveryPullRequestReferences(workspace)) {
const details = await cachedDeliveryPullRequestDetails(workspace, reference);
if (
details.state === "merged"
&& details.headRef === workspace.branchName
&& details.headSha === workspaceHeadSha
&& workspaceHeadSha !== null
) {
mergedPullRequest = true;
break;
}
if (
details.state === "unknown"
|| (details.state === "merged" && (!details.headRef || !details.headSha || !workspaceHeadSha))
) {
pullRequestStateUnknown = true;
}
if (mergedPullRequest) break;
}
}
@@ -2059,13 +2042,6 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
}
}
async function hydrateWorkspace(row: ExecutionWorkspaceRow, runtimeServices: WorkspaceRuntimeService[] = []) {
const workspace = toExecutionWorkspace(row, runtimeServices);
const { git } = await inspectDisplay(workspace);
const assessment = await assessDelivery(row, git);
return toExecutionWorkspace(row, runtimeServices, assessment.deliveryState);
}
async function workspaceHasActiveRun(workspace: Pick<ExecutionWorkspaceRow, "id" | "companyId" | "sourceIssueId">) {
const linkedIssues = await db
.select({
@@ -3059,7 +3035,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
projectWorkspaceId: row.projectWorkspaceId,
});
const runtimeServicesByWorkspaceId = await loadEffectiveRuntimeServicesByExecutionWorkspace(db, row.companyId, [row]);
return hydrateWorkspace(
return toExecutionWorkspace(
row,
(runtimeServicesByWorkspaceId.get(row.id) ?? []).map(toRuntimeService),
);
@@ -3691,6 +3667,25 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
else result.skippedActiveRun += 1;
continue;
}
let deliveryReferences: GitHubPullRequestReference[] | null = null;
if (workspace.providerType === "plugin" && !isClosedExecutionWorkspaceStatus(workspace.status)
&& !reopenPending && !workspace.metadata?.workspaceProviderRemoval && !workspace.metadata?.workspaceProviderRestoration) {
try {
if (!(await pluginWorkspaceSupportsMergeAncestry(db, opts.pluginWorkerManager, workspace))) {
deliveryReferences = await listDeliveryPullRequestReferences(workspace);
if (deliveryReferences.length === 0) {
result.skippedUndelivered += 1;
if (compactionPending && compactedTerminalAt.get(workspace.id) !== latestTerminalAt)
terminalWorkspaceDeadlines.set(workspace.id, compactionDueAt);
else terminalWorkspaceDeadlines.delete(workspace.id);
continue;
}
}
} catch {
result.skippedUndelivered += 1;
continue;
}
}
if (opts.terminalWorkspaceInspectionBackoffMs) {
const revision = createHash("sha256").update(JSON.stringify([
workspace.updatedAt, workspace.metadata,
@@ -3707,6 +3702,17 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
nextAt: now().getTime() + opts.terminalWorkspaceInspectionBackoffMs,
});
}
if (deliveryReferences !== null) {
try {
if (!(await hasMergedDeliveryPullRequest(workspace, deliveryReferences))) {
result.skippedUndelivered += 1;
continue;
}
} catch {
result.skippedUndelivered += 1;
continue;
}
}
const workspacePath = readNullableString(workspace.providerRef) ?? readNullableString(workspace.cwd);
const createdByRuntime = workspace.providerType === "git_worktree"
? isRuntimeOwnedGitBranch(workspace.metadata)
+32 -22
View File
@@ -89,6 +89,14 @@ export function pluginRegistryService(db: Db) {
.then((rows) => rows[0] ?? null);
}
async function writeWorkspaceProviderChange<T>(reconsiderWorkspaces: boolean, write: (database: Db) => Promise<T>): Promise<T> {
return db.transaction(async tx => {
await notifyExternalObjectRefresh(tx);
if (reconsiderWorkspaces) await notifyDeliveryWork(tx, DELIVERY_QUEUES.terminalWorkspaces);
return write(tx as unknown as Db);
});
}
async function nextInstallOrder(): Promise<number> {
const result = await db
.select({ maxOrder: sql<number>`coalesce(max(${plugins.installOrder}), 0)` })
@@ -240,15 +248,14 @@ export function pluginRegistryService(db: Db) {
setClause.categories = data.manifest.categories;
}
return db.transaction(async tx => {
await notifyExternalObjectRefresh(tx);
return tx
.update(plugins)
.set(setClause)
.where(eq(plugins.id, id))
.returning()
.then((rows) => rows[0] ?? null);
});
const workspaceProviderChanged = data.manifest !== undefined
&& !isDeepStrictEqual(plugin.manifestJson.workspaceProvider, data.manifest.workspaceProvider);
return writeWorkspaceProviderChange(workspaceProviderChanged, database => database
.update(plugins)
.set(setClause)
.where(eq(plugins.id, id))
.returning()
.then((rows) => rows[0] ?? null));
},
// ----- Status ---------------------------------------------------------
@@ -258,19 +265,18 @@ export function pluginRegistryService(db: Db) {
const plugin = await getById(id);
if (!plugin) throw notFound("Plugin not found");
return db.transaction(async tx => {
await notifyExternalObjectRefresh(tx);
return tx
.update(plugins)
.set({
status: input.status,
lastError: input.lastError ?? null,
updatedAt: new Date(),
})
.where(eq(plugins.id, id))
.returning()
.then((rows) => rows[0] ?? null);
});
const workspaceProviderActivated = Boolean(plugin.manifestJson.workspaceProvider)
&& plugin.status !== "ready" && input.status === "ready";
return writeWorkspaceProviderChange(workspaceProviderActivated, database => database
.update(plugins)
.set({
status: input.status,
lastError: input.lastError ?? null,
updatedAt: new Date(),
})
.where(eq(plugins.id, id))
.returning()
.then((rows) => rows[0] ?? null));
},
// ----- Uninstall / Remove --------------------------------------------
@@ -490,6 +496,10 @@ export function pluginRegistryService(db: Db) {
))
.then((rows) => rows[0] ?? null);
if (plugin.manifestJson.workspaceProvider && existing?.enabled === false && input.enabled === true) {
await notifyDeliveryWork(db, DELIVERY_QUEUES.terminalWorkspaces);
}
if (existing) {
return db
.update(pluginCompanySettings)
+7
View File
@@ -70,6 +70,13 @@ async function provider(
return plugin;
}
export async function pluginWorkspaceSupportsMergeAncestry(db: Db, workers: WorkspaceWorkers | undefined, workspace: Workspace) {
const pluginId = readWorkspaceProviderBinding(workspace.metadata)?.pluginId;
if (!pluginId) return true;
const selected = await provider(db, workers, workspace.companyId, pluginId);
return selected.manifestJson.workspaceProvider?.supportsMergeAncestry !== false;
}
/** No paths or provider credentials are accepted from task callers. */
export async function callWorkspaceProvider(
db: Db,
@@ -1,30 +0,0 @@
import type { ExecutionWorkspace } from "@paperclipai/shared";
/** Short-lived display cache only. Destructive operations must inspect afresh. */
export function createWorkspaceGitInspectionCache<T>(inspect: (workspace: ExecutionWorkspace) => Promise<T>) {
const entries = new Map<string, { expiresAt: number; promise: Promise<T> }>();
return (workspace: ExecutionWorkspace): Promise<T> => {
const key = JSON.stringify([
workspace.companyId, workspace.id, workspace.updatedAt, workspace.providerType,
workspace.providerRef, workspace.cwd, workspace.repoUrl, workspace.baseRef,
workspace.branchName, workspace.metadata,
]);
const now = Date.now();
const existing = entries.get(key);
if (existing && existing.expiresAt > now) return existing.promise;
for (const [candidate, entry] of entries) {
if (entry.expiresAt <= now) entries.delete(candidate);
}
if (entries.size >= 256) entries.delete(entries.keys().next().value!);
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() + 30_000;
return result;
}, (error) => {
if (entries.get(key) === entry) entries.delete(key);
throw error;
});
return entry.promise;
};
}