diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 50cf963fa4..5b14ed6600 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -144,6 +144,16 @@ The active-lock lifecycle is part of the checkout contract: Stale-lock recovery is crash recovery, not a retry loop. Paperclip must not clear or adopt locks held by non-terminal runs. After stale cleanup, a checkout `409` should mean a real live owner, status/assignee mismatch, unresolved blocker, or active gate still prevents checkout. Agents must treat that `409` as an ownership conflict and stop rather than retrying the same checkout. +### Known execution waits at admission + +A known execution hold is a waiting condition, not a new execution attempt. Every issue wake must read the current effective reconciliation hold under the issue admission lock before creating a run. Resolved recovery bookkeeping can still carry a no-replay hold; only clearing the effective hold makes admission eligible again. The final dispatch gate remains required for changes after admission. + +Repeated automatic signals for an unchanged gate share one durable skipped-wake diagnostic, scoped to company, agent, issue, gate code, and condition identity. The diagnostic retains the first request and counts later observations. This applies to execution reconciliation, dependencies, pause holds, company and agent availability, budget blocks, and disabled heartbeats. These diagnostics do not consume provider attempts and are never proof that a future wake was delivered. All current gates are checked again on the next wake, including the periodic dependency reconciliation sweep. Clearing one gate does not bypass another. + +New comments received during an execution hold retain their individual deferred receipts and ordered comment ids. Release cannot drain those receipts while replay remains blocked; the next eligible wake can adopt them. Authorized external-chat requests also remain deferred with their exact durable receipt. They must use normal promotion and current authorization; a generic wake cannot adopt only their comment ids and discard their actor or session contract. A wait does not authorize replay, reset an incident retry budget, or bypass an interaction's delivery rules. + +The conversation groups repeated empty pre-start reconciliation cancellations into a neutral waiting notice. Started runs, actual startup failures, and run history remain inspectable. No historical run records are deleted. + ### Pre-dispatch configuration validation Pre-dispatch configuration validation is a distinct gate that runs after ownership and checkout are resolved but before the control plane actually dispatches a run. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 61ecfec616..cef78ed009 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -51644,7 +51644,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ), ), ).resolves.toEqual([{ state: "processed" }]); - }); + }, { timeout: 5_000 }); // The row becomes processed inside the mutation transaction, just before // the conversation drain releases its endpoint/thread lease. Synchronize // on that lease boundary before injecting the exact lifecycle commit fault. @@ -51660,7 +51660,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ), ), ).resolves.toEqual([]); - }); + }, { timeout: 5_000 }); if (!first.callbacks.onMessageUpdated) throw new Error("Slack lifecycle callback was not registered"); await first.callbacks.onMessageUpdated({ diff --git a/server/src/__tests__/durable-chat-wakeup.test.ts b/server/src/__tests__/durable-chat-wakeup.test.ts index 2d04815736..3eec7f31d7 100644 --- a/server/src/__tests__/durable-chat-wakeup.test.ts +++ b/server/src/__tests__/durable-chat-wakeup.test.ts @@ -21,6 +21,7 @@ import { chatMessageLinks, chatPublications, issueComments, + issueRecoveryActions, issues, toolApplications, toolConnections, @@ -189,6 +190,50 @@ describe("durable inbound chat scheduler receipts", () => { }; } + async function executionHold(f: { companyId: string; issueId: string; agentId: string }) { + const [action] = await db.insert(issueRecoveryActions).values({ + companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", + ownerType: "board", returnOwnerAgentId: f.agentId, + cause: "legacy_execution_requires_reconciliation", status: "resolved", + fingerprint: randomUUID(), evidence: { automaticRecovery: { replay: "blocked" } }, + nextAction: "Check the stopped execution.", + }).returning(); + return () => db.update(issueRecoveryActions).set({ evidence: {} }).where(eq(issueRecoveryActions.id, action!.id)); + } + + it("defers held inbound chat exactly once and keeps its authority separate from a generic wake", async () => { + const f = await fixture(); + const clearHold = await executionHold(f); + const request = f.request(); + const wake = () => f.heartbeat.wakeup(f.agentId, { + source: "on_demand", triggerDetail: "manual", reason: "issue_commented", + payload: { issueId: f.issueId, commentId: request.commentId }, + contextSnapshot: { issueId: f.issueId, source: "chat:slack", wakeCommentId: request.commentId }, + requestedByActorType: "user", requestedByActorId: request.requestedByActorId, + durableChatRequest: request, + }); + for (let i = 0; i < 3; i++) expect(await wake()).toBeNull(); + const receipts = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agentId)); + expect(receipts).toHaveLength(1); + expect(receipts[0]).toMatchObject({ + id: request.id, idempotencyKey: request.idempotencyKey, requestedAt: request.requestedAt, + status: "deferred_issue_execution", runId: null, requestedByActorId: request.requestedByActorId, + payload: { _paperclipWakeContext: { source: "chat:slack", wakeCommentIds: [request.commentId] } }, + }); + expect(f.authorize).toHaveBeenCalledTimes(1); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.wakeupRequestId, request.id))).toHaveLength(0); + await clearHold(); + const generic = await f.heartbeat.wakeup(f.agentId, { + source: "on_demand", triggerDetail: "manual", reason: "issue_resumed", + payload: { issueId: f.issueId }, requestedByActorType: "user", requestedByActorId: "board-user", + }); + expect(generic?.status).toBe("queued"); + expect(generic?.contextSnapshot?.wakeCommentIds).toBeUndefined(); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.id)))[0]).toMatchObject({ + status: "deferred_issue_execution", runId: null, + }); + }); + async function retryFixture(deferred = false) { const f = await fixture(deferred); const applicationId = randomUUID(), @@ -438,10 +483,12 @@ describe("durable inbound chat scheduler receipts", () => { expect(f.authority).not.toHaveBeenCalled(); }); - it("cancels a revoked deferred retry without reopening or retargeting its original batch", async () => { + it.each([false, true])("cancels a revoked deferred retry without reopening its batch (execution hold=%s)", async (held) => { const f = await retryFixture(true); + const clearHold = held ? await executionHold(f) : null; f.register(); await f.wake(); + await clearHold?.(); f.authority.mockImplementation(async (_tx, input) => { if (input.phase === "promotion") throw conflict("Current chat access was revoked"); @@ -521,10 +568,12 @@ describe("durable inbound chat scheduler receipts", () => { ]); }); - it("preserves the exact retry column and comment batch during deferred promotion", async () => { + it.each([false, true])("preserves the exact retry column and comment batch during promotion (execution hold=%s)", async (held) => { const f = await retryFixture(true); + const clearHold = held ? await executionHold(f) : null; f.register(); await f.wake(); + await clearHold?.(); // A separate fixture-owned run occupies the agent slot after the issue's // predecessor releases it, so this asserts promotion before execution. const blockerId = randomUUID(); diff --git a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts index 6bef458a78..13f770a6b1 100644 --- a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts +++ b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts @@ -17,6 +17,7 @@ import { heartbeatRuns, issueComments, issueRelations, + issueRecoveryActions, issueTreeHoldMembers, issueTreeHolds, issues, @@ -94,8 +95,6 @@ describeEmbeddedPostgres("heartbeat resolved dependency wake reconciliation", () }, 30_000); afterEach(async () => { - vi.clearAllMocks(); - runningProcesses.clear(); // Dependency reconciliation heals missing wakes by enqueuing an // on-demand wake, which dispatches a heartbeat run fire-and-forget (see // startNextQueuedRunForAgent → executeRun in the heartbeat service). That @@ -105,6 +104,8 @@ describeEmbeddedPostgres("heartbeat resolved dependency wake reconciliation", () // insert can land between the events delete and the heartbeat_runs delete and // trip the run_events → runs foreign key. await heartbeatService(db).drainActiveRunExecutions(); + vi.clearAllMocks(); + runningProcesses.clear(); await db.delete(activityLog); await db.delete(heartbeatRunEvents); await db.delete(costEvents); @@ -113,6 +114,7 @@ describeEmbeddedPostgres("heartbeat resolved dependency wake reconciliation", () await db.delete(issueTreeHoldMembers); await db.delete(issueTreeHolds); await db.delete(issueRelations); + await db.delete(issueRecoveryActions); await db.delete(issues); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); @@ -546,6 +548,108 @@ describeEmbeddedPostgres("heartbeat resolved dependency wake reconciliation", () }); }); + async function seedExecutionWait(status: "active" | "resolved" = "resolved") { + const fixture = await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const [action] = await db.insert(issueRecoveryActions).values({ + companyId: fixture.companyId, sourceIssueId: fixture.blockedIssueId, + kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: fixture.agentId, + cause: "legacy_execution_requires_reconciliation", status, + evidence: status === "resolved" ? { automaticRecovery: { replay: "blocked" } } : {}, + fingerprint: randomUUID(), nextAction: "Check the stopped execution before resuming.", + }).returning(); + return { ...fixture, action: action! }; + } + + it.each(["active", "resolved"] as const)("keeps repeated wakes behind a %s execution hold run-free, then resumes once", async (status) => { + const { companyId, agentId, blockedIssueId, action } = await seedExecutionWait(status); + const heartbeat = heartbeatService(db); + // Different producers and wake keys must not create new attempts or notices. + await Promise.all(Array.from({ length: 6 }, (_, i) => heartbeat.wakeup(agentId, { + source: "automation", triggerDetail: "system", reason: "issue_continuation_needed", + requestedByActorType: "system", requestedByActorId: "wait-regression", + idempotencyKey: `producer-${i}`, payload: { issueId: blockedIssueId }, + contextSnapshot: { issueId: blockedIssueId }, + }))); + for (let i = 0; i < 3; i++) { + // Recreate the service to prove the wait is durable across scheduler restarts. + expect((await heartbeatService(db).reconcileResolvedDependencyWakes()).healed).toBe(0); + } + const waits = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); + expect(waits).toHaveLength(1); + expect(waits[0]).toMatchObject({ + status: "skipped", runId: null, reason: "execution_reconciliation_required", coalescedCount: 8, + payload: { issueId: blockedIssueId, executionWait: { recoveryActionId: action.id } }, + }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(0); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + expect(await db.select().from(activityLog).where(and( + eq(activityLog.companyId, companyId), eq(activityLog.action, "issue.blockers_resolved_wake_emitted"), + ))).toHaveLength(0); + + mockAdapterExecute.mockImplementationOnce(async () => { + await db.update(issues).set({ status: "done" }).where(eq(issues.id, blockedIssueId)); + return { exitCode: 0, signal: null, timedOut: false, errorMessage: null, + summary: "Finished the dependency-ready task.", provider: "test", model: "test-model" }; + }); + await db.update(issueRecoveryActions).set({ status: "resolved", evidence: {} }).where(eq(issueRecoveryActions.id, action.id)); + expect((await heartbeat.reconcileResolvedDependencyWakes()).healed).toBe(1); + expect((await heartbeat.reconcileResolvedDependencyWakes()).healed).toBe(0); + await heartbeat.drainActiveRunExecutions(); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(1); + expect(mockAdapterExecute).toHaveBeenCalledTimes(1); + }); + + it("rechecks every gate after an execution hold clears", async () => { + const { companyId, agentId, blockedIssueId, blockerIssueId, action } = await seedExecutionWait(); + const wake = () => heartbeatService(db).wakeup(agentId, { + source: "automation", triggerDetail: "system", reason: "issue_continuation_needed", + requestedByActorType: "system", requestedByActorId: "wait-regression", + payload: { issueId: blockedIssueId }, contextSnapshot: { issueId: blockedIssueId }, + }); + await wake(); + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, blockerIssueId)); + await db.update(issueRecoveryActions).set({ evidence: {} }).where(eq(issueRecoveryActions.id, action.id)); + await wake(); + await wake(); + const waits = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); + expect(waits).toHaveLength(2); + expect(waits.find((row) => row.reason === "issue_dependencies_blocked")?.coalescedCount).toBe(1); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(0); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, blockerIssueId)); + expect((await heartbeatService(db).reconcileResolvedDependencyWakes()).healed).toBe(1); + }); + + it("preserves distinct comments through a hold and adopts them on the next eligible wake", async () => { + const { companyId, agentId, blockedIssueId, action } = await seedExecutionWait(); + const heartbeat = heartbeatService(db); + const commentIds: string[] = []; + for (let i = 0; i < 2; i++) { + const [comment] = await db.insert(issueComments).values({ + companyId, issueId: blockedIssueId, authorUserId: "board-user", body: `Follow-up ${i}`, + }).returning(); + commentIds.push(comment!.id); + expect(await heartbeat.wakeup(agentId, { + source: "on_demand", triggerDetail: "manual", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: "board-user", + payload: { issueId: blockedIssueId, commentId: comment!.id }, + contextSnapshot: { issueId: blockedIssueId, wakeReason: "issue_commented", wakeCommentId: comment!.id }, + })).toBeNull(); + } + const deferred = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); + expect(deferred).toHaveLength(2); + expect(deferred.every((row) => row.status === "deferred_issue_execution" && row.runId === null)).toBe(true); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(0); + await db.update(issueRecoveryActions).set({ evidence: {} }).where(eq(issueRecoveryActions.id, action.id)); + const resumed = await heartbeat.wakeup(agentId, { + source: "on_demand", triggerDetail: "manual", reason: "issue_resumed", + requestedByActorType: "user", requestedByActorId: "board-user", + payload: { issueId: blockedIssueId }, contextSnapshot: { issueId: blockedIssueId }, + }); + expect(resumed?.contextSnapshot?.wakeCommentIds).toEqual(commentIds); + const receipts = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); + expect(receipts.filter((row) => row.status === "coalesced")).toHaveLength(2); + }); + it("retries a resolved dependency wake when the prior wake was skipped as stale", async () => { const { companyId, agentId, blockedIssueId, blockerIssueId } = await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index bbe4fb3395..24b8eee2e2 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -800,6 +800,9 @@ export function createPostgresRunDispatchAdapter( resultJson: { ...parseObject(run.resultJson), stopReason: decision.errorCode, + ...(decision.errorCode === "execution_reconciliation_required" + ? { executionWait: decision.details } + : {}), effectiveTimeoutSec: 0, timeoutConfigured: false, timeoutSource: "stale_queued_run_gate", diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 974c592365..fd8b7e1d4e 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -9,6 +9,7 @@ import { createDb, heartbeatRuns, issueComments, + issueRecoveryActions, issues, } from "@paperclipai/db"; import { @@ -58,6 +59,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { // so the run row must go first. await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); + await db.delete(issueRecoveryActions); await db.delete(issues); await db.delete(agents); await db.delete(companies); @@ -161,6 +163,36 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { return id; } + it("leaves deferred work untouched until the effective execution hold clears", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "blocked" }); + const runId = await seedRun({ companyId, agentId, contextSnapshot: { issueId }, status: "succeeded" }); + const wakeId = await seedDeferredWake({ companyId, agentId, issueId }); + const [hold] = await db.insert(issueRecoveryActions).values({ + companyId, sourceIssueId: issueId, kind: "active_run_watchdog", ownerType: "board", + cause: "legacy_execution_requires_reconciliation", status: "resolved", + fingerprint: runId, evidence: { automaticRecovery: { replay: "blocked" } }, + nextAction: "Check the stopped execution.", + }).returning(); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + let drainCalls = 0; + const drain = async () => { + drainCalls++; + return { outcome: { kind: "released" as const }, postCommitEffects: [] }; + }; + for (let i = 0; i < 3; i++) { + expect((await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, drain)).outcome.kind).toBe("released"); + } + expect(drainCalls).toBe(0); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0]).toMatchObject({ + status: "deferred_issue_execution", runId: null, + }); + await db.update(issueRecoveryActions).set({ evidence: {} }).where(eq(issueRecoveryActions.id, hold!.id)); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, drain); + expect(drainCalls).toBe(1); + }); + // Review test (a): a foreign-company agent id produces the current failed // wake status and the current error text, and creates no run. it("fails a deferred wake whose agent belongs to a different company, without creating a run", async () => { diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 88fb30d6cf..4550810ec7 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { getExecutionBlocker } from "../../../services/execution-blocker.js"; import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -1060,6 +1061,11 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; } + // A release must leave deferred messages intact while replay is held. + if (await getExecutionBlocker(tx, issueRow.companyId, issueRow.id)) { + return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; + } + const locked: LockedIssueExecution = { primaryIssue: toIssueSnapshot(issueRow), run: runSnapshot }; const result = await fn(locked, { host: buildHost(tx, deps), transaction: buildTransaction(tx, deps, db, run) }); return { ...result, run: runSnapshot }; diff --git a/server/src/services/execution-wait.ts b/server/src/services/execution-wait.ts new file mode 100644 index 0000000000..0e749f7f76 --- /dev/null +++ b/server/src/services/execution-wait.ts @@ -0,0 +1,62 @@ +import { createHash } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { agentWakeupRequests, type Db } from "@paperclipai/db"; + +type WakeRequest = typeof agentWakeupRequests.$inferInsert; + +/** + * Record a known gate without inventing an execution attempt. The caller holds + * the company-scoped issue row lock. Only replaceable automatic signals may + * coalesce; messages and authorized interaction receipts keep their identity. + * These receipts are diagnostics, never authority to suppress a future wake: + * admission must read the current gate again before calling this function. + */ +export async function recordExecutionWait( + tx: Db, + input: { + issueId: string; + request: WakeRequest; + condition: Record; + coalesce: boolean; + }, +): Promise<{ created: boolean }> { + const { request, issueId, condition } = input; + const digest = createHash("sha256") + .update(JSON.stringify([request.companyId, request.agentId, issueId, request.reason, condition])) + .digest("hex"); + const key = `execution-wait:${digest}`; + if (input.coalesce) { + const [existing] = await tx.select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where(and( + eq(agentWakeupRequests.companyId, request.companyId), + eq(agentWakeupRequests.agentId, request.agentId), + eq(agentWakeupRequests.status, "skipped"), + eq(agentWakeupRequests.idempotencyKey, key), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + )).limit(1); + if (existing) { + await tx.update(agentWakeupRequests).set({ + coalescedCount: sql`${agentWakeupRequests.coalescedCount} + 1`, + updatedAt: new Date(), + }).where(and( + eq(agentWakeupRequests.companyId, request.companyId), + eq(agentWakeupRequests.id, existing.id), + )); + return { created: false }; + } + } + await tx.insert(agentWakeupRequests).values({ + ...request, + status: "skipped", + runId: null, + finishedAt: new Date(), + idempotencyKey: input.coalesce ? key : request.idempotencyKey, + payload: { + ...request.payload, + issueId, + executionWait: { ...condition, requestedIdempotencyKey: request.idempotencyKey ?? null }, + }, + }); + return { created: true }; +} diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index c4143c93c6..fe6da253eb 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,4 +1,5 @@ import { getExecutionBlocker } from "./execution-blocker.js"; +import { recordExecutionWait } from "./execution-wait.js"; import { legacyExecutionNeedsReconciliation, terminalizeLegacyExecution, @@ -24835,11 +24836,21 @@ export function heartbeatService( "agent_debug_setting"; } + // Automatic signals are replaceable; user input and interaction delivery + // retain distinct durable receipts even when the same gate blocks them. + const coalesceExecutionWait = + opts.requestedByActorType === "system" && + !durableRequest && + !wakeCommentId && + queuedCommentIdsFromRunContext(enrichedContextSnapshot).length === 0 && + !isInteractionResolutionWakePayload(payload ?? {}) && + !hasInteractionContinuationWakeContext(enrichedContextSnapshot); const writeSkippedRequest = async ( skipReason: string, patch: Partial = {}, + waitCondition?: Record, ) => { - await db.insert(agentWakeupRequests).values({ + const request = { ...durableReceiptFields, companyId: agent.companyId, agentId, @@ -24853,7 +24864,18 @@ export function heartbeatService( idempotencyKey: opts.idempotencyKey ?? null, finishedAt: new Date(), ...patch, - }); + }; + if (waitCondition && issueId && isUuidLike(issueId)) { + const waitIssueId = issueId; + return db.transaction(async (tx) => { + await tx.execute(sql`select id from issues where id = ${waitIssueId} and company_id = ${agent.companyId} for update`); + return recordExecutionWait(tx as unknown as Db, { + issueId: waitIssueId, request, condition: waitCondition, coalesce: coalesceExecutionWait, + }); + }); + } + await db.insert(agentWakeupRequests).values(request); + return { created: true }; }; const writeSkippedHeartbeatRequest = async ( skipReason: string, @@ -24893,7 +24915,7 @@ export function heartbeatService( } await writeSkippedRequest("company.inactive", { error: `Wake suppressed because company status is ${companyStatus}`, - }); + }, { companyStatus }); return null; } @@ -25062,7 +25084,9 @@ export function heartbeatService( }, ); if (budgetBlock) { - await writeSkippedRequest("budget.blocked"); + await writeSkippedRequest("budget.blocked", { error: budgetBlock.reason }, { + scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId, + }); throw conflict(budgetBlock.reason, { scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId, @@ -25074,7 +25098,7 @@ export function heartbeatService( if (opts.requestedByActorType !== "user") { await writeSkippedRequest("agent.not_invokable", { error: invokability.message, - }); + }, { status: agent.status, reason: invokability.reason }); } throw conflict(invokability.message, { status: agent.status, @@ -25087,11 +25111,11 @@ export function heartbeatService( const policy = parseHeartbeatPolicy(agent); if (source === "timer" && !policy.enabled) { - await writeSkippedRequest("heartbeat.disabled"); + await writeSkippedRequest("heartbeat.disabled", {}, { enabled: false }); return null; } if (source !== "timer" && !policy.wakeOnDemand) { - await writeSkippedRequest("heartbeat.wakeOnDemand.disabled"); + await writeSkippedRequest("heartbeat.wakeOnDemand.disabled", {}, { wakeOnDemand: false }); return null; } @@ -25131,8 +25155,10 @@ export function heartbeatService( }); if (!treeHoldInteractionWake) { - await writeSkippedRequest("issue_tree_hold_active"); - await logActivity(db, { + const wait = await writeSkippedRequest("issue_tree_hold_active", {}, { + holdId: activePauseHold.holdId, + }); + if (wait.created) await logActivity(db, { companyId: agent.companyId, actorType: "system", actorId: "system", @@ -25485,6 +25511,51 @@ export function heartbeatService( reconciledSourceRunId = sourceRunId; } + // All wake producers share this admission gate. A resolved recovery + // action can still prohibit replay; its durable evidence owns the wait. + // Reconciliation wakes have already proved their authority above and + // must still respect any other effective hold on the same issue. + const executionBlocker = await getExecutionBlocker( + tx as unknown as Db, issue.companyId, issue.id, + ); + if (executionBlocker) { + const condition = { recoveryActionId: executionBlocker.recoveryActionId }; + if (durableRequest || wakeCommentId || hasInteractionContinuationWakeContext(enrichedContextSnapshot)) { + await tx.insert(agentWakeupRequests).values({ + ...durableReceiptFields, + companyId: agent.companyId, agentId, source, triggerDetail, reason, + payload: withQueuedCommentIdsInWakePayload({ + ...payload, + issueId: issue.id, + [DEFERRED_WAKE_CONTEXT_KEY]: enrichedContextSnapshot, + executionWait: condition, + }, [...new Set([ + ...queuedCommentIdsFromRunContext(enrichedContextSnapshot), + ...(wakeCommentId ? [wakeCommentId] : []), + ])]), + status: "deferred_issue_execution", + requestedByActorType: opts.requestedByActorType ?? null, + requestedByActorId: opts.requestedByActorId ?? null, + idempotencyKey: opts.idempotencyKey ?? null, + }); + } else { + await recordExecutionWait(tx as unknown as Db, { + issueId: issue.id, condition, coalesce: coalesceExecutionWait, + request: { + ...durableReceiptFields, + companyId: agent.companyId, agentId, source, triggerDetail, + reason: "execution_reconciliation_required", + error: executionBlocker.nextAction, + payload, + requestedByActorType: opts.requestedByActorType ?? null, + requestedByActorId: opts.requestedByActorId ?? null, + idempotencyKey: opts.idempotencyKey ?? null, + }, + }); + } + return { kind: "deferred" as const }; + } + const issueStateGuard = opts.issueStateGuard; if ( issueStateGuard && @@ -25832,24 +25903,29 @@ export function heartbeatService( !dependencyReadiness.isDependencyReady && !blockedInteractionWake ) { - await tx.insert(agentWakeupRequests).values({ - ...durableReceiptFields, - companyId: agent.companyId, - agentId, - source, - triggerDetail, - reason: "issue_dependencies_blocked", - payload: { - ...(payload ?? {}), - issueId, - unresolvedBlockerIssueIds: - dependencyReadiness.unresolvedBlockerIssueIds, + await recordExecutionWait(tx as unknown as Db, { + issueId: issue.id, + coalesce: coalesceExecutionWait, + condition: { unresolvedBlockerIssueIds: [...dependencyReadiness.unresolvedBlockerIssueIds].sort() }, + request: { + ...durableReceiptFields, + companyId: agent.companyId, + agentId, + source, + triggerDetail, + reason: "issue_dependencies_blocked", + payload: { + ...(payload ?? {}), + issueId, + unresolvedBlockerIssueIds: + dependencyReadiness.unresolvedBlockerIssueIds, + }, + status: "skipped", + requestedByActorType: opts.requestedByActorType ?? null, + requestedByActorId: opts.requestedByActorId ?? null, + idempotencyKey: opts.idempotencyKey ?? null, + finishedAt: new Date(), }, - status: "skipped", - requestedByActorType: opts.requestedByActorType ?? null, - requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, - finishedAt: new Date(), }); return { kind: "skipped" as const }; } @@ -26271,6 +26347,9 @@ export function heartbeatService( // Dedicated interaction wakes carry their own source and session // contract. ID-only adoption must not erase that continuation. return ( + // Durable chat work must keep its receipt, actor, source, and + // session contract through normal promotion and authorization. + !wake.idempotencyKey?.startsWith("chat-inbound:") && !isInteractionResolutionWakePayload(deferredPayload) && !hasInteractionContinuationWakeContext(deferredContext) && (deferredContext.wakeReason ?? wake.reason) === "issue_commented" && diff --git a/server/src/services/native-runtime/native-session-resume.test.ts b/server/src/services/native-runtime/native-session-resume.test.ts index ba66f3c25f..a0114082fd 100644 --- a/server/src/services/native-runtime/native-session-resume.test.ts +++ b/server/src/services/native-runtime/native-session-resume.test.ts @@ -1017,8 +1017,11 @@ const recoveryFakeCodex = resolve( normalizedSessionId, }); expect(continuity).toMatchObject({ - reason: expect.stringContaining( - "run.attach requires a settled Codex provider session", + // The daemon can reject the damaged retained input during startup, + // before attach gets a chance to reject the unsettled provider session. + // Both refusal paths must preserve the exact archived evidence below. + reason: expect.stringMatching( + /run\.attach requires a settled Codex provider session|semantic tool input content digest does not match its transmitted input/, ), previousDriverSessionId: checkpoint.sessionId, }); diff --git a/tests/e2e/acp-stop-continuation.spec.ts b/tests/e2e/acp-stop-continuation.spec.ts index d4c8c651d4..6772749792 100644 --- a/tests/e2e/acp-stop-continuation.spec.ts +++ b/tests/e2e/acp-stop-continuation.spec.ts @@ -71,9 +71,21 @@ for (const { unfinishedWrite, pause } of [{ unfinishedWrite: false, pause: false await page.getByRole("button", { name: "Send", exact: true }).click(); } if (unfinishedWrite) { - await expect(page.getByText("Couldn't start", { exact: false })).toBeVisible(); + await expect(page.getByText("Work cannot start.", { exact: false })).toBeVisible(); + await expect(editor).toHaveText(""); + // Repeated user wakes preserve input without manufacturing attempts. + for (const message of ["go again", "still waiting"]) { + await editor.fill(message); + const response = page.waitForResponse((candidate) => + candidate.request().method() === "POST" && candidate.url().endsWith(`/api/issues/${issue.identifier}/comments`)); + await page.getByRole("button", { name: "Send", exact: true }).click(); + expect((await response).ok()).toBe(true); + await expect(editor).toHaveText(""); + } expect((await json(await request.get(`/api/issues/${issue.id}`))).executionBlocker).toBeTruthy(); await page.waitForTimeout(1000); + await expect(page.getByText("Couldn't start", { exact: false })).toHaveCount(0); + expect(await json(await request.get(`/api/issues/${issue.id}/runs`))).toHaveLength(1); expect(await readFile(path.join(root, "writes"), "utf8")).toBe(writesAtStop); expect((await readFile(path.join(root, "prompts"), "utf8")).trim().split("\n")).toHaveLength(1); } else { diff --git a/ui/src/components/TaskChatThread.test.tsx b/ui/src/components/TaskChatThread.test.tsx index eeefe25a66..cee57a30aa 100644 --- a/ui/src/components/TaskChatThread.test.tsx +++ b/ui/src/components/TaskChatThread.test.tsx @@ -1114,12 +1114,30 @@ describe("TaskChatThread runtime transcript selection", () => { adapterType: "claude_local", createdAt: "2026-08-25T18:00:00.000Z", startedAt: null, finishedAt: "2026-08-25T18:00:00.012Z", }]} />); - expect(container.textContent).toContain("Couldn't start"); + expect(container.textContent).toContain("Waiting to resume"); expect(container.textContent).not.toContain("No user-facing response"); expect(container.textContent).not.toContain("Run completed"); expect(container.querySelector(".text-destructive")).toBeNull(); }); + it.each(["legacy", "native"] as const)("groups repeated %s pre-start holds without hiding executed work", (runtimeMode) => { + const heldRun = (id: string, recoveryActionId: string, startedAt: string | null = null) => ({ + runId: id, runtimeMode, status: "cancelled", errorCode: "execution_reconciliation_required", + agentId: "agent-1", adapterType: runtimeMode === "native" ? "paperclip_runner" : "claude_local", + createdAt: "2026-09-10T18:00:00.000Z", finishedAt: "2026-09-10T18:00:01.000Z", startedAt, + resultJson: { executionWait: { recoveryActionId } }, + }); + render( {}} linkedRuns={[ + ...Array.from({ length: 100 }, (_, i) => heldRun(`wait-${i}`, "hold-1")), + heldRun("ran", "hold-1", "2026-09-10T18:00:00.500Z"), + heldRun("new-hold", "hold-2"), + heldRun("same-new-hold", "hold-2"), + ]} />); + expect(container.textContent?.match(/Waiting to resume/g)).toHaveLength(2); + expect(container.textContent).not.toContain("Couldn't start"); + expect(container.textContent).toContain(runtimeMode === "native" ? "Run cancelled" : "Stopped"); + }); + it("shows cancellation after native progress without offering a retry", () => { nativeTranscriptState.transcriptByRun.set("native-cancelled", [ { diff --git a/ui/src/components/TaskChatThread.tsx b/ui/src/components/TaskChatThread.tsx index 3c20d416e2..046390c3ef 100644 --- a/ui/src/components/TaskChatThread.tsx +++ b/ui/src/components/TaskChatThread.tsx @@ -1380,6 +1380,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { // Raw summary inputs per turn id, so back-to-back same-agent runs can // coalesce into one "Worked" row in the final pass (PAP-362). const turnMergeMetaById = new Map(); + let previousExecutionWaitKey: string | null = null; for (const source of runs) { if (!isTerminalRunStatus(source.status)) continue; if (liveRun && source.id === liveRun.id) continue; @@ -1397,6 +1398,38 @@ export function TaskChatThread(props: TaskChatThreadProps) { settledRunIds.add(source.id); continue; } + // Historical pre-admission cancellations describe a wait, not failed + // work. Collapse repeated observations of that hold, retaining real + // execution and any transcript/comment content between wait episodes. + const executionWait = + source.status === "cancelled" && + !meta?.startedAt && + meta?.errorCode === "execution_reconciliation_required" && + entries.length === 0 && + !lastCommentIdByRun.has(source.id); + if (executionWait) { + const wait = meta?.resultJson?.executionWait; + const waitKey = wait && typeof wait === "object" && "recoveryActionId" in wait + ? String(wait.recoveryActionId) + : "execution_reconciliation_required"; + settledRunIds.add(source.id); + if (previousExecutionWaitKey !== waitKey) { + const id = `${source.id}:execution-wait`; + entriesWithFailures.push({ + ms: toMs(meta?.finishedAt ?? meta?.createdAt), + order: 3, + id, + item: { + id, kind: "marker", variant: "interrupted", tone: "neutral", + label: "Waiting to resume", + detail: "The previous execution needs to be checked before work can continue. See the task’s execution hold for the next action. Individual checks remain in the run history.", + }, + }); + } + previousExecutionWaitKey = waitKey; + continue; + } + previousExecutionWaitKey = null; const acceptedSummary = acceptedSemanticResultSummary(meta?.resultJson); const parsedSource = transcriptToTaskChatItems(entries, { runId: source.id,