diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 45d785b33eb..a72120c36e6 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -202,6 +202,14 @@ and review run. The server reads these records from saved review state; it does not depend on the parent session remembering a separate review session. These records are evidence and do not grant permission to resolve another review. +An automatic chat-completion wake retains its exact delivery receipt when +admission waits on a dependency, budget, or pause. It does not merge into a +replaceable dependency signal. The same gates still prevent execution. +Workspace readiness accepts a completed operation enclosed by a successful +finalize from the same company, workspace, issue, and run. A nested snapshot +that finishes before that finalize finishes does not leave a permanent wait. +Unfinished operations and work after the finalize still require a later save. + ### Answered Slack conversations An accepted scheduled-monitor yield remains unfinished task work. Its committed diff --git a/server/src/__tests__/chat-completion-delivery.test.ts b/server/src/__tests__/chat-completion-delivery.test.ts index 468a7a86f80..2dfd4290972 100644 --- a/server/src/__tests__/chat-completion-delivery.test.ts +++ b/server/src/__tests__/chat-completion-delivery.test.ts @@ -4,7 +4,7 @@ import { randomUUID } from "node:crypto"; import { and, asc, eq } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { agents, agentWakeupRequests, chatCompletionDeliveries as deliveries, chatTaskHandoffs as handoffs, - companies, createDb, documents, environmentLeases, heartbeatRuns, issueComments, issueRecoveryActions, issueThreadInteractions, issues } from "@paperclipai/db"; + companies, createDb, documents, environmentLeases, executionWorkspaces, heartbeatRuns, issueComments, issueRecoveryActions, issueThreadInteractions, issues, workspaceOperations } from "@paperclipai/db"; import { recoveryPostgresSupport as support, startRecoveryPostgres } from "./helpers/recovery-postgres.js"; import { settleInterruptedNativeBootstrap, terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js"; import { getExecutionBlocker } from "../services/execution-blocker.js"; @@ -15,6 +15,8 @@ import { documentService } from "../services/documents.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { chatCompletionDeliveryService, isCompletedOnboardingHandoffWake, prepareChatCompletionTurn, recordChatCompletion, recordChatHandoff } from "../services/chat-completion-delivery.js"; import { heartbeatService, shouldQueueFollowupForRunningIssueWake } from "../services/heartbeat.js"; +import { createHeartbeatQueue } from "../services/heartbeat/queue.js"; +import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js"; (support.supported ? describe : describe.skip)("chat completion delivery", () => { let db: ReturnType; @@ -68,6 +70,75 @@ import { heartbeatService, shouldQueueFollowupForRunningIssueWake } from "../ser })]); expect(JSON.stringify(run.contextSnapshot?.chatCompletionUpdates)).not.toMatch(/Injected|credentials|Plan accepted|Agent approved/); }); + it("retains a distinct completion receipt behind a real dependency gate", async () => { + const f = await seed(); + const blocker = await f.create(); + await issueService(db).update(f.sourceId, { status: "blocked", blockedByIssueIds: [blocker.id] }); + await f.finish(); + const heartbeat = heartbeatService(db); + // The generic dependency signal is replaceable. The outbox's exact wake + // receipt is not: it needs that identity to reconcile a refused admission. + await heartbeat.wakeup(f.agentId, { + source: "automation", triggerDetail: "system", reason: "issue_continuation_needed", + requestedByActorType: "system", requestedByActorId: "dependency-test", + payload: { issueId: f.sourceId }, contextSnapshot: { issueId: f.sourceId }, + }); + const [delivery] = await f.rows(); + const service = chatCompletionDeliveryService(db, heartbeat); + await service.deliver(delivery.id); + const receipts = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + expect(receipts).toEqual(expect.arrayContaining([ + expect.objectContaining({ idempotencyKey: `chat-completion:${delivery.id}:0`, + reason: "issue_dependencies_blocked", status: "skipped", runId: null, + payload: expect.objectContaining({ chatCompletionDeliveryIds: [delivery.id] }) }), + ])); + expect(receipts).toHaveLength(2); + expect(receipts.find(receipt => receipt.idempotencyKey?.startsWith("execution-wait:"))) + .toMatchObject({ coalescedCount: 0 }); + expect((await f.rows())[0]).toMatchObject({ attempts: 0, status: "pending", targetRunId: null, error: null }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); + expect(await issueService(db).getDependencyReadiness(f.sourceId)).toMatchObject({ + isDependencyReady: false, unresolvedBlockerIssueIds: [blocker.id], + }); + }); + it("admits a saved worker completion after its owning finalize encloses the snapshot", async () => { + const f = await seed(), workspaceId = randomUUID(), workerRunId = randomUUID(); + await db.update(issues).set({ responsibleUserId: "operator" }).where(eq(issues.id, f.sourceId)); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: f.companyId, + mode: "isolated_workspace", strategyType: "managed_remote", name: "Saved worker", providerType: "daytona" }); + await db.insert(heartbeatRuns).values({ id: workerRunId, companyId: f.companyId, agentId: f.agentId, + status: "succeeded", contextSnapshot: { issueId: f.task.id } }); + await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, f.task.id)); + await issueService(db).update(f.sourceId, { status: "blocked", blockedByIssueIds: [f.task.id] }); + await db.insert(workspaceOperations).values([ + { companyId: f.companyId, issueId: f.task.id, executionWorkspaceId: workspaceId, heartbeatRunId: workerRunId, + phase: "workspace_finalize", status: "succeeded", startedAt: new Date(10_000), finishedAt: new Date(12_000) }, + { companyId: f.companyId, issueId: f.task.id, executionWorkspaceId: workspaceId, heartbeatRunId: workerRunId, + phase: "remote_workspace_snapshot", status: "succeeded", startedAt: new Date(10_100), finishedAt: new Date(11_900) }, + ]); + await f.finish(); + const deps = createHeartbeatQueueTestDependencies(db), issueSvc = issueService(db); + deps.getAgent.mockImplementation(id => db.select().from(agents).where(eq(agents.id, id)).then(rows => rows[0] ?? null)); + deps.instanceSettings.getExperimental.mockImplementation(() => instanceSettingsService(db).getExperimental()); + deps.issuesSvc.getDependencyReadiness.mockImplementation(issueSvc.getDependencyReadiness); + deps.issuesSvc.listDependencyReadiness.mockImplementation(issueSvc.listDependencyReadiness); + // Admit the real durable run, then hold provider dispatch at the normal drain boundary. + deps.getSchedulingSuppression.mockResolvedValue({ suppressed: true, reason: "task_drain" }); + const queue = createHeartbeatQueue(db, deps); + const service = chatCompletionDeliveryService(db, { wakeup: queue.trackWakeup }); + const [delivery] = await f.rows(); + await service.deliver(delivery.id); + const [queued] = await f.rows(); + expect(queued).toMatchObject({ status: "queued", attempts: 0, error: null, targetRunId: expect.any(String) }); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queued.targetRunId!)); + expect(run).toMatchObject({ status: "queued", contextSnapshot: { issueId: f.sourceId, + wakeReason: "chat_task_completed", chatCompletionDeliveryIds: [delivery.id] } }); + const prepared = await prepareChatCompletionTurn(db, run); + expect(prepared.contextSnapshot?.chatCompletionUpdates).toEqual([ + expect.objectContaining({ id: f.task.id, status: "done", completedAt: expect.any(Date), url: `/issues/${f.task.identifier ?? f.task.id}` }), + ]); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); it("dispatches a committed completion without activity publication and recovers after restart", async () => { const f = await seed(); const run = vi.fn(() => f.service.sweepPending()); diff --git a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts index 5be4f7bd2ff..652ef8e410c 100644 --- a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts @@ -689,14 +689,9 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = }); expect(interactionWake).not.toBeNull(); - await waitForCondition(async () => { - const run = await db - .select({ status: heartbeatRuns.status }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, interactionWake!.id)) - .then((rows) => rows[0] ?? null); - return run?.status === "succeeded"; - }); + // This checks dependency interaction semantics, not provider startup speed. + // Join the admitted execution before inspecting its persisted outcome. + await heartbeat.drainActiveRunExecutions(); const interactionRun = await db .select({ diff --git a/server/src/__tests__/issues-service.test.ts b/server/src/__tests__/issues-service.test.ts index 0d28858b95c..7d8dd0d5823 100644 --- a/server/src/__tests__/issues-service.test.ts +++ b/server/src/__tests__/issues-service.test.ts @@ -4166,6 +4166,7 @@ const dependencyTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSU await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects); + await db.delete(heartbeatRuns); await db.delete(agents); await db.delete(instanceSettings); await db.delete(companies); @@ -4720,6 +4721,42 @@ const dependencyTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSU }); }); + it.each(["covered", "foreign-run", "unfinished", "failed-finalize", "late-finish", "later-operation"])( + "requires a successful owning finalize to cover a nested snapshot: %s", async kind => { + const f = await seedSharedWorkspaceDependency(); + const runId = randomUUID(), foreignRunId = randomUUID(); + await db.insert(heartbeatRuns).values([runId, foreignRunId].map(id => ({ + id, companyId: f.companyId, agentId: f.assigneeAgentId, status: "succeeded", + }))); + const started = new Date("2026-05-23T22:00:00.000Z"); + const finished = new Date("2026-05-23T22:00:02.000Z"); + await db.insert(workspaceOperations).values([ + { companyId: f.companyId, executionWorkspaceId: f.executionWorkspaceId, issueId: f.blockerId, + heartbeatRunId: runId, phase: "workspace_finalize", + status: kind === "failed-finalize" ? "failed" : "succeeded", startedAt: started, finishedAt: finished }, + { companyId: f.companyId, executionWorkspaceId: f.executionWorkspaceId, issueId: f.blockerId, + heartbeatRunId: kind === "foreign-run" ? foreignRunId : runId, phase: "remote_workspace_snapshot", + status: kind === "unfinished" ? "running" : "succeeded", + startedAt: new Date(started.getTime() + 100), + finishedAt: kind === "unfinished" ? null : new Date(finished.getTime() + (kind === "late-finish" ? 100 : -100)) }, + ...(kind === "later-operation" ? [{ + companyId: f.companyId, executionWorkspaceId: f.executionWorkspaceId, issueId: f.blockerId, + heartbeatRunId: runId, phase: "worktree_prepare", status: "succeeded", + startedAt: new Date(finished.getTime() + 100), finishedAt: new Date(finished.getTime() + 200), + }] : []), + ]); + const ready = kind === "covered"; + expect((await svc.listWakeableBlockedDependents(f.blockerId)).map(issue => issue.id)) + .toEqual(ready ? [f.dependentId] : []); + await expect(svc.getDependencyReadiness(f.dependentId)).resolves.toMatchObject({ + isDependencyReady: ready, + pendingFinalizeBlockerIssueIds: ready ? [] : [f.blockerId], + }); + const listed = (await svc.list(f.companyId, { status: "blocked" })).find(issue => issue.id === f.dependentId); + expect(listed?.blockerAttention?.pendingFinalizeBlockerIssueIds ?? []).toEqual(ready ? [] : [f.blockerId]); + }, + ); + it("reports dependency readiness for blocked issue chains", async () => { const companyId = randomUUID(); await seedIssueCompany(db, companyId); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 9f0b0cf7aed..5f8bd897f65 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -2518,6 +2518,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // retain distinct durable receipts even when the same gate blocks them. const coalesceExecutionWait = opts.requestedByActorType === "system" && + opts.allowRunCoalescing !== false && !receiptRequest && !wakeCommentId && queuedCommentIdsFromRunContext(enrichedContextSnapshot).length === 0 && diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 9eb64654b46..6175f0f15dc 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -2562,12 +2562,13 @@ function blockerWorkspaceFacts(companyId: string, workspaceScope: SQL) { } /** An issue's own operation takes precedence over the legacy NULL fallback. - * Only a strictly later successful finalize clears either barrier. */ + * A later successful finalize clears either barrier. A successful operation + * enclosed by its owning finalize is also saved, although it started later. */ function pendingFinalizeBlockerCondition(companyId: string, issueId: SQL, workspaceId: SQL) { return sql`${workspaceId} is not null and exists ( select 1 from (values (1)) seed(value) left join lateral ( - select phase, status, started_at from workspace_operations blocker_operation + select phase, status, started_at, finished_at, heartbeat_run_id from workspace_operations blocker_operation where blocker_operation.company_id = ${companyId}::uuid and blocker_operation.execution_workspace_id = ${workspaceId} and blocker_operation.issue_id = ${issueId} @@ -2578,6 +2579,20 @@ function pendingFinalizeBlockerCondition(companyId: string, issueId: SQL, worksp and coalesce(latest_blocker_operation.status, blocker_workspace_facts.status) = 'succeeded') and (blocker_workspace_facts.finalized_at is null or blocker_workspace_facts.finalized_at <= coalesce(latest_blocker_operation.started_at, blocker_workspace_facts.started_at)) + and not ( + latest_blocker_operation.status = 'succeeded' + and latest_blocker_operation.finished_at is not null + and exists ( + select 1 from workspace_operations owning_finalize + where owning_finalize.company_id = ${companyId}::uuid + and owning_finalize.execution_workspace_id = ${workspaceId} + and owning_finalize.issue_id = ${issueId} + and owning_finalize.heartbeat_run_id = latest_blocker_operation.heartbeat_run_id + and owning_finalize.phase = 'workspace_finalize' and owning_finalize.status = 'succeeded' + and owning_finalize.started_at <= latest_blocker_operation.started_at + and owning_finalize.finished_at >= latest_blocker_operation.finished_at + ) + ) )`; }