mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 14:16:39 +01:00
516 lines
21 KiB
TypeScript
516 lines
21 KiB
TypeScript
import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
|
|
import { notifyExecutionRunWork } from "../execution-control-notifications.js";
|
|
import { logger } from "../../middleware/logger.js";
|
|
import { createHash, randomUUID } from "node:crypto";
|
|
import { stat } from "node:fs/promises";
|
|
import { and, eq, inArray, isNull, sql } from "drizzle-orm";
|
|
import {
|
|
agentWakeupRequests,
|
|
environmentLeases,
|
|
heartbeatRunEvents,
|
|
heartbeatRuns,
|
|
issues,
|
|
issueRecoveryActions,
|
|
nativeRunFinalizations,
|
|
toolInvocations,
|
|
type Db,
|
|
} from "@paperclipai/db";
|
|
import { decideNativeReplacement } from "./native-replacement-evidence.js";
|
|
import { issueService } from "../issues.js";
|
|
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
|
import { buildExecutionContinuation } from "../execution-continuation.js";
|
|
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
|
|
|
|
export const NATIVE_SAFE_REPLACEMENT_REASON = "native_safe_replacement";
|
|
const record = (v: unknown): Record<string, unknown> =>
|
|
v && typeof v === "object" && !Array.isArray(v)
|
|
? (v as Record<string, unknown>)
|
|
: {};
|
|
const processAlive = (pid: number | null) => {
|
|
if (pid === null) return false;
|
|
try {
|
|
process.kill(pid, 0);
|
|
return true;
|
|
} catch (error) {
|
|
return (error as NodeJS.ErrnoException).code !== "ESRCH";
|
|
}
|
|
};
|
|
|
|
export function nativeReplacementPending() {
|
|
return and(
|
|
eq(heartbeatRuns.status, "failed"),
|
|
eq(nativeRunFinalizations.phase, "terminal_failure"),
|
|
inArray(
|
|
nativeRunFinalizations.failureCode,
|
|
["native_provider_terminal_failed", "native_session_cleanup_quarantined", "provider_transport_failed"],
|
|
),
|
|
isNull(nativeRunFinalizations.resultId),
|
|
sql`coalesce(${nativeRunFinalizations.failureDetail}->>'successorRunId', '') = ''`,
|
|
sql`coalesce(${nativeRunFinalizations.failureDetail}->>'replacementDenied', '') = ''`,
|
|
);
|
|
}
|
|
|
|
/** Retryable outbox: terminal_failure plus a new-format failure code is the durable candidate. */
|
|
export async function reconcileSafeNativeReplacements(
|
|
db: Db,
|
|
now = new Date(),
|
|
options: {
|
|
verifyStoppedSession?: (run: typeof heartbeatRuns.$inferSelect) => Promise<{
|
|
evidence: Record<string, unknown>;
|
|
retire: () => boolean;
|
|
} | null>;
|
|
/** Test fault injection at durability boundaries; never exposed by an API. */
|
|
failpoint?: (phase: "successor_inserted" | "lineage_committed") => void;
|
|
} = {},
|
|
) {
|
|
const candidates = await db
|
|
.select({ run: heartbeatRuns, coordinator: nativeRunFinalizations })
|
|
.from(heartbeatRuns)
|
|
.innerJoin(
|
|
nativeRunFinalizations,
|
|
eq(nativeRunFinalizations.runId, heartbeatRuns.id),
|
|
)
|
|
.where(
|
|
nativeReplacementPending(),
|
|
)
|
|
.limit(25);
|
|
let scheduled = 0;
|
|
for (const { run, coordinator } of candidates) {
|
|
try {
|
|
if (
|
|
coordinator.failureDetail?.successorRunId ||
|
|
coordinator.failureDetail?.replacementDenied
|
|
)
|
|
continue;
|
|
const execution = record(
|
|
record(run.runnerProfileJson).nativeExecutionInput,
|
|
);
|
|
const workspace = record(execution.workspace);
|
|
const leases = await db
|
|
.select()
|
|
.from(environmentLeases)
|
|
.where(
|
|
and(
|
|
eq(environmentLeases.companyId, run.companyId),
|
|
eq(environmentLeases.heartbeatRunId, run.id),
|
|
),
|
|
);
|
|
// Teardown is still in progress. The next sweep rechecks its durable outcome.
|
|
if (leases.some((lease) => lease.releasedAt === null)) continue;
|
|
const stoppedSession = coordinator.failureCode !== "native_provider_terminal_failed"
|
|
? await options.verifyStoppedSession?.(run) ?? null : null;
|
|
// A transport label alone is not evidence. Keep inspecting these candidates
|
|
// as process cleanup and the final transcript become durable.
|
|
if (coordinator.failureCode !== "native_provider_terminal_failed" && !stoppedSession) continue;
|
|
const invocations = await db
|
|
.select()
|
|
.from(toolInvocations)
|
|
.where(
|
|
and(
|
|
eq(toolInvocations.companyId, run.companyId),
|
|
eq(toolInvocations.runId, run.id),
|
|
),
|
|
);
|
|
const events = await db
|
|
.select({
|
|
eventType: heartbeatRunEvents.eventType,
|
|
payload: heartbeatRunEvents.payload,
|
|
})
|
|
.from(heartbeatRunEvents)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRunEvents.companyId, run.companyId),
|
|
eq(heartbeatRunEvents.runId, run.id),
|
|
inArray(heartbeatRunEvents.eventType, [
|
|
"harness.diagnostic",
|
|
"tool.execution.started",
|
|
"workspace.changed",
|
|
"workspace.change.updated",
|
|
"delegation.started",
|
|
"terminal.input.sent",
|
|
"artifact.generated",
|
|
"item.started",
|
|
]),
|
|
),
|
|
);
|
|
const safeControlReads = new Set([
|
|
"connections_search",
|
|
"connection_request",
|
|
"paperclip_get_agent",
|
|
"paperclip_get_issue",
|
|
"paperclip_list_issues",
|
|
"paperclip_read_document",
|
|
]);
|
|
const uncertainProviderActions = events.flatMap((event) => {
|
|
const envelope = record(event.payload);
|
|
const p = envelope.prpEvent
|
|
? record(record(envelope.prpEvent).payload)
|
|
: envelope;
|
|
if (event.eventType === "harness.diagnostic")
|
|
return p.classification === "descendant"
|
|
? [`descendant ${String(p.receivedThreadId ?? "unknown")}`]
|
|
: [];
|
|
if (
|
|
[
|
|
"workspace.changed",
|
|
"workspace.change.updated",
|
|
"delegation.started",
|
|
"terminal.input.sent",
|
|
"artifact.generated",
|
|
].includes(event.eventType)
|
|
)
|
|
return [event.eventType];
|
|
if (event.eventType === "tool.execution.started") {
|
|
const name = typeof p.name === "string" ? p.name : "unknown tool";
|
|
const completedTaskControlCallIds = stoppedSession?.evidence.completedTaskControlCallIds;
|
|
if (name === "paperclip_finish" && Array.isArray(completedTaskControlCallIds) &&
|
|
completedTaskControlCallIds.includes(p.executionId)) return [];
|
|
const receiptedRead = invocations.some(
|
|
(row) =>
|
|
(row.id === p.executionId ||
|
|
row.correlationId === p.executionId) &&
|
|
row.toolName === name &&
|
|
row.riskLevel === "read" &&
|
|
row.status === "succeeded" &&
|
|
row.resultHash,
|
|
);
|
|
return p.transport === "process" ||
|
|
(!receiptedRead && !safeControlReads.has(name))
|
|
? [String(p.executionId ?? name)]
|
|
: [];
|
|
}
|
|
return [
|
|
"command_execution",
|
|
"commandExecution",
|
|
"file_change",
|
|
"fileChange",
|
|
"collabAgentToolCall",
|
|
"subAgentActivity",
|
|
].includes(String(p.kind ?? p.type))
|
|
? [String(p.kind ?? p.type)]
|
|
: [];
|
|
});
|
|
let historyComplete = false;
|
|
try {
|
|
const continuation = await buildExecutionContinuation({
|
|
db,
|
|
companyId: run.companyId,
|
|
issueId: coordinator.issueId,
|
|
agentId: run.agentId,
|
|
context: record(run.contextSnapshot),
|
|
summary: null,
|
|
exposeLowTrustRaw: false,
|
|
});
|
|
if (continuation.unresolvedInteractionIds.length > 0) continue;
|
|
historyComplete = true;
|
|
} catch {
|
|
/* The named operator outcome retains incomplete context. */
|
|
}
|
|
const decision = decideNativeReplacement({
|
|
failedSession: true,
|
|
// A facade transport failure can conceal an authoritative protocol
|
|
// rejection. A stopped process and read receipts do not resolve that.
|
|
failureMeaningKnown:
|
|
Boolean(stoppedSession) || (typeof coordinator.failureDetail?.originalFailureCode === "string" &&
|
|
![
|
|
"notification_transport_failed",
|
|
"provider_turn_failed",
|
|
"native_provider_terminal_failed",
|
|
].includes(coordinator.failureDetail.originalFailureCode)),
|
|
predecessorFenced:
|
|
coordinator.leaseOwner === null && run.status === "failed",
|
|
providerStopped:
|
|
!processAlive(run.processPid) &&
|
|
!processAlive(run.processGroupId ? -run.processGroupId : null),
|
|
workspacePreserved:
|
|
typeof workspace.cwd === "string" &&
|
|
(await stat(workspace.cwd).then(
|
|
(s) => s.isDirectory(),
|
|
() => false,
|
|
)),
|
|
historyComplete,
|
|
effectInventoryComplete:
|
|
Boolean(stoppedSession) || (record(run.runnerProfileJson).recoveryEventInventoryVersion === 1 &&
|
|
record(execution.provider).kind === "codex"),
|
|
attempts: coordinator.attempt,
|
|
invocations,
|
|
apiReceipts: record(record(run.resultJson).apiToolReceipts),
|
|
uncertainProviderActions,
|
|
});
|
|
if (!decision.allowed) {
|
|
// Lack of containment can be temporary; do not prevent the next proof.
|
|
if (decision.cause === "provider_ownership_unverified") continue;
|
|
await db.transaction(async (tx) => {
|
|
await tx.execute(
|
|
sql`select set_config('statement_timeout', '15000', true), set_config('lock_timeout', '1000', true)`,
|
|
);
|
|
const [task] = await tx
|
|
.select()
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.id, coordinator.issueId),
|
|
eq(issues.companyId, run.companyId),
|
|
),
|
|
)
|
|
.for("update");
|
|
const [current] = await tx
|
|
.select()
|
|
.from(nativeRunFinalizations)
|
|
.where(eq(nativeRunFinalizations.runId, run.id))
|
|
.for("update");
|
|
if (
|
|
!current ||
|
|
current.phase !== "terminal_failure" ||
|
|
current.failureDetail?.successorRunId ||
|
|
current.failureDetail?.replacementDenied
|
|
)
|
|
return;
|
|
await notifyExecutionRunWork(tx);
|
|
await tx
|
|
.update(nativeRunFinalizations)
|
|
.set({
|
|
failureDetail: {
|
|
...current.failureDetail,
|
|
replacementDenied: decision.cause,
|
|
nextAction: decision.nextAction,
|
|
},
|
|
updatedAt: now,
|
|
})
|
|
.where(eq(nativeRunFinalizations.runId, run.id));
|
|
if (
|
|
!task ||
|
|
task.assigneeAgentId !== run.agentId ||
|
|
["done", "cancelled"].includes(task.status)
|
|
)
|
|
return;
|
|
await issueRecoveryActionService(
|
|
tx as unknown as Db,
|
|
).upsertSourceScoped({
|
|
companyId: run.companyId,
|
|
sourceIssueId: coordinator.issueId,
|
|
kind: "active_run_watchdog",
|
|
ownerType: "board",
|
|
ownerAgentId: null,
|
|
returnOwnerAgentId: run.agentId,
|
|
cause: decision.cause,
|
|
fingerprint: `native-replacement:${run.id}`,
|
|
evidence: { runId: run.id, attempts: coordinator.attempt },
|
|
nextAction: decision.nextAction,
|
|
wakePolicy: null,
|
|
maxAttempts: 3,
|
|
supersedeOnIdentityChange: true,
|
|
});
|
|
});
|
|
continue;
|
|
}
|
|
const created = await db.transaction(async (tx) => {
|
|
await tx.execute(
|
|
sql`select set_config('statement_timeout', '15000', true), set_config('lock_timeout', '1000', true)`,
|
|
);
|
|
const [task] = await tx
|
|
.select()
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.id, coordinator.issueId),
|
|
eq(issues.companyId, run.companyId),
|
|
),
|
|
)
|
|
.for("update");
|
|
const [current] = await tx
|
|
.select()
|
|
.from(nativeRunFinalizations)
|
|
.where(eq(nativeRunFinalizations.runId, run.id))
|
|
.for("update");
|
|
if (
|
|
!task ||
|
|
task.assigneeAgentId !== run.agentId ||
|
|
!["todo", "in_progress", "in_review", "blocked"].includes(task.status) ||
|
|
(task.executionRunId !== null && task.executionRunId !== run.id) ||
|
|
(task.checkoutRunId !== null && task.checkoutRunId !== run.id) ||
|
|
!current ||
|
|
current.phase !== "terminal_failure" ||
|
|
current.failureCode !== coordinator.failureCode ||
|
|
current.failureDetail?.successorRunId ||
|
|
current.failureDetail?.replacementDenied ||
|
|
current.attempt >= 3
|
|
)
|
|
return false;
|
|
if (task.status === "blocked") {
|
|
const failureHolds = await tx.select().from(issueRecoveryActions).where(and(
|
|
eq(issueRecoveryActions.companyId, run.companyId),
|
|
eq(issueRecoveryActions.sourceIssueId, task.id),
|
|
eq(issueRecoveryActions.kind, "active_run_watchdog"),
|
|
eq(issueRecoveryActions.cause, current.failureCode!),
|
|
sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
|
|
)).for("update");
|
|
const ownsBlock = failureHolds.some(hold => {
|
|
const receipt = record(hold.evidence.nativeFailureBlock);
|
|
return receipt.runId === run.id && receipt.statusVersion === task.statusVersion;
|
|
});
|
|
if (!ownsBlock) return false;
|
|
}
|
|
if (stoppedSession) {
|
|
const [currentRun] = await tx.select().from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
|
|
)).for("update");
|
|
if (!currentRun || currentRun.status !== "failed" || currentRun.runnerInstanceId !== run.runnerInstanceId ||
|
|
currentRun.nativeSessionId !== run.nativeSessionId || currentRun.processPid || currentRun.processGroupId) return false;
|
|
}
|
|
if (task.status === "blocked") {
|
|
// Restore only this failure's unchanged projection. The normal issue
|
|
// service still enforces dependency readiness and assignee eligibility.
|
|
await issueService(tx as unknown as Db).update(task.id, { status: "in_progress" }, tx);
|
|
}
|
|
if (stoppedSession) {
|
|
// If the last ownership proof changes, roll back the status restoration.
|
|
if (!stoppedSession.retire()) throw new Error("native_replacement_stopped_session_changed");
|
|
await appendHeartbeatRunEvent(tx as unknown as Db, {
|
|
companyId: run.companyId, runId: run.id, agentId: run.agentId,
|
|
eventType: "native.stopped_text_turn_verified", stream: "system", level: "info",
|
|
message: "The previous runner and provider stopped. The interrupted turn had no external actions; any completion bookkeeping has a verified receipt.",
|
|
payload: stoppedSession.evidence,
|
|
});
|
|
}
|
|
const successorRunId = randomUUID();
|
|
const dueAt = new Date(now.getTime() + 30_000);
|
|
const predecessorContext = { ...record(run.contextSnapshot) };
|
|
// History comes from the failed source run. Consumed wake fields must
|
|
// not grant this automatic retry fresh comment/resume authority.
|
|
for (const key of [
|
|
"explicitUserContinuation", "wakeCommentId", "wakeCommentIds", "commentId",
|
|
"commentIds", "latestCommentId", "resumeIntent", "followUpRequested",
|
|
"paperclipWake", "paperclipWakeComment", "paperclipTaskMarkdown", "paperclipTaskMarkdownCompact",
|
|
"paperclipTaskMarkdownAssignment", "paperclipTaskMarkdownAssignmentCompact", "paperclipTurnContext",
|
|
]) delete predecessorContext[key];
|
|
const context = {
|
|
...predecessorContext,
|
|
issueId: task.id,
|
|
retryOfRunId: run.id,
|
|
wakeReason: NATIVE_SAFE_REPLACEMENT_REASON,
|
|
retryReason: NATIVE_SAFE_REPLACEMENT_REASON,
|
|
forceFreshSession: true,
|
|
recoveryIncidentRootRunId:
|
|
record(run.contextSnapshot).recoveryIncidentRootRunId ?? run.id,
|
|
};
|
|
const [wake] = await tx
|
|
.insert(agentWakeupRequests)
|
|
.values({
|
|
companyId: run.companyId,
|
|
agentId: run.agentId,
|
|
source: "automation",
|
|
triggerDetail: "system",
|
|
reason: NATIVE_SAFE_REPLACEMENT_REASON,
|
|
status: "queued",
|
|
payload: context,
|
|
requestedByActorType: "system",
|
|
idempotencyKey: `native-safe-replacement:${run.id}`,
|
|
})
|
|
.returning();
|
|
await notifyExecutionRunWork(tx);
|
|
await tx.insert(heartbeatRuns).values({
|
|
id: successorRunId,
|
|
companyId: run.companyId,
|
|
agentId: run.agentId,
|
|
invocationSource: "automation",
|
|
triggerDetail: "system",
|
|
status: "scheduled_retry",
|
|
executionStatusDeliveryId: randomUUID(),
|
|
wakeupRequestId: wake!.id,
|
|
responsibleUserId: run.responsibleUserId,
|
|
contextSnapshot: context,
|
|
retryOfRunId: run.id,
|
|
scheduledRetryAt: dueAt,
|
|
scheduledRetryAttempt: current.attempt,
|
|
scheduledRetryReason: NATIVE_SAFE_REPLACEMENT_REASON,
|
|
});
|
|
options.failpoint?.("successor_inserted");
|
|
await tx
|
|
.update(agentWakeupRequests)
|
|
.set({ runId: successorRunId })
|
|
.where(eq(agentWakeupRequests.id, wake!.id));
|
|
await tx
|
|
.update(nativeRunFinalizations)
|
|
.set({
|
|
failureDetail: {
|
|
...current.failureDetail,
|
|
...(stoppedSession ? { stoppedTextTurn: stoppedSession.evidence } : {}),
|
|
successorRunId,
|
|
nextAction:
|
|
"Continue in the linked fresh provider session after the retry delay.",
|
|
},
|
|
nextAttemptAt: dueAt,
|
|
updatedAt: now,
|
|
})
|
|
.where(eq(nativeRunFinalizations.runId, run.id));
|
|
await issueRecoveryActionService(
|
|
tx as unknown as Db,
|
|
).resolveActiveForIssue({
|
|
companyId: run.companyId,
|
|
sourceIssueId: task.id,
|
|
kind: "active_run_watchdog",
|
|
fingerprint: createHash("sha256")
|
|
.update(`${run.id}:${current.failureCode}`)
|
|
.digest("hex"),
|
|
status: "resolved",
|
|
outcome: "handed_back",
|
|
resolutionNote: `Safe continuation is scheduled in run ${successorRunId}.`,
|
|
});
|
|
// A previous sweep can have resolved the UI bookkeeping while retaining
|
|
// an effective no-replay hold. Retire only this exact failure's hold in
|
|
// the same transaction as its verified successor; unrelated holds remain.
|
|
const holds = await tx.select().from(issueRecoveryActions).where(and(
|
|
eq(issueRecoveryActions.companyId, run.companyId),
|
|
eq(issueRecoveryActions.sourceIssueId, task.id),
|
|
eq(issueRecoveryActions.kind, "active_run_watchdog"),
|
|
eq(issueRecoveryActions.cause, current.failureCode!),
|
|
sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
|
|
)).for("update");
|
|
if (holds.some(hold => hold.status === "active" || hold.status === "escalated")) {
|
|
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId: task.id });
|
|
}
|
|
for (const hold of holds) {
|
|
await tx.update(issueRecoveryActions).set({
|
|
status: "resolved", outcome: "handed_back", resolvedAt: now, updatedAt: now,
|
|
wakePolicy: null, monitorPolicy: null,
|
|
nextAction: `Safe continuation is scheduled in run ${successorRunId}.`,
|
|
resolutionNote: "Verified process termination and action receipts allow a fresh session to continue.",
|
|
evidence: { ...hold.evidence, verifiedReplacement: { successorRunId, recordedAt: now.toISOString() },
|
|
...(hold.evidence.automaticRecovery ? { automaticRecovery: {
|
|
...record(hold.evidence.automaticRecovery), replay: "verified_safe_replacement", successorRunId,
|
|
} } : {}),
|
|
},
|
|
}).where(eq(issueRecoveryActions.id, hold.id));
|
|
}
|
|
await tx
|
|
.update(heartbeatRuns)
|
|
.set({ executionStatusDeliveryId: randomUUID() })
|
|
.where(eq(heartbeatRuns.id, run.id));
|
|
// The successor gets task authority only through the existing admission gate.
|
|
await tx
|
|
.update(issues)
|
|
.set({ executionRunId: null, checkoutRunId: null, updatedAt: now })
|
|
.where(
|
|
and(
|
|
eq(issues.id, task.id),
|
|
eq(issues.assigneeAgentId, run.agentId),
|
|
),
|
|
);
|
|
return true;
|
|
});
|
|
if (created) {
|
|
options.failpoint?.("lineage_committed");
|
|
scheduled += 1;
|
|
}
|
|
} catch (error) {
|
|
if (options.failpoint) throw error;
|
|
logger.warn(
|
|
{ runId: run.id },
|
|
"Native replacement remains pending; another sweep will retry its durable decision",
|
|
);
|
|
}
|
|
}
|
|
return { scanned: candidates.length, scheduled };
|
|
}
|