diff --git a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts index a054564853..d61aed3d69 100644 --- a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts +++ b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts @@ -1754,6 +1754,172 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { } }, 120_000); + it("promotes an interaction continuation after removing a coalesced self-authored comment", async () => { + const gateway = await createControlledGatewayServer(); + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const interactionId = randomUUID(); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + const heartbeat = heartbeatService(db); + + try { + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + defaultResponsibleUserId: "responsible-user", + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Local CLI Agent", + role: "engineer", + status: "idle", + adapterType: "openclaw_gateway", + adapterConfig: { + url: gateway.url, + headers: { + "x-openclaw-token": "gateway-token", + }, + payloadTemplate: { + message: "wake now", + }, + waitTimeoutMs: 2_000, + }, + runtimeConfig: {}, + permissions: {}, + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Interaction continuation survives self-comment filtering", + status: "todo", + priority: "medium", + responsibleUserId: "responsible-user", + assigneeAgentId: agentId, + issueNumber: 1, + identifier: `${issuePrefix}-1`, + }); + + const firstRun = await heartbeat.wakeup(agentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { issueId }, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_assigned", + }, + requestedByActorType: "system", + requestedByActorId: null, + }); + + expect(firstRun).not.toBeNull(); + await waitFor(async () => { + const run = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, firstRun!.id)) + .then((rows) => rows[0] ?? null); + return run?.status === "running"; + }); + + const selfComment = await db + .insert(issueComments) + .values({ + companyId, + issueId, + authorUserId: "local-cli-user", + createdByRunId: firstRun!.id, + body: "Completion note from the source run", + }) + .returning() + .then((rows) => rows[0]); + + expect(await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + payload: { issueId, commentId: selfComment.id }, + contextSnapshot: { + issueId, + taskId: issueId, + commentId: selfComment.id, + wakeCommentId: selfComment.id, + wakeReason: "issue_commented", + }, + requestedByActorType: "user", + requestedByActorId: "local-cli-user", + })).toBeNull(); + + expect(await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + payload: { + issueId, + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + mutation: "interaction", + }, + contextSnapshot: { + issueId, + taskId: issueId, + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + wakeReason: "issue_commented", + source: "issue.interaction.respond", + }, + requestedByActorType: "user", + requestedByActorId: "user-1", + })).toBeNull(); + + gateway.releaseFirstWait(); + + await waitFor(async () => { + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)) + .orderBy(asc(heartbeatRuns.createdAt)); + return ( + runs.length === 2 && + runs[0]?.status === "succeeded" && + runs[1]?.status === "succeeded" + ); + }, 90_000); + + const promotedRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)) + .orderBy(asc(heartbeatRuns.createdAt)) + .then((runs) => runs[1] ?? null); + expect(promotedRun?.contextSnapshot).toMatchObject({ + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }); + expect(promotedRun?.contextSnapshot).not.toMatchObject({ + wakeCommentIds: expect.anything(), + }); + expect(promotedRun?.contextSnapshot).not.toMatchObject({ + commentId: selfComment.id, + }); + expect(gateway.getAgentPayloads()).toHaveLength(2); + } finally { + gateway.releaseFirstWait(); + await gateway.close(); + } + }, 120_000); + it("queues exactly one follow-up run when an issue-bound run exits without a comment", async () => { const gateway = await createControlledGatewayServer(); const companyId = randomUUID(); diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index f49ec7cdb4..0fbc086385 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -171,11 +171,11 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { const adapter = createPostgresWakeQueueAdapter(db, stubDeps); const result = await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async (locked, ports) => { - const candidate = await ports.writer.claimNextDeferredWake({ companyId, issueId: locked.primaryIssue.id }); + const candidate = await ports.transaction.findNextDeferredWake({ companyId, issueId: locked.primaryIssue.id }); expect(candidate?.id).toBe(wakeId); - const agent = await ports.reader.findInvokableAgent({ companyId, agentId: foreignAgentId }); + const agent = await ports.transaction.findInvokableAgent({ companyId, agentId: foreignAgentId }); expect(agent).toBeNull(); - const failed = await ports.writer.failDeferredWake({ companyId, wakeId: candidate!.id, now: new Date() }); + const failed = await ports.transaction.failDeferredWake({ companyId, wakeId: candidate!.id, now: new Date() }); expect(failed).toBe(true); return { outcome: { kind: "released" as const }, postCommitEffects: [] }; }); @@ -205,7 +205,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { await adapter.withIssueExecutionLock( { companyId, runId, now: new Date() }, async (_locked, ports) => { - const cancelledUnderWrongCompany = await ports.writer.cancelDeferredWake({ + const cancelledUnderWrongCompany = await ports.transaction.cancelDeferredWake({ companyId: otherCompanyId, wakeId, reason: "cross-company cancel attempt", @@ -213,14 +213,14 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { }); expect(cancelledUnderWrongCompany).toBe(false); - const failedUnderWrongCompany = await ports.writer.failDeferredWake({ + const failedUnderWrongCompany = await ports.transaction.failDeferredWake({ companyId: otherCompanyId, wakeId, now: new Date(), }); expect(failedUnderWrongCompany).toBe(false); - const normalizedUnderWrongCompany = await ports.writer.normalizeDeferredWakeCommentIds({ + const normalizedUnderWrongCompany = await ports.transaction.normalizeDeferredWakeCommentIds({ companyId: otherCompanyId, wakeId, payload: { issueId }, @@ -229,7 +229,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { }); expect(normalizedUnderWrongCompany).toBeNull(); - const reopenedUnderWrongCompany = await ports.writer.reopenIssue({ + const reopenedUnderWrongCompany = await ports.transaction.reopenIssue({ companyId: otherCompanyId, issueId, runId, @@ -264,7 +264,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { reopened: null, }; await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async (_locked, ports) => { - captured.reopened = await ports.writer.reopenIssue({ companyId: otherCompanyId, issueId, runId }); + captured.reopened = await ports.transaction.reopenIssue({ companyId: otherCompanyId, issueId, runId }); return { outcome: { kind: "released" as const }, postCommitEffects: [] }; }); expect(captured.reopened).toBeNull(); @@ -286,7 +286,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { reopened: null, }; await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async (_locked, ports) => { - captured.reopened = await ports.writer.reopenIssue({ companyId, issueId, runId }); + captured.reopened = await ports.transaction.reopenIssue({ companyId, issueId, runId }); return { outcome: { kind: "released" as const }, postCommitEffects: [] }; }); expect(captured.reopened?.status).toBe("todo"); @@ -310,7 +310,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { const adapter = createPostgresWakeQueueAdapter(db, stubDeps); const runId = await seedRun({ companyId, agentId, contextSnapshot: { issueId }, status: "succeeded" }); const result = await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async (_locked, ports) => { - const claimed = await ports.writer.claimDeferredWakeForPromotion({ companyId, wakeId, now: new Date() }); + const claimed = await ports.transaction.claimDeferredWakeForPromotion({ companyId, wakeId, now: new Date() }); expect(claimed).toBe(false); return { outcome: { kind: "released" as const }, postCommitEffects: [] }; }); @@ -345,7 +345,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { const adapter = createPostgresWakeQueueAdapter(db, stubDeps); await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async (locked, ports) => { const finalize = async (wakeId: string) => { - const promoted = await ports.writer.finalizePromotedWake({ + const promoted = await ports.transaction.finalizePromotedWake({ companyId, wakeId, deferredAgent, diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index f3fd525818..fd9e434535 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -11,7 +11,7 @@ import { nativeRunFinalizations, } from "@paperclipai/db"; import { legacyExecutionNeedsReconciliation } from "../../../services/legacy-execution-recovery.js"; -import { evaluateAgentInvokability } from "../../../services/agent-invokability.js"; +import { evaluateAgentInvokabilityFromDb } from "../../../services/agent-invokability.js"; import { issueTreeControlService, isVerifiedIssueTreeControlInteractionWake } from "../../../services/issue-tree-control.js"; import { isAutomaticRecoverySuppressedByPauseHold } from "../../../services/recovery/pause-hold-guard.js"; import { issueService } from "../../../services/issues.js"; @@ -19,18 +19,20 @@ import { issueRecoveryActionService } from "../../../services/issue-recovery-act import { readContinuationAttempt } from "../../../services/recovery/run-liveness-continuations.js"; import { withRecoveryContext } from "../../../services/recovery/status-only-context.js"; import { parseIssueExecutionState } from "../../../services/issue-execution-policy.js"; -import { - buildConfigurationIncompleteRecoveryNoticeSeed, - buildExecutionReviewParticipantRecoveryNoticeSeed, - buildImmediateExecutionPathRecoveryNoticeSeed, - buildWorkspaceValidationRecoveryNoticeSeed, -} from "../../../services/recovery/stranded-notice.js"; import { queuedCommentIdsFromWakePayload, withQueuedCommentIdsInWakePayload, } from "../../../services/issue-queued-comment-queue.js"; import { extractWakeCommentIds } from "../../run-dispatch/index.js"; import { hasInteractionContinuationWakeContext } from "../domain/context.js"; +import { decidePreDrain, type PreDrainFacts } from "../domain/policy.js"; +import { + EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON, + isConfigurationIncompleteFailedRun, + isWorkspaceValidationFailedRun, + parseObject, + readNonEmptyString, +} from "../domain/values.js"; import type { DeferredWakeCandidate, InvokableAgentSnapshot, @@ -39,48 +41,28 @@ import type { LockedIssueExecution, ReleaseTransactionResult, RunSnapshot, - WakeQueueReader, - WakeQueueWriter, + WakeQueueHost, + WakeQueueTransaction, } from "../application/ports.js"; import type { RunSummary } from "../application/types.js"; -import { WakeQueueApplicationError } from "../application/types.js"; const DEFERRED_WAKE_STATUS = "deferred_issue_execution"; const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext"; -const WORKSPACE_VALIDATION_FAILURE_CODE = "workspace_validation_failed"; -const CONFIGURATION_INCOMPLETE_FAILURE_CODE = "configuration_incomplete"; -const WORKSPACE_VALIDATION_RECOVERY_CAUSE = "workspace_validation_failed"; -const CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE = "configuration_incomplete"; -const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE = "execution_review_participant_recovery"; -const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON = "execution_review_participant_recovery"; -const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON = "execution_review_participant_recovery"; const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; type HeartbeatRunRow = typeof heartbeatRuns.$inferSelect; type IssueRow = typeof issues.$inferSelect; -function parseObject(value: unknown): Record { - return typeof value === "object" && value !== null && !Array.isArray(value) - ? (value as Record) - : {}; -} - -function readNonEmptyString(value: unknown): string | null { - return typeof value === "string" && value.trim().length > 0 ? value : null; -} - function normalizeAgentNameKey(value: string | null | undefined): string | null { if (typeof value !== "string") return null; const normalized = value.trim().toLowerCase(); return normalized.length > 0 ? normalized : null; } -function isWorkspaceValidationFailedRun(run: Pick): boolean { - return run.errorCode === WORKSPACE_VALIDATION_FAILURE_CODE; -} - -function isConfigurationIncompleteFailedRun(run: Pick): boolean { - return run.errorCode === CONFIGURATION_INCOMPLETE_FAILURE_CODE || run.errorCode === "model_not_found"; +function toRequestedByActorType(value: string | null): "user" | "agent" | "system" | null { + // The database column is free text; map any value outside the union to + // null instead of widening the type back to string. + return value === "user" || value === "agent" || value === "system" ? value : null; } function toRunSnapshot(row: HeartbeatRunRow): RunSnapshot { @@ -152,7 +134,7 @@ function toDeferredWakeCandidate(row: typeof agentWakeupRequests.$inferSelect): reason: row.reason, source: row.source, triggerDetail: row.triggerDetail, - requestedByActorType: row.requestedByActorType, + requestedByActorType: toRequestedByActorType(row.requestedByActorType), requestedByActorId: row.requestedByActorId, payload, queuedCommentIds, @@ -164,12 +146,23 @@ function toDeferredWakeCandidate(row: typeof agentWakeupRequests.$inferSelect): } export type WakeQueuePostgresAdapterDeps = { - resolveResponsibleUserId: WakeQueueReader["resolveResponsibleUserId"]; - getRoutineEnv: WakeQueueReader["getRoutineEnv"]; - resolveSessionBeforeForWakeup: WakeQueueReader["resolveSessionBeforeForWakeup"]; + resolveResponsibleUserId: WakeQueueHost["resolveResponsibleUserId"]; + getRoutineEnv: WakeQueueHost["getRoutineEnv"]; + resolveSessionBeforeForWakeup: WakeQueueHost["resolveSessionBeforeForWakeup"]; }; -function buildReader(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueReader { +function buildHost(_tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueHost { + return { + resolveResponsibleUserId: deps.resolveResponsibleUserId, + getRoutineEnv: deps.getRoutineEnv, + resolveSessionBeforeForWakeup: deps.resolveSessionBeforeForWakeup, + }; +} + +function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueTransaction { + const treeControlSvc = issueTreeControlService(tx); + const issuesSvc = issueService(tx); + return { async findInvokableAgent({ companyId, agentId }): Promise { const agent = await tx @@ -178,25 +171,11 @@ function buildReader(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueReade .where(and(eq(agents.id, agentId), eq(agents.companyId, companyId))) .then((rows) => rows[0] ?? null); if (!agent) return null; - const companyAgents = await tx - .select({ id: agents.id, companyId: agents.companyId, name: agents.name, reportsTo: agents.reportsTo, status: agents.status }) - .from(agents) - .where(eq(agents.companyId, companyId)); - const invokability = evaluateAgentInvokability(agent, companyAgents); + const invokability = await evaluateAgentInvokabilityFromDb(tx, agent); return { id: agent.id, companyId: agent.companyId, name: agent.name, invokable: invokability.invokable }; }, - resolveResponsibleUserId: deps.resolveResponsibleUserId, - getRoutineEnv: deps.getRoutineEnv, - resolveSessionBeforeForWakeup: deps.resolveSessionBeforeForWakeup, - }; -} -function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWriter { - const treeControlSvc = issueTreeControlService(tx); - const issuesSvc = issueService(tx); - - return { - async claimNextDeferredWake({ companyId, issueId }) { + async findNextDeferredWake({ companyId, issueId }) { const row = await tx .select() .from(agentWakeupRequests) @@ -446,25 +425,6 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite return isAutomaticRecoverySuppressedByPauseHold(tx, companyId, issueId, treeControlSvc); }, - async buildBlockedRecoveryNotice({ noticeKind, issueStatus, finishingRun }) { - if (noticeKind === "workspace_validation") { - return { notice: buildWorkspaceValidationRecoveryNoticeSeed(), recoveryCause: WORKSPACE_VALIDATION_RECOVERY_CAUSE }; - } - if (noticeKind === "configuration_incomplete") { - return { - notice: buildConfigurationIncompleteRecoveryNoticeSeed(finishingRun.configurationIncompletePayload), - recoveryCause: CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE, - }; - } - if (noticeKind === "execution_review_participant") { - return { - notice: buildExecutionReviewParticipantRecoveryNoticeSeed(), - recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE, - }; - } - return { notice: buildImmediateExecutionPathRecoveryNoticeSeed({ status: issueStatus }), recoveryCause: null }; - }, - async queueReviewParticipantRecoveryRun({ companyId, issue, finishingRun, recoveryAgent, sessionBefore, now }) { const executionState = parseIssueExecutionState(issue.executionState); const wakeupRequest = await tx @@ -474,7 +434,7 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite agentId: recoveryAgent.id, source: "automation", triggerDetail: "system", - reason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON, + reason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON, payload: withRecoveryContext( { issueId: issue.id, @@ -506,7 +466,7 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite { issueId: issue.id, taskId: issue.id, - wakeReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON, + wakeReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON, retryReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON, source: "issue.execution_review_recovery", retryOfRunId: finishingRun.id, @@ -542,48 +502,17 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite return toRunSummary(queuedRun); }, - async queueImmediateRecoveryRun({ companyId, issue, finishingRun, recoveryAgent, sessionBefore, now }) { - const retryReason = issue.status === "todo" ? "assignment_recovery" : "issue_continuation_needed"; - const recoveryReason = issue.status === "todo" ? "issue_assignment_recovery" : "issue_continuation_needed"; - const recoverySource = issue.status === "todo" ? "issue.assignment_recovery" : "issue.continuation_recovery"; - const recoveryContextSnapshot = withRecoveryContext( - { - issueId: issue.id, - taskId: issue.id, - wakeReason: recoveryReason, - retryReason, - source: recoverySource, - retryOfRunId: finishingRun.id, - }, - "normal_model", - ); - - const routineEnvContext = await deps.getRoutineEnv({ companyId, issue }); - const responsibleUserId = await deps.resolveResponsibleUserId({ - companyId, - contextSnapshot: recoveryContextSnapshot, - issue, - routineEnvContext, - requestedByActorType: "system", - requestedByActorId: null, - source: "automation", - triggerDetail: "system", - existingRunResponsibleUserId: finishingRun.responsibleUserId, - }); - if (!responsibleUserId) { - throw new WakeQueueApplicationError( - "responsible_user_unresolved", - "Unable to resolve responsible user for recovery heartbeat run", - { - runId: finishingRun.id, - agentId: recoveryAgent.id, - companyId, - issueId: issue.id, - wakeReason: recoveryReason, - }, - ); - } - + async queueImmediateRecoveryRun({ + companyId, + issue, + finishingRun, + recoveryAgent, + reason, + contextSnapshot, + responsibleUserId, + sessionBefore, + now, + }) { const wakeupRequest = await tx .insert(agentWakeupRequests) .values({ @@ -591,7 +520,7 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite agentId: recoveryAgent.id, source: "automation", triggerDetail: "system", - reason: recoveryReason, + reason, payload: withRecoveryContext({ issueId: issue.id, retryOfRunId: finishingRun.id }, "normal_model"), status: "queued", requestedByActorType: "system", @@ -610,7 +539,7 @@ function buildWriter(tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueWrite triggerDetail: "system", status: "queued", wakeupRequestId: wakeupRequest.id, - contextSnapshot: recoveryContextSnapshot, + contextSnapshot, responsibleUserId, sessionIdBefore: sessionBefore, retryOfRunId: finishingRun.id, @@ -767,49 +696,51 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd const issueRow = (contextIssueId ? candidateIssues.find((candidate) => candidate.id === contextIssueId) : candidateIssues[0]) ?? null; - if (!issueRow || (issueRow.executionRunId && issueRow.executionRunId !== run.id)) { + const preDrainFacts: PreDrainFacts = { + issueRowPresent: issueRow !== null, + executionRunIdMatchesRun: !issueRow || !issueRow.executionRunId || issueRow.executionRunId === run.id, + isWorkspaceValidationFailedRun: isWorkspaceValidationFailedRun(run), + isConfigurationIncompleteFailedRun: isConfigurationIncompleteFailedRun(run), + issueStatus: issueRow?.status ?? "", + hasAssigneeUser: Boolean(issueRow?.assigneeUserId), + assigneeAgentMatchesRunAgent: issueRow?.assigneeAgentId === run.agentId, + legacyExecutionNeedsReconciliation: legacyExecutionNeedsReconciliation(run), + // An operator stop never promotes old queued work by itself. The + // next explicit wake adopts those messages atomically when it + // queues a run. + executionCancellationAcknowledged: + run.status === "cancelled" && parseObject(run.resultJson?.executionCancellation).state === "acknowledged", + }; + const preDrain = decidePreDrain(preDrainFacts); + + if (preDrain.kind === "released") { return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; } - if ( - (isWorkspaceValidationFailedRun(run) || isConfigurationIncompleteFailedRun(run)) && - (issueRow.status === "todo" || issueRow.status === "in_progress") && - !issueRow.assigneeUserId && - issueRow.assigneeAgentId === run.agentId - ) { - const configurationIncomplete = isConfigurationIncompleteFailedRun(run); - const notice = configurationIncomplete - ? buildConfigurationIncompleteRecoveryNoticeSeed(runSnapshot.configurationIncompletePayload) - : buildWorkspaceValidationRecoveryNoticeSeed(); + // decidePreDrain only returns "blocked" or "proceed" when the issue row is present. + if (!issueRow) { + throw new Error(`wake-queue: pre-drain decision ${preDrain.kind} reached without an issue row`); + } + + if (preDrain.kind === "blocked") { return { outcome: { kind: "blocked", issue: toIssueSnapshot(issueRow), previousStatus: issueRow.status as "todo" | "in_progress", - notice, - recoveryCause: configurationIncomplete ? CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE : WORKSPACE_VALIDATION_RECOVERY_CAUSE, + noticeKind: preDrain.noticeKind, }, postCommitEffects: [], run: runSnapshot, }; } - if (legacyExecutionNeedsReconciliation(run)) { - return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; - } - - // An operator stop never promotes old queued work by itself. The next - // explicit wake adopts those messages atomically when it queues a run. - if (run.status === "cancelled" && parseObject(run.resultJson?.executionCancellation).state === "acknowledged") { - return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; - } - if (await recordNativeTerminalRecoveryIfNeeded(tx, run, issueRow, input.now)) { return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; } const locked: LockedIssueExecution = { primaryIssue: toIssueSnapshot(issueRow), run: runSnapshot }; - const result = await fn(locked, { reader: buildReader(tx, deps), writer: buildWriter(tx, deps) }); + const result = await fn(locked, { host: buildHost(tx, deps), transaction: buildTransaction(tx, deps) }); return { ...result, run: runSnapshot }; }); }, diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index bce32babfd..24c82d5624 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -8,7 +8,7 @@ import type { RunSummary, } from "./types.js"; -export type { InvokableAgentSnapshot, IssueSnapshot, RunSnapshot, RunSummary }; +export type { InvokableAgentSnapshot, IssueSnapshot, ReleaseRecoveryBlockedNoticeKind, RunSnapshot, RunSummary }; /** The primary issue a locked release resolves to, plus the finishing run the lock step already loaded. */ export type LockedIssueExecution = { @@ -21,9 +21,12 @@ export type ReleaseTransactionResult = { postCommitEffects: PostCommitEffect[]; }; -/** Read-only lookups the release use case needs, each scoped to a company. */ -export interface WakeQueueReader { - findInvokableAgent(input: { companyId: string; agentId: string }): Promise; +/** + * The three host callbacks the release use case needs. These members do + * not run on the module's own transaction, which is why two of them take + * the transaction-scoped issue snapshot instead of an issue id. + */ +export interface WakeQueueHost { /** * Takes the transaction-scoped issue snapshot, not an issue id, so this * port never re-reads the issue on a separate connection while the @@ -64,7 +67,7 @@ export type DeferredWakeCandidate = { reason: string | null; source: string | null; triggerDetail: string | null; - requestedByActorType: string | null; + requestedByActorType: "user" | "agent" | "system" | null; requestedByActorId: string | null; payload: Record; /** The queued comment ids the wake's queued-comment context carries, already extracted from the payload. */ @@ -94,9 +97,14 @@ export type PromoteDeferredWakeInput = { now: Date; }; -/** The transaction-scoped write operations that drain and resolve the deferred-wake queue. */ -export interface WakeQueueWriter { - claimNextDeferredWake(input: { companyId: string; issueId: string }): Promise; +/** + * Every member is bound to the one transaction that `withIssueExecutionLock` + * owns. The interface holds both reads and writes that drain and resolve + * the deferred-wake queue. + */ +export interface WakeQueueTransaction { + findInvokableAgent(input: { companyId: string; agentId: string }): Promise; + findNextDeferredWake(input: { companyId: string; issueId: string }): Promise; getQueuedCommentLiveness(input: { companyId: string; issueId: string; @@ -115,7 +123,7 @@ export interface WakeQueueWriter { normalizeDeferredWakeCommentIds(input: { companyId: string; wakeId: string; - /** The wake's current payload, as already read by `claimNextDeferredWake`, used as the rewrite base. */ + /** The wake's current payload, as already read by `findNextDeferredWake`, used as the rewrite base. */ payload: Record; liveCommentIds: string[]; now: Date; @@ -169,12 +177,6 @@ export interface WakeQueueWriter { /** An open, non-hidden issue that still lists this issue as a `blocks` predecessor. */ hasExplicitBlockerPath(input: { companyId: string; issueId: string }): Promise; isAutomaticRecoverySuppressedByPauseHold(input: { companyId: string; issueId: string }): Promise; - /** Builds the stranded-recovery notice content for a `blocked` outcome; pure formatting, kept behind the writer so `services/recovery/stranded-notice` stays out of the application layer. */ - buildBlockedRecoveryNotice(input: { - noticeKind: ReleaseRecoveryBlockedNoticeKind; - issueStatus: "todo" | "in_progress"; - finishingRun: RunSnapshot; - }): Promise<{ notice: Record; recoveryCause: string | null }>; queueReviewParticipantRecoveryRun(input: { companyId: string; issue: IssueSnapshot; @@ -184,16 +186,18 @@ export interface WakeQueueWriter { now: Date; }): Promise; /** - * Builds the recovery context snapshot, resolves the responsible user - * from it, and queues the run. Throws `WakeQueueApplicationError` with - * code `responsible_user_unresolved` when no responsible user resolves, - * without queuing anything. + * Queues the run with the context snapshot and the responsible user the + * caller already resolved. */ queueImmediateRecoveryRun(input: { companyId: string; issue: IssueSnapshot; finishingRun: RunSnapshot; recoveryAgent: InvokableAgentSnapshot; + /** The wakeup request's reason and the run's context-snapshot wakeReason; the caller derives it from the issue status. */ + reason: string; + contextSnapshot: Record; + responsibleUserId: string; sessionBefore: string | null; now: Date; }): Promise; @@ -207,16 +211,15 @@ export interface WakeQueueWriter { * (workspace-validation block, legacy reconciliation, a native-runtime * terminal failure), the adapter returns that outcome directly without * calling `fn`. Otherwise it calls `fn` with the locked issue and run, and - * with `reader`/`writer` ports bound to the same transaction, so every - * call `fn` makes through them participates in the one transaction this - * method owns. + * with `host`/`transaction` ports, so every call `fn` makes through the + * transaction port participates in the one transaction this method owns. */ export interface IssueLockWriter { withIssueExecutionLock( input: { companyId: string; runId: string; now: Date }, fn: ( locked: LockedIssueExecution, - ports: { reader: WakeQueueReader; writer: WakeQueueWriter }, + ports: { host: WakeQueueHost; transaction: WakeQueueTransaction }, ) => Promise, ): Promise; } @@ -225,8 +228,7 @@ export type StrandedAssignedIssueEscalationInput = { issue: IssueSnapshot; previousStatus: "todo" | "in_progress" | "in_review"; latestRun: RunSnapshot; - notice: Record; - recoveryCause: string | null; + noticeKind: ReleaseRecoveryBlockedNoticeKind; }; export type StrandedRecoveryInPlaceEscalationInput = { diff --git a/server/src/modules/wake-queue/application/types.ts b/server/src/modules/wake-queue/application/types.ts index cb126d644d..90b9df83e4 100644 --- a/server/src/modules/wake-queue/application/types.ts +++ b/server/src/modules/wake-queue/application/types.ts @@ -1,3 +1,5 @@ +import type { ReleaseRecoveryBlockedNoticeKind } from "../domain/policy.js"; + export type RunSummary = { id: string; companyId: string; @@ -76,8 +78,7 @@ export type ReleaseOutcome = kind: "blocked"; issue: IssueSnapshot; previousStatus: "todo" | "in_progress" | "in_review"; - notice: Record; - recoveryCause: string | null; + noticeKind: ReleaseRecoveryBlockedNoticeKind; } | { kind: "blocked_recovery_in_place"; @@ -85,7 +86,7 @@ export type ReleaseOutcome = previousStatus: "todo" | "in_progress" | "in_review"; }; -export type WakeQueueApplicationErrorCode = "responsible_user_unresolved"; +export type WakeQueueApplicationErrorCode = "responsible_user_unresolved" | "deferred_wake_not_advanced"; export class WakeQueueApplicationError extends Error { constructor( diff --git a/server/src/modules/wake-queue/application/use-cases.test.ts b/server/src/modules/wake-queue/application/use-cases.test.ts index 81f0719e00..c6a7a7f677 100644 --- a/server/src/modules/wake-queue/application/use-cases.test.ts +++ b/server/src/modules/wake-queue/application/use-cases.test.ts @@ -10,8 +10,8 @@ import type { RecoveryEscalationPort, RunSnapshot, RunSummary, - WakeQueueReader, - WakeQueueWriter, + WakeQueueHost, + WakeQueueTransaction, } from "./ports.js"; const RUN: RunSnapshot = { @@ -29,7 +29,7 @@ const RUN: RunSnapshot = { const ISSUE: IssueSnapshot = { id: "issue-1", companyId: "company-1", - identifier: "PAP-1", + identifier: "ISSUE-1", status: "in_progress", assigneeAgentId: "finishing-agent", assigneeUserId: null, @@ -81,9 +81,8 @@ function runSummary(id: string): RunSummary { }; } -function createFakeReader(overrides: Partial = {}): WakeQueueReader { +function createFakeHost(overrides: Partial = {}): WakeQueueHost { return { - findInvokableAgent: vi.fn(async () => AGENT), resolveResponsibleUserId: vi.fn(async () => "user-1"), getRoutineEnv: vi.fn(async () => ({ routineId: null, env: null, responsibleUserId: null })), resolveSessionBeforeForWakeup: vi.fn(async () => null), @@ -91,9 +90,10 @@ function createFakeReader(overrides: Partial = {}): WakeQueueRe }; } -function createFakeWriter(overrides: Partial = {}): WakeQueueWriter { +function createFakeTransaction(overrides: Partial = {}): WakeQueueTransaction { return { - claimNextDeferredWake: vi.fn(async () => null), + findInvokableAgent: vi.fn(async () => AGENT), + findNextDeferredWake: vi.fn(async () => null), getQueuedCommentLiveness: vi.fn(async () => ({ liveNonSelfCommentIds: [], containedSelfAuthoredComment: false })), cancelDeferredWake: vi.fn(async () => true), normalizeDeferredWakeCommentIds: vi.fn(async (input) => wakeCandidate({ id: input.wakeId, queuedCommentIds: input.liveCommentIds })), @@ -114,17 +114,16 @@ function createFakeWriter(overrides: Partial = {}): WakeQueueWr hasExistingExecutionPath: vi.fn(async () => false), hasExplicitBlockerPath: vi.fn(async () => false), isAutomaticRecoverySuppressedByPauseHold: vi.fn(async () => false), - buildBlockedRecoveryNotice: vi.fn(async () => ({ notice: {}, recoveryCause: null })), queueReviewParticipantRecoveryRun: vi.fn(async () => runSummary("review-recovery")), queueImmediateRecoveryRun: vi.fn(async () => runSummary("immediate-recovery")), ...overrides, }; } -function createFakeIssueLock(reader: WakeQueueReader, writer: WakeQueueWriter): IssueLockWriter { +function createFakeIssueLock(host: WakeQueueHost, transaction: WakeQueueTransaction): IssueLockWriter { return { withIssueExecutionLock: vi.fn(async (_input, fn) => { - const result = await fn({ primaryIssue: ISSUE, run: RUN }, { reader, writer }); + const result = await fn({ primaryIssue: ISSUE, run: RUN }, { host, transaction }); return { ...result, run: RUN }; }), }; @@ -141,35 +140,36 @@ describe("releaseIssueExecution", () => { it("processes the deferred wakes in requestedAt order", async () => { const claimOrder: string[] = []; const queue = [wakeCandidate({ id: "wake-earliest" }), wakeCandidate({ id: "wake-latest" })]; - const writer = createFakeWriter({ - claimNextDeferredWake: vi.fn(async () => { + const transaction = createFakeTransaction({ + findNextDeferredWake: vi.fn(async () => { const next = queue.shift() ?? null; if (next) claimOrder.push(next.id); return next; }), + // Every wake fails invokability so the loop keeps draining without promoting. + findInvokableAgent: vi.fn(async () => null), }); - // Every wake fails invokability so the loop keeps draining without promoting. - const reader = createFakeReader({ findInvokableAgent: vi.fn(async () => null) }); - const issueLock = createFakeIssueLock(reader, writer); + const host = createFakeHost(); + const issueLock = createFakeIssueLock(host, transaction); const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery: createFakeRecovery() }); await releaseIssueExecution({ companyId: "company-1", runId: "run-1", now: new Date() }); expect(claimOrder).toEqual(["wake-earliest", "wake-latest"]); - expect(writer.failDeferredWake).toHaveBeenCalledTimes(2); + expect(transaction.failDeferredWake).toHaveBeenCalledTimes(2); }); it("stops the loop after the first promotion", async () => { - const claimNextDeferredWake = vi.fn(async () => wakeCandidate({ id: "wake-promotes" })); - const writer = createFakeWriter({ claimNextDeferredWake }); - const reader = createFakeReader(); - const issueLock = createFakeIssueLock(reader, writer); + const findNextDeferredWake = vi.fn(async () => wakeCandidate({ id: "wake-promotes" })); + const transaction = createFakeTransaction({ findNextDeferredWake }); + const host = createFakeHost(); + const issueLock = createFakeIssueLock(host, transaction); const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery: createFakeRecovery() }); const result = await releaseIssueExecution({ companyId: "company-1", runId: "run-1", now: new Date() }); expect(result.outcome.kind).toBe("promoted"); - expect(claimNextDeferredWake).toHaveBeenCalledTimes(1); + expect(findNextDeferredWake).toHaveBeenCalledTimes(1); }); it("continues the loop after a cancel outcome, a fail outcome, and a normalize outcome, then promotes", async () => { @@ -181,7 +181,7 @@ describe("releaseIssueExecution", () => { // normalize: queued comments differ from the live set, then promotes. wakeCandidate({ id: "wake-normalize", queuedCommentIds: ["c1", "c2"] }), ]; - const claimNextDeferredWake = vi.fn(async () => queue.shift() ?? null); + const findNextDeferredWake = vi.fn(async () => queue.shift() ?? null); const findInvokableAgent = vi.fn(async (input: { agentId: string }) => input.agentId === "deferred-agent" ? AGENT : null, ); @@ -190,24 +190,45 @@ describe("releaseIssueExecution", () => { ? { liveNonSelfCommentIds: [], containedSelfAuthoredComment: false } : { liveNonSelfCommentIds: ["c2"], containedSelfAuthoredComment: false }, ); - const writer = createFakeWriter({ claimNextDeferredWake, getQueuedCommentLiveness }); - const reader = createFakeReader({ findInvokableAgent }); - const issueLock = createFakeIssueLock(reader, writer); + const transaction = createFakeTransaction({ findNextDeferredWake, findInvokableAgent, getQueuedCommentLiveness }); + const host = createFakeHost(); + const issueLock = createFakeIssueLock(host, transaction); const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery: createFakeRecovery() }); const result = await releaseIssueExecution({ companyId: "company-1", runId: "run-1", now: new Date() }); - expect(writer.cancelDeferredWake).toHaveBeenCalledTimes(1); - expect(writer.failDeferredWake).toHaveBeenCalledTimes(1); - expect(writer.normalizeDeferredWakeCommentIds).toHaveBeenCalledTimes(1); - expect(claimNextDeferredWake).toHaveBeenCalledTimes(3); + expect(transaction.cancelDeferredWake).toHaveBeenCalledTimes(1); + expect(transaction.failDeferredWake).toHaveBeenCalledTimes(1); + expect(transaction.normalizeDeferredWakeCommentIds).toHaveBeenCalledTimes(1); + expect(findNextDeferredWake).toHaveBeenCalledTimes(3); expect(result.outcome.kind).toBe("promoted"); }); + it("rejects with deferred_wake_not_advanced when the queue read returns the same wake id twice, instead of looping forever", async () => { + // queuedCommentIds with no live comments and no independent continuation + // routes to "cancel_empty", so the drain calls cancelDeferredWake and + // discards its result, then reads the queue again for the same row. + const repeatedCandidate = wakeCandidate({ id: "wake-repeat", queuedCommentIds: ["c1"] }); + const findNextDeferredWake = vi.fn(async () => repeatedCandidate); + const cancelDeferredWake = vi.fn(async () => false); + const transaction = createFakeTransaction({ findNextDeferredWake, cancelDeferredWake }); + const host = createFakeHost(); + const issueLock = createFakeIssueLock(host, transaction); + const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery: createFakeRecovery() }); + + await expect( + releaseIssueExecution({ companyId: "company-1", runId: "run-1", now: new Date() }), + ).rejects.toMatchObject({ + constructor: WakeQueueApplicationError, + code: "deferred_wake_not_advanced", + }); + expect(findNextDeferredWake).toHaveBeenCalledTimes(2); + }); + it("returns the post-commit effects as data without running them", async () => { - const writer = createFakeWriter({ claimNextDeferredWake: vi.fn(async () => wakeCandidate()) }); - const reader = createFakeReader(); - const issueLock = createFakeIssueLock(reader, writer); + const transaction = createFakeTransaction({ findNextDeferredWake: vi.fn(async () => wakeCandidate()) }); + const host = createFakeHost(); + const issueLock = createFakeIssueLock(host, transaction); const recovery = createFakeRecovery(); const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery }); @@ -220,8 +241,8 @@ describe("releaseIssueExecution", () => { it("carries the deferred wake's raw issue, interaction, execution-stage, and accepted-plan context onto the promoted run, and clears only the rendered text projections", async () => { const finalizePromotedWake = vi.fn(async (input: PromoteDeferredWakeInput) => runSummary(input.wakeId)); - const writer = createFakeWriter({ - claimNextDeferredWake: vi.fn(async () => + const transaction = createFakeTransaction({ + findNextDeferredWake: vi.fn(async () => wakeCandidate({ deferredContextSeed: { issueId: ISSUE.id, @@ -239,8 +260,8 @@ describe("releaseIssueExecution", () => { ), finalizePromotedWake, }); - const reader = createFakeReader(); - const issueLock = createFakeIssueLock(reader, writer); + const host = createFakeHost(); + const issueLock = createFakeIssueLock(host, transaction); const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery: createFakeRecovery() }); const result = await releaseIssueExecution({ companyId: "company-1", runId: "run-1", now: new Date() }); @@ -275,14 +296,14 @@ describe("releaseIssueExecution", () => { }), wakeCandidate({ id: "wake-promotes" }), ]; - const claimNextDeferredWake = vi.fn(async () => queue.shift() ?? null); + const findNextDeferredWake = vi.fn(async () => queue.shift() ?? null); const claimDeferredWakeForPromotion = vi.fn(async ({ wakeId }: { wakeId: string }) => wakeId !== "wake-lost-race"); const reopenIssue = vi.fn(async () => null); - const writer = createFakeWriter({ claimNextDeferredWake, claimDeferredWakeForPromotion, reopenIssue }); - const reader = createFakeReader(); + const transaction = createFakeTransaction({ findNextDeferredWake, claimDeferredWakeForPromotion, reopenIssue }); + const host = createFakeHost(); const issueLock: IssueLockWriter = { withIssueExecutionLock: vi.fn(async (_input, fn) => { - const result = await fn({ primaryIssue: doneIssue, run: RUN }, { reader, writer }); + const result = await fn({ primaryIssue: doneIssue, run: RUN }, { host, transaction }); return { ...result, run: RUN }; }), }; @@ -297,9 +318,9 @@ describe("releaseIssueExecution", () => { }); it("throws WakeQueueApplicationError with code responsible_user_unresolved when the responsible user cannot resolve", async () => { - const writer = createFakeWriter({ claimNextDeferredWake: vi.fn(async () => wakeCandidate()) }); - const reader = createFakeReader({ resolveResponsibleUserId: vi.fn(async () => null) }); - const issueLock = createFakeIssueLock(reader, writer); + const transaction = createFakeTransaction({ findNextDeferredWake: vi.fn(async () => wakeCandidate()) }); + const host = createFakeHost({ resolveResponsibleUserId: vi.fn(async () => null) }); + const issueLock = createFakeIssueLock(host, transaction); const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery: createFakeRecovery() }); await expect( @@ -310,22 +331,77 @@ describe("releaseIssueExecution", () => { }); }); + it("resolves the responsible user for an immediate recovery run before queuing it", async () => { + const resolveResponsibleUserId = vi.fn( + async (_input: Parameters[0]) => "resolved-user", + ); + const queueImmediateRecoveryRun = vi.fn( + async (input: Parameters[0]) => runSummary("immediate-recovery"), + ); + const transaction = createFakeTransaction({ queueImmediateRecoveryRun }); + const host = createFakeHost({ resolveResponsibleUserId }); + const issueLock = createFakeIssueLock(host, transaction); + const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery: createFakeRecovery() }); + + const result = await releaseIssueExecution({ companyId: "company-1", runId: "run-1", now: new Date() }); + + expect(result.outcome.kind).toBe("queued_recovery"); + expect(resolveResponsibleUserId).toHaveBeenCalledTimes(1); + const resolveCall = resolveResponsibleUserId.mock.calls[0]![0]; + expect(resolveCall.requestedByActorType).toBe("system"); + expect(resolveCall.requestedByActorId).toBeNull(); + expect(resolveCall.source).toBe("automation"); + expect(resolveCall.triggerDetail).toBe("system"); + expect(resolveCall.existingRunResponsibleUserId).toBe(RUN.responsibleUserId); + // ISSUE.status is "in_progress", so the stalled-continuation labels apply. + expect(resolveCall.contextSnapshot).toEqual({ + issueId: ISSUE.id, + taskId: ISSUE.id, + wakeReason: "issue_continuation_needed", + retryReason: "issue_continuation_needed", + source: "issue.continuation_recovery", + retryOfRunId: RUN.id, + }); + + expect(queueImmediateRecoveryRun).toHaveBeenCalledTimes(1); + const queueCall = queueImmediateRecoveryRun.mock.calls[0]![0]; + expect(queueCall.reason).toBe("issue_continuation_needed"); + expect(queueCall.responsibleUserId).toBe("resolved-user"); + expect(queueCall.contextSnapshot).toBe(resolveCall.contextSnapshot); + }); + + it("throws WakeQueueApplicationError with code responsible_user_unresolved for a recovery run, without queuing it", async () => { + const transaction = createFakeTransaction(); + const host = createFakeHost({ resolveResponsibleUserId: vi.fn(async () => null) }); + const issueLock = createFakeIssueLock(host, transaction); + const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery: createFakeRecovery() }); + + await expect( + releaseIssueExecution({ companyId: "company-1", runId: "run-1", now: new Date() }), + ).rejects.toMatchObject({ + constructor: WakeQueueApplicationError, + code: "responsible_user_unresolved", + }); + expect(transaction.queueImmediateRecoveryRun).not.toHaveBeenCalled(); + }); + it("escalates through the recovery port for a blocked outcome, after the transaction resolves", async () => { - const writer = createFakeWriter({ - claimNextDeferredWake: vi.fn(async () => null), + const transaction = createFakeTransaction({ + findNextDeferredWake: vi.fn(async () => null), hasExistingExecutionPath: vi.fn(async () => false), isAutomaticRecoverySuppressedByPauseHold: vi.fn(async () => false), - buildBlockedRecoveryNotice: vi.fn(async () => ({ notice: { kind: "immediate_execution_path" }, recoveryCause: "immediate_execution_path" })), + // The recovery agent (the finishing run's own agent) is not invokable, which forces "blocked". + findInvokableAgent: vi.fn(async () => null), }); - // The recovery agent (the finishing run's own agent) is not invokable, which forces "blocked". - const reader = createFakeReader({ findInvokableAgent: vi.fn(async () => null) }); - const issueLock = createFakeIssueLock(reader, writer); + const host = createFakeHost(); + const issueLock = createFakeIssueLock(host, transaction); const recovery = createFakeRecovery(); const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery }); const result = await releaseIssueExecution({ companyId: "company-1", runId: "run-1", now: new Date() }); expect(result.outcome.kind).toBe("blocked"); + expect(result.outcome.kind === "blocked" && result.outcome.noticeKind).toBe("immediate_execution_path"); expect(recovery.escalateStrandedAssignedIssue).toHaveBeenCalledTimes(1); }); }); diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index 6a07e32498..355724391b 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -1,20 +1,33 @@ import { enrichPromotedWakeContext } from "../domain/context.js"; -import { decideDeferredWake, decideReleaseRecovery } from "../domain/policy.js"; +import { + decideQueuedCommentAction, + decideReleaseRecovery, + decideWakeOutcome, + deriveImmediateRecoveryContextLabels, +} from "../domain/policy.js"; +import { + EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON, + isConfigurationIncompleteFailedRun, + isWorkspaceValidationFailedRun, + readNonEmptyString, +} from "../domain/values.js"; +import { withRecoveryContext } from "../../../services/recovery/status-only-context.js"; import type { + DeferredWakeCandidate, + InvokableAgentSnapshot, IssueLockWriter, IssueSnapshot, LockedIssueExecution, RecoveryEscalationPort, ReleaseTransactionResult, RunSnapshot, - WakeQueueReader, - WakeQueueWriter, + WakeQueueHost, + WakeQueueTransaction, } from "./ports.js"; import type { PostCommitEffect, ReleaseOutcome } from "./types.js"; import { WakeQueueApplicationError } from "./types.js"; const ISSUE_DISPOSITION_REPAIR_RETRY_REASON = "issue_disposition_repair"; -const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON = "execution_review_participant_recovery"; const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASONS = new Set([ "execution_review_requested", "execution_approval_requested", @@ -31,20 +44,6 @@ const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = new Set([ "cancelled", ]); const STRANDED_ISSUE_RECOVERY_ORIGIN_KIND = "stranded_issue_recovery"; -const WORKSPACE_VALIDATION_FAILURE_CODE = "workspace_validation_failed"; -const CONFIGURATION_INCOMPLETE_FAILURE_CODE = "configuration_incomplete"; - -function readNonEmptyString(value: unknown): string | null { - return typeof value === "string" && value.trim().length > 0 ? value : null; -} - -function isWorkspaceValidationFailedRun(run: Pick): boolean { - return run.errorCode === WORKSPACE_VALIDATION_FAILURE_CODE; -} - -function isConfigurationIncompleteFailedRun(run: Pick): boolean { - return run.errorCode === CONFIGURATION_INCOMPLETE_FAILURE_CODE || run.errorCode === "model_not_found"; -} function isExecutionReviewParticipantRecoveryRun(run: Pick): boolean { return readNonEmptyString(run.contextSnapshot.retryReason) === EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON; @@ -73,6 +72,39 @@ function currentAgentParticipant(issue: IssueSnapshot): { agentId: string } | nu return agentId ? { agentId } : null; } +/** + * Resolves the responsible user for a heartbeat run the module is about to + * queue. The promote path and the immediate-recovery path both call this + * one function; each still checks the result and throws its own error with + * its own metadata when no responsible user resolves. + */ +async function resolveResponsibleUserForQueuedRun( + host: WakeQueueHost, + input: { + companyId: string; + contextSnapshot: Record; + issue: IssueSnapshot; + requestedByActorType: "user" | "agent" | "system" | null; + requestedByActorId: string | null; + source: string; + triggerDetail: string | null; + existingRunResponsibleUserId: string | null; + }, +): Promise { + const routineEnvContext = await host.getRoutineEnv({ companyId: input.companyId, issue: input.issue }); + return host.resolveResponsibleUserId({ + companyId: input.companyId, + contextSnapshot: input.contextSnapshot, + issue: input.issue, + routineEnvContext, + requestedByActorType: input.requestedByActorType, + requestedByActorId: input.requestedByActorId, + source: input.source, + triggerDetail: input.triggerDetail, + existingRunResponsibleUserId: input.existingRunResponsibleUserId, + }); +} + export type ReleaseIssueExecutionInput = { companyId: string; runId: string; @@ -80,6 +112,8 @@ export type ReleaseIssueExecutionInput = { suppressImmediateRecovery?: boolean; }; +type PauseHoldFacts = Awaited>; + /** * Drains the deferred-wake queue for the issue a run just released, in * `requestedAt` order, promoting at most one wake. When the queue empties @@ -89,20 +123,35 @@ export type ReleaseIssueExecutionInput = { */ async function runReleaseDrain( locked: LockedIssueExecution, - ports: { reader: WakeQueueReader; writer: WakeQueueWriter }, + ports: { host: WakeQueueHost; transaction: WakeQueueTransaction }, input: ReleaseIssueExecutionInput, ): Promise { const { run } = locked; - let issue = locked.primaryIssue; + const issue = locked.primaryIssue; const postCommitEffects: PostCommitEffect[] = []; + // Each `continue` path below leaves the wake row off the + // `deferred_issue_execution` status, so the next queue read cannot + // return that same row again. That invariant is what ends this loop. + // The `processedWakeIds` guard below makes a break of the invariant + // fail loudly, instead of holding this transaction open forever. + const processedWakeIds = new Set(); + while (true) { - const candidate = await ports.writer.claimNextDeferredWake({ companyId: run.companyId, issueId: issue.id }); + const candidate = await ports.transaction.findNextDeferredWake({ companyId: run.companyId, issueId: issue.id }); if (!candidate) break; + if (processedWakeIds.has(candidate.id)) { + throw new WakeQueueApplicationError( + "deferred_wake_not_advanced", + "Deferred wake queue read the same wake id twice; the row did not leave the deferred status", + { companyId: run.companyId, issueId: issue.id, wakeId: candidate.id }, + ); + } + processedWakeIds.add(candidate.id); let liveness = { liveNonSelfCommentIds: candidate.queuedCommentIds, containedSelfAuthoredComment: false }; if (candidate.queuedCommentIds.length > 0) { - liveness = await ports.writer.getQueuedCommentLiveness({ + liveness = await ports.transaction.getQueuedCommentLiveness({ companyId: run.companyId, issueId: issue.id, wakeAgentId: candidate.agentId, @@ -111,12 +160,11 @@ async function runReleaseDrain( queuedCommentIds: candidate.queuedCommentIds, }); } - const liveCommentIdsChanged = - liveness.liveNonSelfCommentIds.length !== candidate.queuedCommentIds.length || - liveness.liveNonSelfCommentIds.some((id, index) => id !== candidate.queuedCommentIds[index]); + // A length mismatch is the only way the lists can differ: the adapter derives `liveNonSelfCommentIds` with `.filter`, so it is always a subsequence of `queuedCommentIds`. + const liveCommentIdsDiffer = liveness.liveNonSelfCommentIds.length !== candidate.queuedCommentIds.length; - const deferredAgent = await ports.reader.findInvokableAgent({ companyId: run.companyId, agentId: candidate.agentId }); - const pauseHold = await ports.writer.getPauseHoldFacts({ + const deferredAgent = await ports.transaction.findInvokableAgent({ companyId: run.companyId, agentId: candidate.agentId }); + const pauseHold = await ports.transaction.getPauseHoldFacts({ companyId: run.companyId, issueId: issue.id, wakeAgentId: candidate.agentId, @@ -125,24 +173,21 @@ async function runReleaseDrain( requestedByActorId: candidate.requestedByActorId, }); - let decision = decideDeferredWake({ - queuedComment: { - hasQueuedCommentIds: candidate.queuedCommentIds.length > 0, - liveNonSelfCommentIdsLength: liveness.liveNonSelfCommentIds.length, - queuedCommentIdsLength: candidate.queuedCommentIds.length, - liveCommentIdsChanged, - containedSelfAuthoredComment: liveness.containedSelfAuthoredComment, - preservesIndependentContinuation: candidate.preservesIndependentContinuation, - }, - agent: { agentFound: deferredAgent !== null, invokable: deferredAgent?.invokable ?? false }, - pauseHold: { activePauseHold: pauseHold.activePauseHold, treeHoldInteractionWake: pauseHold.treeHoldInteractionWake }, + const commentAction = decideQueuedCommentAction({ + hasQueuedCommentIds: candidate.queuedCommentIds.length > 0, + liveNonSelfCommentIdsLength: liveness.liveNonSelfCommentIds.length, + liveCommentIdsDiffer, + containedSelfAuthoredComment: liveness.containedSelfAuthoredComment, + preservesIndependentContinuation: candidate.preservesIndependentContinuation, }); - if (decision.kind === "cancel_empty") { - await ports.writer.cancelDeferredWake({ + if (commentAction.kind === "cancel_empty") { + // A `false` result means another writer already moved this row off + // the deferred status, so the next queue read cannot return it again. + await ports.transaction.cancelDeferredWake({ companyId: run.companyId, wakeId: candidate.id, - reason: decision.selfAuthored + reason: commentAction.selfAuthored ? "Deferred wake contained only comments authored by the finishing run" : "Queued messages were discarded before promotion", now: input.now, @@ -151,8 +196,8 @@ async function runReleaseDrain( } let workingCandidate = candidate; - if (decision.kind === "normalize") { - const normalized = await ports.writer.normalizeDeferredWakeCommentIds({ + if (commentAction.kind === "normalize") { + const normalized = await ports.transaction.normalizeDeferredWakeCommentIds({ companyId: run.companyId, wakeId: candidate.id, payload: candidate.payload, @@ -161,29 +206,21 @@ async function runReleaseDrain( }); if (!normalized) continue; workingCandidate = normalized; - // Re-decide with the same agent/pause-hold facts already fetched above; the - // comment-id set now matches, so only fail/cancel-pause-hold/promote can result. - decision = decideDeferredWake({ - queuedComment: { - hasQueuedCommentIds: workingCandidate.queuedCommentIds.length > 0, - liveNonSelfCommentIdsLength: liveness.liveNonSelfCommentIds.length, - queuedCommentIdsLength: liveness.liveNonSelfCommentIds.length, - liveCommentIdsChanged: false, - containedSelfAuthoredComment: liveness.containedSelfAuthoredComment, - preservesIndependentContinuation: workingCandidate.preservesIndependentContinuation, - }, - agent: { agentFound: deferredAgent !== null, invokable: deferredAgent?.invokable ?? false }, - pauseHold: { activePauseHold: pauseHold.activePauseHold, treeHoldInteractionWake: pauseHold.treeHoldInteractionWake }, - }); } - if (decision.kind === "fail_not_invokable") { - await ports.writer.failDeferredWake({ companyId: run.companyId, wakeId: workingCandidate.id, now: input.now }); + // A comment-id rewrite cannot change the agent or pause-hold facts already fetched above, so one decision covers both the rewritten and un-rewritten cases. + const wakeOutcome = decideWakeOutcome({ + agent: { agentFound: deferredAgent !== null, invokable: deferredAgent?.invokable ?? false }, + pauseHold: { activePauseHold: pauseHold.activePauseHold, treeHoldInteractionWake: pauseHold.treeHoldInteractionWake }, + }); + + if (wakeOutcome.kind === "fail_not_invokable") { + await ports.transaction.failDeferredWake({ companyId: run.companyId, wakeId: workingCandidate.id, now: input.now }); continue; } - if (decision.kind === "cancel_pause_hold") { - await ports.writer.cancelDeferredWake({ + if (wakeOutcome.kind === "cancel_pause_hold") { + await ports.transaction.cancelDeferredWake({ companyId: run.companyId, wakeId: workingCandidate.id, reason: "Deferred wake suppressed by active subtree pause hold", @@ -192,150 +229,164 @@ async function runReleaseDrain( continue; } - // decision.kind === "promote" - const invokableAgent = deferredAgent!; + // Unreachable: decideWakeOutcome only returns "promote" when agentFound and invokable are both true. + if (!deferredAgent) throw new Error("wake-queue: promoted a deferred wake with no invokable agent"); - // Claim the wake for promotion before any other write in this branch - // (design choice: claim first, then reopen). A reopen write, or its - // `issue_reopened` post-commit effect, must never survive a lost race on - // this compare-and-set. When the claim fails, a concurrent writer already - // changed the wake's status, so this candidate is gone; move on to the - // next one instead of ending the drain. - const claimedForPromotion = await ports.writer.claimDeferredWakeForPromotion({ - companyId: run.companyId, - wakeId: workingCandidate.id, - now: input.now, - }); - if (!claimedForPromotion) continue; - - let currentIssue = issue; - - if (workingCandidate.deferredCommentIds.length > 0 && (currentIssue.status === "done" || currentIssue.status === "cancelled")) { - const selfAuthorship = await ports.writer.getCommentSelfAuthorship({ - companyId: run.companyId, - issueId: currentIssue.id, - finishingRunId: run.id, - commentIds: workingCandidate.deferredCommentIds, - }); - const shouldReopen = - !selfAuthorship.allSelfAuthored && - (workingCandidate.requestedByActorType === "user" || workingCandidate.wakeReason === "issue_reopened_via_comment"); - if (shouldReopen) { - const reopened = await ports.writer.reopenIssue({ companyId: run.companyId, issueId: currentIssue.id, runId: run.id }); - if (reopened) { - postCommitEffects.push({ - kind: "issue_reopened", - companyId: reopened.companyId, - agentId: invokableAgent.id, - runId: run.id, - issueId: reopened.id, - identifier: reopened.identifier, - reopenedFrom: currentIssue.status, - }); - currentIssue = reopened; - issue = reopened; - } - } - } - - const promotedReason = workingCandidate.reason ?? "issue_execution_promoted"; - const promotedSource = workingCandidate.source ?? "automation"; - const promotedTriggerDetail = workingCandidate.triggerDetail ?? null; - const promotedPayload = { ...workingCandidate.payload }; - delete promotedPayload["_paperclipWakeContext"]; - - const promotedContextSeed: Record = { ...workingCandidate.deferredContextSeed }; - if (pauseHold.activePauseHold) { - promotedContextSeed.treeHoldInteraction = true; - promotedContextSeed.activeTreeHold = { - holdId: pauseHold.holdId, - rootIssueId: pauseHold.rootIssueId, - mode: pauseHold.mode, - reason: pauseHold.reason, - releasePolicy: pauseHold.releasePolicy, - interaction: true, - }; - } - - const { contextSnapshot: promotedContextSnapshot, taskKey: promotedTaskKey } = enrichPromotedWakeContext({ - contextSnapshot: promotedContextSeed, - reason: promotedReason, - source: promotedSource, - triggerDetail: promotedTriggerDetail, - payload: promotedPayload, - }); - - const sessionBefore = - readNonEmptyString(promotedContextSnapshot.resumeSessionDisplayId) ?? - (await ports.reader.resolveSessionBeforeForWakeup({ - companyId: run.companyId, - agentId: invokableAgent.id, - taskKey: promotedTaskKey, - })); - - const promotedRoutineEnvContext = await ports.reader.getRoutineEnv({ - companyId: invokableAgent.companyId, - issue: currentIssue, - }); - const responsibleUserId = await ports.reader.resolveResponsibleUserId({ - companyId: invokableAgent.companyId, - contextSnapshot: promotedContextSnapshot, - issue: currentIssue, - routineEnvContext: promotedRoutineEnvContext, - requestedByActorType: workingCandidate.requestedByActorType as "user" | "agent" | "system" | null, - requestedByActorId: workingCandidate.requestedByActorId, - source: promotedSource, - triggerDetail: promotedTriggerDetail, - existingRunResponsibleUserId: run.responsibleUserId, - }); - if (!responsibleUserId) { - throw new WakeQueueApplicationError( - "responsible_user_unresolved", - "Unable to resolve responsible user for promoted heartbeat run", - { - runId: run.id, - agentId: invokableAgent.id, - companyId: invokableAgent.companyId, - issueId: currentIssue.id, - wakeReason: readNonEmptyString(promotedContextSnapshot.wakeReason), - }, - ); - } - - const promotedRun = await ports.writer.finalizePromotedWake({ - companyId: run.companyId, - wakeId: workingCandidate.id, - deferredAgent: invokableAgent, - issue: currentIssue, - finishingRun: run, - contextSnapshot: promotedContextSnapshot, - reason: promotedReason, - source: promotedSource, - triggerDetail: promotedTriggerDetail, - payload: promotedPayload, - responsibleUserId, - sessionBefore, - now: input.now, - }); - - postCommitEffects.push({ kind: "run_queued", run: promotedRun }); - return { outcome: { kind: "promoted", run: promotedRun }, postCommitEffects }; + const promoted = await promoteDeferredWake(ports, run, issue, workingCandidate, deferredAgent, pauseHold, postCommitEffects, input); + if (!promoted) continue; + return promoted; } - return runReleaseRecoveryTail(issue, run, ports.reader, ports.writer, input, postCommitEffects); + return runReleaseRecoveryTail(issue, run, ports.host, ports.transaction, input, postCommitEffects); +} + +/** + * Finalizes one deferred wake that `decideWakeOutcome` chose to promote. + * Returns `null` when the promotion claim loses a race, so the caller moves + * on to the next queued wake instead of ending the drain. + */ +async function promoteDeferredWake( + ports: { host: WakeQueueHost; transaction: WakeQueueTransaction }, + run: RunSnapshot, + issue: IssueSnapshot, + workingCandidate: DeferredWakeCandidate, + invokableAgent: InvokableAgentSnapshot, + pauseHold: PauseHoldFacts, + postCommitEffects: PostCommitEffect[], + input: ReleaseIssueExecutionInput, +): Promise { + // Claim the wake for promotion before any other write in this branch + // (design choice: claim first, then reopen). A reopen write, or its + // `issue_reopened` post-commit effect, must never survive a lost race on + // this compare-and-set. When the claim fails, a concurrent writer already + // changed the wake's status, so this candidate is gone; the caller moves + // on to the next one instead of ending the drain. + const claimedForPromotion = await ports.transaction.claimDeferredWakeForPromotion({ + companyId: run.companyId, + wakeId: workingCandidate.id, + now: input.now, + }); + if (!claimedForPromotion) return null; + + let currentIssue = issue; + + if (workingCandidate.deferredCommentIds.length > 0 && (currentIssue.status === "done" || currentIssue.status === "cancelled")) { + const selfAuthorship = await ports.transaction.getCommentSelfAuthorship({ + companyId: run.companyId, + issueId: currentIssue.id, + finishingRunId: run.id, + commentIds: workingCandidate.deferredCommentIds, + }); + const shouldReopen = + !selfAuthorship.allSelfAuthored && + (workingCandidate.requestedByActorType === "user" || workingCandidate.wakeReason === "issue_reopened_via_comment"); + if (shouldReopen) { + const reopened = await ports.transaction.reopenIssue({ companyId: run.companyId, issueId: currentIssue.id, runId: run.id }); + if (reopened) { + postCommitEffects.push({ + kind: "issue_reopened", + companyId: reopened.companyId, + agentId: invokableAgent.id, + runId: run.id, + issueId: reopened.id, + identifier: reopened.identifier, + reopenedFrom: currentIssue.status, + }); + currentIssue = reopened; + } + } + } + + const promotedReason = workingCandidate.reason ?? "issue_execution_promoted"; + const promotedSource = workingCandidate.source ?? "automation"; + const promotedTriggerDetail = workingCandidate.triggerDetail ?? null; + const promotedPayload = { ...workingCandidate.payload }; + delete promotedPayload["_paperclipWakeContext"]; + + const promotedContextSeed: Record = { ...workingCandidate.deferredContextSeed }; + if (pauseHold.activePauseHold) { + promotedContextSeed.treeHoldInteraction = true; + promotedContextSeed.activeTreeHold = { + holdId: pauseHold.holdId, + rootIssueId: pauseHold.rootIssueId, + mode: pauseHold.mode, + reason: pauseHold.reason, + releasePolicy: pauseHold.releasePolicy, + interaction: true, + }; + } + + const { contextSnapshot: promotedContextSnapshot, taskKey: promotedTaskKey } = enrichPromotedWakeContext({ + contextSnapshot: promotedContextSeed, + reason: promotedReason, + source: promotedSource, + triggerDetail: promotedTriggerDetail, + payload: promotedPayload, + }); + + const sessionBefore = + readNonEmptyString(promotedContextSnapshot.resumeSessionDisplayId) ?? + (await ports.host.resolveSessionBeforeForWakeup({ + companyId: run.companyId, + agentId: invokableAgent.id, + taskKey: promotedTaskKey, + })); + + const responsibleUserId = await resolveResponsibleUserForQueuedRun(ports.host, { + companyId: invokableAgent.companyId, + contextSnapshot: promotedContextSnapshot, + issue: currentIssue, + requestedByActorType: workingCandidate.requestedByActorType, + requestedByActorId: workingCandidate.requestedByActorId, + source: promotedSource, + triggerDetail: promotedTriggerDetail, + existingRunResponsibleUserId: run.responsibleUserId, + }); + if (!responsibleUserId) { + throw new WakeQueueApplicationError( + "responsible_user_unresolved", + "Unable to resolve responsible user for promoted heartbeat run", + { + runId: run.id, + agentId: invokableAgent.id, + companyId: invokableAgent.companyId, + issueId: currentIssue.id, + wakeReason: readNonEmptyString(promotedContextSnapshot.wakeReason), + }, + ); + } + + const promotedRun = await ports.transaction.finalizePromotedWake({ + companyId: run.companyId, + wakeId: workingCandidate.id, + deferredAgent: invokableAgent, + issue: currentIssue, + finishingRun: run, + contextSnapshot: promotedContextSnapshot, + reason: promotedReason, + source: promotedSource, + triggerDetail: promotedTriggerDetail, + payload: promotedPayload, + responsibleUserId, + sessionBefore, + now: input.now, + }); + + postCommitEffects.push({ kind: "run_queued", run: promotedRun }); + return { outcome: { kind: "promoted", run: promotedRun }, postCommitEffects }; } async function runReleaseRecoveryTail( issue: IssueSnapshot, run: RunSnapshot, - reader: WakeQueueReader, - writer: WakeQueueWriter, + host: WakeQueueHost, + transaction: WakeQueueTransaction, input: ReleaseIssueExecutionInput, postCommitEffects: PostCommitEffect[], ): Promise { const suppressImmediateRecovery = input.suppressImmediateRecovery ?? false; const isStrandedRecoveryOrigin = issue.originKind === STRANDED_ISSUE_RECOVERY_ORIGIN_KIND; - const recoveryAgent = await reader.findInvokableAgent({ companyId: issue.companyId, agentId: run.agentId }); + const recoveryAgent = await transaction.findInvokableAgent({ companyId: issue.companyId, agentId: run.agentId }); const currentParticipant = currentAgentParticipant(issue); const reviewParticipantApplies = @@ -354,22 +405,22 @@ async function runReleaseRecoveryTail( (run.status === "failed" || run.status === "timed_out" || run.status === "cancelled"); const suppressedByPauseHold = (reviewParticipantApplies || immediateApplies) - ? await writer.isAutomaticRecoverySuppressedByPauseHold({ companyId: issue.companyId, issueId: issue.id }) + ? await transaction.isAutomaticRecoverySuppressedByPauseHold({ companyId: issue.companyId, issueId: issue.id }) : false; const hasExistingExecutionPath = reviewParticipantApplies - ? await writer.hasExistingExecutionPath({ + ? await transaction.hasExistingExecutionPath({ companyId: issue.companyId, issueId: issue.id, excludeRunId: run.id, agentId: currentParticipant?.agentId ?? null, }) : immediateApplies - ? await writer.hasExistingExecutionPath({ companyId: issue.companyId, issueId: issue.id, excludeRunId: run.id, agentId: null }) + ? await transaction.hasExistingExecutionPath({ companyId: issue.companyId, issueId: issue.id, excludeRunId: run.id, agentId: null }) : false; const hasExplicitBlockerPath = immediateApplies && !reviewParticipantApplies - ? await writer.hasExplicitBlockerPath({ companyId: issue.companyId, issueId: issue.id }) + ? await transaction.hasExplicitBlockerPath({ companyId: issue.companyId, issueId: issue.id }) : false; const expectedRetryReason: "assignment_recovery" | "issue_continuation_needed" = @@ -379,27 +430,23 @@ async function runReleaseRecoveryTail( suppressImmediateRecovery, reviewParticipant: { applies: reviewParticipantApplies, - hasExistingExecutionPath, - hasPersistedMonitor: Boolean(issue.monitorNextCheckAt), - suppressedByPauseHold, - isStrandedRecoveryOrigin, - recoveryAgentPresent: recoveryAgent !== null, - recoveryAgentInvokable: recoveryAgent?.invokable ?? false, isExecutionReviewParticipantRecoveryRun: isExecutionReviewParticipantRecoveryRun(run), }, immediate: { applies: immediateApplies, isDispositionRepairRetry: readNonEmptyString(run.contextSnapshot.retryReason) === ISSUE_DISPOSITION_REPAIR_RETRY_REASON, + hasExplicitBlockerPath, + isWorkspaceValidationFailedRun: isWorkspaceValidationFailedRun(run), + isConfigurationIncompleteFailedRun: isConfigurationIncompleteFailedRun(run), + automaticRecoveryAlreadyFailed: didAutomaticRecoveryFail(run, expectedRetryReason), + }, + shared: { hasExistingExecutionPath, hasPersistedMonitor: Boolean(issue.monitorNextCheckAt), - hasExplicitBlockerPath, suppressedByPauseHold, isStrandedRecoveryOrigin, recoveryAgentPresent: recoveryAgent !== null, recoveryAgentInvokable: recoveryAgent?.invokable ?? false, - isWorkspaceValidationFailedRun: isWorkspaceValidationFailedRun(run), - isConfigurationIncompleteFailedRun: isConfigurationIncompleteFailedRun(run), - automaticRecoveryAlreadyFailed: didAutomaticRecoveryFail(run, expectedRetryReason), }, }); @@ -415,35 +462,32 @@ async function runReleaseRecoveryTail( } if (decision.kind === "blocked") { - const { notice, recoveryCause } = await writer.buildBlockedRecoveryNotice({ - noticeKind: decision.notice, - issueStatus: issue.status === "todo" ? "todo" : "in_progress", - finishingRun: run, - }); return { outcome: { kind: "blocked", issue, previousStatus: statusForBlock(issue), - notice, - recoveryCause, + noticeKind: decision.notice, }, postCommitEffects, }; } - const sessionBefore = await reader.resolveSessionBeforeForWakeup({ + // Unreachable: decideReleaseRecovery only reaches "queue_review_participant_recovery" or "queue_recovery" when the shared recovery-agent facts are both true. + if (!recoveryAgent) throw new Error("wake-queue: queued a recovery run with no invokable recovery agent"); + + const sessionBefore = await host.resolveSessionBeforeForWakeup({ companyId: issue.companyId, - agentId: recoveryAgent!.id, + agentId: recoveryAgent.id, taskKey: readNonEmptyString(run.contextSnapshot.taskKey) ?? readNonEmptyString(run.contextSnapshot.issueId), }); if (decision.kind === "queue_review_participant_recovery") { - const queuedRun = await writer.queueReviewParticipantRecoveryRun({ + const queuedRun = await transaction.queueReviewParticipantRecoveryRun({ companyId: issue.companyId, issue, finishingRun: run, - recoveryAgent: recoveryAgent!, + recoveryAgent, sessionBefore, now: input.now, }); @@ -451,14 +495,53 @@ async function runReleaseRecoveryTail( return { outcome: { kind: "queued_review_participant_recovery", run: queuedRun }, postCommitEffects }; } - // decision.kind === "queue_recovery"; the adapter builds the recovery - // context snapshot and resolves the responsible user from it, throwing - // WakeQueueApplicationError when no responsible user resolves. - const queuedRun = await writer.queueImmediateRecoveryRun({ + // decision.kind === "queue_recovery"; resolve the responsible user here, + // in the application layer, before the transaction port queues the run. + const { retryReason, recoveryReason, recoverySource } = deriveImmediateRecoveryContextLabels(issue.status); + const recoveryContextSnapshot = withRecoveryContext( + { + issueId: issue.id, + taskId: issue.id, + wakeReason: recoveryReason, + retryReason, + source: recoverySource, + retryOfRunId: run.id, + }, + "normal_model", + ); + + const recoveryResponsibleUserId = await resolveResponsibleUserForQueuedRun(host, { + companyId: issue.companyId, + contextSnapshot: recoveryContextSnapshot, + issue, + requestedByActorType: "system", + requestedByActorId: null, + source: "automation", + triggerDetail: "system", + existingRunResponsibleUserId: run.responsibleUserId, + }); + if (!recoveryResponsibleUserId) { + throw new WakeQueueApplicationError( + "responsible_user_unresolved", + "Unable to resolve responsible user for recovery heartbeat run", + { + runId: run.id, + agentId: recoveryAgent.id, + companyId: issue.companyId, + issueId: issue.id, + wakeReason: recoveryReason, + }, + ); + } + + const queuedRun = await transaction.queueImmediateRecoveryRun({ companyId: issue.companyId, issue, finishingRun: run, - recoveryAgent: recoveryAgent!, + recoveryAgent, + reason: recoveryReason, + contextSnapshot: recoveryContextSnapshot, + responsibleUserId: recoveryResponsibleUserId, sessionBefore, now: input.now, }); @@ -487,8 +570,7 @@ export function createReleaseIssueExecution(deps: { issue: result.outcome.issue, previousStatus: result.outcome.previousStatus, latestRun: result.run, - notice: result.outcome.notice, - recoveryCause: result.outcome.recoveryCause, + noticeKind: result.outcome.noticeKind, }); } else if (result.outcome.kind === "blocked_recovery_in_place") { await deps.recovery.escalateStrandedRecoveryIssueInPlace({ diff --git a/server/src/modules/wake-queue/domain/context.ts b/server/src/modules/wake-queue/domain/context.ts index d9f31ef6e3..67cc586fdc 100644 --- a/server/src/modules/wake-queue/domain/context.ts +++ b/server/src/modules/wake-queue/domain/context.ts @@ -5,6 +5,7 @@ // building the release half's promoted-run snapshot. import { extractWakeCommentIds, WAKE_COMMENT_IDS_KEY } from "../../run-dispatch/index.js"; +import { readNonEmptyString } from "./values.js"; const PAPERCLIP_WAKE_PAYLOAD_KEY = "paperclipWake"; const PAPERCLIP_WAKE_COMMENT_KEY = "paperclipWakeComment"; @@ -21,10 +22,6 @@ const INTERACTION_CONTINUATION_CONTEXT_KEYS = [ "newlyResolvedItemIds", ] as const; -function readNonEmptyString(value: unknown): string | null { - return typeof value === "string" && value.trim().length > 0 ? value : null; -} - function deriveTaskKey( contextSnapshot: Record | null | undefined, payload: Record | null | undefined, diff --git a/server/src/modules/wake-queue/domain/policy.test.ts b/server/src/modules/wake-queue/domain/policy.test.ts index 61eb44d0e6..b6c4a92a92 100644 --- a/server/src/modules/wake-queue/domain/policy.test.ts +++ b/server/src/modules/wake-queue/domain/policy.test.ts @@ -1,147 +1,269 @@ import { describe, expect, it } from "vitest"; import { - decideDeferredWake, + decidePreDrain, + decideQueuedCommentAction, decideReleaseRecovery, - type DeferredWakeFacts, + decideWakeOutcome, + deriveImmediateRecoveryContextLabels, + type DeferredWakeOutcomeFacts, + type DeferredWakeQueuedCommentFacts, + type ImmediateRecoveryContextLabels, + type PreDrainFacts, type ReleaseRecoveryFacts, } from "./policy.js"; -const baseDeferredWakeFacts: DeferredWakeFacts = { - queuedComment: { - hasQueuedCommentIds: false, - liveNonSelfCommentIdsLength: 0, - queuedCommentIdsLength: 0, - liveCommentIdsChanged: false, - containedSelfAuthoredComment: false, - preservesIndependentContinuation: false, - }, - agent: { agentFound: true, invokable: true }, - pauseHold: { activePauseHold: false, treeHoldInteractionWake: false }, +const basePreDrainFacts: PreDrainFacts = { + issueRowPresent: true, + executionRunIdMatchesRun: true, + isWorkspaceValidationFailedRun: false, + isConfigurationIncompleteFailedRun: false, + issueStatus: "in_progress", + hasAssigneeUser: false, + assigneeAgentMatchesRunAgent: true, + legacyExecutionNeedsReconciliation: false, + executionCancellationAcknowledged: false, }; -describe("decideDeferredWake", () => { +describe("decidePreDrain", () => { const cases: Array<{ name: string; - facts: DeferredWakeFacts; - expected: ReturnType; + facts: PreDrainFacts; + expected: ReturnType; + }> = [ + { + name: "released: the issue row is missing", + facts: { ...basePreDrainFacts, issueRowPresent: false }, + expected: { kind: "released" }, + }, + { + name: "released: another run already holds executionRunId", + facts: { ...basePreDrainFacts, executionRunIdMatchesRun: false }, + expected: { kind: "released" }, + }, + { + name: "blocked: a workspace-validation failure on an eligible todo issue", + facts: { ...basePreDrainFacts, isWorkspaceValidationFailedRun: true, issueStatus: "todo" }, + expected: { kind: "blocked", noticeKind: "workspace_validation" }, + }, + { + name: "blocked: a configuration-incomplete failure on an eligible in_progress issue", + facts: { ...basePreDrainFacts, isConfigurationIncompleteFailedRun: true, issueStatus: "in_progress" }, + expected: { kind: "blocked", noticeKind: "configuration_incomplete" }, + }, + { + name: "proceed: a workspace-validation failure on an issue that is not todo or in_progress", + facts: { ...basePreDrainFacts, isWorkspaceValidationFailedRun: true, issueStatus: "in_review" }, + expected: { kind: "proceed" }, + }, + { + name: "proceed: a workspace-validation failure but the issue already has an assigned user", + facts: { + ...basePreDrainFacts, + isWorkspaceValidationFailedRun: true, + issueStatus: "todo", + hasAssigneeUser: true, + }, + expected: { kind: "proceed" }, + }, + { + name: "proceed: a workspace-validation failure but the assigned agent does not match the finishing run's agent", + facts: { + ...basePreDrainFacts, + isWorkspaceValidationFailedRun: true, + issueStatus: "todo", + assigneeAgentMatchesRunAgent: false, + }, + expected: { kind: "proceed" }, + }, + { + name: "released: legacy execution needs reconciliation", + facts: { ...basePreDrainFacts, legacyExecutionNeedsReconciliation: true }, + expected: { kind: "released" }, + }, + { + name: "released: an acknowledged execution cancellation", + facts: { ...basePreDrainFacts, executionCancellationAcknowledged: true }, + expected: { kind: "released" }, + }, + { + name: "proceed: none of the pre-drain conditions apply", + facts: basePreDrainFacts, + expected: { kind: "proceed" }, + }, + { + name: "the blocked-notice check is evaluated before legacy-execution reconciliation", + facts: { + ...basePreDrainFacts, + isWorkspaceValidationFailedRun: true, + issueStatus: "todo", + // If reconciliation were checked first, this would force a "released" + // outcome; the expected "blocked" here proves the blocked-notice + // check, evaluated first, decides the outcome. + legacyExecutionNeedsReconciliation: true, + }, + expected: { kind: "blocked", noticeKind: "workspace_validation" }, + }, + { + name: "the blocked-notice check is evaluated before an acknowledged execution cancellation", + facts: { + ...basePreDrainFacts, + isConfigurationIncompleteFailedRun: true, + issueStatus: "todo", + // If the cancellation check were checked first, this would force a + // "released" outcome; the expected "blocked" here proves the + // blocked-notice check, evaluated first, decides the outcome. + executionCancellationAcknowledged: true, + }, + expected: { kind: "blocked", noticeKind: "configuration_incomplete" }, + }, + ]; + + for (const testCase of cases) { + it(testCase.name, () => { + expect(decidePreDrain(testCase.facts)).toEqual(testCase.expected); + }); + } +}); + +const baseQueuedCommentFacts: DeferredWakeQueuedCommentFacts = { + hasQueuedCommentIds: false, + liveNonSelfCommentIdsLength: 0, + liveCommentIdsDiffer: false, + containedSelfAuthoredComment: false, + preservesIndependentContinuation: false, +}; + +describe("decideQueuedCommentAction", () => { + const cases: Array<{ + name: string; + facts: DeferredWakeQueuedCommentFacts; + expected: ReturnType; }> = [ { name: "cancel_empty: all queued comments discarded and no independent continuation", facts: { - ...baseDeferredWakeFacts, - queuedComment: { - hasQueuedCommentIds: true, - liveNonSelfCommentIdsLength: 0, - queuedCommentIdsLength: 2, - liveCommentIdsChanged: true, - containedSelfAuthoredComment: false, - preservesIndependentContinuation: false, - }, + ...baseQueuedCommentFacts, + hasQueuedCommentIds: true, + liveNonSelfCommentIdsLength: 0, + liveCommentIdsDiffer: true, + containedSelfAuthoredComment: false, + preservesIndependentContinuation: false, }, expected: { kind: "cancel_empty", selfAuthored: false }, }, { name: "cancel_empty: self-authored comments discarded, error text reflects self-authorship", facts: { - ...baseDeferredWakeFacts, - queuedComment: { - hasQueuedCommentIds: true, - liveNonSelfCommentIdsLength: 0, - queuedCommentIdsLength: 1, - liveCommentIdsChanged: true, - containedSelfAuthoredComment: true, - preservesIndependentContinuation: false, - }, + ...baseQueuedCommentFacts, + hasQueuedCommentIds: true, + liveNonSelfCommentIdsLength: 0, + liveCommentIdsDiffer: true, + containedSelfAuthoredComment: true, + preservesIndependentContinuation: false, }, expected: { kind: "cancel_empty", selfAuthored: true }, }, { name: "normalize: no live comments, but an independent continuation reason still rewrites the queued id list", facts: { - ...baseDeferredWakeFacts, - queuedComment: { - hasQueuedCommentIds: true, - liveNonSelfCommentIdsLength: 0, - queuedCommentIdsLength: 1, - liveCommentIdsChanged: true, - containedSelfAuthoredComment: false, - preservesIndependentContinuation: true, - }, + ...baseQueuedCommentFacts, + hasQueuedCommentIds: true, + liveNonSelfCommentIdsLength: 0, + liveCommentIdsDiffer: true, + containedSelfAuthoredComment: false, + preservesIndependentContinuation: true, }, expected: { kind: "normalize" }, }, { - name: "promote: an independent continuation reason keeps the wake alive with the live id set already matching", + name: "proceed: an independent continuation reason keeps the wake alive with the live id set already matching", facts: { - ...baseDeferredWakeFacts, - queuedComment: { - hasQueuedCommentIds: true, - liveNonSelfCommentIdsLength: 0, - queuedCommentIdsLength: 0, - liveCommentIdsChanged: false, - containedSelfAuthoredComment: false, - preservesIndependentContinuation: true, - }, + ...baseQueuedCommentFacts, + hasQueuedCommentIds: true, + liveNonSelfCommentIdsLength: 0, + liveCommentIdsDiffer: false, + containedSelfAuthoredComment: false, + preservesIndependentContinuation: true, }, - expected: { kind: "promote" }, + expected: { kind: "proceed" }, }, { name: "normalize: the live comment id set differs from the queued set", facts: { - ...baseDeferredWakeFacts, - queuedComment: { - hasQueuedCommentIds: true, - liveNonSelfCommentIdsLength: 1, - queuedCommentIdsLength: 2, - liveCommentIdsChanged: true, - containedSelfAuthoredComment: false, - preservesIndependentContinuation: false, - }, + ...baseQueuedCommentFacts, + hasQueuedCommentIds: true, + liveNonSelfCommentIdsLength: 1, + liveCommentIdsDiffer: true, + containedSelfAuthoredComment: false, + preservesIndependentContinuation: false, }, expected: { kind: "normalize" }, }, { - name: "fail_not_invokable: the agent lookup returns not-found", + name: "proceed: no queued comments", + facts: baseQueuedCommentFacts, + expected: { kind: "proceed" }, + }, + { + name: "proceed: queued comments are all still live, matching the queued set", facts: { - ...baseDeferredWakeFacts, - agent: { agentFound: false, invokable: false }, + ...baseQueuedCommentFacts, + hasQueuedCommentIds: true, + liveNonSelfCommentIdsLength: 1, + liveCommentIdsDiffer: false, }, + expected: { kind: "proceed" }, + }, + ]; + + for (const testCase of cases) { + it(testCase.name, () => { + expect(decideQueuedCommentAction(testCase.facts)).toEqual(testCase.expected); + }); + } +}); + +const baseWakeOutcomeFacts: DeferredWakeOutcomeFacts = { + agent: { agentFound: true, invokable: true }, + pauseHold: { activePauseHold: false, treeHoldInteractionWake: false }, +}; + +describe("decideWakeOutcome", () => { + const cases: Array<{ + name: string; + facts: DeferredWakeOutcomeFacts; + expected: ReturnType; + }> = [ + { + name: "fail_not_invokable: the agent lookup returns not-found", + facts: { ...baseWakeOutcomeFacts, agent: { agentFound: false, invokable: false } }, expected: { kind: "fail_not_invokable" }, }, { name: "fail_not_invokable: the agent is found but not invokable", - facts: { - ...baseDeferredWakeFacts, - agent: { agentFound: true, invokable: false }, - }, + facts: { ...baseWakeOutcomeFacts, agent: { agentFound: true, invokable: false } }, expected: { kind: "fail_not_invokable" }, }, { name: "cancel_pause_hold: an active pause hold with no verified tree-hold interaction", - facts: { - ...baseDeferredWakeFacts, - pauseHold: { activePauseHold: true, treeHoldInteractionWake: false }, - }, + facts: { ...baseWakeOutcomeFacts, pauseHold: { activePauseHold: true, treeHoldInteractionWake: false } }, expected: { kind: "cancel_pause_hold" }, }, { name: "promote: an active pause hold but a verified tree-hold interaction wake survives it", - facts: { - ...baseDeferredWakeFacts, - pauseHold: { activePauseHold: true, treeHoldInteractionWake: true }, - }, + facts: { ...baseWakeOutcomeFacts, pauseHold: { activePauseHold: true, treeHoldInteractionWake: true } }, expected: { kind: "promote" }, }, { - name: "promote: no queued comments, an invokable agent, and no pause hold", - facts: baseDeferredWakeFacts, + name: "promote: an invokable agent and no pause hold", + facts: baseWakeOutcomeFacts, expected: { kind: "promote" }, }, ]; for (const testCase of cases) { it(testCase.name, () => { - expect(decideDeferredWake(testCase.facts)).toEqual(testCase.expected); + expect(decideWakeOutcome(testCase.facts)).toEqual(testCase.expected); }); } }); @@ -150,27 +272,23 @@ const baseReleaseRecoveryFacts: ReleaseRecoveryFacts = { suppressImmediateRecovery: false, reviewParticipant: { applies: false, - hasExistingExecutionPath: false, - hasPersistedMonitor: false, - suppressedByPauseHold: false, - isStrandedRecoveryOrigin: false, - recoveryAgentPresent: true, - recoveryAgentInvokable: true, isExecutionReviewParticipantRecoveryRun: false, }, immediate: { applies: false, isDispositionRepairRetry: false, + hasExplicitBlockerPath: false, + isWorkspaceValidationFailedRun: false, + isConfigurationIncompleteFailedRun: false, + automaticRecoveryAlreadyFailed: false, + }, + shared: { hasExistingExecutionPath: false, hasPersistedMonitor: false, - hasExplicitBlockerPath: false, suppressedByPauseHold: false, isStrandedRecoveryOrigin: false, recoveryAgentPresent: true, recoveryAgentInvokable: true, - isWorkspaceValidationFailedRun: false, - isConfigurationIncompleteFailedRun: false, - automaticRecoveryAlreadyFailed: false, }, }; @@ -198,11 +316,8 @@ describe("decideReleaseRecovery", () => { name: "released: immediate recovery applies but an existing execution path already covers it", facts: { ...baseReleaseRecoveryFacts, - immediate: { - ...baseReleaseRecoveryFacts.immediate, - applies: true, - hasExistingExecutionPath: true, - }, + immediate: { ...baseReleaseRecoveryFacts.immediate, applies: true }, + shared: { ...baseReleaseRecoveryFacts.shared, hasExistingExecutionPath: true }, }, expected: { kind: "released" }, }, @@ -234,11 +349,8 @@ describe("decideReleaseRecovery", () => { name: "blocked_recovery_in_place: immediate recovery applies on a stranded-issue-recovery origin", facts: { ...baseReleaseRecoveryFacts, - immediate: { - ...baseReleaseRecoveryFacts.immediate, - applies: true, - isStrandedRecoveryOrigin: true, - }, + immediate: { ...baseReleaseRecoveryFacts.immediate, applies: true }, + shared: { ...baseReleaseRecoveryFacts.shared, isStrandedRecoveryOrigin: true }, }, expected: { kind: "blocked_recovery_in_place" }, }, @@ -246,11 +358,8 @@ describe("decideReleaseRecovery", () => { name: "blocked: immediate recovery applies but the recovery agent is not invokable", facts: { ...baseReleaseRecoveryFacts, - immediate: { - ...baseReleaseRecoveryFacts.immediate, - applies: true, - recoveryAgentInvokable: false, - }, + immediate: { ...baseReleaseRecoveryFacts.immediate, applies: true }, + shared: { ...baseReleaseRecoveryFacts.shared, recoveryAgentInvokable: false }, }, expected: { kind: "blocked", notice: "immediate_execution_path" }, }, @@ -290,11 +399,8 @@ describe("decideReleaseRecovery", () => { name: "released: review-participant recovery applies but a persisted monitor already covers it", facts: { ...baseReleaseRecoveryFacts, - reviewParticipant: { - ...baseReleaseRecoveryFacts.reviewParticipant, - applies: true, - hasPersistedMonitor: true, - }, + reviewParticipant: { ...baseReleaseRecoveryFacts.reviewParticipant, applies: true }, + shared: { ...baseReleaseRecoveryFacts.shared, hasPersistedMonitor: true }, }, expected: { kind: "released" }, }, @@ -302,11 +408,8 @@ describe("decideReleaseRecovery", () => { name: "blocked_recovery_in_place: review-participant recovery applies on a stranded-issue-recovery origin", facts: { ...baseReleaseRecoveryFacts, - reviewParticipant: { - ...baseReleaseRecoveryFacts.reviewParticipant, - applies: true, - isStrandedRecoveryOrigin: true, - }, + reviewParticipant: { ...baseReleaseRecoveryFacts.reviewParticipant, applies: true }, + shared: { ...baseReleaseRecoveryFacts.shared, isStrandedRecoveryOrigin: true }, }, expected: { kind: "blocked_recovery_in_place" }, }, @@ -335,11 +438,10 @@ describe("decideReleaseRecovery", () => { facts: { ...baseReleaseRecoveryFacts, reviewParticipant: { ...baseReleaseRecoveryFacts.reviewParticipant, applies: true }, - immediate: { - ...baseReleaseRecoveryFacts.immediate, - applies: true, - isStrandedRecoveryOrigin: true, - }, + // If the immediate branch were evaluated instead, this flag would force a + // "blocked" outcome; the expected "queue_review_participant_recovery" here + // proves the review-participant branch, checked first, decides the outcome. + immediate: { ...baseReleaseRecoveryFacts.immediate, applies: true, isWorkspaceValidationFailedRun: true }, }, expected: { kind: "queue_review_participant_recovery" }, }, @@ -351,3 +453,45 @@ describe("decideReleaseRecovery", () => { }); } }); + +describe("deriveImmediateRecoveryContextLabels", () => { + const cases: Array<{ + name: string; + issueStatus: string; + expected: ImmediateRecoveryContextLabels; + }> = [ + { + name: "todo: the issue lost its assignment", + issueStatus: "todo", + expected: { + retryReason: "assignment_recovery", + recoveryReason: "issue_assignment_recovery", + recoverySource: "issue.assignment_recovery", + }, + }, + { + name: "not todo: an in_progress issue is a stalled continuation", + issueStatus: "in_progress", + expected: { + retryReason: "issue_continuation_needed", + recoveryReason: "issue_continuation_needed", + recoverySource: "issue.continuation_recovery", + }, + }, + { + name: "not todo: any other status also derives the stalled-continuation labels", + issueStatus: "in_review", + expected: { + retryReason: "issue_continuation_needed", + recoveryReason: "issue_continuation_needed", + recoverySource: "issue.continuation_recovery", + }, + }, + ]; + + for (const testCase of cases) { + it(testCase.name, () => { + expect(deriveImmediateRecoveryContextLabels(testCase.issueStatus)).toEqual(testCase.expected); + }); + } +}); diff --git a/server/src/modules/wake-queue/domain/policy.ts b/server/src/modules/wake-queue/domain/policy.ts index 7fd959a2d8..88feb2a7d2 100644 --- a/server/src/modules/wake-queue/domain/policy.ts +++ b/server/src/modules/wake-queue/domain/policy.ts @@ -1,28 +1,93 @@ // Pure decision rules for the release half of the deferred issue-execution // wake state machine: -// - the per-wake decision (decideDeferredWake), applied to the earliest -// deferred wake queued against the issue a run just released +// - the pre-drain decision (decidePreDrain), applied once per lock +// acquisition, before the caller touches the deferred-wake queue at all +// - the per-wake decision, split into decideQueuedCommentAction and +// decideWakeOutcome, applied to the earliest deferred wake queued +// against the issue a run just released // - the release-recovery decision (decideReleaseRecovery), applied once // the deferred-wake queue is empty and no wake was promoted // The caller reads the database and packs the result into a facts object. // This file only branches on that facts object; it never queries a // database, reads the clock, or reads the wake context payload directly. +export type PreDrainFacts = { + /** True when the transaction found the issue row the lock is for. */ + issueRowPresent: boolean; + /** True when the issue carries no `executionRunId`, or carries the finishing run's own id. False when a different run already holds it. */ + executionRunIdMatchesRun: boolean; + isWorkspaceValidationFailedRun: boolean; + isConfigurationIncompleteFailedRun: boolean; + /** The issue's current status. Only "todo" and "in_progress" can lead to the blocked-notice outcome. */ + issueStatus: string; + hasAssigneeUser: boolean; + assigneeAgentMatchesRunAgent: boolean; + legacyExecutionNeedsReconciliation: boolean; + /** True when the finishing run is cancelled and its stored result carries an acknowledged execution cancellation. */ + executionCancellationAcknowledged: boolean; +}; + +export type PreDrainDecision = + | { kind: "released" } + | { kind: "blocked"; noticeKind: ReleaseRecoveryBlockedNoticeKind } + | { kind: "proceed" }; + +/** + * Decides the pre-drain release outcome for one lock acquisition, before the + * caller runs its own lock function. Check order is fixed: the issue-row + * check runs first, then the blocked-notice check, then legacy-execution + * reconciliation, then acknowledged execution cancellation. Each check + * returns as soon as it applies, so an earlier true condition can hide a + * later one when both hold at the same time. "proceed" means none of the + * four checks applied; the caller then runs its own write-carrying check + * and, if that also clears, calls its lock function. + */ +export function decidePreDrain(facts: PreDrainFacts): PreDrainDecision { + if (!facts.issueRowPresent || !facts.executionRunIdMatchesRun) { + return { kind: "released" }; + } + + if ( + (facts.isWorkspaceValidationFailedRun || facts.isConfigurationIncompleteFailedRun) && + (facts.issueStatus === "todo" || facts.issueStatus === "in_progress") && + !facts.hasAssigneeUser && + facts.assigneeAgentMatchesRunAgent + ) { + return { + kind: "blocked", + noticeKind: facts.isConfigurationIncompleteFailedRun ? "configuration_incomplete" : "workspace_validation", + }; + } + + if (facts.legacyExecutionNeedsReconciliation) { + return { kind: "released" }; + } + + if (facts.executionCancellationAcknowledged) { + return { kind: "released" }; + } + + return { kind: "proceed" }; +} + export type DeferredWakeQueuedCommentFacts = { /** True when the wake carries one or more queued comment ids to check. */ hasQueuedCommentIds: boolean; /** Count of queued comment ids that are still live and not self-authored by the finishing run. */ liveNonSelfCommentIdsLength: number; - /** Count of queued comment ids the wake originally carried. */ - queuedCommentIdsLength: number; /** True when the live, non-self comment id list differs from the queued list. */ - liveCommentIdsChanged: boolean; + liveCommentIdsDiffer: boolean; /** True when every discarded comment id was authored by the finishing run. */ containedSelfAuthoredComment: boolean; /** True when the wake carries an independent reason to continue even with no live comments. */ preservesIndependentContinuation: boolean; }; +export type DeferredWakeQueuedCommentDecision = + | { kind: "cancel_empty"; selfAuthored: boolean } + | { kind: "normalize" } + | { kind: "proceed" }; + export type DeferredWakeAgentFacts = { /** True when the wake's agent exists in the issue's own company. */ agentFound: boolean; @@ -37,42 +102,39 @@ export type DeferredWakePauseHoldFacts = { treeHoldInteractionWake: boolean; }; -export type DeferredWakeFacts = { - queuedComment: DeferredWakeQueuedCommentFacts; +export type DeferredWakeOutcomeFacts = { agent: DeferredWakeAgentFacts; pauseHold: DeferredWakePauseHoldFacts; }; -export type DeferredWakeDecision = - | { kind: "cancel_empty"; selfAuthored: boolean } - | { kind: "normalize" } +export type DeferredWakeOutcomeDecision = | { kind: "fail_not_invokable" } | { kind: "cancel_pause_hold" } | { kind: "promote" }; -/** - * Decides what to do with the earliest deferred wake queued against the - * issue a run just released. The caller applies a "normalize" decision (a - * queued-comment-id rewrite) and calls this function again with facts that - * reflect the rewrite, so a single wake can normalize and then also fail, - * cancel, or promote in the same drain step — matching the order the - * original state machine always evaluated them in. - */ -export function decideDeferredWake(facts: DeferredWakeFacts): DeferredWakeDecision { - const { queuedComment, agent, pauseHold } = facts; - +/** Decides what to do with a deferred wake's queued comment ids. On "normalize", the caller rewrites the ids and then calls `decideWakeOutcome` directly; the rewrite cannot change the agent or pause-hold facts. */ +export function decideQueuedCommentAction( + facts: DeferredWakeQueuedCommentFacts, +): DeferredWakeQueuedCommentDecision { if ( - queuedComment.hasQueuedCommentIds && - queuedComment.liveNonSelfCommentIdsLength === 0 && - !queuedComment.preservesIndependentContinuation + facts.hasQueuedCommentIds && + facts.liveNonSelfCommentIdsLength === 0 && + !facts.preservesIndependentContinuation ) { - return { kind: "cancel_empty", selfAuthored: queuedComment.containedSelfAuthoredComment }; + return { kind: "cancel_empty", selfAuthored: facts.containedSelfAuthoredComment }; } - if (queuedComment.hasQueuedCommentIds && queuedComment.liveCommentIdsChanged) { + if (facts.hasQueuedCommentIds && facts.liveCommentIdsDiffer) { return { kind: "normalize" }; } + return { kind: "proceed" }; +} + +/** Decides the outcome for a deferred wake whose queued-comment action resolved to "proceed" (or finished its rewrite): fail, cancel, or promote. */ +export function decideWakeOutcome(facts: DeferredWakeOutcomeFacts): DeferredWakeOutcomeDecision { + const { agent, pauseHold } = facts; + if (!agent.agentFound || !agent.invokable) { return { kind: "fail_not_invokable" }; } @@ -84,15 +146,19 @@ export function decideDeferredWake(facts: DeferredWakeFacts): DeferredWakeDecisi return { kind: "promote" }; } -export type ReleaseRecoveryReviewParticipantFacts = { - /** True when the issue is in_review, unassigned to a user, and waiting on the finishing run as the current agent participant. */ - applies: boolean; +/** Shared between the review-participant and immediate branches; the caller derives every field from the same expression regardless of which branch applies. */ +export type ReleaseRecoverySharedFacts = { hasExistingExecutionPath: boolean; hasPersistedMonitor: boolean; suppressedByPauseHold: boolean; isStrandedRecoveryOrigin: boolean; recoveryAgentPresent: boolean; recoveryAgentInvokable: boolean; +}; + +export type ReleaseRecoveryReviewParticipantFacts = { + /** True when the issue is in_review, unassigned to a user, and waiting on the finishing run as the current agent participant. */ + applies: boolean; /** True when the finishing run was itself a review-participant-recovery retry. */ isExecutionReviewParticipantRecoveryRun: boolean; }; @@ -102,13 +168,7 @@ export type ReleaseRecoveryImmediateFacts = { applies: boolean; /** True when the finishing run itself carried the disposition-repair retry reason. */ isDispositionRepairRetry: boolean; - hasExistingExecutionPath: boolean; - hasPersistedMonitor: boolean; hasExplicitBlockerPath: boolean; - suppressedByPauseHold: boolean; - isStrandedRecoveryOrigin: boolean; - recoveryAgentPresent: boolean; - recoveryAgentInvokable: boolean; isWorkspaceValidationFailedRun: boolean; isConfigurationIncompleteFailedRun: boolean; /** didAutomaticRecoveryFail(run, expectedRetryReason) for the issue's own status branch. */ @@ -120,6 +180,7 @@ export type ReleaseRecoveryFacts = { suppressImmediateRecovery: boolean; reviewParticipant: ReleaseRecoveryReviewParticipantFacts; immediate: ReleaseRecoveryImmediateFacts; + shared: ReleaseRecoverySharedFacts; }; export type ReleaseRecoveryBlockedNoticeKind = @@ -135,6 +196,32 @@ export type ReleaseRecoveryDecision = | { kind: "queue_review_participant_recovery" } | { kind: "queue_recovery" }; +export type ImmediateRecoveryContextLabels = { + retryReason: "assignment_recovery" | "issue_continuation_needed"; + recoveryReason: "issue_assignment_recovery" | "issue_continuation_needed"; + recoverySource: "issue.assignment_recovery" | "issue.continuation_recovery"; +}; + +/** + * Derives the three labels an immediate-recovery heartbeat run's context + * snapshot carries, from the issue's status. A `todo` issue lost its + * assignment; every other status this decision reaches is a stalled + * continuation. + */ +export function deriveImmediateRecoveryContextLabels(issueStatus: string): ImmediateRecoveryContextLabels { + return issueStatus === "todo" + ? { + retryReason: "assignment_recovery", + recoveryReason: "issue_assignment_recovery", + recoverySource: "issue.assignment_recovery", + } + : { + retryReason: "issue_continuation_needed", + recoveryReason: "issue_continuation_needed", + recoverySource: "issue.continuation_recovery", + }; +} + /** * Decides the release-recovery outcome once the deferred-wake queue is * empty and no wake was promoted. The review-participant branch and the @@ -144,23 +231,23 @@ export type ReleaseRecoveryDecision = * order. */ export function decideReleaseRecovery(facts: ReleaseRecoveryFacts): ReleaseRecoveryDecision { - const { reviewParticipant, immediate } = facts; + const { reviewParticipant, immediate, shared } = facts; if (reviewParticipant.applies) { if ( facts.suppressImmediateRecovery || - reviewParticipant.hasExistingExecutionPath || - reviewParticipant.hasPersistedMonitor || - reviewParticipant.suppressedByPauseHold + shared.hasExistingExecutionPath || + shared.hasPersistedMonitor || + shared.suppressedByPauseHold ) { return { kind: "released" }; } - if (reviewParticipant.isStrandedRecoveryOrigin) { + if (shared.isStrandedRecoveryOrigin) { return { kind: "blocked_recovery_in_place" }; } const shouldBlock = - !reviewParticipant.recoveryAgentInvokable || - !reviewParticipant.recoveryAgentPresent || + !shared.recoveryAgentInvokable || + !shared.recoveryAgentPresent || reviewParticipant.isExecutionReviewParticipantRecoveryRun; if (shouldBlock) { return { kind: "blocked", notice: "execution_review_participant" }; @@ -171,19 +258,15 @@ export function decideReleaseRecovery(facts: ReleaseRecoveryFacts): ReleaseRecov if (immediate.isDispositionRepairRetry) return { kind: "released" }; if (!immediate.applies) return { kind: "released" }; if (facts.suppressImmediateRecovery) return { kind: "released" }; - if ( - immediate.hasExistingExecutionPath || - immediate.hasPersistedMonitor || - immediate.hasExplicitBlockerPath - ) { + if (shared.hasExistingExecutionPath || shared.hasPersistedMonitor || immediate.hasExplicitBlockerPath) { return { kind: "released" }; } - if (immediate.suppressedByPauseHold) return { kind: "released" }; - if (immediate.isStrandedRecoveryOrigin) return { kind: "blocked_recovery_in_place" }; + if (shared.suppressedByPauseHold) return { kind: "released" }; + if (shared.isStrandedRecoveryOrigin) return { kind: "blocked_recovery_in_place" }; const shouldBlockImmediately = - !immediate.recoveryAgentInvokable || - !immediate.recoveryAgentPresent || + !shared.recoveryAgentInvokable || + !shared.recoveryAgentPresent || immediate.isWorkspaceValidationFailedRun || immediate.isConfigurationIncompleteFailedRun || immediate.automaticRecoveryAlreadyFailed; diff --git a/server/src/modules/wake-queue/domain/values.ts b/server/src/modules/wake-queue/domain/values.ts new file mode 100644 index 0000000000..48191b69d2 --- /dev/null +++ b/server/src/modules/wake-queue/domain/values.ts @@ -0,0 +1,26 @@ +// Small, shared domain values for the wake-queue module: string readers, the +// two failed-run codes, and the recovery retry reason they gate. The +// application and adapter layers both need these, so they live here once +// instead of twice. + +export function readNonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value : null; +} + +export function parseObject(value: unknown): Record { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as Record) + : {}; +} + +export const WORKSPACE_VALIDATION_FAILURE_CODE = "workspace_validation_failed"; +export const CONFIGURATION_INCOMPLETE_FAILURE_CODE = "configuration_incomplete"; +export const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON = "execution_review_participant_recovery"; + +export function isWorkspaceValidationFailedRun(run: { errorCode: string | null }): boolean { + return run.errorCode === WORKSPACE_VALIDATION_FAILURE_CODE; +} + +export function isConfigurationIncompleteFailedRun(run: { errorCode: string | null }): boolean { + return run.errorCode === CONFIGURATION_INCOMPLETE_FAILURE_CODE || run.errorCode === "model_not_found"; +} diff --git a/server/src/modules/wake-queue/index.ts b/server/src/modules/wake-queue/index.ts index 4c794e452e..62e369a27b 100644 --- a/server/src/modules/wake-queue/index.ts +++ b/server/src/modules/wake-queue/index.ts @@ -5,7 +5,7 @@ import type { IssueSnapshot, RecoveryEscalationPort, RunSnapshot, - WakeQueueReader, + WakeQueueHost, } from "./application/ports.js"; export type { @@ -14,16 +14,21 @@ export type { RunSummary, } from "./application/types.js"; export { WakeQueueApplicationError } from "./application/types.js"; -export type { IssueSnapshot, RunSnapshot, RecoveryEscalationPort } from "./application/ports.js"; +export type { + IssueSnapshot, + RunSnapshot, + RecoveryEscalationPort, + ReleaseRecoveryBlockedNoticeKind, +} from "./application/ports.js"; export type { ReleaseIssueExecutionInput } from "./application/use-cases.js"; export type WakeQueueDeps = { /** Stays in `heartbeat.ts`; resolves the responsible user for a promoted or recovery run seed. */ - resolveResponsibleUserId: WakeQueueReader["resolveResponsibleUserId"]; + resolveResponsibleUserId: WakeQueueHost["resolveResponsibleUserId"]; /** Stays in `heartbeat.ts`; reads the routine environment context for an execution issue. */ - getRoutineEnv: WakeQueueReader["getRoutineEnv"]; + getRoutineEnv: WakeQueueHost["getRoutineEnv"]; /** Stays in `heartbeat.ts`; resolves the session-before display id for a wakeup. */ - resolveSessionBeforeForWakeup: WakeQueueReader["resolveSessionBeforeForWakeup"]; + resolveSessionBeforeForWakeup: WakeQueueHost["resolveSessionBeforeForWakeup"]; /** `services/recovery`'s stranded-issue escalation, called only after the release transaction commits. */ recovery: RecoveryEscalationPort; }; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 37a38c592f..f91ca1fb9f 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -376,6 +376,10 @@ import { readContinuationAttempt, } from "./recovery/index.js"; import { + buildConfigurationIncompleteRecoveryNoticeSeed, + buildExecutionReviewParticipantRecoveryNoticeSeed, + buildImmediateExecutionPathRecoveryNoticeSeed, + buildWorkspaceValidationRecoveryNoticeSeed, SANDBOX_PROVIDER_PLUGIN_NOT_READY_REASON, type StrandedRecoveryNoticeSeed, } from "./recovery/stranded-notice.js"; @@ -401,7 +405,10 @@ import { import { createWakeQueue, WakeQueueApplicationError, + type IssueSnapshot as WakeQueueIssueSnapshot, type PostCommitEffect as WakeQueuePostCommitEffect, + type ReleaseRecoveryBlockedNoticeKind, + type RunSnapshot as WakeQueueRunSnapshot, } from "../modules/wake-queue/index.js"; import { buildIssueReviewPathLostIdempotencyKey, @@ -8446,6 +8453,61 @@ export function heartbeatService( } } + // The wake-queue module's plain snapshots hold only the fields the release + // decision needs; escalation needs the full row, so this re-reads both by + // id after the release transaction has committed. Returns null when either + // row is gone, so both escalation adapters below skip the escalation call. + async function loadStrandedEscalationRows(input: { + issue: WakeQueueIssueSnapshot; + latestRun: WakeQueueRunSnapshot; + }) { + const [issueRow] = await db + .select() + .from(issues) + .where(and(eq(issues.id, input.issue.id), eq(issues.companyId, input.issue.companyId))); + const [runRow] = await db + .select() + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, input.latestRun.id), eq(heartbeatRuns.companyId, input.latestRun.companyId))); + if (!issueRow || !runRow) return null; + return { issueRow, runRow }; + } + + // Reproduces `adapters/postgres.ts`'s former `buildBlockedRecoveryNotice` + // four-arm switch, now built once here from the full run row this file + // already re-reads through `loadStrandedEscalationRows`. + function buildStrandedRecoveryNoticeForKind( + noticeKind: ReleaseRecoveryBlockedNoticeKind, + input: { issueStatus: "todo" | "in_progress"; runRow: typeof heartbeatRuns.$inferSelect }, + ): { + notice: StrandedRecoveryNoticeSeed; + recoveryCause: + | typeof WORKSPACE_VALIDATION_RECOVERY_CAUSE + | typeof CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE + | typeof EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE + | undefined; + } { + if (noticeKind === "workspace_validation") { + return { notice: buildWorkspaceValidationRecoveryNoticeSeed(), recoveryCause: WORKSPACE_VALIDATION_RECOVERY_CAUSE }; + } + if (noticeKind === "configuration_incomplete") { + const configurationIncomplete = parseObject(parseObject(input.runRow.resultJson).configurationIncomplete); + return { + notice: buildConfigurationIncompleteRecoveryNoticeSeed( + Object.keys(configurationIncomplete).length > 0 ? configurationIncomplete : null, + ), + recoveryCause: CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE, + }; + } + if (noticeKind === "execution_review_participant") { + return { + notice: buildExecutionReviewParticipantRecoveryNoticeSeed(), + recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE, + }; + } + return { notice: buildImmediateExecutionPathRecoveryNoticeSeed({ status: input.issueStatus }), recoveryCause: undefined }; + } + const wakeQueue = createWakeQueue(db, { resolveResponsibleUserId: async (input) => { // `input.issue` is the wake-queue module's own transaction-scoped @@ -8489,45 +8551,28 @@ export function heartbeatService( return resolveSessionBeforeForWakeup(agent, input.taskKey); }, recovery: { - // The wake-queue module's plain snapshots hold only the fields the - // release decision needs; escalation needs the full row, so this - // re-reads both by id after the release transaction has committed. escalateStrandedAssignedIssue: async (input) => { - const [issueRow] = await db - .select() - .from(issues) - .where(and(eq(issues.id, input.issue.id), eq(issues.companyId, input.issue.companyId))); - const [runRow] = await db - .select() - .from(heartbeatRuns) - .where(and(eq(heartbeatRuns.id, input.latestRun.id), eq(heartbeatRuns.companyId, input.latestRun.companyId))); - if (!issueRow || !runRow) return; + const rows = await loadStrandedEscalationRows(input); + if (!rows) return; + const { notice, recoveryCause } = buildStrandedRecoveryNoticeForKind(input.noticeKind, { + issueStatus: input.issue.status === "todo" ? "todo" : "in_progress", + runRow: rows.runRow, + }); await recovery.escalateStrandedAssignedIssue({ - issue: issueRow, + issue: rows.issueRow, previousStatus: input.previousStatus, - latestRun: runRow, - notice: input.notice as StrandedRecoveryNoticeSeed, - recoveryCause: (input.recoveryCause ?? undefined) as - | typeof WORKSPACE_VALIDATION_RECOVERY_CAUSE - | typeof CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE - | typeof EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE - | undefined, + latestRun: rows.runRow, + notice, + recoveryCause, }); }, escalateStrandedRecoveryIssueInPlace: async (input) => { - const [issueRow] = await db - .select() - .from(issues) - .where(and(eq(issues.id, input.issue.id), eq(issues.companyId, input.issue.companyId))); - const [runRow] = await db - .select() - .from(heartbeatRuns) - .where(and(eq(heartbeatRuns.id, input.latestRun.id), eq(heartbeatRuns.companyId, input.latestRun.companyId))); - if (!issueRow || !runRow) return; + const rows = await loadStrandedEscalationRows(input); + if (!rows) return; await recovery.escalateStrandedRecoveryIssueInPlace({ - issue: issueRow, + issue: rows.issueRow, previousStatus: input.previousStatus, - latestRun: runRow, + latestRun: rows.runRow, }); }, },