fix: rotate apply queue batches

.github/workflows/sweep.yml

@@ -2259,6 +2259,8 @@ jobs:
             sync_open_pr_batch="true"
             item_numbers=""
           fi
+          explicit_item_numbers="$item_numbers"
+          auto_selected_apply_batch=false
           sync_comments_only="${{ github.event_name == 'workflow_dispatch' && github.event.inputs.apply_sync_comments_only || 'false' }}"
           product_direction_enabled=false
           case "${CLAWSWEEPER_UNCONFIRMED_PRODUCT_DIRECTION_CLOSE_ENABLED,,}" in
@@ -2276,6 +2278,7 @@ jobs:
           target_slug="$TARGET_REPO"
           target_slug="${target_slug//\//-}"
           cursor_path="results/comment-sync-cursors/${target_slug}.json"
+          apply_cursor_path="results/apply-cursors/${target_slug}.json"
           next_cursor=""
           if [ "$sync_open_pr_batch" = "true" ]; then
             sync_comments_only="true"
@@ -2363,13 +2366,16 @@ jobs:
               --apply-close-reasons "$apply_close_reasons" \
               --stale-min-age-days "$stale_min_age_days" \
               --min-age-days "$min_age_days" \
-              --min-age-minutes "$min_age_minutes")"
+              --min-age-minutes "$min_age_minutes" \
+              --batch-size "$close_processed_limit" \
+              --cursor-path "$apply_cursor_path")"
             if [ -n "$item_numbers" ]; then
+              auto_selected_apply_batch=true
               proposed_count="$(pnpm run --silent workflow -- count-csv --items "$item_numbers")"
               if [ "$proposed_count" -lt "$limit" ]; then
                 limit="$proposed_count"
               fi
-              echo "Auto-selected $proposed_count proposed close(s): $item_numbers"
+              echo "Auto-selected $proposed_count proposed close candidate(s) from apply cursor window: $item_numbers"
             fi
           fi
           item_numbers_arg=()
@@ -2478,17 +2484,32 @@ jobs:
             result_count="$(pnpm run --silent workflow -- count-report --report ".artifacts/apply-reports/apply-report-$checkpoint.json")"
             closed_total=$((closed_total + closed_in_chunk))
             echo "Checkpoint $checkpoint result_count=$result_count closed_in_chunk=$closed_in_chunk closed_total=$closed_total/$limit"
-            publish_changes "chore: apply sweep decisions checkpoint $checkpoint" records apply-report.json
+            apply_publish_paths=(records apply-report.json)
+            if [ "$auto_selected_apply_batch" = "true" ]; then
+              pnpm run workflow -- write-apply-cursor \
+                --cursor-path "$apply_cursor_path" \
+                --report ".artifacts/apply-reports/apply-report-$checkpoint.json" \
+                --target-repo "$TARGET_REPO" \
+                --item-numbers "$item_numbers"
+              apply_publish_paths+=(results/apply-cursors)
+            fi
+            publish_changes "chore: apply sweep decisions checkpoint $checkpoint" "${apply_publish_paths[@]}"
             pnpm run status -- \
               --target-repo "$TARGET_REPO" \
               --state "Apply in progress" \
               --detail "Checkpoint $checkpoint finished. Fresh closes in checkpoint: $closed_in_chunk. Total fresh closes in this run: $closed_total/$limit. Result records in checkpoint: $result_count, including durable review comment syncs." \
               --run-url "https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}"
             publish_status "chore: update sweep apply checkpoint $checkpoint status"
             echo "::endgroup::"
-            if [ "$result_count" -ge "$close_processed_limit" ] && [ "$closed_in_chunk" -gt 0 ]; then
-              echo "Close scan reached its $close_processed_limit-record budget; queueing a fresh-token continuation."
-              continue_apply=true
+            if [ "$result_count" -ge "$close_processed_limit" ]; then
+              if [ "$closed_in_chunk" -gt 0 ]; then
+                echo "Close scan reached its $close_processed_limit-record budget; queueing a fresh-token continuation."
+                continue_apply=true
+              elif [ "$auto_selected_apply_batch" = "true" ]; then
+                echo "Close scan reached its $close_processed_limit-record budget with no fresh closes; cursor is persisted and the next scheduled apply run will advance the next window."
+              else
+                echo "Explicit close scan reached its $close_processed_limit-record budget without fresh closes; stopping to avoid a retry loop."
+              fi
               break
             fi
             if [ "$result_count" -eq 0 ]; then
@@ -2508,6 +2529,10 @@ jobs:
             --detail "Apply/comment-sync run finished with $closed_total fresh closes out of requested limit $limit. See apply-report.json for per-item results." \
             --run-url "https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}"
           publish_status "chore: mark sweep apply finished"
