mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 06:04:11 +01:00
Preserve native workspace recovery across Stop and sandbox reuse (#18332)
Land the thirteen-file native recovery fix independently from #18245. Retain verified Stop evidence and queued user comments. Retire stale preparation ownership atomically and require paired sandbox lease handoff receipts before workspace recovery. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
75666224d4
commit
f5e5b05a33
13 files changed
+1220
-28
No files matched your search
@@ -1,3 +1,8 @@
|
||||
import { hasRequiredWorkspaceRecovery } from "../../../services/workspace-restore-recovery-state.js";
|
||||
import { readNativeWorkspaceSyncReference } from "../../../services/native-runtime/native-workspace-sync.js";
|
||||
import { isPreDispatchExecutionWaitVerified } from "../../../services/pre-dispatch-execution-wait.js";
|
||||
import { isUserCommentContinuationReason } from "../../../services/explicit-native-continuation.js";
|
||||
import { queuedCommentIdsFromRunContext, withQueuedCommentIdsInWakePayload } from "../../../services/issue-queued-comment-queue.js";
|
||||
import { notifyExternalAgentSettlement } from "../../external-agents/index.js";
|
||||
import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js";
|
||||
import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js";
|
||||
@@ -7,7 +12,7 @@ import { hasConversationContinuationPolicy } from "../../../services/conversatio
|
||||
import { getExecutionBlocker } from "../../../services/execution-blocker.js";
|
||||
import { validateReconstructedWorkspaceDispatch } from "../../../services/native-runtime/native-workspace-reconstruction.js";
|
||||
import { getNativeReviewAssignment } from "../../../services/native-runtime/native-review-participant.js";
|
||||
import { and, asc, eq, gte, inArray, lte, or, sql } from "drizzle-orm";
|
||||
import { and, asc, eq, gte, inArray, isNull, lte, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
agentWakeupRequests,
|
||||
@@ -22,7 +27,7 @@ import {
|
||||
nativeRunFinalizations,
|
||||
projects,
|
||||
} from "@paperclipai/db";
|
||||
import { ISSUE_DISPOSITION_REPAIR_RETRY_REASON } from "@paperclipai/shared";
|
||||
import { ISSUE_DISPOSITION_REPAIR_RETRY_REASON, isUuidLike } from "@paperclipai/shared";
|
||||
import { withAccountingTransaction } from "../../../services/accounting-transaction.js";
|
||||
import type { ActivityPublication } from "../../../services/activity-log.js";
|
||||
import { parseObject } from "../../../adapters/utils.js";
|
||||
@@ -918,6 +923,80 @@ export function createPostgresRunDispatchAdapter(
|
||||
return transactionResult.outcome;
|
||||
}
|
||||
|
||||
async function deferUnstartedUserComment(tx: Db, run: HeartbeatRun, issueId: string, now: Date) {
|
||||
if (!run.wakeupRequestId || !await isPreDispatchExecutionWaitVerified(tx, run)) return null;
|
||||
const wait = parseObject(run.resultJson?.executionWait);
|
||||
const [preservation] = await tx.select().from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
|
||||
sql`${issueRecoveryActions.id}::text = ${wait.recoveryActionId}`,
|
||||
)).for("update");
|
||||
const [source] = preservation ? await tx.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId),
|
||||
eq(heartbeatRuns.nativeIssueId, issueId), eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "cancelled"),
|
||||
sql`${heartbeatRuns.id}::text = ${preservation.evidence.runId}`,
|
||||
)) : [];
|
||||
const baseline = readNativeWorkspaceSyncReference(source?.runnerProfileJson?.nativeWorkspaceSync);
|
||||
if (!preservation || !source || !hasRequiredWorkspaceRecovery(source.resultJson) || !baseline || baseline.state === "finalized" ||
|
||||
preservation.fingerprint !== `native-workspace:${source.id}`) return null;
|
||||
const context = parseObject(run.contextSnapshot), commentId = deriveCommentId(context);
|
||||
const commentIds = queuedCommentIdsFromRunContext(context);
|
||||
if (!commentId || !commentIds.includes(commentId) || !isUserCommentContinuationReason(readNonEmptyString(context.wakeReason))) return null;
|
||||
const [wake] = await tx.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
|
||||
eq(agentWakeupRequests.id, run.wakeupRequestId), eq(agentWakeupRequests.runId, run.id),
|
||||
inArray(agentWakeupRequests.status, ["queued", "claimed"]),
|
||||
eq(agentWakeupRequests.requestedByActorType, "user"),
|
||||
)).for("update");
|
||||
const payload = parseObject(wake?.payload);
|
||||
const originalCommentId = readNonEmptyString(payload.commentId);
|
||||
if (!wake?.requestedByActorId || payload.issueId !== issueId ||
|
||||
!originalCommentId || !commentIds.includes(originalCommentId)) return null;
|
||||
const comments = await tx.select().from(issueComments).where(and(
|
||||
eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId),
|
||||
inArray(issueComments.id, commentIds), eq(issueComments.authorType, "user"),
|
||||
sql`${issueComments.deletedAt} is null`, sql`${issueComments.createdByRunId} is null`,
|
||||
));
|
||||
if (comments.length !== commentIds.length || comments.some(comment =>
|
||||
!comment.body.trim() || !comment.authorUserId) ||
|
||||
comments.find(comment => comment.id === originalCommentId)?.authorUserId !== wake.requestedByActorId) return null;
|
||||
const sourceIds = context.queuedCommentSourceWakeIds;
|
||||
if (sourceIds !== undefined && (!Array.isArray(sourceIds) || sourceIds.some(id => typeof id !== "string" || !isUuidLike(id)))) return null;
|
||||
const sourceWakeIds = Array.isArray(sourceIds) ? [...new Set(sourceIds as string[])] : [];
|
||||
const adopted = sourceWakeIds.length ? await tx.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
|
||||
eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.status, "coalesced"),
|
||||
inArray(agentWakeupRequests.id, sourceWakeIds),
|
||||
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
|
||||
)).for("update") : [];
|
||||
if (adopted.length !== sourceWakeIds.length) return null;
|
||||
const mergeTargets = [wake.id, ...adopted.map(receipt => receipt.id)];
|
||||
const coalesced = await tx.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
|
||||
eq(agentWakeupRequests.status, "coalesced"), eq(agentWakeupRequests.requestedByActorType, "user"),
|
||||
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
|
||||
or(and(eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.reason, "issue_execution_same_name")),
|
||||
and(or(isNull(agentWakeupRequests.runId), eq(agentWakeupRequests.runId, run.id)),
|
||||
inArray(sql<string>`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId'`, mergeTargets))),
|
||||
)).for("update");
|
||||
const authors = new Map(comments.map(comment => [comment.id, comment.authorUserId]));
|
||||
const recordedCommentIds = new Set([wake, ...adopted, ...coalesced].flatMap(receipt => {
|
||||
const recordedPayload = parseObject(receipt.payload), actorId = readNonEmptyString(receipt.requestedByActorId);
|
||||
const recordedCommentId = readNonEmptyString(recordedPayload.commentId);
|
||||
return receipt.requestedByActorType === "user" && recordedPayload.issueId === issueId && actorId &&
|
||||
recordedCommentId && authors.get(recordedCommentId) === actorId ? [recordedCommentId] : [];
|
||||
}));
|
||||
if (commentIds.some(id => !recordedCommentIds.has(id))) return null;
|
||||
const [deferred] = await tx.update(agentWakeupRequests).set({
|
||||
status: "deferred_issue_execution", runId: null, claimedAt: null, finishedAt: null, error: null,
|
||||
payload: withQueuedCommentIdsInWakePayload({ ...payload, _paperclipWakeContext: context,
|
||||
executionWait: { ...parseObject(run.resultJson?.executionWait), reason: "execution_recovery",
|
||||
message: "Waiting for execution recovery. Your message is saved." } }, commentIds), updatedAt: now,
|
||||
}).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, run.companyId),
|
||||
eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, run.id),
|
||||
inArray(agentWakeupRequests.status, ["queued", "claimed"]))).returning({ id: agentWakeupRequests.id });
|
||||
return deferred?.id ?? null;
|
||||
}
|
||||
|
||||
async function cancelStaleRunInTx(
|
||||
tx: Db,
|
||||
run: HeartbeatRun,
|
||||
@@ -969,7 +1048,9 @@ export function createPostgresRunDispatchAdapter(
|
||||
// this write must not overwrite whatever status won it.
|
||||
if (!row) return { outcome: "lost_race" };
|
||||
|
||||
if (row.wakeupRequestId) {
|
||||
const deferredWakeRequestId = expectedStatus === "queued" && decision.errorCode === "execution_reconciliation_required"
|
||||
? await deferUnstartedUserComment(tx, row, issueId, now) : null;
|
||||
if (row.wakeupRequestId && !deferredWakeRequestId) {
|
||||
await tx
|
||||
.update(agentWakeupRequests)
|
||||
.set({ status: "skipped", finishedAt: now, error: decision.reason, updatedAt: now })
|
||||
@@ -977,6 +1058,9 @@ export function createPostgresRunDispatchAdapter(
|
||||
and(
|
||||
eq(agentWakeupRequests.id, row.wakeupRequestId),
|
||||
eq(agentWakeupRequests.companyId, row.companyId),
|
||||
eq(agentWakeupRequests.agentId, row.agentId),
|
||||
eq(agentWakeupRequests.runId, row.id),
|
||||
inArray(agentWakeupRequests.status, ["queued", "claimed"]),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user