mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-29 04:25:52 +01:00
DottaandPaperclip
4808a3a2b1
<!-- entire-trail-link-start --> https://entire.com/gh/paperclipai/-/trails/69/changes/paperclip/69 <!-- entire-trail-link-end --> ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Operator Stop must revoke task authority before it stops provider execution. > - Its transaction locks the task and then the run. A concurrent writer can hold the run while it waits for the task. > - PostgreSQL can abort the Stop transaction with a confirmed deadlock. The cancel API then returns 500. > - This pull request retries only that aborted database transaction, at most three times. > - Provider Stop and activity publication stay after commit, so the retry cannot repeat those effects. ## Linked Issues or Issue Description Refs #17639. Its agent-chat project browser test completed the product assertions, then failed during cleanup when the cancel API returned 500. A real PostgreSQL lock-cycle test reproduces the same `40P01` error in the Stop transaction. Related: #13911 addresses audit foreign-key lock contention. This change handles a different concurrent run-writer path without changing the schema or task/run lock order. Stack base: #17639. This is a separate correctness repair for the sleep migration stack, not another polling migration. ## What Changed - Control the runner timeout regression fixture's clock so it reaches the pending append before expiry; retain production cancellation and late-append assertions. - Retry the Stop authority transaction only after PostgreSQL reports `40P01`. Make at most three attempts, with short bounded backoff. - Discard rolled-back local run state, claim flags, and queued activity publications before each retry. Re-read current authority under fresh locks. - Keep provider cancellation and activity publication outside the retried transaction. Preserve other errors and the existing native lock-conflict response. - Add a real two-transaction deadlock test. Add server-side failure tests for retry exhaustion and errors that must not retry. - Let run-control tests use a validated disposable PostgreSQL database URL. ## Verification Published head: `60ba892ae9fc5faaf8cd0833694b4a66c13323c3`, targeting master after #17639 merged at `1ed0af41867ba40fd1d23292d395588d18c2fcbe`. The normal ancestry merge preserves the complete five-path tested cancellation patch byte-for-byte, including the review repair, and all four unrelated current master paths. The production retry logic remains unchanged from this slice's reviewed implementation; the follow-up adds coverage and its audit. - **228 distinct complete cases pass:** all 48 run-control cases and all 180 native-session runtime cases. The two new targeted cases are included in 48 and are not counted again. Server typechecking, build and module-boundary checks also pass. - The new real PostgreSQL cases reach a failed native run's correlated Stop claim, cancellation fence, board handoff retirement and activity write before raising `40P01`. One case preserves retry authority and commits exactly one cancellation activity/live notice. The other changes coordinator authority between attempts, returns 409, preserves the failed run and handoff, and performs no provider/workspace cleanup. - A negative control removes only the publication-list reset. It fails with two live notices for one durable activity. Production was restored byte-for-byte before final validation. The existing actual two-connection deadlock, bounded attempts, non-deadlock rejection and post-commit provider behavior remain covered. - Preserve the initial new-assertion typecheck failure (optional live-event payload) and the corrected final pass. Service-local test files are typechecked; `src/__tests__` is excluded. The negative-control harness initially expected an untruncated Vitest assertion line; the actual named 1-versus-2 failure and restoration were inspected before correcting its result matcher. - [Current-parent acceptance and failure ledger](https://github.com/paperclipai/paperclip/blob/60ba892ae9fc5faaf8cd0833694b4a66c13323c3/doc/plans/2026-10-28-cancellation-deadlock-acceptance.md) records current scope, commands, isolated fixtures, mutation evidence and limits. Earlier combined-stack counts retain their original revision scope. The existing controlled runner timeout/late-append fixture is retained and passes within the complete 180-case runtime suite. - This addresses the prior Ponytail request to exercise native claimed state before deadlock rollback and re-evaluate authority afterward. The advisory loop-invariant snapshot/named-attempt-bound note remains a maintainability suggestion, not a runtime defect. No production loop refactor or new worker duty is introduced. - **All 58 current checks are clean; Greptile and Ponytail both rate this exact head 5/5.** Ponytail explicitly confirms that the native-claim coverage request is resolved and says Ship. The formerly queued dependency/quality review workflow has completed successfully. Old `fffc`/parent checks and reviews are not current passes; the former pending Snyk status is not called successful. Actual failing Ponytail findings must be fixed. Missing/stopped/paused turns without a verdict are authorized nonblockers. Active unfinished reviews retain their original per-head timer, with a 25-minute monitor cutoff respecting the 30-minute maximum. - No full local repository matrix, live provider execution, external alarm, physical host/database suspension or savings claim is made. No schema, dependency, workflow, shared scheduler contract or shared OS resource changes are introduced. ## Risks - Persistent contention still returns an error after the third attempt. This change does not promise to eliminate every possible lock cycle. - Retrying a write after an ambiguous connection failure would be unsafe. The retry accepts only PostgreSQL's confirmed deadlock abort code; it does not accept transport errors. - No schema or public API contract changes. Current task/run authority and cancellation eligibility are checked again on each attempt. ## Model Used OpenAI GPT-6 via Codex, with shell, GitHub CLI, TypeScript, PostgreSQL, and Playwright tools. The exact serving model ID and context window are not exposed in this session. ## 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>
1729 lines
83 KiB
TypeScript
1729 lines
83 KiB
TypeScript
import { resumeDeferredChatWake } from "../durable-chat-wakeup.js";
|
|
import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js";
|
|
import { readNativeStopExecutionEvidence } from "./native-stop-evidence-read.js";
|
|
import { readTaskRetryIntent, reconcileTaskChatRetryReceipts } from "../task-retry.js";
|
|
import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, readNativeLocalProcessStop, recordedLocalExecutionStopped } from "../native-local-process-stop.js";
|
|
import { isHeartbeatRunTerminalStatus } from "./run-lifecycle.js";
|
|
import { pendingCaseHandoffForRun, cancelCaseHandoffsForIssue, isCaseHandoffSourceReleased } from "./case-handoff-lifecycle.js";
|
|
import { publishActivity, type ActivityPublication } from "../activity-log.js";
|
|
import { DEFERRED_WAKE_CONTEXT_KEY } from "./queue.js";
|
|
import { CANCELLABLE_HEARTBEAT_RUN_STATUSES } from "./recovery.js";
|
|
import {
|
|
type WakeupOptions,
|
|
} from "./run-preparation.js";
|
|
import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js";
|
|
import { hasRequiredWorkspaceRecovery, nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js";
|
|
import { preserveNativeFailedWorkspace } from "../native-runtime/native-failed-workspace-recovery.js";
|
|
import { readNativeWorkspaceSyncReference } from "../native-runtime/native-workspace-sync.js";
|
|
import { isAiAuthenticationBlocked } from "../ai-auth-failure.js";
|
|
import {
|
|
nativeRetryCancellationCommitCondition,
|
|
nativeRetryCancellationEligible,
|
|
claimCancellationRequest,
|
|
rethrowNativeCancellationLockConflict,
|
|
} from "../native-runtime/native-cancellation-request.js";
|
|
import {
|
|
canContinueCancelledRun,
|
|
readRunCancellation,
|
|
requestedRunCancellation,
|
|
} from "../run-cancellation.js";
|
|
import { settleSlackConversation } from "../slack-conversation-lifecycle.js";
|
|
import { readQueuedInteractionResponse } from "../queued-interaction-response.js";
|
|
import {
|
|
isAcknowledgedNativeStop,
|
|
acknowledgedNativeStopExecutionHasStopped,
|
|
} from "../acknowledged-native-stop.js";
|
|
import { completeTerminatedRemoteNativeSessionCleanup } from "../../vendor/paperclip-runner/index.js";
|
|
import {
|
|
hasRemoteTerminationReceipt,
|
|
remoteExecutionHasStopped,
|
|
stoppedRemoteCleanupScopes,
|
|
} from "../remote-execution-termination.js";
|
|
import {
|
|
admitExplicitNativeContinuation,
|
|
isUserCommentContinuationReason,
|
|
undeliveredLegacyUserCommentIds,
|
|
} from "../explicit-native-continuation.js";
|
|
import { isCancelledNativeStartup } from "../cancelled-native-startup.js";
|
|
import {
|
|
executionBlockerPredicate,
|
|
getExecutionBlocker,
|
|
} from "../execution-blocker.js";
|
|
import {
|
|
CONVERSATION_CONTINUATION_POLICY,
|
|
runUsedConversationAdapter,
|
|
hasConversationContinuationPolicy,
|
|
} from "../conversation-continuation.js";
|
|
import {
|
|
adapterExecutionControls,
|
|
captureAdapterStopOwnership,
|
|
reportAdapterStopIfUnsettled,
|
|
waitForAdapterStop,
|
|
} from "../adapter-execution-control.js";
|
|
import { AdapterStopTimeoutError } from "../adapter-stop-timeout.js";
|
|
import { budgetCancellationScopePredicate, pendingBudgetRunCancellationPredicate } from "./budget-cancellation.js";
|
|
import { withBudgetSpan } from "../budget-tracing.js";
|
|
import { captureException } from "../../sentry.js";
|
|
import { randomUUID } from "node:crypto";
|
|
import {
|
|
and,
|
|
asc,
|
|
desc,
|
|
eq,
|
|
exists,
|
|
gt,
|
|
inArray,
|
|
isNull,
|
|
isNotNull,
|
|
lt,
|
|
lte,
|
|
ne,
|
|
notInArray,
|
|
or,
|
|
sql,
|
|
} from "drizzle-orm";
|
|
import type { Db } from "@paperclipai/db";
|
|
import {
|
|
type EnvironmentLeaseStatus,
|
|
} from "@paperclipai/shared";
|
|
import {
|
|
agentWakeupRequests,
|
|
browserUseRuns,
|
|
browserUseSessions,
|
|
companies,
|
|
environmentLeases,
|
|
heartbeatRuns,
|
|
heartbeatRunEvents,
|
|
issueComments,
|
|
issueRecoveryActions,
|
|
issues,
|
|
nativeRunFinalizations,
|
|
} from "@paperclipai/db";
|
|
import {
|
|
conflict,
|
|
HttpError,
|
|
notFound,
|
|
} from "../../errors.js";
|
|
import { logger } from "../../middleware/logger.js";
|
|
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
|
|
import {
|
|
queuedCommentIdsFromWakePayload,
|
|
withQueuedCommentIdsInWakePayload,
|
|
withQueuedCommentIdsInRunContext,
|
|
} from "../issue-queued-comment-queue.js";
|
|
import {
|
|
cancelNativeSession,
|
|
closeWarmNativeSessionsForEnvironment,
|
|
} from "../native-runtime/index.js";
|
|
import { runningProcesses } from "../../adapters/index.js";
|
|
import { parseObject } from "../../adapters/utils.js";
|
|
import { isNativeRunnerOwnershipHeld } from "../native-runtime/native-runner-ownership.js";
|
|
import {
|
|
withCurrentBudgetEnforcement,
|
|
type BudgetEnforcementScope,
|
|
} from "../budgets.js";
|
|
import { terminateLocalService } from "../local-service-supervisor.js";
|
|
import { deriveCommentId } from "../../modules/run-dispatch/index.js";
|
|
import { WakeQueueApplicationError } from "../../modules/wake-queue/index.js";
|
|
import { isUuidLike } from "@paperclipai/shared";
|
|
import {
|
|
type ProviderResourceDisposition,
|
|
} from "../environment-runtime.js";
|
|
|
|
import type { createHeartbeatRunState } from "./run-state.js";
|
|
import type { createHeartbeatLifecycle } from "./run-lifecycle.js";
|
|
import type { createHeartbeatRetries, HeartbeatRetryDependencies } from "./retries.js";
|
|
import type { createHeartbeatQueue } from "./queue.js";
|
|
import type { createHeartbeatRecovery } from "./recovery.js";
|
|
import type { createWakeQueue, PostCommitEffect as WakeQueuePostCommitEffect } from "../../modules/wake-queue/index.js";
|
|
import type { environmentRunOrchestrator } from "../environment-run-orchestrator.js";
|
|
|
|
type ProcessRunCancellationSettlement = {
|
|
settled: Promise<void>;
|
|
failed: boolean;
|
|
error?: unknown;
|
|
};
|
|
|
|
// This bounds external Stop concurrency across scopes and heartbeat service
|
|
// instances sharing a database. Durable cancellation rows own recovery.
|
|
const budgetStopDeliveryLimits = new WeakMap<Db, { active: number; waiting: Array<() => void> }>();
|
|
|
|
async function withBudgetStopDelivery<T>(db: Db, work: () => Promise<T>): Promise<T> {
|
|
let limit = budgetStopDeliveryLimits.get(db);
|
|
if (!limit) {
|
|
limit = { active: 0, waiting: [] };
|
|
budgetStopDeliveryLimits.set(db, limit);
|
|
}
|
|
if (limit.active >= 4) await new Promise<void>(resolve => limit.waiting.push(resolve));
|
|
else limit.active += 1;
|
|
try {
|
|
return await withBudgetSpan("budget.cancellation.deliver", work);
|
|
} finally {
|
|
const next = limit.waiting.shift();
|
|
if (next) next();
|
|
else limit.active -= 1;
|
|
}
|
|
}
|
|
|
|
/** Cancellation shares executor ownership and delegates lifecycle and admission effects. */
|
|
export interface HeartbeatRunControlDependencies extends Pick<ReturnType<typeof createHeartbeatRunState>,
|
|
"getRun" | "getAgent" | "getRunCleanupState" | "getRunExecutionWait">,
|
|
Pick<ReturnType<typeof createHeartbeatLifecycle>,
|
|
"mergeRunStopMetadataForAgent" | "setRunStatusFromLive" | "setRunStatus" | "setWakeupStatus" | "appendRunEvent">,
|
|
Pick<HeartbeatRetryDependencies, "finalizeAgentStatus"> {
|
|
options: {
|
|
closeWarmNativeSessionsForRun?: (input: {
|
|
runId: string;
|
|
reason: string;
|
|
}) => Promise<{ closed: number; busy: number; failed: number }>;
|
|
};
|
|
envOrchestrator: Pick<ReturnType<typeof environmentRunOrchestrator>, "releaseForRun">;
|
|
enqueueWakeup: ReturnType<typeof createHeartbeatQueue>["enqueueWakeup"];
|
|
startNextQueuedRunForAgent: ReturnType<typeof createHeartbeatQueue>["startNextQueuedRunForAgent"];
|
|
sweepPendingCleanupLeases: ReturnType<typeof createHeartbeatRecovery>["sweepPendingCleanupLeases"];
|
|
timerClaimWasFirstHeartbeat: ReturnType<typeof createHeartbeatRetries>["timerClaimWasFirstHeartbeat"];
|
|
wakeQueue: Pick<ReturnType<typeof createWakeQueue>, "releaseIssueExecution">;
|
|
applyWakeQueuePostCommitEffects: (effects: WakeQueuePostCommitEffect[]) => Promise<void>;
|
|
getSchedulingSuppression: () => Promise<{ suppressed: boolean }>;
|
|
// Scheduler and route service instances must retain the same in-process barriers.
|
|
activeRunExecutions: Set<string>;
|
|
processRunCancellationSettlements: Map<string, ProcessRunCancellationSettlement>;
|
|
failedProcessRunCancellations: Map<string, ProcessRunCancellationSettlement>;
|
|
}
|
|
|
|
export function leaseReleaseStatusForRunStatus(
|
|
status: string | null | undefined,
|
|
): Extract<EnvironmentLeaseStatus, "released" | "expired" | "failed"> {
|
|
if (status === "cancelled") return "expired";
|
|
return status === "failed" || status === "timed_out" ? "failed" : "released";
|
|
}
|
|
|
|
export function providerResourceDispositionForTerminalRun(
|
|
desired: ProviderResourceDisposition | undefined,
|
|
status: string | null | undefined,
|
|
): ProviderResourceDisposition | undefined {
|
|
if (desired !== "keep_running") return desired;
|
|
return status === "succeeded" ? desired : "stop_and_retain";
|
|
}
|
|
|
|
export async function terminateHeartbeatRunProcess(input: {
|
|
pid: number | null | undefined;
|
|
processGroupId: number | null | undefined;
|
|
graceMs?: number;
|
|
signal?: NodeJS.Signals;
|
|
}) {
|
|
const pid = input.pid ?? null;
|
|
const processGroupId = input.processGroupId ?? null;
|
|
if (typeof pid !== "number" && typeof processGroupId !== "number") return;
|
|
|
|
await terminateLocalService(
|
|
{
|
|
pid:
|
|
typeof pid === "number" && Number.isInteger(pid) && pid > 0
|
|
? pid
|
|
: (processGroupId ?? 0),
|
|
processGroupId:
|
|
typeof processGroupId === "number" &&
|
|
Number.isInteger(processGroupId) &&
|
|
processGroupId > 0
|
|
? processGroupId
|
|
: null,
|
|
},
|
|
{ forceAfterMs: input.graceMs, signal: input.signal },
|
|
);
|
|
}
|
|
|
|
export async function cancelHeartbeatNativeRun(input: {
|
|
db: Db;
|
|
runId: string;
|
|
reason: string;
|
|
runtimeMode: string | null;
|
|
cancellationRequestId?: string;
|
|
cancel?: (
|
|
runId: string,
|
|
reason: string,
|
|
options: { db: Db; scope: "run"; cancellationRequestId?: string },
|
|
) => Promise<{ decision: unknown | null; auditId: string | null }>;
|
|
}) {
|
|
if (input.runtimeMode !== "native") {
|
|
return { decision: null, auditId: null };
|
|
}
|
|
const cancellation = input.cancel
|
|
? await input.cancel(input.runId, input.reason, {
|
|
db: input.db,
|
|
scope: "run",
|
|
...(input.cancellationRequestId ? { cancellationRequestId: input.cancellationRequestId } : {}),
|
|
})
|
|
: await cancelNativeSession(input.runId, input.reason, {
|
|
db: input.db,
|
|
scope: "run",
|
|
...(input.cancellationRequestId ? { cancellationRequestId: input.cancellationRequestId } : {}),
|
|
});
|
|
if (!cancellation.decision || !cancellation.auditId) {
|
|
throw new Error("native_cancellation_outcome_not_audited");
|
|
}
|
|
return cancellation;
|
|
}
|
|
|
|
function readNonEmptyString(value: unknown): string | null {
|
|
return typeof value === "string" && value.trim().length > 0 ? value : null;
|
|
}
|
|
|
|
export type CancelRunOptions = {
|
|
/** Return null if a parked retry was promoted or settled before this cancellation. */
|
|
expectedStatus?: "scheduled_retry";
|
|
budgetEnforcement?: BudgetEnforcementScope;
|
|
/** Optional board request identity, atomically reserved for native Stop. */
|
|
cancellationRequestId?: string;
|
|
cancellationRequestedByUserId?: string | null;
|
|
errorCode?: string;
|
|
resultJson?: Record<string, unknown>;
|
|
eventMessage?: string;
|
|
eventPayload?: Record<string, unknown>;
|
|
/** Per-call graceful process shutdown window, bounded to a safe range. */
|
|
terminationGraceMs?: number;
|
|
/** Caller is immediately scheduling an explicit successor path. */
|
|
suppressImmediateRecovery?: boolean;
|
|
};
|
|
|
|
/** The same durable fence protects both individual Stop and budget fan-out.
|
|
* It records intent only; provider termination still needs its own receipt. */
|
|
function runCancellationFenceSql(input: {
|
|
cancellation: ReturnType<typeof requestedRunCancellation>;
|
|
errorCode: string;
|
|
reassignmentStopRequested?: boolean;
|
|
preserveCancellation?: boolean;
|
|
budgetEnforcement?: BudgetEnforcementScope;
|
|
}) {
|
|
const cancellation = JSON.stringify(input.cancellation);
|
|
const startupCancellation = sql`jsonb_build_object(
|
|
'requestedAt', ${new Date().toISOString()}::text,
|
|
'errorCode', ${input.errorCode}::text,
|
|
'beforeNativeSelection', ${heartbeatRuns.runtimeMode} = 'legacy'
|
|
and ${heartbeatRuns.runtimeModeResolvedAt} is null
|
|
and ${heartbeatRuns.executionStage} = 'preparing'
|
|
and coalesce(${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->>'adapterType' = 'paperclip_runner', false)
|
|
)`;
|
|
let startupFence = input.preserveCancellation
|
|
? sql`coalesce(${heartbeatRuns.resultJson}->'startupCancellation', ${startupCancellation})`
|
|
: startupCancellation;
|
|
if (input.budgetEnforcement) {
|
|
const intent = JSON.stringify({ companyId: input.budgetEnforcement.companyId,
|
|
scopeType: input.budgetEnforcement.scopeType, scopeId: input.budgetEnforcement.scopeId,
|
|
policyId: input.budgetEnforcement.enforcement?.policyId, version: input.budgetEnforcement.enforcement?.version,
|
|
createdBefore: input.budgetEnforcement.createdBefore?.toISOString() });
|
|
startupFence = sql`${startupFence} || jsonb_build_object('budgetEnforcement',
|
|
coalesce(nullif(${heartbeatRuns.resultJson}->'startupCancellation'->'budgetEnforcement', 'null'::jsonb), ${intent}::jsonb))`;
|
|
}
|
|
return sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) ||
|
|
${JSON.stringify(input.reassignmentStopRequested ? { reassignmentStopRequested: true } : {})}::jsonb ||
|
|
jsonb_build_object('cancellation', ${input.preserveCancellation
|
|
? sql`coalesce(${heartbeatRuns.resultJson}->'cancellation', ${cancellation}::jsonb)`
|
|
: sql`${cancellation}::jsonb`}) ||
|
|
jsonb_build_object('startupCancellation', ${startupFence})`;
|
|
}
|
|
|
|
/** Run control binds cancellation, resource release, and saved-input resumption. */
|
|
export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunControlDependencies) {
|
|
const {
|
|
getRun,
|
|
getRunCleanupState,
|
|
getRunExecutionWait,
|
|
options,
|
|
envOrchestrator,
|
|
activeRunExecutions,
|
|
enqueueWakeup,
|
|
getAgent,
|
|
sweepPendingCleanupLeases,
|
|
getSchedulingSuppression,
|
|
wakeQueue,
|
|
applyWakeQueuePostCommitEffects,
|
|
processRunCancellationSettlements,
|
|
mergeRunStopMetadataForAgent,
|
|
setRunStatusFromLive,
|
|
failedProcessRunCancellations,
|
|
setWakeupStatus,
|
|
appendRunEvent,
|
|
finalizeAgentStatus,
|
|
timerClaimWasFirstHeartbeat,
|
|
startNextQueuedRunForAgent,
|
|
setRunStatus,
|
|
} = dependencies;
|
|
|
|
async function releaseEnvironmentLeasesForRun(input: {
|
|
runId: string;
|
|
companyId: string;
|
|
agentId: string;
|
|
status: string | null | undefined;
|
|
failureReason?: string | null;
|
|
providerResourceDisposition?: ProviderResourceDisposition;
|
|
nativeLifecycleTelemetry?: {
|
|
provider: string;
|
|
harness: string;
|
|
lifecycleMode: "per_turn" | "warm";
|
|
sandboxResource: "keep_running" | "stop_and_reuse" | "destroy_after_turn";
|
|
};
|
|
}) {
|
|
let leaseOwnerRun = await getRunCleanupState(input.runId);
|
|
if (leaseOwnerRun && isNativeRunnerOwnershipHeld(leaseOwnerRun)) return;
|
|
const cancelledRun = leaseOwnerRun?.runtimeMode === "native" && leaseOwnerRun.status === "cancelled"
|
|
? await getRun(input.runId) : null;
|
|
if (cancelledRun) leaseOwnerRun = cancelledRun;
|
|
if (leaseOwnerRun && isNativeRunnerOwnershipHeld(leaseOwnerRun)) return;
|
|
// Recovery can finish workspace copy-back outside the executor's finally.
|
|
// Successful copy-back does not earn warm retention for a failed turn.
|
|
const status = leaseOwnerRun?.status ?? input.status;
|
|
let providerResourceDisposition = providerResourceDispositionForTerminalRun(
|
|
input.providerResourceDisposition,
|
|
status,
|
|
);
|
|
if (cancelledRun?.runtimeMode === "native" && status === "cancelled"
|
|
&& readNativeWorkspaceSyncReference(cancelledRun.runnerProfileJson?.nativeWorkspaceSync)?.state === "prepared"
|
|
&& !nativeRunHasNoWorkspaceAccess(cancelledRun)) {
|
|
try {
|
|
const preserved = await preserveNativeFailedWorkspace(db, cancelledRun, { allowPreterminal: true });
|
|
if (!preserved || !hasRequiredWorkspaceRecovery(preserved.resultJson)) return;
|
|
} catch {
|
|
// Keep the active source eligible for the bounded retention sweep.
|
|
// Teardown must not release it after retention lost a lock or DB write.
|
|
logger.warn({ runId: input.runId }, "Deferred cancelled workspace teardown until file retention is recorded");
|
|
return;
|
|
}
|
|
providerResourceDisposition = "stop_and_retain";
|
|
}
|
|
if (providerResourceDisposition === "destroy") {
|
|
const closeResult = await (
|
|
options.closeWarmNativeSessionsForRun ??
|
|
(async ({ runId, reason }) => {
|
|
const environmentIds = await db
|
|
.selectDistinct({ environmentId: environmentLeases.environmentId })
|
|
.from(environmentLeases)
|
|
.where(
|
|
and(
|
|
eq(environmentLeases.heartbeatRunId, runId),
|
|
eq(environmentLeases.status, "active"),
|
|
),
|
|
)
|
|
.then((rows) =>
|
|
rows.flatMap((row) =>
|
|
typeof row.environmentId === "string" &&
|
|
row.environmentId.length > 0
|
|
? [row.environmentId]
|
|
: [],
|
|
),
|
|
);
|
|
const aggregate = { closed: 0, busy: 0, failed: 0 };
|
|
for (const environmentId of environmentIds) {
|
|
const result = await closeWarmNativeSessionsForEnvironment({
|
|
environmentId,
|
|
reason,
|
|
});
|
|
aggregate.closed += result.closed;
|
|
aggregate.busy += result.busy;
|
|
aggregate.failed += result.failed;
|
|
}
|
|
return aggregate;
|
|
})
|
|
)({
|
|
runId: input.runId,
|
|
reason: "terminal heartbeat run destroyed its environment lease",
|
|
}).catch((err) => {
|
|
logger.warn(
|
|
{ err, runId: input.runId },
|
|
"failed to close warm native sessions before environment lease destruction",
|
|
);
|
|
return { closed: 0, busy: 0, failed: 1 };
|
|
});
|
|
// A failed remote checkpoint cannot veto destruction of the isolated
|
|
// sandbox after the run stopped. Provider destruction supplies the exit
|
|
// proof; it does not turn the interrupted checkpoint into a success.
|
|
const remoteLeases = closeResult.failed > 0 && leaseOwnerRun &&
|
|
["cancelled", "failed", "timed_out", "interrupted"].includes(leaseOwnerRun.status)
|
|
? await db.select({ provider: environmentLeases.provider }).from(environmentLeases).where(and(
|
|
eq(environmentLeases.companyId, input.companyId),
|
|
eq(environmentLeases.heartbeatRunId, input.runId),
|
|
))
|
|
: [];
|
|
const canDestroyRemote = remoteLeases.length > 0 &&
|
|
remoteLeases.every(lease => lease.provider && lease.provider !== "local");
|
|
if (closeResult.busy > 0 || (closeResult.failed > 0 && !canDestroyRemote)) {
|
|
logger.warn(
|
|
{ runId: input.runId, warmNativeSessions: closeResult },
|
|
"deferred environment lease destruction until warm native sessions close",
|
|
);
|
|
return;
|
|
}
|
|
}
|
|
const releaseResult = await envOrchestrator
|
|
.releaseForRun({
|
|
heartbeatRunId: input.runId,
|
|
companyId: input.companyId,
|
|
agentId: input.agentId,
|
|
status: leaseReleaseStatusForRunStatus(status),
|
|
failureReason: input.failureReason ?? undefined,
|
|
providerResourceDisposition,
|
|
nativeLifecycleTelemetry: input.nativeLifecycleTelemetry,
|
|
})
|
|
.catch((err) => {
|
|
logger.warn(
|
|
{ err, runId: input.runId },
|
|
"failed to release environment leases for heartbeat run",
|
|
);
|
|
return null;
|
|
});
|
|
for (const releaseError of releaseResult?.errors ?? []) {
|
|
logger.warn(
|
|
{
|
|
err: releaseError.error,
|
|
leaseId: releaseError.leaseId,
|
|
runId: input.runId,
|
|
},
|
|
"failed to release environment lease for heartbeat run",
|
|
);
|
|
}
|
|
await acknowledgeRemoteStop(input.runId, input.companyId);
|
|
}
|
|
|
|
async function acknowledgeRemoteStop(runId: string, companyId: string) {
|
|
// The provider receipt arrives after adapter settlement. A remote ACP child
|
|
// has no host PID, so only this target-aware boundary can acknowledge Stop.
|
|
const state = await getRunCleanupState(runId);
|
|
if (state?.runtimeMode === "native") {
|
|
const scopes = await stoppedRemoteCleanupScopes(db, companyId, runId);
|
|
for (const remoteCleanupScope of scopes ?? []) {
|
|
completeTerminatedRemoteNativeSessionCleanup({ companyId, runId, remoteCleanupScope });
|
|
}
|
|
}
|
|
const stopped = state?.status === "cancelled" ? await getRun(runId) : null;
|
|
// Native interrupt acknowledgement is not physical containment. Publish the
|
|
// shared Stop receipt only after every exact remote lease confirms teardown.
|
|
// Keep workspace preservation separate: a stopped provider can still owe files.
|
|
if (stopped?.status === "cancelled" &&
|
|
parseObject(stopped.resultJson?.executionCancellation).state === "requested" &&
|
|
(stopped.runtimeMode === "native" ? isAcknowledgedNativeStop(stopped)
|
|
: stopped.runtimeMode === "legacy" && await runUsedConversationAdapter(db, stopped)) &&
|
|
await remoteExecutionHasStopped(db, companyId, runId)) {
|
|
await db.update(heartbeatRuns).set({
|
|
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
|
|
executionCancellation: { ...parseObject(stopped.resultJson?.executionCancellation),
|
|
state: "acknowledged", acknowledgedAt: new Date().toISOString(),
|
|
proof: "provider_termination_receipt" },
|
|
...(stopped.runtimeMode === "legacy" ? { conversationContinuation: CONVERSATION_CONTINUATION_POLICY } : {}),
|
|
})}::jsonb`,
|
|
updatedAt: new Date(),
|
|
}).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId),
|
|
eq(heartbeatRuns.status, "cancelled"),
|
|
sql`${heartbeatRuns.resultJson}->'executionCancellation'->>'state' = 'requested'`,
|
|
stopped.runtimeMode === "native" ? sql`${heartbeatRuns.resultJson}->'nativeCancellation' = ${JSON.stringify(stopped.resultJson?.nativeCancellation)}::jsonb` : undefined));
|
|
}
|
|
}
|
|
|
|
async function resumeRemoteStopComments(run: NonNullable<Awaited<ReturnType<typeof getRunExecutionWait>>>, requestId?: string) {
|
|
if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id)) return;
|
|
const [preparationCoordinator] = run.runtimeMode === "legacy" && run.status === "cancelled" && !run.runtimeModeResolvedAt
|
|
? await db.select().from(nativeRunFinalizations).where(and(
|
|
eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id),
|
|
)) : [];
|
|
const cancelledPreparation = run.runtimeMode === "legacy" &&
|
|
await isCancelledNativeStartup(db, run, preparationCoordinator);
|
|
if (run.runtimeMode !== "native" && !cancelledPreparation &&
|
|
parseObject(run.resultJson?.startupCancellation).beforeNativeSelection !== true &&
|
|
!(await remoteExecutionHasStopped(db, run.companyId, run.id)) &&
|
|
!(await recordedLocalExecutionStopped(db, run))) return;
|
|
const issueId = run.nativeIssueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null);
|
|
if (!issueId) return;
|
|
const currentRun = run.runtimeMode === "native" ? await readNativeStopExecutionEvidence(db, run.companyId, run.id) : null;
|
|
const [coordinator] = currentRun ? await db.select({ phase: nativeRunFinalizations.phase,
|
|
leaseOwner: nativeRunFinalizations.leaseOwner }).from(nativeRunFinalizations).where(and(
|
|
eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id),
|
|
)) : [];
|
|
// The ordinary path still enforces task holds, ownership, and native session
|
|
// cleanup before starting a provider.
|
|
const stoppedNativeContinuation = currentRun && isAcknowledgedNativeStop(currentRun) &&
|
|
!activeRunExecutions.has(run.id) && !coordinator?.leaseOwner &&
|
|
["terminal_failure", "applied"].includes(coordinator?.phase ?? "") &&
|
|
await acknowledgedNativeStopExecutionHasStopped(db, currentRun) &&
|
|
!(await getExecutionBlocker(db, run.companyId, issueId));
|
|
const legacyContinuation = run.runtimeMode === "legacy" &&
|
|
hasConversationContinuationPolicy((await getRunExecutionWait(run.companyId, run.id))?.resultJson) &&
|
|
!(await getExecutionBlocker(db, run.companyId, issueId));
|
|
if (run.runtimeMode !== "native" && run.runtimeMode !== "legacy") return;
|
|
const pending = await db.select().from(agentWakeupRequests).where(and(
|
|
eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
|
|
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
|
inArray(agentWakeupRequests.requestedByActorType, ["user", "system"]),
|
|
requestId ? eq(agentWakeupRequests.id, requestId) : undefined,
|
|
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
|
)).orderBy(asc(agentWakeupRequests.requestedAt)).limit(50);
|
|
for (const wake of pending) {
|
|
if (wake.idempotencyKey?.startsWith("chat-inbound:")) {
|
|
await resumeDeferredChatWake(db, { companyId: run.companyId, agentId: run.agentId,
|
|
issueId, requestId: wake.id }).catch(() => {
|
|
logger.warn({ requestId: wake.id }, "saved chat source could not resume after cleanup");
|
|
});
|
|
continue;
|
|
}
|
|
if (wake.requestedByActorType !== "user") continue;
|
|
if (readTaskRetryIntent(wake.payload)) { await resumeTaskRetry(wake.companyId, wake.id); continue; }
|
|
let payload = parseObject(wake.payload);
|
|
if (payload.queuedCommentInterrupt) {
|
|
await resumeQueuedCommentInterrupt(wake.companyId, wake.id);
|
|
continue;
|
|
}
|
|
let context = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]);
|
|
let commentId = deriveCommentId(context, payload);
|
|
let requestedByActorId = wake.requestedByActorId;
|
|
const reason = readNonEmptyString(context.wakeReason) ?? wake.reason;
|
|
if (run.runtimeMode === "native" || cancelledPreparation) {
|
|
const ids = await undeliveredLegacyUserCommentIds(db, run.companyId, issueId, run.agentId,
|
|
queuedCommentIdsFromWakePayload(payload));
|
|
if (!ids.length) continue;
|
|
payload = withQueuedCommentIdsInWakePayload(payload, ids);
|
|
context = withQueuedCommentIdsInRunContext(context, ids);
|
|
commentId = ids.at(-1)!;
|
|
// A coalesced queue can contain several authors. Its saved comments,
|
|
// not the outer wake's first author, authorize the remaining input.
|
|
const [author] = await db.select({ id: issueComments.authorUserId }).from(issueComments).where(and(
|
|
eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), eq(issueComments.id, commentId),
|
|
));
|
|
requestedByActorId = author?.id ?? null;
|
|
}
|
|
if (legacyContinuation || stoppedNativeContinuation) {
|
|
if (!commentId || !run.finishedAt || !requestedByActorId ||
|
|
!isUserCommentContinuationReason(reason)) continue;
|
|
const [comment] = await db.select().from(issueComments).where(and(
|
|
eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId),
|
|
sql`${issueComments.id}::text = ${commentId}`, eq(issueComments.authorType, "user"),
|
|
eq(issueComments.authorUserId, requestedByActorId), isNull(issueComments.deletedAt),
|
|
isNull(issueComments.createdByRunId),
|
|
gt(issueComments.createdAt, run.finishedAt),
|
|
));
|
|
if (!comment?.body.trim()) continue;
|
|
} else {
|
|
let wait = { reason: "execution_recovery", message: "Waiting for execution recovery. Your message is saved." };
|
|
const admitted = await admitExplicitNativeContinuation({ db, companyId: run.companyId, issueId,
|
|
agentId: run.agentId, actorType: wake.requestedByActorType, actorId: requestedByActorId,
|
|
reason, commentId, successorRunId: randomUUID(), dryRun: true,
|
|
queuedCommentRequestId: wake.id,
|
|
onBlocked: (reason, message) => { wait = { reason, message }; },
|
|
});
|
|
if (!admitted) {
|
|
await db.update(agentWakeupRequests).set({
|
|
payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}',
|
|
coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb) || ${JSON.stringify(wait)}::jsonb)`,
|
|
updatedAt: new Date(),
|
|
}).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, run.companyId),
|
|
eq(agentWakeupRequests.status, "deferred_issue_execution")));
|
|
continue;
|
|
}
|
|
}
|
|
// Re-enter ordinary admission with the original user's authority. It
|
|
// atomically adopts the deferred comments and still applies every gate.
|
|
await enqueueWakeup(run.agentId, { source: wake.source as WakeupOptions["source"], triggerDetail: (wake.triggerDetail ?? undefined) as WakeupOptions["triggerDetail"],
|
|
reason, payload, contextSnapshot: context,
|
|
requestedByActorType: "user", requestedByActorId,
|
|
queuedCommentRequestId: wake.id,
|
|
idempotencyKey: `remote-stop-comment:${run.id}:${wake.id}` }, wake.id);
|
|
break;
|
|
}
|
|
}
|
|
|
|
async function resumeQueuedCommentInterrupt(companyId: string, queueId: string, opts?: { retryCleanup?: boolean }) {
|
|
return resumeSavedComments(companyId, queueId, true, opts);
|
|
}
|
|
|
|
async function resumeSavedComments(companyId: string, queueId: string, interrupted = false, opts?: { retryCleanup?: boolean }) {
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(and(
|
|
eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, companyId),
|
|
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
|
));
|
|
if (!wake) return;
|
|
const payload = parseObject(wake.payload);
|
|
let actorId = readNonEmptyString(parseObject(payload.queuedCommentInterrupt).actorId);
|
|
let commentIds = queuedCommentIdsFromWakePayload(payload);
|
|
const issueId = readNonEmptyString(payload.issueId);
|
|
if (!issueId || wake.idempotencyKey?.startsWith("chat-inbound:")) return;
|
|
const response = await readQueuedInteractionResponse(db, companyId, issueId, payload);
|
|
if (!commentIds.length && !response) return;
|
|
// Resolved cards are immutable input. Only an explicit Interrupt click can
|
|
// authorize continuation across a stopped execution; ordinary completion
|
|
// uses normal deferred-wake promotion.
|
|
if (response && !interrupted) return;
|
|
if (!interrupted) {
|
|
commentIds = await undeliveredLegacyUserCommentIds(db, companyId, issueId, wake.agentId, commentIds);
|
|
if (!commentIds.length) return;
|
|
// The queue itself may have begun as a system wake. The saved human
|
|
// comment, not that wake's origin or mutable caller payload, is authority.
|
|
const [comment] = await db.select().from(issueComments).where(and(
|
|
eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
|
|
eq(issueComments.id, commentIds[commentIds.length - 1]!), eq(issueComments.authorType, "user"),
|
|
isNull(issueComments.createdByRunId), isNull(issueComments.deletedAt),
|
|
)).orderBy(desc(issueComments.createdAt)).limit(1);
|
|
if (!comment?.body.trim() || !comment.authorUserId) return;
|
|
actorId = comment.authorUserId;
|
|
}
|
|
if (!actorId) return;
|
|
const agent = await getAgent(wake.agentId);
|
|
if (!agent || agent.companyId !== companyId || (agent.adapterType === "paperclip_runner" && !interrupted && !response?.source.requiresFreshSession)) return;
|
|
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.companyId, companyId),
|
|
eq(heartbeatRuns.agentId, wake.agentId),
|
|
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
|
|
inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
|
|
)).limit(1);
|
|
if (active) return;
|
|
if (interrupted && opts?.retryCleanup) {
|
|
// Only the HTTP click grants an extra cleanup attempt. Periodic retries
|
|
// reuse the intent to deliver, never a fresh provider teardown budget.
|
|
const sourceRun = await db.transaction(async tx => {
|
|
const [task] = await tx.select().from(issues).where(and(
|
|
eq(issues.companyId, companyId), eq(issues.id, issueId),
|
|
)).for("update");
|
|
const [current] = await tx.select().from(agentWakeupRequests).where(and(
|
|
eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, companyId),
|
|
eq(agentWakeupRequests.agentId, wake.agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
|
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
|
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${actorId}`,
|
|
));
|
|
if (!task || task.assigneeAgentId !== wake.agentId || ["done", "cancelled"].includes(task.status) ||
|
|
!current || (!queuedCommentIdsFromWakePayload(current.payload).length &&
|
|
!await readQueuedInteractionResponse(tx as unknown as Db, companyId, issueId, current.payload))) return null;
|
|
const [successor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.companyId, companyId),
|
|
eq(heartbeatRuns.agentId, wake.agentId),
|
|
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
|
|
inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
|
|
)).limit(1);
|
|
if (successor) return null;
|
|
const blocker = await getExecutionBlocker(tx as unknown as Db, companyId, issueId);
|
|
const run = blocker?.runId ? await tx.select().from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, blocker.runId),
|
|
eq(heartbeatRuns.agentId, wake.agentId), eq(heartbeatRuns.runtimeMode, "legacy"),
|
|
inArray(heartbeatRuns.status, ["failed", "timed_out", "interrupted", "cancelled"]),
|
|
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
|
|
)).then(rows => rows[0]) : null;
|
|
if (!run || activeRunExecutions.has(run.id) || adapterExecutionControls.has(run.id)) return null;
|
|
// Older ephemeral leases recorded successful cleanup without a provider
|
|
// receipt. Re-verify them through the recorded teardown path; a timestamp
|
|
// alone never certifies termination. Retained/reusable resources stay put.
|
|
const historical = await tx.select().from(environmentLeases).where(and(
|
|
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
|
eq(environmentLeases.leasePolicy, "ephemeral"), isNotNull(environmentLeases.releasedAt),
|
|
inArray(environmentLeases.status, ["released", "expired", "failed"]),
|
|
)).for("update");
|
|
for (const lease of historical) {
|
|
if (!lease.provider || lease.provider === "local" || !lease.providerLeaseId || hasRemoteTerminationReceipt(lease)) continue;
|
|
// Provider resource IDs identify physical sandboxes. A lease in any
|
|
// company can still own this resource; never destroy it on behalf of
|
|
// this company. This existence-only guard exposes no foreign data.
|
|
const [otherOwner] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
|
ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider),
|
|
eq(environmentLeases.providerLeaseId, lease.providerLeaseId),
|
|
or(isNull(environmentLeases.releasedAt), inArray(environmentLeases.status, ["active", "retained", "pending_cleanup"])),
|
|
)).limit(1);
|
|
if (otherOwner) continue;
|
|
await notifyEnvironmentCleanup(tx);
|
|
await tx.update(environmentLeases).set({ status: "pending_cleanup", updatedAt: new Date() })
|
|
.where(eq(environmentLeases.id, lease.id));
|
|
}
|
|
return run;
|
|
});
|
|
if (sourceRun) await sweepPendingCleanupLeases({ explicitRetry: {
|
|
companyId, runId: sourceRun.id, actorId, reason: "queued_comment_interrupt",
|
|
} });
|
|
}
|
|
const deliveryPayload = response ? { ...payload } : withQueuedCommentIdsInWakePayload(payload, commentIds);
|
|
delete deliveryPayload.queuedCommentInterrupt;
|
|
await enqueueWakeup(wake.agentId, {
|
|
source: "on_demand", triggerDetail: "manual", reason: "issue_commented",
|
|
payload: deliveryPayload, contextSnapshot: response
|
|
? { ...parseObject(payload._paperclipWakeContext), issueId, triggeredBy: "board", actorId,
|
|
responsibleUserId: actorId }
|
|
: withQueuedCommentIdsInRunContext({
|
|
issueId, triggeredBy: "board", actorId, responsibleUserId: actorId,
|
|
}, commentIds),
|
|
requestedByActorType: "user", requestedByActorId: actorId,
|
|
...(interrupted ? { queuedCommentInterruptId: queueId } : { queuedCommentRequestId: queueId }),
|
|
issueStateGuard: { assigneeAgentId: wake.agentId, statuses: ["todo", "in_progress", "in_review", "blocked"] },
|
|
idempotencyKey: `queued-comment-${interrupted ? "interrupt" : "delivery"}:${queueId}`,
|
|
}, queueId);
|
|
}
|
|
|
|
async function resumeTaskRetry(companyId: string, requestId: string) {
|
|
if ((await getSchedulingSuppression()).suppressed) return;
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(and(
|
|
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, requestId),
|
|
eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.reason, "task_retry"),
|
|
eq(agentWakeupRequests.requestedByActorType, "user"),
|
|
));
|
|
const intent = readTaskRetryIntent(wake?.payload);
|
|
if (!wake || !intent || intent.chatActionId || intent.canonicalIntentId || intent.generationId !== wake.id ||
|
|
intent.actorId !== wake.requestedByActorId) return;
|
|
const issueId = readNonEmptyString(wake.payload?.issueId);
|
|
if (!issueId) return;
|
|
await enqueueWakeup(wake.agentId, { source: "on_demand", triggerDetail: "manual", reason: "task_retry",
|
|
requestedByActorType: "user", requestedByActorId: intent.actorId, taskRetryRequestId: wake.id,
|
|
payload: { issueId }, contextSnapshot: { issueId, triggeredBy: "board", actorId: intent.actorId,
|
|
responsibleUserId: intent.actorId },
|
|
issueStateGuard: { assigneeAgentId: wake.agentId, statuses: ["todo", "in_progress", "in_review", "blocked"] },
|
|
idempotencyKey: `task-retry-dispatch:${wake.id}` }, wake.id);
|
|
}
|
|
|
|
async function resumeExecutionWaitComments() {
|
|
if ((await getSchedulingSuppression()).suppressed) return;
|
|
await reconcileTaskChatRetryReceipts(db);
|
|
const retries = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId })
|
|
.from(agentWakeupRequests).innerJoin(issues, and(eq(issues.companyId, agentWakeupRequests.companyId),
|
|
sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`,
|
|
eq(issues.assigneeAgentId, agentWakeupRequests.agentId)))
|
|
.innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active")))
|
|
.where(and(eq(agentWakeupRequests.reason, "task_retry"), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
|
sql`${agentWakeupRequests.payload}->'taskRetry'->>'chatActionId' is null`,
|
|
notInArray(issues.status, ["done", "cancelled"])))
|
|
.orderBy(asc(agentWakeupRequests.updatedAt)).limit(50);
|
|
for (const retry of retries) {
|
|
await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(eq(agentWakeupRequests.id, retry.id));
|
|
await resumeTaskRetry(retry.companyId, retry.id).catch(err => logger.warn({ err, requestId: retry.id }, "Task retry dispatch deferred"));
|
|
}
|
|
const waits = await db.select({ wake: agentWakeupRequests })
|
|
.from(agentWakeupRequests)
|
|
.innerJoin(issues, and(eq(issues.companyId, agentWakeupRequests.companyId),
|
|
sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`,
|
|
eq(issues.assigneeAgentId, agentWakeupRequests.agentId)))
|
|
.innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active")))
|
|
.where(and(exists(db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and(
|
|
eq(issueRecoveryActions.companyId, issues.companyId), eq(issueRecoveryActions.sourceIssueId, issues.id),
|
|
executionBlockerPredicate(),
|
|
))), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
|
eq(agentWakeupRequests.requestedByActorType, "user"),
|
|
sql`${agentWakeupRequests.payload}->'executionWait' is not null`,
|
|
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is null`,
|
|
lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)),
|
|
notInArray(issues.status, ["done", "cancelled"])))
|
|
.orderBy(asc(agentWakeupRequests.updatedAt)).limit(50);
|
|
const seen = new Set<string>();
|
|
for (const { wake } of waits) {
|
|
const issueId = String(wake.payload?.issueId);
|
|
if (seen.has(issueId)) continue;
|
|
seen.add(issueId);
|
|
// Advance the cursor even for invalid evidence so one damaged task cannot
|
|
// starve later requests in this bounded scan. Preserve concurrent edits.
|
|
const [claimed] = await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(and(
|
|
eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, wake.companyId),
|
|
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
|
lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)),
|
|
)).returning({ id: agentWakeupRequests.id });
|
|
if (!claimed) continue;
|
|
// Match normal admission's deterministic current blocker selection. An
|
|
// arbitrary historical action must not choose the retry's source run.
|
|
const blocker = await getExecutionBlocker(db, wake.companyId, issueId);
|
|
const sourceId = blocker?.runId;
|
|
if (!sourceId || !isUuidLike(sourceId)) continue;
|
|
const run = await getRunExecutionWait(wake.companyId, sourceId);
|
|
if (!run || run.companyId !== wake.companyId || run.agentId !== wake.agentId) continue;
|
|
if (canContinueCancelledRun(run)) {
|
|
await resumeSavedComments(wake.companyId, wake.id).catch(err => {
|
|
logger.warn({ err, runId: run.id }, "failed to resume saved input after provider cancellation");
|
|
});
|
|
continue;
|
|
}
|
|
await resumeRemoteStopComments(run, wake.id).catch(err => {
|
|
logger.warn({ err, runId: run.id }, "failed to resume saved execution-wait message");
|
|
});
|
|
}
|
|
}
|
|
|
|
async function releaseIssueExecutionAndPromote(
|
|
run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId">,
|
|
options: { suppressImmediateRecovery?: boolean; resumeStrandedQueue?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] } = {},
|
|
) {
|
|
try {
|
|
const [source] = await db.select({
|
|
errorCode: heartbeatRuns.errorCode,
|
|
livenessState: heartbeatRuns.livenessState,
|
|
resultJson: sql<Record<string, unknown>>`jsonb_build_object(
|
|
'configurationIncomplete', ${heartbeatRuns.resultJson}->'configurationIncomplete'
|
|
)`,
|
|
}).from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
|
|
));
|
|
const sourceHandoff = await pendingCaseHandoffForRun(db, run.id, run.companyId);
|
|
const { postCommitEffects } = await wakeQueue.releaseIssueExecution({
|
|
companyId: run.companyId,
|
|
runId: run.id,
|
|
now: new Date(),
|
|
// A durable authentication card owns recovery. This covers the review path,
|
|
// while continuation classification blocks periodic generic retries.
|
|
suppressImmediateRecovery: options.suppressImmediateRecovery || isAiAuthenticationBlocked(source) || Boolean(sourceHandoff),
|
|
...(options.resumeStrandedQueue ? { resumeStrandedQueue: true } : {}),
|
|
});
|
|
if (options.deferredPostCommitEffects) {
|
|
options.deferredPostCommitEffects.push(...postCommitEffects);
|
|
} else {
|
|
await applyWakeQueuePostCommitEffects(postCommitEffects);
|
|
}
|
|
const [completed] = await db.select({
|
|
id: heartbeatRuns.id,
|
|
status: heartbeatRuns.status,
|
|
startedAt: heartbeatRuns.startedAt,
|
|
nativeIssueId: heartbeatRuns.nativeIssueId,
|
|
contextIssueId: sql<unknown>`${heartbeatRuns.contextSnapshot}->'issueId'`,
|
|
contextTaskId: sql<unknown>`${heartbeatRuns.contextSnapshot}->'taskId'`,
|
|
}).from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
|
|
));
|
|
const issueId = readNonEmptyString(completed?.contextIssueId)
|
|
?? readNonEmptyString(completed?.contextTaskId) ?? completed?.nativeIssueId;
|
|
if (sourceHandoff && completed?.status === "cancelled" && !completed.startedAt && issueId) {
|
|
// A cancelled queue entry has no executeRun.finally. The canonical
|
|
// bootstrap proof still needs to release and wake its durable handoff.
|
|
await isCaseHandoffSourceReleased(db, { runId: completed.id, issueId });
|
|
}
|
|
if (completed?.status === "succeeded" && issueId) {
|
|
await settleSlackConversation(db, run.companyId, issueId).catch((err) => {
|
|
logger.warn({ err, runId: run.id }, "Slack conversation settlement deferred to reconciliation");
|
|
});
|
|
}
|
|
} catch (error) {
|
|
if (
|
|
error instanceof WakeQueueApplicationError &&
|
|
error.code === "responsible_user_unresolved"
|
|
) {
|
|
// Every other `responsible_user_unresolved` HttpError in this file
|
|
// carries `code` inside its own `details`; match that shape here too.
|
|
throw new HttpError(422, error.message, { code: error.code, ...error.details });
|
|
}
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async function listProjectScopedRunIds(companyId: string, projectId: string, createdBefore?: Date, database: Db = db) {
|
|
const runIssueId = sql<
|
|
string | null
|
|
>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`;
|
|
const effectiveProjectId = sql<
|
|
string | null
|
|
>`coalesce(${heartbeatRuns.contextSnapshot} ->> 'projectId', ${issues.projectId}::text)`;
|
|
|
|
const rows = await database
|
|
.selectDistinctOn([heartbeatRuns.id], { id: heartbeatRuns.id })
|
|
.from(heartbeatRuns)
|
|
.leftJoin(
|
|
issues,
|
|
and(
|
|
eq(issues.companyId, companyId),
|
|
sql`${issues.id}::text = ${runIssueId}`,
|
|
),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.companyId, companyId),
|
|
createdBefore ? lt(heartbeatRuns.createdAt, createdBefore) : undefined,
|
|
inArray(heartbeatRuns.status, [
|
|
...CANCELLABLE_HEARTBEAT_RUN_STATUSES,
|
|
]),
|
|
sql`${effectiveProjectId} = ${projectId}`,
|
|
),
|
|
);
|
|
|
|
return rows.map((row) => row.id);
|
|
}
|
|
|
|
async function listProjectScopedWakeupIds(
|
|
companyId: string,
|
|
projectId: string,
|
|
database: Db = db,
|
|
) {
|
|
const wakeIssueId = sql<
|
|
string | null
|
|
>`${agentWakeupRequests.payload} ->> 'issueId'`;
|
|
const effectiveProjectId = sql<
|
|
string | null
|
|
>`coalesce(${agentWakeupRequests.payload} ->> 'projectId', ${issues.projectId}::text)`;
|
|
|
|
const rows = await database
|
|
.selectDistinctOn([agentWakeupRequests.id], {
|
|
id: agentWakeupRequests.id,
|
|
})
|
|
.from(agentWakeupRequests)
|
|
.leftJoin(
|
|
issues,
|
|
and(
|
|
eq(issues.companyId, companyId),
|
|
sql`${issues.id}::text = ${wakeIssueId}`,
|
|
),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, companyId),
|
|
inArray(agentWakeupRequests.status, [
|
|
"queued",
|
|
"deferred_issue_execution",
|
|
]),
|
|
sql`${agentWakeupRequests.runId} is null`,
|
|
sql`${effectiveProjectId} = ${projectId}`,
|
|
),
|
|
);
|
|
|
|
return rows.map((row) => row.id);
|
|
}
|
|
|
|
async function cancelPendingWakeupsForBudgetScope(
|
|
scope: BudgetEnforcementScope,
|
|
tx: Db,
|
|
) {
|
|
const now = new Date();
|
|
let wakeupIds: string[] = [];
|
|
|
|
if (scope.scopeType === "company") {
|
|
wakeupIds = await tx
|
|
.select({ id: agentWakeupRequests.id })
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, scope.companyId),
|
|
inArray(agentWakeupRequests.status, [
|
|
"queued",
|
|
"deferred_issue_execution",
|
|
]),
|
|
sql`${agentWakeupRequests.runId} is null`,
|
|
),
|
|
)
|
|
.then((rows) => rows.map((row) => row.id));
|
|
} else if (scope.scopeType === "agent") {
|
|
wakeupIds = await tx
|
|
.select({ id: agentWakeupRequests.id })
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
eq(agentWakeupRequests.companyId, scope.companyId),
|
|
eq(agentWakeupRequests.agentId, scope.scopeId),
|
|
inArray(agentWakeupRequests.status, [
|
|
"queued",
|
|
"deferred_issue_execution",
|
|
]),
|
|
sql`${agentWakeupRequests.runId} is null`,
|
|
),
|
|
)
|
|
.then((rows) => rows.map((row) => row.id));
|
|
} else {
|
|
wakeupIds = await listProjectScopedWakeupIds(
|
|
scope.companyId,
|
|
scope.scopeId,
|
|
tx,
|
|
);
|
|
}
|
|
|
|
if (wakeupIds.length === 0) return 0;
|
|
|
|
await tx
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
status: "cancelled",
|
|
finishedAt: now,
|
|
error: "Cancelled due to budget pause",
|
|
updatedAt: now,
|
|
})
|
|
.where(and(
|
|
inArray(agentWakeupRequests.id, wakeupIds),
|
|
inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]),
|
|
isNull(agentWakeupRequests.runId),
|
|
scope.createdBefore ? lt(agentWakeupRequests.createdAt, scope.createdBefore) : undefined,
|
|
));
|
|
|
|
return wakeupIds.length;
|
|
}
|
|
|
|
function cancellationTerminationGraceMs(
|
|
configuredGraceSec: number,
|
|
requestedGraceMs: number | undefined,
|
|
) {
|
|
const configuredGraceMs = Math.max(1, configuredGraceSec) * 1000;
|
|
if (requestedGraceMs === undefined || !Number.isFinite(requestedGraceMs)) {
|
|
return configuredGraceMs;
|
|
}
|
|
return Math.max(100, Math.min(30_000, Math.trunc(requestedGraceMs)));
|
|
}
|
|
|
|
async function cancelRunInternal(
|
|
runId: string,
|
|
reason = "Cancelled by control plane",
|
|
options: CancelRunOptions = {},
|
|
) {
|
|
return cancelRun(runId, reason, options);
|
|
}
|
|
|
|
async function acknowledgeBudgetCancellation(run: typeof heartbeatRuns.$inferSelect, terminationProof?: string) {
|
|
if (!parseObject(parseObject(run.resultJson?.startupCancellation).budgetEnforcement).scopeType) return;
|
|
const acknowledged = await db.update(heartbeatRuns).set({ resultJson: sql`jsonb_set(${heartbeatRuns.resultJson},
|
|
'{startupCancellation,budgetEnforcement}', ${heartbeatRuns.resultJson}->'startupCancellation'->'budgetEnforcement'
|
|
|| ${JSON.stringify({ deliveredAt: new Date().toISOString(), ...(terminationProof ? { terminationProof } : {}) })}::jsonb)` })
|
|
.where(and(eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id),
|
|
pendingBudgetRunCancellationPredicate(),
|
|
terminationProof ? and(eq(heartbeatRuns.status, run.status), eq(heartbeatRuns.nextEventSeq, run.nextEventSeq),
|
|
sql`${heartbeatRuns.processPid} is not distinct from ${run.processPid}`,
|
|
sql`${heartbeatRuns.processGroupId} is not distinct from ${run.processGroupId}`,
|
|
sql`${heartbeatRuns.processStartedAt} is not distinct from ${run.processStartedAt?.toISOString() ?? null}::timestamptz`) : undefined,
|
|
)).returning({ id: heartbeatRuns.id });
|
|
if (terminationProof && !acknowledged.length)
|
|
throw conflict("Execution evidence changed before acknowledging budget Stop", { runId: run.id });
|
|
}
|
|
|
|
function hasRecordedStopAcknowledgement(run: typeof heartbeatRuns.$inferSelect) {
|
|
if (parseObject(run.resultJson?.executionCancellation).state === "acknowledged") return true;
|
|
const intent = parseObject(run.resultJson?.nativeCancellation);
|
|
return intent.schema === "paperclip.native-cancellation.v1" && intent.runId === run.id
|
|
&& intent.companyId === run.companyId && intent.issueId === run.nativeIssueId
|
|
&& intent.scope === "run" && intent.reasonCode === "cancellation_run_only"
|
|
&& intent.dispatchState === "acknowledged" && intent.dispatched === true
|
|
&& typeof intent.intentAuditId === "string" && typeof intent.acknowledgementAuditId === "string";
|
|
}
|
|
|
|
async function completedRunTerminationProof(run: typeof heartbeatRuns.$inferSelect): Promise<string | null> {
|
|
// Logical completion can win the race with queued Stop delivery. It does
|
|
// not prove a provider exited, and an in-process owner must still settle.
|
|
if (!isHeartbeatRunTerminalStatus(run.status) || activeRunExecutions.has(run.id) ||
|
|
runningProcesses.has(run.id) || adapterExecutionControls.has(run.id) || processRunCancellationSettlements.has(run.id)) return null;
|
|
const leases = await db.select().from(environmentLeases).where(and(
|
|
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
|
));
|
|
if (leases.some(lease => lease.provider !== "local"))
|
|
return leases.every(hasRemoteTerminationReceipt) ? "provider_termination_receipt" : null;
|
|
if (run.processPid || run.processGroupId || run.processStartedAt)
|
|
return await recordedLocalExecutionStopped(db, run) ? "local_process_exit" : null;
|
|
if (await readNativeLocalProcessStop(db, run.companyId, run.id)) return "local_process_stop_receipt";
|
|
// Normal native finalization clears mutable process fields. Rebind the
|
|
// latest server-owned launch identity before checking that exact local
|
|
// process and group. Any newer start invalidates the older observation.
|
|
const [identity] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload })
|
|
.from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.companyId, run.companyId),
|
|
eq(heartbeatRunEvents.runId, run.id), isNull(heartbeatRunEvents.sourceEventId),
|
|
inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, "native.local_process_stopped"])))
|
|
.orderBy(desc(heartbeatRunEvents.seq)).limit(1);
|
|
const recorded = identity?.payload;
|
|
if (identity?.eventType !== PROCESS_IDENTITY_RECORDED || typeof recorded?.processPid !== "number" ||
|
|
!Number.isSafeInteger(recorded.processPid) || recorded.processPid <= 1 ||
|
|
(recorded.processGroupId !== null && (typeof recorded.processGroupId !== "number" ||
|
|
!Number.isSafeInteger(recorded.processGroupId) || recorded.processGroupId <= 1)) ||
|
|
typeof recorded.processStartedAt !== "string") return null;
|
|
const processStartedAt = new Date(recorded.processStartedAt);
|
|
if (Number.isNaN(processStartedAt.getTime())) return null;
|
|
return await recordedLocalExecutionStopped(db, { ...run, processPid: recorded.processPid,
|
|
processGroupId: recorded.processGroupId, processStartedAt }) ? "local_process_exit" : null;
|
|
}
|
|
|
|
function deliveredBudgetCancellationResult(run: typeof heartbeatRuns.$inferSelect, deliveredAt: Date) {
|
|
const startup = parseObject(run.resultJson?.startupCancellation);
|
|
const intent = parseObject(startup.budgetEnforcement);
|
|
return intent.scopeType ? { startupCancellation: { ...startup,
|
|
budgetEnforcement: { ...intent, deliveredAt: deliveredAt.toISOString() } } } : {};
|
|
}
|
|
|
|
async function cancelRun(
|
|
runId: string,
|
|
reason: string,
|
|
options: CancelRunOptions,
|
|
committedBudgetCancellation = false,
|
|
) {
|
|
let run = await getRun(runId);
|
|
if (!run) throw notFound("Heartbeat run not found");
|
|
if (options.expectedStatus && run.status !== options.expectedStatus) return null;
|
|
const handoffIssueId = run.nativeIssueId ?? run.issueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null);
|
|
const supersedesCaseHandoff = options.resultJson?.cancelledByActorType === "user" || options.errorCode === "issue_reassigned";
|
|
let cancellationClaimed = false;
|
|
let caseHandoffRetryClaimed = false;
|
|
if (handoffIssueId && supersedesCaseHandoff && options.resultJson?.caseHandoffYieldStop !== true) {
|
|
const publications: ActivityPublication[] = [];
|
|
const applySupersession = () => db.transaction(async tx => {
|
|
const [issue] = await tx.select({ id: issues.id, executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId,
|
|
assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId }).from(issues)
|
|
.where(and(eq(issues.companyId, run!.companyId), eq(issues.id, handoffIssueId))).for("update");
|
|
const [current] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId),
|
|
eq(heartbeatRuns.companyId, run!.companyId))).for("update");
|
|
if (!current || (options.expectedStatus && current.status !== options.expectedStatus)) return false;
|
|
run = current;
|
|
if (options.cancellationRequestId) {
|
|
run = await claimCancellationRequest(tx as unknown as Db, runId, run!.companyId, options.cancellationRequestId,
|
|
options.cancellationRequestedByUserId ?? null);
|
|
cancellationClaimed = true;
|
|
}
|
|
// Acceptance takes task -> run locks too. Revoke the source's write
|
|
// authority in the same commit that cancels its outstanding intents,
|
|
// before waiting for process/native Stop. Otherwise a handoff can be
|
|
// accepted in the gap after this cancellation sweep.
|
|
const currentRun = run!;
|
|
const cancellableStatuses: string[] = options.expectedStatus ? [options.expectedStatus] : [...CANCELLABLE_HEARTBEAT_RUN_STATUSES];
|
|
if (!options.expectedStatus && currentRun.runtimeMode === "native" && currentRun.status === "failed" && currentRun.nativeIssueId) {
|
|
// Failed native runs may still own a retry. As in correlated Stop,
|
|
// lock the coordinator NOWAIT so a coordinator-only writer wins
|
|
// safely instead of deadlocking with its run-row claim.
|
|
const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and(
|
|
eq(nativeRunFinalizations.runId, runId), eq(nativeRunFinalizations.companyId, currentRun.companyId),
|
|
eq(nativeRunFinalizations.issueId, currentRun.nativeIssueId))).for("update", { noWait: true }).limit(1);
|
|
caseHandoffRetryClaimed = nativeRetryCancellationEligible({ runId, companyId: currentRun.companyId,
|
|
issueId: currentRun.nativeIssueId, resultJson: currentRun.resultJson, requestId: options.cancellationRequestId }, coordinator);
|
|
if (caseHandoffRetryClaimed) cancellableStatuses.push("failed");
|
|
}
|
|
const requestedAt = new Date().toISOString();
|
|
const guardedCancellation = options.expectedStatus ? {
|
|
cancellation: (currentRun.resultJson?.startupCancellation ? readRunCancellation(currentRun.resultJson) : null)
|
|
?? requestedRunCancellation(options.resultJson ?? {}, reason),
|
|
// Promotion must remain fenced after this transaction releases the
|
|
// run lock, before the ordinary Stop path performs its settlement.
|
|
startupCancellation: { requestedAt, beforeNativeSelection: currentRun.runtimeMode === "legacy" && !currentRun.runtimeModeResolvedAt
|
|
&& currentRun.executionStage === "preparing" && parseObject(currentRun.runnerProfileJson?.adapterDispatch).adapterType === "paperclip_runner" },
|
|
} : {};
|
|
const [fenced] = await tx.update(heartbeatRuns).set({
|
|
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
|
|
executionCancellation: { state: "requested", requestedAt }, ...guardedCancellation,
|
|
})}::jsonb`,
|
|
}).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, run!.companyId),
|
|
inArray(heartbeatRuns.status, cancellableStatuses))).returning();
|
|
if (options.expectedStatus && !fenced) return false;
|
|
if (fenced) run = fenced;
|
|
// A board terminal request deliberately has no source run. It belongs
|
|
// to the current task execution, not whichever historical run is stopped.
|
|
// Queue admission may bind execution before adopting an old checkout.
|
|
const ownsTask = issue && issue.assigneeAgentId === currentRun.agentId && !issue.assigneeUserId &&
|
|
(issue.executionRunId ? issue.executionRunId === runId : issue.checkoutRunId === runId);
|
|
await cancelCaseHandoffsForIssue(tx as unknown as Db, { companyId: run!.companyId, issueId: handoffIssueId,
|
|
sourceRunId: runId, includeUnboundTerminalIntent: Boolean(fenced && ownsTask), reason, publications });
|
|
return true;
|
|
});
|
|
let supersessionApplied = false;
|
|
for (let attempt = 0; ; attempt++) {
|
|
const previousRun = run;
|
|
try {
|
|
supersessionApplied = await applySupersession();
|
|
break;
|
|
} catch (error) {
|
|
const value = parseObject(error);
|
|
const deadlock = value.code === "40P01" || parseObject(value.cause).code === "40P01";
|
|
if (!deadlock || attempt >= 2) rethrowNativeCancellationLockConflict(error);
|
|
// PostgreSQL aborted the entire transaction. Discard its local
|
|
// receipts before re-reading current authority under fresh locks.
|
|
// Provider Stop and publication happen only after this boundary.
|
|
run = previousRun;
|
|
cancellationClaimed = false;
|
|
caseHandoffRetryClaimed = false;
|
|
publications.length = 0;
|
|
await new Promise(resolve => setTimeout(resolve, 25 * (attempt + 1)));
|
|
}
|
|
}
|
|
publications.forEach(publishActivity);
|
|
if (!supersessionApplied) return null;
|
|
}
|
|
if (options.cancellationRequestId && !cancellationClaimed) {
|
|
run = await claimCancellationRequest(db, runId, run.companyId, options.cancellationRequestId, options.cancellationRequestedByUserId ?? null);
|
|
}
|
|
// The caller claim checked retry eligibility under both durable row locks.
|
|
// This is only a cancellation candidate: dispatch rechecks the coordinator
|
|
// under lock, and the final CAS requires its own unchanged acknowledged
|
|
// retry fence. Re-reading only retryable_failure here would strand replay.
|
|
const pendingNativeRetry =
|
|
run.runtimeMode === "native" && run.status === "failed" && (
|
|
caseHandoffRetryClaimed || Boolean(options.cancellationRequestId) || await db
|
|
.select({ runId: nativeRunFinalizations.runId })
|
|
.from(nativeRunFinalizations)
|
|
.where(
|
|
and(
|
|
eq(nativeRunFinalizations.runId, run.id),
|
|
eq(nativeRunFinalizations.companyId, run.companyId),
|
|
eq(nativeRunFinalizations.phase, "retryable_failure"),
|
|
),
|
|
)
|
|
.then((rows) => rows.length > 0)
|
|
);
|
|
if (
|
|
!pendingNativeRetry &&
|
|
!CANCELLABLE_HEARTBEAT_RUN_STATUSES.includes(
|
|
run.status as (typeof CANCELLABLE_HEARTBEAT_RUN_STATUSES)[number],
|
|
)
|
|
) {
|
|
if (committedBudgetCancellation) {
|
|
const terminationProof = hasRecordedStopAcknowledgement(run) ? undefined : await completedRunTerminationProof(run);
|
|
if (terminationProof === null) {
|
|
throw conflict("Budget Stop is committed, but provider termination has not been acknowledged", { runId: run.id });
|
|
}
|
|
await acknowledgeBudgetCancellation(run, terminationProof);
|
|
}
|
|
return run;
|
|
}
|
|
const agent = await getAgent(run.agentId);
|
|
const errorCode = options.errorCode ?? "cancelled";
|
|
const cancellation = ((options.expectedStatus || committedBudgetCancellation) && run.resultJson?.startupCancellation
|
|
? readRunCancellation(run.resultJson) : null) ?? requestedRunCancellation(options.resultJson ?? {}, reason);
|
|
|
|
const pendingProcessCancellation = processRunCancellationSettlements.get(
|
|
run.id,
|
|
);
|
|
if (pendingProcessCancellation) {
|
|
await pendingProcessCancellation.settled;
|
|
if (pendingProcessCancellation.failed)
|
|
throw pendingProcessCancellation.error;
|
|
return getRun(run.id);
|
|
}
|
|
const running = runningProcesses.get(run.id);
|
|
const stopOwnership =
|
|
run.runtimeMode !== "native"
|
|
? captureAdapterStopOwnership(run.id)
|
|
: undefined;
|
|
const control = stopOwnership?.control;
|
|
// Capture the existing adapter owner before waiting on the run lock. Then
|
|
// atomically fence preparation and refresh the selected runtime, so Stop
|
|
// cannot miss a native handoff that won after its first read.
|
|
// Established legacy processes must still be stopped if the database is
|
|
// unavailable. Only native or not-yet-dispatched preparation needs this
|
|
// additional durable fence before its existing cancellation path.
|
|
if (!committedBudgetCancellation && (options.expectedStatus || options.budgetEnforcement || (!options.cancellationRequestId && (run.runtimeMode === "native" || (!run.runtimeModeResolvedAt && !running && !control))))) {
|
|
const fence = async (tx: Db) => {
|
|
const [fenced] = await tx.update(heartbeatRuns).set({
|
|
// Record handoff intent before native cancellation can finalize and release
|
|
// the run. Only its audited stop acknowledgement suppresses recovery.
|
|
resultJson: runCancellationFenceSql({ cancellation, errorCode,
|
|
reassignmentStopRequested: options.errorCode === "issue_reassigned" && options.resultJson?.reassignmentStopConfirmed === true }),
|
|
}).where(and(eq(heartbeatRuns.id, runId), inArray(heartbeatRuns.status,
|
|
options.expectedStatus ? [options.expectedStatus] : pendingNativeRetry ? [...CANCELLABLE_HEARTBEAT_RUN_STATUSES, "failed"] : [...CANCELLABLE_HEARTBEAT_RUN_STATUSES],
|
|
))).returning();
|
|
return fenced ?? null;
|
|
};
|
|
let fenced: typeof run | null;
|
|
try {
|
|
fenced = options.budgetEnforcement
|
|
? await withCurrentBudgetEnforcement(db, options.budgetEnforcement, fence)
|
|
: await fence(db);
|
|
} catch (error) {
|
|
stopOwnership?.release();
|
|
throw error;
|
|
}
|
|
if (!fenced) {
|
|
stopOwnership?.release();
|
|
return options.expectedStatus ? null : getRun(runId);
|
|
}
|
|
run = fenced;
|
|
}
|
|
const resultJson = { ...(agent
|
|
? {
|
|
...mergeRunStopMetadataForAgent(agent, "cancelled", {
|
|
resultJson: parseObject(run.resultJson),
|
|
errorCode,
|
|
errorMessage: reason,
|
|
}),
|
|
...(options.resultJson ?? {}),
|
|
}
|
|
: options.resultJson), cancellation };
|
|
|
|
try {
|
|
let releaseProcessCancellation: (() => void) | undefined;
|
|
const processCancellationSettlement =
|
|
run.runtimeMode !== "native" &&
|
|
!control &&
|
|
running
|
|
? {
|
|
settled: new Promise<void>((resolve) => {
|
|
releaseProcessCancellation = resolve;
|
|
}),
|
|
failed: false,
|
|
error: undefined as unknown,
|
|
}
|
|
: undefined;
|
|
if (processCancellationSettlement) {
|
|
// No await between joining an existing owner above and registering ours.
|
|
processRunCancellationSettlements.set(
|
|
run.id,
|
|
processCancellationSettlement,
|
|
);
|
|
}
|
|
const stopRequestedAt = Date.now();
|
|
const cancellation = await (async () => {
|
|
try {
|
|
if (control) {
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
error: reason,
|
|
errorCode,
|
|
// Stop may have read the row before adapter settlement saved
|
|
// a required-copyback receipt. Merge into current database JSON
|
|
// and leave those server-owned fields to the restore recorder.
|
|
resultJson: preserveWorkspaceRestoreRecoveryMetadataSql({
|
|
...resultJson,
|
|
...(!running ? { executionCancellation: {
|
|
state: "requested", requestedAt: new Date().toISOString(),
|
|
} } : {}),
|
|
}, true),
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, run.id),
|
|
eq(heartbeatRuns.status, "running"),
|
|
),
|
|
);
|
|
control.controller.abort(new Error(reason));
|
|
}
|
|
let terminationSettled = false;
|
|
try {
|
|
await cancelHeartbeatNativeRun({
|
|
db,
|
|
runId: run.id,
|
|
reason,
|
|
runtimeMode: run.runtimeMode,
|
|
...(options.cancellationRequestId ? { cancellationRequestId: options.cancellationRequestId } : {}),
|
|
});
|
|
if (running) {
|
|
await terminateHeartbeatRunProcess({
|
|
pid: running.child.pid,
|
|
processGroupId: running.processGroupId,
|
|
// Codex handles Ctrl-C by cancelling its tool sessions. SIGTERM
|
|
// can leave commands in their separate process groups alive.
|
|
signal: !control && agent?.adapterType === "codex_local" ? "SIGINT" : undefined,
|
|
graceMs: cancellationTerminationGraceMs(
|
|
running.graceSec,
|
|
options.terminationGraceMs,
|
|
),
|
|
});
|
|
}
|
|
terminationSettled = true;
|
|
} finally {
|
|
if (
|
|
(!processCancellationSettlement || terminationSettled) &&
|
|
runningProcesses.get(run.id) === running
|
|
) {
|
|
runningProcesses.delete(run.id);
|
|
}
|
|
}
|
|
|
|
if (control) {
|
|
const stopDiagnostics = {
|
|
runId: run.id,
|
|
adapterType: agent?.adapterType,
|
|
runtimeMode: run.runtimeMode,
|
|
abortRequested: control.controller.signal.aborted,
|
|
};
|
|
try {
|
|
await waitForAdapterStop(control.settled, undefined, stopDiagnostics, control);
|
|
} catch (error) {
|
|
if (!(error instanceof AdapterStopTimeoutError)) throw error;
|
|
// The owned adapter is still settling (for example copying the
|
|
// workspace back) and keeps its Stop proof requirements. Answer
|
|
// this request honestly; report only a Stop that never settles.
|
|
reportAdapterStopIfUnsettled(control.settled, captureException, stopDiagnostics, control, {
|
|
stopRequestedAt,
|
|
});
|
|
throw conflict(error.message, { code: "adapter_stop_pending", runId: run.id });
|
|
}
|
|
const stopped = await getRun(run.id);
|
|
if (stopped && isHeartbeatRunTerminalStatus(stopped.status)) {
|
|
if (
|
|
parseObject(stopped.resultJson?.executionCancellation).state !==
|
|
"acknowledged"
|
|
) {
|
|
throw conflict(
|
|
"Execution ended, but provider termination could not be verified. Inspect the stopped run before continuing.",
|
|
);
|
|
}
|
|
// The owned adapter already finalized this run and its lifecycle.
|
|
// Do not replay the process cancellation side effects below.
|
|
await acknowledgeBudgetCancellation(stopped);
|
|
return { run: stopped, updated: false };
|
|
}
|
|
}
|
|
|
|
const finishedAt = new Date();
|
|
const persistedCancellationResult =
|
|
run.runtimeMode === "native"
|
|
? await getRun(run.id).then((current) =>
|
|
parseObject(current?.resultJson),
|
|
)
|
|
: {};
|
|
return await setRunStatusFromLive(
|
|
run.id,
|
|
"cancelled",
|
|
pendingNativeRetry
|
|
? [...CANCELLABLE_HEARTBEAT_RUN_STATUSES, "failed"]
|
|
: [...CANCELLABLE_HEARTBEAT_RUN_STATUSES],
|
|
{
|
|
finishedAt,
|
|
error: reason,
|
|
errorCode,
|
|
...(resultJson ||
|
|
Object.keys(persistedCancellationResult).length > 0
|
|
? {
|
|
resultJson: {
|
|
...persistedCancellationResult,
|
|
...(resultJson ?? {}),
|
|
// A scheduler placeholder has no process to acknowledge.
|
|
// Preserve its normal release policy instead of treating
|
|
// it as an operator stop of provider work.
|
|
...(processCancellationSettlement && agent && running && (
|
|
(Number.isInteger(running.child.pid) && (running.child.pid ?? 0) > 0) ||
|
|
(Number.isInteger(running.processGroupId) && (running.processGroupId ?? 0) > 0)
|
|
)
|
|
? mergeRunStopMetadataForAgent(agent, "cancelled", {
|
|
resultJson: {
|
|
...resultJson,
|
|
executionCancellation: { state: "acknowledged", acknowledgedAt: finishedAt.toISOString() },
|
|
},
|
|
errorCode, errorMessage: reason,
|
|
})
|
|
: {}),
|
|
// Commit this receipt with the terminal transition. A
|
|
// crash after provider Stop cannot strand an undispatched
|
|
// queued run whose normal path needs no process receipt.
|
|
...deliveredBudgetCancellationResult(run, finishedAt),
|
|
// The native cancellation helper may have advanced a durable
|
|
// pending intent to its acknowledged state after `run` was
|
|
// first read. Never let that stale snapshot overwrite the
|
|
// authoritative post-dispatch acknowledgement.
|
|
...(Object.hasOwn(
|
|
persistedCancellationResult,
|
|
"nativeCancellation",
|
|
)
|
|
? {
|
|
nativeCancellation:
|
|
persistedCancellationResult.nativeCancellation,
|
|
}
|
|
: {}),
|
|
},
|
|
}
|
|
: {}),
|
|
},
|
|
undefined,
|
|
pendingNativeRetry ? nativeRetryCancellationCommitCondition(persistedCancellationResult) : undefined,
|
|
);
|
|
} catch (error) {
|
|
if (processCancellationSettlement) {
|
|
processCancellationSettlement.failed = true;
|
|
processCancellationSettlement.error = error;
|
|
if (activeRunExecutions.has(run.id)) {
|
|
failedProcessRunCancellations.set(
|
|
run.id,
|
|
processCancellationSettlement,
|
|
);
|
|
}
|
|
}
|
|
throw error;
|
|
} finally {
|
|
if (processCancellationSettlement) {
|
|
if (
|
|
processRunCancellationSettlements.get(run.id) ===
|
|
processCancellationSettlement
|
|
) {
|
|
processRunCancellationSettlements.delete(run.id);
|
|
}
|
|
// Always settle waiters, including termination and DB errors. A
|
|
// failed Stop remains an error to its caller, never a cancellation
|
|
// receipt; the signal-bearing executor may then record failure.
|
|
releaseProcessCancellation?.();
|
|
}
|
|
}
|
|
})();
|
|
const cancelled = cancellation.run;
|
|
|
|
if (cancellation.updated && cancelled) {
|
|
await setWakeupStatus(run.wakeupRequestId, "cancelled", {
|
|
finishedAt: cancelled.finishedAt ?? new Date(),
|
|
error: reason,
|
|
});
|
|
await appendRunEvent(cancelled, {
|
|
eventType: "lifecycle",
|
|
stream: "system",
|
|
level: "warn",
|
|
message: options.eventMessage ?? "run cancelled",
|
|
payload: { ...options.eventPayload, cancellation: readRunCancellation(cancelled.resultJson) },
|
|
});
|
|
await releaseIssueExecutionAndPromote(cancelled, {
|
|
suppressImmediateRecovery: options.suppressImmediateRecovery,
|
|
});
|
|
await finalizeAgentStatus(run.agentId, "cancelled", undefined, {
|
|
wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run),
|
|
});
|
|
await startNextQueuedRunForAgent(run.agentId);
|
|
}
|
|
return cancelled;
|
|
} finally {
|
|
stopOwnership?.release();
|
|
}
|
|
}
|
|
|
|
async function cancelActiveForAgentInternal(
|
|
agentId: string,
|
|
reason = "Cancelled due to agent pause",
|
|
errorCode = "cancelled",
|
|
) {
|
|
const runs = await db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.agentId, agentId),
|
|
inArray(heartbeatRuns.status, [...CANCELLABLE_HEARTBEAT_RUN_STATUSES]),
|
|
),
|
|
);
|
|
|
|
for (const run of runs) await cancelRunInternal(run.id, reason, { errorCode });
|
|
|
|
return runs.length;
|
|
}
|
|
|
|
async function cancelPendingWakeupsForAgentsInternal(
|
|
agentIds: string[],
|
|
reason: string,
|
|
) {
|
|
const uniqueAgentIds = [...new Set(agentIds)].filter(
|
|
(agentId) => agentId.length > 0,
|
|
);
|
|
if (uniqueAgentIds.length === 0) return 0;
|
|
|
|
const now = new Date();
|
|
const wakeupIds = await db
|
|
.select({ id: agentWakeupRequests.id })
|
|
.from(agentWakeupRequests)
|
|
.where(
|
|
and(
|
|
inArray(agentWakeupRequests.agentId, uniqueAgentIds),
|
|
inArray(agentWakeupRequests.status, [
|
|
"queued",
|
|
"deferred_issue_execution",
|
|
]),
|
|
sql`${agentWakeupRequests.runId} is null`,
|
|
),
|
|
)
|
|
.then((rows) => rows.map((row) => row.id));
|
|
|
|
if (wakeupIds.length === 0) return 0;
|
|
|
|
await db
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
status: "cancelled",
|
|
finishedAt: now,
|
|
error: reason,
|
|
updatedAt: now,
|
|
})
|
|
.where(inArray(agentWakeupRequests.id, wakeupIds));
|
|
|
|
return wakeupIds.length;
|
|
}
|
|
|
|
async function cancelInvocationsForAgentsInternal(
|
|
agentIds: string[],
|
|
reason: string,
|
|
) {
|
|
const uniqueAgentIds = [...new Set(agentIds)].filter(
|
|
(agentId) => agentId.length > 0,
|
|
);
|
|
let runsCancelled = 0;
|
|
for (const agentId of uniqueAgentIds) {
|
|
runsCancelled += await cancelActiveForAgentInternal(agentId, reason);
|
|
}
|
|
const wakeupsCancelled = await cancelPendingWakeupsForAgentsInternal(
|
|
uniqueAgentIds,
|
|
reason,
|
|
);
|
|
return {
|
|
agentIds: uniqueAgentIds,
|
|
runsCancelled,
|
|
wakeupsCancelled,
|
|
};
|
|
}
|
|
|
|
async function stopBrowserSessionsForBudgetScope(tx: Db, scope: BudgetEnforcementScope) {
|
|
// Browser work can outlive its terminal parent run. Its admitted project
|
|
// snapshot remains authoritative even if the task moves to another project.
|
|
const activeHold = sql`(select ${browserUseRuns.result}->'paperclipBudgetHold'
|
|
from ${browserUseRuns} where ${browserUseRuns.companyId} = ${browserUseSessions.companyId}
|
|
and ${browserUseRuns.sessionId} = ${browserUseSessions.id}
|
|
and ${browserUseRuns.result}->'paperclipBudgetHold'->>'state' = 'held'
|
|
and ${browserUseRuns.result}->'paperclipBudgetHold'->>'invocationId' = ${browserUseRuns.invocationId}::text
|
|
order by ${browserUseRuns.createdAt} desc, ${browserUseRuns.id} desc limit 1)`;
|
|
const project = sql`case when ${activeHold} ? 'projectId' then ${activeHold}->>'projectId'
|
|
else (select ${issues.projectId}::text from ${issues}
|
|
where ${issues.id} = ${browserUseSessions.issueId} and ${issues.companyId} = ${scope.companyId}) end`;
|
|
const sessions = await tx.select({ id: browserUseSessions.id }).from(browserUseSessions).where(and(
|
|
eq(browserUseSessions.companyId, scope.companyId),
|
|
inArray(browserUseSessions.status, ["starting", "running", "idle", "stopping"]),
|
|
scope.scopeType === "agent" ? eq(browserUseSessions.agentId, scope.scopeId) : undefined,
|
|
scope.scopeType === "project" ? sql`${project} = ${scope.scopeId}` : undefined,
|
|
scope.createdBefore ? lt(browserUseSessions.createdAt, scope.createdBefore) : undefined,
|
|
)).for("update");
|
|
if (!sessions.length) return;
|
|
await notifyDeliveryWork(tx, DELIVERY_QUEUES.browser, scope.companyId);
|
|
await tx.update(browserUseSessions).set({ status: "stopping", stopRequested: "end",
|
|
nextPollAt: new Date(), updatedAt: new Date(),
|
|
}).where(and(eq(browserUseSessions.companyId, scope.companyId),
|
|
inArray(browserUseSessions.id, sessions.map(session => session.id))));
|
|
}
|
|
|
|
async function cancelBudgetScopeWork(scope: BudgetEnforcementScope) {
|
|
const reason = "Cancelled due to budget pause";
|
|
const newlyFenced = await withCurrentBudgetEnforcement(db, scope, async (tx) => {
|
|
// Queue admission and every selected run lose authority in this commit,
|
|
// before any provider can delay cancellation delivery.
|
|
await cancelPendingWakeupsForBudgetScope(scope, tx);
|
|
await stopBrowserSessionsForBudgetScope(tx, scope);
|
|
const candidates = scope.scopeType === "project"
|
|
? await listProjectScopedRunIds(scope.companyId, scope.scopeId, scope.createdBefore, tx)
|
|
: await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.companyId, scope.companyId),
|
|
scope.scopeType === "agent" ? eq(heartbeatRuns.agentId, scope.scopeId) : undefined,
|
|
scope.createdBefore ? lt(heartbeatRuns.createdAt, scope.createdBefore) : undefined,
|
|
inArray(heartbeatRuns.status, [...CANCELLABLE_HEARTBEAT_RUN_STATUSES]),
|
|
)).then((rows) => rows.map((row) => row.id));
|
|
if (!candidates.length) return [];
|
|
return tx.update(heartbeatRuns).set({ resultJson: runCancellationFenceSql({
|
|
cancellation: requestedRunCancellation({}, reason), errorCode: "cancelled", preserveCancellation: true, budgetEnforcement: scope,
|
|
}) }).where(and(eq(heartbeatRuns.companyId, scope.companyId), inArray(heartbeatRuns.id, candidates),
|
|
inArray(heartbeatRuns.status, [...CANCELLABLE_HEARTBEAT_RUN_STATUSES]),
|
|
scope.createdBefore ? lt(heartbeatRuns.createdAt, scope.createdBefore) : undefined,
|
|
)).returning({ id: heartbeatRuns.id }).then((rows) => rows.map((row) => row.id));
|
|
});
|
|
// A later policy change revokes permission to fence new work, not the Stop
|
|
// already committed for older work. Recovery must deliver those intents.
|
|
const pending = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
|
pendingBudgetRunCancellationPredicate(), budgetCancellationScopePredicate(scope),
|
|
));
|
|
const runIds = [...new Set([...(newlyFenced ?? []), ...pending.map(run => run.id)])];
|
|
if (!runIds.length) return;
|
|
const failures: unknown[] = [];
|
|
let nextRun = 0;
|
|
await Promise.all(Array.from({ length: Math.min(4, runIds.length) }, async () => {
|
|
while (nextRun < runIds.length) {
|
|
const runId = runIds[nextRun++]!;
|
|
try {
|
|
// The committed intent owns these old runs even if a later budget
|
|
// edit admits new work. Do not revoke or repeat that earlier fence.
|
|
await withBudgetStopDelivery(db, async () => {
|
|
const [attempted] = await db.update(heartbeatRuns).set({ resultJson: sql`jsonb_set(${heartbeatRuns.resultJson},
|
|
'{startupCancellation,budgetEnforcement,lastAttemptAt}', ${JSON.stringify(new Date().toISOString())}::jsonb)` })
|
|
.where(and(eq(heartbeatRuns.companyId, scope.companyId), eq(heartbeatRuns.id, runId),
|
|
pendingBudgetRunCancellationPredicate())).returning({ id: heartbeatRuns.id });
|
|
if (attempted) await cancelRun(runId, reason, {}, true);
|
|
});
|
|
} catch (error) {
|
|
failures.push(error);
|
|
}
|
|
}
|
|
}));
|
|
// Drain the batch before rejecting; one failed provider must not prevent
|
|
// the remaining runs from receiving their already-committed Stop.
|
|
if (failures.length) throw new AggregateError(failures, "Budget cancellation delivery remains pending");
|
|
}
|
|
|
|
return {
|
|
cancelBudgetScopeWork,
|
|
releaseIssueExecutionAndPromote,
|
|
releaseEnvironmentLeasesForRun,
|
|
acknowledgeRemoteStop,
|
|
resumeRemoteStopComments,
|
|
cancelRunInternal,
|
|
resumeExecutionWaitComments,
|
|
resumeQueuedCommentInterrupt,
|
|
resumeSavedLegacyComments: resumeSavedComments,
|
|
resumeTaskRetry,
|
|
cancelActiveForAgentInternal,
|
|
cancelInvocationsForAgentsInternal,
|
|
};
|
|
}
|