+          next_apply_item_numbers="$item_numbers"
+          if [ "$auto_selected_apply_batch" = "true" ] && [ -z "$explicit_item_numbers" ]; then
+            next_apply_item_numbers=""
+          fi
           {
             echo "APPLY_CLOSED_TOTAL=$closed_total"
             echo "APPLY_LIMIT=$limit"
@@ -2520,10 +2545,11 @@ jobs:
             echo "APPLY_CLOSE_DELAY_MS=$close_delay_ms"
             echo "APPLY_PROGRESS_EVERY=$progress_every"
             echo "APPLY_CHECKPOINT_SIZE=$checkpoint_size"
-            echo "APPLY_ITEM_NUMBERS=$item_numbers"
+            echo "APPLY_ITEM_NUMBERS=$next_apply_item_numbers"
             echo "APPLY_TARGET_REPO=$TARGET_REPO"
             echo "APPLY_SYNC_COMMENTS_ONLY=$sync_comments_only"
             echo "APPLY_SYNC_OPEN_PR_BATCH=$sync_open_pr_batch"
+            echo "APPLY_AUTO_SELECTED_BATCH=$auto_selected_apply_batch"
             echo "APPLY_COMMENT_SYNC_MIN_AGE_DAYS=$comment_sync_min_age_days"
           } >> "$GITHUB_ENV"
 
@@ -2545,6 +2571,9 @@ jobs:
           if [ "${APPLY_SYNC_OPEN_PR_BATCH:-false}" = "true" ]; then
             publish_args+=(--path results/comment-sync-cursors)
           fi
+          if [ "${APPLY_AUTO_SELECTED_BATCH:-false}" = "true" ]; then
+            publish_args+=(--path results/apply-cursors)
+          fi
           pnpm run repair:publish-main -- "${publish_args[@]}"
 
       - name: Continue apply sweep

src/clawsweeper.ts

@@ -16973,6 +16973,7 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
   const applyReportEntriesForDir = (
     dir: string,
     location: "items" | "closed",
+    filterRequested = true,
   ): Array<
     ReportEntry & {
       location: "items" | "closed";
@@ -16984,7 +16985,9 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
       .filter(
         (entry) =>
           entry.repo === targetRepo() &&
-          (requestedItemNumberSet.size === 0 || requestedItemNumberSet.has(entry.number)),
+          (!filterRequested ||
+            requestedItemNumberSet.size === 0 ||
+            requestedItemNumberSet.has(entry.number)),
       )
       .map((entry) => ({
         ...entry,
@@ -16998,9 +17001,8 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
       left.number - right.number,
   );
   const files = fileEntries.map((entry) => entry.name);
-  const openFileEntryByNumber = new Map(
-    fileEntries.filter((entry) => entry.location === "items").map((entry) => [entry.number, entry]),
-  );
+  const allOpenFileEntries = applyReportEntriesForDir(itemsDir, "items", false);
+  const openFileEntryByNumber = new Map(allOpenFileEntries.map((entry) => [entry.number, entry]));
   const closedThisRun = new Set<string>();
   if (fileEntries.length === 0 && !existsSync(itemsDir)) {
     console.log("No items directory.");
@@ -17056,16 +17058,19 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
       });
       renameSync(path, join(closedDir, file));
     };
+    const markApplyChecked = (): void => {
+      markdown = replaceFrontMatterValue(markdown, "apply_checked_at", new Date().toISOString());
+      if (!dryRun) writeFileSync(path, markdown, "utf8");
+    };
     const recordApplySkipped = (actionTaken: ActionTaken, reason: string): boolean => {
+      markApplyChecked();
       results.push({ number, action: actionTaken, reason });
       processedCount += 1;
       maybeLogProgress(`skipped #${number}: ${reason}`);
       return processedCount >= processedLimit;
     };
     const markApplySkipped = (actionTaken: ActionTaken, reason: string): boolean => {
       markdown = replaceFrontMatterValue(markdown, "action_taken", actionTaken);
-      markdown = replaceFrontMatterValue(markdown, "apply_checked_at", new Date().toISOString());
-      if (!dryRun) writeFileSync(path, markdown, "utf8");
       return recordApplySkipped(actionTaken, reason);
     };
     const markLabelSyncAuthSkipped = (labelKind: string): boolean =>
@@ -17085,6 +17090,13 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
         action !== "retry_pr_close_coverage_proof" &&
         !shouldProbeClosedState)
     ) {
+      if (
+        !storedHash &&
+        requestedItemNumberSet.has(number) &&
+        recordApplySkipped("kept_open", "review lacks an item snapshot hash")
+      ) {
+        break;
+      }
       continue;
     }
     let isCloseProposal = isApplyCloseCandidateReport(markdown);
@@ -17322,6 +17334,9 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
               }) === null &&
               counterpartOpenClosingPullRequestReason === null &&
               counterpartSameAuthorReason === null;
+            if (result && !fileEntries.some((entry) => entry.number === counterpartNumber)) {
+              fileEntries.push(counterpartEntry);
+            }
           }
         }
       }
