diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 672d8b479ee..36c2cc39842 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2039,6 +2039,13 @@ task with a run refused for that hold. That covers a task created under the tree after the snapshot, a task moved out before release, and a task whose work an overlapping pause refused and released while this one still covered it. The scheduled-retry gate holds the pause row while it saves its refusal, so a concurrent release either sees the refusal or the retry sees no pause. +For an old workspace-wait retry without the JSON marker, a later saved user +message can continue only when an exact system lifecycle warning records that +pause refusal. An earlier Retry Now lifecycle record does not invalidate the +warning. The released pause and its task membership must match. +The parent must retain its host-issued pre-provider workspace-wait proof. +Provider activity, retained execution, an unmatched pause, or an operator Stop +keeps the old run held. The historical cancellation remains unchanged. The queued-run gate checks the hold again after saving its refusal and resumes the task itself if release already finished. A resume restores the assignee's execution only; like a wake that admission refuses during a hold, a non-assignee diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index 1a53d3971cf..1926dafef86 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -1,8 +1,8 @@ import { isRetainedWorkspaceWorkerStopProof } from "./native-runtime/native-workspace-platform-proof.js"; import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js"; import { createHash } from "node:crypto"; -import { and, eq, inArray, isNotNull, isNull, lte, or, sql } from "drizzle-orm"; -import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { and, eq, gt, inArray, isNotNull, isNull, lte, or, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRunEvents, heartbeatRuns, issueTreeHolds, issueTreeHoldMembers, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { isUuidLike } from "@paperclipai/shared"; import { claimedAdapterType } from "./conversation-continuation.js"; import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; @@ -182,17 +182,37 @@ async function isCancelledUnclaimedWorkspaceWaitRetry(db: Db, run: Run): Promise const cancellation = readRunCancellation(run.resultJson); const issueId = run.issueId ?? run.contextSnapshot?.issueId; if (run.runtimeMode !== "legacy" || run.startedAt !== null || run.executionStage !== null - || run.runnerProfileJson !== null || typeof issueId !== "string" || !isUuidLike(issueId) - || run.errorCode !== "cancelled" || run.resultJson?.stopReason !== "cancelled" - || cancellation?.source !== "control_plane" || cancellation.expected !== true || cancellation.initiator.type !== "system") return false; + || run.runnerProfileJson !== null || typeof issueId !== "string" || !isUuidLike(issueId)) return false; + const systemCancellation = run.errorCode === "cancelled" && run.resultJson?.stopReason === "cancelled" + && cancellation?.source === "control_plane" && cancellation.expected === true && cancellation.initiator.type === "system"; + const pausedBeforePromotion = run.errorCode === "issue_paused" && run.resultJson === null && run.finishedAt + ? await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents) + .innerJoin(issueTreeHolds, and(eq(issueTreeHolds.companyId, run.companyId), + sql`${issueTreeHolds.id}::text = ${heartbeatRunEvents.payload}->>'holdId'`, + sql`${issueTreeHolds.rootIssueId}::text = ${heartbeatRunEvents.payload}->>'rootIssueId'`)) + .innerJoin(issueTreeHoldMembers, and(eq(issueTreeHoldMembers.companyId, run.companyId), + eq(issueTreeHoldMembers.holdId, issueTreeHolds.id), eq(issueTreeHoldMembers.issueId, issueId), + eq(issueTreeHoldMembers.skipped, false))) + .where(and(eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + eq(heartbeatRunEvents.eventType, "lifecycle"), + eq(heartbeatRunEvents.stream, "system"), eq(heartbeatRunEvents.level, "warn"), + eq(heartbeatRunEvents.message, "Scheduled retry suppressed because the issue is held by an active subtree pause hold"), + sql`${heartbeatRunEvents.payload}->>'issueId' = ${issueId}`, + sql`${heartbeatRunEvents.payload}->>'scheduledRetryReason' = 'workspace_busy'`, + eq(issueTreeHolds.mode, "pause"), eq(issueTreeHolds.status, "released"), + lte(issueTreeHolds.createdAt, run.finishedAt), isNotNull(issueTreeHolds.releasedAt), + gt(issueTreeHolds.releasedAt, run.finishedAt))).limit(1).then(rows => rows.length > 0) + : false; + if (!systemCancellation && !pausedBeforePromotion) return false; const [source] = await db.select({ agentId: heartbeatRuns.agentId, createdAt: heartbeatRuns.createdAt, retryOfRunId: heartbeatRuns.retryOfRunId, }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.scheduledRetryReason, "workspace_busy"), sql`${heartbeatRuns.contextSnapshot}->>'retryReason' = 'workspace_busy'`, sql`${heartbeatRuns.contextSnapshot}->>'retryOfRunId' = ${heartbeatRuns.retryOfRunId}::text`, - sql`(${heartbeatRuns.resultJson} - array['stopReason', 'cancellation', 'startupCancellation', - 'timeoutFired', 'timeoutSource', 'timeoutConfigured', 'effectiveTimeoutSec']) = '{}'::jsonb`, + pausedBeforePromotion ? isNull(heartbeatRuns.resultJson) + : sql`(${heartbeatRuns.resultJson} - array['stopReason', 'cancellation', 'startupCancellation', + 'timeoutFired', 'timeoutSource', 'timeoutConfigured', 'effectiveTimeoutSec']) = '{}'::jsonb`, noRetainedExecutionPredicate(), noProviderOutputPredicate(), isNull(heartbeatRuns.runnerProfileJson), isNull(heartbeatRuns.usageJson))); if (!source?.retryOfRunId) return false; diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 756410efc6a..9da1c2a25de 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -4,6 +4,8 @@ import os from "node:os"; import { createSystemRecovery, failSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js"; import { instanceSettingsService } from "./instance-settings.js"; import { createPostgresWakeQueueAdapter } from "../modules/wake-queue/adapters/postgres.js"; +import { createPostgresRunDispatchAdapter } from "../modules/run-dispatch/adapters/postgres.js"; +import { issueTreeControlService } from "./issue-tree-control.js"; import { createReleaseIssueExecution } from "../modules/wake-queue/application/use-cases.js"; import * as nativeExecutor from "./native-runtime/native-session-executor.js"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; @@ -1300,6 +1302,109 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes eq(heartbeatRuns.status, "queued")))).toHaveLength(1); }); + async function seedHistoricalPausedWorkspaceRetry(parentHold = false, retryNow = false) { + const f = await seedCancelledWorkspaceWaitRetry(); + const dueAt = new Date("2026-09-11T10:00:00Z"); + const pauseRootIssueId = parentHold ? randomUUID() : f.issueId; + if (parentHold) { + await db.insert(issues).values({ id: pauseRootIssueId, companyId: f.companyId, + title: "Shared workspace parent", status: "in_progress", assigneeAgentId: f.agentId }); + await db.update(issues).set({ parentId: pauseRootIssueId }).where(eq(issues.id, f.issueId)); + } + const [pause] = await db.insert(issueTreeHolds).values({ companyId: f.companyId, + rootIssueId: pauseRootIssueId, mode: "pause", status: "active", createdAt: new Date(dueAt.getTime() - 1_000), + releasePolicy: { strategy: "manual" } }).returning(); + await db.insert(issueTreeHoldMembers).values({ companyId: f.companyId, holdId: pause.id, + issueId: f.issueId, parentIssueId: parentHold ? pauseRootIssueId : null, depth: parentHold ? 1 : 0, + issueTitle: "Paused workspace retry", issueStatus: "blocked", assigneeAgentId: f.agentId }); + await db.update(heartbeatRuns).set({ status: "scheduled_retry", errorCode: null, + finishedAt: null, resultJson: null, scheduledRetryAt: dueAt, scheduledRetryAttempt: 1, + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (retryNow) { + expect(await heartbeatService(db).retryScheduledRetryNow({ issueId: f.issueId, + actor: { actorType: "user", actorId: "board" }, now: dueAt })).toMatchObject({ outcome: "gate_suppressed" }); + const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.sourceRunId)) + .orderBy(heartbeatRunEvents.seq); + expect(events).toMatchObject([ + { seq: 1, eventType: "lifecycle", stream: "system", level: "info", message: "Scheduled retry was requested to run now" }, + { seq: 2, eventType: "lifecycle", stream: "system", level: "warn", + message: "Scheduled retry suppressed because the issue is held by an active subtree pause hold" }, + ]); + } else { + expect(await createPostgresRunDispatchAdapter(db).promoteOrCancelDueRetry({ + companyId: f.companyId, runId: f.sourceRunId, now: dueAt, + })).toMatchObject({ outcome: "gate_suppressed", errorCode: "issue_paused" }); + } + await db.update(heartbeatRuns).set({ resultJson: null }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await issueTreeControlService(db).releaseHold(f.companyId, pauseRootIssueId, pause.id, + { actor: { actorType: "user", actorId: "board" } }); + return { ...f, pauseId: pause.id, dueAt }; + } + + it.each([ + { parentHold: false, retryNow: false }, { parentHold: true, retryNow: false }, + { parentHold: false, retryNow: true }, { parentHold: true, retryNow: true }, + ])("resumes saved input after a historical pause suppressed an unclaimed workspace retry (parent hold: $parentHold, retry now: $retryNow)", async ({ parentHold, retryNow }) => { + const f = await seedHistoricalPausedWorkspaceRetry(parentHold, retryNow); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, + source: "on_demand", reason: "issue_commented", requestedByActorType: "user", requestedByActorId: "board", + status: "deferred_issue_execution", updatedAt: new Date(0), payload: { issueId: f.issueId, commentId: f.commentId, + executionWait: { reason: "execution_recovery" }, + _paperclipWakeContext: { issueId: f.issueId, wakeCommentId: f.commentId, wakeCommentIds: [f.commentId] } }, + }).returning(); + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeExecutionWaitComments(), heartbeat.resumeExecutionWaitComments()]); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.status, "queued"))); + expect(successors).toHaveLength(1); + expect(successors[0].contextSnapshot).toMatchObject({ forceFreshSession: true, wakeCommentIds: [f.commentId] }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0]) + .toMatchObject({ status: "coalesced", runId: successors[0].id }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)))[0]) + .toMatchObject({ status: "cancelled", startedAt: null, errorCode: "issue_paused", resultJson: null }); + }); + + it("holds a Retry Now pause history that also contains provider execution evidence", async () => { + const f = await seedHistoricalPausedWorkspaceRetry(false, true); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, agentId: f.agentId, runId: f.sourceRunId, + eventType: "turn.started", stream: "system", level: "info" }); + expect(await admit(f)).toBeNull(); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.status, "queued")))).toHaveLength(0); + }); + + it.each(["missing_event", "wrong_event", "missing_member", "active_pause", "late_pause", "early_release", + "started", "provider_event", "lease", "operator_stop", "parent_executed", "retry_binding"])( + "holds a historical pause-suppressed workspace retry with %s evidence", async kind => { + const f = await seedHistoricalPausedWorkspaceRetry(); + if (kind === "missing_event") await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.sourceRunId)); + if (kind === "wrong_event") await db.update(heartbeatRunEvents).set({ payload: { holdId: f.pauseId, + rootIssueId: f.issueId, issueId: randomUUID(), scheduledRetryReason: "workspace_busy" } }) + .where(eq(heartbeatRunEvents.runId, f.sourceRunId)); + if (kind === "missing_member") await db.delete(issueTreeHoldMembers).where(eq(issueTreeHoldMembers.holdId, f.pauseId)); + if (kind === "active_pause") await db.update(issueTreeHolds).set({ status: "active", releasedAt: null }) + .where(eq(issueTreeHolds.id, f.pauseId)); + if (kind === "late_pause") await db.update(issueTreeHolds).set({ createdAt: new Date(f.dueAt.getTime() + 1) }) + .where(eq(issueTreeHolds.id, f.pauseId)); + if (kind === "early_release") await db.update(issueTreeHolds).set({ releasedAt: new Date(f.dueAt.getTime() - 1) }) + .where(eq(issueTreeHolds.id, f.pauseId)); + if (kind === "started") await db.update(heartbeatRuns).set({ startedAt: f.dueAt }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "provider_event") await appendHeartbeatRunEvent(db, { companyId: f.companyId, + agentId: f.agentId, runId: f.sourceRunId, eventType: "turn.started", stream: "system", level: "info" }); + if (kind === "lease") await db.insert(environmentLeases).values({ companyId: f.companyId, + heartbeatRunId: f.sourceRunId, provider: "local", leasePolicy: "ephemeral", status: "released", + cleanupStatus: "success", releasedAt: new Date() }); + if (kind === "operator_stop") await db.update(heartbeatRuns).set({ errorCode: "operator_interrupted", + resultJson: { stopReason: "cancelled", cancellation: { source: "operator", expected: true, + initiator: { type: "user", userId: "board" }, reason: "Stopped", recordedAt: f.dueAt.toISOString() } } }) + .where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "parent_executed") await db.update(heartbeatRuns).set({ resultJson: { + executionRecovery: { kind: "workspace_wait", providerWorkStarted: true } } }).where(eq(heartbeatRuns.id, f.parentRunId)); + if (kind === "retry_binding") await db.update(heartbeatRuns).set({ retryOfRunId: null }).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(await admit(f)).toBeNull(); + }); + it.each(["message", "retry"])("admits an unclaimed workspace retry through the heartbeat %s path", async kind => { const f = await seedCancelledWorkspaceWaitRetry(); await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });