diff --git a/server/src/__tests__/heartbeat-held-comment-wake.test.ts b/server/src/__tests__/heartbeat-held-comment-wake.test.ts index 31b747e9c24..8f57d676e30 100644 --- a/server/src/__tests__/heartbeat-held-comment-wake.test.ts +++ b/server/src/__tests__/heartbeat-held-comment-wake.test.ts @@ -148,6 +148,33 @@ describePostgres("a promoted comment refused by a temporary admission hold", () expect(deliveries(f.commentId)).toHaveLength(1); }, 180_000); + it("saves a comment whose wake the hold refuses at enqueue and delivers it once after the hold clears", async () => { + // s7532/t06: the comment arrived during the hold; its wake was refused + // before admission and written as a terminal skip, so it was lost. + const f = await seedHeldComment(); + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorUserId: f.userId, body: "Posted during the hold." }).returning(); + await expect(heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" }, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: comment.id, wakeCommentId: comment.id, + source: "issue.comment", wakeReason: "issue_commented" }, + requestedByActorType: "user", requestedByActorId: f.userId })).rejects.toThrow(/await accounting/); + const saved = (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))) + .filter(wake => (wake.payload as { commentId?: string }).commentId === comment.id); + expect(saved).toEqual([expect.objectContaining({ status: "deferred_issue_execution", runId: null, + payload: expect.objectContaining({ executionWait: expect.objectContaining({ reason: "budget.blocked" }) }) })]); + await settle(f); + expect(deliveries(comment.id)).toHaveLength(0); + + await settleAccounting(f); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }).where(eq(agentWakeupRequests.id, saved[0].id)); + await settle(f); await settle(f); + expect(deliveries(comment.id)).toHaveLength(1); + await settle(f); + expect(deliveries(comment.id)).toHaveLength(1); + }, 180_000); + it("keeps a Retry clicked during the hold pending and starts it exactly once after the hold clears", async () => { const f = await seedHeldComment(); // The run the user retries failed; its saved comment was already discarded. diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 376e00aac93..a1782ae298c 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -94,6 +94,32 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes onBlocked: reason => blocked.push(reason) }); return { result, blocked, reads }; } + it.each([ + { shape: "refused by the cleanup quarantine before launch", errorCode: "native_session_cleanup_quarantined", launched: false, + comment: "local_cleanup", retry: "cleanup_quarantined" }, + { shape: "quarantined after it requested a launch", errorCode: "native_session_cleanup_quarantined", launched: true, + comment: "process_identity_missing", retry: "process_identity_missing" }, + { shape: "failed for another reason without a process record", errorCode: "adapter_failed", launched: false, + comment: "process_identity_missing", retry: "process_identity_missing" }, + ])("routes input after a run $shape to the gate that can clear it", async ({ errorCode, launched, comment, retry }) => { + // s7532/t04: a Retry's run failed at the quarantine fence before it asked + // for a runner. Input then waited for a stop record of a process that + // never existed ("process_identity_missing", Retry still offered) instead + // of the quarantine's own gate, which names the action that clears it. + const f = await seed(); + await db.update(heartbeatRuns).set({ processPid: null, errorCode }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (launched) await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.sourceRunId, agentId: f.agentId, + seq: 1, eventType: "native.process_start_requested", stream: "system", level: "info", message: "launch requested", + payload: { hostname: "another-host" } }); + const { result, blocked } = await readAdmission(f, false); + expect(result).toBeNull(); + expect(blocked).toEqual([comment]); + const retryBlocked: string[] = []; + expect(await admitExplicitNativeContinuation({ ...f, db, dryRun: true, reason: "retry_failed_run", + failedRunId: f.sourceRunId, onBlocked: reason => retryBlocked.push(reason) })).toBeNull(); + expect(retryBlocked).toEqual([retry]); + }); + it("keeps admission and the complete locked session profile while omitting prompt data", async () => { const f = await seed(); const payload = randomBytes(128 * 1024).toString("base64"); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index d96797f44e9..d48da1101de 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -16,7 +16,7 @@ import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; import { canContinueCancelledRun, userStopBoundaryAt } from "./run-cancellation.js"; import { readQueuedInteractionResponse } from "./queued-interaction-response.js"; import { isCancelledNativeStartup, isFailedNativePreparation, isFailedNativePreparationRetryAncestor } from "./cancelled-native-startup.js"; -import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, recordedLocalExecutionStopped, recordUnidentifiedLocalRunnerStop } from "./native-local-process-stop.js"; +import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, nativeRunRefusedBeforeLaunch, recordedLocalExecutionStopped, recordUnidentifiedLocalRunnerStop } from "./native-local-process-stop.js"; import { completeTerminatedRemoteNativeSessionCleanup, nativeSessionCleanupBlock } from "../vendor/paperclip-runner/index.js"; import { hasRecoveredCleanupStopReceipt, hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js"; import { z } from "zod"; @@ -508,6 +508,7 @@ export async function admitExplicitNativeContinuation(input: { if (!run.processPid && !run.processGroupId && !await hasNativeLocalProcessStop(db, companyId, run.id) && !await hasHistoricalSuspendedNativeSession(db, run) && + !await nativeRunRefusedBeforeLaunch(db, run) && !(run.runtimeMode === "native" && await recordUnidentifiedLocalRunnerStop(db, run, { record: !input.dryRun }))) return blocked("process_identity_missing", "The previous run has no verified stop record. Paperclip cannot start this message yet."); if (run.processPid && !processStopped(run.processPid)) return blocked("process_running", "Waiting for the previous process to stop. Your message will start automatically."); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index d5f85b0794e..8acb6924cf7 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -2729,6 +2729,31 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }, ); if (budgetBlock) { + // A user's message refused by a temporary hold (accounting still settling) + // is saved as a deferred receipt, not skipped. It is promoted through + // normal admission once the run it waits behind ends or the hold clears. + if (!executionWaitRequestId && !caseRequest && !durableRequest && issueId && isUuidLike(issueId) && + opts.requestedByActorType === "user" && wakeCommentId && + !(await isBoardOwnedBudgetBlock(db, agent.companyId, budgetBlock))) { + const savedIssueId = issueId; + const condition = { reason: "budget.blocked", message: `${budgetBlock.reason} Your message is saved.`, + scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId }; + await db.transaction(async tx => { + await tx.execute(sql`select id from issues where id = ${savedIssueId} and company_id = ${agent.companyId} for update`); + await notifyDeferredExecutionWork(tx, { companyId: agent.companyId }); + await tx.insert(agentWakeupRequests).values({ + companyId: agent.companyId, agentId, source, triggerDetail, reason, + payload: withQueuedCommentIdsInWakePayload({ + ...payload, issueId: savedIssueId, [DEFERRED_WAKE_CONTEXT_KEY]: enrichedContextSnapshot, executionWait: condition, + }, [...new Set([...queuedCommentIdsFromRunContext(enrichedContextSnapshot), wakeCommentId])]), + status: "deferred_issue_execution", + requestedByActorType: opts.requestedByActorType ?? null, + requestedByActorId: opts.requestedByActorId ?? null, + idempotencyKey: opts.idempotencyKey ?? null, + }); + }); + throw new PersistedExecutionWaitRefusal(budgetBlock.reason, { scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId }); + } const wait = await writeSkippedRequest("budget.blocked", { error: budgetBlock.reason }, { scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId, }); diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 70731d14533..5999ea523b6 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -134,6 +134,21 @@ export async function recordUnidentifiedLocalRunnerStop(db: Db, run: Pick): Promise { + if (run.runtimeMode !== "native" || run.status !== "failed" || run.processPid || run.processGroupId || + run.errorCode !== "native_session_cleanup_quarantined") return false; + const [launch] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED]), + )).limit(1); + return !launch; +} + /** Pre-receipt native runs can retain an exact suspended session after their * mutable process fields were cleared. This is admission evidence for a new * user turn only, never permission to replay the old run or infer its outcomes. diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 19e6a761dd5..e3b0d16f6ab 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -3145,9 +3145,14 @@ export async function verifyStoppedNativeSessionForContinuation( !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(retained.status) || !retained.finishedAt || retained.finishedAt > run.createdAt || retained.controllerLeaseExpiresAt && retained.controllerLeaseExpiresAt > new Date() || - !retained.processStartedAt || !run.processStartedAt || - retained.processStartedAt.getTime() !== run.processStartedAt.getTime() || - retained.processPid !== run.processPid || retained.processGroupId !== run.processGroupId) return null; + // A run refused before launch (no launch event, checked above) and + // with no process record owns no process: only the previous owner's + // stop evidence below can certify the session. A source that does + // carry a process identity must still match the owner's exactly. + (run.processPid || run.processGroupId || run.processStartedAt) && + (!retained.processStartedAt || !run.processStartedAt || + retained.processStartedAt.getTime() !== run.processStartedAt.getTime() || + retained.processPid !== run.processPid || retained.processGroupId !== run.processGroupId)) return null; const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.issueId, run.nativeIssueId), eq(nativeRunFinalizations.runId, retained.id), diff --git a/server/src/services/quarantined-task-retry.test.ts b/server/src/services/quarantined-task-retry.test.ts index 8a02a628709..f08669a6e84 100644 --- a/server/src/services/quarantined-task-retry.test.ts +++ b/server/src/services/quarantined-task-retry.test.ts @@ -244,6 +244,31 @@ describe("task Retry after inherited native cleanup quarantine", () => { } }); + it.each(["stopped", "source_launch_requested", "owner_running"])( + "certifies a quarantined source that never launched from the previous owner's stop (%s)", async mode => { + // s7532/t04: the quarantine fence refused the source before it asked for + // a runner, so it has no process record at all. Only the owner can prove + // the retained session stopped. + const base = await mkdtemp(join(tmpdir(), "quarantined-never-launched-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = base; + try { + const f = await seed(base); + await db.update(heartbeatRuns).set({ processPid: null, processGroupId: null, processStartedAt: null }) + .where(eq(heartbeatRuns.id, f.sourceId)); + if (mode === "source_launch_requested") await appendHeartbeatRunEvent(db, { + companyId: f.companyId, runId: f.sourceId, agentId: f.agentId, eventType: PROCESS_START_REQUESTED }); + if (mode === "owner_running") await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.ownerId)); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceId)); + const proof = await verifyStoppedNativeSessionForContinuation(db, source); + if (mode === "stopped") expect(proof).not.toBeNull(); + else expect(proof).toBeNull(); + } finally { + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(base, { recursive: true, force: true }); + } + }); + it("holds a Retry while the task's own provider session close is still retained, then starts it", async () => { const base = await mkdtemp(join(tmpdir(), "retained-close-retry-")); const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR;