diff --git a/server/src/__tests__/heartbeat-rate-limit-exhausted.test.ts b/server/src/__tests__/heartbeat-rate-limit-exhausted.test.ts index 5f7aab360cda..4c549994a498 100644 --- a/server/src/__tests__/heartbeat-rate-limit-exhausted.test.ts +++ b/server/src/__tests__/heartbeat-rate-limit-exhausted.test.ts @@ -13,8 +13,10 @@ import { countConsecutiveZeroTokenCompletedRuns, isRateLimitExhausted, isRetryableK8sCcrotateThrottleResult, + K8S_REPLACEMENT_LAUNCH_FAILURE_AFTER_THROTTLE_KEY, k8sCcrotateRetryDelayMs, listRecentTerminalRunsForZeroTokenStreak, + reclassifyK8sReplacementLaunchFailureAfterThrottle, } from "../services/heartbeat.js"; import { getEmbeddedPostgresTestSupport, @@ -401,6 +403,118 @@ describe("k8s ccrotate no-progress throttle detection", () => { }); }); +// BLO-34577: the in-run throttle loop relaunches the Job for the same run. On +// 2026-09-18 the relaunch read the previous attempt's Failed pod and returned +// `k8s_pod_schedule_failed` ~100 ms after create; the finalizer recorded that +// code and the pr_review run was dropped without a retry, while the identical +// 429 seen through the throttle path retried. These pin the server-side verdict. +describe("reclassifyK8sReplacementLaunchFailureAfterThrottle (BLO-34577)", () => { + const TENANT_429 = + 'API Error: 429 {"type":"error","error":{"type":"rate_limit_error","message":"All Claude subscription capacity for this tenant is rate-limited"}}'; + const zeroUsage = { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }; + const throttleResult = { + exitCode: 1, + signal: null, + timedOut: false, + errorMessage: TENANT_429, + errorCode: null, + retryNotBefore: "2026-09-18T12:40:00.000Z", + resultJson: { api_error_status: 429, is_error: true }, + usage: zeroUsage, + }; + const launchFailure = { + exitCode: null, + signal: null, + timedOut: false, + errorMessage: + "Pod scheduling failed: Pod ac-ally-96fa0c75-3f2a1b-x9k2q reached phase=Failed: claude exited 1", + errorCode: "k8s_pod_schedule_failed", + }; + + it("finalizes a replacement-launch failure after an in-run throttle with the throttle verdict", () => { + const reclassified = reclassifyK8sReplacementLaunchFailureAfterThrottle({ + launchResult: launchFailure, + throttleResult, + throttleAttempts: 2, + }); + expect(reclassified).not.toBeNull(); + // The verdict is the throttle's: the code the finalizer treats as terminal is gone... + expect(reclassified!.errorCode).not.toBe("k8s_pod_schedule_failed"); + // ...and the result still classifies as the in-run throttle the loop was retrying, + // so the finalizer takes the provider_throttled_no_progress / rate_limit_exhausted arm. + expect(isRetryableK8sCcrotateThrottleResult(reclassified!)).toBe(true); + expect(reclassified!.resultJson).toMatchObject({ api_error_status: 429, is_error: true }); + expect(reclassified!.retryNotBefore).toBe("2026-09-18T12:40:00.000Z"); + // The launch failure is kept as an annotation, not lost. + expect(reclassified!.resultJson?.[K8S_REPLACEMENT_LAUNCH_FAILURE_AFTER_THROTTLE_KEY]).toEqual({ + errorCode: "k8s_pod_schedule_failed", + errorMessage: launchFailure.errorMessage, + throttleAttempts: 2, + throttleErrorCode: null, + }); + expect(reclassified!.errorMessage).toContain("rate-limited"); + expect(reclassified!.errorMessage).toContain("after 2 in-run throttle retries"); + expect(reclassified!.errorMessage).toContain("phase=Failed: claude exited 1"); + }); + + it("leaves an ambiguous k8s_pod_schedule_failed alone when no throttle preceded it", () => { + // Negative control: with no observed throttle the launch failure means what it + // says, and the existing "does not retry ambiguous k8s_pod_schedule_failed" + // contract must keep applying to it. + expect( + reclassifyK8sReplacementLaunchFailureAfterThrottle({ + launchResult: launchFailure, + throttleResult: null, + throttleAttempts: 0, + }), + ).toBeNull(); + expect( + reclassifyK8sReplacementLaunchFailureAfterThrottle({ + launchResult: launchFailure, + throttleResult, + throttleAttempts: 0, + }), + ).toBeNull(); + }); + + it("only reclassifies a launch failure, never another terminal result", () => { + for (const launchResult of [ + { exitCode: 0, signal: null, timedOut: false, usage: { inputTokens: 40, outputTokens: 12 } }, + { exitCode: 1, signal: null, timedOut: false, errorCode: "adapter_failed", errorMessage: "boom" }, + { exitCode: null, signal: null, timedOut: false, errorCode: "k8s_concurrent_run_blocked" }, + ]) { + expect( + reclassifyK8sReplacementLaunchFailureAfterThrottle({ + launchResult, + throttleResult, + throttleAttempts: 1, + }), + ).toBeNull(); + } + }); + + it("does not let a non-throttle prior result stand in as the verdict", () => { + // The loop never retries a result with token usage, so a prior result that + // made progress is not a throttle chain; do not manufacture one. + expect( + reclassifyK8sReplacementLaunchFailureAfterThrottle({ + launchResult: launchFailure, + throttleResult: { ...throttleResult, usage: { inputTokens: 12, outputTokens: 3 } }, + throttleAttempts: 1, + }), + ).toBeNull(); + // Nor a launch failure that somehow reports usage: the pod is claimed to + // have never run, so this is not the shape the reclassifier understands. + expect( + reclassifyK8sReplacementLaunchFailureAfterThrottle({ + launchResult: { ...launchFailure, usage: { inputTokens: 1, outputTokens: 0 } }, + throttleResult, + throttleAttempts: 1, + }), + ).toBeNull(); + }); +}); + describe("countConsecutiveZeroTokenCompletedRuns", () => { it("counts only the newest terminal zero-token prefix", () => { expect(countConsecutiveZeroTokenCompletedRuns([ diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 011828032d5b..a35daf720b15 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -33,6 +33,7 @@ import { INTERACTION_CONTINUATION_INFRA_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_WAKE_REASON, JOB_FAILED_HEARTBEAT_RETRY_MAX_ATTEMPTS, + K8S_REPLACEMENT_LAUNCH_FAILURE_AFTER_THROTTLE_KEY, MAX_TURN_CONTINUATION_RETRY_REASON, MAX_TURN_CONTINUATION_WAKE_REASON, heartbeatService, @@ -3245,6 +3246,51 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { } }); + // BLO-34577: a replacement Job launch that failed before its pod ran, after + // this run already observed a zero-progress 429, is finalized with the throttle + // verdict (provider_throttled_no_progress / rate_limit_exhausted) carrying the + // launch failure as an annotation. That verdict is what re-queues the pr_review + // -- previously the run kept `k8s_pod_schedule_failed` and the review dropped. + describe("BLO-34577 replacement-launch failure after an in-run throttle", () => { + const annotation = { + [K8S_REPLACEMENT_LAUNCH_FAILURE_AFTER_THROTTLE_KEY]: { + errorCode: "k8s_pod_schedule_failed", + errorMessage: "Pod scheduling failed: Pod ac-ally-x reached phase=Failed: claude exited 1", + throttleAttempts: 2, + throttleErrorCode: null, + }, + }; + + it("schedules the bounded retry for pr_review and issue contexts once finalized as the throttle", () => { + for (const contextSnapshot of [ + { wakeReason: "github_pr_opened", reviewKind: "pr_review", githubPrNumber: 3212 }, + { taskKey: "pr_review:Blockcast/pim-multicast-gateway:3212" }, + { issueId: randomUUID(), wakeReason: "issue_assigned" }, + ]) { + expect( + shouldScheduleAutomaticRunRetry({ + errorCode: "provider_throttled_no_progress", + resultJson: { errorFamily: "rate_limit_exhausted", api_error_status: 429, ...annotation }, + contextSnapshot, + }), + ).toBe(true); + } + }); + + it("does not retry when the launch failure was recorded verbatim, annotation or not", () => { + // Negative control: the fix is the VERDICT, not the annotation. A run that + // kept `k8s_pod_schedule_failed` -- and a stale transient family in its + // merged resultJson -- still hits the ambiguous-outcome reject first. + expect( + shouldScheduleAutomaticRunRetry({ + errorCode: "k8s_pod_schedule_failed", + resultJson: { errorFamily: "rate_limit_exhausted", ...annotation }, + contextSnapshot: { wakeReason: "github_pr_opened", reviewKind: "pr_review", githubPrNumber: 3212 }, + }), + ).toBe(false); + }); + }); + // BLO-17456: when a PR-review chain exhausts, the reviewer never posts its // required status, so the PR sits on "Expected — waiting for status" forever. // These drive the real exhaustion path (no mocks): loadConfig() reads diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 9cd25901f547..3cdb22120ad5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -5422,6 +5422,70 @@ export function k8sCcrotateRetryDelayMs(result: { retryNotBefore?: string | null ); } +// BLO-34577: the in-run ccrotate throttle loop re-invokes the k8s adapter for +// the SAME runId after a zero-progress 429. Each replacement attempt is a fresh +// Job launch, and that launch can itself fail before any pod runs +// (`k8s_pod_schedule_failed`). Observed 2026-09-18 on 7 Ally pr_review runs: +// the adapter read the PREVIOUS attempt's Failed pod -- same deterministic Job +// name, deleted in the background -- as the replacement's ~100 ms after create +// and returned `k8s_pod_schedule_failed`. The loop broke on that non-throttle +// result and the finalizer recorded the code verbatim, which +// `shouldScheduleAutomaticRunRetry` and `isNonRetryablePrReviewTerminalOutcome` +// treat as terminal: no retry was minted, the review was dropped, and the gate +// posted `non_retryable_external_lifecycle`. The identical 429 finalized through +// the throttle path retries on the flat rate-limit curve. +// +// The adapter no longer reads a foreign pod (it scopes lookups to the created +// Job's UID), but the server verdict must not depend on that: once this run has +// observed >= 1 zero-progress throttle, a replacement launch that fails before +// its pod runs adds no information about the WORK -- no attempt made model +// progress (the loop only retries zero-token results) and this attempt's pod +// never ran -- so the run's cause is still the throttle. Finalize with the +// throttle verdict and record the launch failure as an annotation, so the +// `provider_throttled_no_progress` / `rate_limit_exhausted` path schedules the +// bounded retry it would have scheduled had the loop simply exhausted. +// +// Deliberately narrow: a `k8s_pod_schedule_failed` with NO prior throttle in +// this run is left exactly as reported. That outcome is ambiguous, and the +// "does not retry ambiguous k8s_pod_schedule_failed" contract still holds. +export const K8S_REPLACEMENT_LAUNCH_FAILURE_AFTER_THROTTLE_KEY = + "replacementLaunchFailureAfterThrottle" as const; + +export function reclassifyK8sReplacementLaunchFailureAfterThrottle(input: { + launchResult: AdapterExecutionResult; + throttleResult: AdapterExecutionResult | null; + throttleAttempts: number; +}): AdapterExecutionResult | null { + const { launchResult, throttleResult, throttleAttempts } = input; + if (!throttleResult || throttleAttempts < 1) return null; + if (launchResult.errorCode !== "k8s_pod_schedule_failed") return null; + // A pod that never ran cannot have made model progress. If usage is reported + // anyway this is not the shape described above; leave the verdict alone. + if (!zeroTokenUsage(launchResult.usage)) return null; + // Only a result the loop itself judged a retryable throttle may stand in as + // the verdict; anything else and this was not a throttle chain. + if (!isRetryableK8sCcrotateThrottleResult(throttleResult)) return null; + const launchErrorMessage = readNonEmptyString(launchResult.errorMessage) ?? launchResult.errorCode; + const throttleErrorMessage = + readNonEmptyString(throttleResult.errorMessage) ?? "Provider throttled before model progress"; + const retryNoun = throttleAttempts === 1 ? "retry" : "retries"; + return { + ...throttleResult, + errorMessage: + `${throttleErrorMessage} (replacement Job launch after ${throttleAttempts} in-run throttle ` + + `${retryNoun} failed before its pod ran: ${launchErrorMessage})`, + resultJson: { + ...(throttleResult.resultJson ?? {}), + [K8S_REPLACEMENT_LAUNCH_FAILURE_AFTER_THROTTLE_KEY]: { + errorCode: launchResult.errorCode, + errorMessage: launchResult.errorMessage ?? null, + throttleAttempts, + throttleErrorCode: throttleResult.errorCode ?? null, + }, + }, + }; +} + // The pinned claude_k8s/opencode_k8s adapters report their launch command as // this exact "kubectl job/" sentinel (not the real invoked command) so // the reservation can learn the expected Job name before the post-create @@ -31155,6 +31219,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } } let ccrotateRetryAttempt = 0; + // BLO-34577: the most recent result the loop judged a retryable + // throttle. A replacement launch that fails before its pod runs is + // finalized with THIS verdict, not the launch failure's. + let lastInRunThrottleResult: Awaited> | null = null; while (true) { const executionReservation = externalRuntimeReservation; if (executionReservation) { @@ -31240,6 +31308,39 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) !isRetryableK8sCcrotateThrottleResult(adapterResult) || ccrotateRetryAttempt >= K8S_CCROTATE_IN_RUN_RETRY_MAX_ATTEMPTS ) { + // BLO-34577: a replacement Job whose launch failed before its + // pod ran, after this run already observed a zero-progress + // throttle, is still the throttle -- see + // reclassifyK8sReplacementLaunchFailureAfterThrottle. Without + // this the run finalized as `k8s_pod_schedule_failed`, which is + // terminal for pr_review, and the review was silently dropped. + const reclassified = isK8sAdapter(agent.adapterType) + ? reclassifyK8sReplacementLaunchFailureAfterThrottle({ + launchResult: adapterResult, + throttleResult: lastInRunThrottleResult, + throttleAttempts: ccrotateRetryAttempt, + }) + : null; + if (reclassified) { + await appendRunEvent(currentRun, seq++, { + eventType: "lifecycle", + stream: "system", + level: "warn", + message: + "replacement Job launch failed after an in-run ccrotate throttle; finalizing with the throttle verdict so the bounded retry is scheduled", + payload: { + launchErrorCode: adapterResult.errorCode ?? null, + launchErrorMessage: adapterResult.errorMessage ?? null, + throttleAttempts: ccrotateRetryAttempt, + maxAttempts: K8S_CCROTATE_IN_RUN_RETRY_MAX_ATTEMPTS, + }, + }); + await onLog( + "stderr", + `[paperclip] Replacement Job launch failed after ${ccrotateRetryAttempt} in-run throttle ${ccrotateRetryAttempt === 1 ? "retry" : "retries"}; recording the run as provider-throttled so it is retried.\n`, + ); + adapterResult = reclassified; + } break; } // BLO-18278: if the provider advertised a reset the in-run loop @@ -31294,6 +31395,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) break; } ccrotateRetryAttempt += 1; + lastInRunThrottleResult = adapterResult; const retryDelayMs = k8sCcrotateRetryDelayMs(adapterResult); if (externalRuntimeReservation) { // The completed Job belongs to the attempt that just returned. diff --git a/vendor/paperclip-adapter-claude-k8s/PROVENANCE.md b/vendor/paperclip-adapter-claude-k8s/PROVENANCE.md index 2cce96d7e391..68a7fcd15681 100644 --- a/vendor/paperclip-adapter-claude-k8s/PROVENANCE.md +++ b/vendor/paperclip-adapter-claude-k8s/PROVENANCE.md @@ -14,7 +14,7 @@ control plane. | Repository vendored from | | | Package | `paperclip-adapter-claude-k8s` | | Version at vendor time | `0.2.5-kkroo.6` | -| Current version | `0.2.6-blockcast.10` — see [Versioning](#versioning) | +| Current version | `0.2.6-blockcast.11` — see [Versioning](#versioning) | | Declared license | MIT, in `package.json` only — see the caveat below | Before this change the image built this package by cloning that repository at a @@ -96,7 +96,7 @@ A manifest of `sha256(path)` over all 41 in-tree files, sorted by path under `LC_ALL=C`, itself hashes to: ``` -7a91abbddc0522fe48ba8480d69be634600217c065341481efc1c1bb3cece4d3 +8c3a5f3d741ff0567bbe9f4a33ff7e9b520396dbc3cebe635d12d23b5f81fdf4 ``` Regenerate with: @@ -193,7 +193,7 @@ after the first Blockcast change that ships, the version alone could no longer tell you which code was running — provenance had to be established by grepping `dist/` for a token. -This directory therefore versions itself: **`0.2.6-blockcast.10`**, set in +This directory therefore versions itself: **`0.2.6-blockcast.11`**, set in `package.json` and `package-lock.json`. The `-blockcast.` prerelease channel says plainly that this is our tree, not an upstream release. diff --git a/vendor/paperclip-adapter-claude-k8s/package-lock.json b/vendor/paperclip-adapter-claude-k8s/package-lock.json index 3812bb532001..aca813770d94 100644 --- a/vendor/paperclip-adapter-claude-k8s/package-lock.json +++ b/vendor/paperclip-adapter-claude-k8s/package-lock.json @@ -1,12 +1,12 @@ { "name": "paperclip-adapter-claude-k8s", - "version": "0.2.6-blockcast.10", + "version": "0.2.6-blockcast.11", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "paperclip-adapter-claude-k8s", - "version": "0.2.6-blockcast.10", + "version": "0.2.6-blockcast.11", "license": "MIT", "dependencies": { "@kubernetes/client-node": "^1.0.0", diff --git a/vendor/paperclip-adapter-claude-k8s/package.json b/vendor/paperclip-adapter-claude-k8s/package.json index 6202459bbaa7..63b0c1385268 100644 --- a/vendor/paperclip-adapter-claude-k8s/package.json +++ b/vendor/paperclip-adapter-claude-k8s/package.json @@ -1,6 +1,6 @@ { "name": "paperclip-adapter-claude-k8s", - "version": "0.2.6-blockcast.10", + "version": "0.2.6-blockcast.11", "description": "Paperclip adapter plugin that runs Claude Code agents as Kubernetes Jobs", "license": "MIT", "repository": { diff --git a/vendor/paperclip-adapter-claude-k8s/src/server/execute.test.ts b/vendor/paperclip-adapter-claude-k8s/src/server/execute.test.ts index bbbfb4806f9e..ea22fee92d7a 100644 --- a/vendor/paperclip-adapter-claude-k8s/src/server/execute.test.ts +++ b/vendor/paperclip-adapter-claude-k8s/src/server/execute.test.ts @@ -92,6 +92,7 @@ const { describeTruncationCause, extractContainerLogDiagnostic, shouldAbortForCancellation, + selectJobOwnedPod, execute, } = await import("./execute.js"); @@ -537,6 +538,16 @@ describe("execute: all-invalid agent.id (N4)", () => { // ─── Helpers shared across execute() integration tests ─────────────────────── +/** + * ownerReferences entry the Job controller stamps on every pod it creates. + * Pod fixtures must carry it (or a controller-uid label) to be read as THIS + * execution's pod — a same-name pod without it is a stale earlier attempt and + * is ignored (BLO-34577, `selectJobOwnedPod`). + */ +function jobOwnerRef(uid: string, name = "ac-job") { + return { apiVersion: "batch/v1", kind: "Job", name, uid, controller: true, blockOwnerDeletion: true }; +} + function makeCtx(overrides: Partial = {}): AdapterExecutionContext { return { runId: "run-test-001", @@ -1498,7 +1509,7 @@ describe("execute: job creation", () => { mockCoreListPods.mockResolvedValue({ items: [ { - metadata: { name: "pod-xyz" }, + metadata: { name: "pod-xyz", ownerReferences: [jobOwnerRef("uid-1")] }, status: { phase: "Pending", conditions: [ @@ -1534,7 +1545,7 @@ describe("execute: waitForPod edge cases", () => { it("throws k8s_pod_schedule_failed when pod reaches phase=Failed immediately", async () => { mockCoreListPods.mockResolvedValue({ items: [{ - metadata: { name: "pod-fail" }, + metadata: { name: "pod-fail", ownerReferences: [jobOwnerRef("uid-1")] }, status: { phase: "Failed", containerStatuses: [{ name: "claude", state: { terminated: { exitCode: 137, reason: "OOMKilled" } } }], @@ -1552,7 +1563,7 @@ describe("execute: waitForPod edge cases", () => { it("uses the startup timeout after the pod is already scheduled", async () => { mockCoreListPods.mockResolvedValue({ items: [{ - metadata: { name: "pod-starting" }, + metadata: { name: "pod-starting", ownerReferences: [jobOwnerRef("uid-1")] }, spec: { nodeName: "k8s-paperclip-1" }, status: { phase: "Pending", @@ -1585,7 +1596,7 @@ describe("execute: waitForPod edge cases", () => { it("throws k8s_pod_schedule_failed when init container exits non-zero", async () => { mockCoreListPods.mockResolvedValue({ items: [{ - metadata: { name: "pod-x" }, + metadata: { name: "pod-x", ownerReferences: [jobOwnerRef("uid-1")] }, status: { phase: "Pending", initContainerStatuses: [{ @@ -1606,7 +1617,7 @@ describe("execute: waitForPod edge cases", () => { it("throws k8s_pod_schedule_failed when init container has ImagePullBackOff", async () => { mockCoreListPods.mockResolvedValue({ items: [{ - metadata: { name: "pod-x" }, + metadata: { name: "pod-x", ownerReferences: [jobOwnerRef("uid-1")] }, status: { phase: "Pending", initContainerStatuses: [{ @@ -1627,7 +1638,7 @@ describe("execute: waitForPod edge cases", () => { it("throws k8s_pod_schedule_failed when main container has CrashLoopBackOff", async () => { mockCoreListPods.mockResolvedValue({ items: [{ - metadata: { name: "pod-x" }, + metadata: { name: "pod-x", ownerReferences: [jobOwnerRef("uid-1")] }, status: { phase: "Pending", initContainerStatuses: [], @@ -1644,6 +1655,199 @@ describe("execute: waitForPod edge cases", () => { expect(result.errorCode).toBe("k8s_pod_schedule_failed"); expect(result.errorMessage).toContain("crash loop"); }); + + // ── BLO-34577: a same-name pod from an EARLIER attempt is not this attempt ── + // + // The Job name is deterministic per (agentId, runId) and the server's in-run + // ccrotate throttle loop re-invokes execute() for the same runId, so the + // replacement Job shares its name with the attempt that just returned. That + // attempt's Job was deleted with propagationPolicy=Background, which leaves + // its Failed pod (claude exit 1 from the 429) matching `job-name=` for a + // while. Reading items[0] surfaced that stale terminal state as THIS + // attempt's k8s_pod_schedule_failed ~100 ms after create — a code the server + // treats as non-retryable — and the PR review was dropped (7 runs on + // 2026-09-18). These drive the real execute() path. + const stale429Pod = () => ({ + metadata: { + name: "ac-job-prev-attempt", + ownerReferences: [jobOwnerRef("uid-from-the-previous-throttled-attempt")], + labels: { "job-name": "ac-job", "controller-uid": "uid-from-the-previous-throttled-attempt" }, + }, + status: { + phase: "Failed", + containerStatuses: [{ name: "claude", state: { terminated: { exitCode: 1, reason: "Error" } } }], + initContainerStatuses: [], + }, + }); + + it("does not read a stale same-name pod from the previous attempt as this attempt's terminal state", async () => { + // Poll 1: only the stale pod exists (the controller has not created ours + // yet). Poll 2+: ours exists too, and is unschedulable — a deterministic, + // short way out of waitForPod that is unmistakably about the NEW pod. + mockCoreListPods + .mockResolvedValueOnce({ items: [stale429Pod()] }) + .mockResolvedValue({ + items: [ + stale429Pod(), + { + metadata: { name: "ac-job-this-attempt", ownerReferences: [jobOwnerRef("uid-1")] }, + status: { + phase: "Pending", + conditions: [ + { type: "PodScheduled", status: "False", reason: "Unschedulable", message: "0/3 nodes are available" }, + ], + containerStatuses: [], + initContainerStatuses: [], + }, + }, + ], + }); + const ctx = makeCtx(); + + const result = await execute(ctx); + + // The verdict is about OUR pod, never the stale one's `claude exited 1`. + expect(result.errorCode).toBe("k8s_pod_schedule_failed"); + expect(result.errorMessage).toContain("unschedulable"); + expect(result.errorMessage).not.toContain("claude exited 1"); + expect(result.errorMessage).not.toContain("ac-job-prev-attempt"); + expect(mockCoreListPods.mock.calls.length).toBeGreaterThanOrEqual(2); + expect(ctx.onLog).toHaveBeenCalledWith( + "stdout", + expect.stringContaining("Ignoring 1 pod(s) named for Job"), + ); + expect(ctx.onLog).toHaveBeenCalledWith( + "stdout", + expect.stringContaining("uid-from-the-previous-throttled-attempt"), + ); + }); + + it("treats a listing that holds ONLY stale pods as 'no pod yet', not as a Failed pod", async () => { + // With podScheduleTimeoutSec=0 the schedule deadline is already past on the + // first poll, so the only way out is the no-pod timeout branch. Under the + // old items[0] read this returned "reached phase=Failed: claude exited 1". + mockCoreListPods.mockResolvedValue({ items: [stale429Pod()] }); + + const result = await execute(makeCtx({ config: { podScheduleTimeoutSec: 0 } } as Partial)); + + expect(result.errorCode).toBe("k8s_pod_schedule_failed"); + expect(result.errorMessage).toContain("Timed out waiting for pod to be scheduled"); + expect(result.errorMessage).not.toContain("claude exited 1"); + }); + + it("still reads THIS attempt's own Failed pod as a failure (negative control)", async () => { + // Same terminal state as the stale fixture, but owned by the Job we + // created: this MUST still surface, or the fix would have hidden every + // genuine fast crash behind "no pod yet". + mockCoreListPods.mockResolvedValue({ + items: [{ + metadata: { name: "ac-job-this-attempt", ownerReferences: [jobOwnerRef("uid-1")] }, + status: { + phase: "Failed", + containerStatuses: [{ name: "claude", state: { terminated: { exitCode: 1, reason: "Error" } } }], + initContainerStatuses: [], + }, + }], + }); + + const result = await execute(makeCtx()); + + expect(result.errorCode).toBe("k8s_pod_schedule_failed"); + expect(result.errorMessage).toContain("ac-job-this-attempt reached phase=Failed: claude exited 1"); + }); + + it("scopes ownership to the ADOPTED Job's uid on the worker-restart reattach path", async () => { + // BLO-27155 reattach: the create 409s and the run adopts its own live Job. + // The pod lookup must then key on the adopted uid, not on a create result. + const jobName = "ac-adopted"; + const liveUid = "live-uid-from-before-the-restart"; + mockBatchCreateJob.mockRejectedValueOnce( + new ApiException(409, "Conflict", { kind: "Status", status: "Failure", reason: "AlreadyExists", code: 409 }, {}) as unknown as Error, + ); + mockBatchReadJob.mockResolvedValue(makeJob({ name: jobName, uid: liveUid, runId: "run-test-001", agentId: "agent-abc" })); + mockCoreListPods.mockResolvedValue({ + items: [ + stale429Pod(), + { + metadata: { name: "ac-adopted-pod", ownerReferences: [jobOwnerRef(liveUid, jobName)] }, + status: { + phase: "Pending", + conditions: [{ type: "PodScheduled", status: "False", reason: "Unschedulable", message: "no nodes" }], + containerStatuses: [], + initContainerStatuses: [], + }, + }, + ], + }); + // The adapter builds its own deterministic name; the reservation must + // agree with it for adoption. Read it back from the create call by probing + // with a rejected create, exactly as the BLO-27155 suite does. + const probe = await execute(makeCtx({ + externalRuntime: { reservationId: "r", slotId: 0, jobName, jobUid: liveUid }, + } as unknown as Partial)); + const builtName = mockBatchCreateJob.mock.calls[0]?.[0]?.body?.metadata?.name as string; + expect(builtName).toBeTruthy(); + expect(probe.errorCode).toBe("k8s_job_create_failed"); // name mismatch => refused, as designed + mockBatchCreateJob.mockRejectedValueOnce( + new ApiException(409, "Conflict", { kind: "Status", status: "Failure", reason: "AlreadyExists", code: 409 }, {}) as unknown as Error, + ); + mockBatchReadJob.mockResolvedValue(makeJob({ name: builtName, uid: liveUid, runId: "run-test-001", agentId: "agent-abc" })); + + const result = await execute(makeCtx({ + externalRuntime: { reservationId: "r", slotId: 0, jobName: builtName, jobUid: liveUid }, + } as unknown as Partial)); + + expect(result.errorCode).toBe("k8s_pod_schedule_failed"); + expect(result.errorMessage).toContain("unschedulable"); + expect(result.errorMessage).not.toContain("claude exited 1"); + }); +}); + +// ─── selectJobOwnedPod (BLO-34577) ─────────────────────────────────────────── + +describe("selectJobOwnedPod", () => { + const owned = (uid: string, name: string, extra: Record = {}) => + ({ metadata: { name, ...extra, ownerReferences: [jobOwnerRef(uid)] }, status: { phase: "Running" } }) as k8s.V1Pod; + + it("returns null and no stale pods for an empty listing", () => { + expect(selectJobOwnedPod([], "uid-1")).toEqual({ owned: null, stale: [] }); + }); + + it("selects the pod whose ownerReferences name this Job uid", () => { + const mine = owned("uid-1", "mine"); + const { owned: got, stale } = selectJobOwnedPod([owned("uid-0", "theirs"), mine], "uid-1"); + expect(got).toBe(mine); + expect(stale.map((p) => p.metadata?.name)).toEqual(["theirs"]); + }); + + it.each(["controller-uid", "batch.kubernetes.io/controller-uid"])( + "falls back to the %s label when ownerReferences are absent", + (label) => { + const mine = { metadata: { name: "mine", labels: { [label]: "uid-1" } }, status: {} } as k8s.V1Pod; + expect(selectJobOwnedPod([mine], "uid-1").owned).toBe(mine); + }, + ); + + it("does not select a pod with neither an owner reference nor a controller-uid label (fail closed)", () => { + const anonymous = { metadata: { name: "anon", labels: { "job-name": "ac-job" } }, status: {} } as k8s.V1Pod; + const { owned: got, stale } = selectJobOwnedPod([anonymous], "uid-1"); + expect(got).toBeNull(); + expect(stale).toEqual([anonymous]); + }); + + it("ignores a non-Job owner that happens to carry the same uid", () => { + const pod = { + metadata: { name: "rs-pod", ownerReferences: [{ apiVersion: "apps/v1", kind: "ReplicaSet", name: "rs", uid: "uid-1" }] }, + status: {}, + } as k8s.V1Pod; + expect(selectJobOwnedPod([pod], "uid-1").owned).toBeNull(); + }); + + it("keeps the first owned pod when the listing carries more than one for this uid", () => { + const first = owned("uid-1", "first"); + const second = owned("uid-1", "second"); + expect(selectJobOwnedPod([first, second], "uid-1")).toEqual({ owned: first, stale: [] }); + }); }); // ─── execute: grace-period fallback (FAR-23) ───────────────────────────────── diff --git a/vendor/paperclip-adapter-claude-k8s/src/server/execute.ts b/vendor/paperclip-adapter-claude-k8s/src/server/execute.ts index 0eba9b34c221..99b7226f7bd6 100644 --- a/vendor/paperclip-adapter-claude-k8s/src/server/execute.ts +++ b/vendor/paperclip-adapter-claude-k8s/src/server/execute.ts @@ -1053,13 +1053,82 @@ export function describePodTerminatedError( return `Pod ${podName} reached phase=${phase}`; } +// Labels the Job controller stamps on every pod it creates. `controller-uid` +// is the legacy key (every supported release); the `batch.kubernetes.io/` +// prefixed key was added in 1.27. Both carry the owning Job's metadata.uid. +const JOB_CONTROLLER_UID_LABELS = ["batch.kubernetes.io/controller-uid", "controller-uid"] as const; + +/** + * Pick, from a `job-name=` pod listing, the pod owned by THIS + * execution's Job — identified by the server-assigned UID returned from + * `createNamespacedJob` (or the adopted Job's UID). + * + * Why this exists (BLO-34577). The Job name is deterministic per + * (agentId, runId) — see job-manifest.ts — and the server's in-run ccrotate + * throttle loop re-invokes `execute()` for the SAME runId after a 429, so the + * replacement Job gets the same name as the attempt that just returned. That + * previous Job was deleted with `propagationPolicy: Background`, which removes + * the Job object immediately but garbage-collects its pod asynchronously. For + * a window after the replacement Job is created, `job-name=` therefore + * matches BOTH the stale pod (phase=Failed, `claude exited 1` from the 429) + * and — once the controller creates it — the new one. Taking `items[0]` read + * the stale pod ~100 ms after create and reported the prior attempt's terminal + * state as this attempt's `k8s_pod_schedule_failed`, a code the server treats + * as non-retryable; the review was dropped. + * + * Ownership is read from `metadata.ownerReferences` (controller: Job, matching + * uid) with the controller-uid labels as the fallback. A pod carrying neither + * is not provably ours and is not selected — fail closed, since selecting the + * wrong pod is exactly the bug. Pure so the matrix is testable without a + * cluster. + */ +export function selectJobOwnedPod( + pods: readonly k8s.V1Pod[], + jobUid: string, +): { owned: k8s.V1Pod | null; stale: k8s.V1Pod[] } { + let owned: k8s.V1Pod | null = null; + const stale: k8s.V1Pod[] = []; + for (const pod of pods) { + const meta = pod.metadata; + const ownerMatch = (meta?.ownerReferences ?? []).some( + (ref) => ref.kind === "Job" && ref.uid === jobUid, + ); + const labelMatch = JOB_CONTROLLER_UID_LABELS.some((key) => meta?.labels?.[key] === jobUid); + if (ownerMatch || labelMatch) { + // The Job controller runs exactly one pod per attempt for this manifest + // (parallelism 1, no restart of a Failed pod is ours to wait on here); + // keep the first owned pod and let the caller's phase logic judge it. + if (!owned) owned = pod; + continue; + } + stale.push(pod); + } + return { owned, stale }; +} + +function describeStalePods(stale: readonly k8s.V1Pod[]): string { + return stale + .map((pod) => { + const name = pod.metadata?.name ?? "unknown"; + const owner = pod.metadata?.ownerReferences?.find((ref) => ref.kind === "Job")?.uid + ?? JOB_CONTROLLER_UID_LABELS.map((key) => pod.metadata?.labels?.[key]).find(Boolean) + ?? ""; + return `${name} (owner uid ${owner}, phase=${pod.status?.phase ?? "Unknown"})`; + }) + .join(", "); +} + /** * Wait for the Job's pod to reach a terminal or running state. * Returns the pod name once logs can be streamed, or throws on failure. + * + * `jobUid` scopes the lookup to the Job this execution created or adopted; + * same-name pods from an earlier attempt are ignored (see selectJobOwnedPod). */ async function waitForPod( namespace: string, jobName: string, + jobUid: string, scheduleTimeoutMs: number, startTimeoutMs: number, onLog: AdapterExecutionContext["onLog"], @@ -1074,12 +1143,21 @@ async function waitForPod( let lastStatus = ""; let lastStatusDetails = "no pod observed yet"; let startDeadline = 0; + let staleLogged = false; while (true) { const podList = await coreApi.listNamespacedPod({ namespace, labelSelector, }); - const pod = podList.items[0]; + const { owned: pod, stale } = selectJobOwnedPod(podList.items, jobUid); + if (stale.length > 0 && !staleLogged) { + staleLogged = true; + await onLog( + "stdout", + `[paperclip] Ignoring ${stale.length} pod(s) named for Job ${jobName} but not owned by this Job (uid ${jobUid}); ` + + `they belong to an earlier attempt still being garbage-collected: ${describeStalePods(stale)}\n`, + ); + } if (!pod) { if (Date.now() >= scheduleDeadline) { @@ -1247,8 +1325,13 @@ async function waitForJobCompletion( /** * Get the exit code from the Job's pod. */ -async function getPodExitCode(namespace: string, jobName: string, kubeconfigPath?: string): Promise { - const state = await getPodTerminatedState(namespace, jobName, kubeconfigPath); +async function getPodExitCode( + namespace: string, + jobName: string, + jobUid: string, + kubeconfigPath?: string, +): Promise { + const state = await getPodTerminatedState(namespace, jobName, jobUid, kubeconfigPath); return state?.exitCode ?? null; } @@ -1278,6 +1361,7 @@ export interface PodLookupResult { async function lookupPodState( namespace: string, jobName: string, + jobUid: string, kubeconfigPath?: string, ): Promise { const coreApi = getCoreApi(kubeconfigPath); @@ -1285,7 +1369,9 @@ async function lookupPodState( namespace, labelSelector: `job-name=${jobName}`, }); - const pod = podList.items[0]; + // Same ownership scoping as waitForPod: a same-name pod from an earlier + // attempt must not be read as this attempt's terminal state (BLO-34577). + const { owned: pod } = selectJobOwnedPod(podList.items, jobUid); if (!pod) return { state: null, phase: null, podMissing: true }; const phase = pod.status?.phase ?? null; @@ -1314,13 +1400,14 @@ async function lookupPodState( async function getPodLookupWithRetry( namespace: string, jobName: string, + jobUid: string, kubeconfigPath?: string, attempts = 4, delayMs = 500, ): Promise { let last: PodLookupResult = { state: null, phase: null, podMissing: true }; for (let i = 0; i < attempts; i++) { - last = await lookupPodState(namespace, jobName, kubeconfigPath); + last = await lookupPodState(namespace, jobName, jobUid, kubeconfigPath); if (last.state) return last; if (last.podMissing) return last; // Pod exists but no terminated state. If it is in a terminal phase the @@ -1335,9 +1422,10 @@ async function getPodLookupWithRetry( async function getPodTerminatedState( namespace: string, jobName: string, + jobUid: string, kubeconfigPath?: string, ): Promise { - return (await lookupPodState(namespace, jobName, kubeconfigPath)).state; + return (await lookupPodState(namespace, jobName, jobUid, kubeconfigPath)).state; } /** @@ -1513,6 +1601,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise