mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-02 08:28:48 +01:00
Merge qualified workspace storage and external owner integration into PR 18419
Co-Authored-By: Paperclip <noreply@paperclip.ing> * commit '6d54fb819d1ac35ea2eec96b99564a27ea2f54b7': fix(server): schedule only outstanding merge confirmations (#18382) fix(workspaces): inspect storage only when needed (#18410)
This commit is contained in:
commit
376dc050a8
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);
|
||||
});
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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;
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user