fix(jobs): stop follow-ups waiting behind an operator hold on a closed PR

src/jobs/closedPullRequestHandoffs.ts

@@ -0,0 +1,160 @@
+import { getPendingPrCommentsKey } from '@propr/core';
+import { resolveRecoveryDeps, type FollowupHandoffRecoveryDeps } from './followupHandoffDeps.js';
+import { releaseFollowupHandoffRecord } from './followupHandoffOperator.js';
+import { recoverInterruptedFollowupHandoffs } from './followupHandoffRecovery.js';
+import { splitHandoffRepository, type HandoffOctokit } from './followupHandoffResumeTypes.js';
+import {
+    HANDOFF_AWAITING_OPERATOR, loadFollowupHandoff, loadFollowupHandoffs, parseHandoffJobData,
+    type FollowupHandoffKey, type FollowupHandoffRecord,
+} from './followupHandoffStore.js';
+
+/**
+ * Follow-ups held for an operator on a pull request that has since been merged
+ * or closed. Nothing can be published there any more, so the hold would block
+ * every later follow-up for the pull request forever: each one waited behind
+ * it, and before waiting in place each queued a replacement every minute.
+ */
+
+export type PullRequestClosure = 'merged' | 'closed';
+
+/** How the pull request was closed, or null while it is open. */
+export async function readPullRequestClosure(
+    octokit: Pick<HandoffOctokit, 'request'>,
+    owner: string,
+    repo: string,
+    pullRequestNumber: number,
+): Promise<PullRequestClosure | null> {
+    const { data } = await octokit.request('GET /repos/{owner}/{repo}/pulls/{pull_number}', {
+        owner, repo, pull_number: pullRequestNumber,
+    });
+    const pull = data as { state?: string; merged?: boolean | null; merged_at?: string | null };
+    if (pull.state !== 'closed') return null;
+    return pull.merged || pull.merged_at ? 'merged' : 'closed';
+}
+
+interface DeletingRedis { del?(key: string): Promise<unknown> }
+
+/**
+ * Releases one operator-held follow-up of a closed pull request and drops the
+ * comments still queued for it. The release is the operator release, forced:
+ * it still requires the agent proven gone and the PR lease, and it only
+ * detaches the worktree HEAD, so retained files and commits stay for the
+ * normal worktree retention cleanup.
+ */
+export async function releaseClosedPullRequestHandoff(
+    record: FollowupHandoffRecord,
+    closure: PullRequestClosure,
+    deps: FollowupHandoffRecoveryDeps = {},
+): Promise<{ released: boolean; reason?: string }> {
+    const key: FollowupHandoffKey = { repository: record.repository, pull_request: record.pull_request };
+    const result = await releaseFollowupHandoffRecord(key, {
+        force: true, reason: closure === 'merged' ? 'released_pr_merged' : 'released_pr_closed',
+    }, deps);
+    if (result.released && deps.redis) {
+        const jobData = parseHandoffJobData(record);
+        const { owner, repo } = splitHandoffRepository(record.repository);
+        await (deps.redis as DeletingRedis).del?.(
+            getPendingPrCommentsKey(jobData.repoOwner ?? owner, jobData.repoName ?? repo, record.pull_request),
+        );
+    }
+    return result;
+}
+
+/** Releases the pull request's follow-up hold, if one is held for an operator; never throws. */
+export async function releaseHeldHandoffOfClosedPullRequest(
+    ref: { repoOwner: string; repoName: string; pullRequestNumber: number },
+    closure: PullRequestClosure,
+    deps: FollowupHandoffRecoveryDeps = {},
+): Promise<{ released: boolean; reason?: string }> {
+    const resolved = await resolveRecoveryDeps(deps);
+    try {
+        const record = await loadFollowupHandoff(resolved.store, {
+            repository: `${ref.repoOwner}/${ref.repoName}`.toLowerCase(), pull_request: ref.pullRequestNumber,
+        });
+        if (record?.status !== HANDOFF_AWAITING_OPERATOR) return { released: false, reason: 'not_held' };
+        return await releaseClosedPullRequestHandoff(record, closure, sharedDeps(deps, resolved));
+    } catch (error) {
+        resolved.log.warn({ ...ref, error: (error as Error).message }, 'Could not release the follow-up hold of a closed pull request');
+        return { released: false, reason: 'error' };
+    } finally {
+        if (resolved.ownRedis) await resolved.redis.quit?.();
+    }
+}
+
+function sharedDeps(
+    deps: FollowupHandoffRecoveryDeps,
+    resolved: Awaited<ReturnType<typeof resolveRecoveryDeps>>,
+): FollowupHandoffRecoveryDeps {
+    return {
+        ...deps, store: resolved.store, redis: resolved.redis, log: resolved.log,
+        getOctokit: resolved.getOctokit, git: resolved.git, inspectLiveness: resolved.inspectLiveness,
+    };
+}
+
+export interface ClosedPullRequestHandoffSummary {
+    inspected: number;
+    released: number;
+    refused: number;
+    errors: number;
+}
+
+/** Each held follow-up's pull request is read from GitHub at most this often. */
+export const CLOSED_PULL_REQUEST_CHECK_INTERVAL_MS = 10 * 60_000;
+const lastChecked = new Map<string, number>();
+
+/**
+ * One pass over the operator-held follow-ups: those whose pull request was
+ * merged or closed are released. A refused release (the agent still running,
+ * unverifiable liveness, the lease held by a live writer) is retried after
+ * the check interval.
+ */
+export async function releaseClosedPullRequestHandoffs(
+    deps: FollowupHandoffRecoveryDeps & { now?: () => number } = {},
+): Promise<ClosedPullRequestHandoffSummary> {
+    const now = deps.now ?? Date.now;
+    const resolved = await resolveRecoveryDeps(deps);
+    const summary: ClosedPullRequestHandoffSummary = { inspected: 0, released: 0, refused: 0, errors: 0 };
+    try {
+        const held = await loadFollowupHandoffs(resolved.store, { status: HANDOFF_AWAITING_OPERATOR });
+        const current = new Set<string>();
+        for (const record of held) {
+            const checkKey = `${record.repository}#${record.pull_request}#${record.incarnation}`;
+            current.add(checkKey);
+            const checkedAt = lastChecked.get(checkKey);
+            if (checkedAt !== undefined && now() - checkedAt < CLOSED_PULL_REQUEST_CHECK_INTERVAL_MS) continue;
+            lastChecked.set(checkKey, now());
+            summary.inspected += 1;
+            try {
+                const { owner, repo } = splitHandoffRepository(record.repository);
+                const closure = await readPullRequestClosure(await resolved.getOctokit(), owner, repo, record.pull_request);
+                if (!closure) continue;
+                const result = await releaseClosedPullRequestHandoff(record, closure, sharedDeps(deps, resolved));
+                if (result.released) {
+                    summary.released += 1;
+                    resolved.log.info({ repository: record.repository, pullRequest: record.pull_request, taskId: record.task_id, closure },
+                        'Released the follow-up hold of a closed pull request');
+                } else {
+                    summary.refused += 1;
+                    resolved.log.warn({ repository: record.repository, pullRequest: record.pull_request, reason: result.reason },
+                        'Could not yet release the follow-up hold of a closed pull request');
+                }
+            } catch (error) {
+                summary.errors += 1;
+                resolved.log.warn({ repository: record.repository, pullRequest: record.pull_request, error: (error as Error).message },
+                    'Failed to check a held follow-up against its pull request');
+            }
+        }
+        for (const key of lastChecked.keys()) if (!current.has(key)) lastChecked.delete(key);
+        return summary;
+    } finally {
+        if (resolved.ownRedis) await resolved.redis.quit?.();
+    }
+}
+
+/** The worker's periodic handoff pass: interrupted publications, then holds on closed pull requests. */
+export async function recoverFollowupHandoffsAndClosedHolds(): Promise<Record<string, unknown>> {
+    return {
+        ...await recoverInterruptedFollowupHandoffs(),
+        closedPullRequestHolds: await releaseClosedPullRequestHandoffs(),
+    };
+}

