mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-03 17:51:58 +01:00
fix(heartbeat): wait for inherited remote workspace writers (#18691)
## Thinking Path > - Paperclip manages AI agents and their work. > - A child task can inherit its parent's remote workspace. > - The remote workspace permits one writer until that writer saves its files. > - Reservation reports an active writer as a generic setup failure. > - The existing shared workspace wait does not cover inherited isolated workspaces or a race at reservation. > - This pull request sends verified active writer contention through the existing scheduled retry path. > - The child keeps a live execution path while unfinished files remain protected. ## Linked Issues or Issue Description **What happened?** A native sandbox task inherited an isolated remote workspace from its running parent. Setup failed with `The workspace still has an unfinished writer or save operation.` The child became blocked before provider execution and had no scheduled continuation after the parent saved. **Expected behavior** The child waits for the active writer through the existing workspace retry path. It keeps its workspace binding. A stale writer or a stopped writer with unfinished saves still requires reconciliation. **Steps to reproduce** 1. Start a native sandbox task with remote workspace storage. 2. Create a child that inherits its isolated workspace or operator branch while the parent is still writing. 3. Start the child. On the old code, atomic reservation raises a generic setup failure. **Paperclip version or commit** Observed on `410b0e27015495ab8951768b5b0b94727de33785`. Regression reproduced on source based on `554a770c6f8e48e327e19f5208b575a5ec5bce30`. **Deployment mode** Authenticated managed server, external PostgreSQL, native Codex with a Daytona sandbox. The failed attempt stopped before runtime selection or provider startup. Related: #18611 permits inherited remote bindings. #16371 adds waits for reused shared execution workspaces. This repair handles active writer contention at atomic remote reservation. #17156 proposes a separate change to task workspace inheritance. ## What Changed - Return a typed conflict with the exact active writer from the remote reservation owner. Keep company, binding, access, and writer checks in that transaction. - Translate a recent running writer into the existing heartbeat workspace wait. Keep stale, unknown, stopped, dirty, and uncertain save fences. The writer task lookup uses its primary key. - Preserve the admitted assignee role across preparation so reassignment cannot turn an old owner into a guest retry. - Add PostgreSQL regressions for isolated workspaces, operator branches, native writer identity, unchanged saved authority, admission after save, durable retry, Stop, and reassignment before contention reaches the wait handler. - Update the development guide and workspace verification recipes. ## Verification - Red: both inherited isolated/operator preparation cases fail on the old code. They receive `HttpError(remote_workspace_writer_pending)` instead of a workspace wait. No provider or allocation starts. - Green: those same cases pass after the repair. The assembled public heartbeat cases also pass: one scheduled retry owns the task lock, no sandbox lease is created, the original writer and saved authority remain unchanged, and Stop prevents retry promotion. - Ownership race red/green: changing the assignee during preparation produced `wasIssueAssignee: false` before the review fix. The same regression now preserves its original role. A real transaction fault boundary also reassigns the task after reservation rejects it and before the wait handler runs; promotion cancels that retry with `issue_reassigned`. - Test isolation: an initial combined placement cached the synthetic S3 run-log backend and failed twelve later preparation cases. The assembled native tests now have their own process. The ordered reproduction and all four complete affected suites pass after that change: 171 tests, zero skipped. - Four additional saved-file lifecycle cases pass against disposable Docker PostgreSQL 17 with independent filesystem execution and local object storage. They restore inherited shared, isolated, and operator-branch files on a new allocation and retain the second-writer fence. The other 42 cases in that file were outside the selection. - `pnpm -r typecheck`, `pnpm build`, `node scripts/check-maps.mjs`, and `git diff --check` pass on the final source. - The earlier `pnpm test:run` passed all six chat shards: 1,828 tests. Its remaining general-server phase was interrupted after final-head CI passed. That local command began before the review fix and is not a complete local full-suite pass. - Final head `096235d8fc80033f92d8de1fcfa9835795ba0456`: 54 successful checks/statuses, three skipped checks, and two neutral checks. None is pending or failed. All build, typecheck, general/serialized tests, Runner checks, and eight browser E2E shards pass in [CI](https://github.com/paperclipai/paperclip/actions/runs/41544310900). The PR is mergeable with no conflicts. - Greptile is 5/5 on that exact head. Its ownership finding is resolved and no review thread remains open. Paperclip Review also reports 5/5 with no findings; its check conclusion is neutral. Its earlier status-conflict recovery completed through the ordinary review request path. - Local environment: macOS arm64, Node 26.4.0, pnpm 9.15.4, isolated native PostgreSQL and the disposable PostgreSQL 17 lifecycle fixture. The acquisition boundary is controlled in the focused preparation cases. No live Codex/Daytona replay is claimed: the local environment and test configuration lack the required test credentials. - No deployment, merge, or original task recovery is claimed. ## Risks - Incorrect contention classification could retry a stopped or unverified writer. Only a running writer with a same-company task and recent activity enters the wait path. Stale and terminal dirty cases retain the conflict. - Retries must not start another executor or release unsaved files. Reservation remains atomic. The tests check unchanged writer authority, no allocation before release, one durable retry, retained binding, Stop, and reassignment before the conflict is caught. - No API schema, database migration, provider configuration, model, credential, trust, or approval changes. ## Model Used - OpenAI `gpt-6.1-sol` through Codex. The context window size is not exposed in this session. Used reasoning, tool use, code execution, and browser diagnostics. No sub-agents were used. ## 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 — targeted and chat results are recorded above; the full local command was interrupted - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
ca3f99449e
commit
40ea7a9b89
9 files changed
+254
-5
No files matched your search
@@ -1775,6 +1775,13 @@ The original `sourceIssueId` remains provenance; it does not prevent authorized
|
||||
bound tasks from continuing saved work. Explicitly selected roots follow the
|
||||
same binding contract and the filesystem access boundary below. Task-assignment
|
||||
grants are not filesystem permissions. The single-writer fence still applies.
|
||||
If reservation finds a current running writer, the bound task uses the existing
|
||||
workspace wait and scheduled retry path, including for isolated workspaces and
|
||||
operator branches. Reservation does not acquire another allocation or change
|
||||
the writer's saved files. A stale writer or a stopped writer with unfinished
|
||||
saves still requires reconciliation; contention retries cannot clear that fence.
|
||||
The wait retains the run's admitted assignment role so reassignment during
|
||||
preparation cannot resume the former assignee through the guest-run path.
|
||||
|
||||
Use the focused `remote-workspace-*` and `remote-workspaces.integration` suites
|
||||
for storage, import, retention, and lifecycle regression checks. These checks do
|
||||
|
||||
@@ -81,6 +81,16 @@ An inherited isolated workspace or operator branch can be reserved through the
|
||||
task's current persisted binding. The original source task remains provenance.
|
||||
Reservation checks task/run, binding, trust, and retained source visibility before
|
||||
claiming the single writer; an unbound task cannot claim that workspace.
|
||||
A live writer conflict at atomic reservation uses the existing workspace wait,
|
||||
including inherited isolated workspaces and operator branches. The retry retains
|
||||
the binding and current admission gates; Stop cancels it. Stale or stopped
|
||||
writers with unfinished saves remain fenced. The wait retains the admitted
|
||||
assignment role; reassignment during preparation cancels the old assignee's retry.
|
||||
The
|
||||
[preparation tests](../../server/src/services/heartbeat/workspace-preparation.test.ts)
|
||||
and [heartbeat tests](../../server/src/__tests__/heartbeat-remote-workspace-busy.test.ts)
|
||||
cover reservation and persisted waits with PostgreSQL; live sandbox
|
||||
provider acceptance remains separate.
|
||||
The [reservation tests](../../server/src/services/remote-workspace-reservation.test.ts)
|
||||
exercise that service boundary with real PostgreSQL, including competing children,
|
||||
private sources, foreign companies, and closed records. This is automated evidence,
|
||||
|
||||
@@ -94,6 +94,12 @@ is controlled test input, not a live sandbox qualification.
|
||||
Create a child that inherits a saved isolated workspace or operator branch.
|
||||
Verify it continues the same files after the prior writer saves, including on a
|
||||
new allocation. Two bound children must not acquire concurrent writer authority.
|
||||
Start the child while its parent is still writing. Verify it waits with one
|
||||
scheduled retry and retains its workspace binding without acquiring an allocation.
|
||||
Stop the waiting task and verify that the retry cannot execute. A stale writer
|
||||
or a stopped writer with unfinished saves must remain fenced.
|
||||
Reassign the child during preparation before reservation reports contention;
|
||||
retry promotion must cancel the former assignee's run without starting it.
|
||||
The [reservation regression](../../server/src/services/remote-workspace-reservation.test.ts)
|
||||
and [saved-file lifecycle suite](../../server/src/services/remote-workspaces.integration.test.ts)
|
||||
exercise inheritance, access checks, and file continuity with real PostgreSQL.
|
||||
|
||||
@@ -0,0 +1,115 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, authUsers, companies, companyMemberships, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS,
|
||||
environmentLeases, environments, executionWorkspaces, heartbeatRuns, issues, type Db } from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { RemoteWorkspaceWriterActiveError } from "../services/remote-workspaces.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
|
||||
describe.skipIf(!support.supported)("inherited remote workspace heartbeat wait", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: Db;
|
||||
let home: string;
|
||||
let heartbeat: ReturnType<typeof heartbeatService>;
|
||||
|
||||
beforeAll(async () => {
|
||||
home = await mkdtemp(join(tmpdir(), "heartbeat-remote-workspace-busy-"));
|
||||
database = await startEmbeddedPostgresTestDatabase("heartbeat-remote-workspace-busy-");
|
||||
db = createDb(database.connectionString);
|
||||
vi.stubEnv("PAPERCLIP_HOME", home);
|
||||
vi.stubEnv("DATABASE_URL", database.connectionString);
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "71".repeat(32));
|
||||
vi.stubEnv("PAPERCLIP_STORAGE_PROVIDER", "s3");
|
||||
vi.stubEnv("PAPERCLIP_STORAGE_S3_BUCKET", "unused-reservation-fixture");
|
||||
await instanceSettingsService(db).updateExperimental({ enableIsolatedWorkspaces: true });
|
||||
heartbeat = heartbeatService(db);
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
|
||||
afterAll(async () => {
|
||||
await heartbeat?.drainActiveRunExecutions();
|
||||
await db?.$client.end();
|
||||
await database?.cleanup();
|
||||
vi.unstubAllEnvs();
|
||||
if (home) await rm(home, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ mode: "isolated_workspace", action: "Stop" },
|
||||
{ mode: "operator_branch", action: "Stop" },
|
||||
{ mode: "isolated_workspace", action: "reassignment" },
|
||||
] as const)("persists a $mode writer wait and honors $action", async ({ mode, action }) => {
|
||||
const userId = randomUUID();
|
||||
await db.insert(authUsers).values({ id: userId, name: "Workspace owner", email: `${userId}@example.test`,
|
||||
createdAt: new Date(), updatedAt: new Date() });
|
||||
const [company] = await db.insert(companies).values({ name: "Remote writer wait", issuePrefix: `W${randomUUID().slice(0, 6)}`,
|
||||
requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: userId }).returning();
|
||||
await db.insert(companyMemberships).values({ companyId: company.id, principalType: "user", principalId: userId,
|
||||
status: "active", membershipRole: "owner" });
|
||||
const [environment] = await db.insert(environments).values({ name: `Remote wait ${randomUUID()}`,
|
||||
driver: "sandbox", status: "active", config: { provider: "daytona" } }).returning();
|
||||
const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Native worker", status: "idle",
|
||||
adapterType: "paperclip_runner", defaultEnvironmentId: environment.id,
|
||||
adapterConfig: { provider: "codex", model: "gpt-6.1-sol" }, runtimeConfig: { heartbeat: { maxConcurrentRuns: 2 } } }).returning();
|
||||
const [parent] = await db.insert(issues).values({ companyId: company.id, title: "Running coordinator",
|
||||
status: "in_progress", assigneeAgentId: agent.id, responsibleUserId: userId }).returning();
|
||||
const [writer] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id,
|
||||
nativeIssueId: parent.id, runtimeMode: "native", status: "running", startedAt: new Date(), lastOutputAt: new Date() }).returning();
|
||||
const authority = { schema: "paperclip.remote-workspace.v1" as const, epoch: 3, snapshotId: randomUUID(),
|
||||
source: { kind: "empty" as const }, location: null, writerRunId: writer.id, dirty: true };
|
||||
const [workspace] = await db.insert(executionWorkspaces).values({ companyId: company.id, sourceIssueId: parent.id,
|
||||
name: "Inherited remote workspace", mode, strategyType: "project_primary", providerType: "cloud_sandbox",
|
||||
cwd: `/workspaces/${randomUUID()}`, metadata: { remoteWorkspace: authority } }).returning();
|
||||
await db.update(issues).set({ executionWorkspaceId: workspace.id }).where(eq(issues.id, parent.id));
|
||||
const [child] = await db.insert(issues).values({ companyId: company.id, parentId: parent.id,
|
||||
title: "Use the inherited files", status: "todo", assigneeAgentId: agent.id, responsibleUserId: userId,
|
||||
executionWorkspaceId: workspace.id, executionWorkspacePreference: "reuse_existing", executionWorkspaceSettings: { mode } }).returning();
|
||||
const originalTransaction = db.transaction.bind(db);
|
||||
let reassignmentInjected = false;
|
||||
const transactionSpy = vi.spyOn(db, "transaction").mockImplementation(async (callback, config) => {
|
||||
try { return await originalTransaction(callback, config); }
|
||||
catch (error) {
|
||||
if (action === "reassignment" && error instanceof RemoteWorkspaceWriterActiveError) {
|
||||
const [replacement] = await db.insert(agents).values({ companyId: company.id,
|
||||
name: "New task owner", adapterType: "process" }).returning();
|
||||
await db.update(issues).set({ assigneeAgentId: replacement.id }).where(eq(issues.id, child.id));
|
||||
reassignmentInjected = true;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
try {
|
||||
const run = await heartbeat.wakeup(agent.id, { source: "on_demand", triggerDetail: "manual", reason: "manual",
|
||||
requestedByActorType: "user", requestedByActorId: userId, payload: { issueId: child.id }, contextSnapshot: { issueId: child.id } });
|
||||
expect(run).not.toBeNull();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const [deferred] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id));
|
||||
expect(deferred).toMatchObject({ status: "cancelled", errorCode: "workspace_busy",
|
||||
contextSnapshot: { workspaceBusyDeferredWhileAssignee: true },
|
||||
resultJson: { executionRecovery: { kind: "workspace_wait", providerWorkStarted: false },
|
||||
workspaceBusy: { executionWorkspaceId: workspace.id, holderRunId: writer.id } } });
|
||||
const retries = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run!.id));
|
||||
expect(retries).toHaveLength(1);
|
||||
expect(retries[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "workspace_busy" });
|
||||
const [bound] = await db.select().from(issues).where(eq(issues.id, child.id));
|
||||
expect(bound).toMatchObject({ executionRunId: retries[0].id, executionWorkspaceId: workspace.id });
|
||||
expect(bound.status).not.toBe("blocked");
|
||||
expect((await db.select().from(environmentLeases)).filter(lease => lease.heartbeatRunId === run!.id)).toEqual([]);
|
||||
expect(readRemoteWorkspaceAuthority((await db.select().from(executionWorkspaces)
|
||||
.where(eq(executionWorkspaces.id, workspace.id)))[0].metadata)).toEqual(authority);
|
||||
if (action === "Stop") await heartbeat.cancelRun(retries[0].id);
|
||||
expect((await heartbeat.promoteDueScheduledRetries(new Date(Date.now() + 5 * 60_000))).runIds).not.toContain(retries[0].id);
|
||||
if (action === "reassignment") {
|
||||
expect(reassignmentInjected).toBe(true);
|
||||
expect(await heartbeat.getRun(retries[0].id)).toMatchObject({ status: "cancelled", errorCode: "issue_reassigned" });
|
||||
}
|
||||
} finally { transactionSpy.mockRestore(); }
|
||||
});
|
||||
});
|
||||
@@ -2734,6 +2734,7 @@ export function heartbeatService(
|
||||
task: {
|
||||
issueRef,
|
||||
issueId,
|
||||
wasIssueAssignee: issueContext?.assigneeAgentId === agent.id,
|
||||
context,
|
||||
responsibleUserId,
|
||||
previousSessionParams,
|
||||
|
||||
@@ -43,6 +43,7 @@ import { selectHeartbeatEnvironment } from "./environment-selection.js";
|
||||
import { assertNativeSessionPreparationAuthority } from "../native-runtime/native-execution-lease.js";
|
||||
import { currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js";
|
||||
import { projectService } from "../projects.js";
|
||||
import { WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, WorkspaceBusyDeferral } from "./retries.js";
|
||||
|
||||
const execFile = promisify(execFileCallback);
|
||||
const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_WORKSPACE_PREPARATION_TEST_DATABASE_URL");
|
||||
@@ -107,7 +108,7 @@ describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace
|
||||
const resolver = createHeartbeatWorkspaceResolver(db);
|
||||
const input: HeartbeatWorkspacePreparationInput = {
|
||||
run, agent,
|
||||
task: { issueRef: issue, issueId: issue.id, context: { issueId: issue.id, projectId: project.id }, responsibleUserId: null, previousSessionParams: null },
|
||||
task: { issueRef: issue, issueId: issue.id, wasIssueAssignee: true, context: { issueId: issue.id, projectId: project.id }, responsibleUserId: null, previousSessionParams: null },
|
||||
policy: {
|
||||
trustPreset: { kind: "standard", preset: "standard", boundary: null, sourcePresets: {} },
|
||||
isolatedWorkspacesEnabled: true,
|
||||
@@ -289,6 +290,80 @@ describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace
|
||||
}
|
||||
});
|
||||
|
||||
async function inheritedRemoteFixture(mode: "isolated_workspace" | "operator_branch") {
|
||||
const { input, issue, envOrchestrator } = await fixture();
|
||||
configureRemotePersistence();
|
||||
process.env.PAPERCLIP_SECRETS_MASTER_KEY = "71".repeat(32);
|
||||
input.agent.adapterType = "paperclip_runner";
|
||||
input.environment.selectedEnvironmentForConfig = { ...input.environment.localEnvironment,
|
||||
driver: "sandbox", config: { provider: "daytona" } };
|
||||
input.policy.requestedExecutionWorkspaceMode = mode;
|
||||
input.policy.effectiveExecutionWorkspaceMode = mode;
|
||||
const [parent] = await db.insert(issues).values({ companyId: issue.companyId, projectId: issue.projectId,
|
||||
projectWorkspaceId: issue.projectWorkspaceId, title: "Still running coordinator", assigneeAgentId: input.agent.id }).returning();
|
||||
const [writer] = await db.insert(heartbeatRuns).values({ companyId: issue.companyId, agentId: input.agent.id,
|
||||
nativeIssueId: parent.id, runtimeMode: "native", status: "running", startedAt: new Date(), lastOutputAt: new Date() }).returning();
|
||||
const authority = { schema: "paperclip.remote-workspace.v1" as const, epoch: 3,
|
||||
snapshotId: randomUUID(), source: { kind: "empty" as const }, location: null, writerRunId: writer.id, dirty: true };
|
||||
const [workspace] = await db.insert(executionWorkspaces).values({ companyId: issue.companyId,
|
||||
sourceIssueId: parent.id, projectId: issue.projectId, projectWorkspaceId: issue.projectWorkspaceId,
|
||||
name: "Inherited remote workspace", mode, strategyType: "project_primary", providerType: "cloud_sandbox",
|
||||
cwd: `/workspaces/${randomUUID()}`, metadata: { remoteWorkspace: authority } }).returning();
|
||||
await db.update(issues).set({ executionWorkspaceId: workspace.id, executionWorkspacePreference: "reuse_existing" })
|
||||
.where(eq(issues.id, issue.id));
|
||||
input.task.issueRef = { ...issue, executionWorkspaceId: workspace.id, executionWorkspacePreference: "reuse_existing" };
|
||||
input.reuse.existingExecutionWorkspace = toExecutionWorkspace(workspace);
|
||||
input.reuse.reusableExistingExecutionWorkspace = toExecutionWorkspace(workspace);
|
||||
input.reuse.requestedShouldReuseExisting = true;
|
||||
input.reuse.workspaceReuseRequest = resolveExecutionWorkspaceReuseRequestForIssue({ issueExecutionWorkspaceId: workspace.id,
|
||||
issueExecutionWorkspacePreference: "reuse_existing", existingExecutionWorkspaceStatus: "active" });
|
||||
return { input, issue, envOrchestrator, parent, writer, workspace, authority };
|
||||
}
|
||||
|
||||
it.each(["isolated_workspace", "operator_branch"] as const)("waits for the live writer of an inherited remote %s at reservation", async mode => {
|
||||
const { input, issue, envOrchestrator, parent, writer, workspace, authority } = await inheritedRemoteFixture(mode);
|
||||
const acquire = vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(new Error("writer released; acquisition admitted"));
|
||||
const readAuthority = async () => readRemoteWorkspaceAuthority((await db.select().from(executionWorkspaces)
|
||||
.where(eq(executionWorkspaces.id, workspace.id)))[0].metadata);
|
||||
|
||||
await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({
|
||||
constructor: WorkspaceBusyDeferral, code: "workspace_busy", executionWorkspaceId: workspace.id,
|
||||
holder: { runId: writer.id, issueId: parent.id }, wasIssueAssignee: true,
|
||||
});
|
||||
expect(acquire).not.toHaveBeenCalled();
|
||||
expect(await readAuthority()).toEqual(authority);
|
||||
|
||||
await db.update(heartbeatRuns).set({ lastOutputAt: new Date(Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS - 1_000) })
|
||||
.where(eq(heartbeatRuns.id, writer.id));
|
||||
await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({ status: 409 });
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, writer.id));
|
||||
await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({ status: 409 });
|
||||
expect(await readAuthority()).toEqual(authority);
|
||||
expect(acquire).not.toHaveBeenCalled();
|
||||
|
||||
const saved = { ...authority, writerRunId: null, dirty: false };
|
||||
await db.update(executionWorkspaces).set({ metadata: { remoteWorkspace: saved } }).where(eq(executionWorkspaces.id, workspace.id));
|
||||
await expect(prepareHeartbeatWorkspace(db, input)).rejects.toThrow("writer released; acquisition admitted");
|
||||
expect(acquire).toHaveBeenCalledOnce();
|
||||
expect(await readAuthority()).toEqual({ ...saved, epoch: 4, writerRunId: input.run.id });
|
||||
const [bound] = await db.select().from(issues).where(eq(issues.id, issue.id));
|
||||
expect(bound.executionWorkspaceId).toBe(workspace.id);
|
||||
});
|
||||
|
||||
it("retains admitted ownership when a remote writer wait races with reassignment", async () => {
|
||||
const { input, issue } = await inheritedRemoteFixture("isolated_workspace");
|
||||
const [replacement] = await db.insert(agents).values({ companyId: issue.companyId,
|
||||
name: "New task owner", adapterType: "process" }).returning();
|
||||
input.controllerLease.assertOwned = async () => {
|
||||
await db.update(issues).set({ assigneeAgentId: replacement.id }).where(eq(issues.id, issue.id));
|
||||
};
|
||||
await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({
|
||||
code: "workspace_busy", wasIssueAssignee: true,
|
||||
});
|
||||
const [current] = await db.select().from(issues).where(eq(issues.id, issue.id));
|
||||
expect(current.assigneeAgentId).toBe(replacement.id);
|
||||
});
|
||||
|
||||
it("rejects automatic Core services before remote reservation or acquisition", async () => {
|
||||
const { input, envOrchestrator } = await fixture();
|
||||
configureRemotePersistence();
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
import { assertEphemeralWorkspaceExecution, readRemoteWorkspaceAuthority, reserveRemoteWorkspace, shouldUseRemoteWorkspaces } from "../remote-workspaces.js";
|
||||
import { assertEphemeralWorkspaceExecution, readRemoteWorkspaceAuthority, RemoteWorkspaceWriterActiveError, reserveRemoteWorkspace, shouldUseRemoteWorkspaces } from "../remote-workspaces.js";
|
||||
import { WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, WorkspaceBusyDeferral } from "./retries.js";
|
||||
import { WORKSPACE_BUSY_RETRY_REASON } from "../../modules/run-dispatch/index.js";
|
||||
import { readFreshWorkspaceReadiness } from "../fresh-workspace-continuation.js";
|
||||
import { assertNativeSessionPreparationAuthority } from "../native-runtime/native-execution-lease.js";
|
||||
import { resolveRemoteWorkspaceSource } from "./remote-workspace-source.js";
|
||||
@@ -105,6 +107,7 @@ export interface HeartbeatWorkspacePreparationInput {
|
||||
executionWorkspacePreference: string | null;
|
||||
}) | null;
|
||||
issueId: string | null;
|
||||
wasIssueAssignee: boolean;
|
||||
context: Record<string, unknown>;
|
||||
responsibleUserId: string | null;
|
||||
previousSessionParams: Record<string, unknown> | null;
|
||||
@@ -1007,6 +1010,12 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace
|
||||
expectedBindingRevision: remoteFreshBinding!.revision },
|
||||
recoverySourceRunIds: [remoteFresh.sourceRunId],
|
||||
} : {}),
|
||||
}).catch(error => {
|
||||
if (!(error instanceof RemoteWorkspaceWriterActiveError)
|
||||
|| error.writer.lastActivityAt.getTime() < Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS) throw error;
|
||||
throw new WorkspaceBusyDeferral({ holder: error.writer, executionWorkspaceId: error.workspaceId,
|
||||
deferralAttempt: run.scheduledRetryReason === WORKSPACE_BUSY_RETRY_REASON ? run.scheduledRetryAttempt ?? 0 : 0,
|
||||
wasIssueAssignee: input.task.wasIssueAssignee });
|
||||
});
|
||||
persistedExecutionWorkspace = await executionWorkspacesSvc.getStoredById(agent.companyId, reserved.id);
|
||||
if (!persistedExecutionWorkspace?.cwd) throw new Error("remote_workspace_reservation_missing");
|
||||
|
||||
@@ -15,12 +15,13 @@ import {
|
||||
import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
|
||||
import { getStorageService, type StorageService } from "../storage/index.js";
|
||||
import { loadConfig } from "../config.js";
|
||||
import { conflict, forbidden, unprocessable } from "../errors.js";
|
||||
import { conflict, forbidden, HttpError, unprocessable } from "../errors.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { createGitRemoteAuthProvider } from "./git-credentials.js";
|
||||
import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js";
|
||||
import { assertExecutionWorkspaceRecordAccess, assertTaskWorkspaceExecutionAccess } from "./task-workspace-access.js";
|
||||
import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
import { canonicalExecutionUuid } from "./execution-identity.js";
|
||||
import { persistActivity } from "./activity-log.js";
|
||||
import {
|
||||
captureRemoteWorkspaceArchive, cleanupRemoteWorkspaceArchive, prepareRemoteWorkspaceDirectory, prepareRemoteRepository, remoteWorkspaceRepositoryPath,
|
||||
@@ -130,6 +131,17 @@ async function currentWorkspace(tx: Transaction, companyId: string, workspaceId:
|
||||
return workspace;
|
||||
}
|
||||
|
||||
export class RemoteWorkspaceWriterActiveError extends HttpError {
|
||||
constructor(
|
||||
readonly workspaceId: string,
|
||||
readonly writer: {
|
||||
runId: string; agentId: string; issueId: string; issueIdentifier: string | null; lastActivityAt: Date;
|
||||
},
|
||||
) {
|
||||
super(409, "The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" });
|
||||
}
|
||||
}
|
||||
|
||||
export async function reserveRemoteWorkspace(input: {
|
||||
db: Db; companyId: string; issueId: string; runId: string; agentId: string; environmentId: string;
|
||||
projectId: string | null; projectWorkspaceId: string | null; existingWorkspaceId: string | null;
|
||||
@@ -182,8 +194,21 @@ export async function reserveRemoteWorkspace(input: {
|
||||
)).limit(1);
|
||||
if (uncertain) throw conflict("The prior provisioning operation needs verification before this workspace can run again.", { code: "remote_workspace_provision_unverified" });
|
||||
if (authority.writerRunId && authority.writerRunId !== input.runId) {
|
||||
const [prior] = await tx.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(and(
|
||||
const [prior] = await tx.select({ status: heartbeatRuns.status, agentId: heartbeatRuns.agentId,
|
||||
issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId,
|
||||
contextIssueId: sql<string | null>`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId')`,
|
||||
lastOutputAt: heartbeatRuns.lastOutputAt, startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt,
|
||||
}).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, authority.writerRunId)));
|
||||
if (prior?.status === "running") {
|
||||
const writerIssueId = prior.nativeIssueId ?? prior.issueId ?? prior.contextIssueId;
|
||||
const [writerIssue] = writerIssueId ? await tx.select({ id: issues.id, identifier: issues.identifier }).from(issues)
|
||||
.where(and(eq(issues.companyId, input.companyId), eq(issues.id, canonicalExecutionUuid(sql`${writerIssueId}::text`)))) : [];
|
||||
if (writerIssue) throw new RemoteWorkspaceWriterActiveError(existing.id, {
|
||||
runId: authority.writerRunId, agentId: prior.agentId, issueId: writerIssue.id, issueIdentifier: writerIssue.identifier,
|
||||
lastActivityAt: prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt,
|
||||
});
|
||||
}
|
||||
const leases = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId),
|
||||
eq(environmentLeases.heartbeatRunId, authority.writerRunId)));
|
||||
const unusedReservation = !authority.dirty && prior && ["failed", "cancelled", "interrupted", "timed_out", "succeeded"].includes(prior.status)
|
||||
|
||||
@@ -1412,7 +1412,8 @@ describe("automatic retained workspace recovery", () => {
|
||||
const localEnvironment = await environmentService(db).ensureLocalEnvironment(f.companyId);
|
||||
const prepared = await prepareHeartbeatWorkspace(db, {
|
||||
run: successor, agent,
|
||||
task: { issueRef: issue, issueId: f.issueId, context: { issueId: f.issueId }, responsibleUserId: "local-board",
|
||||
task: { issueRef: issue, issueId: f.issueId, wasIssueAssignee: issue.assigneeAgentId === agent.id,
|
||||
context: { issueId: f.issueId }, responsibleUserId: "local-board",
|
||||
previousSessionParams: null, freshWorkspace: readiness },
|
||||
policy: { trustPreset: { kind: "standard", preset: "standard", boundary: null, sourcePresets: {} },
|
||||
isolatedWorkspacesEnabled: false, effectiveExecutionWorkspaceMode: "shared_workspace",
|
||||
|
||||
Reference in new issue
Block a user