mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix(server): recognize recoverable result finalization during upgrades (#19651)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work.
> - Native execution saves an accepted result before it finishes
workspace export and status projection.
> - The upgrade report currently recognizes only a live execution in the
observed phase.
> - A durable accepted result can therefore block an upgrade after its
execution lease is released.
> - Existing finalization owners already resume that saved work with
task binding, physical worker stop proof, and an atomic owner claim.
> - This pull request lets the report inspect those existing contracts
without taking authority.
> - Qualified remote work can continue through a normal server
replacement.
## Linked Issues or Issue Description
Refs #18037 and #19596.
**What happened?**
An accepted remote native result remains in result acceptance, workspace
finalization, retry, or committed projection. The upgrade report marks
it blocking because it is no longer a result-less observed execution
with a live execution lease.
**Expected behavior**
Report a saved result as recoverable only when the existing replacement
owners can resume it. Preserve all task, result, contract, remote
workspace, provider lease, physical owner, decision, and workspace
barrier checks. Keep unsupported or unknown work blocking.
**Steps to reproduce**
Run the new PostgreSQL post-result readiness fixture. It persists a real
encrypted Runner header and an accepted terminal result with a bound
remote workspace. The unchanged report returns one blocker.
**Paperclip version or commit**
Regression base: ee6c101854. Related shutdown recovery is in #19596.
## What Changed
- Delegate saved-result eligibility to the native finalization owner.
- Require the accepted terminal result and contract, current task
binding, remote authority, retained provider lease or successful
workspace receipt, and a supported physical worker identity.
- Require an applied decision and assessment for committed projection.
Keep finalizer leases beyond the supported outage budget blocking.
- Reuse the existing result recovery binding with selected metadata. Do
not load complete run or result payloads for the new observation.
- Keep real signed stop verification, owner compare-and-swap, retry
deadlines, and actual finalization with their existing owners.
- Share a two-second scheduling budget across saved-result inspections,
leaving unchecked candidates blocking after the budget expires.
- Add PostgreSQL, encrypted-header, signed stop-proof, competing-owner,
projection, and independent-work cases. Clarify the operator contract.
## Verification
- The valid accepted-result fixture reproduced the original blocking
report, then passed with this change.
- The initial eight post-result cases passed against PostgreSQL. They
include a real Ed25519 stop receipt, the existing verifier, a competing
PostgreSQL ownership claim, one copyback, and receipt reuse.
- Seventeen focused existing PostgreSQL shutdown and workspace ownership
cases passed.
- A broader five-file run passed 91 cases. Its one failure was an
invalid negative fixture: PostgreSQL correctly refused a contract change
protected by the accepted-result foreign key. The test now asserts that
refusal.
- A subsequent standalone run timed out in PostgreSQL bootstrap before
tests ran. The final run used the existing shared PostgreSQL command and
isolated per-suite template databases: all 92 tests passed across five
files (159.95 seconds). It covers both readiness suites, platform stop
proof, workspace finalization, and graceful shutdown. That scope passed
on `ce4e3995b6`. A follow-up restricts provider leases to the active
status and adds released/unknown-status controls. The new fixture now
uses the existing PostgreSQL bootstrap timeout constant. The affected
eight-case rerun passed on `850859b44d`. The explicit configured-stack
guard also passed its six platform-proof cases. Broad local source
typecheck still has stale borrowed dependency declarations; fresh CI
supplies that coverage.
- Review's locked-query regression reproduced RED: three saved-result
candidates took 6.081 seconds. With the shared scheduling budget, both
readiness suites passed all 16 cases (49.36 seconds) on final code
committed as `70b0b7b8f7`. The lock case leaves all three candidates
blocking, releases its connection, and observes all three as recoverable
after the lock is released.
- Greptile reviewed final `70b0b7b8f7` at 5/5 with no open threads.
Current-head CI typecheck, build, tests, and native review are pending.
No hosted replacement or enrollment acceptance is claimed.
## Risks
A false positive could select an unsafe upgrade. Missing evidence and
read errors remain blocking. Readiness does not stop admission, detach
an agent, acquire a lease, request a stop receipt, or authorize
takeover. The replacement still verifies physical stop and rechecks
current task ownership before copyback. Saved results retaining a
previous controller boot remain conservatively blocking, including
finalizers resumed by a later boot. This PR does not change enrollment
policy or migrate data.
## Model Used
OpenAI GPT-6.1, model ID `gpt-6.1-sol`, with reasoning, tool use, and
code execution. The host did not expose the context-window size.
## 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 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
- [ ] 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:
DottaandPaperclip authored and GitHub committed
1 parent
374a918d67
commit
bee0f0e2fa
7 files changed
+425
-15
No files matched your search
@@ -154,12 +154,22 @@ The checkpoint owner authenticates/decrypts that header and delegates version an
|
||||
inspection to Runner; the identity must match the run, session and runner. Ordinary
|
||||
sandbox runs bind their workspace identity; plugin tasks bind their active dispatched
|
||||
environment lease and need no workspace-sync descriptor.
|
||||
Accepted terminal results can also qualify through their existing finalization
|
||||
owner. This requires the exact accepted result and contract, a remote workspace,
|
||||
and its retained provider lease or successful workspace barrier. Committed work
|
||||
also needs its applied decision and assessment. Unknown physical owners and
|
||||
finalizer leases beyond the five-minute budget remain blockers. The observation
|
||||
does not request stop proof or take a lease. Copyback still requires the old
|
||||
worker's signed stop receipt and an atomic owner claim. Committed projections
|
||||
use the existing accepted-result and decision receipts.
|
||||
Legacy/local executions, missing identities/checkpoints, foreign/expired owners,
|
||||
in-flight admission, and an incomplete census prevent a positive report. It
|
||||
returns aggregate counts only. The request does not stop admission, detach a
|
||||
runner, or authorize new work. Ordinary drain reads do not perform this SQL.
|
||||
The explicit probe uses a read-only transaction, a two-second statement timeout,
|
||||
and at most 1,001 run rows. Header reads are capped at 256 KiB each and 8 MiB total;
|
||||
The explicit probe uses read-only transactions, a two-second statement timeout,
|
||||
and at most 1,001 run rows. Post-result inspections share a two-second scheduling
|
||||
budget; after it expires, unchecked candidates remain blocking. An in-flight
|
||||
inspection retains its SQL statement timeout. Header reads are capped at 256 KiB each and 8 MiB total;
|
||||
oversized, malformed, legacy-format or local evidence falls back to the ordinary drain.
|
||||
This bounded header probe does not validate every journal record or prove a live sandbox. This is an observation; the existing shutdown and
|
||||
startup owners still perform the actual handoff and enforce Stop and fencing.
|
||||
|
||||
@@ -0,0 +1,262 @@
|
||||
import { generateKeyPairSync, randomUUID, sign } from "node:crypto";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { eq, inArray, sql } from "drizzle-orm";
|
||||
import { agents, companies, completionContracts, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, environments, executionWorkspaces,
|
||||
heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, workAssessments, workspaceOperations } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js";
|
||||
import { currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js";
|
||||
import { nativeRunnerCheckpointWriter } from "../native-runtime/native-runner-checkpoint.js";
|
||||
import { readUpgradeReadiness } from "./upgrade-readiness.js";
|
||||
import { currentWorkspaceWorker } from "../native-runtime/native-workspace-platform-proof.js";
|
||||
import { withNativeWorkspaceFinalizationOwnership } from "../native-runtime/native-workspace-finalization-ownership.js";
|
||||
import { resumeNativeWorkspaceFinalization } from "../native-runtime/native-workspace-finalizer.js";
|
||||
import { projectCommittedRun } from "../native-runtime/native-run-projection.js";
|
||||
|
||||
describe("post-result upgrade readiness", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
const runs: string[] = [];
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("post-result-upgrade-");
|
||||
db = createDb(database.connectionString);
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "a".repeat(64));
|
||||
for (const [key, value] of Object.entries({ RAILWAY_PROJECT_ID: "project", RAILWAY_ENVIRONMENT_ID: "environment",
|
||||
RAILWAY_SERVICE_ID: "service", RAILWAY_DEPLOYMENT_ID: "deployment", PAPERCLIP_CLOUD_STACK_ID: "stack",
|
||||
PAPERCLIP_CLOUD_API_ORIGIN: "https://cloud.example", PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "fixture-token" })) vi.stubEnv(key, value);
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
afterEach(async () => { if (runs.length) await db.update(heartbeatRuns).set({ status: "succeeded" }).where(inArray(heartbeatRuns.id, runs)); });
|
||||
afterAll(async () => { await db?.$client.end(); await database?.cleanup(); vi.unstubAllEnvs(); });
|
||||
|
||||
async function seed() {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
|
||||
const contractId = randomUUID(), workspaceId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID();
|
||||
const resultId = randomUUID(), runnerInstanceId = randomUUID(), nativeSessionId = randomUUID();
|
||||
const controller = await currentNativeControllerIdentity();
|
||||
const leaseOwner = `execution:${randomUUID()}`;
|
||||
const remote = { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId, providerLeaseId: "sandbox",
|
||||
remoteCwd: "/workspace", epoch: 1, state: "prepared", snapshotId: null };
|
||||
const authority = { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId: null, source: { kind: "empty" },
|
||||
location: { environmentId, leaseId, providerLeaseId: "sandbox", remoteCwd: "/workspace" }, writerRunId: runId, dirty: true };
|
||||
await db.insert(companies).values({ id: companyId, name: "Upgrade", issuePrefix: `P${companyId.slice(0, 6)}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Runner", adapterType: "paperclip_runner" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Accepted work", status: "in_progress", assigneeAgentId: agentId });
|
||||
await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1,
|
||||
schemaVersion: "paperclip.completion-contract.v1", policyVersion: "fixture", risk: "standard",
|
||||
completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: {},
|
||||
canonicalSha256: "a".repeat(64), createdByActorType: "system", createdByActorId: "test" });
|
||||
await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId, mode: "isolated_workspace",
|
||||
strategyType: "local_directory", name: "Remote", cwd: "/workspace", providerType: "cloud_sandbox", metadata: { remoteWorkspace: authority } });
|
||||
await db.insert(environments).values({ id: environmentId, name: `Sandbox ${environmentId}`, driver: "sandbox" });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId, status: "running",
|
||||
runtimeMode: "native", nativePhase: "observed", runnerInstanceId, nativeSessionId, completionContractId: contractId,
|
||||
processPid: process.pid, processStartedAt: controller.processStartedAt,
|
||||
runnerProfileJson: { remoteWorkspace: remote, nativeExecutionInput: { binding: { executionWorkspaceId: workspaceId } } },
|
||||
contextSnapshot: { executionWorkspaceId: workspaceId } });
|
||||
runs.push(runId);
|
||||
await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId, executionWorkspaceId: workspaceId }).where(eq(issues.id, issueId));
|
||||
await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, executionWorkspaceId: workspaceId,
|
||||
issueId, heartbeatRunId: runId, provider: "daytona", providerLeaseId: "sandbox", leasePolicy: "retain_on_failure" });
|
||||
await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "observed", leaseOwner,
|
||||
leaseExpiresAt: new Date(Date.now() + 60_000), controllerBootId: controller.bootId, controllerPid: controller.pid,
|
||||
controllerProcessStartedAt: controller.processStartedAt, controllerGeneration: 1 });
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "post-result-header-"));
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: root, identity: { runId, runnerInstanceId,
|
||||
normalizedSessionId: nativeSessionId, environmentLeaseId: workspaceId, turnId: "turn", itemId: "item" },
|
||||
expectedRunnerVersion: "0.3.0", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
const writer = nativeRunnerCheckpointWriter(db, { companyId, issueId, runId, leaseOwner, bootId: controller.bootId,
|
||||
authorityRunId: runId, generation: 1 });
|
||||
core.setCheckpointPersistence(writer);
|
||||
try { core.issueBootstrapTicket(); await core.flushCheckpoint(); }
|
||||
finally { await core.stop(); await rm(root, { recursive: true, force: true }); }
|
||||
await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId,
|
||||
serverFingerprint: resultId, schemaStatus: "accepted", resultJson: { terminal: { runTerminalState: "succeeded" } }, canonicalSha256: "b".repeat(64) });
|
||||
await db.update(nativeRunFinalizations).set({ phase: "workspace_finalizing", resultId }).where(eq(nativeRunFinalizations.runId, runId));
|
||||
return { companyId, issueId, runId, agentId, contractId, resultId, workspaceId, environmentId, leaseId, remote, authority, controller,
|
||||
probe: () => readUpgradeReadiness(db, { activeRunIds: new Set([runId]), counters: () => ({ activeRuns: 1, pendingWakes: 0, draining: false }) }) };
|
||||
}
|
||||
|
||||
it("certifies accepted remote work without changing its lease, result or task admission", async () => {
|
||||
const f = await seed();
|
||||
const before = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(await f.probe()).toMatchObject({ ready: true, recoverableRuns: 1, blockingRuns: 0, draining: false });
|
||||
expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual(before);
|
||||
});
|
||||
|
||||
it.each(["result_accepted", "workspace_finalizing", "retryable_failure"])("retains the real recovery owner for %s", async phase => {
|
||||
const f = await seed();
|
||||
await db.update(nativeRunFinalizations).set({ phase, leaseOwner: null, leaseExpiresAt: null,
|
||||
...(phase === "retryable_failure" ? { failureCode: "native_workspace_sync_out_failed", nextAttemptAt: new Date(Date.now() + 60_000) } : {}),
|
||||
}).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(await f.probe()).toMatchObject({ ready: true, recoverableRuns: 1 });
|
||||
});
|
||||
|
||||
it("requires the accepted terminal result, current task, remote lease and exact physical owner", async () => {
|
||||
const f = await seed();
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const profile = run.runnerProfileJson!;
|
||||
const owner = { token: randomUUID(), controllerBootId: f.controller.bootId, hostname: os.hostname(),
|
||||
pid: f.controller.pid, processStartedAt: f.controller.processStartedAt.toISOString(), worker: currentWorkspaceWorker() };
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect((await f.probe()).ready).toBe(true);
|
||||
for (const physical of [null, { ...owner, token: "" }, { ...owner, worker: undefined }, { ...owner, controllerBootId: randomUUID() },
|
||||
{ ...owner, hostname: "another-host" }, { ...owner, processStartedAt: new Date(0).toISOString() },
|
||||
{ ...owner, worker: { ...owner.worker, deploymentId: "another-deployment" } }, { ...owner, worker: { ...owner.worker, extra: true } }]) {
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeWorkspaceFinalizationOwner: physical ?? {} } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
}
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId));
|
||||
for (const patch of [{ controllerGeneration: 0 }, { controllerBootId: "another-boot" }, { phase: "unknown" }, { phase: "terminal_failure" }]) {
|
||||
await db.update(nativeRunFinalizations).set(patch).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
await db.update(nativeRunFinalizations).set({ controllerGeneration: 1, controllerBootId: f.controller.bootId, phase: "workspace_finalizing" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
}
|
||||
await expect(db.update(heartbeatRuns).set({ completionContractId: null }).where(eq(heartbeatRuns.id, f.runId)))
|
||||
.rejects.toMatchObject({ cause: { code: "23503" } });
|
||||
expect((await f.probe()).ready).toBe(true);
|
||||
for (const terminal of [null, {}, { runTerminalState: "active" }, { runTerminalState: "unknown" }]) {
|
||||
await db.update(nativeRunResults).set({ resultJson: { terminal } }).where(eq(nativeRunResults.id, f.resultId));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
}
|
||||
await db.update(nativeRunResults).set({ resultJson: { terminal: { runTerminalState: "succeeded" } } }).where(eq(nativeRunResults.id, f.resultId));
|
||||
await db.update(nativeRunResults).set({ schemaStatus: "rejected" }).where(eq(nativeRunResults.id, f.resultId));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
await db.update(nativeRunResults).set({ schemaStatus: "accepted" }).where(eq(nativeRunResults.id, f.resultId));
|
||||
const foreignCheckout = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: foreignCheckout, companyId: f.companyId, agentId: f.agentId, status: "succeeded" });
|
||||
for (const patch of [{ executionRunId: null }, { checkoutRunId: foreignCheckout }, { status: "cancelled" }, { assigneeAgentId: null }]) {
|
||||
await db.update(issues).set(patch).where(eq(issues.id, f.issueId));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
await db.update(issues).set({ executionRunId: f.runId, checkoutRunId: f.runId, status: "in_progress", assigneeAgentId: f.agentId }).where(eq(issues.id, f.issueId));
|
||||
}
|
||||
for (const patch of [{ cleanupStatus: "pending" }, { status: "expired" }, { status: "released" }, { status: "unknown" }, { expiresAt: new Date(0) },
|
||||
{ providerLeaseId: "another-sandbox" }, { heartbeatRunId: null }]) {
|
||||
await db.update(environmentLeases).set(patch).where(eq(environmentLeases.id, f.leaseId));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
await db.update(environmentLeases).set({ status: "active", cleanupStatus: null, expiresAt: null,
|
||||
providerLeaseId: "sandbox", heartbeatRunId: f.runId }).where(eq(environmentLeases.id, f.leaseId));
|
||||
}
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeWorkspaceSync: { state: "prepared" } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await db.update(nativeRunFinalizations).set({ leaseOwner: `native-finalizer:${randomUUID()}`, leaseExpiresAt: new Date(Date.now() + 600_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect((await f.probe()).ready).toBe(true);
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", "");
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", "fixture-token");
|
||||
});
|
||||
|
||||
it("allows independent admitted work while observing both saved results", async () => {
|
||||
const first = await seed(), second = await seed();
|
||||
expect(await readUpgradeReadiness(db, { activeRunIds: new Set([first.runId, second.runId]),
|
||||
counters: () => ({ activeRuns: 2, pendingWakes: 0, draining: false }) })).toMatchObject({ ready: true, recoverableRuns: 2 });
|
||||
});
|
||||
|
||||
it("bounds saved-result inspection across several locked candidates", async () => {
|
||||
const fixtures = [await seed(), await seed(), await seed()];
|
||||
const probeDb = createDb(database.connectionString, { maxConnections: 1 });
|
||||
const lock = await db.$client.reserve();
|
||||
try {
|
||||
await lock.unsafe("begin");
|
||||
await lock.unsafe("lock table workspace_operations in access exclusive mode");
|
||||
const started = performance.now();
|
||||
const report = readUpgradeReadiness(probeDb, { activeRunIds: new Set(fixtures.map(f => f.runId)),
|
||||
counters: () => ({ activeRuns: 3, pendingWakes: 0, draining: false }) });
|
||||
expect(await db.execute(sql`select 1 as value`)).toMatchObject([{ value: 1 }]);
|
||||
expect(await report).toMatchObject({ ready: false, activeRuns: 3, recoverableRuns: 0, blockingRuns: 3 });
|
||||
expect(performance.now() - started).toBeLessThan(5_000);
|
||||
expect(await probeDb.execute(sql`show statement_timeout`)).toMatchObject([{ statement_timeout: "0" }]);
|
||||
} finally {
|
||||
await lock.unsafe("rollback"); lock.release(); await probeDb.$client.end();
|
||||
}
|
||||
expect(await readUpgradeReadiness(db, { activeRunIds: new Set(fixtures.map(f => f.runId)),
|
||||
counters: () => ({ activeRuns: 3, pendingWakes: 0, draining: false }) }))
|
||||
.toMatchObject({ ready: true, recoverableRuns: 3, blockingRuns: 0 });
|
||||
}, 15_000);
|
||||
|
||||
it("requires the applied decision and saved workspace before certifying committed projection", async () => {
|
||||
const f = await seed(), assessmentId = randomUUID(), decisionId = randomUUID();
|
||||
await db.insert(workAssessments).values({ id: assessmentId, companyId: f.companyId, issueId: f.issueId, runId: f.runId,
|
||||
contractId: f.contractId, resultId: f.resultId, triggerKind: "native_result", triggerActorCompanyId: f.companyId,
|
||||
priorIssueStatus: "in_progress", priorStatusVersion: 0, policyVersion: "fixture", assessmentJson: {}, inputDigest: assessmentId });
|
||||
await db.insert(statusDecisions).values({ id: decisionId, companyId: f.companyId, issueId: f.issueId, runId: f.runId,
|
||||
assessmentId, decisionVersion: 1, policyVersion: "fixture", fromStatus: "in_progress", toStatus: "done",
|
||||
reasonCode: "completion_contract_satisfied", decisionJson: {}, decisionDigest: decisionId, applicationState: "applied", appliedAt: new Date() });
|
||||
await db.update(nativeRunFinalizations).set({ phase: "committed", assessmentId, decisionId, leaseOwner: null, leaseExpiresAt: null })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await db.update(issues).set({ status: "done", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
const [barrier] = await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
heartbeatRunId: f.runId, executionWorkspaceId: f.workspaceId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date() }).returning();
|
||||
expect((await f.probe()).ready).toBe(true);
|
||||
for (const patch of [{ applicationState: "proposed" }, { applicationState: "superseded" }]) {
|
||||
await db.update(statusDecisions).set(patch).where(eq(statusDecisions.id, decisionId));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
}
|
||||
await db.update(statusDecisions).set({ applicationState: "applied" }).where(eq(statusDecisions.id, decisionId));
|
||||
await db.update(workspaceOperations).set({ status: "running" }).where(eq(workspaceOperations.id, barrier.id));
|
||||
expect((await f.probe()).ready).toBe(false);
|
||||
await db.update(workspaceOperations).set({ status: "succeeded" }).where(eq(workspaceOperations.id, barrier.id));
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await projectCommittedRun({ db, run, coordinator });
|
||||
await projectCommittedRun({ db, run, coordinator });
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toMatchObject({ status: "succeeded", nativePhase: "committed" });
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("waits for a real signed stopped-worker receipt, then resumes workspace ownership once", async () => {
|
||||
const f = await seed();
|
||||
const worker = currentWorkspaceWorker()!;
|
||||
const owner = { token: randomUUID(), controllerBootId: f.controller.bootId, hostname: os.hostname(),
|
||||
pid: f.controller.pid, processStartedAt: f.controller.processStartedAt.toISOString(), worker };
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const { privateKey, publicKey } = generateKeyPairSync("ed25519");
|
||||
vi.stubEnv("PAPERCLIP_CLOUD_RUNTIME_IDENTITY_JWKS", JSON.stringify({ keys: [{ ...publicKey.export({ format: "jwk" }), kid: "fixture", alg: "EdDSA", use: "sig" }] }));
|
||||
let stopped = false;
|
||||
const fetch = vi.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) => {
|
||||
expect(JSON.parse(String(init?.body))).toEqual({ runId: f.runId, ownerToken: owner.token, worker });
|
||||
if (!stopped) return new Response(null, { status: 409 });
|
||||
const header = Buffer.from(JSON.stringify({ alg: "EdDSA", typ: "paperclip-workspace-worker-stop+jwt", kid: "fixture" })).toString("base64url");
|
||||
const iat = Math.floor(Date.now() / 1000);
|
||||
const payload = Buffer.from(JSON.stringify({ v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1",
|
||||
sub: "stack", runId: f.runId, ownerToken: owner.token, worker, stopped: true, iat, exp: iat + 60 })).toString("base64url");
|
||||
return Response.json({ receipt: `${header}.${payload}.${sign(null, Buffer.from(`${header}.${payload}`), privateKey).toString("base64url")}` });
|
||||
});
|
||||
try {
|
||||
expect((await f.probe()).ready).toBe(true);
|
||||
expect(fetch).not.toHaveBeenCalled();
|
||||
vi.stubEnv("RAILWAY_DEPLOYMENT_ID", "replacement");
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson,
|
||||
nativeWorkspaceFinalizationOwner: { ...owner, hostname: "previous-container", controllerBootId: "previous-boot" } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const scope = { db, companyId: f.companyId, runId: f.runId };
|
||||
const copyback = vi.fn();
|
||||
expect(await withNativeWorkspaceFinalizationOwnership(scope, copyback)).toEqual({ acquired: false, reason: "platform_stop_pending" });
|
||||
expect(copyback).not.toHaveBeenCalled();
|
||||
stopped = true;
|
||||
let enter!: () => void, release!: () => void;
|
||||
const entered = new Promise<void>(resolve => { enter = resolve; });
|
||||
const joined = new Promise<void>(resolve => { release = resolve; });
|
||||
const first = withNativeWorkspaceFinalizationOwnership(scope, async ownership => {
|
||||
await ownership.assertHeld(); copyback(); enter(); await joined;
|
||||
await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId,
|
||||
executionWorkspaceId: f.workspaceId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date() });
|
||||
});
|
||||
try {
|
||||
await entered;
|
||||
expect(await resumeNativeWorkspaceFinalization({ db, runId: f.runId })).toBeNull();
|
||||
} finally { release(); await first; }
|
||||
expect(await resumeNativeWorkspaceFinalization({ db, runId: f.runId })).toMatchObject({ status: "succeeded" });
|
||||
expect(copyback).toHaveBeenCalledOnce();
|
||||
const [recovered] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(recovered.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toBeUndefined();
|
||||
expect(recovered.runnerProfileJson?.nativeWorkspaceOwnerStopProof).toMatchObject({ runId: f.runId, ownerToken: owner.token, stopped: true, worker });
|
||||
} finally { fetch.mockRestore(); vi.stubEnv("RAILWAY_DEPLOYMENT_ID", "deployment"); }
|
||||
});
|
||||
});
|
||||
@@ -7,8 +7,10 @@ import { readNativeWorkspaceSyncReference } from "../native-runtime/native-works
|
||||
import { REMOTE_RUNNER_RECONNECT_GRACE_MS } from "../native-runtime/native-runner-recovery-contract.js";
|
||||
import { currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js";
|
||||
import { readNativeControllerRecoveryInventory } from "../native-runtime/native-controller-recovery-inventory.js";
|
||||
import { readNativeFinalizationUpgradeReadiness } from "../native-runtime/native-finalization-upgrade-readiness.js";
|
||||
|
||||
const MAX_UPGRADE_RUNS = 1_000;
|
||||
const POST_RESULT_INSPECTION_BUDGET_MS = 2_000;
|
||||
type Counters = { activeRuns: number; pendingWakes: number; draining: boolean };
|
||||
type Candidate = { id: string; recoverable: boolean };
|
||||
|
||||
@@ -20,17 +22,19 @@ export async function readUpgradeReadiness(db: Db, input: {
|
||||
activeRunIds: ReadonlySet<string>;
|
||||
counters: () => Counters;
|
||||
}) {
|
||||
const inspectionDeadline = performance.now() + POST_RESULT_INSPECTION_BUDGET_MS;
|
||||
const before = input.counters();
|
||||
const beforeIds = [...input.activeRunIds];
|
||||
const owner = await currentNativeControllerIdentity();
|
||||
const rows = await db.transaction(async tx => {
|
||||
await tx.execute(sql`set local statement_timeout = '2s'`);
|
||||
return tx.execute<Candidate & { companyId: string; issueId: string; runnerId: string; sessionId: string; workspace: unknown; remoteWorkspace: unknown; hasRemoteWorkspace: boolean; pluginLeaseId: string | null; material: Record<string, unknown> | null }>(sql`
|
||||
return tx.execute<Candidate & { companyId: string; issueId: string; runnerId: string; sessionId: string; workspace: unknown; remoteWorkspace: unknown; hasRemoteWorkspace: boolean; hasResult: boolean; pluginLeaseId: string | null; material: Record<string, unknown> | null }>(sql`
|
||||
select r.id, r.company_id as "companyId", r.native_issue_id as "issueId",
|
||||
r.runner_instance_id as "runnerId", r.native_session_id as "sessionId",
|
||||
r.runner_profile_json->'nativeWorkspaceSync' as workspace,
|
||||
r.runner_profile_json->'remoteWorkspace' as "remoteWorkspace",
|
||||
r.runner_profile_json ? 'remoteWorkspace' as "hasRemoteWorkspace", p.id as "pluginLeaseId",
|
||||
f.result_id is not null as "hasResult",
|
||||
case when pg_column_size(c.material) <= ${NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES}
|
||||
and sum(pg_column_size(c.material)) over () <= ${NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES} then c.material else null end as material, (
|
||||
r.runtime_mode = 'native' and a.adapter_type = 'paperclip_runner'
|
||||
@@ -58,19 +62,25 @@ export async function readUpgradeReadiness(db: Db, input: {
|
||||
where r.status = 'running'
|
||||
limit ${MAX_UPGRADE_RUNS + 1}`);
|
||||
}, { accessMode: "read only" });
|
||||
const candidates = rows.map(row => {
|
||||
const candidates: Candidate[] = [];
|
||||
for (const row of rows) {
|
||||
const remote = remoteWorkspaceRunReferenceSchema.safeParse(row.remoteWorkspace);
|
||||
// A remote reference owns the logical workspace identity. Corruption or conflicting
|
||||
// plugin authority must not fall back to a different recoverable-looking record.
|
||||
const workspaceId = row.hasRemoteWorkspace
|
||||
? remote.success && !row.pluginLeaseId ? remote.data.workspaceId : null
|
||||
: row.pluginLeaseId ?? readNativeWorkspaceSyncReference(row.workspace)?.workspaceId;
|
||||
const identity = row.recoverable && row.material
|
||||
const identity = (row.recoverable || row.hasResult) && row.material
|
||||
? inspectNativeRunnerUpgradeHeader({ companyId: row.companyId, issueId: row.issueId, runId: row.id }, row.material) : null;
|
||||
return { id: row.id, recoverable: Boolean(workspaceId && identity && identity.runId === row.id
|
||||
const bound = Boolean(workspaceId && identity && identity.runId === row.id
|
||||
&& identity.runnerInstanceId === row.runnerId && identity.normalizedSessionId === row.sessionId
|
||||
&& identity.environmentLeaseId === workspaceId) };
|
||||
});
|
||||
&& identity.environmentLeaseId === workspaceId);
|
||||
const finalization = bound && row.hasResult && remote.success && !row.pluginLeaseId
|
||||
&& performance.now() < inspectionDeadline
|
||||
? await readNativeFinalizationUpgradeReadiness(db, { companyId: row.companyId, runId: row.id,
|
||||
controller: owner, supportedOutageMs: 300_000 }).catch(() => false) : false;
|
||||
candidates.push({ id: row.id, recoverable: bound && (row.recoverable || finalization) });
|
||||
}
|
||||
// Additive evidence does not strengthen or redefine the existing ready flag.
|
||||
const retained = await readNativeControllerRecoveryInventory(db).catch(() => ({
|
||||
version: 1 as const, observedAt: new Date().toISOString(), coverage: "unknown" as const,
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
import { and, desc, eq, sql } from "drizzle-orm";
|
||||
import { agents, environmentLeases, environments, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations,
|
||||
nativeRunResults, statusDecisions, workAssessments, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared";
|
||||
import type { NativeControllerIdentity } from "./native-restart-recovery.js";
|
||||
import { currentNativeWorkspaceRecoveryResult } from "./native-workspace-recovery-binding.js";
|
||||
import { nativeWorkspaceFinalizationOwnerCanUpgrade } from "./native-workspace-finalization-ownership.js";
|
||||
import { currentWorkspaceWorker, workspaceWorkerStopProofSupported } from "./native-workspace-platform-proof.js";
|
||||
import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js";
|
||||
|
||||
export async function readNativeFinalizationUpgradeReadiness(db: Db, input: {
|
||||
companyId: string; runId: string; controller: NativeControllerIdentity; supportedOutageMs: number;
|
||||
}): Promise<boolean> {
|
||||
const worker = currentWorkspaceWorker();
|
||||
if (!workspaceWorkerStopProofSupported(worker)) return false;
|
||||
return db.transaction(async tx => {
|
||||
await tx.execute(sql`set local statement_timeout = '2s'`);
|
||||
const [row] = await tx.select({
|
||||
run: { id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId, runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status,
|
||||
errorCode: heartbeatRuns.errorCode, nativePhase: heartbeatRuns.nativePhase, completionContractId: heartbeatRuns.completionContractId,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
contextSnapshot: sql<Record<string, unknown>>`jsonb_strip_nulls(jsonb_build_object(
|
||||
'nativeReviewInteractionId', ${heartbeatRuns.contextSnapshot}->'nativeReviewInteractionId',
|
||||
'nativeReviewDecisionId', ${heartbeatRuns.contextSnapshot}->'nativeReviewDecisionId'))`,
|
||||
runnerProfileJson: sql<Record<string, unknown>>`jsonb_build_object(
|
||||
'remoteWorkspace', ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace',
|
||||
'nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync' is not null
|
||||
and ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync' <> 'null'::jsonb,
|
||||
'nativeWorkspaceFinalizationOwner', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceFinalizationOwner')` },
|
||||
coordinator: { companyId: nativeRunFinalizations.companyId, issueId: nativeRunFinalizations.issueId,
|
||||
runId: nativeRunFinalizations.runId, phase: nativeRunFinalizations.phase, resultId: nativeRunFinalizations.resultId,
|
||||
controllerBootId: nativeRunFinalizations.controllerBootId, controllerPid: nativeRunFinalizations.controllerPid,
|
||||
controllerProcessStartedAt: nativeRunFinalizations.controllerProcessStartedAt, controllerGeneration: nativeRunFinalizations.controllerGeneration,
|
||||
leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt,
|
||||
assessmentId: nativeRunFinalizations.assessmentId, decisionId: nativeRunFinalizations.decisionId },
|
||||
issue: { id: issues.id, companyId: issues.companyId, status: issues.status, executionRunId: issues.executionRunId,
|
||||
checkoutRunId: issues.checkoutRunId, assigneeAgentId: issues.assigneeAgentId, executionWorkspaceId: issues.executionWorkspaceId },
|
||||
adapter: agents.adapterType,
|
||||
})
|
||||
.from(heartbeatRuns).innerJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.runId, heartbeatRuns.id),
|
||||
eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId)))
|
||||
.innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, heartbeatRuns.companyId)))
|
||||
.innerJoin(agents, and(eq(agents.id, heartbeatRuns.agentId), eq(agents.companyId, heartbeatRuns.companyId)))
|
||||
.where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))).limit(1);
|
||||
if (!row) return false;
|
||||
const { run, coordinator, issue } = row;
|
||||
if (run.runtimeMode !== "native" || run.status !== "running" || row.adapter !== "paperclip_runner"
|
||||
|| isNativeRunnerOwnershipHeld(run) || !coordinator.resultId || run.nativeIssueId !== issue.id
|
||||
|| coordinator.controllerBootId !== input.controller.bootId || coordinator.controllerPid !== input.controller.pid
|
||||
|| coordinator.controllerProcessStartedAt?.getTime() !== input.controller.processStartedAt.getTime()
|
||||
|| coordinator.controllerGeneration < 1
|
||||
|| !["result_accepted", "workspace_finalizing", "retryable_failure", "committed"].includes(coordinator.phase)) return false;
|
||||
if (coordinator.leaseOwner?.startsWith("native-finalizer:") && coordinator.leaseExpiresAt
|
||||
&& coordinator.leaseExpiresAt.getTime() > Date.now() + input.supportedOutageMs) return false;
|
||||
const profile = run.runnerProfileJson;
|
||||
const reference = remoteWorkspaceRunReferenceSchema.safeParse(profile?.remoteWorkspace);
|
||||
if (!reference.success || profile.nativeWorkspaceSync === true
|
||||
|| !nativeWorkspaceFinalizationOwnerCanUpgrade(profile, input.controller, worker)) return false;
|
||||
const remote = reference.data;
|
||||
const [workspace] = await tx.select({ id: executionWorkspaces.id, providerType: executionWorkspaces.providerType,
|
||||
metadata: sql<Record<string, unknown>>`jsonb_build_object('remoteWorkspace', ${executionWorkspaces.metadata}->'remoteWorkspace')` })
|
||||
.from(executionWorkspaces).where(and(eq(executionWorkspaces.id, remote.workspaceId),
|
||||
eq(executionWorkspaces.companyId, input.companyId))).limit(1);
|
||||
if (!workspace || workspace.providerType !== "cloud_sandbox") return false;
|
||||
let authority;
|
||||
try { authority = readRemoteWorkspaceAuthority(workspace.metadata); } catch { return false; }
|
||||
if (!authority) return false;
|
||||
const [barrier] = await tx.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(eq(workspaceOperations.companyId, input.companyId),
|
||||
eq(workspaceOperations.heartbeatRunId, run.id), eq(workspaceOperations.issueId, issue.id),
|
||||
eq(workspaceOperations.executionWorkspaceId, workspace.id), eq(workspaceOperations.phase, "workspace_finalize"),
|
||||
eq(workspaceOperations.status, "succeeded"))).orderBy(desc(workspaceOperations.createdAt)).limit(1);
|
||||
const [result] = await tx.select({ id: nativeRunResults.id, completionContractId: nativeRunResults.completionContractId,
|
||||
terminalState: sql<string>`${nativeRunResults.resultJson}->'terminal'->>'runTerminalState'` }).from(nativeRunResults).where(and(eq(nativeRunResults.id, coordinator.resultId),
|
||||
eq(nativeRunResults.companyId, input.companyId), eq(nativeRunResults.runId, run.id), eq(nativeRunResults.issueId, issue.id),
|
||||
eq(nativeRunResults.schemaStatus, "accepted"))).limit(1);
|
||||
if (!result || result.completionContractId !== run.completionContractId) return false;
|
||||
if (!["succeeded", "failed", "cancelled"].includes(result.terminalState)) return false;
|
||||
if (coordinator.phase === "committed") {
|
||||
if (!barrier || !coordinator.decisionId || !coordinator.assessmentId) return false;
|
||||
const [decision] = await tx.select({ id: statusDecisions.id }).from(statusDecisions).innerJoin(workAssessments,
|
||||
and(eq(workAssessments.id, statusDecisions.assessmentId), eq(workAssessments.companyId, statusDecisions.companyId),
|
||||
eq(workAssessments.issueId, statusDecisions.issueId), eq(workAssessments.runId, statusDecisions.runId)))
|
||||
.where(and(eq(statusDecisions.id, coordinator.decisionId), eq(statusDecisions.companyId, input.companyId),
|
||||
eq(statusDecisions.issueId, issue.id), eq(statusDecisions.runId, run.id), eq(statusDecisions.applicationState, "applied"),
|
||||
eq(workAssessments.id, coordinator.assessmentId), eq(workAssessments.resultId, result.id),
|
||||
eq(workAssessments.contractId, result.completionContractId))).limit(1);
|
||||
return Boolean(decision);
|
||||
}
|
||||
if (!await currentNativeWorkspaceRecoveryResult(tx as unknown as Db, { companyId: input.companyId, run, coordinator, issue })) return false;
|
||||
if (barrier) return true;
|
||||
const location = authority.location;
|
||||
if (!location || authority.writerRunId !== run.id || authority.epoch !== remote.epoch || location.leaseId !== remote.leaseId
|
||||
|| location.providerLeaseId !== remote.providerLeaseId || location.remoteCwd !== remote.remoteCwd
|
||||
|| issue.executionWorkspaceId !== workspace.id) return false;
|
||||
const [lease] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).innerJoin(environments,
|
||||
eq(environments.id, environmentLeases.environmentId)).where(and(eq(environmentLeases.id, location.leaseId),
|
||||
eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.environmentId, location.environmentId),
|
||||
eq(environmentLeases.executionWorkspaceId, workspace.id), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
eq(environmentLeases.issueId, issue.id), eq(environmentLeases.providerLeaseId, location.providerLeaseId),
|
||||
eq(environments.driver, "sandbox"), eq(environments.status, "active"),
|
||||
eq(environmentLeases.status, "active"),
|
||||
sql`${environmentLeases.releasedAt} is null and ${environmentLeases.cleanupStatus} is null`,
|
||||
sql`(${environmentLeases.expiresAt} is null or ${environmentLeases.expiresAt} > clock_timestamp())`)).limit(1);
|
||||
return Boolean(lease);
|
||||
}, { accessMode: "read only" });
|
||||
}
|
||||
@@ -17,6 +17,7 @@ import { withWorkspaceLockKeepalive } from "./native-workspace-lock-keepalive.js
|
||||
import {
|
||||
currentNativeControllerIdentity,
|
||||
evaluateNativeControllerTakeover,
|
||||
type NativeControllerIdentity,
|
||||
} from "./native-restart-recovery.js";
|
||||
|
||||
import { currentWorkspaceWorker, requestWorkspaceWorkerStopProof, type WorkspaceWorker, type WorkspaceWorkerStopProof } from "./native-workspace-platform-proof.js";
|
||||
@@ -75,6 +76,18 @@ function readOwner(value: unknown): Owner | null {
|
||||
? owner as Owner : null;
|
||||
}
|
||||
|
||||
export function nativeWorkspaceFinalizationOwnerCanUpgrade(profile: Record<string, unknown> | null,
|
||||
identity: NativeControllerIdentity, worker: WorkspaceWorker | undefined): boolean {
|
||||
const raw = profile?.[OWNER_KEY];
|
||||
if (raw == null) return true;
|
||||
const owner = readOwner(raw);
|
||||
return Boolean(owner && owner.token.length > 0 && owner.token.length <= 200 && worker && owner.worker
|
||||
&& Object.keys(owner.worker).length === Object.keys(worker).length && owner.controllerBootId === identity.bootId
|
||||
&& owner.hostname === os.hostname() && owner.pid === identity.pid
|
||||
&& Date.parse(owner.processStartedAt) === identity.processStartedAt.getTime()
|
||||
&& Object.keys(worker).every(key => owner.worker![key as keyof WorkspaceWorker] === worker[key as keyof WorkspaceWorker]));
|
||||
}
|
||||
|
||||
/**
|
||||
* The advisory lock serializes controllers without holding row locks across I/O.
|
||||
* The durable receipt additionally prevents a disconnected lock session from
|
||||
|
||||
@@ -62,6 +62,15 @@ export type WorkspaceWorkerStopCheck =
|
||||
| { status: "pending" }
|
||||
| { status: "unsupported" };
|
||||
|
||||
export function workspaceWorkerStopProofSupported(worker: WorkspaceWorker | undefined, env: NodeJS.ProcessEnv = process.env): boolean {
|
||||
const context = getCloudStackContext(env);
|
||||
if (!context?.stackId || !context.cloudOrigin || !env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN?.trim() || !validWorker(worker)) return false;
|
||||
try {
|
||||
const origin = new URL(context.cloudOrigin);
|
||||
return origin.protocol === "https:" && !origin.username && !origin.password;
|
||||
} catch { return false; }
|
||||
}
|
||||
|
||||
|
||||
export async function requestWorkspaceWorkerStopProof(binding: WorkspaceStopBinding, options: {
|
||||
env?: NodeJS.ProcessEnv; fetchImpl?: typeof fetch; now?: Date;
|
||||
@@ -69,7 +78,7 @@ export async function requestWorkspaceWorkerStopProof(binding: WorkspaceStopBind
|
||||
const env = options.env ?? process.env;
|
||||
const context = getCloudStackContext(env);
|
||||
const token = env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN?.trim();
|
||||
if (!context?.stackId || !context.cloudOrigin || !token || !validWorker(binding.worker)) return { status: "unsupported" };
|
||||
if (!workspaceWorkerStopProofSupported(binding.worker, env) || !context?.stackId || !context.cloudOrigin || !token) return { status: "unsupported" };
|
||||
let origin: URL;
|
||||
try { origin = new URL(context.cloudOrigin); } catch { return { status: "unsupported" }; }
|
||||
if (origin.protocol !== "https:" || origin.username || origin.password) return { status: "unsupported" };
|
||||
|
||||
@@ -2,13 +2,12 @@ import { and, eq, gt, inArray, isNull, ne, not, or, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
|
||||
import { getNativeReviewAssignment } from "./native-review-participant.js";
|
||||
|
||||
/** Both operator and platform recovery must still own the task's accepted work.
|
||||
* Call with coordinator -> issue -> run rows locked in a short transaction. */
|
||||
export async function currentNativeWorkspaceRecoveryResult(db: Db, input: {
|
||||
companyId: string;
|
||||
issue: typeof issues.$inferSelect;
|
||||
run: typeof heartbeatRuns.$inferSelect;
|
||||
coordinator: typeof nativeRunFinalizations.$inferSelect;
|
||||
issue: Pick<typeof issues.$inferSelect, "id" | "companyId" | "status" | "executionRunId" | "checkoutRunId" | "assigneeAgentId">;
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "nativeIssueId" | "runtimeMode" | "status" | "agentId"
|
||||
| "contextSnapshot" | "completionContractId" | "createdAt">;
|
||||
coordinator: Pick<typeof nativeRunFinalizations.$inferSelect, "companyId" | "issueId" | "runId" | "phase" | "resultId">;
|
||||
}) {
|
||||
const { companyId, issue, run, coordinator } = input;
|
||||
if (issue.companyId !== companyId || run.companyId !== companyId || coordinator.companyId !== companyId
|
||||
@@ -22,7 +21,7 @@ export async function currentNativeWorkspaceRecoveryResult(db: Db, input: {
|
||||
contextSnapshot: run.contextSnapshot, actingRunId: run.id, issueExecutionRunId: issue.executionRunId,
|
||||
allowResolvedByRunId: run.id,
|
||||
})) return null;
|
||||
const [result] = await db.select().from(nativeRunResults).where(and(
|
||||
const [result] = await db.select({ id: nativeRunResults.id, completionContractId: nativeRunResults.completionContractId }).from(nativeRunResults).where(and(
|
||||
eq(nativeRunResults.id, coordinator.resultId), eq(nativeRunResults.companyId, companyId),
|
||||
eq(nativeRunResults.issueId, issue.id), eq(nativeRunResults.runId, run.id), eq(nativeRunResults.schemaStatus, "accepted"),
|
||||
));
|
||||
|
||||
Reference in new issue
Block a user