mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 06:04:11 +01:00
fix(server): wake native workspace finalization from committed work (#17734)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native runs can accept a result before workspace copy-back finishes. > - The periodic orphan pass repeatedly tries this recovery, including cases that require operator action. > - Copy-back needs exact ownership evidence. A generic timer cannot grant that authority. > - This pull request wakes workspace recovery from committed changes and local owner completion. > - Saved retries use the existing scheduler. Empty work, completed barriers, and operator holds become quiet. ## Linked Issues or Issue Description Refs #17717. This is slice 52 of the recurring-server sleep stack, targeting master after parent #17717 merged at `4c56702e06c698cc10d4a86514669aba0f4d9a9b`. Related work: #17709 (status reassessment), #17687 (unsafe export recovery), and #16478 (workspace repair holds). This PR covers native workspace copy-back and validation. Result commit/projection and resource settlement remain separate recovery duties. It does not claim full-instance sleep. ## What Changed - Register a company/run/task-scoped workspace worker. Reconstruct accepted, unfinished work from existing records. Use saved retry and lease deadlines. - Notify from both accepted-result writers (control-plane and PRP coordinator), workspace receipts, coordinator changes, operator confirmation, retained workspace repair, and relevant task updates. Notifications follow the outer transaction. - Distinguish active local ownership, busy external ownership, pending Cloud evidence, operator-required evidence, and changed bindings. Only concrete external ownership/evidence waits receive a scoped retry. - Preserve advisory locks, signed stop proof, exact binding checks, physical callback joins, and existing failure policy. Join physical operations during shutdown. - Respect the existing heartbeat execution-suppression policy before discovery and execution; wake suppressed work on settings activation. - Skip physical workspace recovery in the enabled periodic orphan pass. Keep direct/startup reconciliation and the remaining result/cleanup recovery paths. - Keep routine company-only finalization hints quiet. Ordinary work requires issue/run scope; explicit settings and owner-release events keep their company-wide contract. - Preserve the parent's entire Runner/package tree, profile 80 and all historical identities. Add the dated obligation, producer, lifecycle and limitation audit. ## Verification - Current local source: `4059b24e3ea5c623be54ade18ec9b6ab78eb0289`, prepared by a normal merge of the current parent. The sole manual conflict keeps ordinary workspace fanout inside the existing issue/run guard. All 1,198 parent-only, 15 owned-only and 10 automatic shared paths are preserved. The complete package tree matches the parent byte-for-byte. - Controlled regression: the original workspace-only unscoped fanout performs three candidate SELECTs for routine company hints. The same zero-query assertion passes with scoped fanout. A real accepted result then produces one successful workspace settlement and returns to quiet. The failed baseline remains in the audit. - The first complete local group passes **1,693 cases in 14 files**, including all 19 workspace-worker cases. It covers both accepted-result producers, physical ownership, export retry/failure, finalization recovery, arbiter corpus, issue recovery actions, control-plane port, workspace recovery, heartbeat recovery, startup, retained cleanup and native execution. - Separate process-recovery and external-chat response files qualify the changed recovery/settlement paths without sharing their general test database with other files. Both pass in full: 410 process-recovery cases in 162.79 seconds and 273 external-chat cases in 53.79 seconds. The total is **2,376 distinct passes in 16 complete files**, with no skips. Normal merge `5c46c52873158529736b8349e4d44658865108ee` adopts the parent squash with this entire tested child tree unchanged. - Production server typechecking, build, module boundaries, migration order and source-profile checks pass. Thirteen dedicated PostgreSQL 17 fixtures use the unchanged migration 413 schema. Embedded-only paths use the existing recovery helper with fresh per-call fixtures. All temporary imports are restored byte-exact. No test assertion or timeout is relaxed. - Runner validation belongs to the unchanged parent package tree: 130 passing installation/compatibility/transport cases, 22 installed ACPX/Grok contract cases and 26 native-provider Rust cases. One existing opt-in closed Pi snapshot case is skipped because its distribution root is not configured. These are source-scoped checks, with no live provider qualification claim. - The owner audit at `doc/plans/2026-10-26-native-workspace-finalization-sleep-audit.md` preserves earlier failures and their disposition. Current-head acceptance: all 57 checks are clean, including protected CI and security. Greptile gives 5/5 on `7df5c9c44ea9e171be3c61b4430d3727b10f3337` with no actionable findings. Earlier scores do not substitute for this result. No Ponytail verdict is present under the authorized missing-review policy. - No claim of a full local repository suite, deployed app/database sleep, expanded Cloud wake coverage or measured cost savings is made. ## Risks - Supported wake producers are in-process application services. This adds no cross-replica or independent SQL-writer transport. - A real unresolved export still needs recovery. Busy owners and pending Cloud stop evidence retain a 30-second scoped check. No timeout authorizes copy-back. - Result projection and resource release still use their existing recovery. Some existing cleanup failures remain best effort; this PR does not infer cleanup success from a workspace barrier. - Cloud alarm coverage and host-sleep safety remain unchanged. No deployed suspension or cost savings have been measured. - Parent #17717 is merged. Current base is master at `4c56702e06c698cc10d4a86514669aba0f4d9a9b`. Fresh current-head CI and review must pass before this slice merges. ## Model Used OpenAI Codex, GPT-6 family, with repository tools and code execution. The exact serving model ID and context-window size are not exposed in this session. No subagent was used for this slice. ## 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 --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
26 files changed
+987
-66
No files matched your search
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
|
||||
|
||||
/** Topics for existing durable queues; register intent before their writes. */
|
||||
export const DELIVERY_QUEUES = {
|
||||
nativeWorkspaceFinalization: "native-workspace-finalization",
|
||||
retainedNativeSessionCleanup: "retained-native-session-cleanup",
|
||||
nativeStatusReassessment: "native-status-reassessment",
|
||||
nativeReviewResponse: "native-review-response",
|
||||
|
||||
@@ -1024,6 +1024,7 @@ export function heartbeatService(
|
||||
drainRunningRunsForShutdown,
|
||||
registerScheduledRetryCancellationWork,
|
||||
registerRetainedNativeSessionCleanupWork,
|
||||
registerNativeWorkspaceFinalizationWork,
|
||||
} = createHeartbeatRecovery(db, {
|
||||
pluginWorkerManager: options.pluginWorkerManager,
|
||||
enterShutdown: () => { shutdownInProgress = true; },
|
||||
@@ -4093,6 +4094,14 @@ export function heartbeatService(
|
||||
triggerIssueMonitor,
|
||||
|
||||
reportRunActivity: clearDetachedRunWarning,
|
||||
registerNativeWorkspaceFinalizationWork: (
|
||||
coordinator: Parameters<typeof registerNativeWorkspaceFinalizationWork>[0], canRun: () => boolean,
|
||||
) => registerNativeWorkspaceFinalizationWork(coordinator,
|
||||
() => canRun() && readTaskDrain(new Date()) === null,
|
||||
async () => {
|
||||
const state = await getSchedulingSuppression(true);
|
||||
return !state.suppressed || state.reason === "task_drain";
|
||||
}),
|
||||
registerRetainedNativeSessionCleanupWork: (
|
||||
coordinator: Parameters<typeof registerRetainedNativeSessionCleanupWork>[0], canRun: () => boolean,
|
||||
) => registerRetainedNativeSessionCleanupWork(coordinator, () => canRun() && readTaskDrain(new Date()) === null),
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { nativeWorkspaceFinalizationWork } from "../native-runtime/native-workspace-finalization-work.js";
|
||||
import { attemptNativeWorkspaceFinalization } from "../native-runtime/native-workspace-finalizer.js";
|
||||
import { retainedNativeSessionCleanupWork } from "../native-runtime/retained-native-session-cleanup-work.js";
|
||||
import { reconcileRetainedNativeSessionCleanups } from "../native-runtime/retained-native-session-cleanup.js";
|
||||
import { createScheduledRetryCancellationRecovery } from "./scheduled-retry-cancellation.js";
|
||||
@@ -1930,6 +1932,17 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
return operation;
|
||||
}
|
||||
const retainedNativeCleanup = retainedNativeSessionCleanupWork(db, cleanupRetainedNativeSession);
|
||||
const nativeWorkspaceFinalization = nativeWorkspaceFinalizationWork(db, {
|
||||
environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager,
|
||||
onWorkspaceSettled: settleRecoveredNativeWorkspace,
|
||||
attempt: input => {
|
||||
const operation = attemptNativeWorkspaceFinalization(input);
|
||||
const tracked = operation.then(() => undefined, () => undefined);
|
||||
activeRunExecutionPromises.add(tracked);
|
||||
void tracked.then(() => activeRunExecutionPromises.delete(tracked));
|
||||
return operation;
|
||||
},
|
||||
});
|
||||
|
||||
function scheduleRetainedNativeSessionCleanup() {
|
||||
// The per-database sweep joins startup and periodic callers. One bounded
|
||||
@@ -1954,7 +1967,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
|
||||
let settledProcessesDiscovered = false;
|
||||
|
||||
async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean }) {
|
||||
async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean }) {
|
||||
if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db);
|
||||
const staleThresholdMs = opts?.staleThresholdMs ?? 0;
|
||||
const now = new Date();
|
||||
@@ -1979,6 +1992,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
skipNativeResponsePresentation: opts?.skipNativeResponsePresentation,
|
||||
skipNativeReviewResponse: opts?.skipNativeReviewResponse,
|
||||
skipNativeStatusReassessment: opts?.skipNativeStatusReassessment,
|
||||
skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization,
|
||||
environmentRuntime,
|
||||
pluginWorkerManager: dependencies.pluginWorkerManager,
|
||||
onWorkspaceSettled: settleRecoveredNativeWorkspace,
|
||||
@@ -2592,6 +2606,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
|
||||
return {
|
||||
registerRetainedNativeSessionCleanupWork: retainedNativeCleanup.register,
|
||||
registerNativeWorkspaceFinalizationWork: nativeWorkspaceFinalization.register,
|
||||
registerScheduledRetryCancellationWork: scheduledRetryCancellation.register,
|
||||
sweepPendingCleanupLeases,
|
||||
markNativeOwnershipUnverified,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
|
||||
import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js";
|
||||
import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
|
||||
import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js";
|
||||
@@ -626,6 +627,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
|
||||
await notifyDirectoryCleanup(tx);
|
||||
await notifyInstructionCollection(tx);
|
||||
await notifyCancelledWorkspacePreservation(tx);
|
||||
await notifyNativeWorkspaceFinalization(tx);
|
||||
}
|
||||
if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx);
|
||||
await notifyToolConnectionHealth(tx, "*");
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
|
||||
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
|
||||
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
|
||||
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
|
||||
@@ -11502,6 +11503,7 @@ export function issueService(db: Db) {
|
||||
if (patch.status !== undefined || patch.statusVersion !== undefined || patch.executionState !== undefined
|
||||
|| patch.executionRunId !== undefined || patch.checkoutRunId !== undefined
|
||||
|| patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined || blockedByIssueIds !== undefined) {
|
||||
await notifyNativeWorkspaceFinalization(tx, { companyId: receiptExisting.companyId, issueId: id });
|
||||
await notifyNativeStatusReassessment(tx, { companyId: receiptExisting.companyId, issueId: id });
|
||||
}
|
||||
const updated = await tx
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
|
||||
import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js";
|
||||
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
|
||||
import { notifyNativeReviewResponse } from "./native-review-response-notifications.js";
|
||||
@@ -8,6 +9,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i
|
||||
|
||||
export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) {
|
||||
if (scope.issueId || scope.runId) {
|
||||
await notifyNativeWorkspaceFinalization(transaction, scope);
|
||||
await notifyRetainedNativeSessionCleanup(transaction, scope);
|
||||
await notifyNativeStatusReassessment(transaction, scope);
|
||||
await notifyNativeResponsePresentation(transaction, scope);
|
||||
|
||||
@@ -37,7 +37,7 @@ import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
||||
import { issueService } from "../issues.js";
|
||||
import { emitAgentTaskRun } from "../agent-task-run-telemetry.js";
|
||||
import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js";
|
||||
import { resumeNativeWorkspaceFinalization } from "./native-workspace-finalizer.js";
|
||||
import { resumeNativeWorkspaceFinalization, recordNativeWorkspaceFinalizationFailure } from "./native-workspace-finalizer.js";
|
||||
import { dismissObsoleteNativePolicyReviews } from "./obsolete-policy-reviews.js";
|
||||
import {
|
||||
cleanupNativeWorkspaceSync,
|
||||
@@ -387,6 +387,7 @@ type NativeReconciliationOptions = {
|
||||
skipNativeResponsePresentation?: boolean;
|
||||
skipNativeReviewResponse?: boolean;
|
||||
skipNativeStatusReassessment?: boolean;
|
||||
skipNativeWorkspaceFinalization?: boolean;
|
||||
pluginWorkerManager?: WorkspaceWorkers;
|
||||
environmentRuntime?: EnvironmentRuntimeService;
|
||||
onWorkspaceSettled?: (input: {
|
||||
@@ -619,13 +620,18 @@ async function reconcileNativeFinalizationsPass(
|
||||
const barrier = await db.select({ status: workspaceOperations.status })
|
||||
.from(workspaceOperations)
|
||||
.where(and(
|
||||
eq(workspaceOperations.companyId, row.companyId),
|
||||
eq(workspaceOperations.issueId, row.issueId),
|
||||
eq(workspaceOperations.heartbeatRunId, row.runId),
|
||||
eq(workspaceOperations.phase, "workspace_finalize"),
|
||||
))
|
||||
.orderBy(desc(workspaceOperations.createdAt))
|
||||
|
||||
|
||||
.orderBy(desc(sql`${workspaceOperations.status} = 'succeeded'`), desc(workspaceOperations.createdAt))
|
||||
.limit(1)
|
||||
.then((entries) => entries[0] ?? null);
|
||||
if (!barrier || barrier.status !== "succeeded") {
|
||||
if (options.skipNativeWorkspaceFinalization) continue;
|
||||
const recoveryDecision = resolveNativeReconciliationStatus({
|
||||
facts: { workspaceOperationPending: true },
|
||||
priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus,
|
||||
@@ -645,23 +651,7 @@ async function reconcileNativeFinalizationsPass(
|
||||
const workspaceFinalizeStatus =
|
||||
operation.status === "succeeded" ? "succeeded" : "failed";
|
||||
if (workspaceFinalizeStatus === "failed") {
|
||||
const unrecoverable = operation.stderrExcerpt?.includes(
|
||||
"workspace_sync_out_unrecoverable",
|
||||
);
|
||||
const pluginFailure = operation.stderrExcerpt?.trim();
|
||||
const pluginRepair = pluginFailure === "plugin_workspace_branch_mismatch" || pluginFailure === "plugin_workspace_owner_mismatch";
|
||||
const failure = await recordNativeFinalizationFailure({
|
||||
db,
|
||||
runId: row.runId,
|
||||
error: new Error(
|
||||
pluginRepair ? `native_${pluginFailure}` : unrecoverable
|
||||
? "native_workspace_sync_out_unrecoverable"
|
||||
: "native_workspace_sync_out_failed",
|
||||
),
|
||||
projectRunStatus: true,
|
||||
failureScope: "workspace",
|
||||
permanent: unrecoverable,
|
||||
});
|
||||
const failure = await recordNativeWorkspaceFinalizationFailure(db, row.runId, operation);
|
||||
results.push({
|
||||
...failure,
|
||||
reconciliationAction: "resume_workspace_operation" as const,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js";
|
||||
import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "../native-response-presentation-notifications.js";
|
||||
import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
|
||||
import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
|
||||
@@ -391,6 +392,7 @@ export class NativeRunCoordinatorStore {
|
||||
return { disposition: "duplicate" as const, resultId: existing.id };
|
||||
}
|
||||
|
||||
await notifyNativeWorkspaceFinalization(tx, this.#binding);
|
||||
const [inserted] = await tx
|
||||
.insert(nativeRunResults)
|
||||
.values({
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyNativeWorkspaceFinalization, releaseNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js";
|
||||
import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
|
||||
import { beginRunPhase } from "../execution-preparation-timing.js";
|
||||
import { logActivity } from "../activity-log.js";
|
||||
@@ -29,6 +30,7 @@ export async function tryLockNativeWorkspaceFinalization(tx: Db, input: { compan
|
||||
}
|
||||
|
||||
const activeOwnerTokens = new Set<string>();
|
||||
const activeRunOwners = new Map<string, { token: string; db: Db }>();
|
||||
// Positive same-boot evidence, retained only when durable cleanup could not be
|
||||
// confirmed. Absence from the active set alone never proves another token joined.
|
||||
const joinedOwnerTokens = new Set<string>();
|
||||
@@ -40,6 +42,13 @@ type Owner = {
|
||||
processStartedAt: string;
|
||||
worker?: WorkspaceWorker;
|
||||
};
|
||||
export type NativeWorkspaceFinalizationWaitReason =
|
||||
| "busy"
|
||||
| "owner_active"
|
||||
| "platform_stop_pending"
|
||||
| "owner_unverified"
|
||||
| "binding_changed";
|
||||
|
||||
export type NativeWorkspaceFinalizationOwnership = {
|
||||
token: string;
|
||||
/** Recheck both the physical lock connection and the durable owner before publishing. */
|
||||
@@ -72,7 +81,7 @@ function readOwner(value: unknown): Owner | null {
|
||||
export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
input: { db: Db; companyId: string; runId: string; retainedRepairActionId?: string },
|
||||
action: (ownership: NativeWorkspaceFinalizationOwnership) => Promise<T>,
|
||||
): Promise<{ acquired: false; reason?: "owner_unverified" } | { acquired: true; value: T }> {
|
||||
): Promise<{ acquired: false; reason: NativeWorkspaceFinalizationWaitReason } | { acquired: true; value: T }> {
|
||||
const identity = await currentNativeControllerIdentity();
|
||||
const owner: Owner = {
|
||||
token: randomUUID(), controllerBootId: identity.bootId, hostname: os.hostname(), pid: identity.pid,
|
||||
@@ -81,6 +90,7 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
const scope = and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId),
|
||||
input.retainedRepairActionId ? inArray(heartbeatRuns.runtimeMode, ["native", "legacy"]) : eq(heartbeatRuns.runtimeMode, "native"));
|
||||
const owned = and(scope, sql`${heartbeatRuns.runnerProfileJson}->${OWNER_KEY}->>'token' = ${owner.token}`);
|
||||
const runKey = nativeWorkspaceFinalizationLockKey(input.companyId, input.runId);
|
||||
let claimed = false;
|
||||
let actionPromise: Promise<T> | undefined;
|
||||
const finishAcquisition = beginRunPhase("execution.workspace_ownership_wait");
|
||||
@@ -88,7 +98,10 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
return await withDedicatedDbConnection(input.db, (dedicated) => dedicated.transaction(async (lock) => {
|
||||
const rows = await lock.execute(sql`select pg_try_advisory_xact_lock(hashtextextended(${nativeWorkspaceFinalizationLockKey(input.companyId, input.runId)}, 0)) as acquired,
|
||||
(select setting::bigint from pg_settings where name = 'idle_in_transaction_session_timeout') as idle_timeout_ms`);
|
||||
if (rows[0]?.acquired !== true) return { acquired: false } as const;
|
||||
if (rows[0]?.acquired !== true) return {
|
||||
acquired: false,
|
||||
reason: !input.retainedRepairActionId && activeRunOwners.get(runKey)?.db === input.db ? "owner_active" : "busy",
|
||||
} as const;
|
||||
return withWorkspaceLockKeepalive(lock as unknown as Db, Number(rows[0].idle_timeout_ms), async () => {
|
||||
// Reads and durable progress use the ordinary pool, not the lock transaction.
|
||||
const [run] = await input.db.select({ profile: heartbeatRuns.runnerProfileJson,
|
||||
@@ -130,6 +143,7 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
}) : null;
|
||||
};
|
||||
let platformProof: WorkspaceWorkerStopProof | null = null;
|
||||
let platformStopPending = false;
|
||||
if (rawPrior != null) {
|
||||
const requireStopVerification = async () => {
|
||||
if (input.retainedRepairActionId) return { acquired: false, reason: "owner_unverified" } as const;
|
||||
@@ -150,7 +164,7 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
sql`(${issueRecoveryActions.nextAction} is distinct from ${nextAction} or ${issueRecoveryActions.evidence}#>>'{workspaceOwnerRecovery,state}' is distinct from 'operator_required')`,
|
||||
));
|
||||
}
|
||||
return { acquired: false } as const;
|
||||
return { acquired: false, reason: platformStopPending ? "platform_stop_pending" : "owner_unverified" } as const;
|
||||
}
|
||||
if (run.issueId) await recovery.upsertSourceScoped({
|
||||
companyId: input.companyId, sourceIssueId: run.issueId,
|
||||
@@ -161,11 +175,20 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
nextAction,
|
||||
wakePolicy: null, maxAttempts: 1, supersedeOnIdentityChange: true,
|
||||
});
|
||||
return { acquired: false } as const;
|
||||
return { acquired: false, reason: platformStopPending ? "platform_stop_pending" : "owner_unverified" } as const;
|
||||
};
|
||||
// The harness checks the exact owned deployment, including orphaned children.
|
||||
// A live deployment, unsigned answer or older receipt stays fenced.
|
||||
if (prior?.worker) platformProof = await requestWorkspaceWorkerStopProof({ runId: input.runId, ownerToken: prior.token, worker: prior.worker });
|
||||
if (prior?.controllerBootId === identity.bootId && prior.pid === identity.pid
|
||||
&& activeOwnerTokens.has(prior.token)) {
|
||||
return { acquired: false, reason: input.retainedRepairActionId ? "owner_unverified"
|
||||
: activeRunOwners.get(runKey)?.db === input.db ? "owner_active" : "busy" } as const;
|
||||
}
|
||||
if (prior?.worker) {
|
||||
const check = await requestWorkspaceWorkerStopProof({ runId: input.runId, ownerToken: prior.token, worker: prior.worker });
|
||||
platformProof = check.status === "verified" ? check.proof : null;
|
||||
platformStopPending = check.status === "pending";
|
||||
}
|
||||
if (!prior) return requireStopVerification();
|
||||
if (!platformProof && prior.hostname !== owner.hostname) return requireStopVerification();
|
||||
const knownJoinedHere = prior.controllerBootId === identity.bootId
|
||||
@@ -194,6 +217,7 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
// only our token even if the server committed but its response was lost.
|
||||
const claimOwner = async (db: Db) => {
|
||||
activeOwnerTokens.add(owner.token);
|
||||
activeRunOwners.set(runKey, { token: owner.token, db: input.db });
|
||||
claimed = true;
|
||||
const [claimedRun] = await db.update(heartbeatRuns).set({
|
||||
runnerProfileJson: sql`jsonb_set(coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb), array[${OWNER_KEY}], ${JSON.stringify(owner)}::jsonb) || ${JSON.stringify(platformProof ? { nativeWorkspaceOwnerStopProof: platformProof } : {})}::jsonb`,
|
||||
@@ -212,7 +236,14 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
if (claimedRun) automaticResultId = result.id;
|
||||
return claimedRun;
|
||||
}) : await claimOwner(input.db);
|
||||
if (!claimedRun) return { acquired: false } as const;
|
||||
if (!claimedRun) {
|
||||
|
||||
|
||||
activeOwnerTokens.delete(owner.token);
|
||||
if (activeRunOwners.get(runKey)?.token === owner.token) activeRunOwners.delete(runKey);
|
||||
claimed = false;
|
||||
return { acquired: false, reason: "binding_changed" } as const;
|
||||
}
|
||||
if (platformProof && run.issueId) await logActivity(input.db, {
|
||||
companyId: input.companyId, actorType: "system", actorId: "workspace-recovery",
|
||||
action: "issue.workspace_owner_stop_verified", entityType: "issue", entityId: run.issueId,
|
||||
@@ -263,9 +294,11 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
await actionPromise?.catch(() => undefined);
|
||||
if (claimed) {
|
||||
activeOwnerTokens.delete(owner.token);
|
||||
if (activeRunOwners.get(runKey)?.token === owner.token) activeRunOwners.delete(runKey);
|
||||
joinedOwnerTokens.add(owner.token);
|
||||
try {
|
||||
await input.db.transaction(async tx => {
|
||||
await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, runId: input.runId });
|
||||
await notifyNativeStatusReassessment(tx, { companyId: input.companyId, runId: input.runId });
|
||||
await tx.update(heartbeatRuns).set({
|
||||
runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) - ${OWNER_KEY}`,
|
||||
@@ -275,6 +308,8 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
} catch {
|
||||
// A reconnect in this exact controller boot can use the positive join
|
||||
// receipt. A new boot still needs durable barrier or operator proof.
|
||||
} finally {
|
||||
releaseNativeWorkspaceFinalization(input.db, { companyId: input.companyId, runId: input.runId });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
import { and, asc, eq, gt, ne, notExists } from "drizzle-orm";
|
||||
import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
|
||||
import { onNativeWorkspaceFinalizationRelease, type NativeWorkspaceFinalizationScope } from "../native-workspace-finalization-notifications.js";
|
||||
import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
|
||||
import { nativeRunnerOwnershipNotHeldCondition } from "./native-runner-ownership.js";
|
||||
import { attemptNativeWorkspaceFinalization, recordNativeWorkspaceFinalizationFailure } from "./native-workspace-finalizer.js";
|
||||
|
||||
type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number };
|
||||
type AttemptInput = Parameters<typeof attemptNativeWorkspaceFinalization>[0];
|
||||
|
||||
|
||||
const OWNERSHIP_RECHECK_MS = 30_000;
|
||||
|
||||
export function nativeWorkspaceFinalizationWork(db: Db, options: {
|
||||
environmentRuntime?: AttemptInput["environmentRuntime"];
|
||||
pluginWorkerManager?: AttemptInput["pluginWorkerManager"];
|
||||
attempt?: typeof attemptNativeWorkspaceFinalization;
|
||||
onWorkspaceSettled(input: { runId: string; companyId: string; agentId: string; succeeded: boolean }): Promise<void>;
|
||||
}) {
|
||||
function candidates(scope?: NativeWorkspaceFinalizationScope, after?: string) {
|
||||
return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
issueId: nativeRunFinalizations.issueId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
|
||||
leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt,
|
||||
}).from(nativeRunFinalizations)
|
||||
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId),
|
||||
eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId)))
|
||||
.innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId)))
|
||||
.innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId),
|
||||
eq(nativeRunResults.runId, nativeRunFinalizations.runId), eq(nativeRunResults.companyId, nativeRunFinalizations.companyId),
|
||||
eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), eq(nativeRunResults.schemaStatus, "accepted")))
|
||||
.where(and(eq(heartbeatRuns.runtimeMode, "native"), nativeRunnerOwnershipNotHeldCondition(),
|
||||
ne(nativeRunFinalizations.phase, "terminal_failure"),
|
||||
notExists(db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, nativeRunFinalizations.companyId),
|
||||
eq(workspaceOperations.heartbeatRunId, nativeRunFinalizations.runId), eq(workspaceOperations.issueId, nativeRunFinalizations.issueId),
|
||||
eq(workspaceOperations.phase, "workspace_finalize"), eq(workspaceOperations.status, "succeeded"),
|
||||
))),
|
||||
scope ? eq(nativeRunFinalizations.companyId, scope.companyId) : undefined,
|
||||
scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined,
|
||||
scope?.runId ? eq(nativeRunFinalizations.runId, scope.runId) : undefined,
|
||||
after ? gt(nativeRunFinalizations.runId, after) : undefined,
|
||||
)).orderBy(asc(nativeRunFinalizations.runId)).limit(100);
|
||||
}
|
||||
function entry(row: Awaited<ReturnType<typeof candidates>>[number]): Entry {
|
||||
return { id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId,
|
||||
dueAt: Math.max(row.nextAttemptAt?.getTime() ?? 0, row.leaseOwner ? row.leaseExpiresAt?.getTime() ?? 0 : 0) };
|
||||
}
|
||||
async function load(scope?: NativeWorkspaceFinalizationScope, signal?: AbortSignal, canRun = () => true): Promise<Entry[]> {
|
||||
const entries: Entry[] = [];
|
||||
let after: string | undefined;
|
||||
while (!signal?.aborted && canRun()) {
|
||||
const rows = await candidates(scope, after);
|
||||
entries.push(...rows.map(entry));
|
||||
if (rows.length < 100) return entries;
|
||||
after = rows[rows.length - 1].id;
|
||||
}
|
||||
throw new Error("Native workspace finalization discovery stopped");
|
||||
}
|
||||
async function check(current: Entry, signal?: AbortSignal): Promise<Entry | null> {
|
||||
if (signal?.aborted) throw new Error("Native workspace finalization stopped");
|
||||
const scope = { companyId: current.companyId, runId: current.id };
|
||||
const [row] = await candidates(scope);
|
||||
if (!row) return null;
|
||||
const due = entry(row);
|
||||
if (due.dueAt > Date.now()) return due;
|
||||
if (signal?.aborted) throw new Error("Native workspace finalization stopped");
|
||||
const outcome = await (options.attempt ?? attemptNativeWorkspaceFinalization)({
|
||||
db, runId: current.id, environmentRuntime: options.environmentRuntime, pluginWorkerManager: options.pluginWorkerManager,
|
||||
});
|
||||
if (outcome.kind === "retry") return { ...due, dueAt: outcome.retryAt.getTime() };
|
||||
if (outcome.kind === "waiting") {
|
||||
return outcome.reason === "busy" || outcome.reason === "platform_stop_pending"
|
||||
? { ...due, dueAt: Date.now() + OWNERSHIP_RECHECK_MS } : null;
|
||||
}
|
||||
if (outcome.kind !== "completed" || outcome.operation.status === "succeeded") return null;
|
||||
const failure = await recordNativeWorkspaceFinalizationFailure(db, current.id, outcome.operation);
|
||||
if (failure.phase === "terminal_failure") {
|
||||
await options.onWorkspaceSettled({ runId: row.id, companyId: row.companyId, agentId: row.agentId, succeeded: false });
|
||||
return null;
|
||||
}
|
||||
const [remaining] = await candidates(scope);
|
||||
return remaining ? entry(remaining) : null;
|
||||
}
|
||||
function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean,
|
||||
canRecover: () => Promise<boolean> = async () => true) {
|
||||
const worker = registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.nativeWorkspaceFinalization, batchSize: 1,
|
||||
decodeScope: (hint: string): NativeWorkspaceFinalizationScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
|
||||
matches: (item: Entry, scope?: NativeWorkspaceFinalizationScope) => !scope || item.companyId === scope.companyId
|
||||
&& (!scope.issueId || item.issueId === scope.issueId) && (!scope.runId || item.id === scope.runId),
|
||||
load: async (scope, signal) => await canRecover() ? load(scope, signal, canRun) : [],
|
||||
check: async (item, signal) => {
|
||||
if (!canRun()) throw new Error("Native workspace finalization paused");
|
||||
if (!await canRecover()) return null;
|
||||
return check(item, signal);
|
||||
},
|
||||
}, canRun);
|
||||
const stop = onNativeWorkspaceFinalizationRelease(db, scope => worker.invalidate(JSON.stringify(scope)));
|
||||
return { ...worker, stop };
|
||||
}
|
||||
return { load, check, register };
|
||||
}
|
||||
@@ -213,7 +213,7 @@ describe("native workspace finalization recovery", () => {
|
||||
it("releases only the verified owner and resumes the saved result without changing its disposition", async () => {
|
||||
const seed = await seedAbandonedCopyback();
|
||||
const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seed.runId));
|
||||
expect(await withNativeWorkspaceFinalizationOwnership(seed, async () => "unexpected")).toEqual({ acquired: false });
|
||||
expect(await withNativeWorkspaceFinalizationOwnership(seed, async () => "unexpected")).toEqual({ acquired: false, reason: "owner_unverified" });
|
||||
const observed: Array<Promise<unknown>> = [];
|
||||
const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.nativeStatusReassessment, () => {
|
||||
observed.push(db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns)
|
||||
@@ -337,7 +337,7 @@ describe("native workspace finalization recovery", () => {
|
||||
const worker = { provider: "railway" as const, projectId: "p", environmentId: "e", serviceId: "s", deploymentId: "d" };
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seed.runId));
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, nativeWorkspaceFinalizationOwner: { ...seed.owner, worker } } }).where(eq(heartbeatRuns.id, seed.runId));
|
||||
const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: seed.owner.token, worker, stopped: true, iat: 1, exp: 61 });
|
||||
const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: seed.owner.token, worker, stopped: true, iat: 1, exp: 61 } });
|
||||
try { await reconcileNativeFinalizations(db, [seed.runId]); } finally { probe.mockRestore(); }
|
||||
}
|
||||
|
||||
@@ -477,7 +477,7 @@ describe("native workspace finalization recovery", () => {
|
||||
await entered;
|
||||
await db.execute(sql`select pg_terminate_backend(pid) from pg_locks where locktype = 'advisory' and database = (select oid from pg_database where datname = current_database()) and pid <> pg_backend_pid()`);
|
||||
const second = await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "must not run");
|
||||
expect(second).toEqual({ acquired: false });
|
||||
expect(second).toEqual({ acquired: false, reason: "busy" });
|
||||
const [held] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
expect(held.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toBeTruthy();
|
||||
} finally { finish(); }
|
||||
@@ -504,7 +504,7 @@ describe("native workspace finalization recovery", () => {
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...beforeReconnect.runnerProfileJson, nativeWorkspaceFinalizationOwner: unknown } })
|
||||
.where(eq(heartbeatRuns.id, seeded.runId));
|
||||
expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async () => "must not run"))
|
||||
.toEqual({ acquired: false });
|
||||
.toEqual({ acquired: false, reason: "owner_unverified" });
|
||||
}
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: beforeReconnect.runnerProfileJson }).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
expect(await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "resumed"))
|
||||
@@ -529,7 +529,7 @@ describe("native workspace finalization recovery", () => {
|
||||
token: randomUUID(), hostname: os.hostname(), pid: parent.pid, processStartedAt: startedAt,
|
||||
} } }).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
const work = vi.fn(async () => "second physical writer");
|
||||
expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, work)).toEqual({ acquired: false });
|
||||
expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, work)).toEqual({ acquired: false, reason: "owner_unverified" });
|
||||
expect(work).not.toHaveBeenCalled();
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId));
|
||||
expect(action).toMatchObject({ ownerType: "board", cause: "native_workspace_finalization_owner_unverified", wakePolicy: null });
|
||||
@@ -564,7 +564,7 @@ describe("native workspace finalization recovery", () => {
|
||||
await db.update(issueRecoveryActions).set({ fingerprint: nativeSha256({ runId: seed.runId, owner }), evidence: { runId: seed.runId, owner, preserved: "recovery evidence" } }).where(eq(issueRecoveryActions.id, seed.actionId));
|
||||
const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seed.runId));
|
||||
const proof = { v: 1 as const, iss: "paperclip-cloud" as const, aud: "paperclip-workspace-worker-stop/v1" as const, sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true as const, iat: 1, exp: 61 };
|
||||
const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue(proof);
|
||||
const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof });
|
||||
try {
|
||||
const work = vi.fn(async () => {
|
||||
const [recovery] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, seed.actionId));
|
||||
@@ -590,18 +590,18 @@ describe("native workspace finalization recovery", () => {
|
||||
const worker = { provider: "railway" as const, projectId: "p", environmentId: "e", serviceId: "s", deploymentId: "d" };
|
||||
const owner = { ...seed.owner, worker };
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, seed.runId));
|
||||
const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue(null);
|
||||
const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "pending" });
|
||||
const work = vi.fn();
|
||||
try {
|
||||
expect(await withNativeWorkspaceFinalizationOwnership(seed, work)).toEqual({ acquired: false });
|
||||
expect(await withNativeWorkspaceFinalizationOwnership(seed, work)).toEqual({ acquired: false, reason: "platform_stop_pending" });
|
||||
const [recovery] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, seed.issueId), eq(issueRecoveryActions.fingerprint, nativeSha256({ runId: seed.runId, owner }))));
|
||||
expect(recovery.evidence?.workspaceOwnerRecovery).toEqual({ state: "operator_required" });
|
||||
const replacement = { ...owner, token: randomUUID() };
|
||||
probe.mockImplementation(async () => {
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: replacement } }).where(eq(heartbeatRuns.id, seed.runId));
|
||||
return { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 };
|
||||
return { status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 } };
|
||||
});
|
||||
expect(await withNativeWorkspaceFinalizationOwnership(seed, work)).toEqual({ acquired: false });
|
||||
expect(await withNativeWorkspaceFinalizationOwnership(seed, work)).toEqual({ acquired: false, reason: "binding_changed" });
|
||||
expect(work).not.toHaveBeenCalled();
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seed.runId));
|
||||
expect(run.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toEqual(replacement);
|
||||
@@ -635,11 +635,11 @@ describe("native workspace finalization recovery", () => {
|
||||
createdAt: new Date(run.createdAt.getTime() + (changed === "newer run" ? 60_000 : -60_000)),
|
||||
});
|
||||
}
|
||||
return { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 };
|
||||
return { status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 } };
|
||||
});
|
||||
const copyback = vi.fn(async () => fs.writeFile(file, "old sandbox work"));
|
||||
try {
|
||||
expect(await withNativeWorkspaceFinalizationOwnership(seed, copyback)).toEqual({ acquired: false });
|
||||
expect(await withNativeWorkspaceFinalizationOwnership(seed, copyback)).toEqual({ acquired: false, reason: "binding_changed" });
|
||||
expect(copyback).not.toHaveBeenCalled();
|
||||
expect(await fs.readFile(file, "utf8")).toBe("new task work");
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seed.runId));
|
||||
@@ -653,7 +653,7 @@ describe("native workspace finalization recovery", () => {
|
||||
const worker = { provider: "railway" as const, projectId: "p", environmentId: "e", serviceId: "s", deploymentId: "d" };
|
||||
const owner = { ...seed.owner, worker };
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, seed.runId));
|
||||
const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 });
|
||||
const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 } });
|
||||
const write = vi.fn();
|
||||
try {
|
||||
await expect(withNativeWorkspaceFinalizationOwnership(seed, async ownership => {
|
||||
@@ -677,7 +677,7 @@ describe("native workspace finalization recovery", () => {
|
||||
...(changed === "closed" ? { status: "done" } : { executionRunId: null }),
|
||||
}).where(eq(issues.id, seed.issueId));
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, seed.runId));
|
||||
const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 });
|
||||
const probe = vi.spyOn(platformProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof: { v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "stack", runId: seed.runId, ownerToken: owner.token, worker, stopped: true, iat: 1, exp: 61 } });
|
||||
try {
|
||||
await reconcileNativeFinalizations(db, [seed.runId]);
|
||||
expect(probe).toHaveBeenCalledOnce();
|
||||
@@ -692,7 +692,7 @@ describe("native workspace finalization recovery", () => {
|
||||
const owner = { token: randomUUID(), hostname: "other-controller.invalid", pid: 42, processStartedAt: new Date().toISOString() };
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
const work = vi.fn();
|
||||
expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, work)).toEqual({ acquired: false });
|
||||
expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, work)).toEqual({ acquired: false, reason: "owner_unverified" });
|
||||
expect(work).not.toHaveBeenCalled();
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId));
|
||||
expect(action).toMatchObject({ ownerType: "board", cause: "native_workspace_finalization_owner_unverified", wakePolicy: null, evidence: { runId: seeded.runId, owner } });
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { recordNativeFinalizationFailure } from "./native-run-finalizer.js";
|
||||
import { inspectPluginWorkspaceForFinalization } from "./native-plugin-workspace-finalization.js";
|
||||
import type { WorkspaceWorkers } from "../plugin-workspaces.js";
|
||||
import { restoreNativeWorkspaceBestEffort } from "./native-workspace-best-effort.js";
|
||||
@@ -49,7 +50,7 @@ function workspaceSyncFailure(
|
||||
* older runs retain the local workspace validation path. The caller owns the
|
||||
* durable retry and failure policy.
|
||||
*/
|
||||
export async function resumeNativeWorkspaceFinalization(input: {
|
||||
export async function attemptNativeWorkspaceFinalization(input: {
|
||||
db: Db;
|
||||
runId: string;
|
||||
environmentRuntime?: EnvironmentRuntimeService;
|
||||
@@ -95,8 +96,10 @@ export async function resumeNativeWorkspaceFinalization(input: {
|
||||
if (!admission || admission.resultId !== bound.resultId) {
|
||||
throw new Error("native_workspace_finalization_binding_missing");
|
||||
}
|
||||
if (admission.phase === "terminal_failure"
|
||||
|| (admission.nextAttemptAt && admission.nextAttemptAt > new Date())) return null;
|
||||
if (admission.phase === "terminal_failure") return { kind: "inactive" } as const;
|
||||
if (admission.nextAttemptAt && admission.nextAttemptAt > new Date()) {
|
||||
return { kind: "retry", retryAt: admission.nextAttemptAt } as const;
|
||||
}
|
||||
const successful = await input.db.select().from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, bound.companyId),
|
||||
eq(workspaceOperations.heartbeatRunId, input.runId),
|
||||
@@ -105,7 +108,7 @@ export async function resumeNativeWorkspaceFinalization(input: {
|
||||
eq(workspaceOperations.status, "succeeded"),
|
||||
)).orderBy(desc(workspaceOperations.createdAt)).limit(1).then((rows) => rows[0] ?? null);
|
||||
// A stale failure from a pre-fencing controller cannot invalidate exported work.
|
||||
if (successful) return successful;
|
||||
if (successful) return { kind: "completed", operation: successful } as const;
|
||||
const previous = await input.db.select().from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, bound.companyId),
|
||||
eq(workspaceOperations.heartbeatRunId, input.runId),
|
||||
@@ -140,7 +143,7 @@ export async function resumeNativeWorkspaceFinalization(input: {
|
||||
executionWorkspaceId: workspace?.id ?? null,
|
||||
issueId: bound.issueId,
|
||||
});
|
||||
return recorder.recordOperation({
|
||||
const operation = await recorder.recordOperation({
|
||||
phase: "workspace_finalize",
|
||||
cwd,
|
||||
metadata: {
|
||||
@@ -301,6 +304,26 @@ export async function resumeNativeWorkspaceFinalization(input: {
|
||||
};
|
||||
},
|
||||
});
|
||||
return { kind: "completed", operation } as const;
|
||||
});
|
||||
return owned.acquired ? owned.value : { kind: "waiting", reason: owned.reason } as const;
|
||||
}
|
||||
|
||||
|
||||
|
||||
export async function resumeNativeWorkspaceFinalization(input: Parameters<typeof attemptNativeWorkspaceFinalization>[0]) {
|
||||
const result = await attemptNativeWorkspaceFinalization(input);
|
||||
return result.kind === "completed" ? result.operation : null;
|
||||
}
|
||||
|
||||
export function recordNativeWorkspaceFinalizationFailure(db: Db, runId: string, operation: { stderrExcerpt: string | null }) {
|
||||
const unrecoverable = operation.stderrExcerpt?.includes("workspace_sync_out_unrecoverable");
|
||||
const pluginFailure = operation.stderrExcerpt?.trim();
|
||||
const pluginRepair = pluginFailure === "plugin_workspace_branch_mismatch" || pluginFailure === "plugin_workspace_owner_mismatch";
|
||||
return recordNativeFinalizationFailure({
|
||||
db, runId,
|
||||
error: new Error(pluginRepair ? `native_${pluginFailure}` : unrecoverable
|
||||
? "native_workspace_sync_out_unrecoverable" : "native_workspace_sync_out_failed"),
|
||||
projectRunStatus: true, failureScope: "workspace", permanent: unrecoverable,
|
||||
});
|
||||
return owned.acquired ? owned.value : null;
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js";
|
||||
import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js";
|
||||
import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
|
||||
import os from "node:os";
|
||||
@@ -66,6 +67,7 @@ export async function resumeNativeWorkspaceAfterOwnerStop(input: {
|
||||
const now = new Date();
|
||||
const proof = { owner: prior, resultId: result.id, actorId: input.actorId,
|
||||
stopEvidence: confirmation.stopEvidence, confirmedAt: now.toISOString() };
|
||||
await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId });
|
||||
await notifyNativeStatusReassessment(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId });
|
||||
await tx.update(heartbeatRuns).set({
|
||||
runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} - ${OWNER_KEY}`, updatedAt: now,
|
||||
|
||||
@@ -22,10 +22,30 @@ describe("workspace platform stop proof", () => {
|
||||
it("fails closed on live, unknown, unreachable and unsigned platform responses", async () => {
|
||||
for (const response of [{ stopped: false }, { stopped: true }, { receipt: "forged" }]) {
|
||||
const fetchImpl = vi.fn().mockResolvedValue(Response.json(response));
|
||||
expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl })).toBeNull();
|
||||
expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl })).toEqual({ status: "pending" });
|
||||
}
|
||||
expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl: vi.fn().mockRejectedValue(new Error("offline")) })).toBeNull();
|
||||
expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl: vi.fn().mockResolvedValue(Response.json({ receipt: receipt() })) })).toMatchObject({ stopped: true });
|
||||
expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl: vi.fn().mockRejectedValue(new Error("offline")) })).toEqual({ status: "pending" });
|
||||
expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl: vi.fn().mockResolvedValue(Response.json({ receipt: receipt() })) })).toMatchObject({ status: "verified", proof: { stopped: true } });
|
||||
});
|
||||
it("does not schedule platform verification without a supported authenticated endpoint", async () => {
|
||||
for (const unavailable of [
|
||||
{},
|
||||
{ ...env, PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "" },
|
||||
{ ...env, PAPERCLIP_CLOUD_API_ORIGIN: "http://cloud.example" },
|
||||
{ ...env, PAPERCLIP_CLOUD_API_ORIGIN: "https://user:password@cloud.example" },
|
||||
{ ...env, PAPERCLIP_CLOUD_API_ORIGIN: "not a URL" },
|
||||
]) {
|
||||
const fetchImpl = vi.fn();
|
||||
expect(await requestWorkspaceWorkerStopProof(binding, { env: unavailable, now, fetchImpl }))
|
||||
.toEqual({ status: "unsupported" });
|
||||
expect(fetchImpl).not.toHaveBeenCalled();
|
||||
}
|
||||
});
|
||||
it("keeps an authenticated but temporarily unavailable verification pending", async () => {
|
||||
const fetchImpl = vi.fn().mockResolvedValue(new Response(null, { status: 503 }));
|
||||
expect(await requestWorkspaceWorkerStopProof(binding, { env, now, fetchImpl }))
|
||||
.toEqual({ status: "pending" });
|
||||
expect(fetchImpl).toHaveBeenCalledOnce();
|
||||
});
|
||||
it("records a complete platform identity, and does not guess legacy ownership", () => {
|
||||
expect(currentWorkspaceWorker({ RAILWAY_PROJECT_ID: "project", RAILWAY_ENVIRONMENT_ID: "environment", RAILWAY_SERVICE_ID: "service", RAILWAY_DEPLOYMENT_ID: "old-deployment" })).toEqual(worker);
|
||||
|
||||
@@ -43,23 +43,31 @@ export function verifyWorkspaceWorkerStopReceipt(
|
||||
return proof;
|
||||
} catch { return null; }
|
||||
}
|
||||
/** Unsupported platforms, legacy receipts and unavailable evidence never authorize takeover. */
|
||||
export type WorkspaceWorkerStopCheck =
|
||||
| { status: "verified"; proof: WorkspaceWorkerStopProof }
|
||||
| { status: "pending" }
|
||||
| { status: "unsupported" };
|
||||
|
||||
|
||||
export async function requestWorkspaceWorkerStopProof(binding: WorkspaceStopBinding, options: {
|
||||
env?: NodeJS.ProcessEnv; fetchImpl?: typeof fetch; now?: Date;
|
||||
} = {}): Promise<WorkspaceWorkerStopProof | null> {
|
||||
} = {}): Promise<WorkspaceWorkerStopCheck> {
|
||||
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 null;
|
||||
if (!context?.stackId || !context.cloudOrigin || !token || !validWorker(binding.worker)) 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" };
|
||||
try {
|
||||
const origin = new URL(context.cloudOrigin);
|
||||
if (origin.protocol !== "https:" || origin.username || origin.password) return null;
|
||||
const response = await (options.fetchImpl ?? fetch)(`${origin.origin}/v1/tenant/workspace-worker-stop`, {
|
||||
method: "POST", redirect: "error", headers: { authorization: `Bearer ${token}`, "x-paperclip-cloud-stack-id": context.stackId, "content-type": "application/json" },
|
||||
body: JSON.stringify(binding), signal: AbortSignal.timeout(10_000),
|
||||
});
|
||||
if (!response.ok) return null;
|
||||
if (!response.ok) return { status: "pending" };
|
||||
const payload = await response.json() as { receipt?: unknown };
|
||||
return typeof payload.receipt === "string" ? verifyWorkspaceWorkerStopReceipt(payload.receipt, binding, env, options.now) : null;
|
||||
} catch { return null; }
|
||||
const proof = typeof payload?.receipt === "string"
|
||||
? verifyWorkspaceWorkerStopReceipt(payload.receipt, binding, env, options.now) : null;
|
||||
return proof ? { status: "verified", proof } : { status: "pending" };
|
||||
} catch { return { status: "pending" }; }
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js";
|
||||
import { createAgentIdentityRedactor } from "../agent-identity-redaction.js";
|
||||
import { and, asc, eq, gt, or } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
@@ -408,6 +409,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
ignoredAttentionRequests: normalizationDiagnostics.ignoredAttentionRequests,
|
||||
},
|
||||
} as unknown as Record<string, unknown>;
|
||||
await notifyNativeWorkspaceFinalization(tx, this.#binding);
|
||||
const inserted = await tx.insert(nativeRunResults).values({
|
||||
companyId: this.#binding.companyId,
|
||||
issueId: this.#binding.issueId,
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export type NativeWorkspaceFinalizationScope = { companyId: string; runId?: string; issueId?: string };
|
||||
|
||||
export function notifyNativeWorkspaceFinalization(tx: object, scope?: NativeWorkspaceFinalizationScope) {
|
||||
return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeWorkspaceFinalization,
|
||||
scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*");
|
||||
}
|
||||
|
||||
|
||||
|
||||
const listeners = new WeakMap<object, Set<(scope: NativeWorkspaceFinalizationScope) => void>>();
|
||||
export function onNativeWorkspaceFinalizationRelease(db: object, listener: (scope: NativeWorkspaceFinalizationScope) => void) {
|
||||
let owned = listeners.get(db);
|
||||
if (!owned) { owned = new Set(); listeners.set(db, owned); }
|
||||
owned.add(listener);
|
||||
return () => {
|
||||
owned.delete(listener);
|
||||
if (!owned.size && listeners.get(db) === owned) listeners.delete(db);
|
||||
};
|
||||
}
|
||||
export function releaseNativeWorkspaceFinalization(db: object, scope: NativeWorkspaceFinalizationScope) {
|
||||
for (const listener of listeners.get(db) ?? []) listener(scope);
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
|
||||
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
@@ -480,6 +481,8 @@ export function workspaceOperationService(db: Db) {
|
||||
async function writeState<T>(write: (database: Db) => Promise<T>): Promise<T> {
|
||||
if (recordInput.phase !== "workspace_finalize" || !input.heartbeatRunId) return write(db);
|
||||
return db.transaction(async tx => {
|
||||
await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId,
|
||||
runId: input.heartbeatRunId!, issueId: input.issueId ?? undefined });
|
||||
await notifyNativeStatusReassessment(tx, { companyId: input.companyId,
|
||||
runId: input.heartbeatRunId!, issueId: input.issueId ?? undefined });
|
||||
return write(tx as unknown as Db);
|
||||
|
||||
@@ -1079,11 +1079,11 @@ describe("automatic retained workspace recovery", () => {
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...source.runnerProfileJson, nativeWorkspaceFinalizationOwner: prior } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
// This is the verified platform boundary. Signature and identity validation
|
||||
// have independent coverage in native-workspace-platform-proof.test.ts.
|
||||
const proof = vi.spyOn(workerProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({
|
||||
const proof = vi.spyOn(workerProof, "requestWorkspaceWorkerStopProof").mockResolvedValue({ status: "verified", proof: {
|
||||
v: 1, iss: "paperclip-cloud", aud: "paperclip-workspace-worker-stop/v1", sub: "test-stack", stopped: true,
|
||||
iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 30,
|
||||
runId: f.runId, ownerToken: prior.token, worker,
|
||||
});
|
||||
} });
|
||||
try {
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(proof).toHaveBeenCalledWith({ runId: f.runId, ownerToken: prior.token, worker });
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
|
||||
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { beginIdleTrackedWork } from "./task-admission.js";
|
||||
@@ -247,6 +248,7 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ
|
||||
const next = { ...current!, revision: current!.revision + 1, phase: "reconcile_effects" as const,
|
||||
state: "complete" as const, nextAttemptAt: null };
|
||||
const activity = await db.transaction(async tx => {
|
||||
await notifyNativeWorkspaceFinalization(tx, input);
|
||||
await notifyNativeStatusReassessment(tx, input);
|
||||
await tx.insert(workspaceOperations).values({ companyId: input.companyId, issueId: input.issueId,
|
||||
heartbeatRunId: input.runId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date(),
|
||||
|
||||
Reference in new issue
Block a user