From 51b0e01ead0adca19eb2e7b459ca80ef5c62e280 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 11 Sep 2026 17:12:55 -0500 Subject: [PATCH 1/2] fix: resume saved user messages after execution recovery (#13270) Preserve verified native process-stop evidence and retry saved user messages through normal continuation admission after recovery cleanup. Show the current wait reason and serialize delivery so a saved message starts one fresh turn. Validated with 410 focused tests, typecheck, build, token gates, all PR CI checks, and Greptile 5/5. Co-Authored-By: Paperclip --- doc/SPEC-implementation.md | 2 +- doc/execution-semantics.md | 6 + doc/run-log-events.md | 15 + packages/shared/src/types/issue.ts | 2 + .../issue-queued-comments-routes.test.ts | 15 + server/src/routes/issues.ts | 3 + .../explicit-native-continuation.test.ts | 102 ++++++- .../services/explicit-native-continuation.ts | 27 +- server/src/services/heartbeat.ts | 273 +++++++++++++----- .../services/issue-queued-comment-queue.ts | 2 + .../src/services/native-local-process-stop.ts | 60 ++++ .../native-runtime/native-restart-recovery.ts | 5 + ...unner-restart-recovery.integration.test.ts | 5 + .../native-session-executor.test.ts | 23 +- .../native-runtime/native-session-executor.ts | 12 + .../task-chat/TaskChatQueuedMessages.test.tsx | 9 + .../task-chat/TaskChatQueuedMessages.tsx | 5 + 17 files changed, 471 insertions(+), 95 deletions(-) create mode 100644 server/src/services/native-local-process-stop.ts diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index f59b4f5b5a..1d60893b7a 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -41,7 +41,7 @@ These decisions close open questions from `SPEC.md` for V1. | Communication | Tasks + comments only (no separate chat system) | | Task ownership | Single assignee; atomic checkout required for `in_progress` transition | | Task watchdogs | A task watchdog is an explicitly configured, issue-subtree-scoped verification and recovery capacity. It may restore live task paths inside the watched subtree; for issue-thread interaction resolution it is an ordinary agent subject to the same audience and containment checks, not board authority, active-run output monitoring, or general liveness recovery. | -| Recovery | Liveness/watchdog recovery preserves explicit ownership: continue interrupted local conversations with bounded fresh turns and preserved history, never replay tool calls automatically; retain native ownership and real execution gates; otherwise open visible source-scoped recovery actions by default, use issue-backed recovery only for independent repair work, or require human escalation (see `doc/execution-semantics.md`) | +| Recovery | Liveness/watchdog recovery preserves explicit ownership: continue interrupted local conversations with bounded fresh turns and preserved history, never replay tool calls automatically; retain native ownership and real execution gates; preserve verified stop evidence and reconsider saved post-stop user messages after cleanup; otherwise open visible source-scoped recovery actions by default, use issue-backed recovery only for independent repair work, or require human escalation (see `doc/execution-semantics.md`) | | Agent adapters | Built-in `process`, `http`, local CLI/session adapters, and OpenClaw gateway support; external adapters can also be loaded through the adapter plugin flow | | Plugin framework | Local/self-hosted early plugin runtime is in scope; cloud marketplace and packaged public distribution remain out of scope | | Auth | Mode-dependent human auth (`local_trusted` implicit board in current code; authenticated mode uses sessions), API keys for agents | diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 8040a2f9e0..14807e2186 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -840,6 +840,12 @@ Real gates still apply: company and task ownership, active provider ownership, b An operator Stop waits for provider termination. Remote sandbox providers may return a stopped/deleted receipt after their control-plane operation completes. Paperclip binds that receipt to the company, run, and exact lease; successful file cleanup, a terminal run row, or an in-sandbox shutdown event is not sufficient. Legacy conversational runs receive their cancellation acknowledgement after all remote leases have confirmed termination. Stop alone never creates a continuation. A user message queued during remote cleanup is reconsidered when the provider confirms termination; it still passes normal admission and adopts pending comment IDs in order. Once stopped, the next explicit wake uses the same queue. A compatible saved ACP session can resume, and an unavailable or incompatible session can start fresh with the full task context. Run credentials and scratch paths remain scoped to the new run. A subtree pause requires Resume; a message does not bypass it. +For native conversations, an authenticated user message sent after the previous run finishes can retire its execution recovery holds and start a fresh turn. Hold retirement and the new run are atomic. The previous transcript, tool outcomes, and recovery history remain intact. This starts a new conversation; it does not replay tool calls with unknown outcomes. + +Local recovery records a server-authored stop receipt before it clears a verified absent process identity. A new execution request invalidates that receipt before any process can spawn; recording a new process identity also invalidates it. Missing process IDs without a receipt still block admission. Remote execution continues to require termination receipts for every lease. + +If cleanup or another execution gate is still pending, the message stays in its existing queue receipt. Startup and periodic scheduling reconsider up to 50 due receipts per pass, at most once per 30 seconds per receipt, without calling a model or resetting recovery attempts. Cleanup callbacks use the same admission path. The issue lock prevents concurrent workers from delivering an adopted or discarded receipt again. The queued-message area shows the current wait reason. Pauses, approvals, budgets, ownership, and external chat authorization remain enforced. A message sent before the run finished does not grant new post-stop authority. + Historical legacy interruption holds for conversational adapters no longer block new messages or Resume. Classification uses the run’s saved adapter invocation or continuation policy, never the agent’s current adapter settings. Missing historical adapter evidence retains the hold. A terminal row with a live predecessor process or unreleased environment lease still blocks actual admission and Resume. Retry scheduling can happen before cleanup, but grants no execution authority. Recovery folds their obsolete no-replay bookkeeping without changing task ownership, status, or automatically waking old work. The audit trail remains readable. Native integrity and ownership holds, and non-conversational adapter holds, remain enforced. The server projection remains available for diagnostics. Normal working, finishing, and interaction waits add no badges or cards to task lists or feeds. Active transcript headers keep saying Working during automatic retry and execution confirmation; attempts, causes, and recovery decisions belong in the run log. Recovery uses the existing transcript and run log rather than adding a reconciliation form. A cancelled run that never started says “Couldn't start” instead of implying that the agent answered. diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 62105ed90b..2e069de502 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -42,6 +42,21 @@ proofs, encryption keys, environment variables, provider credentials, command arguments, or an unsanitized stderr stream. Detailed failed-attempt diagnostics remain in the bounded `native_run_finalizations.recovery_history` ledger. +## Native Local Process Stop Evidence + +The server writes `native.local_process_stopped` in the same transaction that +clears a local run's process identity, after it verifies that its PID and process +group are absent. The payload contains only those process IDs. Remote process +IDs are never checked against the control-plane host. + +The server writes `native.process_start_requested` before a backend can spawn, +and `native.process_identity_recorded` when it stores a new native process +identity. Either invalidates an earlier local stop receipt, including a crash +before the new PID callback. Continuation admission accepts only the latest +server-authored event among these three types; provider +source events cannot supply stop authority. These records stay in the local run +log and do not add Telemetry or OpenTelemetry data. + ## Sandbox Startup Run-Log Event Paperclip writes one `run.startup.step` event to the run log for each bring-up diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index c7ef92579d..ea0a8e0440 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -989,6 +989,8 @@ export interface IssueQueuedCommentQueue { protocol: IssueQueuedCommentProtocol; steeringDisposition: IssueQueuedCommentSteeringDisposition; entries: IssueQueuedCommentEntry[]; + /** Current admission condition for a saved user continuation. */ + executionWait?: { reason: string; message: string } | null; } interface IssueCommentMetadataRowBase { diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index 41e7bd3992..b2f3b45a88 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -213,6 +213,21 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { return queueRunId; } + it("projects the recovery wait reason only while the message is deferred", async () => { + const seeded = await seedQueue(); + const executionWait = { reason: "remote_cleanup", message: "Waiting for the previous environment to stop." }; + await db.update(agentWakeupRequests).set({ + payload: sql`coalesce(${agentWakeupRequests.payload}, '{}'::jsonb) || ${JSON.stringify({ executionWait })}::jsonb`, + }).where(eq(agentWakeupRequests.id, seeded.wakeId)); + const waiting = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`); + expect(waiting.status).toBe(200); + expect(waiting.body.executionWait).toEqual(executionWait); + await promoteQueue(seeded); + const admitted = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`); + expect(admitted.status).toBe(200); + expect(admitted.body.executionWait).toBeUndefined(); + }); + it("returns the authoritative order, preserves full Markdown edits, and rejects stale revisions", async () => { const seeded = await seedQueue(); const initial = await request(app(seeded.companyId)) diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 6f1dd618b3..124e06a8a8 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -6889,8 +6889,11 @@ export function issueRoutes( ?? (await getNativeSessionSteeringState(steering.steeringRunId) .then((state) => state.disposition) .catch(() => "temporarily_unavailable" as const)); + const wait = queueState?.state === "deferred" ? readObject(readObject(wake?.payload).executionWait) : {}; return buildQueuedCommentQueueSnapshot({ issueId: input.issue.id, + executionWait: typeof wait.reason === "string" && typeof wait.message === "string" + ? { reason: wait.reason, message: wait.message } : null, queueId: wake?.id ?? null, state: queueState?.state ?? null, activeRunId: input.activeRun?.id ?? null, diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 6778021e9d..a535c0c867 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -1,3 +1,5 @@ +import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; +import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; import { remoteTerminationReceipt } from "./remote-execution-termination.js"; import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; @@ -10,7 +12,7 @@ import { import { startEmbeddedPostgresTestDatabase, getEmbeddedPostgresTestSupport } from "../__tests__/helpers/embedded-postgres.js"; import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; import { buildExecutionContinuation } from "./execution-continuation.js"; -import { heartbeatService, type HeartbeatEnvironmentRuntime } from "./heartbeat.js"; +import { heartbeatService, persistHeartbeatRunProcessMetadata, type HeartbeatEnvironmentRuntime } from "./heartbeat.js"; import { getExecutionBlocker } from "./execution-blocker.js"; const support = await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("explicit native conversation continuation", () => { @@ -47,6 +49,104 @@ const support = await getEmbeddedPostgresTestSupport(); agentId: f.agentId, status: "queued", contextSnapshot: { issueId: f.issueId, previousRunId: result.previousRunId, forceFreshSession: true } }); return result; }); + it("preserves local stop proof after process metadata is cleared and invalidates it on another launch", async () => { + const f = await seed(); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.transaction(async tx => { + expect(await recordNativeLocalProcessStop(tx as unknown as typeof db, source)).toBe(true); + await tx.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, source.id)); + }); + expect(await hasNativeLocalProcessStop(db, f.companyId, source.id)).toBe(true); + expect(await hasNativeLocalProcessStop(db, randomUUID(), source.id)).toBe(false); + expect(await admit(f, true)).toMatchObject({ previousRunId: source.id }); + await persistHeartbeatRunProcessMetadata(db, source.id, { pid: 999999999, processGroupId: null, startedAt: new Date().toISOString() }); + await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, source.id)); + expect(await admit(f, true)).toBeNull(); + expect(await recordNativeLocalProcessStop(db, source)).toBe(true); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: source.id, agentId: f.agentId, + eventType: PROCESS_START_REQUESTED }); + // No PID was stored for the new launch, as when the server dies after spawn. + expect(await admit(f, true)).toBeNull(); + }); + + it.each(["live", "remote", "provider_event"])("does not accept invalid local stop proof: %s", async kind => { + const f = await seed(); + if (kind === "remote") { + const [environment] = await db.insert(environments).values({ name: "Remote stop", driver: "sandbox" }).returning(); + await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.sourceRunId, + environmentId: environment.id, provider: "daytona", status: "active", leasePolicy: "ephemeral" }); + } + if (kind === "live") await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId)); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "provider_event") { + await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: source.id, agentId: f.agentId, + eventType: "native.local_process_stopped", seq: 1, sourceEventId: randomUUID() }); + } else expect(await recordNativeLocalProcessStop(db, source)).toBe(false); + expect(await hasNativeLocalProcessStop(db, f.companyId, source.id)).toBe(false); + }); + + it("resumes saved local messages after restart exactly once and keeps the same wait receipt while blocked", async () => { + const f = await seed(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId)); + // A prior cancelled admission is also held, but cannot select the native + // retry source. The newest blocker must win just as it does on Send. + const oldAdmissionId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: oldAdmissionId, companyId: f.companyId, agentId: f.agentId, + status: "cancelled", errorCode: "execution_reconciliation_required", contextSnapshot: { issueId: f.issueId }, + finishedAt: new Date("2026-09-11T09:00:00Z") }); + await db.update(issueRecoveryActions).set({ updatedAt: new Date(0), evidence: { runId: oldAdmissionId, + automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", cause: "uncertain_external_action", fingerprint: randomUUID(), status: "active", + nextAction: "Waiting for the current run.", evidence: { runId: f.sourceRunId } }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: f.sourceRunId }); + await heartbeatService(db).wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, commentId: f.commentId }, + contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId } }); + const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + expect(waiting.payload?.executionWait).toMatchObject({ reason: "process_running" }); + const makeDue = () => db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, waiting.id)); + await makeDue(); + await heartbeatService(db).resumeExecutionWaitComments(); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(1); + await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, f.sourceRunId)); + const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.transaction(async tx => { + await recordNativeLocalProcessStop(tx as unknown as typeof db, stopped); + await tx.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, stopped.id)); + }); + const holdId = randomUUID(); + await db.insert(issueTreeHolds).values({ id: holdId, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause", status: "active" }); + await db.insert(issueTreeHoldMembers).values({ companyId: f.companyId, holdId, issueId: f.issueId, depth: 0, issueTitle: "Deploy", issueStatus: "blocked" }); + for (let attempt = 0; attempt < 2; attempt++) { + await makeDue(); + await heartbeatService(db).resumeExecutionWaitComments(); + } + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(1); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + const [paused] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id)); + expect(paused.payload?.executionWait).toMatchObject({ reason: "issue_tree_hold_active" }); + await db.update(issueTreeHolds).set({ status: "released" }).where(eq(issueTreeHolds.id, holdId)); + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1, maxDailyRuns: 0 } } }).where(eq(agents.id, f.agentId)); + for (let attempt = 0; attempt < 2; attempt++) { + await makeDue(); + await heartbeatService(db).resumeExecutionWaitComments(); + } + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(1); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, f.agentId)); + await makeDue(); + await Promise.all([heartbeatService(db).resumeExecutionWaitComments(), heartbeatService(db).resumeExecutionWaitComments()]); + const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + expect(runs).toHaveLength(1); + expect(runs[0].contextSnapshot).toMatchObject({ forceFreshSession: true, previousRunId: f.sourceRunId, + explicitUserContinuation: { commentId: f.commentId } }); + const [adopted] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id)); + expect(adopted).toMatchObject({ status: "coalesced", runId: runs[0].id }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + }); + it.each([true, false])("acknowledges a legacy remote Stop only after confirmed lease cleanup: %s", async confirmed => { const f = await seed(); await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId)); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 9cae39bd3a..7e99eaa317 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -1,3 +1,4 @@ +import { hasNativeLocalProcessStop } from "./native-local-process-stop.js"; import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js"; import { z } from "zod"; @@ -29,8 +30,10 @@ export async function admitExplicitNativeContinuation(input: { actorType: string | null | undefined; actorId: string | null | undefined; reason: string | null; commentId: string | null; successorRunId: string; dryRun?: boolean; + onBlocked?: (reason: string, message: string) => void; }): Promise<{ previousRunId: string; commentId: string } | null> { const { db, companyId, issueId, agentId, actorId, commentId } = input; + const blocked = (reason: string, message: string) => { input.onBlocked?.(reason, message); return null; }; if (input.actorType !== "user" || !actorId || !commentId || !["issue_commented", "issue_reopened_via_comment"].includes(input.reason ?? "")) return null; if (!z.string().guid().safeParse(commentId).success) return null; @@ -60,20 +63,21 @@ export async function admitExplicitNativeContinuation(input: { eq(approvals.id, issueApprovals.approvalId), eq(approvals.companyId, companyId), )).where(and(eq(issueApprovals.companyId, companyId), eq(issueApprovals.issueId, issueId), inArray(approvals.status, ["pending", "revision_requested"]))).limit(1); - if (pendingInteraction || pendingApproval) return null; + if (pendingInteraction || pendingApproval) return blocked("decision_pending", "A pending approval or question must be resolved before this message can start."); const sources: Run[] = []; for (const action of actions) { const runId = action.evidence.runId ?? action.evidence.sourceRunId; - if (typeof runId !== "string") return null; + if (typeof runId !== "string") return blocked("source_missing", "The stopped run could not be identified. Your message is saved."); // Text comparison keeps malformed historical evidence a hold, not a UUID cast error. const [run] = await db.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), sql`${heartbeatRuns.id}::text = ${runId}`, )); if (!run || run.agentId !== agentId || !terminal.includes(run.status) || (run.nativeIssueId ?? run.contextSnapshot?.issueId) !== issueId || - !run.finishedAt || comment.createdAt <= run.finishedAt) return null; - if (adapterExecutionControls.has(run.id)) return null; + !run.finishedAt) return blocked("source_unavailable", "The previous execution has not finished or its owner changed. Your message is saved."); + if (comment.createdAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue."); + if (adapterExecutionControls.has(run.id)) return blocked("execution_settling", "Waiting for the previous run to stop. Your message will start automatically."); const unusedAdmission = run.status === "cancelled" && !run.startedAt && run.errorCode === "execution_reconciliation_required" && !run.processPid && !run.processGroupId && !run.nativeSessionId; @@ -82,7 +86,7 @@ export async function admitExplicitNativeContinuation(input: { eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id), )).for("update"); if (coordinator && (coordinator.phase !== "terminal_failure" || coordinator.leaseOwner || - coordinator.resultId || coordinator.failureDetail?.successorRunId)) return null; + coordinator.resultId || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling", "Waiting for the previous run to finish recovery. Your message will start automatically."); const leases = await db.select() .from(environmentLeases).where(and( eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id), @@ -90,17 +94,18 @@ export async function admitExplicitNativeContinuation(input: { const remote = leases.some(lease => lease.provider !== "local"); if (remote) { // Never interpret remote PIDs using the control-plane host's process table. - if (!leases.every(hasRemoteTerminationReceipt)) return null; + if (!leases.every(hasRemoteTerminationReceipt)) return blocked("remote_cleanup", "Waiting for the previous environment to stop. Your message will start automatically."); if (!input.dryRun && !leases.every(lease => completeTerminatedRemoteNativeSessionCleanup({ companyId, runId: run.id, remoteCleanupScope: remoteLeaseCleanupScope(lease)!, }))) return null; } else { - if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed")) return null; + if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed")) return blocked("local_cleanup", "Waiting for the previous environment to finish cleanup. Your message will start automatically."); if (!unusedAdmission) { // A missing process identity is not evidence that a provider exited. - if (!run.processPid && !run.processGroupId) return null; - if (run.processPid && !processStopped(run.processPid)) return null; - if (run.processGroupId && !processStopped(-run.processGroupId)) return null; + if (!run.processPid && !run.processGroupId && + !await hasNativeLocalProcessStop(db, companyId, run.id)) 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."); + if (run.processGroupId && !processStopped(-run.processGroupId)) return blocked("process_running", "Waiting for the previous process to stop. Your message will start automatically."); } } sources.push(run); @@ -113,7 +118,7 @@ export async function admitExplicitNativeContinuation(input: { inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]), ne(heartbeatRuns.id, input.successorRunId), )).limit(1); - if (active) return null; + if (active) return blocked("execution_active", "Waiting for the current run. Your message is saved."); const previous = nativeSources.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime())[0]!; // Prove required task history is available before retiring any hold. await buildExecutionContinuation({ db, companyId, issueId, agentId, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 32c26a8ffd..089df9c702 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,8 +1,9 @@ +import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js"; import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js"; import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js"; import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; -import { getExecutionBlocker } from "./execution-blocker.js"; +import { executionBlockerPredicate, getExecutionBlocker } from "./execution-blocker.js"; import { CONVERSATION_CONTINUATION_POLICY, runUsedConversationAdapter, hasConversationContinuationPolicy, isConversationAdapter } from "./conversation-continuation.js"; import { recordExecutionWait } from "./execution-wait.js"; import { @@ -49,6 +50,7 @@ import { asc, desc, eq, + exists, getTableColumns, gt, gte, @@ -8599,19 +8601,27 @@ export async function persistHeartbeatRunProcessMetadata( () => null, ); const startedAt = new Date(observedStartedAt ?? meta.startedAt); - return db - .update(heartbeatRuns) - .set({ - processPid: meta.pid, - processGroupId: meta.processGroupId, - processStartedAt: Number.isNaN(startedAt.getTime()) - ? new Date() - : startedAt, - updatedAt: new Date(), - }) - .where(eq(heartbeatRuns.id, runId)) - .returning() - .then((rows) => rows[0] ?? null); + return db.transaction(async tx => { + const run = await tx + .update(heartbeatRuns) + .set({ + processPid: meta.pid, + processGroupId: meta.processGroupId, + processStartedAt: Number.isNaN(startedAt.getTime()) + ? new Date() + : startedAt, + updatedAt: new Date(), + }) + .where(eq(heartbeatRuns.id, runId)) + .returning() + .then((rows) => rows[0] ?? null); + if (run?.runtimeMode === "native") await appendHeartbeatRunEvent(tx as unknown as Db, { + companyId: run.companyId, runId, agentId: run.agentId, + eventType: PROCESS_IDENTITY_RECORDED, stream: "system", level: "info", + message: "Process identity recorded; prior stop evidence no longer applies.", + }); + return run; + }); } async function terminateHeartbeatRunProcess(input: { @@ -9949,9 +9959,9 @@ export function heartbeatService( } } - async function resumeRemoteStopComments(run: typeof heartbeatRuns.$inferSelect) { - if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id) || - !(await remoteExecutionHasStopped(db, run.companyId, run.id))) return; + async function resumeRemoteStopComments(run: typeof heartbeatRuns.$inferSelect, requestId?: string) { + if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id)) return; + if (run.runtimeMode !== "native" && !(await remoteExecutionHasStopped(db, run.companyId, run.id))) return; const issueId = run.nativeIssueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null); if (!issueId) return; const legacyContinuation = run.runtimeMode === "legacy" && run.status === "cancelled" && @@ -9962,9 +9972,11 @@ export function heartbeatService( eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.requestedByActorType, "user"), + requestId ? eq(agentWakeupRequests.id, requestId) : undefined, sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, - )).orderBy(asc(agentWakeupRequests.requestedAt)); + )).orderBy(asc(agentWakeupRequests.requestedAt)).limit(50); for (const wake of pending) { + if (wake.idempotencyKey?.startsWith("chat-inbound:")) continue; const payload = parseObject(wake.payload); const context = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]); const commentId = deriveCommentId(context, payload); @@ -9978,19 +9990,76 @@ export function heartbeatService( isNull(issueComments.createdByRunId), gt(issueComments.createdAt, run.finishedAt), )); if (!comment?.body.trim()) continue; - } else if (!await admitExplicitNativeContinuation({ db, companyId: run.companyId, issueId, - agentId: run.agentId, actorType: wake.requestedByActorType, actorId: wake.requestedByActorId, - reason: wake.reason, commentId, successorRunId: randomUUID(), dryRun: true })) 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: wake.requestedByActorId, + reason: wake.reason, commentId, successorRunId: randomUUID(), dryRun: true, + 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: wake.reason, payload, contextSnapshot: context, requestedByActorType: "user", requestedByActorId: wake.requestedByActorId, - idempotencyKey: `remote-stop-comment:${run.id}:${wake.id}` }); + idempotencyKey: `remote-stop-comment:${run.id}:${wake.id}` }, wake.id); break; } } + async function resumeExecutionWaitComments() { + if ((await getSchedulingSuppression()).suppressed) return; + 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`, + lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)), + notInArray(issues.status, ["done", "cancelled"]))) + .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); + const seen = new Set(); + 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 getRun(sourceId); + if (!run || run.companyId !== wake.companyId || run.agentId !== wake.agentId) continue; + await resumeRemoteStopComments(run, wake.id).catch(err => { + logger.warn({ err, runId: run.id }, "failed to resume saved execution-wait message"); + }); + } + } + async function hasUnsafeTextProjectionDatabase() { if (!unsafeTextProjectionPromise) { unsafeTextProjectionPromise = db @@ -18016,31 +18085,35 @@ export function heartbeatService( }); continue; } - const cleared = await db - .update(heartbeatRuns) - .set({ - processPid: null, - processGroupId: null, - processStartedAt: null, - updatedAt: now, - }) - .where( - and( - eq(heartbeatRuns.id, run.id), - eq(heartbeatRuns.runtimeMode, "native"), - run.processPid === null - ? isNull(heartbeatRuns.processPid) - : eq(heartbeatRuns.processPid, run.processPid), - run.processGroupId === null - ? isNull(heartbeatRuns.processGroupId) - : eq(heartbeatRuns.processGroupId, run.processGroupId), - run.processStartedAt === null - ? isNull(heartbeatRuns.processStartedAt) - : eq(heartbeatRuns.processStartedAt, run.processStartedAt), - ), - ) - .returning({ id: heartbeatRuns.id }) - .then((rows) => rows[0] ?? null); + const cleared = await db.transaction(async tx => { + const cleared = await tx + .update(heartbeatRuns) + .set({ + processPid: null, + processGroupId: null, + processStartedAt: null, + updatedAt: now, + }) + .where( + and( + eq(heartbeatRuns.id, run.id), + eq(heartbeatRuns.runtimeMode, "native"), + run.processPid === null + ? isNull(heartbeatRuns.processPid) + : eq(heartbeatRuns.processPid, run.processPid), + run.processGroupId === null + ? isNull(heartbeatRuns.processGroupId) + : eq(heartbeatRuns.processGroupId, run.processGroupId), + run.processStartedAt === null + ? isNull(heartbeatRuns.processStartedAt) + : eq(heartbeatRuns.processStartedAt, run.processStartedAt), + ), + ) + .returning({ id: heartbeatRuns.id }) + .then((rows) => rows[0] ?? null); + if (cleared) await recordNativeLocalProcessStop(tx as unknown as Db, run); + return cleared; + }); if (cleared) claimableNativeRunIds.add(cleared.id); } @@ -18474,6 +18547,7 @@ export function heartbeatService( async function resumeQueuedRuns() { if ((await getSchedulingSuppression()).suppressed) return; + await resumeExecutionWaitComments(); const cutoff = await getWorktreeExecutionCutoff(); const queuedRuns = await db @@ -19106,31 +19180,35 @@ export function heartbeatService( } runningProcesses.delete(run.id); if (run.processPid || run.processGroupId || run.processStartedAt) { - const cleared = await db - .update(heartbeatRuns) - .set({ - processPid: null, - processGroupId: null, - processStartedAt: null, - updatedAt: new Date(), - }) - .where( - and( - eq(heartbeatRuns.id, run.id), - eq(heartbeatRuns.runtimeMode, "native"), - run.processPid === null - ? isNull(heartbeatRuns.processPid) - : eq(heartbeatRuns.processPid, run.processPid), - run.processGroupId === null - ? isNull(heartbeatRuns.processGroupId) - : eq(heartbeatRuns.processGroupId, run.processGroupId), - run.processStartedAt === null - ? isNull(heartbeatRuns.processStartedAt) - : eq(heartbeatRuns.processStartedAt, run.processStartedAt), - ), - ) - .returning() - .then((rows) => rows[0] ?? null); + const cleared = await db.transaction(async tx => { + const cleared = await tx + .update(heartbeatRuns) + .set({ + processPid: null, + processGroupId: null, + processStartedAt: null, + updatedAt: new Date(), + }) + .where( + and( + eq(heartbeatRuns.id, run.id), + eq(heartbeatRuns.runtimeMode, "native"), + run.processPid === null + ? isNull(heartbeatRuns.processPid) + : eq(heartbeatRuns.processPid, run.processPid), + run.processGroupId === null + ? isNull(heartbeatRuns.processGroupId) + : eq(heartbeatRuns.processGroupId, run.processGroupId), + run.processStartedAt === null + ? isNull(heartbeatRuns.processStartedAt) + : eq(heartbeatRuns.processStartedAt, run.processStartedAt), + ), + ) + .returning() + .then((rows) => rows[0] ?? null); + if (cleared) await recordNativeLocalProcessStop(tx as unknown as Db, run); + return cleared; + }); if (!cleared) { const current = await getRun(run.id); if (current) { @@ -24818,7 +24896,7 @@ export function heartbeatService( } } - async function enqueueWakeup(agentId: string, opts: WakeupOptions = {}) { + async function enqueueWakeup(agentId: string, opts: WakeupOptions = {}, executionWaitRequestId?: string) { const source = opts.source ?? "on_demand"; const triggerDetail = opts.triggerDetail ?? null; const contextSnapshot: Record = { @@ -24946,6 +25024,20 @@ export function heartbeatService( patch: Partial = {}, waitCondition?: Record, ) => { + if (executionWaitRequestId) { + await db.update(agentWakeupRequests).set({ + payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', + coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb) || ${JSON.stringify({ + reason: skipReason, message: patch.error ?? (skipReason === "issue_tree_hold_active" + ? "This task is paused. Resume it to send your saved message." + : "Waiting for task execution to be enabled. Your message is saved."), + })}::jsonb)`, + updatedAt: new Date(), + }).where(and(eq(agentWakeupRequests.id, executionWaitRequestId), + eq(agentWakeupRequests.companyId, agent.companyId), eq(agentWakeupRequests.agentId, agentId), + eq(agentWakeupRequests.status, "deferred_issue_execution"))); + return { created: false }; + } const request = { ...durableReceiptFields, companyId: agent.companyId, @@ -25191,7 +25283,7 @@ export function heartbeatService( const invokability = await getAgentInvokability(agent); if (!invokability.invokable) { - if (opts.requestedByActorType !== "user") { + if (opts.requestedByActorType !== "user" || executionWaitRequestId) { await writeSkippedRequest("agent.not_invokable", { error: invokability.message, }, { status: agent.status, reason: invokability.reason }); @@ -25305,6 +25397,20 @@ export function heartbeatService( sql`select id from issues where id = ${issueId} and company_id = ${agent.companyId} for update`, ); + if (executionWaitRequestId) { + const [pending] = await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, executionWaitRequestId), eq(agentWakeupRequests.companyId, agent.companyId), + eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"), + eq(agentWakeupRequests.requestedByActorType, "user"), + eq(agentWakeupRequests.requestedByActorId, opts.requestedByActorId ?? ""), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + )); + // The issue lock serializes cleanup callbacks and periodic workers. + // An adopted, discarded, or edited receipt is no longer authority. + if (!pending || !wakeCommentId || !queuedCommentIdsFromWakePayload(pending.payload).includes(wakeCommentId)) { + return { kind: "deferred" as const }; + } + } let automaticParentRunId: string | null = null; if ( source === "automation" && @@ -25609,10 +25715,18 @@ export function heartbeatService( reconciledSourceRunId = sourceRunId; } + let continuationWait = { reason: "execution_recovery", message: "Waiting for execution recovery. Your message is saved." }; const deferBlockedExecution = async ( executionBlocker: NonNullable>>, ) => { - const condition = { recoveryActionId: executionBlocker.recoveryActionId }; + const condition = { recoveryActionId: executionBlocker.recoveryActionId, ...continuationWait }; + if (executionWaitRequestId) { + await tx.update(agentWakeupRequests).set({ + payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', ${JSON.stringify(condition)}::jsonb)`, + updatedAt: new Date(), + }).where(eq(agentWakeupRequests.id, executionWaitRequestId)); + return { kind: "deferred" as const }; + } if (durableRequest || wakeCommentId || hasInteractionContinuationWakeContext(enrichedContextSnapshot)) { await tx.insert(agentWakeupRequests).values({ ...durableReceiptFields, @@ -25659,6 +25773,7 @@ export function heartbeatService( agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId, reason, commentId: wakeCommentId ?? null, successorRunId: explicitContinuationRunId, dryRun: true, + onBlocked: (reason, message) => { continuationWait = { reason, message }; }, }))) return deferBlockedExecution(executionBlocker); const issueStateGuard = opts.issueStateGuard; @@ -26375,6 +26490,11 @@ export function heartbeatService( tx, ); if (dailyCapBlock) { + if (executionWaitRequestId && executionBlocker) { + continuationWait = { reason: dailyCapBlock.reason, + message: "The agent has reached its daily limit. Your message is saved until work can resume." }; + return deferBlockedExecution(executionBlocker); + } const now = new Date(); await tx.insert(agentWakeupRequests).values({ ...durableReceiptFields, @@ -26469,7 +26589,7 @@ export function heartbeatService( !wake.idempotencyKey?.startsWith("chat-inbound:") && !isInteractionResolutionWakePayload(deferredPayload) && !hasInteractionContinuationWakeContext(deferredContext) && - (deferredContext.wakeReason ?? wake.reason) === "issue_commented" && + ["issue_commented", "issue_reopened_via_comment"].includes(String(deferredContext.wakeReason ?? wake.reason)) && queuedCommentIdsFromWakePayload(wake.payload).length > 0 ); }); @@ -27995,6 +28115,7 @@ export function heartbeatService( releaseEnvironmentLeasesForRun, resumeRemoteStopComments, + resumeExecutionWaitComments, sweepStaleIssueLocks, diff --git a/server/src/services/issue-queued-comment-queue.ts b/server/src/services/issue-queued-comment-queue.ts index 0fb59fc625..398a9c0b25 100644 --- a/server/src/services/issue-queued-comment-queue.ts +++ b/server/src/services/issue-queued-comment-queue.ts @@ -177,6 +177,7 @@ export function buildQueuedCommentQueueSnapshot ({ comment: comment as unknown as IssueComment, position, diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts new file mode 100644 index 0000000000..da8b952252 --- /dev/null +++ b/server/src/services/native-local-process-stop.ts @@ -0,0 +1,60 @@ +import { and, desc, eq, inArray, isNull } from "drizzle-orm"; +import { environmentLeases, heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db"; +import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; + +export const PROCESS_START_REQUESTED = "native.process_start_requested"; +export const PROCESS_IDENTITY_RECORDED = "native.process_identity_recorded"; +const LOCAL_PROCESS_STOPPED = "native.local_process_stopped"; + +function absent(pid: number): boolean { + try { + process.kill(pid, 0); + return false; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "ESRCH"; + } +} + +/** + * Preserve the host's observation before recovery clears the process fields. + * Call in the same transaction as that compare-and-set. Never inspect remote + * process IDs in the control-plane host's process table. + */ +export async function recordNativeLocalProcessStop(db: Db, run: typeof heartbeatRuns.$inferSelect) { + if (run.runtimeMode !== "native" || (!run.processPid && !run.processGroupId)) return false; + const leases = await db.select({ provider: environmentLeases.provider }) + .from(environmentLeases) + .where(and( + eq(environmentLeases.companyId, run.companyId), + eq(environmentLeases.heartbeatRunId, run.id), + )); + if (leases.some(lease => lease.provider !== "local")) return false; + if ((run.processPid && !absent(run.processPid)) || + (run.processGroupId && !absent(-run.processGroupId))) return false; + await appendHeartbeatRunEvent(db, { + companyId: run.companyId, + runId: run.id, + agentId: run.agentId, + eventType: LOCAL_PROCESS_STOPPED, + stream: "system", + level: "info", + message: "Recovery verified the local process stopped before clearing its identity.", + payload: { processPid: run.processPid, processGroupId: run.processGroupId }, + }); + return true; +} + +/** Only server-authored evidence counts. A later launch invalidates the receipt. */ +export async function hasNativeLocalProcessStop(db: Db, companyId: string, runId: string) { + const [event] = await db.select({ eventType: heartbeatRunEvents.eventType }) + .from(heartbeatRunEvents) + .where(and( + eq(heartbeatRunEvents.companyId, companyId), + eq(heartbeatRunEvents.runId, runId), + isNull(heartbeatRunEvents.sourceEventId), + inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, LOCAL_PROCESS_STOPPED]), + )) + .orderBy(desc(heartbeatRunEvents.seq)) + .limit(1); + return event?.eventType === LOCAL_PROCESS_STOPPED; +} diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index aac0377126..30ba177efa 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,3 +1,4 @@ +import { recordNativeLocalProcessStop } from "../native-local-process-stop.js"; import { randomUUID } from "node:crypto"; import { and, desc, eq, inArray, isNull, lte, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -909,6 +910,10 @@ export async function claimNativeRestartRecoveries(input: { } as const; } + if (claimKind !== "reattach_existing_runner") { + await recordNativeLocalProcessStop(tx as unknown as Db, row.run); + } + await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index f5e8f76a7f..cb2d54a132 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -1,3 +1,4 @@ +import { hasNativeLocalProcessStop } from "../native-local-process-stop.js"; import { randomUUID } from "node:crypto"; import { spawn } from "node:child_process"; import { existsSync } from "node:fs"; @@ -580,6 +581,10 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( if (!claim || claim.kind !== "resume_dead_runner") { throw new Error("Expected dead-runner recovery claim"); } + expect(await hasNativeLocalProcessStop(fixture.db, companyId, fixture.runId)).toBe(true); + const [stoppedRun] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(stoppedRun.processPid).toBeNull(); + expect(stoppedRun.processGroupId).toBeNull(); restored = createRunnerdCodexTransport({ ...options, diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 71f5dc5bcb..e52d9e195e 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -4119,7 +4119,7 @@ function leaseDb( returning: () => Promise>; }; result.returning = () => - Promise.resolve([{ runId: coordinator.runId }]); + Promise.resolve([{ runId: coordinator.runId, nextEventSeq: 2 }]); return result; }, }; @@ -4162,12 +4162,20 @@ function leaseDb( return query; }, }); + const insert = (table: unknown) => ({ + values: (values: Record) => { + updates.push({ table, values }); + return { returning: async () => [values] }; + }, + }); const tx = { + insert, execute: async () => [], select, update, }; return { + insert, select, transaction: async (operation: (transaction: Db) => Promise) => operation(tx as unknown as Db), @@ -6463,13 +6471,16 @@ describe("native process ownership", () => { highestContiguousSourceSeq: 1, }; }); - state.createBackend.mockImplementationOnce((_input, options) => ({ - kind: "test", - onSpawn: options.onSpawn, - })); + const updates: Array<{ table: unknown; values: Record }> = []; + state.createBackend.mockImplementationOnce((_input, options) => { + expect(updates).toContainEqual({ table: heartbeatRunEvents, values: expect.objectContaining({ + eventType: "native.process_start_requested", runId: execution.binding.runId, + }) }); + return { kind: "test", onSpawn: options.onSpawn }; + }); await executePaperclipNativeSession({ - db: leaseDb(), + db: leaseDb(execution, {}, {}, updates), execution, runnerInstanceId: "runner", onSpawn, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6f9fb0a1b0..63ecfa3f5d 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { PROCESS_START_REQUESTED } from "../native-local-process-stop.js"; import { remoteLeaseCleanupScope } from "../remote-execution-termination.js"; import { resolveConnectorAssignments, isConnectorSkill } from "../connector-runtime.js"; import { @@ -7598,6 +7599,17 @@ async function executePaperclipNativeSessionWithinScope( input.db, input.execution.binding, ); + // Invalidate prior stop evidence before a backend can spawn. A crash between + // spawn and the PID callback must not make an old receipt authorize a turn. + await appendHeartbeatRunEvent(input.db, { + companyId: input.execution.binding.companyId, + runId: input.execution.binding.runId, + agentId: input.execution.binding.agentId, + eventType: PROCESS_START_REQUESTED, + stream: "system", + level: "info", + message: "Native execution requested; prior local stop evidence no longer applies.", + }); const runnerdBackend = input.useRunnerd && input.backend === undefined ? await createRunnerdBackend({ diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx index 055c61ac66..755a6754b7 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx @@ -78,6 +78,15 @@ describe("TaskChatQueuedMessages", () => { return props; } + it("shows the saved message's wait reason and removes it when admission succeeds", () => { + const message = "Waiting for the previous environment to stop. Your message will start automatically."; + render({ queue: { ...queue, executionWait: { reason: "remote_cleanup", message } } }); + expect(container.querySelector('[role="status"]')?.textContent).toContain(message); + expect(container.textContent).toContain("First queued message"); + render(); + expect(container.textContent).not.toContain(message); + }); + it("renders each queued message once as a compact one-line row", () => { render(); const pane = container.querySelector( diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx index 5d6dcf4453..725ef7ee4c 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx @@ -365,6 +365,11 @@ export function TaskChatQueuedMessages({ data-testid="task-chat-queued-messages" aria-label="Queued messages" > + {queue.executionWait && ( +
+ {queue.executionWait.message} +
+ )} Date: Fri, 11 Sep 2026 15:29:16 -0700 Subject: [PATCH 2/2] fix(test): await issue lock before retry race assertions (#13273) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Retry decisions must respect changes to an issue owner. > - Database tests verify this with two concurrent transactions. > - The test started the competing operation before its fixture held the lock. > - That race can fail a correct source verification run and block deployment. > - This PR waits for lock acquisition before starting the competing operation. ## Linked Issues or Issue Description Refs #13257 for the deployment verification work that exposed this test race. No duplicate fix was found. **What happened?** [Cloud verification job 103424158921](https://github.com/paperclipai/paperclip/actions/runs/34648268409/job/103424158921) failed because the test did not observe a concurrent issue-row lock waiter. The fixture and retry operation both started without an ordering guarantee. **Expected behavior** The fixture must hold the issue lock before the competing retry operation starts. The test must still prove that the retry waits for the lock and observes the reassignment. **Steps to reproduce** 1. Add a temporary 50 ms delay before the fixture acquires the issue lock. 2. Run the promoteOrCancelDueRetry issue-lock test. 3. The original fixture fails with the same missing-waiter error as CI. 4. The synchronized fixture passes with that delay. The delay is not part of this PR. ## What Changed - Separate fixture readiness from its transaction completion promise. - Wait for readiness at both callers before starting the retry decision. - Propagate transaction failure during setup through Promise.race. ## Verification - The original fixture fails under the temporary delayed-lock probe. The fixed fixture passes the same probe. - All 31 tests in server/src/modules/run-dispatch/adapters/postgres.test.ts pass after removing the probe. - The real concurrent waiter, lock-order, and reassignment assertions remain intact. No timeout was increased. - Full local typecheck and build pass (221s and 50s). The full local test command stopped in its server phase after 10,610 passes, 65 skips, and 14 failures: 13 existing macOS skill-cache rename/permission failures and one unchanged Telegram test assertion. The Telegram test passes in a focused rerun. Later local phases did not run after that failure. Current-head Greptile is 5/5 with no findings. All 32 current-head checks pass, including full Linux typecheck, build, native verification, server suites, browser suites, and canary packaging. The unchanged GitHub browser mock assertion passed its single failed-shard retry. - git diff --check passes. ## Risks - Only test synchronization changes. Production database behavior is unchanged. - Awaiting the transaction itself during setup would deadlock the test. Returning the completion promise inside an object avoids that problem. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository tools, and code execution. The exact serving model ID and context window are not exposed by this environment. ## 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 — all 31 database adapter tests pass; full local verification limits are disclosed above - [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 --- .../src/modules/run-dispatch/adapters/postgres.test.ts | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/server/src/modules/run-dispatch/adapters/postgres.test.ts b/server/src/modules/run-dispatch/adapters/postgres.test.ts index 8a4e9a764f..7c50cca50e 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.test.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.test.ts @@ -319,8 +319,10 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => { .set({ assigneeAgentId: newAssigneeAgentId }) .where(eq(issues.id, issueId)); }); - await locked; - return transaction; + // Await lock acquisition before starting the competing operation. Keep + // completion separate so setup does not wait for that operation to finish. + await Promise.race([locked, transaction]); + return { done: transaction }; } describe("evaluateScheduledRetryGate", () => { @@ -572,7 +574,7 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => { contextSnapshot: { issueId, wakeReason: "issue_assigned" }, }); - const holderDone = reassignIssueAndLockRunOnceAConcurrentWaiterBlocks( + const { done: holderDone } = await reassignIssueAndLockRunOnceAConcurrentWaiterBlocks( issueId, runId, replacementAgentId, @@ -733,7 +735,7 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => { // Acquire the issue row lock first and hold it until it observes a // concurrent `for update` waiter — the promote call below — proving // this is a real block, not a race the assertion got lucky on. - const holderDone = reassignIssueAndLockRunOnceAConcurrentWaiterBlocks( + const { done: holderDone } = await reassignIssueAndLockRunOnceAConcurrentWaiterBlocks( issueId, runId, newAgentId,