diff --git a/.changeset/usage-limit-active-lanes.md b/.changeset/usage-limit-active-lanes.md new file mode 100644 index 0000000000..25d3a486d5 --- /dev/null +++ b/.changeset/usage-limit-active-lanes.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: A provider rate limit now pauses every task actually running on that provider, including on renamed boards. +category: fix +dev: `taskUsesProvider`'s executor and merger lane checks resolve the workflow's wip/review columns (from the same per-workflow IR cache the planner lane already uses) instead of comparing against `"in-progress"`/`"in-review"`; both fail soft to the legacy literal. diff --git a/packages/engine/src/__tests__/usage-limit-detector.test.ts b/packages/engine/src/__tests__/usage-limit-detector.test.ts index 544778a31a..a2444ca817 100644 --- a/packages/engine/src/__tests__/usage-limit-detector.test.ts +++ b/packages/engine/src/__tests__/usage-limit-detector.test.ts @@ -333,3 +333,217 @@ describe("usage-limit parking and recovery round trip (U11)", () => { expect(store.pauseTask).not.toHaveBeenCalledWith("FN-106", false); }); }); + +/* +FNXC:WorkflowLifecycleColumns 2026-07-31-22:35: + +THE EXECUTOR AND MERGER LANES WERE STILL LITERALS. `taskUsesProvider` resolves a task's active lane to decide +which providers it is running on; the PLANNER lane was converted to traits (the note in that function +describes the exact failure and says it was fixed), and the executor/merger halves were left as +`task.column === "in-progress"` / `=== "in-review"`. + +So on a renamed board an actively-executing card resolved NO providers, and a provider rate limit never +paused it — the engine kept hammering the limited provider with that card. Measured before the fix on a +renamed board (`building` = wip), executor limit triggered by a peer: only the TRIGGERING task was paused, +and only because of the always-include-the-trigger fallback. The peer running on the same rate-limited +provider was left going. + +HOW THIS WAS FOUND, because the route matters more than the bug: a scan for "legacy literal within a few +lines of a role-resolved call" flagged this file. The FIRST thing I suspected there — the `done`/`archived` +terminal filter — turned out to be a FALSE POSITIVE: its revert stayed green, because the lane check already +excludes finished cards. Chasing why the revert would not go red is what surfaced the real defect one line +over. A revert proof that stays green is information, not a dead end. +*/ +describe("a provider rate limit pauses every card actually running on that provider", () => { + const RENAMED_IR = { + version: "v2", id: "wf-renamed", name: "renamed", nodes: [], edges: [], + columns: [ + { id: "queued", name: "Queued", traits: [{ trait: "intake" }, { trait: "hold", config: { release: "capacity" } }] }, + { id: "building", name: "Building", traits: [{ trait: "wip", config: { limitSetting: "maxConcurrent" } }] }, + { id: "checking", name: "Checking", traits: [{ trait: "merge" }, { trait: "merge-blocker" }] }, + { id: "shipped", name: "Shipped", traits: [{ trait: "complete" }] }, + ], + }; + + const card = (id: string, column: string) => ({ + id, column, dependencies: [], steps: [], currentStep: 0, log: [], + modelProvider: "openai-codex", modelId: "gpt-5", + }); + + function resolvingStore(tasks: any[], ir: unknown) { + const store = createMockStore(tasks); + const selection = { workflowId: "wf-renamed", stepIds: [] }; + store.getTaskWorkflowSelection = () => (ir ? selection : undefined); + store.getTaskWorkflowSelectionAsync = async () => (ir ? selection : undefined); + store.getWorkflowDefinition = async () => (ir ? { id: "wf-renamed", ir } : undefined); + return store; + } + + async function pausedIds(store: any, agentType = "executor", trigger = "FN-TRIGGER"): Promise { + await new UsageLimitPauser(store).onUsageLimitHit(agentType, trigger, "rate_limit_error: Rate limit exceeded", "openai-codex"); + return (store.pauseTask.mock.calls as unknown[][]).map((call) => call[0] as string); + } + + it("pauses a PEER executing in the renamed WIP column", async () => { + // Pre-fix: only FN-TRIGGER, via the always-include fallback. FN-PEER kept running on the limited provider. + const store = resolvingStore( + [card("FN-TRIGGER", "building"), card("FN-PEER", "building"), card("FN-SHIPPED", "shipped")], + RENAMED_IR, + ); + + const paused = await pausedIds(store); + + expect(paused).toContain("FN-PEER"); + // The terminal lane is still excluded — by the lane check itself, which is why the `done`/`archived` + // filter above it did not need converting. + expect(paused).not.toContain("FN-SHIPPED"); + }); + + it("keeps the LEGACY wip id when a resolvable workflow declares no wip column", async () => { + /* + FNXC:WorkflowLifecycleColumns 2026-07-31-23:55 (PR #2672 review — greptile P1): + THE FALLBACK IS PER ROLE, not per object, and nothing pinned that until now. + + A workflow that RESOLVES but declares no `wip` column previously suppressed the + legacy id — because the fallback keyed on whether `activeLanes` existed at all — + so a card in `in-progress` resolved no providers and kept running on the limited + one. A missing ROLE is not the same fact as a missing WORKFLOW; only the second + means "no basis to judge". + + Found by mutation: reverting to the per-object form left all 57 existing tests + green, so the fix greptile asked for was unproven. + */ + const NO_WIP_IR = { + version: "v2", id: "wf-nowip", name: "no wip", nodes: [], edges: [], + columns: [ + { id: "queued", name: "Queued", traits: [{ trait: "intake" }, { trait: "hold", config: { release: "capacity" } }] }, + { id: "checking", name: "Checking", traits: [{ trait: "merge" }, { trait: "merge-blocker" }] }, + { id: "shipped", name: "Shipped", traits: [{ trait: "complete" }] }, + ], + }; + const store = resolvingStore( + [card("FN-TRIGGER", "in-progress"), card("FN-LEGACY-WIP", "in-progress")], + NO_WIP_IR, + ); + + const paused = await pausedIds(store); + + expect(paused).toContain("FN-LEGACY-WIP"); + }); + + it("pauses a merger-lane peer in the renamed REVIEW column", async () => { + const store = resolvingStore( + [card("FN-TRIGGER", "checking"), card("FN-PEER", "checking"), card("FN-WIP", "building")], + RENAMED_IR, + ); + + const paused = await pausedIds(store, "merger"); + + expect(paused).toContain("FN-PEER"); + // A wip card is not on the merger lane, so a merger-provider limit must leave it alone. + expect(paused).not.toContain("FN-WIP"); + }); + + it("does NOT pause a card parked in the renamed planning lane on an executor limit", async () => { + // The paired negative: "pause everything with that provider" must not pass for "resolve the lane". + const store = resolvingStore([card("FN-TRIGGER", "building"), card("FN-PARKED", "queued")], RENAMED_IR); + + expect(await pausedIds(store)).not.toContain("FN-PARKED"); + }); + + it("keeps the legacy literals when no workflow resolves", async () => { + const store = resolvingStore([card("FN-TRIGGER", "in-progress"), card("FN-PEER", "in-progress")], undefined); + + expect(await pausedIds(store)).toContain("FN-PEER"); + }); +}); + +/* +FNXC:WorkflowLifecycleColumns 2026-07-31-23:35 (PR #2672 review, greptile P1 + P2): + +TWO PROPERTIES MY FIRST VERSION GOT WRONG. + +1. THE FALLBACK IS PER ROLE, NOT PER OBJECT. I keyed it on whether `activeLanes` existed at all, so a + workflow that RESOLVES but declares no wip (or no review) column suppressed the legacy id and resolved no + providers — reintroducing the exact bug this change fixes, for the partial-vocabulary case. A missing ROLE + is not the same fact as a missing WORKFLOW; only the second means "no basis". + +2. RESOLVE ONLY THE CANDIDATES. Resolving every task before filtering made a rate limit on a 400-card board + pay 400 resolutions to pause a handful, including terminal and paused cards that can never be affected. +*/ +describe("lane resolution degrades per role and costs only what it must", () => { + /** Declares a wip lane but NO review lane — the partial vocabulary that broke the first version. */ + const NO_REVIEW_IR = { + version: "v2", id: "wf-partial", name: "partial", nodes: [], edges: [], + columns: [ + { id: "queued", name: "Queued", traits: [{ trait: "intake" }, { trait: "hold", config: { release: "capacity" } }] }, + { id: "in-progress", name: "In progress", traits: [{ trait: "wip", config: { limitSetting: "maxConcurrent" } }] }, + ], + }; + + const card = (id: string, column: string) => ({ + id, column, dependencies: [], steps: [], currentStep: 0, log: [], + modelProvider: "openai-codex", modelId: "gpt-5", + }); + + function storeFor(tasks: any[], ir: unknown, workflowId = "wf-partial") { + const store = createMockStore(tasks); + const selection = { workflowId, stepIds: [] }; + const reads = { definition: 0 }; + /* + Records WHICH task ids were asked about, not how many reads happened. A read count is + concurrency-dependent here — the shared IR cache fills after an await, so same-workflow tasks race and the + number lands anywhere between 1 and N. Asking "was this task resolved at all?" is exact and race-free, + which is what a filtering assertion actually needs. + */ + const asked: string[] = []; + store.getTaskWorkflowSelectionAsync = async (id: string) => { + asked.push(id); + return selection; + }; + store.getTaskWorkflowSelection = (id: string) => { + asked.push(id); + return selection; + }; + store.getWorkflowDefinition = async () => { + reads.definition += 1; + return ir ? { id: workflowId, ir } : undefined; + }; + (store as any).reads = reads; + (store as any).asked = asked; + return store; + } + + it("keeps the legacy REVIEW id when the workflow declares a wip lane but no review lane", async () => { + // Pre-fix: `activeLanes` was truthy, `review` undefined, so an `in-review` card resolved no providers and + // kept running on the rate-limited provider. + const store = storeFor([card("FN-TRIGGER", "in-review"), card("FN-PEER", "in-review")], NO_REVIEW_IR); + + await new UsageLimitPauser(store).onUsageLimitHit("merger", "FN-TRIGGER", "rate_limit_error: Rate limit exceeded", "openai-codex"); + + expect((store.pauseTask.mock.calls as unknown[][]).map((c) => c[0])).toContain("FN-PEER"); + }); + + it("does not resolve a workflow for terminal or paused tasks", async () => { + // Only FN-LIVE is a plausible candidate; the other three can never be affected, so paying a resolution + // for them is cost with no possible outcome. + const store = storeFor( + [ + card("FN-LIVE", "in-progress"), + card("FN-DONE", "done"), + card("FN-ARCHIVED", "archived"), + { ...card("FN-PAUSED", "in-progress"), paused: true }, + ], + NO_REVIEW_IR, + ); + + await new UsageLimitPauser(store).onUsageLimitHit("executor", "FN-LIVE", "rate_limit_error: Rate limit exceeded", "openai-codex"); + + // Exact, not a count: the three tasks that can never be affected are never resolved at all. + const asked = (store as any).asked as string[]; + expect(asked).toContain("FN-LIVE"); + expect(asked).not.toContain("FN-DONE"); + expect(asked).not.toContain("FN-ARCHIVED"); + expect(asked).not.toContain("FN-PAUSED"); + }); +}); diff --git a/packages/engine/src/usage-limit-detector.ts b/packages/engine/src/usage-limit-detector.ts index 0016b22c15..6c4dfa2136 100644 --- a/packages/engine/src/usage-limit-detector.ts +++ b/packages/engine/src/usage-limit-detector.ts @@ -125,6 +125,12 @@ export class UsageLimitPauser { settings: Awaited>, agentType: string, preImplementationColumns?: ReadonlySet, + /* + FNXC:WorkflowLifecycleColumns 2026-07-31-22:10: + The WIP and REVIEW lanes of this task's own workflow, resolved by the caller from the same per-workflow + IR cache as `preImplementationColumns`. Optional so an unresolvable workflow keeps the legacy literals. + */ + activeLanes?: { wip?: string; review?: string }, ): boolean { /* FNXC:WorkflowLifecycleColumns 2026-07-29-15:20 (P0 audit after the Planning-column merge): @@ -149,12 +155,36 @@ export class UsageLimitPauser { resolveValidatorSessionModel(task.validatorModelProvider, task.validatorModelId, settings).provider, ] : []) : agentType === "executor" - ? (task.column === "in-progress" ? [ + /* + FNXC:WorkflowLifecycleColumns 2026-07-31-22:15: + THE EXECUTOR AND MERGER LANES ARE RESOLVED TOO. The note above describes exactly this failure for the + PLANNER lane and says it was fixed there — these two were left as literals, so the same silent hole + stayed open one lane over: on a renamed board an actively-executing card in the WIP column resolved NO + providers, so a provider rate limit never paused it and the engine kept hammering the limited provider + with that card. + + MEASURED before the fix, renamed board (`building` = wip), executor limit on a peer card: only the + TRIGGERING task was paused, and that only because of the always-include-the-trigger fallback below. + The peer executing on the same rate-limited provider was left running. + + FNXC:WorkflowLifecycleColumns 2026-07-31-23:10 (PR #2672 review, greptile P1): + THE FALLBACK IS PER ROLE, not per object. My first version keyed it on whether `activeLanes` existed + at all, so a workflow that RESOLVES but declares no wip (or no review) column suppressed the legacy + id and resolved no providers — reintroducing this very bug for the partial-vocabulary case. A missing + ROLE is not the same fact as a missing WORKFLOW, and only the second one means "no basis". + + Falling back per role can only ADD pauses, for a card sitting in a legacy-named column on a board that + declares no such role — which is the safe direction here and the one the note above states: the cost + of a wrong include is pausing work that was fine. + */ + ? (((activeLanes?.wip ?? "in-progress") === task.column) ? [ resolveExecutorSessionModel(task.modelProvider, task.modelId, settings).provider, resolveValidatorSessionModel(task.validatorModelProvider, task.validatorModelId, settings).provider, ] : []) : agentType === "merger" - ? (task.column === "in-review" ? [resolveMergerSessionModel(settings, undefined, task).provider] : []) + ? (((activeLanes?.review ?? "in-review") === task.column) + ? [resolveMergerSessionModel(settings, undefined, task).provider] + : []) : []; const resolvedProviders = providersByActiveLane; return resolvedProviders.some((candidate) => candidate?.trim().toLowerCase() === provider); @@ -206,6 +236,32 @@ export class UsageLimitPauser { */ const irCache = new Map(); const preImplementationByTask = new Map>(); + /* + FNXC:WorkflowLifecycleColumns 2026-07-31-22:20: + Resolved for EVERY agent type, because the executor and merger lanes need it and only the + pre-implementation lane is planner-only. Shares `irCache`, so this is still one IR read per WORKFLOW + across the whole fan-out — a task whose workflow cannot be resolved yields `undefined` and the callee + keeps the legacy literal for it. + */ + const activeLanesByTask = new Map(); + /* + FNXC:WorkflowLifecycleColumns 2026-07-31-23:20 (PR #2672 review, greptile P2): + RESOLVE ONLY THE CANDIDATES. The first version resolved every task on the board before filtering, so a + rate limit on a 400-card board paid 400 resolutions to pause a handful — and terminal or paused cards, + which can never be affected, were resolved too. The cheap predicates run first now, so the resolution + cost tracks the number of plausible tasks rather than the size of the board. + + KNOWN AND UNCHANGED: concurrent misses on the same workflow can still both read, because `irCache` fills + after the await. That is a property this path shares with the planner-lane resolution above it, and + serialising to fix it would trade a bounded duplicate read for latency on the pause — which is the + operation an operator is waiting on. Named rather than silently accepted. + */ + const laneCandidates = tasks.filter((task) => + task.paused !== true && task.column !== "done" && task.column !== "archived"); + await Promise.all(laneCandidates.map(async (task) => { + const columns = await resolveTaskLifecycleColumns(this.store, task.id, irCache).catch(() => undefined); + activeLanesByTask.set(task.id, columns ? { wip: columns.wip, review: columns.review } : undefined); + })); if (agentType === PLANNER_AGENT_ROLE) { await Promise.all(tasks.map(async (task) => { const columns = await resolveTaskLifecycleColumns(this.store, task.id, irCache).catch(() => undefined); @@ -232,7 +288,14 @@ export class UsageLimitPauser { && task.column !== "archived" && task.paused !== true && providerId !== "unknown" - && this.taskUsesProvider(task, providerId, settings, agentType, preImplementationByTask.get(task.id))); + && this.taskUsesProvider( + task, + providerId, + settings, + agentType, + preImplementationByTask.get(task.id), + activeLanesByTask.get(task.id), + )); // Always include the task that produced the 429 even if its actual provider // came from a runtime fallback not represented in persisted task settings.