mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(workspaces): release an unchanged remote writer without a snapshot and wake its waiters (#20171)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - A remote (sandbox) task workspace has one writer run at a time. The
writer holds the workspace until its files are saved as a snapshot
> - A run that a user comment interrupts right after its claim did no
work. No provider turn started and no command ran in the sandbox
> - Paperclip still kept its sandbox, then took a full snapshot of the
unchanged workspace. On a 2 GB repository this took 3 minutes
> - The next runs failed setup with `workspace_busy` ("ended, but its
workspace save has not settled"). The retry then waited for the
one-minute settling re-check after the save. The user's message waited
more than 4 minutes
> - This pull request releases such a writer at once, at the saved
version it started from, when durable evidence proves that it changed
nothing. It also wakes parked `workspace_busy` waiters when a hold
settles
> - The benefit is that an interrupted run no longer delays the next
message by a full workspace save
## Linked Issues or Issue Description
**What happened?**
A remote-workspace run was claimed (`native.coordinator.claim`). One
second later a board comment interrupted it (`operator_interrupted`). It
made no provider turn and no tool calls. Cancelled-workspace
preservation still retained its sandbox, and workspace recovery took a
full `remote_workspace_snapshot` of the unchanged workspace (3 minutes
for a 2 GB repository). Two follow-up runs were cancelled with
`workspace_busy`. Their retry was due during the save, and it parked for
the one-minute settling re-check after the save finished.
**Expected behavior**
A writer that provably changed nothing releases the workspace at once
and keeps its existing `snapshotId`. The next run is admitted without
`workspace_busy`. When a save does run, waiters start as soon as it
commits.
**Steps to reproduce**
1. Run a task on a remote sandbox workspace that already has a saved
version.
2. Interrupt the run with a board comment right after the claim, before
the provider turn.
3. See the full snapshot operation on the interrupted run and
`workspace_busy` on the follow-up runs.
**Paperclip version or commit**
Staging served revision b43cccbb0. Reproduced on `master` by the new
regression tests.
## What Changed
- `prepareRemoteWorkspace` records `remoteWorkspaceUnchangedBaseline`
(epoch, writer run, saved version) when preparation started from a clean
saved version and added nothing to it: no recovered attachments, no
repository outside the saved version and no workspace migration.
- New `proveRemoteWorkspaceUnchanged`
(`remote-workspace-unchanged-proof.ts`). It uses only durable evidence.
The writer run must have ended without a result, and its owner fence
must be clear. It must have no process fields, no execution, provider or
tool events (the shared `executionStartedEventPredicate`), no native
result, no runner checkpoint rows (the controller writes one before
every runner command), and no provider turn in a checkpoint that names
this run. It must have no workspace operation other than a restore or a
`workspace_finalize`/`workspace_recovery` receipt, so provisioning and
saves count as writes. Every task repository must be in the saved
version. Any missing or ambiguous evidence refuses.
- New `releaseUnchangedRemoteWorkspaceWriter`. It sets the authority to
clean with no writer, keeps `snapshotId`, marks the run reference
`saved`, logs `workspace.snapshot_skipped_unchanged` and sends the same
notifications as a final save. It never waits on a writer run that
another transaction holds (`skip locked`).
- The release applies in three places: the next run's
`reserveRemoteWorkspace`, cancelled/failed-workspace preservation
(`preserveNativeFailedWorkspace`), and `saveRemoteWorkspace` with the
final boundary. The final save returns the same version on a repeated
call, so a recovery that raced the release finishes without a capture.
- The final-save commit and the unchanged release call
`wakeWorkspaceBusyWaitersInTx`. A parked waiter whose holder was that
writer is due at once, not at the one-minute settling re-check.
- `doc/execution-semantics.md` describes the release and the wake.
## Verification
- `pnpm --filter @paperclipai/server exec vitest run
src/services/remote-workspace-unchanged-release.test.ts` (13 tests,
embedded Postgres, real archive capture):
- Staging shape (claim, then interrupt with no provider work): the next
reservation is admitted. No snapshot operation is recorded. The run
reference is `saved` at the old `snapshotId`. Red on `master`: "The
workspace still has an unfinished writer or save operation."
- A later final save, behind the `workspace_finalize` receipt that
production finalizers publish first, returns the same version with no
capture, a repeat returns it again, and a parked waiter becomes due now.
Red on `master`: a new snapshot is captured.
- Negatives that still do a full save: `turn.started`, `provider.event`,
`tool.execution.started`, `native.process_start_requested`, a
provisioning command that wrote into the workspace, a recorded process,
a live owner fence, and a writer that is still running.
- Upgrade safety: no baseline (a writer prepared by the previous
release), a baseline without its saved version, and a baseline with
another schema each keep the full save.
- A full final save wakes the parked waiter. Red on `master`.
- `native-cancelled-workspace-preservation.test.ts`: a new case shows
that a cancelled writer with a baseline and no work is released, with no
retained lease and no recovery action. Red on `master` (the
restore-recovery hold was recorded). The other 44 cases pass unchanged.
- I also ran the related remote-workspace, workspace-recovery, native
preservation and run-dispatch suites locally.
## Risks
- A false "unchanged" verdict would drop unsaved files. The proof
requires a positive baseline from preparation and refuses on any write
record. Only `.paperclip-runtime` (excluded from snapshots) is written
by runner staging before a process starts. A deploy that lands mid-run
keeps the full save, because the proof reads only durable rows.
- Previous-release state: a writer prepared without the baseline (or
with an unknown or partial one) keeps today's full save. An older server
ignores the new metadata key.
- Waiters now start sooner after a final save. Arrival order among
waiters is unchanged (the park check still admits the earliest first).
- Other ways a hold can settle (for example a stop receipt that turns an
unused reservation free) still rely on the one-minute re-check.
## Model Used
- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used in Claude Code
with tool use (shell, file edits and a read-only browser for staging
evidence).
## Checklist
- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
8e5959c759
commit
098f7d370b
6 files changed
+488
-7
No files matched your search
@@ -569,7 +569,24 @@ writer, measured from the time the writer ended. After the cutoff, preparation
|
||||
fails with the unfinished-writer error and leaves the held files unchanged.
|
||||
This wait is also one parked retry, for example after a reassignment stops the
|
||||
writer. It is re-checked every minute while the ended writer's hold settles, is
|
||||
admitted once the hold releases, and stops parking at the same cutoff.
|
||||
admitted once the hold releases, and stops parking at the same cutoff. The
|
||||
writer's final save commit and an unchanged release (below) make these waiters
|
||||
due immediately instead of at the next re-check.
|
||||
|
||||
A writer that ended (failed, cancelled, interrupted or timed out) without
|
||||
changing its files releases the hold at once, with no snapshot. Its workspace
|
||||
keeps the saved version it started from. Preparation records that starting
|
||||
point only when it began from a clean saved version and added nothing to it:
|
||||
no recovered attachments, no repository outside the saved version and no
|
||||
workspace migration. The release requires durable evidence that nothing ran
|
||||
after that point: no execution or provider event, process, runner checkpoint,
|
||||
native result or provider turn in a checkpoint for that run; no workspace
|
||||
operation other than a restore or a finalize or recovery receipt; and no task
|
||||
repository outside the saved version. The writer's owner fence must also be clear. Any missing or ambiguous evidence,
|
||||
including a writer prepared by an earlier release without that starting point,
|
||||
keeps the full save. The next run's reservation, cancelled-workspace
|
||||
preservation and a later final save each apply this release. A repeated save
|
||||
returns the same version.
|
||||
|
||||
A writer stopped mid-run, for example by reassignment, keeps its unsaved files
|
||||
in its stopped sandbox. When the run used a reusable (warm) sandbox, the stop
|
||||
|
||||
@@ -460,6 +460,31 @@ import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
|
||||
it("hands the workspace on at its saved version when the stopped writer never started provider work", async () => {
|
||||
const f = await seed();
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
await db.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspaceUnchangedBaseline: {
|
||||
schema: "paperclip.remote-workspace-unchanged-baseline.v1", epoch: 2, runId: f.runId, snapshotId: f.reference.snapshotId,
|
||||
} } }).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
|
||||
try {
|
||||
await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready;
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(run.resultJson).not.toHaveProperty("workspaceRestoreRecovery");
|
||||
expect(run.runnerProfileJson?.remoteWorkspace).toMatchObject({ state: "saved", snapshotId: f.reference.snapshotId });
|
||||
const [released] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
const authority = readRemoteWorkspaceAuthority(released.metadata)!;
|
||||
expect(authority).toMatchObject({ epoch: 2, snapshotId: f.reference.snapshotId, writerRunId: null, dirty: false });
|
||||
expect(await readRemoteWorkspaceWriterHold(db, { companyId: f.companyId, authority })).toEqual({ kind: "free" });
|
||||
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)))[0])
|
||||
.toMatchObject({ status: "active", leasePolicy: "ephemeral" });
|
||||
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toHaveLength(0);
|
||||
// A later pass finds nothing left to keep.
|
||||
await preserveCancelledNativeWorkspaces(db);
|
||||
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toHaveLength(0);
|
||||
} finally { await worker.stop(); }
|
||||
});
|
||||
|
||||
it("rolls preservation back with its outer transaction", async () => {
|
||||
const f = await seed();
|
||||
await expect(db.transaction(async tx => {
|
||||
|
||||
@@ -18,6 +18,7 @@ import { preserveRemoteWorkspacePreparationFailure } from "./native-workspace-fi
|
||||
import { pinFailedObservedWorkspace } from "./native-observed-workspace-preservation.js";
|
||||
import { hasConfirmedSandboxStopAndRetain } from "../sandbox-stop-and-retain.js";
|
||||
import { preparedRemoteWorkspaceSource } from "../remote-workspace-lifecycle.js";
|
||||
import { releaseUnchangedRemoteWorkspaceWriterById } from "../remote-workspaces.js";
|
||||
|
||||
const reusableLeasePolicies = ["reuse_by_environment", "reuse_by_execution_workspace"];
|
||||
|
||||
@@ -137,6 +138,12 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeo
|
||||
)).limit(1);
|
||||
if (!accepted) return;
|
||||
}
|
||||
// A run that ended without changing its files has nothing to keep: its
|
||||
// workspace is handed on at the saved version it started from.
|
||||
if (!exhaustedAcceptedCopyback && !coordinator.resultId && coordinator.phase === "terminal_failure"
|
||||
&& run.runnerProfileJson?.nativeWorkspaceSync == null
|
||||
&& await releaseUnchangedRemoteWorkspaceWriterById(tx, { companyId: run.companyId,
|
||||
workspaceId: currentReference.workspaceId, runId: run.id })) return;
|
||||
// A stopped reusable sandbox is retained only with its exact stop receipt
|
||||
// while the workspace is still bound to this writer. Otherwise record nothing.
|
||||
if (!exhaustedAcceptedCopyback && source.status !== "active" && reusableLeasePolicies.includes(source.leasePolicy) && (!hasConfirmedSandboxStopAndRetain(source)
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import {
|
||||
executionWorkspaceRepositories, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, nativeRunnerCheckpoints,
|
||||
nativeRunResults, workspaceOperations, type Db,
|
||||
} from "@paperclipai/db";
|
||||
import { remoteWorkspaceRunReferenceSchema, type RemoteWorkspaceAuthority } from "@paperclipai/shared";
|
||||
import { executionStartedEventPredicate } from "./pre-dispatch-execution-evidence.js";
|
||||
import { runOwnerFenceHeldSql } from "./remote-workspace-writer-hold.js";
|
||||
|
||||
/**
|
||||
* A writer's prepared files equal its saved version only while nothing ran in
|
||||
* them. Preparation records that starting point; everything after it leaves a
|
||||
* durable record (an operation, a runner checkpoint, a run event or a
|
||||
* process) before it can write. A run with none of those changed nothing.
|
||||
*/
|
||||
export const UNCHANGED_BASELINE_KEY = "remoteWorkspaceUnchangedBaseline";
|
||||
export const UNCHANGED_BASELINE_SCHEMA = "paperclip.remote-workspace-unchanged-baseline.v1";
|
||||
|
||||
export type UnchangedBaseline = {
|
||||
schema: typeof UNCHANGED_BASELINE_SCHEMA; epoch: number; runId: string; snapshotId: string;
|
||||
/** Set once the writer was released without a new save. */
|
||||
releasedAt?: string;
|
||||
};
|
||||
|
||||
type Workspace = Pick<typeof executionWorkspaces.$inferSelect, "id" | "companyId" | "metadata">;
|
||||
type Repository = { id: string; relativePath: string; repoUrl: string; ref: string | null; pinnedCommit?: string };
|
||||
|
||||
const object = (value: unknown): Record<string, unknown> =>
|
||||
value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
|
||||
/** Missing or malformed fields (an older or partial record) never prove anything. */
|
||||
export function readUnchangedBaseline(metadata: Record<string, unknown> | null | undefined): UnchangedBaseline | null {
|
||||
const value = object(metadata?.[UNCHANGED_BASELINE_KEY]);
|
||||
if (value.schema !== UNCHANGED_BASELINE_SCHEMA || !Number.isSafeInteger(value.epoch) || (value.epoch as number) < 1
|
||||
|| typeof value.runId !== "string" || !value.runId || typeof value.snapshotId !== "string" || !value.snapshotId
|
||||
|| (value.releasedAt !== undefined && typeof value.releasedAt !== "string")) return null;
|
||||
return value as UnchangedBaseline;
|
||||
}
|
||||
|
||||
/** Operations that cannot change the saved files: a restore writes exactly the
|
||||
* saved version, and finalize and recovery rows are receipts that wrap a save
|
||||
* (whose own snapshot operation is still counted). */
|
||||
const FILE_NEUTRAL_OPERATION_PHASES = ["remote_workspace_restore", "workspace_finalize", "workspace_recovery"];
|
||||
/** Terminal states of a run that ended without a result. */
|
||||
const ENDED_WITHOUT_RESULT = ["failed", "cancelled", "interrupted", "timed_out"];
|
||||
|
||||
export type UnchangedProofRefusal =
|
||||
| "baseline_missing" | "authority_mismatch" | "writer_not_ended" | "writer_owner_live" | "process_recorded"
|
||||
| "reference_mismatch" | "provider_session_started" | "execution_recorded" | "result_recorded"
|
||||
| "runner_commanded" | "workspace_operation_recorded" | "repository_unsaved";
|
||||
|
||||
/**
|
||||
* Whether the ended writer of `authority` provably left its files exactly as
|
||||
* the saved version it started from. The caller holds the workspace row lock.
|
||||
* Any missing, ambiguous or older-release evidence refuses; a refusal keeps
|
||||
* today's full save.
|
||||
*/
|
||||
export async function proveRemoteWorkspaceUnchanged(tx: Pick<Db, "select">, input: {
|
||||
workspace: Workspace; authority: RemoteWorkspaceAuthority;
|
||||
/** The caller owns the writer's finalization, so its own fence is expected. */
|
||||
callerOwnsWriter?: boolean;
|
||||
isSavedRepository: (repository: Repository) => boolean;
|
||||
}): Promise<{ proven: true; baseline: UnchangedBaseline } | { proven: false; reason: UnchangedProofRefusal }> {
|
||||
const { workspace, authority } = input;
|
||||
const runId = authority.writerRunId;
|
||||
const baseline = readUnchangedBaseline(workspace.metadata);
|
||||
if (!baseline || baseline.releasedAt || !runId || baseline.runId !== runId) return { proven: false, reason: "baseline_missing" };
|
||||
if (!authority.dirty || !authority.location || authority.epoch !== baseline.epoch || authority.snapshotId !== baseline.snapshotId)
|
||||
return { proven: false, reason: "authority_mismatch" };
|
||||
|
||||
const [run] = await tx.select({ status: heartbeatRuns.status, processPid: heartbeatRuns.processPid,
|
||||
processGroupId: heartbeatRuns.processGroupId, processStartedAt: heartbeatRuns.processStartedAt,
|
||||
sessionIdAfter: heartbeatRuns.sessionIdAfter, runnerProfileJson: heartbeatRuns.runnerProfileJson,
|
||||
ownerFenceHeld: runOwnerFenceHeldSql() }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, runId)));
|
||||
if (!run || !ENDED_WITHOUT_RESULT.includes(run.status)) return { proven: false, reason: "writer_not_ended" };
|
||||
if (!input.callerOwnsWriter && run.ownerFenceHeld) return { proven: false, reason: "writer_owner_live" };
|
||||
if (run.processPid !== null || run.processGroupId !== null || run.processStartedAt !== null || run.sessionIdAfter !== null)
|
||||
return { proven: false, reason: "process_recorded" };
|
||||
const profile = object(run.runnerProfileJson);
|
||||
const reference = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace);
|
||||
if (profile.nativeWorkspaceSync != null || !reference.success || reference.data.state !== "prepared"
|
||||
|| reference.data.workspaceId !== workspace.id || reference.data.epoch !== authority.epoch
|
||||
|| reference.data.snapshotId !== authority.snapshotId || reference.data.leaseId !== authority.location.leaseId
|
||||
|| reference.data.remoteCwd !== authority.location.remoteCwd) return { proven: false, reason: "reference_mismatch" };
|
||||
// A rebound admission checkpoint may name this run; it proves nothing ran
|
||||
// only while it holds no turn, result or terminal state of its own.
|
||||
const checkpoint = object(profile.sessionCheckpoint);
|
||||
if (object(checkpoint.identity).runId === runId && (checkpoint.activeTurnId != null || checkpoint.semanticResult != null
|
||||
|| checkpoint.terminal != null || (checkpoint.terminalTurns != null
|
||||
&& (!Array.isArray(checkpoint.terminalTurns) || checkpoint.terminalTurns.length > 0))))
|
||||
return { proven: false, reason: "provider_session_started" };
|
||||
|
||||
const [event] = await tx.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, workspace.companyId), eq(heartbeatRunEvents.runId, runId),
|
||||
executionStartedEventPredicate())).limit(1);
|
||||
if (event) return { proven: false, reason: "execution_recorded" };
|
||||
const [result] = await tx.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and(
|
||||
eq(nativeRunResults.companyId, workspace.companyId), eq(nativeRunResults.runId, runId))).limit(1);
|
||||
if (result) return { proven: false, reason: "result_recorded" };
|
||||
// The controller writes this checkpoint before every runner command.
|
||||
const [commanded] = await tx.select({ runId: nativeRunnerCheckpoints.runId }).from(nativeRunnerCheckpoints).where(and(
|
||||
eq(nativeRunnerCheckpoints.companyId, workspace.companyId), eq(nativeRunnerCheckpoints.runId, runId))).limit(1);
|
||||
if (commanded) return { proven: false, reason: "runner_commanded" };
|
||||
// Provisioning and saves record their operation before they touch files.
|
||||
const [operation] = await tx.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.heartbeatRunId, runId),
|
||||
sql`(${workspaceOperations.executionWorkspaceId} is null or ${workspaceOperations.executionWorkspaceId} = ${workspace.id})`,
|
||||
sql`${workspaceOperations.phase} not in (${sql.join(FILE_NEUTRAL_OPERATION_PHASES.map(phase => sql`${phase}`), sql`, `)})`)).limit(1);
|
||||
if (operation) return { proven: false, reason: "workspace_operation_recorded" };
|
||||
// A requested repository is cloned before its save is recorded. Only
|
||||
// repositories already in the saved version are proven untouched.
|
||||
const repositories = await tx.select().from(executionWorkspaceRepositories).where(and(
|
||||
eq(executionWorkspaceRepositories.companyId, workspace.companyId),
|
||||
eq(executionWorkspaceRepositories.executionWorkspaceId, workspace.id)));
|
||||
if (repositories.some(row => !row.pinnedCommit || !input.isSavedRepository({ id: row.id, relativePath: row.relativePath,
|
||||
repoUrl: row.repoUrl, ref: row.requestedRef === "HEAD" ? null : row.requestedRef, pinnedCommit: row.pinnedCommit })))
|
||||
return { proven: false, reason: "repository_unsaved" };
|
||||
return { proven: true, baseline };
|
||||
}
|
||||
@@ -0,0 +1,223 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { cp, mkdir, mkdtemp, realpath, rm, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { promisify } from "node:util";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
agents, closeRegisteredClients, companies, createDb, environmentLeases, environments, executionWorkspaces,
|
||||
heartbeatRunEvents, heartbeatRuns, issues, workspaceOperations,
|
||||
} from "@paperclipai/db";
|
||||
import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
|
||||
import type { EnvironmentLease } from "@paperclipai/shared";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { createStorageService } from "../storage/service.js";
|
||||
import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js";
|
||||
import {
|
||||
prepareRemoteWorkspace, readRemoteWorkspaceAuthority, RemoteWorkspaceWriterActiveError, reserveRemoteWorkspace, saveRemoteWorkspace,
|
||||
} from "./remote-workspaces.js";
|
||||
import { UNCHANGED_BASELINE_KEY } from "./remote-workspace-unchanged-proof.js";
|
||||
|
||||
const exec = promisify(execFile);
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
|
||||
describe.skipIf(!support.supported)("remote workspace writer that ended without changing its files", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let root: string;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("workspace-unchanged-release-");
|
||||
db = createDb(database.connectionString);
|
||||
root = await realpath(await mkdtemp(path.join(os.tmpdir(), "workspace-unchanged-release-")));
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "63".repeat(32));
|
||||
vi.stubEnv("DATABASE_URL", database.connectionString);
|
||||
}, 60_000);
|
||||
afterAll(async () => {
|
||||
if (database) await closeRegisteredClients(database.connectionString);
|
||||
await database?.cleanup();
|
||||
if (root) await rm(root, { recursive: true, force: true });
|
||||
vi.unstubAllEnvs();
|
||||
});
|
||||
|
||||
/** A saved workspace, then a follow-up writer B prepared on the same sandbox (staging PAP-3367 shape). */
|
||||
async function preparedFollowUp(options: { runtimeProvisionCommand?: string } = {}) {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Unchanged release", issuePrefix: `U${companyId.slice(0, 6)}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Codie", adapterType: "paperclip_runner" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Daily repair", status: "in_progress", assigneeAgentId: agentId });
|
||||
const [environment] = await db.insert(environments).values({ name: issueId, driver: "sandbox" }).returning();
|
||||
const storage = { ...createStorageService(createLocalDiskStorageProvider(path.join(root, issueId, "objects"))), provider: "s3" as const };
|
||||
const remoteCwd = path.join(root, issueId, "remote");
|
||||
await mkdir(remoteCwd, { recursive: true });
|
||||
const providerLeaseId = randomUUID();
|
||||
const transfer: NonNullable<NonNullable<AdapterSandboxExecutionTarget["runner"]>["syncIn"]> = async operations => {
|
||||
for (const operation of operations) for (const file of operation.files) {
|
||||
await mkdir(path.dirname(file.targetPath), { recursive: true });
|
||||
await cp(file.sourcePath, file.targetPath, { recursive: file.kind === "directory" });
|
||||
}
|
||||
return { operations: operations.map(operation => ({ operationId: operation.operationId,
|
||||
filesTransferred: operation.files.length, bytesTransferred: 0 })) };
|
||||
};
|
||||
let captures = 0;
|
||||
const target = (leaseId: string): AdapterSandboxExecutionTarget => ({
|
||||
kind: "remote", transport: "sandbox", leaseId, remoteCwd, timeoutMs: 30_000,
|
||||
runner: { syncIn: transfer, syncOut: async operations => { captures += 1; return transfer(operations); },
|
||||
execute: async input => {
|
||||
const result = await exec(input.command, input.args ?? [], { cwd: input.cwd ?? remoteCwd,
|
||||
env: { ...process.env, ...input.env }, timeout: input.timeoutMs, maxBuffer: 8 * 1024 * 1024 });
|
||||
return { exitCode: 0, stdout: result.stdout, stderr: result.stderr, signal: null, timedOut: false,
|
||||
pid: process.pid, startedAt: new Date().toISOString() };
|
||||
} },
|
||||
});
|
||||
const writer = async (runId: string, existingWorkspaceId: string | null, runtimeProvisionCommand?: string) => {
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, runtimeMode: "legacy",
|
||||
status: "running", contextSnapshot: { issueId } });
|
||||
const [lease] = await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: runId,
|
||||
environmentId: environment.id, provider: "daytona", providerLeaseId, status: "active" }).returning();
|
||||
const reserved = await reserveRemoteWorkspace({ db, storage, companyId, issueId, runId, agentId,
|
||||
environmentId: environment.id, projectId: null, projectWorkspaceId: null, existingWorkspaceId, source: { kind: "empty" } });
|
||||
const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId,
|
||||
lease: lease as EnvironmentLease, target: target(lease.id), runtimeProvisionCommand });
|
||||
return { lease: prepared.lease, workspaceId: reserved.id, remoteRoot: prepared.remoteRoot };
|
||||
};
|
||||
// Run A does real work and saves it.
|
||||
const runA = randomUUID();
|
||||
const a = await writer(runA, null);
|
||||
await writeFile(path.join(a.remoteRoot, "report.md"), "earlier work\n");
|
||||
const [workspaceA] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, a.workspaceId));
|
||||
const savedA = await saveRemoteWorkspace({ db, storage, workspace: workspaceA, runId: runA, lease: a.lease,
|
||||
target: target(a.lease.id), assertOwned: async () => undefined });
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runA));
|
||||
// Run B is the follow-up: claimed, then interrupted before any provider work.
|
||||
const runB = randomUUID();
|
||||
const b = await writer(runB, a.workspaceId, options.runtimeProvisionCommand);
|
||||
const end = async (patch: Partial<typeof heartbeatRuns.$inferInsert> = {}) => {
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), errorCode: "operator_interrupted", ...patch })
|
||||
.where(eq(heartbeatRuns.id, runB));
|
||||
};
|
||||
const workspace = async () => (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, a.workspaceId)))[0]!;
|
||||
const saveB = async () => saveRemoteWorkspace({ db, storage, workspace: await workspace(), runId: runB, lease: b.lease,
|
||||
target: target(b.lease.id), assertOwned: async () => undefined });
|
||||
const snapshotOperations = () => db.select().from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.heartbeatRunId, runB), eq(workspaceOperations.phase, "remote_workspace_snapshot")));
|
||||
const reserveNext = async () => {
|
||||
const runC = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: runC, companyId, agentId, issueId, runtimeMode: "legacy",
|
||||
status: "running", contextSnapshot: { issueId } });
|
||||
return { runC, reserved: await reserveRemoteWorkspace({ db, storage, companyId, issueId, runId: runC, agentId,
|
||||
environmentId: environment.id, projectId: null, projectWorkspaceId: null, existingWorkspaceId: a.workspaceId, source: { kind: "empty" } }) };
|
||||
};
|
||||
const parkWaiter = async () => {
|
||||
const waiterId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: waiterId, companyId, agentId, issueId, runtimeMode: "legacy",
|
||||
status: "scheduled_retry", scheduledRetryReason: "workspace_busy", scheduledRetryAt: new Date(Date.now() + 5 * 60_000),
|
||||
contextSnapshot: { issueId, workspaceBusyWait: { key: `execution:${a.workspaceId}`, holderRunId: runB,
|
||||
waitingSince: new Date().toISOString(), issueExecutionWorkspaceId: a.workspaceId, issueProjectWorkspaceId: null } } });
|
||||
return async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, waiterId)))[0]!.scheduledRetryAt!;
|
||||
};
|
||||
const event = (eventType: string) => db.insert(heartbeatRunEvents).values({ companyId, runId: runB, agentId,
|
||||
seq: 1, stream: "system", level: "info", eventType });
|
||||
return { companyId, runA, runB, savedA, workspace, end, saveB, snapshotOperations, reserveNext, parkWaiter, event,
|
||||
captures: () => captures };
|
||||
}
|
||||
|
||||
async function expectFullSave(f: Awaited<ReturnType<typeof preparedFollowUp>>) {
|
||||
await expect(f.reserveNext()).rejects.toBeInstanceOf(RemoteWorkspaceWriterActiveError);
|
||||
const before = f.captures();
|
||||
const saved = await f.saveB();
|
||||
expect(saved).not.toHaveProperty("unchanged");
|
||||
expect(saved.snapshotId).not.toBe(f.savedA.snapshotId);
|
||||
expect(f.captures()).toBeGreaterThan(before);
|
||||
expect(await f.snapshotOperations()).toHaveLength(1);
|
||||
}
|
||||
|
||||
it("admits the next run at once, without a save, when the interrupted writer never started provider work", async () => {
|
||||
const f = await preparedFollowUp();
|
||||
await f.end();
|
||||
const { runC, reserved } = await f.reserveNext();
|
||||
const authority = readRemoteWorkspaceAuthority(reserved.metadata);
|
||||
expect(authority).toMatchObject({ writerRunId: runC, snapshotId: f.savedA.snapshotId, dirty: false, epoch: 3 });
|
||||
expect(await f.snapshotOperations()).toEqual([]);
|
||||
const [runB] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runB));
|
||||
expect(runB.runnerProfileJson?.remoteWorkspace).toMatchObject({ state: "saved", snapshotId: f.savedA.snapshotId, epoch: 2 });
|
||||
});
|
||||
|
||||
it("settles a later save of that writer without capturing, and wakes runs parked behind it", async () => {
|
||||
const f = await preparedFollowUp();
|
||||
await f.end();
|
||||
const dueAt = await f.parkWaiter();
|
||||
// Production finalizers publish their wrapper receipt before the save.
|
||||
await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: (await f.workspace()).id,
|
||||
heartbeatRunId: f.runB, phase: "workspace_finalize", status: "running", metadata: { owningService: "native_workspace_finalizer" } });
|
||||
const before = f.captures();
|
||||
const saved = await f.saveB();
|
||||
expect(saved).toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true });
|
||||
expect(f.captures()).toBe(before);
|
||||
expect((await dueAt()).getTime()).toBeLessThanOrEqual(Date.now());
|
||||
// A repeated save (a recovery retry) returns the same version.
|
||||
expect(await f.saveB()).toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true });
|
||||
expect(await f.snapshotOperations()).toEqual([]);
|
||||
const { runC } = await f.reserveNext();
|
||||
expect(readRemoteWorkspaceAuthority((await f.workspace()).metadata)?.writerRunId).toBe(runC);
|
||||
});
|
||||
|
||||
it.each(["turn.started", "provider.event", "tool.execution.started", "native.process_start_requested"])(
|
||||
"keeps the full save after a %s event", async eventType => {
|
||||
const f = await preparedFollowUp();
|
||||
await f.event(eventType);
|
||||
await f.end();
|
||||
await expectFullSave(f);
|
||||
});
|
||||
|
||||
it("keeps the full save when setup wrote into the workspace", async () => {
|
||||
const f = await preparedFollowUp({ runtimeProvisionCommand: "echo provisioned > provisioned.txt" });
|
||||
await f.end();
|
||||
await expectFullSave(f);
|
||||
});
|
||||
|
||||
it("keeps the full save while the writer's owner may still act, and while it is still running", async () => {
|
||||
const f = await preparedFollowUp();
|
||||
await f.end({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) });
|
||||
await expect(f.reserveNext()).rejects.toBeInstanceOf(RemoteWorkspaceWriterActiveError);
|
||||
await db.update(heartbeatRuns).set({ status: "running", controllerLeaseExpiresAt: null, finishedAt: null })
|
||||
.where(eq(heartbeatRuns.id, f.runB));
|
||||
const saved = await f.saveB();
|
||||
expect(saved).not.toHaveProperty("unchanged");
|
||||
expect(await f.snapshotOperations()).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("keeps the full save when a process was recorded", async () => {
|
||||
const f = await preparedFollowUp();
|
||||
await f.end({ processPid: 4242, processStartedAt: new Date() });
|
||||
await expectFullSave(f);
|
||||
});
|
||||
|
||||
it("wakes parked runs when a full final save settles", async () => {
|
||||
const f = await preparedFollowUp();
|
||||
await f.event("turn.started");
|
||||
await f.end();
|
||||
const dueAt = await f.parkWaiter();
|
||||
await f.saveB();
|
||||
expect((await dueAt()).getTime()).toBeLessThanOrEqual(Date.now());
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ label: "a writer prepared by the previous release (no baseline)", baseline: undefined },
|
||||
{ label: "a baseline missing its saved version", baseline: "drop_snapshot" },
|
||||
{ label: "a baseline from another schema", baseline: "other_schema" },
|
||||
] as const)("keeps the full save for $label", async ({ baseline }) => {
|
||||
const f = await preparedFollowUp();
|
||||
const current = await f.workspace();
|
||||
const recorded = current.metadata?.[UNCHANGED_BASELINE_KEY] as Record<string, unknown> | undefined;
|
||||
expect(recorded).toMatchObject({ runId: f.runB, epoch: 2, snapshotId: f.savedA.snapshotId });
|
||||
const { [UNCHANGED_BASELINE_KEY]: _dropped, ...previousRelease } = current.metadata ?? {};
|
||||
const next = baseline === undefined ? previousRelease
|
||||
: baseline === "drop_snapshot" ? { ...previousRelease, [UNCHANGED_BASELINE_KEY]: { ...recorded, snapshotId: undefined } }
|
||||
: { ...previousRelease, [UNCHANGED_BASELINE_KEY]: { ...recorded, schema: "paperclip.remote-workspace-unchanged-baseline.v0" } };
|
||||
await db.update(executionWorkspaces).set({ metadata: next }).where(eq(executionWorkspaces.id, current.id));
|
||||
await f.end();
|
||||
await expectFullSave(f);
|
||||
});
|
||||
});
|
||||
@@ -42,6 +42,9 @@ import { attachPreservedRemoteWorkspaceSnapshots, retainRemoteRecoveryAttachment
|
||||
import { notifyRemoteWorkspaceSnapshots } from "./remote-workspace-snapshot-retention.js";
|
||||
import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js";
|
||||
import { preserveRemoteWorkspacePreparationFailure } from "./native-runtime/native-workspace-finalization-state.js";
|
||||
import { proveRemoteWorkspaceUnchanged, readUnchangedBaseline, UNCHANGED_BASELINE_KEY, UNCHANGED_BASELINE_SCHEMA,
|
||||
type UnchangedBaseline } from "./remote-workspace-unchanged-proof.js";
|
||||
import { wakeWorkspaceBusyWaitersInTx } from "../modules/run-dispatch/index.js";
|
||||
|
||||
export { readRemoteWorkspaceAuthority, type RemoteWorkspaceAuthority, type RemoteWorkspaceSource } from "@paperclipai/shared";
|
||||
export { remoteWorkspaceRepositoryPath } from "./remote-workspace-transport.js";
|
||||
@@ -206,9 +209,9 @@ export async function reserveRemoteWorkspace(input: {
|
||||
selectedSharedSource = selected !== null;
|
||||
}
|
||||
const [selectedRow] = selected ? await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, selected))).for("update") : [];
|
||||
const existing = selectedRow ?? null;
|
||||
let existing = selectedRow ?? null;
|
||||
if (selected && !existing && !input.fresh) throw new Error("remote_workspace_missing");
|
||||
const authority = readRemoteWorkspaceAuthority(existing?.metadata);
|
||||
let authority = readRemoteWorkspaceAuthority(existing?.metadata);
|
||||
if (existing && !authority) throw conflict("Import this existing workspace before enabling remote storage.", { code: "remote_workspace_migration_required" });
|
||||
if (existing && authority) {
|
||||
if (["archived", "cleanup_failed"].includes(existing.status)) throw conflict("Workspace ownership changed.");
|
||||
@@ -236,6 +239,12 @@ export async function reserveRemoteWorkspace(input: {
|
||||
await settleUnverifiedProvision(tx as unknown as Db, operation, input.runId);
|
||||
provenProvisions.push(operation.id);
|
||||
}
|
||||
// An ended writer that provably changed nothing hands the workspace on
|
||||
// at its saved version, so this run is admitted instead of waiting.
|
||||
if (authority.dirty && authority.writerRunId && authority.writerRunId !== input.runId) {
|
||||
const released = await releaseUnchangedRemoteWorkspaceWriter(tx, { workspace: existing, authority, wakeWaiters: false });
|
||||
if (released) ({ workspace: existing, authority } = released);
|
||||
}
|
||||
const hold = await readRemoteWorkspaceWriterHold(tx as unknown as Db, { companyId: input.companyId, authority, runId: input.runId });
|
||||
if (hold.kind === "held") throw new RemoteWorkspaceWriterActiveError(existing.id, hold.writer);
|
||||
if (hold.kind === "unattributed") throw conflict("The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" });
|
||||
@@ -551,6 +560,13 @@ export async function prepareRemoteWorkspace(input: {
|
||||
|| current.location.remoteCwd !== remoteRoot)) throw new Error("remote_workspace_authority_changed");
|
||||
const authority: RemoteWorkspaceAuthority = { ...current, dirty: true, location: { environmentId: input.lease.environmentId!,
|
||||
leaseId: input.lease.id, providerLeaseId: input.lease.providerLeaseId!, remoteCwd: remoteRoot } };
|
||||
// Files published here equal the saved version only when this preparation
|
||||
// started from a clean saved version and added nothing to it. Later writes
|
||||
// (provisioning, runner commands, saves) leave their own durable records.
|
||||
const unchangedBaseline: UnchangedBaseline | undefined = initial.snapshotId && !initial.dirty && migration === null
|
||||
&& current.snapshotId === initial.snapshotId && recoveredFiles.length === 0
|
||||
&& ((input.source ?? initial.source).repositories ?? []).every(repository => isSavedWorkspaceRepository(input.workspace, initial, repository))
|
||||
? { schema: UNCHANGED_BASELINE_SCHEMA, epoch: authority.epoch, runId: input.runId, snapshotId: initial.snapshotId } : undefined;
|
||||
const [lease] = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.id, input.lease.id),
|
||||
eq(environmentLeases.companyId, workspace.companyId))).for("update");
|
||||
if (!lease || lease.status !== "active" || lease.environmentId !== input.lease.environmentId
|
||||
@@ -559,7 +575,7 @@ export async function prepareRemoteWorkspace(input: {
|
||||
const [savedLease] = await tx.update(environmentLeases).set({ metadata: { ...lease.metadata, remoteCwd: remoteRoot, remoteWorkspaceId: workspace.id }, updatedAt: new Date() })
|
||||
.where(eq(environmentLeases.id, lease.id)).returning();
|
||||
const [savedWorkspace] = await tx.update(executionWorkspaces).set({ cwd: remoteRoot, providerRef: remoteRoot,
|
||||
metadata: { ...workspace.metadata, remoteWorkspace: authority,
|
||||
metadata: { ...workspace.metadata, remoteWorkspace: authority, [UNCHANGED_BASELINE_KEY]: unchangedBaseline,
|
||||
// A live copy names the run it is shown to; the next preparation retires
|
||||
// it once that run has succeeded. Retired records stay so a sandbox
|
||||
// restored from an older save is cleaned up too.
|
||||
@@ -700,6 +716,57 @@ async function settleSaveInvocation(input: { db: Db; workspace: Workspace; runId
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Hand a workspace on at its saved version when its writer ended without
|
||||
* changing it (see proveRemoteWorkspaceUnchanged). The caller holds the
|
||||
* workspace row lock. Returns null, changing nothing, when the proof fails or
|
||||
* the writer run is locked by another transaction; its full save still applies.
|
||||
*/
|
||||
export async function releaseUnchangedRemoteWorkspaceWriter(tx: Transaction, input: {
|
||||
workspace: typeof executionWorkspaces.$inferSelect; authority: RemoteWorkspaceAuthority;
|
||||
callerOwnsWriter?: boolean; wakeWaiters?: boolean;
|
||||
}): Promise<{ workspace: typeof executionWorkspaces.$inferSelect; authority: RemoteWorkspaceAuthority } | null> {
|
||||
const { workspace, authority } = input;
|
||||
const runId = authority.writerRunId;
|
||||
if (!runId || !authority.dirty || !authority.snapshotId || !readUnchangedBaseline(workspace.metadata)) return null;
|
||||
// Never wait on a writer run another transaction holds; that owner settles it.
|
||||
const [writer] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, runId))).for("update", { skipLocked: true });
|
||||
if (!writer) return null;
|
||||
const proof = await proveRemoteWorkspaceUnchanged(tx as unknown as Db, { workspace, authority,
|
||||
callerOwnsWriter: input.callerOwnsWriter, isSavedRepository: repository => isSavedWorkspaceRepository(workspace, authority, repository) });
|
||||
if (!proof.proven) {
|
||||
logger.info({ runId, workspaceId: workspace.id, reason: proof.reason }, "Remote workspace keeps its full save");
|
||||
return null;
|
||||
}
|
||||
const at = new Date();
|
||||
const released: RemoteWorkspaceAuthority = { ...authority, dirty: false, writerRunId: null };
|
||||
const [saved] = await tx.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspace: released,
|
||||
[UNCHANGED_BASELINE_KEY]: { ...proof.baseline, releasedAt: at.toISOString() } }, updatedAt: at })
|
||||
.where(eq(executionWorkspaces.id, workspace.id)).returning();
|
||||
await persistRunReference(tx, workspace.companyId, runId, reference(workspace.id, released, "saved"));
|
||||
await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`);
|
||||
await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement);
|
||||
if (input.wakeWaiters !== false) await wakeWorkspaceBusyWaitersInTx(tx as unknown as Db, { companyId: workspace.companyId,
|
||||
runId, contextSnapshot: null, scheduledRetryReason: null, now: at });
|
||||
await persistActivity(tx as unknown as Db, { companyId: workspace.companyId, actorType: "system", actorId: "remote-workspaces",
|
||||
action: "workspace.snapshot_skipped_unchanged", entityType: "execution_workspace", entityId: workspace.id, runId,
|
||||
details: { snapshotId: authority.snapshotId, epoch: authority.epoch } });
|
||||
return { workspace: saved!, authority: released };
|
||||
}
|
||||
|
||||
/** Locks the workspace and releases `runId` as its unchanged ended writer, if provable. */
|
||||
export async function releaseUnchangedRemoteWorkspaceWriterById(tx: Transaction, input: {
|
||||
companyId: string; workspaceId: string; runId: string;
|
||||
}) {
|
||||
const [workspace] = await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId),
|
||||
eq(executionWorkspaces.id, input.workspaceId))).for("update");
|
||||
const authority = readRemoteWorkspaceAuthority(workspace?.metadata);
|
||||
if (!workspace || !authority || authority.writerRunId !== input.runId) return null;
|
||||
return releaseUnchangedRemoteWorkspaceWriter(tx, { workspace, authority });
|
||||
}
|
||||
|
||||
export async function saveRemoteWorkspace(input: {
|
||||
db: Db; workspace: Workspace; runId: string; lease: EnvironmentLease; target: AdapterExecutionTarget | null;
|
||||
assertOwned: () => Promise<void>; storage?: StorageService; boundary?: string;
|
||||
@@ -711,11 +778,28 @@ export async function saveRemoteWorkspace(input: {
|
||||
if (!expected?.location || expected.location.providerLeaseId !== input.lease.providerLeaseId || expected.location.leaseId !== input.lease.id) throw new Error("remote_workspace_save_binding_mismatch");
|
||||
await input.assertOwned();
|
||||
const invocationId = randomUUID();
|
||||
const operation = await input.db.transaction(async tx => {
|
||||
const reserved = await input.db.transaction(async tx => {
|
||||
const issueId = await lockWorkspaceRunParents(tx, input.workspace.companyId, input.runId);
|
||||
const workspace = await currentWorkspace(tx, input.workspace.companyId, input.workspace.id);
|
||||
const current = readRemoteWorkspaceAuthority(workspace.metadata);
|
||||
if (!current || current.epoch !== expected.epoch || (current.writerRunId !== input.runId && current.dirty)) throw new Error("remote_workspace_authority_changed");
|
||||
// A writer that ended without changing its files keeps its saved version.
|
||||
// No capture runs; a repeated call returns the same version.
|
||||
if (boundary === "final") {
|
||||
const baseline = readUnchangedBaseline(workspace.metadata);
|
||||
const replayed = baseline?.releasedAt && baseline.runId === input.runId && baseline.epoch === current.epoch
|
||||
&& baseline.snapshotId === current.snapshotId && !current.dirty && current.writerRunId === null;
|
||||
const released = replayed ? { authority: current } : current.writerRunId === input.runId
|
||||
? await releaseUnchangedRemoteWorkspaceWriter(tx, { workspace, authority: current, callerOwnsWriter: true }) : null;
|
||||
if (released?.authority.snapshotId) {
|
||||
const [snapshot] = await tx.select().from(executionWorkspaceSnapshots).where(and(
|
||||
eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id),
|
||||
eq(executionWorkspaceSnapshots.id, released.authority.snapshotId)));
|
||||
if (!snapshot?.sha256) throw new Error("remote_workspace_snapshot_missing");
|
||||
return { kind: "unchanged" as const, saved: { snapshotId: snapshot.id, generation: snapshot.generation, sha256: snapshot.sha256,
|
||||
savedAt: snapshot.updatedAt.toISOString(), unchanged: true as const } };
|
||||
}
|
||||
}
|
||||
const [previous] = await tx.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, workspace.companyId),
|
||||
eq(workspaceOperations.executionWorkspaceId, workspace.id), eq(workspaceOperations.heartbeatRunId, input.runId),
|
||||
eq(workspaceOperations.phase, "remote_workspace_snapshot"), sql`${workspaceOperations.metadata}->>'boundary' = ${boundary}`,
|
||||
@@ -729,7 +813,7 @@ export async function saveRemoteWorkspace(input: {
|
||||
const [registered] = await tx.update(workspaceOperations).set({ ...reopened, metadata: { ...previous.metadata,
|
||||
saveWork: { schema: SAVE_WORK_SCHEMA, activeIds: [...(work?.activeIds ?? []), invocationId],
|
||||
unknown: work?.unknown ?? true, quiescedAt: null } }, updatedAt: new Date() }).where(eq(workspaceOperations.id, previous.id)).returning();
|
||||
return registered!;
|
||||
return { kind: "registered" as const, operation: registered! };
|
||||
}
|
||||
// The workspace lock serializes generation reservations. Failed/deleted
|
||||
// snapshots still own their generation, and an operation can be committed
|
||||
@@ -753,8 +837,10 @@ export async function saveRemoteWorkspace(input: {
|
||||
saveWork: { schema: SAVE_WORK_SCHEMA, activeIds: [invocationId], unknown: false, quiescedAt: null },
|
||||
exportCleanup: { ...current.location, state: "pending" } },
|
||||
}).returning();
|
||||
return created!;
|
||||
return { kind: "registered" as const, operation: created! };
|
||||
});
|
||||
if (reserved.kind === "unchanged") return reserved.saved;
|
||||
const { operation } = reserved;
|
||||
let captured: Awaited<ReturnType<typeof captureRemoteWorkspaceArchive>> | null = null;
|
||||
let publicationScratch: string | undefined;
|
||||
let failure: { code: string; cause?: string } | undefined;
|
||||
@@ -809,6 +895,9 @@ export async function saveRemoteWorkspace(input: {
|
||||
if (boundary === "final") {
|
||||
await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement);
|
||||
// The hold ends with this commit; runs parked behind it start now.
|
||||
await wakeWorkspaceBusyWaitersInTx(tx as unknown as Db, { companyId: workspace.companyId, runId: input.runId,
|
||||
contextSnapshot: null, scheduledRetryReason: null, now: at });
|
||||
}
|
||||
const receipt = object(workspace.metadata?.remoteWorkspaceSnapshotRepositories);
|
||||
const priorRepositories = receipt.snapshotId === current.snapshotId && Array.isArray(receipt.repositories)
|
||||
|
||||
Reference in new issue
Block a user