import { WorkflowEntrypoint } from "cloudflare:workers"; import type { WorkflowEvent, WorkflowStep } from "cloudflare:workers"; import { applyRun, failRun, planRun } from "./service"; import { abandonRequeue, checkRequeue } from "./requeue"; import { MAX_CI_POLLS, MAX_ENQUEUE_ATTEMPTS } from "./requeue-policy"; /** `requeue` skips planning: an applied proposal from a queue-triggered run only needs the CI wait and re-enqueue. */ export type MergeFixParams = { runId: string; phase?: "fix" | "requeue" }; /** Plan (model calls) and apply are separate steps without retries: a retry could re-bill the models or push twice. */ export class MergeFixWorkflow extends WorkflowEntrypoint { override async run(event: Readonly>, step: WorkflowStep): Promise { const { runId, phase = "fix" } = event.payload; if (phase === "requeue") { await requeue(runId, step); return; } try { const plan = await step.do("plan", { retries: { limit: 0, delay: "1 second" }, timeout: "10 minutes" }, () => planRun(runId), ); if (!plan.autoApply) return; const applied = await step.do("apply", { retries: { limit: 0, delay: "1 second" }, timeout: "5 minutes" }, () => applyRun(runId), ); if (!applied.autoApply) return; await requeue(runId, step); } catch (error) { await step.do("fail", () => failRun(runId, error instanceof Error ? error.message : "The fix failed")); } } } async function requeue(runId: string, step: WorkflowStep): Promise { try { for (let poll = 1; poll <= MAX_CI_POLLS + MAX_ENQUEUE_ATTEMPTS; poll++) { await step.sleep(`requeue-wait-${poll}`, "2 minutes"); const final = poll === MAX_CI_POLLS + MAX_ENQUEUE_ATTEMPTS; const next = await step.do(`requeue-${poll}`, { retries: { limit: 2, delay: "30 seconds" } }, () => checkRequeue(runId, poll, final), ); if (next === "done") return; } } catch (error) { await step.do("requeue-abandon", () => abandonRequeue(runId, error instanceof Error ? error.message : "Re-enqueue failed"), ); } }