mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
eebdfe71b1
commit
3cc1032f6a
12 files changed
+245
-7
No files matched your search
@@ -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
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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") {
|
||||
|
||||
@@ -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" };
|
||||
|
||||
@@ -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<string, unknown> | null | undefined,
|
||||
interactionWakeReasons: ReadonlySet<string>,
|
||||
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
|
||||
);
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -71,6 +71,24 @@ export function queuedInteractionSteeringBody(interaction: {
|
||||
|
||||
export type QueuedInteractionResponse = NonNullable<Awaited<ReturnType<typeof readQueuedInteractionResponse>>>;
|
||||
|
||||
/** A stored human answer can reach its assignee while dependency work stays blocked. */
|
||||
export async function isVerifiedDependencyInteractionResponse(db: Pick<Db, "select">, input: {
|
||||
companyId: string; issueId: string; agentId: string; context: unknown;
|
||||
}): Promise<boolean> {
|
||||
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;
|
||||
|
||||
@@ -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 = <T extends { issueId: string }>(rows: T[]) => new Map(rows.map(row => [row.issueId, row]));
|
||||
const runMap = byIssue(runs), waitMap = new Map<string, typeof waits[number]>();
|
||||
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<string, RetryDependencies>();
|
||||
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);
|
||||
|
||||
Reference in new issue
Block a user