diff --git a/packages/adapter-utils/src/server-utils.test.ts b/packages/adapter-utils/src/server-utils.test.ts index 1aafda0e3ed..0f5b3da9bb4 100644 --- a/packages/adapter-utils/src/server-utils.test.ts +++ b/packages/adapter-utils/src/server-utils.test.ts @@ -3008,6 +3008,34 @@ describe("renderPaperclipWakePrompt", () => { expect(JSON.parse(stringifyPaperclipWakePayload(payload)!)).toMatchObject({ dispositionRepair: payload.dispositionRepair }); }); + it("tells the agent what it can do when a child or blocker stalls on the board", () => { + const payload = { + reason: "issue_blocker_stalled", + issue: { id: "parent-1", identifier: "CHA-5", title: "Coordinate the rollout", status: "blocked", priority: "medium" }, + blockerStall: { + stalledIssueId: "child-1", + stalledIssueIdentifier: "CHA-8", + stalledIssueTitle: "Ship the migration", + stalledIssueStatus: "blocked", + relation: "child", + recoveryAction: { id: "action-1", kind: "run_recovery", cause: "execution_recovery_budget_exhausted", + status: "escalated", ownerType: "board", nextAction: "Retry the run or reassign the task" }, + }, + }; + expect(JSON.parse(stringifyPaperclipWakePayload(payload) ?? "{}")).toMatchObject({ + blockerStall: { stalledIssueIdentifier: "CHA-8", relation: "child", + recoveryAction: { cause: "execution_recovery_budget_exhausted", nextAction: "Retry the run or reassign the task" } }, + }); + const prompt = renderPaperclipWakePrompt(payload); + expect(prompt).toContain("## Stalled dependency"); + expect(prompt).toContain("CHA-8 Ship the migration (blocked)"); + expect(prompt).toContain("execution_recovery_budget_exhausted"); + expect(prompt).toContain("Retry the run or reassign the task"); + expect(prompt).toMatch(/re-delegate/i); + expect(prompt).toMatch(/escalate/i); + expect(prompt).toMatch(/block this issue/i); + }); + it("includes continuation and child issue summaries in structured wake context", () => { const payload = { reason: "issue_children_completed", diff --git a/packages/adapter-utils/src/server-utils.ts b/packages/adapter-utils/src/server-utils.ts index e7debfb0deb..4fbd4ae8938 100644 --- a/packages/adapter-utils/src/server-utils.ts +++ b/packages/adapter-utils/src/server-utils.ts @@ -704,6 +704,22 @@ type PaperclipWakeChildIssueSummary = { summary: string | null; }; +/** `issue_blocker_stalled`: the child or blocker that stalled on a board-owned recovery action. */ +type PaperclipWakeBlockerStall = { + stalledIssueId: string | null; + stalledIssueIdentifier: string | null; + stalledIssueTitle: string | null; + stalledIssueStatus: string | null; + relation: "child" | "blocker"; + recoveryAction: { + kind: string | null; + cause: string | null; + status: string | null; + ownerType: string | null; + nextAction: string | null; + }; +}; + type PaperclipWakeBlockerSummary = { id: string | null; identifier: string | null; @@ -817,6 +833,7 @@ type PaperclipWakePayload = { annotationDeltas: PaperclipWakeAnnotationDelta[]; childIssueSummaries: PaperclipWakeChildIssueSummary[]; childIssueSummaryTruncated: boolean; + blockerStall: PaperclipWakeBlockerStall | null; commentIds: string[]; latestCommentId: string | null; comments: PaperclipWakeComment[]; @@ -1322,6 +1339,29 @@ function normalizePaperclipWakeChildIssueSummary( return { id, identifier, title, status, priority, summary }; } +function normalizePaperclipWakeBlockerStall(value: unknown): PaperclipWakeBlockerStall | null { + const stall = parseObject(value); + const text = (field: unknown) => asString(field, "").trim() || null; + const stalledIssueId = text(stall.stalledIssueId); + const stalledIssueIdentifier = text(stall.stalledIssueIdentifier); + if (!stalledIssueId && !stalledIssueIdentifier) return null; + const action = parseObject(stall.recoveryAction); + return { + stalledIssueId, + stalledIssueIdentifier, + stalledIssueTitle: text(stall.stalledIssueTitle), + stalledIssueStatus: text(stall.stalledIssueStatus), + relation: stall.relation === "blocker" ? "blocker" : "child", + recoveryAction: { + kind: text(action.kind), + cause: text(action.cause), + status: text(action.status), + ownerType: text(action.ownerType), + nextAction: text(action.nextAction), + }, + }; +} + function normalizePaperclipWakeBlockerSummary( value: unknown, ): PaperclipWakeBlockerSummary | null { @@ -1895,6 +1935,7 @@ export function normalizePaperclipWakePayload( payload.childIssueSummaryTruncated, false, ), + blockerStall: normalizePaperclipWakeBlockerStall(payload.blockerStall), commentIds, latestCommentId: asString(payload.latestCommentId, "").trim() || null, comments, @@ -3067,6 +3108,26 @@ function renderPaperclipWakePromptBody( } } + if (normalized.blockerStall) { + const stall = normalized.blockerStall; + const label = stall.stalledIssueIdentifier ?? stall.stalledIssueId ?? "unknown"; + const action = stall.recoveryAction; + lines.push( + "", + "## Stalled dependency", + "", + `Your ${stall.relation === "child" ? "child issue" : "blocker"} ${label}${stall.stalledIssueTitle ? ` ${stall.stalledIssueTitle}` : ""}${stall.stalledIssueStatus ? ` (${stall.stalledIssueStatus})` : ""} cannot continue on its own: it waits on a recovery action that needs the board.`, + `- recovery: ${action.kind ?? "unknown"}${action.cause ? ` — ${action.cause}` : ""}${action.status ? ` (${action.status})` : ""}`, + ...(action.nextAction ? [`- next action for the board: ${action.nextAction}`] : []), + "", + "Choose one, then stop:", + `- Re-delegate: if the work can be done another way, create or reassign a task for it and stop depending on ${label}.`, + `- Escalate: comment on this issue with what is stuck and what the board must do on ${label} (its Retry or reassignment), so the board sees it here.`, + `- Block this issue with a reason that names ${label} and its recovery action, if nothing else can move until the board acts.`, + `Do not redo ${label}'s work yourself while it is assigned, and do not wait silently.`, + ); + } + if (normalized.childIssueSummaries.length > 0) { lines.push("", "Direct child issue summaries:"); for (const child of normalized.childIssueSummaries) { diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 731b9d044bc..8c480ccc8e5 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -614,6 +614,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0435_premium_puff_adder.sql", "0436_legal_reavers.sql", "0447_legacy_controller_process.sql", + "0450_petite_frightful_four.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0450_petite_frightful_four.sql b/packages/db/src/migrations/0450_petite_frightful_four.sql new file mode 100644 index 00000000000..6b004b13aba --- /dev/null +++ b/packages/db/src/migrations/0450_petite_frightful_four.sql @@ -0,0 +1 @@ +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "agent_wakeup_requests_issue_blocker_stalled_uq" ON "agent_wakeup_requests" USING btree ("company_id","idempotency_key") WHERE "agent_wakeup_requests"."idempotency_key" LIKE 'issue_blocker_stalled:%' AND "agent_wakeup_requests"."status" IN ('queued', 'deferred_issue_execution', 'claimed', 'coalesced', 'completed'); \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0445_snapshot.json b/packages/db/src/migrations/meta/0450_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0445_snapshot.json rename to packages/db/src/migrations/meta/0450_snapshot.json index 7ab3519b394..af424af109a 100644 --- a/packages/db/src/migrations/meta/0445_snapshot.json +++ b/packages/db/src/migrations/meta/0450_snapshot.json @@ -1,6 +1,6 @@ { - "id": "f0d8430b-7abf-4232-86a9-11a268a0d20e", - "prevId": "93cbf124-1264-4bc3-a27d-b651ce9640e7", + "id": "b8f43f1a-3230-4cda-991d-0e7b6323114b", + "prevId": "4bd9d289-9cb8-4262-9dd7-04ce2b6020c4", "version": "7", "dialect": "postgresql", "tables": { @@ -4102,6 +4102,28 @@ "method": "btree", "with": {} }, + "agent_wakeup_requests_issue_blocker_stalled_uq": { + "name": "agent_wakeup_requests_issue_blocker_stalled_uq", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "idempotency_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"agent_wakeup_requests\".\"idempotency_key\" LIKE 'issue_blocker_stalled:%' AND \"agent_wakeup_requests\".\"status\" IN ('queued', 'deferred_issue_execution', 'claimed', 'coalesced', 'completed')", + "concurrently": true, + "method": "btree", + "with": {} + }, "agent_wakeup_requests_company_payload_issue_idx": { "name": "agent_wakeup_requests_company_payload_issue_idx", "columns": [ @@ -4252,6 +4274,18 @@ "primaryKey": false, "notNull": false }, + "harness_verified_digest": { + "name": "harness_verified_digest", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "harness_verified_at": { + "name": "harness_verified_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, "reports_to": { "name": "reports_to", "type": "uuid", @@ -21965,6 +21999,112 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.controller_instances": { + "name": "controller_instances", + "schema": "", + "columns": { + "boot_id": { + "name": "boot_id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "epoch": { + "name": "epoch", + "type": "bigserial", + "primaryKey": false, + "notNull": true + }, + "process_space": { + "name": "process_space", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "pid": { + "name": "pid", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "started_at": { + "name": "started_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "renewed_at": { + "name": "renewed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "lease_expires_at": { + "name": "lease_expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "fenced_at": { + "name": "fenced_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "fenced_by_epoch": { + "name": "fenced_by_epoch", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "stopped_at": { + "name": "stopped_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "controller_instances_epoch_idx": { + "name": "controller_instances_epoch_idx", + "columns": [ + { + "expression": "epoch", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "controller_instances_lease_idx": { + "name": "controller_instances_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.cost_events": { "name": "cost_events", "schema": "", @@ -34856,6 +34996,12 @@ "primaryKey": false, "notNull": false }, + "controller_process": { + "name": "controller_process", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "execution_stage": { "name": "execution_stage", "type": "text", @@ -35125,6 +35271,22 @@ "method": "btree", "with": {} }, + "heartbeat_runs_legacy_controller_lease_idx": { + "name": "heartbeat_runs_legacy_controller_lease_idx", + "columns": [ + { + "expression": "controller_lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'legacy' AND \"heartbeat_runs\".\"status\" = 'running' AND \"heartbeat_runs\".\"controller_boot_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_cost_accounting_pending_idx": { "name": "heartbeat_runs_cost_accounting_pending_idx", "columns": [ @@ -44766,6 +44928,12 @@ "primaryKey": false, "notNull": false }, + "lease_holder": { + "name": "lease_holder", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "controller_boot_id": { "name": "controller_boot_id", "type": "text", @@ -44930,6 +45098,38 @@ "method": "btree", "with": {} }, + "native_run_finalizations_finalizer_lease_idx": { + "name": "native_run_finalizations_finalizer_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"lease_owner\" like 'native-finalizer:%'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_cleanup_lease_idx": { + "name": "native_run_finalizations_cleanup_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"lease_owner\" like 'native-cleanup:%'", + "concurrently": false, + "method": "btree", + "with": {} + }, "native_run_finalizations_accepted_result_lease_idx": { "name": "native_run_finalizations_accepted_result_lease_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 5d8bc8d9a27..fbb1f107ccc 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3137,6 +3137,13 @@ "when": 1794147993995, "tag": "0449_controller_instances", "breakpoints": true + }, + { + "idx": 450, + "version": "7", + "when": 1794170960477, + "tag": "0450_petite_frightful_four", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/agent_wakeup_requests.ts b/packages/db/src/schema/agent_wakeup_requests.ts index 83006f2616c..11c82c37122 100644 --- a/packages/db/src/schema/agent_wakeup_requests.ts +++ b/packages/db/src/schema/agent_wakeup_requests.ts @@ -99,6 +99,13 @@ export const agentWakeupRequests = pgTable( chatCompletionIdempotencyUq: uniqueIndex("agent_wakeup_requests_chat_completion_uq") .on(table.companyId, table.idempotencyKey) .where(sql`${table.idempotencyKey} LIKE 'chat-completion:%'`), + // One delivered or pending wake per stall (issue_blocker_stalled). Refused + // wakes (skipped, failed, cancelled) are not covered, so a stall retries + // once its agent takes wakes again; concurrent sweeps cannot both deliver. + issueBlockerStalledIdempotencyUq: uniqueIndex("agent_wakeup_requests_issue_blocker_stalled_uq") + .on(table.companyId, table.idempotencyKey) + .concurrently() + .where(sql`${table.idempotencyKey} LIKE 'issue_blocker_stalled:%' AND ${table.status} IN ('queued', 'deferred_issue_execution', 'claimed', 'coalesced', 'completed')`), companyPayloadIssueIdx: index("agent_wakeup_requests_company_payload_issue_idx").on( table.companyId, sql`(${table.payload} ->> 'issueId')`, diff --git a/packages/shared/src/constants.ts b/packages/shared/src/constants.ts index 341858b027a..ec1cd7fca0c 100644 --- a/packages/shared/src/constants.ts +++ b/packages/shared/src/constants.ts @@ -417,6 +417,14 @@ export const ISSUE_RECOVERY_ACTION_OWNER_TYPES = [ ] as const; export type IssueRecoveryActionOwnerType = (typeof ISSUE_RECOVERY_ACTION_OWNER_TYPES)[number]; +/** + * Wake reason for an agent whose child or blocker stalled on a recovery action + * that needs the board. Woken once per stall (one recovery action). + */ +export const ISSUE_BLOCKER_STALLED_WAKE_REASON = "issue_blocker_stalled" as const; +export const ISSUE_BLOCKER_STALL_RELATIONS = ["child", "blocker"] as const; +export type IssueBlockerStallRelation = (typeof ISSUE_BLOCKER_STALL_RELATIONS)[number]; + export const ISSUE_RECOVERY_ACTION_OUTCOMES = [ "restored", "handed_back", diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 1cfff379faf..9ab71a2cf39 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -404,6 +404,8 @@ export { ISSUE_DISPOSITION_REPAIR_RETRY_REASON, ISSUE_RECOVERY_ACTION_STATUSES, ISSUE_RECOVERY_ACTION_OWNER_TYPES, + ISSUE_BLOCKER_STALLED_WAKE_REASON, + ISSUE_BLOCKER_STALL_RELATIONS, ISSUE_RECOVERY_ACTION_OUTCOMES, pluginOperationIssueOriginKind, isPluginOperationIssueOriginKind, @@ -597,6 +599,7 @@ export { type IssueRecoveryActionKind, type IssueRecoveryActionStatus, type IssueRecoveryActionOwnerType, + type IssueBlockerStallRelation, type IssueRecoveryActionOutcome, type IssueRelationType, type IssueTreeControlMode, @@ -1122,6 +1125,7 @@ export type { IssueBlockedInboxSeverity, IssueBlockedInboxState, IssueRecoveryAction, + IssueBlockerStalledWakeContext, IssueWatchdog, IssueWatchdogStatus, IssueWatchdogSummary, diff --git a/packages/shared/src/types/index.ts b/packages/shared/src/types/index.ts index 4aa18840240..c2e73fb144a 100644 --- a/packages/shared/src/types/index.ts +++ b/packages/shared/src/types/index.ts @@ -724,6 +724,7 @@ export type { IssueBlockedInboxSeverity, IssueBlockedInboxState, IssueRecoveryAction, + IssueBlockerStalledWakeContext, SuccessfulRunHandoffState, SuccessfulRunHandoffStateKind, IssueScheduledRetry, diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index f78d79c7449..d053454552a 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -25,6 +25,7 @@ import type { IssueRecoveryActionKind, IssueRecoveryActionOutcome, IssueRecoveryActionOwnerType, + IssueBlockerStallRelation, IssueRecoveryActionStatus, IssueWorkMode, IssueThreadInteractionContinuationPolicy, @@ -623,6 +624,23 @@ export interface IssueUnblockDescriptor { action: string; } +/** Wake context of `issue_blocker_stalled`: the stalled issue and the recovery action it waits on. */ +export interface IssueBlockerStalledWakeContext { + stalledIssueId: string; + stalledIssueIdentifier: string | null; + stalledIssueTitle: string; + stalledIssueStatus: string; + relation: IssueBlockerStallRelation; + recoveryAction: { + id: string; + kind: string; + cause: string; + status: string; + ownerType: string; + nextAction: string; + }; +} + export interface IssueRecoveryAction { /** Read-only activity of the exact native run named by the wake policy. */ nativeRunActivity?: { @@ -974,6 +992,8 @@ export interface Issue { blockedTransitionAt?: Date | null; blockedOwnerNotifiedAt?: Date | null; activeRecoveryAction?: IssueRecoveryAction | null; + /** Children and blockers stalled on a recovery action that needs the board (issue detail only). */ + stalledDependencies?: IssueBlockerStalledWakeContext[]; successfulRunHandoff?: SuccessfulRunHandoffState | null; executionBlocker?: ExecutionBlocker | null; interruption?: TaskInterruptionPresentation | null; diff --git a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts index 3918fcb3ea1..85cc945b7f5 100644 --- a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts @@ -40,6 +40,7 @@ import { import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; 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"; const mockAdapterExecute = vi.hoisted(() => @@ -815,6 +816,73 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = expect(noActiveRuns).toBe(true); }); + it("admits an issue_blocker_stalled wake while a blocker is unresolved", async () => { + // The waiter's blocker stalled on a board-owned recovery action. Parking + // this wake until the blocker resolves would hide the stall again. + const companyId = randomUUID(); + const agentId = randomUUID(); + const blockerId = randomUUID(); + const dependentId = randomUUID(); + await db.insert(companies).values({ + id: companyId, name: "Paperclip", 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: "CodexCoder", role: "engineer", status: "active", adapterType: "codex_local", + adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, permissions: {}, + }); + await db.insert(issues).values([ + { id: blockerId, companyId, title: "Stalled blocker", status: "blocked", priority: "high", responsibleUserId: "responsible-user" }, + { id: dependentId, companyId, title: "Waiting dependent", status: "blocked", priority: "medium", assigneeAgentId: agentId, + responsibleUserId: "responsible-user" }, + ]); + await db.insert(issueRelations).values({ companyId, issueId: blockerId, relatedIssueId: dependentId, type: "blocks" }); + const [action] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: blockerId, kind: "run_recovery", + status: "resolved", outcome: "blocked", ownerType: "board", cause: "native_session_retry_exhausted", + fingerprint: "stall", nextAction: "Retry", resolvedAt: new Date() }).returning(); + const stallWake = (requestedByActorType: "system" | "agent") => heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_blocker_stalled", + payload: { issueId: dependentId, stalledIssueId: blockerId }, + idempotencyKey: `issue_blocker_stalled:${dependentId}:${blockerId}:${action.id}:${requestedByActorType}`, + requestedByActorType, + requestedByActorId: requestedByActorType === "system" ? "issue-blocker-stall" : agentId, + contextSnapshot: { issueId: dependentId, wakeReason: "issue_blocker_stalled", + blockerStall: { stalledIssueId: blockerId, relation: "blocker", recoveryAction: { id: action.id } } }, + }); + + // The reason string alone is not trusted: an agent cannot use it to skip the blocker gate. + expect(await stallWake("agent")).toBeNull(); + + const stalledWake = await stallWake("system"); + expect(stalledWake).not.toBeNull(); + await heartbeat.drainActiveRunExecutions(); + const [run] = await db.select({ status: heartbeatRuns.status, contextSnapshot: heartbeatRuns.contextSnapshot }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, stalledWake!.id)); + expect(run?.status).toBe("succeeded"); + expect(run?.contextSnapshot).toMatchObject({ dependencyBlockedInteraction: true, unresolvedBlockerIssueIds: [blockerId] }); + + // A resource wait replaces the run with a server-created retry that keeps + // the stall context but not the wake reason. The retry gate still admits it. + const retryGate = async (contextSnapshot: Record) => { + const [retry] = await db.insert(heartbeatRuns).values({ companyId, agentId, invocationSource: "automation", + status: "scheduled_retry", scheduledRetryAt: new Date(), scheduledRetryReason: "transient_failure", + retryOfRunId: stalledWake!.id, contextSnapshot }).returning(); + return createPostgresRunDispatchAdapter(db).evaluateScheduledRetryGate({ companyId, runId: retry.id, + retryReasonOverride: "transient_failure", now: new Date() }); + }; + const replacement: Record = { ...(run!.contextSnapshot as Record), + wakeReason: "bounded_transient_heartbeat_retry", retryReason: "transient_failure", retryOfRunId: stalledWake!.id }; + expect(await retryGate(replacement)).toMatchObject({ allowed: true }); + const { blockerStall: _stall, ...withoutStall } = replacement; + expect(await retryGate(withoutStall)).toMatchObject({ allowed: false, errorCode: "issue_dependencies_blocked" }); + // Once the stall's action is no longer current, the retry waits for the blocker like any other. + await db.update(issueRecoveryActions).set({ outcome: "restored" }).where(eq(issueRecoveryActions.id, action.id)); + expect(await retryGate(replacement)).toMatchObject({ allowed: false, errorCode: "issue_dependencies_blocked" }); + }); + it("defers issue_blockers_resolved as a follow-up when the same issue is already running", async () => { const companyId = randomUUID(); const agentId = randomUUID(); diff --git a/server/src/__tests__/issue-blocker-stall-wakeups.test.ts b/server/src/__tests__/issue-blocker-stall-wakeups.test.ts new file mode 100644 index 00000000000..d77227daabe --- /dev/null +++ b/server/src/__tests__/issue-blocker-stall-wakeups.test.ts @@ -0,0 +1,333 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + activityLog, agents, agentWakeupRequests, companies, createDb, heartbeatRuns, issueRecoveryActions, issueRelations, issues, type Db, +} from "@paperclipai/db"; +import { ISSUE_BLOCKER_STALLED_WAKE_REASON } from "@paperclipai/shared"; +import { recoveryService } from "../services/recovery/service.js"; +import { issueBlockerStalledWakeKey, listStalledDependencies, recoveryActionSweepCondition, wakeWaitersOfStalledIssues } from "../services/recovery/blocker-stall-wakeups.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RECOVERY_ACTION_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); + +// Restart chaos w-20261108T181313-s9701 t05: a child blocked on a board-owned +// recovery action (retries exhausted), and its parent sat silently blocked for +// 50+ minutes because only completion or resolution wakes a parent. +(support.supported ? describe : describe.skip)("issue_blocker_stalled wakes", () => { + let temporary: Awaited> | undefined; + let db: Db; + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("blocker-stall-"); + db = createDb(external ?? temporary!.connectionString); + db.$client.options.onnotice = () => {}; + }, 30_000); + afterEach(async () => { vi.restoreAllMocks(); await db.execute(sql`truncate companies cascade`); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + + // Records each wake the way the queue does, so the idempotency check sees it. + function wakeRecorder() { + const calls: Array<{ agentId: string; input: Record }> = []; + const enqueueWakeup = vi.fn(async (agentId: string, input: Record) => { + calls.push({ agentId, input }); + const [row] = await db.insert(agentWakeupRequests).values({ companyId: input.payload.companyId ?? companyIdOf.get(agentId)!, + agentId, source: input.source, triggerDetail: input.triggerDetail, reason: input.reason, payload: input.payload, + idempotencyKey: input.idempotencyKey, status: "queued" }).returning(); + return { id: row.id }; + }); + return { calls, enqueueWakeup }; + } + const companyIdOf = new Map(); + + async function fixture(options: { ownerType?: "board" | "agent"; status?: "active" | "escalated" } = {}) { + const companyId = randomUUID(), parentAgent = randomUUID(), childAgent = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Stall", issuePrefix: `S${companyId.slice(0, 6)}` }); + await db.insert(agents).values([ + { id: parentAgent, companyId, name: "Parent owner", role: "engineer", status: "idle" }, + { id: childAgent, companyId, name: "Child owner", role: "engineer", status: "idle" }, + ]); + companyIdOf.set(parentAgent, companyId); companyIdOf.set(childAgent, companyId); + const [parent] = await db.insert(issues).values({ companyId, title: "Parent", status: "blocked", assigneeAgentId: parentAgent }).returning(); + const [child] = await db.insert(issues).values({ companyId, title: "Child", status: "blocked", parentId: parent.id, + assigneeAgentId: childAgent }).returning(); + const [action] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: child.id, + kind: "run_recovery", status: options.status ?? "escalated", ownerType: options.ownerType ?? "board", + cause: "execution_recovery_budget_exhausted", fingerprint: "fixture", nextAction: "Retry the run or reassign the task" }).returning(); + return { companyId, parentAgent, childAgent, parent, child, action }; + } + + it("wakes the parent's assignee once per stall, with the stalled child and its recovery action", async () => { + const f = await fixture(); + const { calls, enqueueWakeup } = wakeRecorder(); + const service = recoveryService(db, { enqueueWakeup: enqueueWakeup as never }); + await service.reconcileActiveRecoveryActions({ companyId: f.companyId }); + await service.reconcileActiveRecoveryActions({ companyId: f.companyId }); + const parentWakes = calls.filter(call => call.input.reason === ISSUE_BLOCKER_STALLED_WAKE_REASON); + expect(parentWakes).toHaveLength(1); + expect(parentWakes[0]).toMatchObject({ agentId: f.parentAgent, input: { + idempotencyKey: issueBlockerStalledWakeKey({ waiterIssueId: f.parent.id, stalledIssueId: f.child.id, recoveryActionId: f.action.id }), + contextSnapshot: { issueId: f.parent.id, wakeReason: ISSUE_BLOCKER_STALLED_WAKE_REASON, blockerStall: { + stalledIssueId: f.child.id, stalledIssueStatus: "blocked", relation: "child", + recoveryAction: { id: f.action.id, status: "escalated", ownerType: "board", nextAction: "Retry the run or reassign the task" }, + } }, + } }); + const activity = await db.select().from(activityLog).where(eq(activityLog.action, "issue.blocker_stalled_wake")); + expect(activity).toHaveLength(1); + expect(activity[0]).toMatchObject({ entityId: f.parent.id, actorType: "system" }); + + // Unstalled, then stalled again: a new recovery action is a new stall. + await db.update(issueRecoveryActions).set({ status: "resolved", resolvedAt: new Date() }).where(eq(issueRecoveryActions.id, f.action.id)); + const [again] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.child.id, + kind: "run_recovery", status: "escalated", ownerType: "board", cause: "execution_recovery_budget_exhausted", + fingerprint: "fixture-2", nextAction: "Retry" }).returning(); + await service.reconcileActiveRecoveryActions({ companyId: f.companyId }); + const later = calls.filter(call => call.input.reason === ISSUE_BLOCKER_STALLED_WAKE_REASON); + expect(later).toHaveLength(2); + expect(later[1]!.input.contextSnapshot.blockerStall.recoveryAction.id).toBe(again.id); + }); + + it("wakes the assignee of an issue the stalled one blocks", async () => { + const f = await fixture(); + const dependentAgent = randomUUID(); + await db.insert(agents).values({ id: dependentAgent, companyId: f.companyId, name: "Dependent owner", role: "engineer", status: "idle" }); + companyIdOf.set(dependentAgent, f.companyId); + const [dependent] = await db.insert(issues).values({ companyId: f.companyId, title: "Dependent", status: "blocked", + assigneeAgentId: dependentAgent }).returning(); + await db.insert(issueRelations).values({ companyId: f.companyId, issueId: f.child.id, relatedIssueId: dependent.id, type: "blocks" }); + const { calls, enqueueWakeup } = wakeRecorder(); + await recoveryService(db, { enqueueWakeup: enqueueWakeup as never }).reconcileActiveRecoveryActions({ companyId: f.companyId }); + const stalled = calls.filter(call => call.input.reason === ISSUE_BLOCKER_STALLED_WAKE_REASON); + expect(stalled.map(call => [call.agentId, call.input.contextSnapshot.blockerStall.relation]).sort()) + .toEqual([[dependentAgent, "blocker"], [f.parentAgent, "child"]].sort()); + }); + + it("wakes only the swept waiter for a shared stalled blocker, so N dependents cost N, not N squared", async () => { + const f = await fixture(); + const dependentAgent = randomUUID(); + await db.insert(agents).values({ id: dependentAgent, companyId: f.companyId, name: "Dependent owner", role: "engineer", status: "idle" }); + companyIdOf.set(dependentAgent, f.companyId); + const dependents = await db.insert(issues).values([1, 2, 3].map(index => ({ companyId: f.companyId, + title: `Dependent ${index}`, status: "blocked" as const, assigneeAgentId: dependentAgent }))).returning(); + await db.insert(issueRelations).values(dependents.map(dependent => ({ companyId: f.companyId, issueId: f.child.id, + relatedIssueId: dependent.id, type: "blocks" as const }))); + const { calls, enqueueWakeup } = wakeRecorder(); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId, issueId: dependents[0]!.id }); + expect(calls.map(call => call.input.payload.issueId)).toEqual([dependents[0]!.id]); + // The stalled issue's own pass reaches every waiter, skipping those already woken. + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId, issueId: f.child.id }); + expect(calls.map(call => call.input.payload.issueId).sort()) + .toEqual([f.parent.id, ...dependents.map(dependent => dependent.id)].sort()); + }); + + it("pages past waiters whose agents take no wakes, so later waiters are reached", async () => { + const f = await fixture(); + const [sleepy, awake] = [randomUUID(), randomUUID()]; + await db.insert(agents).values([ + { id: sleepy, companyId: f.companyId, name: "Sleepy", role: "engineer", status: "paused" }, + { id: awake, companyId: f.companyId, name: "Awake", role: "engineer", status: "idle" }, + ]); + companyIdOf.set(sleepy, f.companyId); companyIdOf.set(awake, f.companyId); + const rows = Array.from({ length: 101 }, (_, index) => ({ id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, + companyId: f.companyId, title: `Waiter ${index}`, status: "todo" as const, assigneeAgentId: index < 100 ? sleepy : awake })); + await db.insert(issues).values(rows); + await db.insert(issueRelations).values(rows.map(row => ({ companyId: f.companyId, issueId: f.child.id, + relatedIssueId: row.id, type: "blocks" as const }))); + const { calls, enqueueWakeup } = wakeRecorder(); + const result = await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId, issueId: f.child.id }); + expect(calls.map(call => call.input.payload.issueId)).toContain(rows[100]!.id); + expect(result.deferred).toBe(100); + }); + + it("wakes the dependent when the blocker's board action resolved as blocked (t05 shape)", async () => { + // t05: CHA-8 blocks CHA-5. Its board-owned action was resolved with outcome + // "blocked" six seconds after creation, as the issue moved to blocked for + // the board, so no open action remained to find. + const f = await fixture(); + await db.update(issueRecoveryActions).set({ status: "resolved", outcome: "blocked", resolvedAt: new Date() }) + .where(eq(issueRecoveryActions.id, f.action.id)); + await db.update(issues).set({ parentId: null }).where(eq(issues.id, f.child.id)); + await db.insert(issueRelations).values({ companyId: f.companyId, issueId: f.child.id, relatedIssueId: f.parent.id, type: "blocks" }); + const { calls, enqueueWakeup } = wakeRecorder(); + const service = recoveryService(db, { enqueueWakeup: enqueueWakeup as never }); + // The per-issue blocked sweep visits the stalled blocker itself. + await service.reconcileResolvedDependencyWakeBackstop({ companyId: f.companyId, issueId: f.child.id }); + await service.reconcileResolvedDependencyWakeBackstop({ companyId: f.companyId, issueId: f.child.id }); + const stalled = calls.filter(call => call.input.reason === ISSUE_BLOCKER_STALLED_WAKE_REASON); + expect(stalled).toHaveLength(1); + expect(stalled[0]).toMatchObject({ agentId: f.parentAgent, input: { contextSnapshot: { blockerStall: { + stalledIssueId: f.child.id, relation: "blocker", recoveryAction: { id: f.action.id, status: "resolved" } } } } }); + + // Retry moves it out of blocked: not a stall any more, and no new wake. + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, f.child.id)); + await service.reconcileResolvedDependencyWakeBackstop({ companyId: f.companyId, issueId: f.child.id }); + await service.reconcileActiveRecoveryActions({ companyId: f.companyId }); + expect(calls.filter(call => call.input.reason === ISSUE_BLOCKER_STALLED_WAKE_REASON)).toHaveLength(1); + }); + + it("lists the waiting issue's stalled children and blockers for its page", async () => { + const f = await fixture(); + const [blocker] = await db.insert(issues).values({ companyId: f.companyId, title: "Blocker", status: "blocked" }).returning(); + await db.insert(issueRelations).values({ companyId: f.companyId, issueId: blocker.id, relatedIssueId: f.parent.id, type: "blocks" }); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: blocker.id, kind: "run_recovery", + status: "resolved", outcome: "blocked", ownerType: "board", cause: "native_session_retry_exhausted", + fingerprint: "blocker", nextAction: "Retry", resolvedAt: new Date() }); + const listed = await listStalledDependencies(db, { companyId: f.companyId, issueId: f.parent.id }); + expect(listed.map(entry => [entry.stalledIssueId, entry.relation]).sort()) + .toEqual([[blocker.id, "blocker"], [f.child.id, "child"]].sort()); + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, blocker.id)); + expect((await listStalledDependencies(db, { companyId: f.companyId, issueId: f.parent.id })).map(entry => entry.stalledIssueId)) + .toEqual([f.child.id]); + }); + + it("never wakes a waiter parked in the backlog", async () => { + const f = await fixture(); + await db.update(issues).set({ status: "backlog" }).where(eq(issues.id, f.parent.id)); + const { calls, enqueueWakeup } = wakeRecorder(); + await recoveryService(db, { enqueueWakeup: enqueueWakeup as never }).reconcileActiveRecoveryActions({ companyId: f.companyId }); + expect(calls).toEqual([]); + }); + + it("does not let a merged run swallow a second stall, and guards the waiter's state", async () => { + const f = await fixture(); + const [second] = await db.insert(issues).values({ companyId: f.companyId, title: "Second child", status: "blocked", + parentId: f.parent.id, assigneeAgentId: f.childAgent }).returning(); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: second.id, kind: "run_recovery", + status: "escalated", ownerType: "board", cause: "execution_recovery_budget_exhausted", fingerprint: "second", nextAction: "Retry" }); + const { calls, enqueueWakeup } = wakeRecorder(); + await recoveryService(db, { enqueueWakeup: enqueueWakeup as never }).reconcileActiveRecoveryActions({ companyId: f.companyId }); + expect(calls).toHaveLength(2); + for (const call of calls) { + expect(call.input).toMatchObject({ allowRunCoalescing: false, + issueStateGuard: { statuses: expect.not.arrayContaining(["backlog", "done", "cancelled"]), assigneeAgentId: f.parentAgent } }); + } + }); + + it("defers while the agent takes no wakes, and retries a refused wake once the refusal is old", async () => { + const f = await fixture(); + await db.update(agents).set({ runtimeConfig: { heartbeat: { wakeOnDemand: false } } }).where(eq(agents.id, f.parentAgent)); + const { calls, enqueueWakeup } = wakeRecorder(); + const result = await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(result).toMatchObject({ woken: 0, deferred: 1 }); + expect(result.retryAt).toBeGreaterThan(Date.now() + 9 * 60_000); + expect(calls).toEqual([]); + + // Wakes enabled again; an earlier refusal row exists for this stall. + await db.update(agents).set({ runtimeConfig: {} }).where(eq(agents.id, f.parentAgent)); + const key = issueBlockerStalledWakeKey({ waiterIssueId: f.parent.id, stalledIssueId: f.child.id, recoveryActionId: f.action.id }); + const [refused] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.parentAgent, source: "automation", + triggerDetail: "system", reason: "heartbeat.wakeOnDemand.disabled", payload: { issueId: f.parent.id }, idempotencyKey: key, + status: "skipped" }).returning(); + const refusedPass = await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toEqual([]); // a fresh refusal is not retried at once + expect(refusedPass.retryAt).toBe(refused.requestedAt.getTime() + 10 * 60_000); + await db.update(agentWakeupRequests).set({ requestedAt: new Date(Date.now() - 11 * 60_000) }).where(eq(agentWakeupRequests.id, refused.id)); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(1); + }); + + it("does not wake again while a resource-wait retry carries the stall", async () => { + const f = await fixture(); + const { calls, enqueueWakeup } = wakeRecorder(); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(1); + // A workspace-busy deferral cancels the wake and schedules a keyless retry + // that keeps the stall context. + await db.update(agentWakeupRequests).set({ status: "cancelled", requestedAt: new Date(Date.now() - 11 * 60_000) }) + .where(eq(agentWakeupRequests.companyId, f.companyId)); + const [retry] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.parentAgent, + invocationSource: "automation", status: "scheduled_retry", scheduledRetryAt: new Date(), + contextSnapshot: { ...calls[0]!.input.contextSnapshot, wakeReason: "workspace_busy_retry", retryOfRunId: randomUUID() } }).returning(); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(1); + // A retry chain that was itself cancelled, or failed before the agent ran, no longer counts. + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, retry.id)); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(2); + }); + + it("reaches a waiter through its own sweep when the stalled blocker is unassigned", async () => { + const f = await fixture(); + await db.update(issues).set({ parentId: null, assigneeAgentId: null }).where(eq(issues.id, f.child.id)); + await db.insert(issueRelations).values({ companyId: f.companyId, issueId: f.child.id, relatedIssueId: f.parent.id, type: "blocks" }); + const { calls, enqueueWakeup } = wakeRecorder(); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId, issueId: f.parent.id }); + expect(calls.map(call => call.agentId)).toEqual([f.parentAgent]); + }); + + it("makes progress past the per-pass cap", async () => { + const f = await fixture(); + const [second] = await db.insert(issues).values({ companyId: f.companyId, title: "Second child", status: "blocked", + parentId: f.parent.id, assigneeAgentId: f.childAgent }).returning(); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: second.id, kind: "run_recovery", + status: "escalated", ownerType: "board", cause: "c", fingerprint: "second", nextAction: "Retry" }); + const { calls, enqueueWakeup } = wakeRecorder(); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId, maxStalls: 1 }); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId, maxStalls: 1 }); + expect(new Set(calls.map(call => call.input.contextSnapshot.blockerStall.stalledIssueId))).toEqual(new Set([f.child.id, second.id])); + }); + + it("delivers one wake when two sweeps race for the same stall", async () => { + const f = await fixture(); + const { calls, enqueueWakeup } = wakeRecorder(); + const results = await Promise.all([ + wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }), + wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId, issueId: f.child.id }), + ]); + const rows = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.reason, ISSUE_BLOCKER_STALLED_WAKE_REASON)); + expect(rows).toHaveLength(1); + expect(results.reduce((total, result) => total + result.woken, 0)).toBe(1); + expect(calls.length).toBeGreaterThanOrEqual(1); + }); + + it("keeps a company with only a resolved-as-blocked stall in the recovery-action sweep", async () => { + const f = await fixture(); + await db.update(issueRecoveryActions).set({ status: "resolved", outcome: "blocked", resolvedAt: new Date() }) + .where(eq(issueRecoveryActions.id, f.action.id)); + const selected = async () => (await db.selectDistinct({ companyId: issueRecoveryActions.companyId }).from(issueRecoveryActions) + .where(recoveryActionSweepCondition())).map(row => row.companyId); + expect(await selected()).toContain(f.companyId); + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, f.child.id)); + expect(await selected()).not.toContain(f.companyId); + }); + + it("withholds a private stalled issue's details from a waiter agent that cannot read it", async () => { + const f = await fixture(); + await db.update(issues).set({ visibility: "private", title: "Secret acquisition" }).where(eq(issues.id, f.child.id)); + const { calls, enqueueWakeup } = wakeRecorder(); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(1); + const stall = calls[0]!.input.contextSnapshot.blockerStall; + expect(JSON.stringify(stall)).not.toContain("Secret acquisition"); + expect(JSON.stringify(stall)).not.toContain("execution_recovery_budget_exhausted"); + expect(stall).toMatchObject({ stalledIssueId: f.child.id, stalledIssueIdentifier: null }); + const [activity] = await db.select().from(activityLog).where(eq(activityLog.action, "issue.blocker_stalled_wake")); + expect(activity.details).toMatchObject({ detailsWithheld: true, stalledIssueIdentifier: null }); + }); + + it("ignores an action resolved any other way", async () => { + const f = await fixture(); + await db.update(issueRecoveryActions).set({ status: "resolved", outcome: "restored", resolvedAt: new Date() }) + .where(eq(issueRecoveryActions.id, f.action.id)); + const { calls, enqueueWakeup } = wakeRecorder(); + const service = recoveryService(db, { enqueueWakeup: enqueueWakeup as never }); + await service.reconcileResolvedDependencyWakeBackstop({ companyId: f.companyId, issueId: f.child.id }); + await service.reconcileActiveRecoveryActions({ companyId: f.companyId }); + expect(calls.filter(call => call.input.reason === ISSUE_BLOCKER_STALLED_WAKE_REASON)).toEqual([]); + }); + + it("does not wake for an agent-owned action, a terminal or unassigned waiter, or a hidden waiter", async () => { + const agentOwned = await fixture({ ownerType: "agent", status: "active" }); + const doneParent = await fixture(); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, doneParent.parent.id)); + const unassigned = await fixture(); + await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, unassigned.parent.id)); + const hidden = await fixture(); + await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, hidden.parent.id)); + const { calls, enqueueWakeup } = wakeRecorder(); + const service = recoveryService(db, { enqueueWakeup: enqueueWakeup as never }); + for (const f of [agentOwned, doneParent, unassigned, hidden]) await service.reconcileActiveRecoveryActions({ companyId: f.companyId }); + expect(calls.filter(call => call.input.reason === ISSUE_BLOCKER_STALLED_WAKE_REASON)).toEqual([]); + }); +}); diff --git a/server/src/__tests__/issue-recovery-work-lifecycle.test.ts b/server/src/__tests__/issue-recovery-work-lifecycle.test.ts index 52f465daa8e..50a8629fa25 100644 --- a/server/src/__tests__/issue-recovery-work-lifecycle.test.ts +++ b/server/src/__tests__/issue-recovery-work-lifecycle.test.ts @@ -123,6 +123,25 @@ describe("issue recovery obligation lifecycle", () => { expect(f.check.mock.calls.length).toBe(checked * 2); expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.coordinator.nextWakeAt()).toBeNull(); }); + it("retries a recovery-action pass whose blocker stall wake failed to enqueue", async () => { + const f = fixture(); f.rows.actions.push({ id: "selected", companyId: "selected" }); + f.recoveryActions.mockResolvedValueOnce({ blockerStallWakeFailures: 1 } as never); + await f.worker.ready; + expect(f.recoveryActions).toHaveBeenCalledTimes(1); + expect(f.coordinator.nextWakeAt()).toBe(1_030_000); + await vi.advanceTimersByTimeAsync(30_000); + expect(f.recoveryActions).toHaveBeenCalledTimes(2); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("comes back for a stall held while its agent takes no wakes", async () => { + const f = fixture(); f.rows.actions.push({ id: "selected", companyId: "selected" }); + f.recoveryActions.mockResolvedValueOnce({ blockerStallRetryAt: 1_600_000 } as never); + await f.worker.ready; + expect(f.coordinator.nextWakeAt()).toBe(1_600_000); + await vi.advanceTimersByTimeAsync(600_000); + expect(f.recoveryActions).toHaveBeenCalledTimes(2); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); it("keeps an incomplete watchdog subtree as concrete retry debt", async () => { const f = fixture(); f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" }); f.state.incomplete = true; await f.worker.ready; expect(f.coordinator.nextWakeAt()).toBe(1_030_000); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 0dda710dadf..3a4806de488 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -51,6 +51,7 @@ import { legacyDispositionEpisode, legacyDispositionFingerprint } from "../../.. import { appendHeartbeatRunEvent } from "../../../services/heartbeat-run-events.js"; import { emitAgentTaskRun } from "../../../services/agent-task-run-telemetry.js"; import { issueService } from "../../../services/issues.js"; +import { verifyBlockerStallWake } from "../../../services/recovery/blocker-stall-wakeups.js"; import { issueTreeControlService, isVerifiedIssueTreeControlInteractionWake, @@ -479,8 +480,16 @@ export function createPostgresRunDispatchAdapter( const dependencyReadiness = await issuesSvcForRead.listDependencyReadiness(input.companyId, [issueId]); const readiness = dependencyReadiness.get(issueId); + // A verified stall wake runs while its stalled dependency is unresolved, + // including after a resource wait replaced it with a scheduled retry. + const stallAdmitted = readiness && !readiness.isDependencyReady && input.contextSnapshot.blockerStall + ? await verifyBlockerStallWake(db, { companyId: input.companyId, issueId, contextSnapshot: input.contextSnapshot, + requestedByActorType: input.wakeupRequestId ? await db.select({ actor: agentWakeupRequests.requestedByActorType }) + .from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, input.companyId), + eq(agentWakeupRequests.id, input.wakeupRequestId))).then(rows => rows[0]?.actor ?? null) : null }) + : false; facts.dependenciesBlocked = - readiness && !readiness.isDependencyReady + readiness && !readiness.isDependencyReady && !stallAdmitted ? { unresolvedBlockerIssueIds: readiness.unresolvedBlockerIssueIds, unresolvedBlockerCount: readiness.unresolvedBlockerCount, @@ -646,9 +655,13 @@ export function createPostgresRunDispatchAdapter( const retryReasonKind = classifyRetryReasonKind(retryReason); // Dependency edges can change after scheduled promotion without changing // the displayed status. Read them again under the queued/final issue lock. - const readiness = issue && retryReasonKind === "native_safe_replacement" + const replacementReadiness = issue && retryReasonKind === "native_safe_replacement" ? (await issueService(dbOrTx).listDependencyReadiness(input.companyId, [issueId])).get(issueId) : null; + const readiness = replacementReadiness && !replacementReadiness.isDependencyReady && context.blockerStall + && await verifyBlockerStallWake(dbOrTx as unknown as Db, { companyId: input.companyId, issueId, + contextSnapshot: context, requestedByActorType: null }) + ? null : replacementReadiness; return { runId: input.runId, runAgentId: input.agentId, diff --git a/server/src/modules/run-dispatch/domain/wake-context.ts b/server/src/modules/run-dispatch/domain/wake-context.ts index 91545b4bdb3..645d10f78ae 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.ts @@ -168,6 +168,23 @@ export function allowsIssueInteractionWake( return Boolean(deriveCommentId(contextSnapshot)); } +/** + * Whether a wake may run while its issue still has unresolved blockers: a + * verified comment 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`); + * the wake reason string alone never qualifies. Pause holds are separate and + * do not use this rule. + */ +export function allowsDependencyBlockedWake( + contextSnapshot: Record | null | undefined, + interactionWakeReasons: ReadonlySet, + verifiedBlockerStall: boolean, +): boolean { + return allowsIssueInteractionWake(contextSnapshot, interactionWakeReasons) || verifiedBlockerStall; +} + export function isResolvedInteractionContinuationWakeContext(contextSnapshot: unknown): boolean { const context = parseObject(contextSnapshot); const interactionId = readNonEmptyString(context.interactionId); diff --git a/server/src/modules/run-dispatch/index.ts b/server/src/modules/run-dispatch/index.ts index 2cd18c72398..5f391cca374 100644 --- a/server/src/modules/run-dispatch/index.ts +++ b/server/src/modules/run-dispatch/index.ts @@ -30,6 +30,7 @@ export { extractWakeCommentIds, deriveCommentId, allowsIssueInteractionWake, + allowsDependencyBlockedWake, isResolvedInteractionContinuationWakeContext, } from "./domain/wake-context.js"; export type { diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 928d55caa0b..e3931c8e5bd 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -1,3 +1,4 @@ +import { listStalledDependencies } from "../services/recovery/blocker-stall-wakeups.js"; import { confirmMcpAttachmentCleanup, recordMcpAttachmentCleanup, type McpAttachmentCleanup } from "../services/public-mcp/upload-cleanup.js"; import { notifyDeferredExecutionWork } from "../services/deferred-execution-notifications.js"; import { notifyNativeRecoveryNotices } from "../services/native-recovery-notice-notifications.js"; @@ -1445,6 +1446,7 @@ const ISSUE_WAKE_DIAGNOSTIC_KNOWN_REASONS = new Set([ "chat_task_completed", "issue_assigned", "issue_blockers_resolved", + "issue_blocker_stalled", "issue_commented", "issue_comment_mentioned", "issue_dependencies_blocked", @@ -9830,7 +9832,7 @@ export function issueRoutes( timing.time("work_products", () => workProductsSvc.listForIssue(issue.id)), timing.time("created_from", () => resolveVisibleIssueCreationSource(req, issue)), ]); - const [recoveryActionsByRelationIssue, revalidatedActiveRecoveryAction, mentionedProjects] = await Promise.all([ + const [recoveryActionsByRelationIssue, revalidatedActiveRecoveryAction, mentionedProjects, stalledDependencies] = await Promise.all([ timing.time("relation_recovery", () => relationRecoveryActionMap(recoveryActionsSvc, issue.companyId, relations)), timing.time("revalidate_recovery", () => revalidateActiveSourceRecoveryForRead({ issue, @@ -9841,6 +9843,14 @@ export function issueRoutes( timing.time("mentioned_projects", () => mentionedProjectIds.length > 0 ? projectsSvc.listByIds(issue.companyId, mentionedProjectIds) : Promise.resolve([])), + // Children and blockers stalled on the board, shown on this page so the + // board can act without opening each one. Unreadable issues are omitted. + timing.time("stalled_dependencies", async () => { + const stalled = await listStalledDependencies(db, { companyId: issue.companyId, issueId: issue.id }); + if (stalled.length === 0) return stalled; + const readable = await readableIssueEdgeIds(req, issue.companyId, stalled.map(entry => entry.stalledIssueId)); + return stalled.filter(entry => readable.has(entry.stalledIssueId)); + }), ]); const relationsWithRecoveryActions = withRecoveryActionsOnRelationSummaries( relations, recoveryActionsByRelationIssue, @@ -9867,6 +9877,8 @@ export function issueRoutes( interruption: await taskInterruptionPresentation(db, issue), scheduledRetry, activeRecoveryAction: revalidatedActiveRecoveryAction, + // Always present, so a cleared stall replaces the cached list. + stalledDependencies, blockedBy: visibleRelations.blockedBy, blocks: visibleRelations.blocks, relatedWork: visibleReferenceSummary, diff --git a/server/src/services/agent-conversations.ts b/server/src/services/agent-conversations.ts index de7b9719b6c..e63bfe57f13 100644 --- a/server/src/services/agent-conversations.ts +++ b/server/src/services/agent-conversations.ts @@ -80,6 +80,7 @@ export function isConversationExecutionWake( return isConversation(issue) && ( reason === "issue_blockers_resolved" || reason === "issue_children_completed" || + reason === "issue_blocker_stalled" || reason === "issue_unblock_requested" ); } diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 5f3c2fe48d2..c17ae70b663 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { verifyBlockerStallWake } from "../recovery/blocker-stall-wakeups.js"; import { CONTEXT_COMPACTION_WAKE_REASON, ContextCompactionError, isContextCompactionRun, readContextCompactionRequest } from "../context-compaction.js"; import { isProcessRunAdmissionClosed } from "./run-admission.js"; import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js"; @@ -172,7 +173,7 @@ import { WORKSPACE_WORKTREE_REQUIRES_PROJECT_REMEDIATION, } from "../execution-workspace-policy.js"; import { readContinuationAttempt } from "../recovery/index.js"; -import { allowsIssueInteractionWake, wakeChatConversationsConflict } from "../../modules/run-dispatch/index.js"; +import { allowsDependencyBlockedWake, wakeChatConversationsConflict } from "../../modules/run-dispatch/index.js"; import { withAgentStartLock } from "../agent-start-lock.js"; import { evaluateAgentInvokability, @@ -1021,11 +1022,20 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ); const readiness = dependencyReadiness.get(issueId); const unresolvedBlockerCount = readiness?.unresolvedBlockerCount ?? 0; + // The run's wake request records who asked for it. + const blockerStallVerified = unresolvedBlockerCount > 0 && await verifyBlockerStallWake(db, { + companyId: run.companyId, issueId, contextSnapshot: context, + requestedByActorType: run.wakeupRequestId + ? (await db.select({ actor: agentWakeupRequests.requestedByActorType }).from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, run.wakeupRequestId)).then(rows => rows[0]?.actor ?? null)) + : null, + }); if ( unresolvedBlockerCount > 0 && - !allowsIssueInteractionWake( + !allowsDependencyBlockedWake( context, ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS, + blockerStallVerified, ) ) { await cancelQueuedRunForBlockedDependencies( @@ -3905,12 +3915,17 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // Blocked descendants should stay idle until the final blocker resolves. // Human comment/mention wakes are the exception: they may run in a // bounded interaction mode so the assignee can answer or triage. + // A stalled child or blocker (issue_blocker_stalled, once per stall) + // is the other exception: the assignee must be able to re-delegate, + // escalate or block with a reason while that dependency is stuck. const blockedInteractionWake = dependencyReadiness && !dependencyReadiness.isDependencyReady && - allowsIssueInteractionWake( + allowsDependencyBlockedWake( enrichedContextSnapshot, ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS, + await verifyBlockerStallWake(tx as unknown as Db, { companyId: issue.companyId, issueId: issue.id, + contextSnapshot: enrichedContextSnapshot, requestedByActorType: opts.requestedByActorType }), ); if (blockedInteractionWake) { diff --git a/server/src/services/heartbeat/run-preparation.ts b/server/src/services/heartbeat/run-preparation.ts index ab1338b01da..6c6fe418cab 100644 --- a/server/src/services/heartbeat/run-preparation.ts +++ b/server/src/services/heartbeat/run-preparation.ts @@ -178,6 +178,7 @@ const ISSUE_RESPONSIBLE_USER_WAKE_REASONS = new Set([ "issue_reopened_via_comment", "issue_blockers_resolved", "issue_children_completed", + "issue_blocker_stalled", "issue_status_changed", "issue_tree_restored", "issue_recovery_action_restored", @@ -2752,6 +2753,7 @@ export async function buildPaperclipWakePayload(input: { : [], childIssueSummaryTruncated: input.contextSnapshot.childIssueSummaryTruncated === true, + blockerStall: input.contextSnapshot.blockerStall ?? null, dispositionRepair: buildReviewPathRecoveryDisposition(input.contextSnapshot) ?? (input.contextSnapshot.legacyDispositionEpisode ? { diff --git a/server/src/services/issue-read-timing.ts b/server/src/services/issue-read-timing.ts index a3244e6f653..574196a85e8 100644 --- a/server/src/services/issue-read-timing.ts +++ b/server/src/services/issue-read-timing.ts @@ -6,7 +6,7 @@ type IssueReadPhase = "lookup" | "authorization" | "project_goal" | "ancestors" | "mentions" | "documents" | "relations" | "blockers" | "review" | "references" | "handoff" | "retry" | "recovery" | "cases" | "inbox" | "channel" | "workspace" | "work_products" | "created_from" | "execution_blocker" - | "relation_recovery" | "revalidate_recovery" | "mentioned_projects"; + | "relation_recovery" | "revalidate_recovery" | "mentioned_projects" | "stalled_dependencies"; export function createIssueReadTiming() { const start = performance.now(); diff --git a/server/src/services/recovery/blocker-stall-wakeups.ts b/server/src/services/recovery/blocker-stall-wakeups.ts new file mode 100644 index 00000000000..9b178731e57 --- /dev/null +++ b/server/src/services/recovery/blocker-stall-wakeups.ts @@ -0,0 +1,332 @@ +/** + * Wake the issues waiting on a stalled one (`issue_blocker_stalled`). + * + * A parent hears about a child only when the child completes + * (`issue_children_completed`), and a dependent hears about a blocker only when + * every blocker is done (`issue_blockers_resolved`). When a child or blocker + * instead stalls on a recovery action that needs the board (for example a run + * whose retries are exhausted), nothing told the waiting agent: restart chaos + * w-20261108T181313-s9701 t05 left a parent silently blocked for 50+ minutes. + * + * A stall is the issue's latest recovery action needing the board, on an + * issue that is not done or cancelled: + * - open (`active` or `escalated`) and board-owned or escalated; or + * - board-owned and resolved with outcome `blocked` while the issue is still + * `blocked`. Recovery resolves its action this way when it gives the issue + * to the board (t05: created and resolved within six seconds), so an open + * action alone misses the common case. + * Each open, agent-assigned waiter (the parent, and every issue the stalled + * one `blocks`) is woken once per stall. The idempotency key names the + * recovery action, so a repeated sweep never wakes twice, and a later stall + * (a new action after a Retry or another failure) wakes again. + */ +import { and, eq, gt, inArray, isNull, ne, notInArray, or, sql } from "drizzle-orm"; +import { agents, agentWakeupRequests, heartbeatRuns, issueRecoveryActions, issueRelations, issues, type Db } from "@paperclipai/db"; +import { ISSUE_BLOCKER_STALLED_WAKE_REASON, type IssueBlockerStalledWakeContext, type IssueBlockerStallRelation } from "@paperclipai/shared"; +import { logActivity } from "../activity-log.js"; +import { parseHeartbeatPolicy } from "../heartbeat/timer-policy.js"; +import { authorizationService } from "../authorization.js"; + +const TERMINAL_ISSUE_STATUSES = ["done", "cancelled"]; +/** Waiters that may be woken: not finished, and not parked in the backlog by the board. */ +const WAKEABLE_WAITER_STATUSES = ["todo", "in_progress", "in_review", "blocked"]; +const MAX_STALLS_PER_PASS = 200; +/** A wake row with one of these statuses was delivered or is still pending. */ +const DELIVERED_OR_PENDING_WAKE_STATUSES = ["queued", "deferred_issue_execution", "claimed", "coalesced", "completed"]; +/** Waiters of one stall read per page, and scanned per pass before a cursor resumes the rest. */ +const MAX_WAITERS_PER_STALL = 100; +const MAX_WAITERS_SCANNED_PER_STALL = 500; +const waiterCursors = new Map(); +/** Runs that still hold, or already delivered, a stall prompt. A failed setup never did. */ +const DELIVERING_RUN_STATUSES = ["queued", "scheduled_retry", "running", "succeeded"]; +/** A refused wake (skipped, failed, cancelled) is retried, but not more often than this. */ +const REFUSED_WAKE_RETRY_MS = 10 * 60_000; +const NON_INVOKABLE_AGENT_STATUSES = ["paused", "terminated", "pending_approval"]; +/** Company-wide passes resume after the last stall a capped pass reached, so every stall is visited. */ +const companyCursors = new Map(); + +export type BlockerStallWakeup = (agentId: string, input: { + source: "automation"; + triggerDetail: "system"; + reason: string; + payload: Record; + idempotencyKey: string; + requestedByActorType: "system"; + requestedByActorId: string; + contextSnapshot: Record; + allowRunCoalescing: false; + issueStateGuard: { statuses: string[]; assigneeAgentId: string }; +}) => Promise; + +export function issueBlockerStalledWakeKey(input: { waiterIssueId: string; stalledIssueId: string; recoveryActionId: string }) { + return [ISSUE_BLOCKER_STALLED_WAKE_REASON, input.waiterIssueId, input.stalledIssueId, input.recoveryActionId].join(":"); +} + +/** + * Recovery-action rows that keep a company in the recovery-action sweep: open + * actions, and a board action resolved as "blocked" on an issue that is still + * blocked (a stall with no open action left). Without the second case, an + * unassigned stalled blocker with a todo waiter is visited by no sweep. + */ +export function recoveryActionSweepCondition() { + return or( + inArray(issueRecoveryActions.status, ["active", "escalated"]), + and(eq(issueRecoveryActions.status, "resolved"), eq(issueRecoveryActions.outcome, "blocked"), + eq(issueRecoveryActions.ownerType, "board"), + sql`exists (select 1 from ${issues} stalled where stalled.id = ${issueRecoveryActions.sourceIssueId} + and stalled.company_id = ${issueRecoveryActions.companyId} and stalled.status = 'blocked')`), + )!; +} + +type StallFilter = { companyId?: string | null; issueId?: string | null; issueIds?: readonly string[]; afterActionId?: string | null }; + +/** The stalled issues matching `filter` and the recovery action each one waits on. */ +async function readStalls(db: Db, filter: StallFilter, limit = MAX_STALLS_PER_PASS) { + if (filter.issueIds && filter.issueIds.length === 0) return []; + return db.select({ + actionId: issueRecoveryActions.id, companyId: issueRecoveryActions.companyId, + kind: issueRecoveryActions.kind, cause: issueRecoveryActions.cause, status: issueRecoveryActions.status, + ownerType: issueRecoveryActions.ownerType, nextAction: issueRecoveryActions.nextAction, + issueId: issues.id, identifier: issues.identifier, title: issues.title, issueStatus: issues.status, parentId: issues.parentId, + projectId: issues.projectId, assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId, + }).from(issueRecoveryActions) + .innerJoin(issues, and(eq(issues.id, issueRecoveryActions.sourceIssueId), eq(issues.companyId, issueRecoveryActions.companyId))) + .where(and( + filter.companyId ? eq(issueRecoveryActions.companyId, filter.companyId) : undefined, + filter.issueId ? eq(issueRecoveryActions.sourceIssueId, filter.issueId) : undefined, + filter.issueIds ? inArray(issueRecoveryActions.sourceIssueId, [...filter.issueIds]) : undefined, + filter.afterActionId ? gt(issueRecoveryActions.id, filter.afterActionId) : undefined, + notInArray(issues.status, TERMINAL_ISSUE_STATUSES), + or( + and(inArray(issueRecoveryActions.status, ["active", "escalated"]), + or(eq(issueRecoveryActions.ownerType, "board"), eq(issueRecoveryActions.status, "escalated"))), + and(eq(issueRecoveryActions.status, "resolved"), eq(issueRecoveryActions.outcome, "blocked"), + eq(issueRecoveryActions.ownerType, "board"), eq(issues.status, "blocked")), + ), + // Only the issue's latest action describes its current state. + sql`not exists (select 1 from ${issueRecoveryActions} newer + where newer.company_id = ${issueRecoveryActions.companyId} + and newer.source_issue_id = ${issueRecoveryActions.sourceIssueId} + and newer.created_at > ${issueRecoveryActions.createdAt})`, + )) + .orderBy(issueRecoveryActions.id) + .limit(limit); +} + +type StallRow = Awaited>[number]; + +/** What a waiter's agent may see of a stalled issue it cannot read: that it is stalled, nothing more. */ +function redactedStallContext(stall: StallRow, relation: IssueBlockerStallRelation): IssueBlockerStalledWakeContext { + return { + stalledIssueId: stall.issueId, stalledIssueIdentifier: null, stalledIssueTitle: "A task you cannot read", + stalledIssueStatus: stall.issueStatus, relation, + recoveryAction: { id: stall.actionId, kind: "", cause: "", status: stall.status, ownerType: stall.ownerType, + nextAction: "Ask the board to act on it; its details are not available to you." }, + }; +} + +function stallContext(stall: StallRow, relation: IssueBlockerStallRelation): IssueBlockerStalledWakeContext { + return { + stalledIssueId: stall.issueId, + stalledIssueIdentifier: stall.identifier, + stalledIssueTitle: stall.title, + stalledIssueStatus: stall.issueStatus, + relation, + recoveryAction: { id: stall.actionId, kind: stall.kind, cause: stall.cause, status: stall.status, + ownerType: stall.ownerType, nextAction: stall.nextAction }, + }; +} + +/** + * The children and blockers of `issueId` that are stalled on the board, for + * the waiting issue's own page: the board sees them without opening each one. + */ +export async function listStalledDependencies(db: Db, input: { companyId: string; issueId: string }): Promise { + const children = await db.select({ id: issues.id }).from(issues) + .where(and(eq(issues.companyId, input.companyId), eq(issues.parentId, input.issueId), isNull(issues.hiddenAt))); + const blockers = await db.select({ id: issueRelations.issueId }).from(issueRelations) + .where(and(eq(issueRelations.companyId, input.companyId), eq(issueRelations.type, "blocks"), + eq(issueRelations.relatedIssueId, input.issueId))); + const childIds = new Set(children.map(row => row.id)); + const stalls = await readStalls(db, { companyId: input.companyId, issueIds: [...new Set([...childIds, ...blockers.map(row => row.id)])] }, 50); + return stalls.map(stall => stallContext(stall, childIds.has(stall.issueId) ? "child" : "blocker")); +} + +export async function wakeWaitersOfStalledIssues(db: Db, enqueueWakeup: BlockerStallWakeup, + opts: { companyId?: string | null; issueId?: string | null; maxStalls?: number } = {}) { + const result = { stalls: 0, woken: 0, alreadyWoken: 0, deferred: 0, failed: 0, retryAt: null as number | null }; + const retryAt = (at: number) => { result.retryAt = result.retryAt === null ? at : Math.min(result.retryAt, at); }; + const maxStalls = opts.maxStalls ?? MAX_STALLS_PER_PASS; + let stalled: StallRow[]; + if (opts.issueId) { + // The per-issue sweep checks the issue itself and its own children and + // blockers, so a stalled dependency that this sweep never visits (for + // example an unassigned blocker) still reaches the waiting issue. + const scope = opts.companyId ? [eq(issues.companyId, opts.companyId)] : []; + const children = await db.select({ id: issues.id }).from(issues).where(and(...scope, eq(issues.parentId, opts.issueId))); + const blockers = await db.select({ id: issueRelations.issueId }).from(issueRelations) + .where(and(eq(issueRelations.type, "blocks"), eq(issueRelations.relatedIssueId, opts.issueId), + opts.companyId ? eq(issueRelations.companyId, opts.companyId) : undefined)); + stalled = await readStalls(db, { companyId: opts.companyId, + issueIds: [...new Set([opts.issueId, ...children.map(row => row.id), ...blockers.map(row => row.id)])] }, maxStalls); + } else { + const cursorKey = opts.companyId ?? "*"; + stalled = await readStalls(db, { companyId: opts.companyId, afterActionId: companyCursors.get(cursorKey) ?? null }, maxStalls); + if (stalled.length >= maxStalls) companyCursors.set(cursorKey, stalled.at(-1)!.actionId); + else companyCursors.delete(cursorKey); + } + const access = authorizationService(db); + const agentCanRead = async (agentId: string, onBehalfOfUserId: string | null, stall: StallRow) => (await access.decide({ + actor: { type: "agent", agentId, companyId: stall.companyId, onBehalfOfUserId }, + action: "issue:read", + resource: { type: "issue", companyId: stall.companyId, issueId: stall.issueId, projectId: stall.projectId, + parentIssueId: stall.parentId, assigneeAgentId: stall.assigneeAgentId, assigneeUserId: stall.assigneeUserId, + status: stall.issueStatus }, + })).allowed; + const agentAcceptsWakes = new Map(); + const acceptsWakes = async (agentId: string) => { + if (!agentAcceptsWakes.has(agentId)) { + const [agent] = await db.select().from(agents).where(eq(agents.id, agentId)); + agentAcceptsWakes.set(agentId, Boolean(agent) && !NON_INVOKABLE_AGENT_STATUSES.includes(agent!.status) + && parseHeartbeatPolicy(agent!).wakeOnDemand); + } + return agentAcceptsWakes.get(agentId)!; + }; + + for (const stall of stalled) { + result.stalls += 1; + const cursorKey = `${stall.actionId}:${opts.issueId ?? "*"}`; + let after = waiterCursors.get(cursorKey) ?? null; + let scanned = 0; + for (;;) { + const waiters = await db.select({ id: issues.id, identifier: issues.identifier, assigneeAgentId: issues.assigneeAgentId, + status: issues.status, responsibleUserId: issues.responsibleUserId, + relation: sql<"child" | "blocker">`case when ${issues.id} = ${stall.parentId ?? null} then 'child' else 'blocker' end` }) + .from(issues) + .where(and( + eq(issues.companyId, stall.companyId), + ne(issues.id, stall.issueId), + inArray(issues.status, WAKEABLE_WAITER_STATUSES), + isNull(issues.hiddenAt), + sql`${issues.assigneeAgentId} is not null`, + // A waiter's own sweep wakes only that waiter for its children and + // blockers; the stalled issue's own pass covers all of its waiters. + opts.issueId && stall.issueId !== opts.issueId ? eq(issues.id, opts.issueId) : undefined, + after ? gt(issues.id, after) : undefined, + sql`not exists (select 1 from ${agentWakeupRequests} w where w.company_id = ${stall.companyId} + and w.idempotency_key = 'issue_blocker_stalled:' || ${issues.id}::text || ${`:${stall.issueId}:${stall.actionId}`} + and w.status in (${sql.join(DELIVERED_OR_PENDING_WAKE_STATUSES.map(status => sql`${status}`), sql`, `)}))`, + or( + stall.parentId ? eq(issues.id, stall.parentId) : sql`false`, + sql`exists (select 1 from ${issueRelations} r where r.company_id = ${stall.companyId} + and r.type = 'blocks' and r.issue_id = ${stall.issueId} and r.related_issue_id = ${issues.id})`, + ), + )).orderBy(issues.id).limit(MAX_WAITERS_PER_STALL); + for (const waiter of waiters) { + const idempotencyKey = issueBlockerStalledWakeKey({ waiterIssueId: waiter.id, stalledIssueId: stall.issueId, recoveryActionId: stall.actionId }); + // (company_id, payload->>'issueId') is indexed; the key narrows to this stall. + // A delivered or pending wake ends it. A refused one (wake-on-demand off, + // agent paused) is retried once the refusal is old, so re-enabling wakes + // still delivers the stall. + const [existing] = await db.select({ id: agentWakeupRequests.id, status: agentWakeupRequests.status, + requestedAt: agentWakeupRequests.requestedAt }).from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, stall.companyId), + sql`${agentWakeupRequests.payload}->>'issueId' = ${waiter.id}`, + eq(agentWakeupRequests.idempotencyKey, idempotencyKey), + or(inArray(agentWakeupRequests.status, DELIVERED_OR_PENDING_WAKE_STATUSES), + gt(agentWakeupRequests.requestedAt, new Date(Date.now() - REFUSED_WAKE_RETRY_MS))))).limit(1); + if (existing) { + result.alreadyWoken += 1; + // A recent refusal is retried once it is old; the worker comes back then. + if (!(DELIVERED_OR_PENDING_WAKE_STATUSES as readonly string[]).includes(existing.status)) + retryAt(existing.requestedAt.getTime() + REFUSED_WAKE_RETRY_MS); + continue; + } + // A resource wait cancels the stall wake's run and schedules a retry + // with no idempotency key; the retry keeps the stall in its context. + const [carried] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, stall.companyId), eq(heartbeatRuns.agentId, waiter.assigneeAgentId!), + inArray(heartbeatRuns.status, DELIVERING_RUN_STATUSES), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${waiter.id}`, + sql`${heartbeatRuns.contextSnapshot}#>>'{blockerStall,stalledIssueId}' = ${stall.issueId}`, + sql`${heartbeatRuns.contextSnapshot}#>>'{blockerStall,recoveryAction,id}' = ${stall.actionId}`)).limit(1); + if (carried) { result.alreadyWoken += 1; continue; } + if (!(await acceptsWakes(waiter.assigneeAgentId!))) { + result.deferred += 1; + retryAt(Date.now() + REFUSED_WAKE_RETRY_MS); + continue; + } + // A private stalled issue's details never reach an agent that cannot read it. + const readable = await agentCanRead(waiter.assigneeAgentId!, waiter.responsibleUserId, stall); + const context = readable ? stallContext(stall, waiter.relation) : redactedStallContext(stall, waiter.relation); + try { + const wake = await enqueueWakeup(waiter.assigneeAgentId!, { + source: "automation", triggerDetail: "system", reason: ISSUE_BLOCKER_STALLED_WAKE_REASON, + payload: { issueId: waiter.id, ...context }, idempotencyKey, + requestedByActorType: "system", requestedByActorId: "issue-blocker-stall", + contextSnapshot: { issueId: waiter.id, taskId: waiter.id, wakeReason: ISSUE_BLOCKER_STALLED_WAKE_REASON, + source: "issue.blocker_stalled", blockerStall: context }, + // One stall per run: coalescing would keep only the last blockerStall + // context while marking every merged stall as delivered. + allowRunCoalescing: false, + issueStateGuard: { statuses: WAKEABLE_WAITER_STATUSES, assigneeAgentId: waiter.assigneeAgentId! }, + }); + // Null is a normal deferral (for example wake-on-demand disabled): no + // wake row exists yet, so a later sweep tries again. + if (!wake) { result.deferred += 1; continue; } + await logActivity(db, { + companyId: stall.companyId, actorType: "system", actorId: "issue-blocker-stall", + action: "issue.blocker_stalled_wake", entityType: "issue", entityId: waiter.id, + agentId: waiter.assigneeAgentId, issueId: waiter.id, + details: { stalledIssueId: stall.issueId, stalledIssueIdentifier: readable ? stall.identifier : null, + relation: waiter.relation, recoveryActionId: stall.actionId, + recoveryActionKind: readable ? stall.kind : null, recoveryActionStatus: stall.status, detailsWithheld: !readable }, + }); + result.woken += 1; + } catch { + // The next sweep retries: no wake row with this key exists yet. + result.failed += 1; + } + } + scanned += waiters.length; + if (waiters.length < MAX_WAITERS_PER_STALL) { waiterCursors.delete(cursorKey); break; } + after = waiters.at(-1)!.id; + // Skipped waiters (agents taking no wakes) never pin later ones: the next pass resumes here. + if (scanned >= MAX_WAITERS_SCANNED_PER_STALL) { waiterCursors.set(cursorKey, after); retryAt(Date.now()); break; } + } + } + return result; +} + +/** + * Whether a wake's `issue_blocker_stalled` context is real, so it may run past + * the waiter's unresolved blockers. The reason string alone is not trusted: + * the wake must come from the system, name a recovery action that is the + * stalled issue's current stall, and that issue must be a child or blocker of + * the waking issue. Anything else falls back to the ordinary dependency gate. + */ +export async function verifyBlockerStallWake(db: Db, input: { + companyId: string; + issueId: string; + contextSnapshot: Record | null | undefined; + requestedByActorType: string | null | undefined; +}): Promise { + const context = input.contextSnapshot; + // A resource wait (AI connection busy, workspace busy) replaces the run with + // a server-created retry that keeps the context but not the wake reason or + // request. It carries the stall that admitted the original run. + const serverRetry = typeof context?.retryOfRunId === "string" && context.retryOfRunId.length > 0; + if (!context || (context.wakeReason !== ISSUE_BLOCKER_STALLED_WAKE_REASON && !serverRetry)) return false; + if (!serverRetry && input.requestedByActorType !== "system") return false; + const stall = context.blockerStall as Partial | undefined; + const stalledIssueId = typeof stall?.stalledIssueId === "string" ? stall.stalledIssueId : null; + const actionId = typeof stall?.recoveryAction?.id === "string" ? stall.recoveryAction.id : null; + if (!stalledIssueId || !actionId || stalledIssueId === input.issueId) return false; + const [current] = await readStalls(db, { companyId: input.companyId, issueIds: [stalledIssueId] }, 1); + if (!current || current.actionId !== actionId) return false; + if (current.parentId === input.issueId) return true; + const [edge] = await db.select({ id: issueRelations.id }).from(issueRelations) + .where(and(eq(issueRelations.companyId, input.companyId), eq(issueRelations.type, "blocks"), + eq(issueRelations.issueId, stalledIssueId), eq(issueRelations.relatedIssueId, input.issueId))).limit(1); + return Boolean(edge); +} diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 16c9654b8c6..c96598f047e 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,3 +1,4 @@ +import { wakeWaitersOfStalledIssues, type BlockerStallWakeup } from "./blocker-stall-wakeups.js"; import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { OWNER_UNPROVABLE_CAUSE } from "./owner-gone-attestation.js"; import { promoteDeferredWorkIntents } from "../work-intent-outbox.js"; @@ -245,6 +246,7 @@ type RecoveryWakeupOptions = { statuses: string[]; assigneeAgentId: string; }; + allowRunCoalescing?: boolean; }; type RecoveryWakeup = ( @@ -261,6 +263,7 @@ type ResolvedDependencyWakeBackstopOptions = { blockerIssueId?: string | null; issueId?: string; source?: ResolvedDependencyWakeBackstopSource; + onDeferredDeadline?: (at: number) => void; }; const recoveryRunFields = { @@ -3635,7 +3638,11 @@ export function recoveryService( // longer schedules another agent-owned wake for them. result.skipped += 1; } - return result; + // A child or blocker stalled on a board-owned action: tell whoever waits on it, once per stall. + const stalls = await wakeWaitersOfStalledIssues(db, deps.enqueueWakeup as BlockerStallWakeup, { companyId: opts?.companyId }); + // A failed delivery is reported so the recovery worker retries this pass. + return { ...result, blockerStallWakes: stalls.woken, blockerStallWakeFailures: stalls.failed, + blockerStallRetryAt: stalls.retryAt }; } async function escalateDispositionRepair(input: { @@ -6076,6 +6083,16 @@ export function recoveryService( ); } + // A stalled child or blocker: tell whoever waits on it, once per stall + // (blocker-stall-wakeups.ts). This sweep visits each blocked issue, which + // covers a board action already resolved as "blocked". + if (!opts?.runId && !opts?.blockerIssueId) { + const stalls = await wakeWaitersOfStalledIssues(db, deps.enqueueWakeup as BlockerStallWakeup, + { companyId: opts?.companyId ?? null, issueId: opts?.issueId ?? null }); + // Counted with the dependency wakes, so the worker retries a failed one. + result.enqueueFailed += stalls.failed; + if (stalls.retryAt !== null) opts?.onDeferredDeadline?.(stalls.retryAt); + } return result; } diff --git a/server/src/services/recovery/work.ts b/server/src/services/recovery/work.ts index e039b13757c..dec58947ca1 100644 --- a/server/src/services/recovery/work.ts +++ b/server/src/services/recovery/work.ts @@ -5,6 +5,7 @@ import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator import { DELIVERY_QUEUES, type DeliveryQueue } from "../delivery-work-notifications.js"; import type { IssueRecoveryScope } from "../issue-recovery-notifications.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { recoveryActionSweepCondition } from "./blocker-stall-wakeups.js"; type Entry = { id: string; companyId: string; dueAt: number }; type Scope = { companyId: string; issueId: string; onDeferredDeadline: (at: number) => void }; @@ -16,7 +17,7 @@ export function registerIssueRecoveryWork( owner: { stranded(scope: Scope): Promise; orphanBlockers(scope: { companyId: string }): Promise; - recoveryActions(scope: { companyId: string }): Promise; + recoveryActions(scope: { companyId: string }): Promise<{ blockerStallWakeFailures?: number; blockerStallRetryAt?: number | null } | unknown>; dependencies(scope: Scope): Promise<{ enqueueFailed: number }>; locks(scope: Scope): Promise; watchdogs(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>; @@ -59,7 +60,7 @@ export function registerIssueRecoveryWork( }, }, canRun); } - function registerCompany(queue: DeliveryQueue, kind: "blockers" | "actions", check: (scope: { companyId: string }) => Promise) { + function registerCompany(queue: DeliveryQueue, kind: "blockers" | "actions", check: (scope: { companyId: string }) => Promise) { const cursors = new Map(); let enabled = false; return registerScopedDeadlineWork(coordinator, { @@ -74,7 +75,7 @@ export function registerIssueRecoveryWork( const cursor = cursors.get(key); const rows = kind === "actions" ? await db.selectDistinct({ id: issueRecoveryActions.companyId, companyId: issueRecoveryActions.companyId }) - .from(issueRecoveryActions).where(and(inArray(issueRecoveryActions.status, ["active", "escalated"]), + .from(issueRecoveryActions).where(and(recoveryActionSweepCondition(), scope ? eq(issueRecoveryActions.companyId, scope.companyId) : undefined, cursor ? gt(issueRecoveryActions.companyId, cursor) : undefined)) .orderBy(asc(issueRecoveryActions.companyId)).limit(PAGE_SIZE) @@ -92,12 +93,25 @@ export function registerIssueRecoveryWork( if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete }; }, - check: async entry => { if (enabled) await check({ companyId: entry.companyId }); return null; }, + check: async entry => { + if (!enabled) return null; + const next = await check({ companyId: entry.companyId }); + return typeof next === "number" ? { ...entry, dueAt: next } : null; + }, }, canRun); } const assignments = register(DELIVERY_QUEUES.strandedAssignments, ["todo", "in_progress", "in_review"], owner.stranded); - const orphanBlockers = registerCompany(DELIVERY_QUEUES.orphanBlockers, "blockers", owner.orphanBlockers); - const recoveryActions = registerCompany(DELIVERY_QUEUES.activeRecoveryActions, "actions", owner.recoveryActions); + const orphanBlockers = registerCompany(DELIVERY_QUEUES.orphanBlockers, "blockers", async scope => { await owner.orphanBlockers(scope); }); + const recoveryActions = registerCompany(DELIVERY_QUEUES.activeRecoveryActions, "actions", async scope => { + const result = await owner.recoveryActions(scope); + // A stall wake that failed to enqueue must be retried, not dropped with the entry. + const failures = result && typeof result === "object" && "blockerStallWakeFailures" in result + ? Number(result.blockerStallWakeFailures) : 0; + if (failures > 0) throw new Error("Blocker stall wake remains undelivered"); + // A stall held back because the agent takes no wakes is checked again later. + const retryAt = result && typeof result === "object" && "blockerStallRetryAt" in result ? result.blockerStallRetryAt : null; + return typeof retryAt === "number" ? retryAt : null; + }); const dependencies = register(DELIVERY_QUEUES.resolvedDependencies, ["blocked"], async scope => { const result = await owner.dependencies(scope); if (result.enqueueFailed > 0) throw new Error("Resolved dependency wake remains undelivered"); diff --git a/ui/src/components/IssueChatThread.tsx b/ui/src/components/IssueChatThread.tsx index 274e7c3fea0..13a1d44a56d 100644 --- a/ui/src/components/IssueChatThread.tsx +++ b/ui/src/components/IssueChatThread.tsx @@ -48,6 +48,7 @@ import type { IssueAttachment, IssueDocumentSummary, IssueBlockerAttention, + IssueBlockerStalledWakeContext, IssueRecoveryAction, IssueQueuedCommentQueue, IssueRelationIssueSummary, @@ -267,6 +268,7 @@ import { X, } from "lucide-react"; import { IssueBlockedNotice } from "./IssueBlockedNotice"; +import { StalledDependencyNotice } from "./StalledDependencyNotice"; import { IssueAssignedBacklogNotice } from "./IssueAssignedBacklogNotice"; import { IssueRecoveryActionCard, @@ -585,6 +587,8 @@ interface IssueChatThreadProps { /** Company-wide set of issue ids with a live (queued/running) run. */ liveIssueIds?: ReadonlySet; blockerAttention?: IssueBlockerAttention | null; + /** Children and blockers stalled on the board (`stalledDependencies` on the issue). */ + stalledDependencies?: IssueBlockerStalledWakeContext[]; successfulRunHandoff?: SuccessfulRunHandoffState | null; scheduledRetry?: IssueScheduledRetry | null; recoveryAction?: IssueRecoveryAction | null; @@ -5874,6 +5878,7 @@ export function IssueChatThread({ blockedBy = [], liveIssueIds, blockerAttention = null, + stalledDependencies, successfulRunHandoff = null, scheduledRetry = null, recoveryAction = null, @@ -6848,6 +6853,7 @@ export function IssueChatThread({ : null } /> + vi.fn()); +const retryTask = vi.hoisted(() => vi.fn()); + +vi.mock("@/lib/router", () => ({ + Link: ({ children, to, ...props }: AnchorHTMLAttributes & { to: string }) => ( + {children} + ), +})); +vi.mock("../api/issues", () => ({ issuesApi: { get: getIssue, retryTask } })); + +// eslint-disable-next-line @typescript-eslint/no-explicit-any +(globalThis as any).IS_REACT_ACT_ENVIRONMENT = true; + +const stall = (overrides: Partial = {}): IssueBlockerStalledWakeContext => ({ + stalledIssueId: "child-1", + stalledIssueIdentifier: "CHA-8", + stalledIssueTitle: "Ship the migration", + stalledIssueStatus: "blocked", + relation: "child", + recoveryAction: { id: "action-1", kind: "active_run_watchdog", cause: "native_session_retry_exhausted", + status: "resolved", ownerType: "board", nextAction: "Automatic recovery stopped. Recorded work is preserved." }, + ...overrides, +}); + +let root: ReturnType | null = null; +let container: HTMLDivElement | null = null; +afterEach(() => { + act(() => root?.unmount()); + container?.remove(); + root = null; container = null; + vi.clearAllMocks(); +}); + +async function render(stalled: IssueBlockerStalledWakeContext[] | undefined) { + container = document.createElement("div"); + document.body.appendChild(container); + root = createRoot(container); + const client = new QueryClient({ defaultOptions: { queries: { retry: false } } }); + await act(async () => { + root!.render(); + }); + await act(async () => { await Promise.resolve(); }); + return container; +} + +async function waitFor(read: () => T | null): Promise { + for (let attempt = 0; attempt < 50; attempt += 1) { + const value = read(); + if (value) return value; + await act(async () => { await new Promise(resolve => setTimeout(resolve, 10)); }); + } + return read(); +} + +describe("StalledDependencyNotice", () => { + it("renders nothing without stalled dependencies", async () => { + const node = await render([]); + expect(node.querySelector("[data-testid='stalled-dependency-notice']")).toBeNull(); + expect(getIssue).not.toHaveBeenCalled(); + }); + + it("names the stalled child, its board action and a link, and offers the child's own Retry", async () => { + getIssue.mockResolvedValue({ id: "child-1", identifier: "CHA-8", status: "blocked", interruption: { canRetry: true, observedRunId: "run-1" } }); + retryTask.mockResolvedValue({ accepted: true, presentation: null }); + const node = await render([stall()]); + const notice = node.querySelector("[data-testid='stalled-dependency-notice']")!; + expect(notice.textContent).toContain("A dependency needs the board"); + expect(notice.textContent).toContain("CHA-8"); + expect(notice.textContent).toContain("Ship the migration"); + expect(notice.textContent).toContain("Automatic recovery stopped"); + expect(notice.querySelector("a")?.getAttribute("href")).toContain("CHA-8"); + const button = await waitFor(() => notice.querySelector("button[aria-label='Retry CHA-8']")); + expect(button).not.toBeNull(); + await act(async () => { button!.click(); }); + expect(retryTask).toHaveBeenCalledWith("child-1", expect.objectContaining({ observedRunId: "run-1" })); + }); + + it("explains a refused Retry and lets the board try again", async () => { + getIssue.mockResolvedValue({ id: "child-1", identifier: "CHA-8", status: "blocked", interruption: { canRetry: true, observedRunId: "run-1" } }); + retryTask.mockRejectedValue(new ApiError("Retry is not allowed while the run is still stopping", 409, {})); + const node = await render([stall()]); + const button = await waitFor(() => node.querySelector("button[aria-label='Retry CHA-8']")); + await act(async () => { button!.click(); }); + const alert = await waitFor(() => node.querySelector("[role='alert']")); + expect(alert?.textContent).toBeTruthy(); + expect(node.querySelector("button[aria-label='Retry CHA-8']")?.disabled).toBe(false); + }); + + it("shows a lost response as taking longer, with Retry still available", async () => { + getIssue.mockResolvedValue({ id: "child-1", identifier: "CHA-8", status: "blocked", interruption: { canRetry: true, observedRunId: "run-1" } }); + retryTask.mockRejectedValue(new Error("network down")); + const node = await render([stall()]); + const button = await waitFor(() => node.querySelector("button[aria-label='Retry CHA-8']")); + await act(async () => { button!.click(); }); + const status = await waitFor(() => [...node.querySelectorAll("[role='status']")].find(element => element.textContent?.includes("Taking longer")) ?? null); + expect(status).toBeTruthy(); + expect(node.querySelector("button[aria-label='Retry CHA-8']")?.disabled).toBe(false); + }); + + it("says when the child's state cannot be read and offers a reload", async () => { + getIssue.mockRejectedValueOnce(new Error("offline")).mockResolvedValue({ id: "child-1", identifier: "CHA-8", status: "blocked", + interruption: { canRetry: true, observedRunId: "run-1" } }); + const node = await render([stall()]); + const alert = await waitFor(() => node.querySelector("[role='alert']")); + expect(alert?.textContent).toContain("Could not load this task's state"); + const reload = [...node.querySelectorAll("button")].find(button => button.textContent === "Reload")!; + await act(async () => { reload.click(); }); + expect(await waitFor(() => node.querySelector("button[aria-label='Retry CHA-8']"))).not.toBeNull(); + }); + + it("drops a stall once the child's refreshed state has moved on", async () => { + getIssue.mockResolvedValueOnce({ id: "child-1", identifier: "CHA-8", status: "blocked", interruption: { canRetry: true, observedRunId: "run-1" } }) + .mockResolvedValue({ id: "child-1", identifier: "CHA-8", status: "in_progress", interruption: { canRetry: false } }); + retryTask.mockResolvedValue({ accepted: true, presentation: { canRetry: false } }); + const node = await render([stall()]); + const button = await waitFor(() => node.querySelector("button[aria-label='Retry CHA-8']")); + await act(async () => { button!.click(); }); + expect(await waitFor(() => node.querySelector("[data-testid='stalled-dependency-notice']") ? null : true)).toBe(true); + }); + + it("omits Retry when the child's task state does not offer it", async () => { + getIssue.mockResolvedValue({ id: "child-1", identifier: "CHA-8", status: "blocked", interruption: { canRetry: false } }); + const node = await render([stall({ relation: "blocker" })]); + await waitFor(() => (getIssue.mock.calls.length > 0 ? true : null)); + await act(async () => { await new Promise(resolve => setTimeout(resolve, 20)); }); + expect(node.textContent).toContain("Blocker"); + expect(node.querySelector("button")).toBeNull(); + }); +}); diff --git a/ui/src/components/StalledDependencyNotice.tsx b/ui/src/components/StalledDependencyNotice.tsx new file mode 100644 index 00000000000..26ee0484c65 --- /dev/null +++ b/ui/src/components/StalledDependencyNotice.tsx @@ -0,0 +1,107 @@ +import type { Issue, IssueBlockerStalledWakeContext } from "@paperclipai/shared"; +import { useQueries } from "@tanstack/react-query"; +import { AlertTriangle, RotateCcw } from "lucide-react"; +import { Link } from "@/lib/router"; +import { Button } from "@/components/ui/button"; +import { issuesApi } from "../api/issues"; +import { queryKeys } from "../lib/queryKeys"; +import { createIssueDetailPath } from "../lib/issueDetailBreadcrumb"; +import { useTaskRetry } from "../hooks/useTaskRetry"; +import { describeError } from "../api/errors"; + +/** + * Children and blockers of this task that stopped on a recovery action that + * needs the board (for example, retries exhausted). The task's agent is woken + * once per stall (`issue_blocker_stalled`); this lets the board see and act on + * each one from here, with the child's own Retry, without opening it. + */ +export function StalledDependencyNotice({ stalled }: { stalled: IssueBlockerStalledWakeContext[] | null | undefined }) { + // Most tasks have none: render nothing and read nothing. + if (!stalled || stalled.length === 0) return null; + return ; +} + +function StalledDependencyList({ entries }: { entries: IssueBlockerStalledWakeContext[] }) { + // The child's own task state decides whether Retry is offered, as on its page. + const children = useQueries({ + queries: entries.map(entry => { + const ref = entry.stalledIssueIdentifier ?? entry.stalledIssueId; + return { queryKey: queryKeys.issues.detail(ref), queryFn: () => issuesApi.get(ref) }; + }), + }); + // The waiter's saved list refreshes only with the waiter. Once a child's own + // refreshed state has moved on (its Retry started it, or it was unblocked), + // drop it here rather than show a cleared stall. + const rows = entries.map((entry, index) => ({ entry, child: children[index] })) + .filter(({ entry, child }) => !child?.data || child.data.status === entry.stalledIssueStatus); + if (rows.length === 0) return null; + return ( +
+
+ + {rows.length === 1 ? "A dependency needs the board" : `${rows.length} dependencies need the board`} +
+
    + {rows.map(({ entry, child }) => void child?.refetch()} />)} +
+
+ ); +} + +function StalledDependencyRow({ entry, child, readError, isFetching, refetch }: { + entry: IssueBlockerStalledWakeContext; child: Issue | undefined; readError: Error | null; isFetching: boolean; refetch: () => void; +}) { + const ref = entry.stalledIssueIdentifier ?? entry.stalledIssueId; + const retry = useTaskRetry(child); + const canRetry = child?.interruption?.canRetry === true; + // Same states as the task's own Retry (ExecutionBlockerNotice): a refused + // request explains why, and a lost response can be retried. + const failed = Boolean(retry.error) && !retry.optimistic && !retry.takingLonger; + return ( +
  • + + {entry.relation === "child" ? "Child" : "Blocker"} + + + {entry.stalledIssueIdentifier ?? "Task"} + + {entry.stalledIssueTitle} + {entry.recoveryAction.nextAction ? ( + {entry.recoveryAction.nextAction} + ) : null} + {readError && !child ? ( + + Could not load this task's state: {describeError(readError).body} + + + ) : null} + {failed ? ( + {describeError(retry.error).body} + ) : null} + {retry.takingLonger ? ( + Taking longer than usual… + ) : null} + {canRetry || failed || retry.takingLonger ? ( + + ) : null} +
    • + ); +} diff --git a/ui/src/pages/IssueDetail.tsx b/ui/src/pages/IssueDetail.tsx index cb01c98e642..d7161c31b98 100644 --- a/ui/src/pages/IssueDetail.tsx +++ b/ui/src/pages/IssueDetail.tsx @@ -1207,6 +1207,7 @@ type IssueDetailChatTabProps = { blockedBy: Issue["blockedBy"]; liveIssueIds: ReadonlySet; blockerAttention: Issue["blockerAttention"] | null; + stalledDependencies?: Issue["stalledDependencies"]; successfulRunHandoff: Issue["successfulRunHandoff"] | null; scheduledRetry: Issue["scheduledRetry"] | null; recoveryAction: Issue["activeRecoveryAction"]; @@ -1373,6 +1374,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ blockedBy, liveIssueIds, blockerAttention, + stalledDependencies, successfulRunHandoff, scheduledRetry, recoveryAction, @@ -2688,6 +2690,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ blockedBy={blockedBy ?? []} liveIssueIds={liveIssueIds} blockerAttention={blockerAttention} + stalledDependencies={stalledDependencies} successfulRunHandoff={successfulRunHandoff} scheduledRetry={scheduledRetry} recoveryAction={isPreviousServerStopCause(recoveryAction?.cause) ? null : recoveryAction ?? null} @@ -8562,6 +8565,7 @@ export function TaskDetailSurface({ conversation, tasksTab }: { tasksTab?: TaskS blockedBy={issue.blockedBy ?? []} liveIssueIds={liveIssueIds} blockerAttention={issue.blockerAttention ?? null} + stalledDependencies={issue.stalledDependencies} successfulRunHandoff={issue.successfulRunHandoff ?? null} scheduledRetry={issue.scheduledRetry ?? null} recoveryAction={issue.interruption?.responsibility === "system" || isPreviousServerStopCause(issue.activeRecoveryAction?.cause) ? null : issue.activeRecoveryAction ?? null}