From 3cc1032f6a97af2e7456f5b8e9c61a292caa4630 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:27:34 -0600 Subject: [PATCH] fix: deliver saved answers to dependency-blocked tasks (#20363) ## Thinking Path > - Paperclip lets people direct agent work through tasks and structured answers. > - Dependencies stop task execution until prerequisite work finishes. > - A person must still be able to answer or redirect the assignee of a blocked task. > - Ordinary comments already get a bounded answer/triage turn. A saved question response has no comment ID, so the same gates reject it. > - Later dispatch also requires an active task and tries to check it out. That conflicts with bounded interaction on a blocked task. > - This pull request verifies the stored human response at each gate and keeps the turn in bounded interaction mode. > - The assignee can receive the answer without starting dependency work or removing its blockers. ## Linked Issues or Issue Description Refs #19923. That change covers dependency waits and comment retries. This change covers stored interaction responses that have no comment ID. **What happened?** A parent task waits on a parked child. The assignee asked a setup question. After a person answers, the response stays queued. Sending that saved response is refused as `issue_dependencies_blocked`. Ordinary new messages can reach the same assignee. Passing admission alone is insufficient. The dispatch status check rejects a blocked task, and automatic checkout then tries to start work whose prerequisites are unfinished. **Expected behavior** Deliver the stored human answer or confirmation to its requesting assignee as a bounded answer/triage turn. Keep the task blocked and keep its dependency relationships. Preserve current-owner, pause, approval, budget, recovery, and execution gates. **Steps to reproduce** 1. Assign a parent task and make a backlog child block it. 2. Save a question on the parent and answer it as a person. 3. Dispatch its resolved-interaction wake, or send the saved answer from the task queue. 4. Observe that the answer does not reach the provider while an ordinary comment can run in answer/triage mode. **Paperclip version or commit** Reproduced on master at `653ab8b18f76`. This branch is rebased onto master at `0e3fdebea8`. ## What Changed - Verify the stored response by company, task, requesting agent, status, kind, human resolver, and resolution time. A context flag alone is insufficient. - Allow a verified response through dependency admission and claim in the existing bounded interaction mode. - Recheck that response during scheduled retry promotion and final provider dispatch. - Keep bounded responses from checking out the blocked task or setting it to `in_progress`, including when the last blocker finishes during preparation. - Preserve continuation classification through workspace and AI-account resource waits, so the final response and owner checks still run. - Correct the task-case API summary so the existing `case for task` discovery regression passes on the first search page. This fixes a pre-existing failure exposed by full CI, without changing search ranking or assertions. - Ignore obsolete Retry notices after newer conversation work, without acknowledging saved input or changing execution authority. - Document the response contract. Add real PostgreSQL and assembled-heartbeat regressions. ## Verification - Final commit `ba4b337cf1bb0236b7c0a742665b222ee894ff51`: [required CI passed, attempt 2](https://github.com/paperclipai/paperclip/actions/runs/42706363120/attempts/2), including typecheck, production build, server/chat tests, Runner checks, and all 8 browser shards. PR status: 59 passing checks and 3 normal skips. - Red: answered questions and accepted/rejected confirmations return no admitted run before the fix. All three new positive cases fail at the dependency gate. A resource-wait retry also reproduces as refused before its gate is fixed. - Green: at delivery revision `9375a603e4`, focused heartbeat integration and classification passed 14 tests; retry scheduling and dispatch safety passed 241 tests. These use real PostgreSQL and controlled adapter and resource-availability boundaries. The wider initial dependency, native-continuation, and queue suites also passed before the final refinements. - Negative cases keep pending, missing, wrong-task, and agent-resolved responses blocked. Changing the response proof immediately before dispatch cancels the run without calling the provider. Policy controls retain the current assignee, task status, and pending approval checks. - Positive cases execute once, leave the parent blocked, preserve the dependency edge, and leave the child in backlog with no child execution. The same saved response survives a resource wait; removing its human proof refuses promotion. - The resource-wait provider-boundary and blocker-completion regressions also failed before their fixes. All 14 focused integration and classifier cases passed at delivery revision `9375a603e4`, including a changed response during retry preparation. - Red: a retained Retry for an older run makes a newer successful conversation display as failed. Green: all 78 database-backed query and interruption-presentation tests pass, including a later real failure and independent queued input controls. - Server layered typecheck and module-boundary checks passed at `11992aad0c7fbaf13e7548d71ee633560bd8bb6f`, before the one-line API summary correction. That CI passed build, typecheck and every browser shard, but exposed the pre-existing case-search assertion and a report-assets test timeout. Final-head CI passed as linked above. - Fresh Greptile review is 5/5 on final commit `ba4b337cf1bb0236b7c0a742665b222ee894ff51`, with no unresolved threads. - The existing API discovery suite reproduced its first-page assertion before the summary correction; all 3 cases pass afterward. The timed-out report-assets privacy test passed locally in isolation and the full support job passed on the final-head rerun. Runner shutdowns interrupted the first attempt; affected jobs were rerun once without code changes. A separate onboarding timeout also passed on that rerun. - `pnpm verify --explain` requests broad CI because the execution contract is a global input. The final CI run supplied that coverage. - Separately recovered the reported staging conversation through an ordinary message and the existing steering control. The agent acknowledged the message and the saved-answer queue drained. The fresh Notebook action subsequently succeeded. This was on the previously deployed revision; it does not claim this code is deployed. ## Risks - This changes admission and dispatch. Each gate checks the persisted response; no caller-supplied marker grants the exception. - The bounded turn does not check out the task, resolve a dependency, or approve another action. Existing pause, ownership, budget, approval, and stopped-execution controls still apply. - Each additional lookup is one primary-key read and runs only for a response on a dependency-blocked path. There is no new polling worker. - No migration is required. Older queued contexts get the exception only when their stored human response still verifies. No prior run or action is replayed by this change. - This PR does not roll out the code to staging. ## Model Used OpenAI Codex, GPT-6 family, with repository tools, code execution, and browser inspection. The runtime does not expose the exact model ID, context-window limit, or reasoning-mode label; those values are not asserted. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/execution-semantics.md | 13 ++ .../heartbeat-dependency-scheduling.test.ts | 118 ++++++++++++++++++ .../__tests__/task-interruption-query.test.ts | 34 +++++ .../modules/run-dispatch/adapters/postgres.ts | 11 +- .../run-dispatch/domain/policy.test.ts | 15 +++ .../src/modules/run-dispatch/domain/policy.ts | 5 +- .../run-dispatch/domain/wake-context.test.ts | 8 ++ .../run-dispatch/domain/wake-context.ts | 10 +- server/src/services/heartbeat.ts | 6 + server/src/services/heartbeat/queue.ts | 6 +- .../services/queued-interaction-response.ts | 18 +++ .../task-interruption-presentation.ts | 8 +- 12 files changed, 245 insertions(+), 7 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index a4608f08d47..2fee42e1a17 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3602,6 +3602,19 @@ ownership, active controllers, cleanup leases, pause, approval, budget, and norm execution gates still apply. Dependency-blocked interaction mode remains limited to its existing answer/triage contract. +A saved human answer or confirmation can reach the requesting assignee while +dependencies remain unfinished. Admission and dispatch verify the stored response +against its company, task, requesting agent, resolved status, and human resolver. +The response uses dependency-blocked answer/triage mode. It does not check out the +task, set it to `in_progress`, resolve its dependencies, or approve another action. +An absent, changed, pending, or agent-resolved response grants no such exception. +That bounded response remains a continuation after a workspace or AI-account wait, +so final dispatch still verifies it. Finishing a dependency during preparation +does not turn the admitted response into an attempt to check out the task. +Task interruption notices ignore Retry receipts for a superseded run. A later +successful conversation must not inherit the old failure banner. This read-only +projection neither retires Retry authority nor acknowledges independent saved input. + Snapshot preservation and native copyback are distinct receipts. A historical workspace recovered as an immutable snapshot does not make its old native sync reference finalized. Once recovery settles, the current inbound owner can diff --git a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts index 85cc945b7f5..134ce9cc8cc 100644 --- a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts @@ -29,6 +29,7 @@ import { issueDocuments, issueRelations, issueRecoveryActions, + issueThreadInteractions, issueTreeHolds, issues, workspaceOperations, @@ -42,6 +43,8 @@ import { heartbeatService } from "../services/heartbeat.ts"; import { runningProcesses } from "../adapters/index.ts"; import { createPostgresRunDispatchAdapter } from "../modules/run-dispatch/adapters/postgres.js"; import { taskInterruptionPresentation } from "../services/task-interruption-presentation.js"; +import * as workspacePreparation from "../services/heartbeat/workspace-preparation.js"; +import { WorkspaceBusyDeferral } from "../services/heartbeat/retries.js"; const mockAdapterExecute = vi.hoisted(() => vi.fn(async () => ({ @@ -130,6 +133,7 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = await new Promise((resolve) => setTimeout(resolve, 50)); } await heartbeat.drainActiveRunExecutions(); + vi.restoreAllMocks(); mockAdapterExecute.mockReset(); mockAdapterExecute.mockImplementation(async () => ({ exitCode: 0, @@ -564,6 +568,120 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = expect(recoveredIntent).toMatchObject({ status: "coalesced" }); }); + it.each(["answered", "accepted", "rejected", "pending", "missing", "wrong_issue", "agent_resolved", "changed_before_dispatch", + "workspace_wait", "workspace_wait_changed", "blocker_finishes"] as const)( + "delivers only a saved human response while dependencies remain blocked: %s", async response => { + const companyId = randomUUID(), agentId = randomUUID(), blockerId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Response fixture", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "responsible-user" }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "responsible-user", + status: "active", membershipRole: "member" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Parent assignee", role: "engineer", status: "active", + adapterType: "codex_local", adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, permissions: {} }); + await db.insert(issues).values([ + { id: blockerId, companyId, title: "Parked trial", status: "backlog", responsibleUserId: "responsible-user" }, + { id: issueId, companyId, title: "Finish trial prerequisites", status: "blocked", assigneeAgentId: agentId, + responsibleUserId: "responsible-user" }, + ]); + await db.insert(issueRelations).values({ companyId, issueId: blockerId, relatedIssueId: issueId, type: "blocks" }); + const confirmation = response === "accepted" || response === "rejected"; + const status = confirmation ? response : response === "pending" ? "pending" : "answered"; + const [interaction] = await db.insert(issueThreadInteractions).values({ companyId, + issueId: response === "wrong_issue" ? blockerId : issueId, createdByAgentId: agentId, + kind: confirmation ? "request_confirmation" : "ask_user_questions", status, + title: "Setup access", resolvedByUserId: response === "pending" || response === "agent_resolved" ? null : "responsible-user", + resolvedByAgentId: response === "agent_resolved" ? agentId : null, + resolvedAt: response === "pending" ? null : new Date(), + payload: confirmation ? { version: 1, prompt: "Continue setup?" } + : { version: 1, questions: [{ id: "access", prompt: "Can setup continue?", selectionMode: "single", + options: [{ id: "ready", label: "Access is ready" }] }] }, + result: response === "pending" ? null : confirmation ? { version: 1, outcome: response } + : { version: 1, answers: [{ questionId: "access", optionIds: ["ready"] }] }, + }).returning(); + const context = { issueId, mutation: "interaction", interactionId: response === "missing" ? randomUUID() : interaction.id, + interactionKind: interaction.kind, interactionStatus: status, wakeReason: "issue_commented" }; + const resourceWait = response === "workspace_wait" || response === "workspace_wait_changed"; + let deferredSourceId: string | null = null; + const workspace = resourceWait ? vi.spyOn(workspacePreparation, "prepareHeartbeatWorkspace") + .mockRejectedValueOnce(new WorkspaceBusyDeferral({ holder: { runId: randomUUID(), agentId, + issueId: blockerId, issueIdentifier: null }, executionWorkspaceId: randomUUID(), + wasIssueAssignee: true, deferralAttempt: 0 })) : null; + const service = response === "changed_before_dispatch" || resourceWait || response === "blocker_finishes" ? heartbeatService(db, { + beforeChatControlRecoveryCheck: async ({ stage }) => { + if (response === "blocker_finishes" && stage === "claim") + await db.update(issues).set({ status: "done" }).where(eq(issues.id, blockerId)); + }, + beforeResolvedInteractionContinuationDispatchCheck: async ({ runId }) => { + if (response === "changed_before_dispatch" || response === "workspace_wait_changed" && deferredSourceId && runId !== deferredSourceId) + await db.update(issueThreadInteractions).set({ resolvedByUserId: null, resolvedByAgentId: agentId }) + .where(eq(issueThreadInteractions.id, interaction.id)); + }, + }) : heartbeat; + const wake = await service.wakeup(agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: context, contextSnapshot: context, requestedByActorType: "user", requestedByActorId: "responsible-user" }); + let dispatchedRunId = wake?.id; + if (resourceWait) { + expect(wake).not.toBeNull(); + await service.drainActiveRunExecutions(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake!.id)))[0]) + .toMatchObject({ status: "cancelled", errorCode: "workspace_busy" }); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, wake!.id)); + expect(retry).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "workspace_busy" }); + deferredSourceId = wake!.id; + dispatchedRunId = retry.id; + workspace!.mockRestore(); + expect((await service.promoteDueScheduledRetries(new Date(Date.now() + 5 * 60_000))).runIds).toContain(retry.id); + await service.resumeQueuedRuns(); + } + const shouldDeliver = ["answered", "accepted", "rejected", "workspace_wait", "blocker_finishes"].includes(response); + if (shouldDeliver) { + expect(wake).not.toBeNull(); + await service.drainActiveRunExecutions(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, dispatchedRunId!)); + expect({ status: run.status, error: run.error, errorCode: run.errorCode }) + .toEqual({ status: "succeeded", error: null, errorCode: null }); + expect(run).toMatchObject({ status: "succeeded", contextSnapshot: { + interactionId: interaction.id, interactionStatus: status, dependencyBlockedInteraction: true, + unresolvedBlockerIssueIds: [blockerId], + } }); + expect(mockAdapterExecute).toHaveBeenCalledTimes(1); + if (response === "answered") { + // A resource wait keeps the same bounded interaction and must not + // strand the answer again at scheduled promotion. + const [retry] = await db.insert(heartbeatRuns).values({ companyId, agentId, + invocationSource: "automation", status: "scheduled_retry", scheduledRetryAt: new Date(), + scheduledRetryReason: "interaction_continuation_infra_retry", retryOfRunId: run.id, + contextSnapshot: { ...run.contextSnapshot, wakeReason: "interaction_continuation_infra_retry", + retryReason: "interaction_continuation_infra_retry" }, + }).returning(); + const retryGate = () => createPostgresRunDispatchAdapter(db).evaluateScheduledRetryGate({ + companyId, runId: retry.id, retryReasonOverride: "interaction_continuation_infra_retry", now: new Date(), + }); + expect(await retryGate()).toMatchObject({ allowed: true }); + await db.update(issueThreadInteractions).set({ resolvedByUserId: null, resolvedByAgentId: agentId }) + .where(eq(issueThreadInteractions.id, interaction.id)); + expect(await retryGate()).toMatchObject({ allowed: false, errorCode: "issue_dependencies_blocked" }); + } + } else if (response === "changed_before_dispatch" || response === "workspace_wait_changed") { + expect(wake).not.toBeNull(); + await service.drainActiveRunExecutions(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, dispatchedRunId!)))[0]) + .toMatchObject({ status: "cancelled", errorCode: "issue_not_in_progress" }); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + } else { + expect(wake).toBeNull(); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + } + expect(await db.select().from(issueRelations).where(eq(issueRelations.relatedIssueId, issueId))).toHaveLength(1); + expect((await db.select().from(issues).where(eq(issues.id, blockerId)))[0].status) + .toBe(response === "blocker_finishes" ? "done" : "backlog"); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0].status).toBe("blocked"); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${blockerId}`)).toHaveLength(0); + }, + ); + it("keeps blocked descendants idle until their blockers resolve", async () => { const companyId = randomUUID(); const agentId = randomUUID(); diff --git a/server/src/__tests__/task-interruption-query.test.ts b/server/src/__tests__/task-interruption-query.test.ts index 226bf36dfb6..badec8c72b5 100644 --- a/server/src/__tests__/task-interruption-query.test.ts +++ b/server/src/__tests__/task-interruption-query.test.ts @@ -124,6 +124,40 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => expect((await taskInterruptionPresentation(db, parent))?.retryWait?.reason).not.toBe("dependency_blocked"); }); + it("does not report a superseded Retry as failure after a newer conversation succeeds", async () => { + const companyId = randomUUID(), agentId = randomUUID(), oldRunId = randomUUID(), newRunId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Continued conversation", issuePrefix: `C${companyId.slice(0, 7)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Agent", adapterType: "paperclip_runner" }); + const [task] = await db.insert(issues).values({ companyId, title: "Continue setup", status: "in_review", + assigneeAgentId: agentId }).returning(); + await db.insert(heartbeatRuns).values([ + { id: oldRunId, companyId, agentId, nativeIssueId: task.id, scopeKind: "issue", runtimeMode: "native", + status: "failed", createdAt: new Date(Date.now() - 120_000), finishedAt: new Date(Date.now() - 110_000) }, + { id: newRunId, companyId, agentId, nativeIssueId: task.id, scopeKind: "issue", runtimeMode: "native", + status: "succeeded", createdAt: new Date(Date.now() - 60_000), finishedAt: new Date(Date.now() - 30_000) }, + ]); + const [retry] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "on_demand", reason: "task_retry", + status: "deferred_issue_execution", requestedByActorType: "user", requestedAt: new Date(Date.now() - 90_000), + payload: { issueId: task.id, taskRetry: { sourceRunId: oldRunId } } }).returning(); + for (const presentation of [await taskInterruptionPresentation(db, task), + (await taskInterruptionPresentations(db, companyId, [task])).get(task.id)]) { + expect(presentation).toMatchObject({ state: "resolved", canRetry: false, intentId: null, observedRunId: newRunId }); + } + // Reads do not acknowledge input or retire execution authority. + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, retry.id)))[0]?.status) + .toBe("deferred_issue_execution"); + // A later provider failure still belongs to the current run. + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "native_provider_terminal_failed" }) + .where(eq(heartbeatRuns.id, newRunId)); + expect(await taskInterruptionPresentation(db, task)).toMatchObject({ state: "failed", canRetry: true, observedRunId: newRunId }); + await db.update(heartbeatRuns).set({ status: "succeeded", errorCode: null }).where(eq(heartbeatRuns.id, newRunId)); + // Unconsumed ordinary input is independent of the obsolete Retry. + const [message] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "on_demand", + reason: "issue_commented", status: "deferred_issue_execution", requestedByActorType: "user", + requestedAt: new Date(Date.now() - 20_000), payload: { issueId: task.id } }).returning(); + expect(await taskInterruptionPresentation(db, task)).toMatchObject({ state: "failed", intentId: message.id }); + }); + // G1 wave s9802 t02: an accepted Retry held by the unpriced-usage refusal // read as a budget wait. It is an accounting matter; name it so. it("labels a Retry held by unpriced usage as accounting, not budget", async () => { diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 3a4806de488..5174c71137e 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -42,6 +42,7 @@ import { budgetService, budgetServiceInTransaction, recheckAdmittedRunBudgetRead import { lockRunBudgetRecoveryOwnerInTransaction, recheckRunBudgetReservationInTransaction } from "../../../services/budget-reservations.js"; import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js"; import { isUnstartedTaskQuestionAnswer } from "../../../services/unstarted-task-question-answer.js"; +import { isVerifiedDependencyInteractionResponse } from "../../../services/queued-interaction-response.js"; import { isHeartbeatWakeOnDemandEnabled } from "../../../services/heartbeat-policy.js"; import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; @@ -488,8 +489,12 @@ export function createPostgresRunDispatchAdapter( .from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, input.wakeupRequestId))).then(rows => rows[0]?.actor ?? null) : null }) : false; + const responseAdmitted = readiness && !readiness.isDependencyReady + && input.contextSnapshot.dependencyBlockedInteraction === true + && await isVerifiedDependencyInteractionResponse(dbOrTx, { companyId: input.companyId, + issueId, agentId: input.agentId, context: input.contextSnapshot }); facts.dependenciesBlocked = - readiness && !readiness.isDependencyReady && !stallAdmitted + readiness && !readiness.isDependencyReady && !stallAdmitted && !responseAdmitted ? { unresolvedBlockerIssueIds: readiness.unresolvedBlockerIssueIds, unresolvedBlockerCount: readiness.unresolvedBlockerCount, @@ -680,6 +685,10 @@ export function createPostgresRunDispatchAdapter( issueExecutionRunId: issue?.executionRunId ?? null, issueCheckoutRunId: issue?.checkoutRunId ?? null, isResolvedInteractionContinuation, + isDependencyBlockedInteractionResponse: Boolean(isResolvedInteractionContinuation && issue?.status === "blocked" + && context.dependencyBlockedInteraction === true + && await isVerifiedDependencyInteractionResponse(dbOrTx, { companyId: input.companyId, issueId, + agentId: input.agentId, context })), isUnstartedTaskQuestionAnswer: Boolean(isResolvedInteractionContinuation && issue?.status === "todo" && await isUnstartedTaskQuestionAnswer(dbOrTx, { companyId: input.companyId, issueId, interactionId: context.interactionId })), isConnectionContinuation: (isResolvedInteractionContinuation && context.interactionKind === "connection_intent") diff --git a/server/src/modules/run-dispatch/domain/policy.test.ts b/server/src/modules/run-dispatch/domain/policy.test.ts index 8fc730f76b4..a97b3d294fd 100644 --- a/server/src/modules/run-dispatch/domain/policy.test.ts +++ b/server/src/modules/run-dispatch/domain/policy.test.ts @@ -267,6 +267,21 @@ describe("decideScheduledRetryGate", () => { }); describe("decideQueuedRunStaleness", () => { + it("keeps a verified response bounded to a blocked task and its current assignee", () => { + const response = { ...baseStalenessFacts(), isResolvedInteractionContinuation: true, + issueStatus: "blocked", isDependencyBlockedInteractionResponse: true }; + expect(decideQueuedRunStaleness(response, NOW)).toEqual({ stale: false }); + expect(decideQueuedRunStaleness({ ...response, isDependencyBlockedInteractionResponse: false }, NOW)) + .toMatchObject({ stale: true, errorCode: "issue_not_in_progress" }); + expect(decideQueuedRunStaleness({ ...response, issueAssigneeAgentId: "other-agent" }, NOW)) + .toMatchObject({ stale: true, errorCode: "issue_assignee_changed" }); + for (const issueStatus of ["backlog", "todo", "done", "cancelled"]) { + expect(decideQueuedRunStaleness({ ...response, issueStatus }, NOW)).toMatchObject({ stale: true }); + } + expect(decideQueuedRunStaleness({ ...response, pendingResponse: "approval" }, NOW)) + .toMatchObject({ stale: true, errorCode: "issue_waiting_for_response" }); + }); + it("is not stale when every rule passes", () => { expect(decideQueuedRunStaleness(baseStalenessFacts(), NOW)).toEqual({ stale: false, diff --git a/server/src/modules/run-dispatch/domain/policy.ts b/server/src/modules/run-dispatch/domain/policy.ts index 59d955278cd..655025d3b81 100644 --- a/server/src/modules/run-dispatch/domain/policy.ts +++ b/server/src/modules/run-dispatch/domain/policy.ts @@ -147,6 +147,8 @@ export type QueuedRunFacts = { isResolvedInteractionContinuation: boolean; /** Verified under the issue lock: an answer to a question no run asked, on a todo task that has never started. */ isUnstartedTaskQuestionAnswer?: boolean; + /** The stored human response was verified again for bounded dependency interaction. */ + isDependencyBlockedInteractionResponse?: boolean; /** A connection resolution or tool refresh can resume an agent waiting in review. */ isConnectionContinuation?: boolean; isInteractionWake: boolean; @@ -514,9 +516,10 @@ export function decideQueuedRunStaleness( if (facts.isResolvedInteractionContinuation || facts.isConnectionContinuation) { // A todo task that has never started has no parked execution to protect. const startsUnstartedTask = facts.issueStatus === "todo" && facts.isUnstartedTaskQuestionAnswer === true; + const answersBlockedTask = facts.issueStatus === "blocked" && facts.isDependencyBlockedInteractionResponse === true; const earlyStatus = decideIssueStatus({ status: facts.issueStatus, - requiresInProgress: facts.issueStatus !== "in_review" && !startsUnstartedTask, + requiresInProgress: facts.issueStatus !== "in_review" && !startsUnstartedTask && !answersBlockedTask, terminalBypass: true, }); if (earlyStatus === "not_in_progress") { diff --git a/server/src/modules/run-dispatch/domain/wake-context.test.ts b/server/src/modules/run-dispatch/domain/wake-context.test.ts index 02f73e5d1df..7fb92f8d916 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.test.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.test.ts @@ -9,6 +9,14 @@ import { } from "./wake-context.js"; describe("wake context", () => { + it.each(["workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"])( + "keeps a bounded response classified through a %s resource wait", retryReason => { + const context = { interactionId: "response", interactionStatus: "answered", mutation: "interaction", + dependencyBlockedInteraction: true, wakeReason: `${retryReason}_retry`, retryReason }; + expect(isResolvedInteractionContinuationWakeContext(context)).toBe(true); + expect(isResolvedInteractionContinuationWakeContext({ ...context, dependencyBlockedInteraction: false })).toBe(false); + expect(isResolvedInteractionContinuationWakeContext({ ...context, mutation: undefined })).toBe(false); + }); it("recognizes a retired available connection without accepting ordinary expiry", () => { const context = { interactionId: "connection", interactionStatus: "expired", interactionKind: "connection_intent", connectionIntentResolution: "existing_connection", mutation: "interaction", wakeReason: "issue_commented" }; diff --git a/server/src/modules/run-dispatch/domain/wake-context.ts b/server/src/modules/run-dispatch/domain/wake-context.ts index 645d10f78ae..40bf97bd0d5 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.ts @@ -29,6 +29,9 @@ export const RESOLVED_INTERACTION_CONTINUATION_STATUSES = new Set([ "cancelled", "rejected", ]); +const RESOLVED_INTERACTION_RESOURCE_WAIT_REASONS = new Set([ + WORKSPACE_BUSY_RETRY_REASON, AI_CONNECTION_BUSY_RETRY_REASON, AI_CONNECTION_POOL_WAIT_RETRY_REASON, +]); // A run silent for this long no longer counts as the shared-workspace holder // it was waiting behind. Kept here so the deferral gate and the parked-wait @@ -170,7 +173,7 @@ export function allowsIssueInteractionWake( /** * Whether a wake may run while its issue still has unresolved blockers: a - * verified comment interaction, or a stalled child or blocker + * verified comment or human-response interaction, or a stalled child or blocker * (`issue_blocker_stalled`, once per stall), which the assignee must be able * to re-delegate, escalate or block on while that dependency is stuck. The * caller verifies the stall against server state (`verifyBlockerStallWake`); @@ -181,8 +184,9 @@ export function allowsDependencyBlockedWake( contextSnapshot: Record | null | undefined, interactionWakeReasons: ReadonlySet, verifiedBlockerStall: boolean, + verifiedInteractionResponse = false, ): boolean { - return allowsIssueInteractionWake(contextSnapshot, interactionWakeReasons) || verifiedBlockerStall; + return allowsIssueInteractionWake(contextSnapshot, interactionWakeReasons) || verifiedBlockerStall || verifiedInteractionResponse; } export function isResolvedInteractionContinuationWakeContext(contextSnapshot: unknown): boolean { @@ -201,6 +205,8 @@ export function isResolvedInteractionContinuationWakeContext(contextSnapshot: un return ( (mutation === "interaction" && wakeReason === "issue_commented") || (mutation === "interaction" && retryReason === RUNNER_REVISION_WAIT_RETRY_REASON) || + (mutation === "interaction" && context.dependencyBlockedInteraction === true && retryReason !== null + && RESOLVED_INTERACTION_RESOURCE_WAIT_REASONS.has(retryReason)) || wakeReason === INTERACTION_CONTINUATION_INFRA_WAKE_REASON || retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON ); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 6fbd9ad75eb..e00e1c6f1c1 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -4,6 +4,7 @@ import { isProcessRunAdmissionClosed } from "./heartbeat/run-admission.js"; import type { StopInvocationOptions } from "./heartbeat/run-control.js"; import { isContextCompactionRun } from "./context-compaction.js"; import { isUnstartedTaskQuestionAnswer } from "./unstarted-task-question-answer.js"; +import { isVerifiedDependencyInteractionResponse } from "./queued-interaction-response.js"; import { registerDeferredReceiptWork } from "./heartbeat/deferred-receipt-work.js"; import { reconcileTaskChatRetryReceipts } from "./task-retry.js"; import { registerIssueRecoveryWork } from "./recovery/work.js"; @@ -2230,9 +2231,14 @@ export function heartbeatService( .listDependencyReadiness(agent.companyId, [issueId]) .then((rows) => rows.get(issueId) ?? null) : null; + const dependencyResponse = issueId && issueContext?.status === "blocked" + && context.dependencyBlockedInteraction === true + && await isVerifiedDependencyInteractionResponse(db, { companyId: agent.companyId, + issueId, agentId: agent.id, context }); if ( issueId && issueContext && + !dependencyResponse && isResolvedInteractionContinuationWakeContext(context) ) { try { diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 230c913591c..4881298f030 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -42,7 +42,7 @@ import { isCompletedOnboardingHandoffWake, } from "../chat-completion-delivery.js"; import { hasWorkspaceRestoreFailure } from "@paperclipai/shared"; -import { readQueuedInteractionResponse } from "../queued-interaction-response.js"; +import { isVerifiedDependencyInteractionResponse, readQueuedInteractionResponse } from "../queued-interaction-response.js"; import { isConversation, isConversationReset, @@ -1107,6 +1107,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende context, ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS, blockerStallVerified, + await isVerifiedDependencyInteractionResponse(db, { companyId: run.companyId, + issueId, agentId: run.agentId, context }), ) ) { await cancelQueuedRunForBlockedDependencies( @@ -3997,6 +3999,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS, await verifyBlockerStallWake(tx as unknown as Db, { companyId: issue.companyId, issueId: issue.id, contextSnapshot: enrichedContextSnapshot, requestedByActorType: opts.requestedByActorType }), + await isVerifiedDependencyInteractionResponse(tx, { companyId: issue.companyId, + issueId: issue.id, agentId, context: enrichedContextSnapshot }), ); if (blockedInteractionWake) { diff --git a/server/src/services/queued-interaction-response.ts b/server/src/services/queued-interaction-response.ts index 930fa56c851..5486716ffc6 100644 --- a/server/src/services/queued-interaction-response.ts +++ b/server/src/services/queued-interaction-response.ts @@ -71,6 +71,24 @@ export function queuedInteractionSteeringBody(interaction: { export type QueuedInteractionResponse = NonNullable>>; +/** A stored human answer can reach its assignee while dependency work stays blocked. */ +export async function isVerifiedDependencyInteractionResponse(db: Pick, input: { + companyId: string; issueId: string; agentId: string; context: unknown; +}): Promise { + const id = queuedInteractionId(input.context); + if (!id) return false; + const context = record(input.context); + const [interaction] = await db.select({ status: issueThreadInteractions.status, kind: issueThreadInteractions.kind, + resolvedByUserId: issueThreadInteractions.resolvedByUserId, resolvedAt: issueThreadInteractions.resolvedAt }) + .from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.id, id), eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issueId), eq(issueThreadInteractions.createdByAgentId, input.agentId), + )).limit(1); + return Boolean(interaction?.resolvedByUserId && interaction.resolvedAt + && interaction.status === context.interactionStatus + && (context.interactionKind === undefined || interaction.kind === context.interactionKind)); +} + export async function readQueuedInteractionResponse(db: Db, companyId: string, issueId: string, payload: unknown) { const id = queuedInteractionId(payload); if (!id) return null; diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index 9f6c9130273..38132ec4429 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -477,7 +477,11 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i const ownerStopRequired = new Set(ownerStops.map(row => `${row.issueId}:${row.runId}`)); const byIssue = (rows: T[]) => new Map(rows.map(row => [row.issueId, row])); const runMap = byIssue(runs), waitMap = new Map(); - const retryWaitMap = byIssue(waits.filter(wait => wait.sourceRunId)); + // A Retry belongs to the run it observed. New conversation work supersedes + // that authority, even before dispatch retires the old receipt. Retained + // ordinary input is independent and must still be shown as waiting. + const currentWaits = waits.filter(wait => !wait.sourceRunId || wait.sourceRunId === runMap.get(wait.issueId)?.id); + const retryWaitMap = byIssue(currentWaits.filter(wait => wait.sourceRunId)); const retryIssueIds = [...retryWaitMap.keys()]; const retryDependencyMap = new Map(); if (retryIssueIds.length) { @@ -497,7 +501,7 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i savingFilesOnly: ids.every(id => saving.has(id)) }); } } - for (const wait of waits) { + for (const wait of currentWaits) { const current = waitMap.get(wait.issueId); if (!current || wait.requestedAt > current.requestedAt || (wait.requestedAt.getTime() === current.requestedAt.getTime() && wait.id > current.id)) waitMap.set(wait.issueId, wait);