@@ -17330,6 +17345,7 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
       return result;
     };
     if (syncCommentsOnly && state !== "open") {
+      markApplyChecked();
       results.push({ number, action: "skipped_already_closed", reason: `state is ${state}` });
       processedCount += 1;
       maybeLogProgress(`skipped comment sync #${number}: already ${state}`);
@@ -17529,16 +17545,13 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
         : reviewedAuthorAssociation;
       markdown = replaceFrontMatterValue(markdown, "author_association", authorAssociation);
       markdown = replaceFrontMatterValue(markdown, "action_taken", "skipped_maintainer_authored");
-      markdown = replaceFrontMatterValue(markdown, "apply_checked_at", new Date().toISOString());
-      if (!dryRun) writeFileSync(path, markdown, "utf8");
-      results.push({
-        number,
-        action: "skipped_maintainer_authored",
-        reason: `author association is ${authorAssociation}`,
-      });
-      processedCount += 1;
-      maybeLogProgress(`skipped #${number}: maintainer authored`);
-      if (processedCount >= processedLimit) break;
+      if (
+        recordApplySkipped(
+          "skipped_maintainer_authored",
+          `author association is ${authorAssociation}`,
+        )
+      )
+        break;
       continue;
     }
     const updatedSinceReview = Boolean(storedUpdatedAt && item.updatedAt !== storedUpdatedAt);
@@ -17693,6 +17706,7 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
           "applied_at",
           commentUpdatedAt(existingReviewComment) ?? new Date().toISOString(),
         );
+        markdown = replaceFrontMatterValue(markdown, "apply_checked_at", new Date().toISOString());
         archiveClosed(markdown);
         closedCount += 1;
         processedCount += 1;
@@ -18031,6 +18045,7 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
         syncReasons.push("recorded existing durable comment metadata");
       }
       markdown = updateReviewCommentMetadata(markdown, syncedComment, markedReviewComment);
+      markdown = replaceFrontMatterValue(markdown, "apply_checked_at", new Date().toISOString());
       if (!dryRun) writeFileSync(path, markdown, "utf8");
       results.push({
         number,
@@ -18045,6 +18060,7 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
     }
     if (proofBlockedForCommentSync) {
       if (!needsReviewCommentSync) {
+        markdown = replaceFrontMatterValue(markdown, "apply_checked_at", new Date().toISOString());
         if (!dryRun) writeFileSync(path, markdown, "utf8");
         results.push({
           number,
@@ -18062,6 +18078,7 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
       !needsReviewCommentSync &&
       (!isCloseProposal || syncCommentsOnly)
     ) {
+      markdown = replaceFrontMatterValue(markdown, "apply_checked_at", new Date().toISOString());
       if (!dryRun) writeFileSync(path, markdown, "utf8");
       results.push({
         number,
@@ -18078,25 +18095,18 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
     }
     if (closedCount >= limit) break;
     if (applyKind !== "all" && item.kind !== applyKind) {
-      results.push({
-        number,
-        action: "kept_open",
-        reason: `type is ${item.kind}; apply kind is ${applyKind}`,
-      });
-      processedCount += 1;
-      maybeLogProgress(`skipped #${number}: type is ${item.kind}`);
-      if (processedCount >= processedLimit) break;
+      if (recordApplySkipped("kept_open", `type is ${item.kind}; apply kind is ${applyKind}`))
+        break;
       continue;
     }
     if (!closeReasonEnabled(closeReason, applyCloseReasons)) {
-      results.push({
-        number,
-        action: "kept_open",
-        reason: `close reason ${closeReason} is not enabled for this apply run`,
-      });
-      processedCount += 1;
-      maybeLogProgress(`skipped #${number}: close reason ${closeReason} not enabled`);
-      if (processedCount >= processedLimit) break;
+      if (
+        recordApplySkipped(
+          "kept_open",
+          `close reason ${closeReason} is not enabled for this apply run`,
+        )
+      )
+        break;
       continue;
     }
     const currentReportValidation = validateCloseDecision(
@@ -18130,14 +18140,7 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
       staleMinAgeDays,
     });
     if (ageSkipReason) {
-      results.push({
-        number,
-        action: "kept_open",
-        reason: ageSkipReason,
-      });
-      processedCount += 1;
-      maybeLogProgress(`skipped #${number}: ${ageSkipReason}`);
-      if (processedCount >= processedLimit) break;
+      if (recordApplySkipped("kept_open", ageSkipReason)) break;
       continue;
     }
     const prCloseCoverageBlock =
@@ -18226,6 +18229,7 @@ async function applyDecisionsCommand(args: Args): Promise<void> {
     markdown = replaceFrontMatterValue(markdown, "close_comment_sha256", sha256(reviewComment));
     markdown = replaceFrontMatterValue(markdown, "action_taken", "closed");
     markdown = replaceFrontMatterValue(markdown, "applied_at", new Date().toISOString());
+    markdown = replaceFrontMatterValue(markdown, "apply_checked_at", new Date().toISOString());
     archiveClosed(markdown);
     closedCount += 1;
     processedCount += 1;