From 391dd15ff26f43c27f3a3414b876dd4345791560 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 17 Oct 2026 16:53:29 -0700 Subject: [PATCH] fix: recover deferred messages and completed unblock waits (#16673) Recover accepted deferred messages through the latest finished issue run and recognize exact completed agent unblock notifications as passive idle waits. Preserve assignment handoffs, pause and Stop, failed outcomes, and cleanup ownership. Honor local and SSH release contracts and atomically settle cleanup claims. Add PostgreSQL recovery, concurrency, and interruption regressions. Validation: recursive typecheck and build pass; 678 focused tests pass; full CI passes; Apex review is 5/5 with no unresolved findings. Co-Authored-By: Paperclip --- doc/execution-semantics.md | 4 + ...nvironment-runtime-driver-contract.test.ts | 18 ++ .../src/__tests__/environment-runtime.test.ts | 5 +- .../src/__tests__/environment-service.test.ts | 22 +++ .../heartbeat-pending-cleanup-sweep.test.ts | 45 +++++ ...heartbeat-stale-queue-invalidation.test.ts | 31 +++ .../src/__tests__/idle-issue-schedule.test.ts | 63 +++++- .../wake-queue/adapters/postgres.test.ts | 179 ++++++++++++++++++ .../modules/wake-queue/adapters/postgres.ts | 27 ++- .../modules/wake-queue/application/ports.ts | 2 +- .../wake-queue/application/use-cases.ts | 6 +- server/src/services/environment-runtime.ts | 5 +- server/src/services/environments.ts | 11 +- server/src/services/execution-blocker.ts | 29 ++- server/src/services/heartbeat/queue.test.ts | 53 +++++- server/src/services/heartbeat/queue.ts | 67 +++---- server/src/services/heartbeat/run-control.ts | 3 +- server/src/services/idle-issue-schedule.ts | 7 +- server/src/services/idle-work-proof.ts | 31 +++ 19 files changed, 546 insertions(+), 62 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 04f1d2c477d..a9119cabd7b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -89,6 +89,8 @@ Entering `blocked` requires a routable waiting path. An issue may transition int When a structured unblock descriptor is the waiting path, Paperclip immediately notifies the named owner: an agent owner gets a wake, a user or board owner gets an inbox notification. Prose-only blocked — free-text that names an owner or action in a comment without any of the paths above — routes to nobody. It is rejected at the API or auto-classified as `needs_attention` with a board notification, never silently accepted as a healthy waiting state. +An agent-owned unblock request can become a passive wait after its exact notification finishes. Idle sleep requires the current owner, action, blocked transition, and same-company wake receipt to match. A coalesced notification also requires its target run to finish cleanup and accounting. The issue and its unblock action stay unchanged; pending work and ready dependencies still prevent sleep. + Ordinary task questions are current input waits only while they still belong to the current human direction. A newer non-deleted human message on the same task makes an earlier ordinary question historical, without answering, cancelling, @@ -492,6 +494,8 @@ Agent names and structured @-mentions are context, not assignment or wake reques Normal issue-thread feedback can still wake the current assignee according to the comment policy; self-comments and closed-task comments retain their existing suppression rules. Explicit assignment, review, interaction, and blocker-resolution wakes retain their own routing. New mention wake requests are ignored at admission, including calls from legacy producers. Already accepted queue entries retain their existing execution rules: an old entry can combine an assignment or other feedback with a later mention, so its last wake reason does not prove that the entry is mention-only. +Restart recovery revisits accepted deferred messages through the latest issue run, including historical mention queues for another agent. Ordinary messages for a former owner retain their assignment handoff. Native sources require their matching committed finalization, settled environment leases, and released finalizer and copyback ownership before recovery clears old issue claims. A successful retained warm release can satisfy the lease check. Released local bookkeeping without a provider allocation and a matching SSH bookkeeping release need no provider cleanup receipt. Plugin releases record their successful acknowledgement. Pending or failed cleanup and durable owners still hold work. A successful fenced cleanup completion clears its in-flight claim in the same database write, so a restart before retry bookkeeping cannot strand saved input. The normal release checks preserve cleanup holds, pauses, and operator Stops. Recovery checks for a competing turn under the issue lock before it promotes saved input. + Accepted agent feedback must survive a child changing to `done` before its active run exits. Deferred wake promotion may reopen that completed child only for its current assignee, with explicit agent resume intent and live tracked comments from another author. Claim promotion before reopening. Cancelled tasks, deleted comments, self-authored comments, empty continuations, and agent continuations without explicit intent keep their existing suppression rules. Normal pause, ownership, authorization, and budget gates still apply. Pause and tree-control previews should make the same distinction visible. They should report whether the affected subtree contains live running work, queued wakes, agent-owned work, or only human-owned/static issues, so a pause after a handoff does not look like it interrupted agent execution when no agent execution path existed. diff --git a/server/src/__tests__/environment-runtime-driver-contract.test.ts b/server/src/__tests__/environment-runtime-driver-contract.test.ts index f2616aae722..27dc88c7220 100644 --- a/server/src/__tests__/environment-runtime-driver-contract.test.ts +++ b/server/src/__tests__/environment-runtime-driver-contract.test.ts @@ -27,6 +27,8 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { environmentRuntimeService } from "../services/environment-runtime.js"; +import { environmentService } from "../services/environments.js"; +import { hasPendingRunCleanup } from "../services/execution-blocker.js"; import { secretService } from "../services/secrets.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); @@ -184,6 +186,22 @@ describeEmbeddedPostgres("environment runtime driver contract", () => { }; } + it("settles native cleanup through the SSH bookkeeping release contract", async () => { + const { companyId, environment, runId } = await seedEnvironment({ driver: "ssh", config: {} }); + const lease = await environmentService(db).acquireLease({ companyId, environmentId: environment.id, + heartbeatRunId: runId, provider: "ssh", providerLeaseId: "ssh://worker@example.invalid:22/workspace", + metadata: { driver: "ssh" }, + }); + await db.update(heartbeatRuns).set({ status: "succeeded", runtimeMode: "native", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, runId)); + expect(await hasPendingRunCleanup(db, companyId, runId)).toBe(true); + const released = await environmentRuntimeService(db).releaseRunLeases(runId); + expect(released).toHaveLength(1); + expect(released[0]?.lease).toMatchObject({ id: lease.id, status: "released", cleanupStatus: null }); + expect(released[0]?.lease.releasedAt).not.toBeNull(); + expect(await hasPendingRunCleanup(db, companyId, runId)).toBe(false); + }); + async function runContract(testCase: RuntimeContractCase) { const cleanup = await testCase.setup?.(); try { diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index ca6944dc2f2..1385e9167e6 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -6800,6 +6800,7 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { }, }); + await db.delete(environments).where(eq(environments.driver, "local")); await environmentService(db).update(environment.id, { driver: "local", config: {}, @@ -6823,7 +6824,7 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { }), }), }); - expect(released[0]?.lease.status).toBe("released"); + expect(released[0]?.lease).toMatchObject({ status: "released", cleanupStatus: "success" }); }); async function seedPluginDriverEnvironment(pluginId: string) { @@ -7042,7 +7043,7 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { timedOut: false, stdout: "ok\n", }); - expect(destroyed?.status).toBe("failed"); + expect(destroyed).toMatchObject({ status: "failed", cleanupStatus: "success" }); expect(workerManager.call).toHaveBeenCalledWith(pluginId, "environmentResumeLease", { driverKey: "fake-plugin", companyId, diff --git a/server/src/__tests__/environment-service.test.ts b/server/src/__tests__/environment-service.test.ts index 50d10441fae..59774e35ec5 100644 --- a/server/src/__tests__/environment-service.test.ts +++ b/server/src/__tests__/environment-service.test.ts @@ -189,6 +189,28 @@ describeEmbeddedPostgres("environmentService leases", () => { }, ); + it.each(["expired", "retained"] as const)("keeps cleanup completion fenced and atomic for %s leases", async status => { + const { companyId, environmentId, runId } = await seedEnvironment(); + const attemptId = randomUUID(); + const lease = await svc.acquireLease({ companyId, environmentId, heartbeatRunId: runId, + provider: "fake", leasePolicy: "reuse_by_environment" }); + await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", + metadata: { pendingCleanupAttemptId: attemptId, pendingCleanupInFlight: true, + pendingCleanupLeaseExpiresAtMs: Date.now() + 60_000, userNote: "preserve" }, + }).where(eq(environmentLeases.id, lease.id)); + expect(await svc.releaseLease(lease.id, status, { expectedPendingCleanupAttemptId: randomUUID(), + cleanupStatus: "success" })).toBeNull(); + expect(await svc.getLeaseById(lease.id)).toMatchObject({ status: "pending_cleanup", cleanupStatus: "failed", + metadata: { pendingCleanupInFlight: true, pendingCleanupAttemptId: attemptId } }); + expect(await svc.releaseLease(lease.id, "pending_cleanup", { expectedPendingCleanupAttemptId: attemptId, + cleanupStatus: "failed" })).toMatchObject({ metadata: { pendingCleanupInFlight: true } }); + expect(await svc.releaseLease(lease.id, status, { expectedPendingCleanupAttemptId: attemptId, + cleanupStatus: "success" })).toMatchObject({ status, cleanupStatus: "success", metadata: { + pendingCleanupAttemptId: attemptId, pendingCleanupInFlight: false, pendingCleanupLeaseExpiresAtMs: 0, + userNote: "preserve", + } }); + }); + it("acquires and releases a lease for a run", async () => { const { companyId, environmentId, runId } = await seedEnvironment(); diff --git a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts index a118b6945c1..2b5ec8ea3b9 100644 --- a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts +++ b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts @@ -36,6 +36,7 @@ vi.mock("../middleware/logger.js", () => ({ import { logger } from "../middleware/logger.ts"; import { heartbeatService, type HeartbeatEnvironmentRuntime } from "../services/heartbeat.ts"; +import { hasPendingRunCleanup } from "../services/execution-blocker.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -164,6 +165,50 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { return id; } + it("settles the cleanup claim before a later cooldown write fails", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + const agentId = randomUUID(), runId = randomUUID(); + await db.insert(agents).values({ id: agentId, companyId, name: "Cleanup worker", role: "engineer", + status: "idle", adapterType: "process", adapterConfig: {}, runtimeConfig: {}, permissions: {} }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "manual", + status: "succeeded", runtimeMode: "native", finishedAt: new Date() }); + const leaseId = await insertOrphanEphemeralLease({ companyId, environmentId, + updatedAt: new Date(Date.now() - 60 * 60 * 1000) }); + await db.update(environmentLeases).set({ heartbeatRunId: runId }).where(eq(environmentLeases.id, leaseId)); + const teardown = vi.fn(async () => undefined); + const heartbeat = heartbeatService(db, { environmentRuntime: { + retryPendingSandboxTeardown: teardown, + } as unknown as HeartbeatEnvironmentRuntime }); + await db.execute(sql`CREATE FUNCTION test_cleanup_cooldown_failure() RETURNS trigger AS $$ + BEGIN + IF OLD.status = 'expired' AND NEW.metadata ? 'pendingCleanupRetryAfterMs' THEN + RAISE EXCEPTION 'fixture cooldown write failed'; + END IF; + RETURN NEW; + END; + $$ LANGUAGE plpgsql`); + await db.execute(sql`CREATE TRIGGER test_cleanup_cooldown_failure BEFORE UPDATE ON environment_leases + FOR EACH ROW EXECUTE FUNCTION test_cleanup_cooldown_failure()`); + try { + await expect(heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 })).rejects.toMatchObject({ + cause: { message: "fixture cooldown write failed" }, + }); + expect(teardown).toHaveBeenCalledTimes(1); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); + expect(lease).toMatchObject({ status: "expired", cleanupStatus: "success", + metadata: { pendingCleanupInFlight: false } }); + expect(await hasPendingRunCleanup(db, companyId, runId)).toBe(false); + expect(await heartbeatService(db, { environmentRuntime: { + retryPendingSandboxTeardown: teardown, + } as unknown as HeartbeatEnvironmentRuntime }).sweepPendingCleanupLeases({ backoffMs: 0 })) + .toMatchObject({ swept: 0, destroyed: 0 }); + expect(teardown).toHaveBeenCalledTimes(1); + } finally { + await db.execute(sql`DROP TRIGGER test_cleanup_cooldown_failure ON environment_leases`); + await db.execute(sql`DROP FUNCTION test_cleanup_cooldown_failure()`); + } + }); + it("test_pending_cleanup_sweep_retries_and_destroys_lease", async () => { const { companyId, environmentId } = await seedCompanyAndEnvironment(); const leaseId = await insertPendingCleanupLease({ diff --git a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts index a8c7882c71b..e2dc1d6f0f3 100644 --- a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts +++ b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts @@ -216,6 +216,37 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { return { companyId, agentId }; } + it.each(["legacy", "native"])("delivers an accepted non-assignee queue after its %s owner finished", async runtimeMode => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const [recipient] = await db.insert(agents).values({ companyId, name: "Saved feedback recipient", role: "engineer", + status: "idle", adapterType: "codex_local", adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true } }, permissions: {}, + }).returning(); + const [issue] = await db.insert(issues).values({ companyId, title: "Existing accepted feedback", status: "in_progress", assigneeAgentId: agentId }).returning(); + const [owner] = await db.insert(heartbeatRuns).values({ companyId, agentId, invocationSource: "assignment", status: "succeeded", + runtimeMode, nativeIssueId: runtimeMode === "native" ? issue.id : null, startedAt: new Date(), finishedAt: new Date(), contextSnapshot: { issueId: issue.id }, + }).returning(); + if (runtimeMode === "native") await db.insert(nativeRunFinalizations).values({ runId: owner.id, companyId, issueId: issue.id, phase: "committed" }); + const [comment] = await db.insert(issueComments).values({ companyId, issueId: issue.id, body: "Saved feedback for the recipient", + authorAgentId: agentId, createdByRunId: owner.id, + }).returning(); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId: recipient.id, source: "automation", + reason: "issue_execution_deferred", status: "deferred_issue_execution", requestedByActorType: "agent", requestedByActorId: agentId, + payload: withQueuedCommentIdsInWakePayload({ issueId: issue.id, _paperclipWakeContext: { + issueId: issue.id, wakeReason: "issue_comment_mentioned", wakeSource: "automation", + } }, [comment.id]), + }).returning(); + await heartbeat.resumeQueuedRuns(); + const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); + expect(delivered.runId).toBeTruthy(); + await heartbeat.waitForRunExecutionDrain(delivered.runId!); + expect(countExecuteCallsForRun(delivered.runId!)).toBe(1); + expect((await heartbeat.getRun(delivered.runId!))?.contextSnapshot).toMatchObject({ wakeCommentIds: [comment.id] }); + await heartbeat.resumeQueuedRuns(); + expect(countExecuteCallsForRun(delivered.runId!)).toBe(1); + expect((await db.select().from(issueComments).where(eq(issueComments.id, comment.id)))[0]?.body).toBe(comment.body); + expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0]?.assigneeAgentId).toBe(agentId); + }); + it.each(["preparing", "verifying", "resuming"] as const)("keeps assigned work queued through %s, including recovery with timers disabled", async lifecycleState => { const { companyId, agentId } = await seedCompanyAndAgent({ heartbeatConfig: { enabled: false } }); await db.update(agents).set({ status: "paused", lifecycleState }).where(eq(agents.id, agentId)); diff --git a/server/src/__tests__/idle-issue-schedule.test.ts b/server/src/__tests__/idle-issue-schedule.test.ts index 6406a4b9624..70ca3651521 100644 --- a/server/src/__tests__/idle-issue-schedule.test.ts +++ b/server/src/__tests__/idle-issue-schedule.test.ts @@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { ONBOARDING_FIRST_TASK_ORIGIN_KIND, PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared"; -import { agents, agentApiKeys, companies, createDb, heartbeatRuns, issues, issueComments, issueRelations, issueThreadInteractions, routines, routineTriggers } from "@paperclipai/db"; +import { agents, agentApiKeys, agentWakeupRequests, companies, createDb, heartbeatRuns, issues, issueComments, issueRelations, issueThreadInteractions, routines, routineTriggers } from "@paperclipai/db"; import { beginIdleTrackedWork, idleWorkSnapshot } from "../services/task-admission.js"; import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -75,6 +75,67 @@ if (!support.supported) console.warn(`Skipping issue schedule Postgres tests: ${ expect(await readIssueSchedules()).toEqual(present); }); + async function completedUnblockRequest() { + const f = await seed(); + const transition = new Date(now - 5000); + const descriptor = { owner: { agentId: f.agentId }, action: "Provide the missing input" }; + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Waiting for unblock input", + status: "blocked", assigneeAgentId: f.agentId, unblockDescriptor: descriptor, + blockedTransitionAt: transition, blockedOwnerNotifiedAt: new Date(now - 4000), + }).returning(); + await db.update(heartbeatRuns).set({ finishedAt: new Date(now - 1000), contextSnapshot: { issueId: issue.id } }).where(eq(heartbeatRuns.id, f.runId)); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, + source: "automation", reason: "issue_unblock_requested", status: "coalesced", runId: f.runId, + finishedAt: new Date(now - 2000), idempotencyKey: `issue-unblock:${issue.id}:${transition.toISOString()}`, + payload: { issueId: issue.id, action: descriptor.action }, + }).returning(); + return { ...f, issue, wake, descriptor }; + } + + it("allows a blocked task after its exact agent unblock request finishes", async () => { + const f = await completedUnblockRequest(); + expect(await readScheduled()).toEqual(present); + expect(await readIssueSchedules()).toEqual(none); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0]).toMatchObject({ status: "blocked", unblockDescriptor: f.descriptor }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wake.id)))[0]).toMatchObject({ status: "coalesced", runId: f.runId }); + }); + + it("keeps an explicit board-owned unblock request passive", async () => { + const f = await seed(); + await db.insert(issues).values({ companyId: f.companyId, title: "Waiting for the board", status: "blocked", + assigneeAgentId: f.agentId, unblockDescriptor: { owner: "board", action: "Provide input" }, + }); + expect(await readIssueSchedules()).toEqual(none); + }); + + it.each(["missing_receipt", "unnotified", "old_notification", "new_transition", "changed_action", "changed_owner", "foreign_owner", "foreign_receipt", "foreign_run", "wrong_issue", "wrong_run_issue", "missing_run", "unfinished_wake", "cancelled_receipt", "unfinished_run", "retry", "accounting", "unknown_descriptor"])( + "retains a blocked task with %s unblock evidence", async kind => { + const f = await completedUnblockRequest(); + if (kind === "missing_receipt") await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wake.id)); + if (kind === "unnotified") await db.update(issues).set({ blockedOwnerNotifiedAt: null }).where(eq(issues.id, f.issue.id)); + if (kind === "old_notification") await db.update(issues).set({ blockedOwnerNotifiedAt: new Date(now - 6000) }).where(eq(issues.id, f.issue.id)); + if (kind === "new_transition") await db.update(issues).set({ blockedTransitionAt: new Date(now - 3000), blockedOwnerNotifiedAt: new Date(now - 2000) }).where(eq(issues.id, f.issue.id)); + if (kind === "changed_action") await db.update(issues).set({ unblockDescriptor: { ...f.descriptor, action: "Different input" } }).where(eq(issues.id, f.issue.id)); + if (kind === "changed_owner" || kind === "foreign_owner") { + const other = await seed(); + if (kind === "changed_owner") await db.update(agents).set({ companyId: f.companyId }).where(eq(agents.id, other.agentId)); + await db.update(issues).set({ unblockDescriptor: { ...f.descriptor, owner: { agentId: other.agentId } } }).where(eq(issues.id, f.issue.id)); + } + if (kind === "foreign_receipt") await db.update(agentWakeupRequests).set({ companyId: (await seed()).companyId }).where(eq(agentWakeupRequests.id, f.wake.id)); + if (kind === "foreign_run") await db.update(heartbeatRuns).set({ companyId: (await seed()).companyId }).where(eq(heartbeatRuns.id, f.runId)); + if (kind === "wrong_issue") await db.update(agentWakeupRequests).set({ payload: { ...f.wake.payload, issueId: randomUUID() } }).where(eq(agentWakeupRequests.id, f.wake.id)); + if (kind === "wrong_run_issue") await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: randomUUID() } }).where(eq(heartbeatRuns.id, f.runId)); + if (kind === "missing_run") await db.update(agentWakeupRequests).set({ runId: null }).where(eq(agentWakeupRequests.id, f.wake.id)); + if (kind === "unfinished_wake") await db.update(agentWakeupRequests).set({ finishedAt: null }).where(eq(agentWakeupRequests.id, f.wake.id)); + if (kind === "cancelled_receipt") await db.update(agentWakeupRequests).set({ status: "cancelled" }).where(eq(agentWakeupRequests.id, f.wake.id)); + if (kind === "unfinished_run") await db.update(heartbeatRuns).set({ finishedAt: null }).where(eq(heartbeatRuns.id, f.runId)); + if (kind === "retry") await db.update(heartbeatRuns).set({ scheduledRetryAt: new Date(now + 1000) }).where(eq(heartbeatRuns.id, f.runId)); + if (kind === "accounting") await db.update(heartbeatRuns).set({ costAccountingPending: true }).where(eq(heartbeatRuns.id, f.runId)); + if (kind === "unknown_descriptor") await db.execute(sql`UPDATE issues SET unblock_descriptor = unblock_descriptor || '{"unknown":true}'::jsonb WHERE id = ${f.issue.id}`); + expect(await readIssueSchedules()).toEqual(present); + }, + ); + it.each(["typed", "legacy-typed", "interaction", "run", "origin", "agent-state"])( "allows an untouched onboarding task, but retains it after %s activity", async activity => { const f = await seed(); diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 1e13681d345..471b5cf18d0 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -15,6 +15,7 @@ import { issueComments, issueRecoveryActions, issues, + nativeRunFinalizations, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, @@ -65,6 +66,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { await db.delete(environmentLeases); // `heartbeat_runs.wakeup_request_id` references `agent_wakeup_requests.id`, // so the run row must go first. + await db.delete(nativeRunFinalizations); await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); await db.delete(issueRecoveryActions); @@ -175,6 +177,183 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { return id; } + it.each(["queued", "running", "scheduled_retry", "cancelled"])("preserves a stranded queue when another agent has a newer %s turn", async status => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const otherAgentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const runId = await seedRun({ companyId, agentId, status: "succeeded", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + await seedRun({ companyId, agentId: otherAgentId, status, contextSnapshot: { issueId } }); + const wakeId = await seedDeferredWake({ companyId, agentId: otherAgentId, issueId }); + let drained = false; + await createPostgresWakeQueueAdapter(db, stubDeps).withIssueExecutionLock( + { companyId, runId, now: new Date(), resumeStrandedQueue: true }, + async () => { drained = true; return { outcome: { kind: "released" }, postCommitEffects: [] }; }, + ); + expect(drained).toBe(false); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + }); + + it("ignores a foreign company's run when rechecking a stranded queue", async () => { + const companyId = await seedCompany(); + const foreignCompanyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const foreignAgentId = await seedAgent({ companyId: foreignCompanyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const runId = await seedRun({ companyId, agentId, status: "succeeded", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + await seedRun({ companyId: foreignCompanyId, agentId: foreignAgentId, status: "running", contextSnapshot: { issueId } }); + let drained = false; + await createPostgresWakeQueueAdapter(db, stubDeps).withIssueExecutionLock( + { companyId, runId, now: new Date(), resumeStrandedQueue: true }, + async () => { drained = true; return { outcome: { kind: "released" }, postCommitEffects: [] }; }, + ); + expect(drained).toBe(true); + }); + + it.each(["unfinished", "running"])("does not drain a stranded queue through an %s source", async kind => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const runId = await seedRun({ companyId, agentId, status: kind === "running" ? "running" : "succeeded", contextSnapshot: { issueId } }); + if (kind === "running") await db.update(heartbeatRuns).set({ finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + const wakeId = await seedDeferredWake({ companyId, agentId, issueId }); + let drained = false; + await createPostgresWakeQueueAdapter(db, stubDeps).withIssueExecutionLock( + { companyId, runId, now: new Date(), resumeStrandedQueue: true }, + async () => { drained = true; return { outcome: { kind: "released" }, postCommitEffects: [] }; }, + ); + expect(drained).toBe(false); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + }); + + it("recovers a historical mention without consuming a former owner's task messages", async () => { + const companyId = await seedCompany(); + const ownerId = await seedAgent({ companyId }); + const formerOwnerId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, status: "blocked", assigneeAgentId: ownerId }); + const runId = await seedRun({ companyId, agentId: ownerId, status: "succeeded", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + const [message] = await db.insert(issueComments).values({ companyId, issueId, authorType: "user", authorUserId: "responsible-user", body: "Saved task input" }).returning(); + const [mention] = await db.insert(issueComments).values({ companyId, issueId, authorAgentId: ownerId, createdByRunId: runId, body: "Accepted feedback" }).returning(); + const handoffId = await seedDeferredWake({ companyId, agentId: formerOwnerId, issueId, + payload: { _paperclipWakeContext: { issueId, wakeReason: "issue_commented", wakeCommentIds: [message.id] } }, + }); + const mentionId = await seedDeferredWake({ companyId, agentId: formerOwnerId, issueId, + payload: { _paperclipWakeContext: { issueId, wakeReason: "issue_comment_mentioned", wakeCommentIds: [mention.id] } }, + }); + const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(db, stubDeps), + recovery: { escalateStrandedAssignedIssue: async () => undefined, escalateStrandedRecoveryIssueInPlace: async () => undefined }, + }); + const result = await release({ companyId, runId, now: new Date(), suppressImmediateRecovery: true, resumeStrandedQueue: true }); + expect(result.outcome.kind).toBe("promoted"); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, handoffId)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + const [promoted] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, mentionId)); + expect(promoted).toMatchObject({ status: "queued", runId: expect.any(String) }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, promoted.runId!)))[0].contextSnapshot) + .toMatchObject({ wakeCommentIds: [mention.id] }); + expect((await db.select().from(issueComments).where(eq(issueComments.id, message.id)))[0].body).toBe(message.body); + }); + + it.each(["missing", "retryable_failure"])("holds native queue recovery with %s finalization evidence", async phase => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const runId = await seedRun({ companyId, agentId, status: "succeeded", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + if (phase !== "missing") await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase }); + const wakeId = await seedDeferredWake({ companyId, agentId, issueId }); + let drained = false; + await createPostgresWakeQueueAdapter(db, stubDeps).withIssueExecutionLock( + { companyId, runId, now: new Date(), resumeStrandedQueue: true }, + async () => { drained = true; return { outcome: { kind: "released" }, postCommitEffects: [] }; }, + ); + expect(drained).toBe(false); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + }); + + it.each(["active_lease", "pending_cleanup", "failed_cleanup", "unknown_cleanup", "local_provider_resource", "ssh_provider_mismatch", "cleanup_claim", "finalizer_owner", "workspace_owner"])( + "waits for native cleanup and finalizer ownership before recovering saved input (%s)", async kind => { + const companyId = await seedCompany(), agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, status: "blocked", assigneeAgentId: agentId }); + const runId = await seedRun({ companyId, agentId, status: "succeeded", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, finishedAt: new Date(), + ...(kind === "workspace_owner" ? { runnerProfileJson: { nativeWorkspaceFinalizationOwner: { + token: "copyback-owner", hostname: "previous-worker", pid: 123, processStartedAt: new Date(Date.now() - 1000).toISOString(), + } } } : {}), + }).where(eq(heartbeatRuns.id, runId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "committed", + ...(kind === "finalizer_owner" ? { leaseOwner: "finalizer-worker", leaseExpiresAt: new Date(Date.now() - 1000) } : {}), + }); + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId)); + const [lease] = ["finalizer_owner", "workspace_owner"].includes(kind) ? [] : await db.insert(environmentLeases).values({ + companyId, heartbeatRunId: runId, provider: kind === "unknown_cleanup" ? "daytona" : kind === "ssh_provider_mismatch" ? "ssh" : "local", + providerLeaseId: kind === "local_provider_resource" ? randomUUID() : kind === "ssh_provider_mismatch" ? "ssh://worker@example.invalid:22/workspace" : null, + status: kind === "active_lease" ? "active" : kind === "pending_cleanup" ? "pending_cleanup" : "released", + releasedAt: kind === "active_lease" ? null : new Date(), + cleanupStatus: kind === "failed_cleanup" ? "failed" : ["unknown_cleanup", "local_provider_resource", "ssh_provider_mismatch"].includes(kind) ? null : "success", + metadata: kind === "cleanup_claim" ? { pendingCleanupInFlight: true } : {}, + }).returning(); + const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorType: "user", authorUserId: "responsible-user", body: "Saved input after cleanup" }).returning(); + const wakeId = await seedDeferredWake({ companyId, agentId, issueId, requestedByActorId: "responsible-user", + payload: { _paperclipWakeContext: { issueId, wakeReason: "issue_commented", wakeCommentIds: [comment.id] } }, + }); + const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(db, stubDeps), + recovery: { escalateStrandedAssignedIssue: async () => undefined, escalateStrandedRecoveryIssueInPlace: async () => undefined }, + }); + const input = { companyId, runId, now: new Date(), suppressImmediateRecovery: true, resumeStrandedQueue: true }; + const held = await release(input); + expect(held).toMatchObject({ outcome: { kind: "released" }, postCommitEffects: [] }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]) + .toMatchObject({ executionRunId: runId, checkoutRunId: runId }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect(await db.select().from(heartbeatRuns)).toHaveLength(1); + if (lease) await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date(), metadata: {} }).where(eq(environmentLeases.id, lease.id)); + if (kind === "finalizer_owner") await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, runId)); + if (kind === "workspace_owner") await db.update(heartbeatRuns).set({ runnerProfileJson: {} }).where(eq(heartbeatRuns.id, runId)); + expect((await release(input)).outcome.kind).toBe("promoted"); + await release(input); + expect(await db.select().from(heartbeatRuns)).toHaveLength(2); + expect((await db.select().from(issueComments).where(eq(issueComments.id, comment.id)))[0].body).toBe(comment.body); + }, + ); + + it.each(["released", "retained", "local_release", "ssh_release", "unrelated_cleanup"])("allows native queue recovery after settled %s cleanup", async kind => { + const companyId = await seedCompany(), agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, status: "blocked", assigneeAgentId: agentId }); + const runId = await seedRun({ companyId, agentId, status: "succeeded", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "committed" }); + if (kind === "unrelated_cleanup") { + const foreignCompany = await seedCompany(), foreignAgent = await seedAgent({ companyId: foreignCompany }); + const foreignRun = await seedRun({ companyId: foreignCompany, agentId: foreignAgent, status: "running" }); + await db.insert(environmentLeases).values({ companyId: foreignCompany, heartbeatRunId: foreignRun, provider: "local", status: "pending_cleanup", cleanupStatus: "failed" }); + } else { + await db.insert(environmentLeases).values({ companyId, heartbeatRunId: runId, + provider: kind === "local_release" ? "local" : kind === "ssh_release" ? "ssh" : "daytona", + providerLeaseId: kind === "ssh_release" ? "ssh://worker@example.invalid:22/workspace" : null, + metadata: kind === "ssh_release" ? { driver: "ssh" } : {}, + status: ["local_release", "ssh_release"].includes(kind) ? "released" : kind, + leasePolicy: kind === "retained" ? "reuse_by_environment" : "ephemeral", cleanupStatus: ["local_release", "ssh_release"].includes(kind) ? null : "success", + releasedAt: kind === "retained" ? null : new Date(), + }); + } + const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorType: "user", authorUserId: "responsible-user", body: "Saved input for the next turn" }).returning(); + await seedDeferredWake({ companyId, agentId, issueId, requestedByActorId: "responsible-user", + payload: { _paperclipWakeContext: { issueId, wakeReason: "issue_commented", wakeCommentIds: [comment.id] } }, + }); + const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(db, stubDeps), + recovery: { escalateStrandedAssignedIssue: async () => undefined, escalateStrandedRecoveryIssueInPlace: async () => undefined }, + }); + expect((await release({ companyId, runId, now: new Date(), suppressImmediateRecovery: true, resumeStrandedQueue: true })).outcome.kind).toBe("promoted"); + }); + it.each([ "completed", "multiple_comments", "repeated_reference", "parent_reference", "mixed_issue_references", "mixed_foreign_references", "mixed_unknown_references", "no_comments", "missing_comment", "human_comment", "other_run_comment", "foreign_comment", "other_issue_comment", "deleted_comment", diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 5d7a7b9eade..3767c09740a 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -3,7 +3,7 @@ import { readRunCancellation } from "../../../services/run-cancellation.js"; import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js"; import { instanceSettingsService } from "../../../services/instance-settings.js"; import { currentConversationCommentCondition } from "../../../services/agent-conversations.js"; -import { getExecutionBlocker } from "../../../services/execution-blocker.js"; +import { getExecutionBlocker, hasPendingRunCleanup } from "../../../services/execution-blocker.js"; import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { extractIssueReferenceIdentifiers, isUuidLike } from "@paperclipai/shared"; @@ -1051,6 +1051,9 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd throw new Error(`wake-queue: run ${input.runId} was not found while releasing issue execution`); } const runSnapshot = toRunSnapshot(run); + if (input.resumeStrandedQueue && (!run.finishedAt || !["succeeded", "failed", "cancelled", "timed_out", "interrupted"].includes(run.status))) { + return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; + } const contextIssueId = readNonEmptyString(parseObject(run.contextSnapshot).issueId); // Lock the context issue (if any) and every issue that still references this @@ -1078,6 +1081,19 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd `, ); + if (input.resumeStrandedQueue && run.runtimeMode === "native") { + const [finalized] = contextIssueId && run.nativeIssueId === contextIssueId + ? await tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, run.id), + eq(nativeRunFinalizations.issueId, run.nativeIssueId), eq(nativeRunFinalizations.phase, "committed"), + )).limit(1) + : []; + if (!finalized) return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; + } + if (input.resumeStrandedQueue && await hasPendingRunCleanup(tx, input.companyId, run.id)) { + return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; + } + const candidateIssues = await tx .select() .from(issues) @@ -1159,11 +1175,14 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd // agent's review participation retains its separate recovery path. const [successor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, input.companyId), - eq(heartbeatRuns.agentId, run.agentId), - sql`${heartbeatRuns.id} <> ${run.id}`, + input.resumeStrandedQueue ? undefined : eq(heartbeatRuns.agentId, run.agentId), + input.resumeStrandedQueue ? undefined : sql`${heartbeatRuns.id} <> ${run.id}`, or(eq(heartbeatRuns.nativeIssueId, issueRow.id), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueRow.id}`), - inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), + or(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), + input.resumeStrandedQueue ? sql`(${heartbeatRuns.createdAt}, ${heartbeatRuns.id}) > ( + SELECT source.created_at, source.id FROM heartbeat_runs source + WHERE source.company_id = ${input.companyId} AND source.id = ${run.id})` : undefined), )).limit(1); if (successor) return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index ff3430ff496..6180288d797 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -235,7 +235,7 @@ export interface WakeQueueTransaction { */ export interface IssueLockWriter { withIssueExecutionLock( - input: { companyId: string; runId: string; now: Date }, + input: { companyId: string; runId: string; now: Date; resumeStrandedQueue?: boolean }, fn: ( locked: LockedIssueExecution, ports: { host: WakeQueueHost; transaction: WakeQueueTransaction }, diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index 0cbceec1b63..afe739e327f 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -120,6 +120,7 @@ export type ReleaseIssueExecutionInput = { runId: string; now: Date; suppressImmediateRecovery?: boolean; + resumeStrandedQueue?: boolean; }; type PauseHoldFacts = Awaited>; @@ -200,7 +201,7 @@ async function runReleaseDrain( !candidate.preservesIndependentContinuation && candidate.queuedCommentIds.length > 0 && ["issue_commented", "issue_reopened_via_comment"].includes(candidate.wakeReason ?? candidate.reason ?? ""); if (ordinaryTaskComment && candidate.agentId !== issue.assigneeAgentId) { - if (run.agentId !== issue.assigneeAgentId) { + if (input.resumeStrandedQueue || run.agentId !== issue.assigneeAgentId) { // The old owner can release before assignment admission adopts these // exact IDs. Leave its receipt intact, skip it for this drain, and let // a current-assignee wake behind it proceed. @@ -968,7 +969,8 @@ export function createReleaseIssueExecution(deps: { input: ReleaseIssueExecutionInput, ): Promise<{ outcome: ReleaseOutcome; postCommitEffects: PostCommitEffect[] }> { const result = await deps.issueLock.withIssueExecutionLock( - { companyId: input.companyId, runId: input.runId, now: input.now }, + { companyId: input.companyId, runId: input.runId, now: input.now, + ...(input.resumeStrandedQueue ? { resumeStrandedQueue: true } : {}) }, (locked, ports) => runReleaseDrain(locked, ports, input), ); diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index a95d58cec6e..a7c2f0098af 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -3689,7 +3689,7 @@ function createPluginEnvironmentDriver( providerLeaseId: input.lease.providerLeaseId, leaseMetadata: input.lease.metadata ?? undefined, }); - return await environmentsSvc.releaseLease(input.lease.id, input.status); + return await environmentsSvc.releaseLease(input.lease.id, input.status, { cleanupStatus: "success" }); }, async resumeRunLease(input) { @@ -3737,6 +3737,9 @@ function createPluginEnvironmentDriver( }); return await environmentsSvc.releaseLease(input.lease.id, "failed", { failureReason: input.failureReason ?? "lease_destroyed", + cleanupStatus: "success", + ...(input.lease.status === "pending_cleanup" && typeof input.lease.metadata?.pendingCleanupAttemptId === "string" + ? { expectedPendingCleanupAttemptId: input.lease.metadata.pendingCleanupAttemptId } : {}), }); }, diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index 0ac6c69fcbc..ab5e80764eb 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -1653,6 +1653,11 @@ export function environmentService(db: Db) { }, ) => { const now = new Date(); + const releaseMetadata = options?.remoteExecutionTermination + ? sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ remoteExecutionTermination: options.remoteExecutionTermination })}::jsonb` + : sql`${environmentLeases.metadata} - 'remoteExecutionTermination'`; + const cleanupSettled = options?.expectedPendingCleanupAttemptId && options.cleanupStatus === "success" + && status !== "pending_cleanup"; const row = await db .update(environmentLeases) .set({ @@ -1664,9 +1669,9 @@ export function environmentService(db: Db) { ...(options?.cleanupStatus !== undefined ? { cleanupStatus: options.cleanupStatus } : {}), // A later release without a receipt cannot reuse an earlier stop's // authority (for example after a same-run lease resume). - metadata: options?.remoteExecutionTermination - ? sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ remoteExecutionTermination: options.remoteExecutionTermination })}::jsonb` - : sql`${environmentLeases.metadata} - 'remoteExecutionTermination'`, + metadata: cleanupSettled + ? sql`coalesce(${releaseMetadata}, '{}'::jsonb) || '{"pendingCleanupInFlight":false,"pendingCleanupLeaseExpiresAtMs":0}'::jsonb` + : releaseMetadata, }) .where(and(eq(environmentLeases.id, id), options?.expectedPendingCleanupAttemptId ? and(eq(environmentLeases.status, "pending_cleanup"), diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index b451fd41e30..38d615dfbd5 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -2,13 +2,40 @@ import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js"; import { and, desc, eq, gt, inArray, not, or, sql } from "drizzle-orm"; import { conversationRecoveryActionPredicate, getConversationOwnershipBlocker } from "./conversation-continuation.js"; import { z } from "zod"; -import { agentWakeupRequests, chatConversations, chatEndpoints, heartbeatRuns, issueComments, issues, issueRecoveryActions, toolConnections, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, chatConversations, chatEndpoints, environmentLeases, heartbeatRuns, issueComments, issues, issueRecoveryActions, nativeRunFinalizations, toolConnections, type Db } from "@paperclipai/db"; import { canContinueCancelledRun, readRunCancellation } from "./run-cancellation.js"; import { canRetryStoppedRun } from "./cancelled-native-startup.js"; import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; import { EXECUTION_RECONCILIATION_CAUSES, type ExecutionBlocker } from "@paperclipai/shared"; import { hasRequiredWorkspaceRecovery, LEGACY_WORKSPACE_RECOVERY_SCHEMA, nativeRunHasNoWorkspaceAccess } from "./workspace-restore-recovery-state.js"; +export async function hasPendingRunCleanup(db: Db, companyId: string, runId: string): Promise { + const [state] = await db.execute<{ pending: boolean }>(sql`SELECT + ${heartbeatRuns.runnerProfileJson}->>'nativeWorkspaceFinalizationOwner' IS NOT NULL + OR EXISTS (SELECT 1 FROM ${environmentLeases} + WHERE ${environmentLeases.companyId} = ${companyId} AND ${environmentLeases.heartbeatRunId} = ${runId} + AND (( ${environmentLeases.releasedAt} IS NULL AND NOT COALESCE( + ${environmentLeases.status} = 'retained' AND ${environmentLeases.leasePolicy} = 'reuse_by_environment' + AND ${environmentLeases.cleanupStatus} = 'success', false)) + OR ${environmentLeases.status} IN ('active', 'pending_cleanup') OR ${environmentLeases.cleanupStatus} = 'failed' + OR COALESCE(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') <> 'false' + OR (${heartbeatRuns.runtimeMode} = 'native' AND NOT COALESCE( + (${environmentLeases.status} IN ('released', 'expired', 'failed') + AND ${environmentLeases.releasedAt} IS NOT NULL AND (${environmentLeases.cleanupStatus} = 'success' + OR (${environmentLeases.cleanupStatus} IS NULL AND ( + (${environmentLeases.provider} = 'local' AND ${environmentLeases.providerLeaseId} IS NULL) + OR (${environmentLeases.provider} = 'ssh' AND ${environmentLeases.metadata}->>'driver' = 'ssh' + AND ${environmentLeases.providerLeaseId} LIKE 'ssh://%'))))) + OR (${environmentLeases.status} = 'retained' AND ${environmentLeases.leasePolicy} = 'reuse_by_environment' + AND ${environmentLeases.cleanupStatus} = 'success'), false)))) + OR EXISTS (SELECT 1 FROM ${nativeRunFinalizations} + WHERE ${nativeRunFinalizations.companyId} = ${companyId} AND ${nativeRunFinalizations.runId} = ${runId} + AND (${nativeRunFinalizations.leaseOwner} IS NOT NULL + OR ${nativeRunFinalizations.phase} NOT IN ('committed', 'applied', 'terminal_failure'))) AS pending + FROM ${heartbeatRuns} WHERE ${heartbeatRuns.companyId} = ${companyId} AND ${heartbeatRuns.id} = ${runId} LIMIT 1`); + return state?.pending !== false; +} + /** Resolved recovery bookkeeping can still carry an effective no-replay hold. */ export function executionBlockerPredicate() { return and( diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index 3e676e5af0a..00186fd2299 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -1,6 +1,6 @@ import { claimNativeSessionResumptions } from "../native-runtime/native-finalization-reconciler.js"; import type { Db } from "@paperclipai/db"; -import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueComments, issues, nativeRunFinalizations } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -151,6 +151,57 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { return { company, agent, issue, wake, run }; } + async function strandedMention(runtimeMode = "legacy") { + const seeded = await fixture("succeeded"); + await db.update(heartbeatRuns).set({ runtimeMode, nativeIssueId: runtimeMode === "native" ? seeded.issue.id : null, finishedAt: new Date() }).where(eq(heartbeatRuns.id, seeded.run.id)); + if (runtimeMode === "native") await db.insert(nativeRunFinalizations).values({ runId: seeded.run.id, companyId: seeded.company.id, issueId: seeded.issue.id, phase: "committed" }); + await db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, seeded.issue.id)); + const [recipient] = await db.insert(agents).values({ companyId: seeded.company.id, name: "Mention recipient", role: "engineer", status: "idle", adapterType: "process", adapterConfig: {} }).returning(); + const [comment] = await db.insert(issueComments).values({ companyId: seeded.company.id, issueId: seeded.issue.id, body: "Saved feedback", authorAgentId: seeded.agent.id }).returning(); + const [pending] = await db.insert(agentWakeupRequests).values({ companyId: seeded.company.id, agentId: recipient.id, + source: "automation", reason: "issue_execution_deferred", status: "deferred_issue_execution", + payload: { issueId: seeded.issue.id, _paperclipWakeContext: { issueId: seeded.issue.id, wakeReason: "issue_comment_mentioned", wakeCommentIds: [comment.id] } }, + }).returning(); + return { ...seeded, recipient, pending }; + } + + it.each(["legacy", "native"])("resumes an accepted non-assignee queue through its %s owner", async runtimeMode => { + const { run, pending } = await strandedMention(runtimeMode); + const deps = callbacks(db); + await createHeartbeatQueue(db, deps).resumeQueuedRuns(); + expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ id: run.id }), { suppressImmediateRecovery: true, resumeStrandedQueue: true }, + ); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, pending.id)))[0]).toMatchObject({ + status: "deferred_issue_execution", runId: null, payload: pending.payload, + }); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + + it("preserves a former owner's ordinary messages for the assignment handoff", async () => { + const { pending } = await strandedMention(); + const payload = { ...pending.payload, _paperclipWakeContext: { + ...(pending.payload!._paperclipWakeContext as Record), wakeReason: "issue_commented", + } }; + await db.update(agentWakeupRequests).set({ payload }).where(eq(agentWakeupRequests.id, pending.id)); + const deps = callbacks(db); + await createHeartbeatQueue(db, deps).resumeQueuedRuns(); + expect(deps.releaseIssueExecutionAndPromote).not.toHaveBeenCalled(); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, pending.id)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null, payload }); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + + it.each(["queued", "running", "scheduled_retry"])("does not replay a finishing owner while another agent has a %s turn", async status => { + const { company, issue, recipient } = await strandedMention(); + await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: recipient.id, status, invocationSource: "assignment", contextSnapshot: { issueId: issue.id } }); + const deps = callbacks(db); + deps.getAgent.mockResolvedValue({ ...recipient, status: "paused" }); + await createHeartbeatQueue(db, deps).resumeQueuedRuns(); + expect(deps.releaseIssueExecutionAndPromote).not.toHaveBeenCalled(); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + it.each(["claimed", "cancelled"] as const)("releases a running claim while preserving a %s wake", async (wakeStatus) => { const { run, issue, wake } = await fixture(); await db.update(agentWakeupRequests).set({ status: wakeStatus }).where(eq(agentWakeupRequests.id, wake.id)); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 6a61871516b..96bfefaf901 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -30,7 +30,7 @@ import { undeliveredLegacyUserCommentIds, } from "../explicit-native-continuation.js"; import { canRetryStoppedRun } from "../cancelled-native-startup.js"; -import { getExecutionBlocker } from "../execution-blocker.js"; +import { getExecutionBlocker, hasPendingRunCleanup } from "../execution-blocker.js"; import { isConversationAdapter } from "../conversation-continuation.js"; import { recordExecutionWait } from "../execution-wait.js"; import { @@ -77,7 +77,6 @@ import { chatEndpoints, companies, costEvents, - environmentLeases, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, @@ -85,7 +84,6 @@ import { issueRecoveryActions, issueRelations, issues, - nativeRunFinalizations, } from "@paperclipai/db"; import { conflict, @@ -198,7 +196,7 @@ export interface HeartbeatQueueDependencies extends Pick Promise<{ suppressed: boolean; reason: "worktree_instance" | "database_restore_in_progress" | "task_drain" | null }>; canStartNewExternalBotWork: (agent: typeof agents.$inferSelect) => Promise; executeRun: (runId: string) => Promise; - releaseIssueExecutionAndPromote: (run: Pick, options?: { suppressImmediateRecovery?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] }) => Promise; + releaseIssueExecutionAndPromote: (run: Pick, options?: { suppressImmediateRecovery?: boolean; resumeStrandedQueue?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] }) => Promise; applyWakeQueuePostCommitEffects: (effects: WakeQueuePostCommitEffect[]) => Promise; publishRunLifecyclePluginEvent: (run: HeartbeatRun) => void; cancelRunInternal: (runId: string, reason?: string, options?: { errorCode?: string }) => Promise; @@ -1171,27 +1169,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende liveRunExecutions.has(previousRunId)) { return { ownsIssue, blocked: true }; } - const [pendingLease] = await tx.select({ id: environmentLeases.id }) - .from(environmentLeases).where(and( - eq(environmentLeases.companyId, run.companyId), - eq(environmentLeases.heartbeatRunId, previousRunId), - or(and(isNull(environmentLeases.releasedAt), - // Warm release deliberately retains the sandbox. Its successful - // receipt settles the old run without destroying the resource. - sql`not coalesce(${environmentLeases.status} = 'retained' - and ${environmentLeases.leasePolicy} = 'reuse_by_environment' - and ${environmentLeases.cleanupStatus} = 'success', false)`), - eq(environmentLeases.status, "pending_cleanup"), - eq(environmentLeases.cleanupStatus, "failed")), - )).limit(1); - const [finalization] = await tx.select({ - phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner, - }).from(nativeRunFinalizations).where(and( - eq(nativeRunFinalizations.companyId, run.companyId), - eq(nativeRunFinalizations.runId, previousRunId), - )); - if (pendingLease || (finalization && (finalization.leaseOwner || - !["committed", "applied", "terminal_failure"].includes(finalization.phase)))) { + if (await hasPendingRunCleanup(tx, run.companyId, previousRunId)) { return { ownsIssue, blocked: true }; } } @@ -1776,7 +1754,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende .from(agentWakeupRequests) .innerJoin(issues, and(eq(issues.companyId, agentWakeupRequests.companyId), sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`, - eq(issues.assigneeAgentId, agentWakeupRequests.agentId))) + or(eq(issues.assigneeAgentId, agentWakeupRequests.agentId), + sql`${agentWakeupRequests.payload} #>> '{_paperclipWakeContext,wakeReason}' = 'issue_comment_mentioned'`))) .innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active"))) .where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(issues.executionRunId), @@ -1785,35 +1764,37 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende sql`${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}' <> '[]'::jsonb`, ), sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`), sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is null`, + sql`NOT EXISTS (SELECT 1 FROM heartbeat_runs live_issue_run + WHERE live_issue_run.company_id = ${issues.companyId} + AND (live_issue_run.context_snapshot->>'issueId' = ${issues.id}::text + OR live_issue_run.native_issue_id = ${issues.id}) + AND live_issue_run.status IN ('queued', 'running', 'scheduled_retry'))`, cutoff ? gte(agentWakeupRequests.requestedAt, cutoff) : undefined)) .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); for (const { wake } of strandedQueues) { if (!queuedCommentIdsFromWakePayload(wake.payload).length && !await readQueuedInteractionResponse(db, wake.companyId, String(wake.payload?.issueId), wake.payload)) continue; const [latest] = await db.select().from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, wake.companyId), eq(heartbeatRuns.agentId, wake.agentId), + eq(heartbeatRuns.companyId, wake.companyId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${String(wake.payload?.issueId)}`, + or(isNull(heartbeatRuns.nativeIssueId), sql`${heartbeatRuns.nativeIssueId}::text = ${String(wake.payload?.issueId)}`), )).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(and( eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.status, "deferred_issue_execution"), )); - if (!latest || !isHeartbeatRunTerminalStatus(latest.status)) continue; - if (latest.runtimeMode === "native" && typeof wake.payload?.dotAssignmentFollowUp === "string") { - await releaseIssueExecutionAndPromote(latest, { suppressImmediateRecovery: true }).catch(err => { - logger.warn({ err, queueId: wake.id }, "failed to promote unread Dot comment after recovery"); - }); - continue; + if (!latest || !latest.finishedAt || !isHeartbeatRunTerminalStatus(latest.status)) continue; + if (latest.runtimeMode !== "legacy" && latest.runtimeMode !== "native") continue; + if (latest.runtimeMode === "legacy") { + const cancelledAdmission = latest.status === "cancelled" && !latest.startedAt && + latest.errorCode === "execution_reconciliation_required"; + if ((latest.status !== "cancelled" || cancelledAdmission) && await getExecutionBlocker(db, wake.companyId, String(wake.payload?.issueId))) { + await resumeSavedLegacyComments(wake.companyId, wake.id).catch(err => { + logger.warn({ err, queueId: wake.id }, "failed to deliver saved legacy comment after recovery stopped"); + }); + } } - if (latest.runtimeMode !== "legacy") continue; - const cancelledAdmission = latest.status === "cancelled" && !latest.startedAt && - latest.errorCode === "execution_reconciliation_required"; - if ((latest.status !== "cancelled" || cancelledAdmission) && await getExecutionBlocker(db, wake.companyId, String(wake.payload?.issueId))) { - await resumeSavedLegacyComments(wake.companyId, wake.id).catch(err => { - logger.warn({ err, queueId: wake.id }, "failed to deliver saved legacy comment after recovery stopped"); - }); - } - await releaseIssueExecutionAndPromote(latest, { suppressImmediateRecovery: true }).catch(err => { - logger.warn({ err, queueId: wake.id }, "failed to promote stranded legacy comments"); + await releaseIssueExecutionAndPromote(latest, { suppressImmediateRecovery: true, resumeStrandedQueue: true }).catch(err => { + logger.warn({ err, queueId: wake.id }, "failed to promote stranded comments"); }); } diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index f4b0529ddc9..7f4a8bf8a24 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -687,7 +687,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont async function releaseIssueExecutionAndPromote( run: Pick, - options: { suppressImmediateRecovery?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] } = {}, + options: { suppressImmediateRecovery?: boolean; resumeStrandedQueue?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] } = {}, ) { try { const source = await getRun(run.id); @@ -698,6 +698,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont // A durable authentication card owns recovery. This covers the review path, // while continuation classification blocks periodic generic retries. suppressImmediateRecovery: options.suppressImmediateRecovery || isAiAuthenticationBlocked(source), + ...(options.resumeStrandedQueue ? { resumeStrandedQueue: true } : {}), }); if (options.deferredPostCommitEffects) { options.deferredPostCommitEffects.push(...postCommitEffects); diff --git a/server/src/services/idle-issue-schedule.ts b/server/src/services/idle-issue-schedule.ts index e3944d75a0c..c234403e873 100644 --- a/server/src/services/idle-issue-schedule.ts +++ b/server/src/services/idle-issue-schedule.ts @@ -4,7 +4,7 @@ import type { Db } from "@paperclipai/db"; import { AGENT_STATUSES, issueExecutionMonitorPolicySchema, PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared"; import type { ScheduledWake } from "./idle-scheduled-work.js"; import { onboardingFirstTaskAwaitingUserSql } from "./recovery/onboarding-first-task-wait.js"; -import { responseDrivenIssueWait } from "./idle-work-proof.js"; +import { completedAgentUnblockRequest, responseDrivenIssueWait } from "./idle-work-proof.js"; const processGeneration = randomUUID(); const limit = 10_000; @@ -28,6 +28,7 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st company_status: string; company_updated: Date | string; agent_status: string | null; agent_updated: Date | string | null; scoped: boolean; unsafe_dependencies: boolean; dependency_ready: boolean; untouched_onboarding: boolean; response_driven_wait: boolean; + unblock_request_finished: boolean; response_wait_generation: unknown; }>(sql`SELECT i.id, i.company_id, i.status, i.updated_at, i.assignee_agent_id, i.assignee_user_id, i.checkout_run_id, i.execution_run_id, i.execution_policy, i.execution_state, i.unblock_descriptor, @@ -36,6 +37,7 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st a.status AS agent_status, a.updated_at AS agent_updated, (i.assignee_agent_id IS NULL OR a.id IS NOT NULL) AS scoped, (${sql.raw(responseDrivenIssueWait('i'))}) AS response_driven_wait, + (${sql.raw(completedAgentUnblockRequest('i'))}) AS unblock_request_finished, (SELECT jsonb_agg(jsonb_build_array(w.id, w.updated_at, w.kind, w.status, w.source_run_id, w.effective_resolver_policy, w.continuation_policy, w.addressee_agent_id, w.addressee_user_id) ORDER BY w.id) FROM issue_thread_interactions w WHERE w.issue_id = i.id AND w.status = 'pending') AS response_wait_generation, @@ -82,7 +84,8 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st const passive = row.status === "backlog" || (row.status === "todo" && !row.assignee_agent_id) || (row.status === "todo" && row.assignee_agent_id && row.agent_status === "idle" && row.untouched_onboarding === true) - || (row.status === "blocked" && (row.unblock_descriptor === null || unblock?.owner === "board")) + || (row.status === "blocked" && (row.unblock_descriptor === null || unblock?.owner === "board" + || row.unblock_request_finished === true)) || (["in_progress", "in_review"].includes(row.status) && row.response_driven_wait === true) || Boolean(row.assignee_user_id && !row.assignee_agent_id); if (!passive) return { blocked: true }; diff --git a/server/src/services/idle-work-proof.ts b/server/src/services/idle-work-proof.ts index 38ef426696d..cff70af709f 100644 --- a/server/src/services/idle-work-proof.ts +++ b/server/src/services/idle-work-proof.ts @@ -15,6 +15,37 @@ export function finishedIdleWake(alias: string): string { AND ${finishedIdleRun('wake_target')})))`; } +export function completedAgentUnblockRequest(issue: string): string { + return `${issue}.blocked_transition_at IS NOT NULL + AND ${issue}.blocked_owner_notified_at >= ${issue}.blocked_transition_at + AND jsonb_typeof(${issue}.unblock_descriptor) = 'object' + AND ${issue}.unblock_descriptor = jsonb_build_object('owner', ${issue}.unblock_descriptor->'owner', + 'action', ${issue}.unblock_descriptor->'action') + AND jsonb_typeof(${issue}.unblock_descriptor->'owner') = 'object' + AND ${issue}.unblock_descriptor->'owner' = jsonb_build_object('agentId', ${issue}.unblock_descriptor->'owner'->'agentId') + AND jsonb_typeof(${issue}.unblock_descriptor->'owner'->'agentId') = 'string' + AND jsonb_typeof(${issue}.unblock_descriptor->'action') = 'string' + AND length(trim(${issue}.unblock_descriptor->>'action')) > 0 + AND EXISTS (SELECT 1 FROM agents unblock_owner + WHERE unblock_owner.company_id = ${issue}.company_id + AND unblock_owner.id::text = ${issue}.unblock_descriptor->'owner'->>'agentId') + AND EXISTS (SELECT 1 FROM agent_wakeup_requests unblock_wake + WHERE unblock_wake.company_id = ${issue}.company_id + AND unblock_wake.agent_id::text = ${issue}.unblock_descriptor->'owner'->>'agentId' + AND unblock_wake.idempotency_key = 'issue-unblock:' || ${issue}.id::text || ':' + || to_char(${issue}.blocked_transition_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"') + AND unblock_wake.payload->>'issueId' = ${issue}.id::text + AND unblock_wake.payload->>'action' = ${issue}.unblock_descriptor->>'action' + AND unblock_wake.status IN ('completed', 'coalesced') AND unblock_wake.finished_at IS NOT NULL + AND EXISTS (SELECT 1 FROM heartbeat_runs unblock_target + WHERE unblock_target.id = unblock_wake.run_id AND unblock_target.company_id = ${issue}.company_id + AND unblock_target.agent_id = unblock_wake.agent_id + AND (unblock_target.context_snapshot->>'issueId' = ${issue}.id::text + OR unblock_target.native_issue_id = ${issue}.id) + AND (unblock_target.native_issue_id IS NULL OR unblock_target.native_issue_id = ${issue}.id) + AND ${finishedIdleRun('unblock_target')}))`; +} + export function connectionDeliveryWaitingForResponse(delivery: string): string { return `EXISTS (SELECT 1 FROM issue_thread_interactions waiting_delivery JOIN issues waiting_issue ON waiting_issue.id = waiting_delivery.issue_id