src/jobs/followupHandoffOperator.ts

@@ -210,7 +210,7 @@ async function confirmUnconfirmedRetryOwnership(
  */
 export async function releaseFollowupHandoffRecord(
     key: FollowupHandoffKey,
-    options: { force?: boolean },
+    options: { force?: boolean; /** Recorded status reason; defaults to an operator release. */ reason?: string },
     deps: FollowupHandoffRecoveryDeps = {},
 ): Promise<{ released: boolean; reason?: string }> {
     const resolved = await resolveRecoveryDeps(deps);
@@ -253,7 +253,8 @@ export async function releaseFollowupHandoffRecord(
             const refusal = await branchReleaseRefusal(resolved.git, fresh);
             if (refusal) return { released: false, reason: refusal };
             const advanced = await advanceFollowupHandoff(resolved.store, fresh, {
-                status: HANDOFF_RELEASED, status_reason: options.force ? 'released_by_operator_force' : 'released_by_operator',
+                status: HANDOFF_RELEASED,
+                status_reason: options.reason ?? (options.force ? 'released_by_operator_force' : 'released_by_operator'),
             });
             return { released: Boolean(advanced), reason: advanced ? undefined : 'superseded' };
         });

src/jobs/prCommentHandoffIntegration.ts

@@ -11,6 +11,7 @@ import {
 import { adoptPredecessorFollowupHandoff, settleFollowupHandoffAfterFailure } from './followupHandoffRecovery.js';
 import { delayPRCommentJobInPlace, predecessorWaitDelay, schedulePRCommentRecovery } from './prCommentCollisionRecovery.js';
 import { releasePRProcessingLock, startPRProcessingLockHeartbeat } from './prProcessingLock.js';
+import { releaseHeldHandoffOfClosedPullRequest, type PullRequestClosure } from './closedPullRequestHandoffs.js';
 import type { ProcessingState } from './prPublicationRecovery.js';
 import type { PullRequestPublication } from './prPublication.js';
 
@@ -277,6 +278,14 @@ export interface PredecessorHandoffGateParams {
     lockKey: string;
     lockToken: string;
     correlatedLogger: Logger;
+    /**
+     * How the pull request was closed, or null while it is open. Read only when
+     * a predecessor blocks this writer: a merged or closed pull request ends the
+     * wait instead of waiting forever behind a hold nobody can publish.
+     */
+    pullRequestClosure?: () => Promise<PullRequestClosure | null>;
+    /** Releases the closed pull request's operator-held follow-up; injectable for tests. */
+    releaseClosedHold?: typeof releaseHeldHandoffOfClosedPullRequest;
 }
 
 /** Removes the given comment IDs from a captured set in place, returning how many were removed. */
@@ -380,6 +389,8 @@ export async function gatePredecessorHandoff(params: PredecessorHandoffGateParam
         if (skipped) await releasePRProcessingLock(redisClient, params.lockKey, params.lockToken);
         return skipped;
     }
+    const closed = await endWaitOnClosedPullRequest(params, blocked);
+    if (closed) return closed;
     // Wait in place as the same job and task, backing off: an operator hold
     // can stand for days, and a fresh replacement every minute left thousands
     // of cancelled tasks behind a single unanswered follow-up.
@@ -399,6 +410,29 @@ export async function gatePredecessorHandoff(params: PredecessorHandoffGateParam
     return { status: 'rescheduled', reason: blocked, replacementTaskId };
 }
 
+/**
+ * A blocked writer on a merged or closed pull request stops instead of
+ * waiting: nothing can be published there, and the hold it waits behind would
+ * never be settled. Its captured comments are not returned to the pending
+ * list, exactly as for a job that finds the pull request closed later on, and
+ * the hold itself is released once this job has let go of the PR lease.
+ */
+async function endWaitOnClosedPullRequest(params: PredecessorHandoffGateParams, blocked: string): Promise<JobResult | undefined> {
+    const { ref, redisClient, correlatedLogger } = params;
+    let closure: PullRequestClosure | null = null;
+    try {
+        closure = await params.pullRequestClosure?.() ?? null;
+    } catch (error) {
+        correlatedLogger.warn({ error: (error as Error).message }, 'Could not read the pull request state while blocked by a predecessor; waiting');
+    }
+    if (!closure) return undefined;
+    await releasePRProcessingLock(redisClient, params.lockKey, params.lockToken);
+    const release = await (params.releaseClosedHold ?? releaseHeldHandoffOfClosedPullRequest)(ref, closure, { redis: redisClient as never });
+    correlatedLogger.info({ pullRequestNumber: ref.pullRequestNumber, closure, blocked, holdReleased: release.released, releaseReason: release.reason },
+        'Pull request is closed; ending the follow-up instead of waiting behind its predecessor');
+    return { status: 'cancelled', reason: 'cancelled_pr_closed', pullRequestNumber: ref.pullRequestNumber };
+}
+
 export interface HandoffErrorSettlementParams {
     ref: HandoffRef;
     taskId: string;

src/jobs/processPullRequestCommentJob.ts

@@ -54,6 +54,8 @@ import { findPRContinuation, type Contribution } from './prContinuation.js';
 import { suspendObsoleteValidationForImplementation } from './followupCiSuspension.js';
 import { stoppedReviewRecap } from './notificationRecap.js';
 import { resumeAutomaticPRCommentRetry } from './prCommentAutomaticRetry.js';
+import { readPullRequestClosure } from './closedPullRequestHandoffs.js';
+import type { HandoffOctokit } from './followupHandoffResumeTypes.js';
 
 const redisClient = new Redis({
     host: process.env.REDIS_HOST || '127.0.0.1',
@@ -414,8 +416,15 @@ async function processAdmittedPRCommentJob(job: Job<CommentJobData>): Promise<Jo
     const predecessorGate = await gatePredecessorHandoff({
         job, ref: { repoOwner, repoName, pullRequestNumber }, taskId, stateManager, redisClient,
         pickedUpComments: context.pickedUpComments, commentsToProcess, lockKey, lockToken, correlatedLogger,
+        pullRequestClosure: () => readPullRequestClosure(octokit as unknown as HandoffOctokit, repoOwner, repoName, pullRequestNumber),
     });
-    if (predecessorGate) return predecessorGate;
+    if (predecessorGate) {
+        if (predecessorGate.reason === 'cancelled_pr_closed') {
+            await createPRCommentTaskStateIfMissing({ job, taskId, stateManager, preexistingState, modelName, correlatedLogger });
+            await stateManager.markTaskCancelled(taskId, 'system', { reason: formatTaskTerminalReason('cancelled_pr_closed'), terminalReason: 'cancelled_pr_closed' });
+        }
+        return predecessorGate;
+    }
 
     const executionController = new AbortController();
     const stopLockHeartbeat = startPRProcessingLockHeartbeat({

src/worker.ts

@@ -40,7 +40,7 @@ import { ultrafixFailedStepFinalizerOptions } from './jobs/ultrafixFailedStepFin
 import { getTaskReplacementService } from './taskReplacement/index.js';
 import { recoverNonterminalGoals } from './goalRecovery.js';
 import { reconcileFollowupCiSuspensions } from './jobs/followupCiSuspension.js';
-import { recoverInterruptedFollowupHandoffs } from './jobs/followupHandoffRecovery.js';
+import { recoverFollowupHandoffsAndClosedHolds } from './jobs/closedPullRequestHandoffs.js';
 import { cancelClosedPullRequestValidation } from './jobs/closedPullRequestCiCancellation.js';
 import { prepareAgentRegistryAtStartup, processAgentImagePreparationJob, startAgentImageWatchdog } from './workerAgentPreparation.js';
 
@@ -403,7 +403,7 @@ async function startWorker(options: WorkerOptions = {}): Promise<StartedWorker>
             closedPullRequests: await cancelClosedPullRequestValidation(),
             epicQueues: await reconcileEpicExecutionQueues(),
         }),
-        recoverFollowupHandoffs: () => recoverInterruptedFollowupHandoffs(),
+        recoverFollowupHandoffs: () => recoverFollowupHandoffsAndClosedHolds(),
     });
 
     const usageTipsRunner = await startUsageTipsSelectionRunner();