diff --git a/.github/workflows/sweep.yml b/.github/workflows/sweep.yml index 841219817a..af98480579 100644 --- a/.github/workflows/sweep.yml +++ b/.github/workflows/sweep.yml @@ -2263,7 +2263,10 @@ jobs: progress_every="10" checkpoint_size="${{ github.event_name == 'workflow_dispatch' && github.event.inputs.apply_checkpoint_size || '5' }}" comment_sync_processed_limit=1000 - close_processed_limit=300 + base_close_processed_limit=300 + max_close_processed_limit=900 + close_processed_limit="$base_close_processed_limit" + adaptive_apply_scan_reason="base_window" sync_batch_size="${{ github.event_name == 'workflow_dispatch' && github.event.inputs.apply_limit || '25' }}" item_numbers="${{ github.event_name == 'repository_dispatch' && github.event.client_payload.item_number || github.event.inputs.apply_item_numbers || '' }}" sync_open_pr_batch="${{ github.event_name == 'schedule' && github.event.schedule == '6,21,36,51 * * * *' && 'true' || 'false' }}" @@ -2311,6 +2314,17 @@ jobs: echo "Capping apply checkpoint size at 5 to keep each GitHub App token within its lifetime." checkpoint_size=5 fi + if [ "$sync_comments_only" != "true" ] && [ -z "$item_numbers" ]; then + mkdir -p .artifacts + adaptive_batch_env=".artifacts/apply-adaptive-batch.env" + pnpm run --silent workflow -- adaptive-apply-batch-size \ + --status-path "results/sweep-status/${target_slug}.json" \ + --base-size "$base_close_processed_limit" \ + --max-size "$max_close_processed_limit" > "$adaptive_batch_env" + cat "$adaptive_batch_env" + close_processed_limit="$(awk -F= '$1 == "close_processed_limit" { print $2 }' "$adaptive_batch_env")" + adaptive_apply_scan_reason="$(awk -F= '$1 == "adaptive_apply_scan_reason" { print $2 }' "$adaptive_batch_env")" + fi mkdir -p .artifacts/apply-reports publish_changes() { message="$1" @@ -2430,7 +2444,7 @@ jobs: if [ "$proposed_count" -lt "$limit" ]; then limit="$proposed_count" fi - echo "Auto-selected $proposed_count proposed close candidate(s) from apply cursor window: $item_numbers; apply-ready count: ${apply_ready_count:-unknown}" + echo "Auto-selected $proposed_count proposed close candidate(s) from $close_processed_limit-record apply cursor window ($adaptive_apply_scan_reason): $item_numbers; apply-ready count: ${apply_ready_count:-unknown}" fi fi item_numbers_arg=() @@ -2466,7 +2480,7 @@ jobs: pnpm run status -- \ --target-repo "$TARGET_REPO" \ --state "Apply in progress" \ - --detail "Starting apply/comment-sync run for up to $limit fresh $apply_kind closes. Close reasons: $apply_close_reasons. Existing Codex automated review comments are updated in place when closing or when comment-only sync is stale by ${comment_sync_min_age_days} day(s); checkpoints commit every $checkpoint_size fresh closes; close delay is ${close_delay_ms}ms; sync-comments-only=$sync_comments_only; item numbers=${item_numbers:-all}." \ + --detail "Starting apply/comment-sync run for up to $limit fresh $apply_kind closes. Close reasons: $apply_close_reasons. Scan window: $close_processed_limit records ($adaptive_apply_scan_reason). Existing Codex automated review comments are updated in place when closing or when comment-only sync is stale by ${comment_sync_min_age_days} day(s); checkpoints commit every $checkpoint_size fresh closes; close delay is ${close_delay_ms}ms; sync-comments-only=$sync_comments_only; item numbers=${item_numbers:-all}." \ --run-url "https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}" if ! publish_changes "chore: mark sweep apply in progress" records results/sweep-status; then echo "Best-effort apply start status update failed" @@ -2659,6 +2673,8 @@ jobs: 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" + echo "APPLY_CLOSE_PROCESSED_LIMIT=$close_processed_limit" + echo "APPLY_ADAPTIVE_SCAN_REASON=$adaptive_apply_scan_reason" } >> "$GITHUB_ENV" - name: Commit apply results diff --git a/docs/scheduler.md b/docs/scheduler.md index 0606eef37b..9ef97cd3f2 100644 --- a/docs/scheduler.md +++ b/docs/scheduler.md @@ -459,6 +459,15 @@ continuation with a fresh GitHub App token after any checkpoint that closes at least one item. A saturated scan that closes nothing stops without chaining so the same records cannot create an unbounded runner loop. +Untargeted cursor-based close apply starts with a 300-record scan window. If +the previous cursor window was a full close-mode scan, closed nothing, skipped +at least 80% of processed records, and did not hit a live-fetch, runtime-budget, +or missing-cursor failure, the next automatic window expands to inspect more +records, capped at 900. This changes only the deterministic scan window: +`apply_limit`, checkpoint size, close gates, live-state checks, and maintainer +policy gates stay unchanged. The workflow logs and sweep status detail include +the selected scan window and reason. + ## Continuation and Recovery When a normal or hot review run fills its planned capacity, the publish job diff --git a/src/clawsweeper.ts b/src/clawsweeper.ts index 98203badc3..14a4ca9f0d 100644 --- a/src/clawsweeper.ts +++ b/src/clawsweeper.ts @@ -760,6 +760,7 @@ interface WorkflowStatusSummary { detail: string; runUrl: string | undefined; applyHealth: Record | undefined; + lastCloseApplyHealth: Record | undefined; plannedCount: number | undefined; plannedCapacity: number | undefined; plannedShards: number | undefined; @@ -1923,10 +1924,14 @@ function writeSweepStatus(options: { }): void { const profile = options.profile ?? targetProfile(); const updatedAt = new Date().toISOString(); + const previousStatus = readSweepStatusSummary(profile); const applyHealth = - options.applyHealth === undefined - ? readSweepStatusSummary(profile)?.applyHealth - : options.applyHealth; + options.applyHealth === undefined ? previousStatus?.applyHealth : options.applyHealth; + const previousCloseApplyHealth = + previousStatus?.lastCloseApplyHealth ?? + (previousStatus?.applyHealth?.mode === "close" ? previousStatus.applyHealth : undefined); + const lastCloseApplyHealth = + applyHealth && applyHealth.mode === "close" ? applyHealth : previousCloseApplyHealth; const payload = { schema_version: 1, slug: profile.slug, @@ -1949,6 +1954,7 @@ function writeSweepStatus(options: { bot_owned_proof_decisions_requested: options.botOwnedProofDecisionsRequested ?? null, bot_owned_proof_dispatches: options.botOwnedProofDispatches ?? null, apply_health: applyHealth ?? null, + last_close_apply_health: lastCloseApplyHealth ?? null, updated_at: updatedAt, }; const outputPath = sweepStatusPath(profile); @@ -8294,6 +8300,7 @@ function readSweepStatusSummary(profile = targetProfile()): WorkflowStatusSummar detail: stringOrUndefined(parsed.detail) ?? "No workflow status has been published yet.", runUrl: stringOrUndefined(parsed.run_url), applyHealth: recordOrUndefined(parsed.apply_health), + lastCloseApplyHealth: recordOrUndefined(parsed.last_close_apply_health), plannedCount: numberOrUndefined(parsed.planned_count), plannedCapacity: numberOrUndefined(parsed.planned_capacity), plannedShards: numberOrUndefined(parsed.planned_shards), @@ -8388,6 +8395,7 @@ function workflowStatusSummary(block: string): WorkflowStatusSummary { detail, runUrl, applyHealth: undefined, + lastCloseApplyHealth: undefined, plannedCount: numberOrUndefined(planMatch?.[1]), plannedShards: numberOrUndefined(planMatch?.[2]), plannedCapacity: numberOrUndefined(planMatch?.[3]), diff --git a/src/repair/workflow-utils.ts b/src/repair/workflow-utils.ts index fe3cc950f2..2ce666d96e 100644 --- a/src/repair/workflow-utils.ts +++ b/src/repair/workflow-utils.ts @@ -102,6 +102,25 @@ type ApplyCycleSummary = { scheduled_interval_minutes: number | null; label: string; }; + +type AdaptiveApplyBatchSizeOptions = { + statusPath: string; + baseSize: number; + maxSize: number; +}; + +type AdaptiveApplyBatchSize = { + closeProcessedLimit: number; + baseCloseProcessedLimit: number; + maxCloseProcessedLimit: number; + adaptive: boolean; + reason: string; + previousProcessed: number | null; + previousProcessedLimit: number | null; + previousClosed: number | null; + previousSkipped: number | null; +}; + const args = parseArgs(process.argv.slice(2)); function runCli(): void { @@ -158,6 +177,25 @@ function runCli(): void { )}\n`, ); break; + case "adaptive-apply-batch-size": { + const result = adaptiveApplyBatchSize({ + statusPath: requiredString("status-path"), + baseSize: numberArg("base-size", 300), + maxSize: numberArg("max-size", 900), + }); + printOutput({ + close_processed_limit: String(result.closeProcessedLimit), + base_close_processed_limit: String(result.baseCloseProcessedLimit), + max_close_processed_limit: String(result.maxCloseProcessedLimit), + adaptive_apply_scan: result.adaptive ? "true" : "false", + adaptive_apply_scan_reason: result.reason, + previous_apply_processed: optionalNumberOutput(result.previousProcessed), + previous_apply_processed_limit: optionalNumberOutput(result.previousProcessedLimit), + previous_apply_closed: optionalNumberOutput(result.previousClosed), + previous_apply_skipped: optionalNumberOutput(result.previousSkipped), + }); + break; + } case "count-command-actions": console.log( countCommandActions( @@ -308,6 +346,89 @@ function defaultCapacityReason( return "under capacity: due backlog below planned capacity"; } +export function adaptiveApplyBatchSize( + options: AdaptiveApplyBatchSizeOptions, +): AdaptiveApplyBatchSize { + const baseSize = positiveInteger(options.baseSize, "baseSize"); + const maxSize = Math.max(baseSize, positiveInteger(options.maxSize, "maxSize")); + const base: AdaptiveApplyBatchSize = { + closeProcessedLimit: baseSize, + baseCloseProcessedLimit: baseSize, + maxCloseProcessedLimit: maxSize, + adaptive: false, + reason: "base_window", + previousProcessed: null, + previousProcessedLimit: null, + previousClosed: null, + previousSkipped: null, + }; + const status = readJsonObjectIfPresent(options.statusPath); + if (!status) return base; + const health = closeApplyHealthFromStatus(status); + if (!health) return base; + const previousProcessed = nonNegativeIntegerOrNull(health.processed); + const previousProcessedLimit = nonNegativeIntegerOrNull(health.processed_limit); + const previousClosed = nonNegativeIntegerOrNull(health.closed); + const previousSkipped = nonNegativeIntegerOrNull(health.skipped); + const withPrevious = { + ...base, + previousProcessed, + previousProcessedLimit, + previousClosed, + previousSkipped, + }; + if (health.cursor_required !== true) return withPrevious; + if ( + previousProcessed === null || + previousProcessedLimit === null || + previousClosed === null || + previousSkipped === null + ) { + return withPrevious; + } + + const fullWindow = previousProcessedLimit > 0 && previousProcessed >= previousProcessedLimit; + const zeroClose = previousClosed === 0; + const skipHeavy = previousProcessed > 0 && previousSkipped / previousProcessed >= 0.8; + const attentionReasons = new Set(stringArray(health.attention_reasons)); + const unsafeAttention = [ + "cursor_required_but_missing_after_full_window", + "skipped_live_fetch_failed", + "skipped_runtime_budget", + ].some((reason) => attentionReasons.has(reason)); + if (!fullWindow || !zeroClose || !skipHeavy || unsafeAttention) return withPrevious; + + const grown = Math.min(maxSize, Math.max(baseSize, previousProcessedLimit * 2)); + if (grown <= baseSize) return withPrevious; + return { + ...withPrevious, + closeProcessedLimit: grown, + adaptive: true, + reason: "previous_full_zero_close_skip_window", + }; +} + +function closeApplyHealthFromStatus(status: LooseRecord): LooseRecord | null { + const current = recordOrNull(status.apply_health); + if (current?.mode === "close") return current; + const preserved = recordOrNull(status.last_close_apply_health); + return preserved?.mode === "close" ? preserved : null; +} + +function optionalNumberOutput(value: number | null): string { + return value === null ? "" : String(value); +} + +function positiveInteger(value: number, name: string): number { + if (!Number.isInteger(value) || value < 1) throw new Error(`${name} must be a positive integer`); + return value; +} + +function nonNegativeIntegerOrNull(value: JsonValue | undefined): number | null { + const parsed = Number(value); + return Number.isInteger(parsed) && parsed >= 0 ? parsed : null; +} + export function plannedItemNumberCsv(plan: LooseRecord): string { const candidates: JsonValue[] = Array.isArray(plan.candidates) ? plan.candidates : []; return candidates @@ -1359,6 +1480,19 @@ function readJsonObject(filePath: string): LooseRecord { return parsed; } +function readJsonObjectIfPresent(filePath: string): LooseRecord | null { + if (!filePath || !fs.existsSync(filePath)) return null; + try { + return readJsonObject(filePath); + } catch { + return null; + } +} + +function recordOrNull(value: JsonValue | undefined): LooseRecord | null { + return isJsonObject(value) ? value : null; +} + function printOutput(values: Record): void { for (const [key, value] of Object.entries(values)) console.log(`${key}=${value}`); } @@ -1416,7 +1550,7 @@ function positiveNumber(value: string, fallback: number): number { return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; } -function stringArray(value: JsonValue): string[] { +function stringArray(value: JsonValue | undefined): string[] { return Array.isArray(value) ? value.filter((entry): entry is string => typeof entry === "string") : []; diff --git a/test/clawsweeper.test.ts b/test/clawsweeper.test.ts index c23b411bdf..8ff8af0afb 100644 --- a/test/clawsweeper.test.ts +++ b/test/clawsweeper.test.ts @@ -1996,6 +1996,9 @@ test("sweep status writer preserves non-apply health and clears stale apply upda /state\.startsWith\("Apply "\)\s+\?\s+null\s+:\s+readSweepStatusSummary\(profile\)\?\.applyHealth/, ); assert.match(source, /apply_health: applyHealth \?\? null/); + assert.match(source, /last_close_apply_health: lastCloseApplyHealth \?\? null/); + assert.match(source, /previousStatus\?\.lastCloseApplyHealth/); + assert.match(source, /previousStatus\?\.applyHealth\?\.mode === "close"/); }); test("review parser strips environment access caveats from risks", () => { diff --git a/test/repair/workflow-utils.test.ts b/test/repair/workflow-utils.test.ts index 51423feb6f..8dd84adae3 100644 --- a/test/repair/workflow-utils.test.ts +++ b/test/repair/workflow-utils.test.ts @@ -7,6 +7,7 @@ import test from "node:test"; import { applyCursorAdvanceCount, + adaptiveApplyBatchSize, artifactItemNumbers, automationLimit, commentSyncBatchOutput, @@ -639,6 +640,136 @@ test("workflow utilities expose review capacity telemetry from plans", () => { ); }); +test("workflow utilities expand automatic apply scan after a skip-heavy zero-close window", () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "clawsweeper-workflow-")); + const statusPath = path.join(root, "results/sweep-status/openclaw-openclaw.json"); + write( + statusPath, + JSON.stringify({ + apply_health: { + mode: "close", + cursor_required: true, + processed: 300, + processed_limit: 300, + closed: 0, + skipped: 285, + attention_reasons: ["skipped_changed_since_review"], + }, + }), + ); + + const result = adaptiveApplyBatchSize({ statusPath, baseSize: 300, maxSize: 900 }); + + assert.equal(result.closeProcessedLimit, 600); + assert.equal(result.adaptive, true); + assert.equal(result.reason, "previous_full_zero_close_skip_window"); + assert.equal(result.previousProcessed, 300); + assert.equal(result.previousSkipped, 285); +}); + +test("workflow utilities use preserved close health after comment-sync status updates", () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "clawsweeper-workflow-")); + const statusPath = path.join(root, "results/sweep-status/openclaw-openclaw.json"); + write( + statusPath, + JSON.stringify({ + apply_health: { + mode: "comment_sync", + cursor_required: true, + processed: 25, + processed_limit: 25, + closed: 0, + skipped: 0, + attention_reasons: [], + }, + last_close_apply_health: { + mode: "close", + cursor_required: true, + processed: 300, + processed_limit: 300, + closed: 0, + skipped: 300, + attention_reasons: ["skipped_changed_since_review"], + }, + }), + ); + + const result = adaptiveApplyBatchSize({ statusPath, baseSize: 300, maxSize: 900 }); + + assert.equal(result.closeProcessedLimit, 600); + assert.equal(result.adaptive, true); + assert.equal(result.previousProcessed, 300); +}); + +test("workflow utilities cap adaptive apply scan and reset on productive or unsafe windows", () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "clawsweeper-workflow-")); + const statusPath = path.join(root, "results/sweep-status/openclaw-openclaw.json"); + const size = () => adaptiveApplyBatchSize({ statusPath, baseSize: 300, maxSize: 900 }); + + write( + statusPath, + JSON.stringify({ + apply_health: { + mode: "close", + cursor_required: true, + processed: 600, + processed_limit: 600, + closed: 0, + skipped: 600, + attention_reasons: ["skipped_protected_label"], + }, + }), + ); + assert.equal(size().closeProcessedLimit, 900); + + write( + statusPath, + JSON.stringify({ + apply_health: { + mode: "close", + cursor_required: true, + processed: 300, + processed_limit: 300, + closed: 1, + skipped: 299, + attention_reasons: [], + }, + }), + ); + assert.deepEqual( + { limit: size().closeProcessedLimit, reason: size().reason, adaptive: size().adaptive }, + { limit: 300, reason: "base_window", adaptive: false }, + ); + + write( + statusPath, + JSON.stringify({ + apply_health: { + mode: "close", + cursor_required: true, + processed: 300, + processed_limit: 300, + closed: 0, + skipped: 300, + attention_reasons: ["skipped_live_fetch_failed"], + }, + }), + ); + assert.deepEqual( + { limit: size().closeProcessedLimit, reason: size().reason, adaptive: size().adaptive }, + { limit: 300, reason: "base_window", adaptive: false }, + ); + + assert.equal( + adaptiveApplyBatchSize({ + statusPath: path.join(root, "missing.json"), + baseSize: 300, + maxSize: 900, + }).closeProcessedLimit, + 300, + ); +}); + test("workflow utilities select eligible proposed close records", () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), "clawsweeper-workflow-")); const oldDate = "2024-01-01T00:00:00Z"; diff --git a/test/sweep-workflow.test.ts b/test/sweep-workflow.test.ts index 5baef57c86..b6411598ac 100644 --- a/test/sweep-workflow.test.ts +++ b/test/sweep-workflow.test.ts @@ -139,7 +139,11 @@ test("apply workflow bounds checkpoints and requeues with a fresh token", () => assert.match(inputBlock, /apply_limit:[\s\S]*default: "5"/); assert.match(inputBlock, /apply_checkpoint_size:[\s\S]*default: "5"/); assert.match(applyStep, /Capping apply checkpoint size at 5/); - assert.match(applyStep, /close_processed_limit=300/); + assert.match(applyStep, /base_close_processed_limit=300/); + assert.match(applyStep, /max_close_processed_limit=900/); + assert.match(applyStep, /close_processed_limit="\$base_close_processed_limit"/); + assert.match(applyStep, /adaptive-apply-batch-size/); + assert.match(applyStep, /--status-path "results\/sweep-status\/\$\{target_slug\}\.json"/); assert.match(applyStep, /processed-limit "\$close_processed_limit"/); assert.match(applyStep, /comment_sync_processed_limit=1000/); assert.match(applyStep, /--processed-limit "\$comment_sync_processed_limit"/); @@ -163,6 +167,14 @@ test("apply workflow bounds checkpoints and requeues with a fresh token", () => assert.match(applyStep, /--apply-health-file "\.artifacts\/apply-health-final\.json"/); assert.match(applyStep, /--state "Apply idle"/); assert.match(applyStep, /--batch-size "\$close_processed_limit"/); + assert.match( + applyStep, + /Scan window: \$close_processed_limit records \(\$adaptive_apply_scan_reason\)/, + ); + assert.match( + applyStep, + /Auto-selected \$proposed_count proposed close candidate\(s\) from \$close_processed_limit-record apply cursor window/, + ); assert.match(applyStep, /--cursor-path "\$apply_cursor_path"/); assert.match(applyStep, /write-apply-cursor/); assert.match(applyStep, /--item-numbers "\$item_numbers"/);