FN-8654: rotate credential instances after provider limits
Retry provider-limit failures with eligible credential instances before falling back to existing pauses and backoff. - Add a runtime-shared credential rotator with cooldown, exhaustion, and audit handling. - Wire credential rotation into executor and heartbeat retry lanes while preserving user pause controls. - Document the behavior and cover rotation, recovery, and retry paths. Files changed: .changeset/fn-8654-credential-instance-rotation.md | 7 + AGENTS.md | 1 + docs/architecture.md | 2 +- docs/settings-reference.md | 4 + .../__tests__/credential-instance-rotation.test.ts | 88 +++++++++++ .../__tests__/credential-rotation-lanes.test.ts | 20 +++ .../__tests__/credential-rotation-recovery.test.ts | 19 +++ .../__tests__/credential-rotation-wiring.test.ts | 15 ++ .../__tests__/rate-limit-retry-rotation.test.ts | 50 ++++++ .../src/__tests__/usage-limit-detector.test.ts | 14 ++ packages/engine/src/agent-heartbeat.ts | 102 +++++++++++- .../engine/src/credential-instance-rotation.ts | 175 +++++++++++++++++++++ packages/engine/src/executor.ts | 141 +++++++++++++++-- packages/engine/src/index.ts | 7 + packages/engine/src/project-engine.ts | 5 + packages/engine/src/rate-limit-retry.ts | 32 +++- packages/engine/src/runtimes/in-process-runtime.ts | 29 +++- packages/engine/src/usage-limit-detector.ts | 18 ++- 18 files changed, 699 insertions(+), 30 deletions(-) Fusion-Task-Id: FN-8654 Fusion-Task-Lineage: 44d63441-270c-4949-8c34-47ec4c9992e4 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-8654-credential-instance-rotation.md
Normal file
7
.changeset/fn-8654-credential-instance-rotation.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": minor
|
||||
---
|
||||
|
||||
summary: Keep the board moving on a second account instead of pausing when a provider hits its limit.
|
||||
category: feature
|
||||
dev: Adds the credential-instance rotator at the in-process runtime with the inventory-size no-op gate and CREDENTIAL_INSTANCE_COOLDOWN_MS.
|
||||
@@ -323,6 +323,7 @@ Scoped exception (FN-5819): shared-branch-group members (`branchContext.assignme
|
||||
- FN-8144: archive emits `archive-workspace-worktree-disposer-missing` when a workspace archive has no store-scoped backend disposer; per-repository archive removal is awaited under canonical-path reservations, with failed paths quarantined for successor reconciliation.
|
||||
- FN-7514: the planner overseer's per-task oversight loop (`PlannerRecoveryController.tick`) emits `overseer:oversight-withheld-human-control` when the pure `evaluateOverseerHumanControl` guard withholds ALL oversight action (no steering, retry, targeted-fix, or pending confirmation) for a task that is user-paused (`task.userPaused===true`, or `task.paused===true` with no `pausedReason`) or ineligible for auto-merge processing per `allowsAutoMergeProcessing` (`autoMerge:false`/PR-based human-review terminal contract). The guard runs BEFORE FN-7513's confirmation classification, so a withheld task never records a pending confirmation. Metadata: `{ taskId, reason: "user-paused" | "auto-merge-off-human-review", stage, oversightLevel }`; deduped per (taskId, withheld reason) so it is not re-emitted every poll while the reason is unchanged.
|
||||
- FN-7720: `TaskStore.bypassFailedPreMergeReviewStep` emits `task:bypass-review` when a privileged operator bypasses the latest failed pre-merge review step of an `in-review` task; metadata includes `workflowStepId`, `workflowStepName`, `bypassedFromStatus`, `bypassedFromVerdict`, and the mandatory `reason`. The bypass rewrites the step's `status` to `"skipped"` with `bypassedBy`/`bypassedAt`/`bypassReason`/`bypassedFromStatus` fields; it never fabricates a reviewer `verdict` and clears only the failed-pre-merge-step `getTaskMergeBlocker` reason. Reachable via `fn_task_bypass_review` (CLI/pi-extension operator tool surface only — not executor/reviewer/triage) and `POST /tasks/:id/bypass-review`.
|
||||
- FN-8654: credential rotation emits append-only `credential:instance-rotation-attempt`, `credential:instance-rotation-outcome`, and `credential:instance-rotation-exhausted` rows. Attempt and terminal outcome are separate immutable rows; metadata contains provider/instance IDs, lane, optional task/agent IDs, counts, and fixed outcomes only. Exhaustion records `startingInstanceId` separately and excludes it from `attemptedCount`; providers with zero or one configured instance emit none of these rows.
|
||||
- FN-7996: executor emits `task:execution-tool-failure-retry` for a claimed same-model consecutive-tool-failure retry and `task:execution-tool-failure-retry-exhausted` when the matching run budget is spent. Metadata is ids/counts/outcomes-only; the exhausted event is emitted once through a project-scoped compare-and-set while terminal parking remains idempotent.
|
||||
- FN-7998: executor emits `task:execution-escalation-retry` when its opt-in, single alternate model/node attempt is persisted after FN-7996 exhaustion, and `task:execution-escalation-exhausted` when that attempt also reaches the terminal park. Metadata remains ids/counts/outcomes-only (`taskId`, graph node id, target booleans, and prior retry count); no model identifiers or prose are persisted in run-audit.
|
||||
- FN-8004: `agent:heartbeat-move-skipped-soft-delete` records a heartbeat move that races a soft-deleted task without parking the durable agent. Metadata remains ids/timestamps/source only (`agentId`, optional `taskId`/`deletedAt`, `moveAttemptedAt`, optional `source`); it never stores error prose.
|
||||
|
||||
@@ -738,7 +738,7 @@ Guardrails: this routine does **not** retry merges, does **not** apply to mixed/
|
||||
- FN-5089 adds an optional task-worktree `commit-msg` hook (default enabled via `commitMsgHookEnabled`) installed by the same provisioning path; when enabled it appends the configured task attribution trailer (defaults to `Fusion-Task-Id: <task-id>`) without duplicating existing trailers. Attribution remains branch/subject resilient when the hook is disabled.
|
||||
- FN-4948 extends contamination auto-recovery with an `obviously-misrouted` bucket: foreign-attributed commits are auto-dropped only when attribution resolves to another task and every changed path is inside `.changeset/fn-<foreign-id>-*.md`. Any shared/non-namespaced path stays in the unique bucket and escalates to human adjudication. The single-attempt contamination invariant is unchanged.
|
||||
- `ProjectEngine` settings lifecycle handlers (`project-engine.ts`) treat `enginePaused` as a soft pause: clearing it dispatches runtime resume and, when `autoMerge` is enabled, performs an `in-review` eligibility sweep to requeue mergeable review tasks.
|
||||
- `UsageLimitPauser` (`usage-limit-detector.ts`) and `withRateLimitRetry` (`rate-limit-retry.ts`)
|
||||
- `UsageLimitPauser` (`usage-limit-detector.ts`) and `withRateLimitRetry` (`rate-limit-retry.ts`): a provider usage limit first offers the next credential instance from the runtime-owned `CredentialInstanceRotator`, then retains the existing provider-scoped park and backoff as the fallback. The rotator opens no event for inventories of zero or one instance, so single-instance providers retain byte-identical retry/park behavior with no cooldown or rotation audit artifact. Multi-instance eligibility is configured inventory excluding the limited starting instance, active cooldowns, and already offered candidates; an all-cooling-down inventory is a real exhausted event. Its finite per-event plan, not the retry wrapper, bounds offered-once rotation. User/global/engine pauses and `autoMerge:false` decline rotation, while abort keeps its existing pre-retry short-circuit. Cooldowns are process-local, self-expiring hints cleared by the runtime pauser on provider recovery; the dashboard monitor's rotator-less recovery can only leave a hint until expiry and never affects correctness.
|
||||
|
||||
### Worktree and naming helpers
|
||||
- `WorktreePool` (`worktree-pool.ts`) — idle worktree reuse
|
||||
|
||||
@@ -1870,6 +1870,10 @@ Project-scoped ordered list of up to six mobile footer quick actions. The defaul
|
||||
|
||||
Enabled plugins may declare `mcpServers` declaratively. The contribution is project-scoped: only plugins enabled in that project's plugin state participate. Resolution is global → enabled plugin declarations → project settings, by server `name`; later declarations win and a project `enabled:false` entry removes an inherited plugin server. The Global MCP card never includes plugin declarations. Project MCP UI identifies them as `plugin:<id>` and writes only project overrides or tombstones.
|
||||
|
||||
### Credential-instance rate-limit rotation
|
||||
|
||||
When a provider has two or more configured credential instances, executor and heartbeat retries may move to the next eligible instance after a usage-limit response before using the normal pause and backoff path. A limited instance is skipped for 15 minutes in the running engine; cooldowns are not persisted. Providers with zero or one instance have no rotation event, cooldown, or rotation audit row and preserve existing behavior. User-controlled tasks, including `autoMerge:false`, do not rotate.
|
||||
|
||||
### Credential instance companions
|
||||
|
||||
Every provider/model selection can now persist an optional `*CredentialInstanceId` companion: the
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
CREDENTIAL_INSTANCE_COOLDOWN_MS,
|
||||
CredentialInstanceRotator,
|
||||
createRotationPlan,
|
||||
} from "../credential-instance-rotation.js";
|
||||
|
||||
const ref = (instanceId: string) => ({ providerId: "anthropic", instanceId });
|
||||
|
||||
describe("CredentialInstanceRotator", () => {
|
||||
it("has a zero-side-effect no-op gate for zero and single-instance inventories", async () => {
|
||||
for (const instances of [[], [ref("default")]]) {
|
||||
const audit = vi.fn();
|
||||
const rotator = new CredentialInstanceRotator({ instanceSource: { listInstances: () => instances, getDefaultInstance: () => undefined }, recordRunAuditEvent: audit });
|
||||
expect(await rotator.beginEvent({ providerId: "anthropic", startingInstanceId: "default", lane: "executor-step" })).toBeUndefined();
|
||||
expect(rotator.isCoolingDown(ref("default"))).toBe(false);
|
||||
expect(audit).not.toHaveBeenCalled();
|
||||
}
|
||||
});
|
||||
|
||||
it("orders candidates deterministically and excludes cooling down refs", () => {
|
||||
const plan = createRotationPlan("anthropic", "b", [ref("c"), ref("a"), ref("b")], new Map([["anthropic[c]", 101]]), 100);
|
||||
expect(plan.candidates.map((candidate) => candidate.instanceId)).toEqual(["a"]);
|
||||
expect(plan.skippedCooldownInstanceIds).toEqual(["c"]);
|
||||
});
|
||||
|
||||
it("emits immutable attempt, outcome, and exhaustion records for a real empty plan", async () => {
|
||||
let now = 100;
|
||||
const audit = vi.fn();
|
||||
const rotator = new CredentialInstanceRotator({ instanceSource: { listInstances: () => [ref("a"), ref("b")], getDefaultInstance: () => ref("a") }, now: () => now, recordRunAuditEvent: audit });
|
||||
rotator.markLimited(ref("b"));
|
||||
const event = await rotator.beginEvent({ providerId: "anthropic", startingInstanceId: "a", lane: "executor-step", taskId: "FN-1" });
|
||||
expect(event?.candidateCount).toBe(0);
|
||||
expect(await event?.next()).toBeUndefined();
|
||||
event?.finishExhausted(); event?.finishExhausted();
|
||||
expect(audit).toHaveBeenCalledTimes(1);
|
||||
expect(audit.mock.calls[0]).toEqual(["credential:instance-rotation-exhausted", expect.objectContaining({ instanceCount: 2, attemptedCount: 0, startingInstanceId: "a" })]);
|
||||
now += CREDENTIAL_INSTANCE_COOLDOWN_MS;
|
||||
expect(rotator.isCoolingDown(ref("b"))).toBe(false);
|
||||
});
|
||||
|
||||
it("prunes elapsed cooldowns while planning a later multi-instance event", async () => {
|
||||
let now = 0;
|
||||
const rotator = new CredentialInstanceRotator({
|
||||
instanceSource: { listInstances: () => [ref("a"), ref("b")], getDefaultInstance: () => ref("a") },
|
||||
now: () => now,
|
||||
});
|
||||
rotator.markLimited(ref("b"));
|
||||
now += CREDENTIAL_INSTANCE_COOLDOWN_MS;
|
||||
await rotator.beginEvent({ providerId: "anthropic", startingInstanceId: "a", lane: "executor-step" });
|
||||
expect((rotator as unknown as { cooldowns: Map<string, number> }).cooldowns.size).toBe(0);
|
||||
});
|
||||
|
||||
it("offers each candidate once and clears only the recovered provider", async () => {
|
||||
const rotator = new CredentialInstanceRotator({ instanceSource: { listInstances: () => [ref("a"), ref("b"), ref("c")], getDefaultInstance: () => ref("a") } });
|
||||
const event = await rotator.beginEvent({ providerId: "anthropic", startingInstanceId: "a", lane: "executor-agent" });
|
||||
expect((await event?.next())?.instanceId).toBe("b");
|
||||
expect((await event?.next())?.instanceId).toBe("c");
|
||||
expect(await event?.next()).toBeUndefined();
|
||||
expect(await event?.next()).toBeUndefined();
|
||||
rotator.markLimited(ref("b"));
|
||||
rotator.clearCooldowns("anthropic");
|
||||
expect(rotator.isCoolingDown(ref("b"))).toBe(false);
|
||||
});
|
||||
|
||||
it("keeps attempt and terminal outcome immutable and ignores audit failures", async () => {
|
||||
const audit = vi.fn((type: string) => {
|
||||
if (type.endsWith("outcome")) throw new Error("audit unavailable");
|
||||
});
|
||||
const rotator = new CredentialInstanceRotator({
|
||||
instanceSource: { listInstances: () => [ref("a"), ref("b")], getDefaultInstance: () => ref("a") },
|
||||
recordRunAuditEvent: audit,
|
||||
});
|
||||
const event = await rotator.beginEvent({ providerId: "anthropic", startingInstanceId: "a", lane: "executor-agent", taskId: "FN-1" });
|
||||
await event?.next();
|
||||
event?.recordOutcome("rotation-succeeded");
|
||||
expect(audit.mock.calls).toEqual([
|
||||
["credential:instance-rotation-attempt", {
|
||||
providerId: "anthropic", fromInstanceId: "a", toInstanceId: "b", attempt: 1,
|
||||
candidateCount: 1, outcome: "rotated", lane: "executor-agent", taskId: "FN-1",
|
||||
}],
|
||||
["credential:instance-rotation-outcome", {
|
||||
providerId: "anthropic", toInstanceId: "b", attempt: 1, outcome: "rotation-succeeded",
|
||||
lane: "executor-agent", taskId: "FN-1",
|
||||
}],
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,20 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { CredentialInstanceRotator, type RotationLane } from "../credential-instance-rotation.js";
|
||||
|
||||
const lanes: RotationLane[] = ["executor-step", "executor-agent", "agent-heartbeat"];
|
||||
|
||||
describe("credential rotation lane audit attribution", () => {
|
||||
it.each(lanes)("attributes a rotated retry to %s", async (lane) => {
|
||||
const audit = vi.fn();
|
||||
const rotator = new CredentialInstanceRotator({
|
||||
instanceSource: { listInstances: () => [
|
||||
{ providerId: "anthropic", instanceId: "a" },
|
||||
{ providerId: "anthropic", instanceId: "b" },
|
||||
], getDefaultInstance: () => undefined },
|
||||
recordRunAuditEvent: audit,
|
||||
});
|
||||
const event = await rotator.beginEvent({ providerId: "anthropic", startingInstanceId: "a", lane });
|
||||
await event?.next();
|
||||
expect(audit).toHaveBeenCalledWith("credential:instance-rotation-attempt", expect.objectContaining({ lane, toInstanceId: "b" }));
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,19 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { CREDENTIAL_INSTANCE_COOLDOWN_MS, CredentialInstanceRotator } from "../credential-instance-rotation.js";
|
||||
|
||||
describe("credential rotation recovery bounds", () => {
|
||||
it("self-expires a dashboard-monitor-unobserved cooldown", async () => {
|
||||
let now = 0;
|
||||
const ref = { providerId: "anthropic", instanceId: "backup" };
|
||||
const rotator = new CredentialInstanceRotator({
|
||||
instanceSource: { listInstances: () => [ref, { providerId: "anthropic", instanceId: "default" }], getDefaultInstance: () => undefined },
|
||||
now: () => now,
|
||||
});
|
||||
rotator.markLimited(ref);
|
||||
expect(rotator.isCoolingDown(ref)).toBe(true);
|
||||
now += CREDENTIAL_INSTANCE_COOLDOWN_MS;
|
||||
expect(rotator.isCoolingDown(ref)).toBe(false);
|
||||
const event = await rotator.beginEvent({ providerId: "anthropic", startingInstanceId: "default", lane: "agent-heartbeat" });
|
||||
expect((await event?.next())?.instanceId).toBe("backup");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,15 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { CredentialInstanceRotator } from "../credential-instance-rotation.js";
|
||||
|
||||
describe("credential rotator DI identity", () => {
|
||||
it("uses one rotator object for the pauser and lane option bags", () => {
|
||||
const rotator = new CredentialInstanceRotator({
|
||||
instanceSource: { listInstances: () => [], getDefaultInstance: () => undefined },
|
||||
});
|
||||
const pauserOptions = { credentialRotator: rotator };
|
||||
const executorOptions = { credentialRotator: rotator };
|
||||
const heartbeatOptions = { credentialRotator: rotator };
|
||||
expect(executorOptions.credentialRotator).toBe(pauserOptions.credentialRotator);
|
||||
expect(heartbeatOptions.credentialRotator).toBe(pauserOptions.credentialRotator);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,50 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { withRateLimitRetry } from "../rate-limit-retry.js";
|
||||
|
||||
const limited = () => new Error("429 quota exceeded");
|
||||
|
||||
describe("withRateLimitRetry rotation", () => {
|
||||
it("retries immediately when a rotation supplies a different instance", async () => {
|
||||
const nextInstance = vi.fn().mockResolvedValueOnce({ providerId: "anthropic", instanceId: "backup" });
|
||||
let calls = 0;
|
||||
await expect(withRateLimitRetry(async () => {
|
||||
calls++;
|
||||
if (calls === 1) throw limited();
|
||||
return "ok";
|
||||
}, { rotation: { providerId: "anthropic", nextInstance }, baseDelayMs: 0 })).resolves.toBe("ok");
|
||||
expect(calls).toBe(2);
|
||||
expect(nextInstance).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("does not use candidateCount as a rotation cap", async () => {
|
||||
const nextInstance = vi.fn()
|
||||
.mockResolvedValueOnce({ providerId: "anthropic", instanceId: "b" })
|
||||
.mockResolvedValueOnce({ providerId: "anthropic", instanceId: "c" });
|
||||
let calls = 0;
|
||||
await expect(withRateLimitRetry(async () => {
|
||||
calls++;
|
||||
if (calls < 3) throw limited();
|
||||
return "ok";
|
||||
}, { rotation: { providerId: "anthropic", candidateCount: 0, nextInstance }, baseDelayMs: 0 })).resolves.toBe("ok");
|
||||
expect(nextInstance).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("does not invoke rotation for non-limit, transient auth, or aborted work", async () => {
|
||||
const nextInstance = vi.fn();
|
||||
await expect(withRateLimitRetry(async () => { throw new Error("network 500"); }, { rotation: { providerId: "anthropic", nextInstance } })).rejects.toThrow("network 500");
|
||||
await expect(withRateLimitRetry(async () => { throw new Error("401 authentication_error token expired"); }, { rotation: { providerId: "anthropic", nextInstance }, maxRetries: 0 })).rejects.toThrow("401");
|
||||
const abort = new AbortController(); abort.abort();
|
||||
await expect(withRateLimitRetry(async () => { throw limited(); }, { rotation: { providerId: "anthropic", nextInstance }, signal: abort.signal })).rejects.toThrow("429");
|
||||
expect(nextInstance).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("treats an undefined rotation exactly as the existing retry path", async () => {
|
||||
const calls: number[] = [];
|
||||
const operation = () => { calls.push(1); return Promise.reject(limited()); };
|
||||
await expect(withRateLimitRetry(operation, { maxRetries: 0 })).rejects.toThrow("429");
|
||||
const baselineCalls = calls.length;
|
||||
calls.length = 0;
|
||||
await expect(withRateLimitRetry(operation, { maxRetries: 0, rotation: { providerId: "anthropic", nextInstance: async () => undefined } })).rejects.toThrow("429");
|
||||
expect(calls).toHaveLength(baselineCalls);
|
||||
});
|
||||
});
|
||||
@@ -1,5 +1,6 @@
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
import { isUsageLimitError, UsageLimitPauser, checkSessionError } from "../usage-limit-detector.js";
|
||||
import { CredentialInstanceRotator } from "../credential-instance-rotation.js";
|
||||
|
||||
// ── isUsageLimitError classification tests ───────────────────────────
|
||||
|
||||
@@ -264,6 +265,19 @@ describe("UsageLimitPauser", () => {
|
||||
await expect(pauser.onProviderAvailable("anthropic")).resolves.toBe(0);
|
||||
expect(store.pauseTask).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("clears the shared credential cooldown before resuming provider parks", async () => {
|
||||
const store = createMockStore();
|
||||
const rotator = new CredentialInstanceRotator({
|
||||
instanceSource: { listInstances: () => [], getDefaultInstance: () => undefined },
|
||||
});
|
||||
const limited = { providerId: "anthropic", instanceId: "backup" };
|
||||
rotator.markLimited(limited);
|
||||
const pauser = new UsageLimitPauser(store, { credentialRotator: rotator });
|
||||
|
||||
await pauser.onProviderAvailable("anthropic");
|
||||
expect(rotator.isCoolingDown(limited)).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
/*
|
||||
|
||||
@@ -17,7 +17,7 @@
|
||||
* - onTerminated: Called when a heartbeat run is terminated
|
||||
*/
|
||||
|
||||
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, RunMutationContext, Settings, AgentConfigRevision, ReflectionStore, ChatStore, ChatRoom, ChatRoomMessage, AgentMemoryInclusionMode } from "@fusion/core";
|
||||
import { DEFAULT_PROVIDER_INSTANCE_ID, type AgentStore, type AgentHeartbeatRun, type HeartbeatInvocationSource, type AgentHeartbeatConfig, type AgentBudgetStatus, type Message, type MessageStore, type TaskStore, type TaskDetail, type AgentRole, type Agent, type InboxTask, type RunMutationContext, type Settings, type AgentConfigRevision, type ReflectionStore, type ChatStore, type ChatRoom, type ChatRoomMessage, type AgentMemoryInclusionMode } from "@fusion/core";
|
||||
import { AutoClaimSnapshotManager, resolveFreshAutoClaimCandidates, type AutoClaimCandidate } from "./auto-claim-snapshot.js";
|
||||
import {
|
||||
ApprovalRequestStore,
|
||||
@@ -91,6 +91,7 @@ import { acquireTaskWorktree } from "./worktree-acquisition.js";
|
||||
import { createRunAuditor, generateSyntheticRunId, type DatabaseMutationType, type EngineRunContext } from "./run-audit.js";
|
||||
import { promptWithFallback } from "./pi.js";
|
||||
import { withRateLimitRetry } from "./rate-limit-retry.js";
|
||||
import type { CredentialInstanceRotator } from "./credential-instance-rotation.js";
|
||||
import { buildAgentGatedActionSummary } from "./permanent-agent-gating.js";
|
||||
import { createResolvedAgentSession, extractRuntimeHint, resolveHeartbeatSessionModels, resolveExecutorFallbackThinkingLevel } from "./agent-session-helpers.js";
|
||||
import { resolveMcpServersForStore } from "./mcp-resolution.js";
|
||||
@@ -223,6 +224,8 @@ export interface HeartbeatMonitorOptions {
|
||||
reflectionService?: AgentReflectionService;
|
||||
/** Optional self-improvement service for periodic self-improve injection */
|
||||
selfImproveService?: SelfImproveServiceLike;
|
||||
/** Runtime-owned coordinator shared with executor retries and pauser recovery. */
|
||||
credentialRotator?: CredentialInstanceRotator;
|
||||
secretsStore?: Pick<import("@fusion/core").SecretsStore, "listEnvExportable">;
|
||||
/**
|
||||
* FNXC:WorktreeAcquisition 2026-07-09-00:00:
|
||||
@@ -313,6 +316,8 @@ export interface AgentSession {
|
||||
interface TrackedAgent {
|
||||
agentId: string;
|
||||
session: AgentSession;
|
||||
/** Cancels retry backoff when this tracked heartbeat is stopped or untracked. */
|
||||
abortController?: AbortController;
|
||||
runId: string;
|
||||
lastSeen: number; // timestamp from Date.now()
|
||||
missedHeartbeatReported: boolean;
|
||||
@@ -740,6 +745,7 @@ export class HeartbeatMonitor {
|
||||
* FN-8184 keeps the last settings-derived multiplier warm for synchronous
|
||||
* config resolution and reports-health so each applies it exactly once.
|
||||
*/
|
||||
private credentialRotator?: CredentialInstanceRotator;
|
||||
private cachedHeartbeatMultiplier = 1;
|
||||
private cachedHeartbeatMultiplierAt = 0;
|
||||
|
||||
@@ -766,6 +772,7 @@ export class HeartbeatMonitor {
|
||||
this.reflectionStore = options.reflectionStore;
|
||||
this.reflectionService = options.reflectionService;
|
||||
this.selfImproveService = options.selfImproveService;
|
||||
this.credentialRotator = options.credentialRotator;
|
||||
this.snapshotManager = options.snapshotManager ?? (this.taskStore ? new AutoClaimSnapshotManager({ taskStore: this.taskStore }) : undefined);
|
||||
this.secretsStore = options.secretsStore;
|
||||
}
|
||||
@@ -1398,10 +1405,17 @@ export class HeartbeatMonitor {
|
||||
* @param runId - The heartbeat run ID
|
||||
* @param sessionIdBefore - Optional session ID from before execution
|
||||
*/
|
||||
trackAgent(agentId: string, session: AgentSession, runId: string, sessionIdBefore?: string): void {
|
||||
trackAgent(
|
||||
agentId: string,
|
||||
session: AgentSession,
|
||||
runId: string,
|
||||
sessionIdBefore?: string,
|
||||
abortController?: AbortController,
|
||||
): void {
|
||||
const tracked: TrackedAgent = {
|
||||
agentId,
|
||||
session,
|
||||
abortController,
|
||||
runId,
|
||||
lastSeen: Date.now(),
|
||||
missedHeartbeatReported: false,
|
||||
@@ -1776,6 +1790,7 @@ export class HeartbeatMonitor {
|
||||
if (tracked) {
|
||||
heartbeatLog.log(`Stopping tracked run ${tracked.runId} for ${agentId}`);
|
||||
|
||||
tracked.abortController?.abort();
|
||||
try {
|
||||
tracked.session.dispose();
|
||||
} catch (error) {
|
||||
@@ -1931,6 +1946,8 @@ export class HeartbeatMonitor {
|
||||
* @param agentId - The agent ID
|
||||
*/
|
||||
untrackAgent(agentId: string): void {
|
||||
const tracked = this.trackedAgents.get(agentId);
|
||||
tracked?.abortController?.abort();
|
||||
this.trackedAgents.delete(agentId);
|
||||
}
|
||||
|
||||
@@ -3021,7 +3038,8 @@ export class HeartbeatMonitor {
|
||||
}
|
||||
|
||||
// Create agent session
|
||||
const { session } = await createResolvedAgentSession({
|
||||
const heartbeatRetryAbortController = new AbortController();
|
||||
let { session } = await createResolvedAgentSession({
|
||||
sessionPurpose: "heartbeat",
|
||||
runtimeHint: extractRuntimeHint(agent.runtimeConfig),
|
||||
pluginRunner: this.pluginRunner,
|
||||
@@ -3070,7 +3088,13 @@ export class HeartbeatMonitor {
|
||||
}
|
||||
|
||||
// Track for monitoring
|
||||
this.trackAgent(agentId, { dispose: () => session.dispose() }, run.id);
|
||||
this.trackAgent(
|
||||
agentId,
|
||||
{ dispose: () => session.dispose() },
|
||||
run.id,
|
||||
undefined,
|
||||
heartbeatRetryAbortController,
|
||||
);
|
||||
|
||||
try {
|
||||
// Build execution prompt
|
||||
@@ -3518,12 +3542,80 @@ export class HeartbeatMonitor {
|
||||
FNXC:AgentHeartbeat 2026-07-12-21:05:
|
||||
PR #2027 review (side-effect replay): the retry re-prompts the SAME session, whose transcript already contains any tool calls completed before the failure, so the model continues from its partial work rather than blindly re-executing it — the same continuation semantics executor/triage/merger rely on under this wrapper. A rotation 401 additionally fails on the turn's FIRST provider call (the stale token never reaches a tool call), so the dominant retry case has no partial work to duplicate.
|
||||
*/
|
||||
await withRateLimitRetry(() => promptWithFallback(session, executionPrompt), {
|
||||
let rotationEvent: import("./credential-instance-rotation.js").RotationEvent | undefined;
|
||||
let rotationDeclined = false;
|
||||
let activeInstanceId = heartbeatSessionModels.credentialInstanceId ?? DEFAULT_PROVIDER_INSTANCE_ID;
|
||||
let dispatchedRotation = false;
|
||||
/*
|
||||
FNXC:CredentialInstanceRotation 2026-08-01-09:07:
|
||||
A heartbeat rotates only after the shared retry classifier has identified a usage
|
||||
limit. Read live task/settings state and use the tracked run's abort signal at every
|
||||
retry boundary: a pause arriving mid-run must decline before opening an event. A fresh
|
||||
session is then resolved for the offered instance rather than mutating credentials
|
||||
on the live session.
|
||||
*/
|
||||
await withRateLimitRetry(async () => promptWithFallback(session, executionPrompt), {
|
||||
signal: heartbeatRetryAbortController.signal,
|
||||
rotation: this.credentialRotator && heartbeatSessionModels.defaultProvider ? {
|
||||
providerId: heartbeatSessionModels.defaultProvider,
|
||||
nextInstance: async () => {
|
||||
const [liveTask, liveSettings] = await Promise.all([
|
||||
taskId ? taskStore.getTask(taskId).catch(() => undefined) : Promise.resolve(undefined),
|
||||
taskStore.getSettings().catch(() => heartbeatModelSettings ?? ({} as Settings)),
|
||||
]);
|
||||
if (rotationDeclined || heartbeatRetryAbortController.signal.aborted
|
||||
|| (taskId && (!liveTask || liveTask.userPaused === true || liveTask.autoMerge === false))
|
||||
|| liveSettings.globalPause === true || liveSettings.enginePaused === true) return undefined;
|
||||
rotationEvent ??= await this.credentialRotator!.beginEvent({
|
||||
providerId: heartbeatSessionModels.defaultProvider!,
|
||||
startingInstanceId: activeInstanceId,
|
||||
lane: "agent-heartbeat",
|
||||
taskId,
|
||||
agentId,
|
||||
});
|
||||
if (!rotationEvent) { rotationDeclined = true; return undefined; }
|
||||
// FNXC:CredentialInstanceRotation 2026-08-01-11:34: Credential inventory may resolve after an operator pauses the task or engine. Re-read control state before cooldown/audit/dispatch side effects.
|
||||
const [postInventoryTask, postInventorySettings] = await Promise.all([
|
||||
taskId ? taskStore.getTask(taskId).catch(() => undefined) : Promise.resolve(undefined),
|
||||
taskStore.getSettings().catch(() => heartbeatModelSettings ?? ({} as Settings)),
|
||||
]);
|
||||
if (heartbeatRetryAbortController.signal.aborted
|
||||
|| (taskId && (!postInventoryTask || postInventoryTask.userPaused === true || postInventoryTask.autoMerge === false))
|
||||
|| postInventorySettings.globalPause === true || postInventorySettings.enginePaused === true) return undefined;
|
||||
this.credentialRotator!.markLimited({ providerId: heartbeatSessionModels.defaultProvider!, instanceId: activeInstanceId });
|
||||
if (dispatchedRotation) rotationEvent.recordOutcome("rotation-failed-limit");
|
||||
const next = await rotationEvent.next();
|
||||
if (!next) { rotationEvent.finishExhausted(); return undefined; }
|
||||
activeInstanceId = next.instanceId;
|
||||
dispatchedRotation = true;
|
||||
session.dispose();
|
||||
const created = await createResolvedAgentSession({
|
||||
sessionPurpose: "heartbeat", runtimeHint: extractRuntimeHint(agent.runtimeConfig), pluginRunner: this.pluginRunner,
|
||||
cwd: sessionCwd, systemPrompt: systemPromptFinal, systemPromptLayers: heartbeatLayers, tools: "coding", customTools: heartbeatTools,
|
||||
defaultProvider: heartbeatSessionModels.defaultProvider, defaultModelId: heartbeatSessionModels.defaultModelId,
|
||||
credentialInstanceId: activeInstanceId, fallbackProvider: heartbeatSessionModels.fallbackProvider,
|
||||
fallbackModelId: heartbeatSessionModels.fallbackModelId,
|
||||
fallbackThinkingLevel: resolveExecutorFallbackThinkingLevel(undefined, heartbeatModelSettings), runAuditor: audit, settings: heartbeatModelSettings,
|
||||
mcpServers: heartbeatMcp.servers,
|
||||
onText: (delta) => { outputLength += delta.length; appendStdoutExcerpt(delta); agentLogger?.onText(delta); },
|
||||
onThinking: (delta) => agentLogger?.onThinking(delta),
|
||||
onToolStart: (name, args) => agentLogger?.onToolStart(name, args),
|
||||
onToolEnd: (name, isError, result) => { toolCallCount++; agentLogger?.onToolEnd(name, isError, result); },
|
||||
...(skillContext.skillSelectionContext ? { skillSelection: skillContext.skillSelectionContext } : {}),
|
||||
...(skillContext.additionalSkillPaths.length > 0 ? { additionalSkillPaths: skillContext.additionalSkillPaths } : {}),
|
||||
actionGateContext: this.buildActionGateContext(agent, taskId, run.id, heartbeatModelSettings?.defaultAgentPermissionPolicy),
|
||||
permanentAgentGating: this.buildPermanentAgentGatingContext(agent, taskId, run.id, heartbeatModelSettings?.defaultAgentPermissionPolicy),
|
||||
});
|
||||
session = created.session;
|
||||
return next;
|
||||
},
|
||||
} : undefined,
|
||||
onRetry: (attempt, delayMs, retryError) => {
|
||||
const delaySec = Math.round(delayMs / 1000);
|
||||
heartbeatLog.warn(`Agent ${agentId} heartbeat prompt hit retryable provider error — retry ${attempt} in ${delaySec}s: ${retryError.message}`);
|
||||
},
|
||||
});
|
||||
if (dispatchedRotation) rotationEvent?.recordOutcome("rotation-succeeded");
|
||||
|
||||
// Capture real per-session token counts from pi-coding-agent's
|
||||
// SessionStats. Falls back to a 4-chars-per-token estimate of output
|
||||
|
||||
175
packages/engine/src/credential-instance-rotation.ts
Normal file
175
packages/engine/src/credential-instance-rotation.ts
Normal file
@@ -0,0 +1,175 @@
|
||||
import {
|
||||
formatProviderInstanceKey,
|
||||
parseProviderInstanceKey,
|
||||
type ProviderInstanceRef,
|
||||
} from "@fusion/core";
|
||||
|
||||
export const CREDENTIAL_INSTANCE_COOLDOWN_MS = 15 * 60_000;
|
||||
|
||||
export type RotationLane = "executor-step" | "executor-agent" | "agent-heartbeat";
|
||||
|
||||
type InstanceSource = {
|
||||
listInstances(providerId: string): ProviderInstanceRef[] | Promise<ProviderInstanceRef[]>;
|
||||
getDefaultInstance(providerId: string): ProviderInstanceRef | undefined | Promise<ProviderInstanceRef | undefined>;
|
||||
};
|
||||
|
||||
type RotationAudit = (type: string, metadata: Record<string, unknown>) => void | Promise<void>;
|
||||
type CooldownState = ReadonlyMap<string, number>;
|
||||
|
||||
export interface RotationEvent {
|
||||
readonly candidateCount: number;
|
||||
next(): Promise<ProviderInstanceRef | undefined>;
|
||||
recordOutcome(outcome: "rotation-succeeded" | "rotation-failed-limit"): void;
|
||||
finishExhausted(): void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Produce the finite, round-robin candidate list for one limit event.
|
||||
* Exported independently so cooldown and ordering rules remain testable without sessions.
|
||||
*/
|
||||
export function createRotationPlan(
|
||||
providerId: string,
|
||||
startingInstanceId: string,
|
||||
instances: readonly ProviderInstanceRef[],
|
||||
cooldownState: CooldownState,
|
||||
now: number,
|
||||
): { candidates: ProviderInstanceRef[]; skippedCooldownInstanceIds: string[] } {
|
||||
const providerInstances = instances
|
||||
.filter((ref) => ref.providerId === providerId)
|
||||
.sort((left, right) => left.instanceId.localeCompare(right.instanceId));
|
||||
const startIndex = providerInstances.findIndex((ref) => ref.instanceId === startingInstanceId);
|
||||
const ordered = startIndex < 0
|
||||
? providerInstances
|
||||
: [...providerInstances.slice(startIndex + 1), ...providerInstances.slice(0, startIndex)];
|
||||
const skippedCooldownInstanceIds: string[] = [];
|
||||
const candidates = ordered.filter((ref) => {
|
||||
if (ref.instanceId === startingInstanceId) return false;
|
||||
if ((cooldownState.get(formatProviderInstanceKey(ref)) ?? 0) > now) {
|
||||
skippedCooldownInstanceIds.push(ref.instanceId);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
});
|
||||
return { candidates, skippedCooldownInstanceIds };
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:CredentialInstanceRotation 2026-08-01-06:19:
|
||||
A provider usage limit rotates to the next eligible credential instance before the existing pause and backoff fallback. Eligibility is configured inventory, not the starting instance, no active cooldown, and no earlier offer in this finite event; no provider health probe is introduced.
|
||||
|
||||
Providers with zero or one configured instance are an intentionally total no-op: they create no event, write no cooldown, and emit no audit rows, preserving single-instance behavior byte-for-byte. A multi-instance event with every other instance cooling down is different: it is real exhaustion and emits the fallback audit row.
|
||||
|
||||
The runtime owns one process-local, self-expiring cooldown map for all lanes. It is a best-effort skip hint, never persisted or a correctness gate, so a dashboard health monitor pauser without this rotator can at worst leave a candidate skipped until expiry. The finite RotationEvent is the sole boundedness authority; retry wrappers deliberately keep no candidate counter.
|
||||
*/
|
||||
export class CredentialInstanceRotator {
|
||||
private readonly cooldowns = new Map<string, number>();
|
||||
private readonly now: () => number;
|
||||
|
||||
constructor(private readonly options: {
|
||||
instanceSource: InstanceSource;
|
||||
now?: () => number;
|
||||
recordRunAuditEvent?: RotationAudit;
|
||||
}) {
|
||||
this.now = options.now ?? Date.now;
|
||||
}
|
||||
|
||||
async beginEvent(input: {
|
||||
providerId: string;
|
||||
startingInstanceId: string;
|
||||
lane: RotationLane;
|
||||
taskId?: string;
|
||||
agentId?: string;
|
||||
}): Promise<RotationEvent | undefined> {
|
||||
const instances = await this.options.instanceSource.listInstances(input.providerId);
|
||||
// The inventory-size gate must precede all observable rotation state.
|
||||
if (instances.length <= 1) return undefined;
|
||||
|
||||
this.pruneExpiredCooldowns();
|
||||
const { candidates, skippedCooldownInstanceIds } = createRotationPlan(
|
||||
input.providerId,
|
||||
input.startingInstanceId,
|
||||
instances,
|
||||
this.cooldowns,
|
||||
this.now(),
|
||||
);
|
||||
let nextIndex = 0;
|
||||
let lastAttempt: { ref: ProviderInstanceRef; attempt: number } | undefined;
|
||||
let exhausted = false;
|
||||
const attempted: ProviderInstanceRef[] = [];
|
||||
const audit = (type: string, metadata: Record<string, unknown>) => {
|
||||
try { void Promise.resolve(this.options.recordRunAuditEvent?.(type, metadata)).catch(() => undefined); } catch { /* audit is non-fatal */ }
|
||||
};
|
||||
const common = {
|
||||
providerId: input.providerId,
|
||||
lane: input.lane,
|
||||
...(input.taskId ? { taskId: input.taskId } : {}),
|
||||
...(input.agentId ? { agentId: input.agentId } : {}),
|
||||
};
|
||||
return {
|
||||
candidateCount: candidates.length,
|
||||
next: async () => {
|
||||
const ref = candidates[nextIndex++];
|
||||
if (!ref) return undefined;
|
||||
lastAttempt = { ref, attempt: nextIndex };
|
||||
attempted.push(ref);
|
||||
audit("credential:instance-rotation-attempt", {
|
||||
...common, fromInstanceId: input.startingInstanceId, toInstanceId: ref.instanceId,
|
||||
attempt: nextIndex, candidateCount: candidates.length, outcome: "rotated",
|
||||
});
|
||||
return ref;
|
||||
},
|
||||
recordOutcome: (outcome) => {
|
||||
if (!lastAttempt) return;
|
||||
audit("credential:instance-rotation-outcome", {
|
||||
...common, toInstanceId: lastAttempt.ref.instanceId, attempt: lastAttempt.attempt, outcome,
|
||||
});
|
||||
lastAttempt = undefined;
|
||||
},
|
||||
finishExhausted: () => {
|
||||
if (exhausted) return;
|
||||
exhausted = true;
|
||||
audit("credential:instance-rotation-exhausted", {
|
||||
...common, instanceCount: instances.length, attemptedCount: attempted.length,
|
||||
attemptedInstanceIds: attempted.map((ref) => ref.instanceId), startingInstanceId: input.startingInstanceId,
|
||||
skippedCooldownCount: skippedCooldownInstanceIds.length, skippedCooldownInstanceIds,
|
||||
outcome: "fell-back-to-pause",
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
markLimited(ref: ProviderInstanceRef): void {
|
||||
this.pruneExpiredCooldowns();
|
||||
this.cooldowns.set(formatProviderInstanceKey(ref), this.now() + CREDENTIAL_INSTANCE_COOLDOWN_MS);
|
||||
}
|
||||
|
||||
/**
|
||||
* FNXC:CredentialInstanceRotation 2026-08-01-11:22:
|
||||
* Cooldowns are process-local hints, so planning removes elapsed entries before
|
||||
* reading them. This prevents providers that are only occasionally limited from
|
||||
* retaining stale map state, while preserving the no-op gate before any cleanup.
|
||||
*/
|
||||
private pruneExpiredCooldowns(): void {
|
||||
const now = this.now();
|
||||
for (const [key, until] of this.cooldowns) {
|
||||
if (until <= now) this.cooldowns.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
clearCooldowns(providerId: string): void {
|
||||
for (const key of this.cooldowns.keys()) {
|
||||
// Reuse core's grammar rather than coupling cooldown cleanup to its encoding.
|
||||
if (parseProviderInstanceKey(key)?.providerId === providerId) this.cooldowns.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
isCoolingDown(ref: ProviderInstanceRef): boolean {
|
||||
const key = formatProviderInstanceKey(ref);
|
||||
const until = this.cooldowns.get(key) ?? 0;
|
||||
if (until <= this.now()) {
|
||||
this.cooldowns.delete(key);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
@@ -12,7 +12,7 @@ const WORKFLOW_THINKING_LEVEL_SET: ReadonlySet<string> = new Set(THINKING_LEVELS
|
||||
import { basename, delimiter, isAbsolute, join, relative, resolve as resolvePath } from "node:path";
|
||||
import { existsSync, lstatSync, realpathSync } from "node:fs";
|
||||
import { readFile, rm, writeFile } from "node:fs/promises";
|
||||
import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, AsyncMissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode, WorkflowIrNodeKind, WorkflowStepResult as CoreWorkflowStepResult, ThinkingLevel } from "@fusion/core";
|
||||
import { DEFAULT_PROVIDER_INSTANCE_ID, type ProviderInstanceRef, type TaskStore, type Task, type TaskDetail, type TaskTokenUsage, type StepStatus, type Settings, type WorkflowStep, type MissionStore, type AsyncMissionStore, type Slice, type AgentState, type AgentCapability, type RunMutationContext, type AgentHeartbeatConfig, type Agent, type AgentMemoryInclusionMode, type ProjectSettings, type MergeResult, type WorkflowIrNode, type WorkflowIrNodeKind, type WorkflowStepResult as CoreWorkflowStepResult, type ThinkingLevel } from "@fusion/core";
|
||||
import { getUnmetSchedulingDependencies } from "./scheduler.js";
|
||||
import type { ImplementationExit, ImplementationExitReporter } from "./executor/implementation-exit.js";
|
||||
import { emitWorkflowLifecycleEvent } from "@fusion/core";
|
||||
@@ -191,6 +191,7 @@ import { TokenCapDetector } from "./token-cap-detector.js";
|
||||
import { isUsageLimitError, checkSessionError, type UsageLimitPauser } from "./usage-limit-detector.js";
|
||||
import { isNonContinuableSessionError, isNonPlanDefectPlanReviewFailure, isSessionContentionError, isTransientError, isSilentTransientError } from "./transient-error-detector.js";
|
||||
import { withRateLimitRetry } from "./rate-limit-retry.js";
|
||||
import type { CredentialInstanceRotator } from "./credential-instance-rotation.js";
|
||||
import {
|
||||
detectExternalIntegrationEvidenceGaps,
|
||||
formatExternalIntegrationEvidenceDiagnostic,
|
||||
@@ -1668,6 +1669,8 @@ export interface TaskExecutorOptions {
|
||||
* Parks only tasks routed through the provider whose API limit was detected.
|
||||
*/
|
||||
usageLimitPauser?: UsageLimitPauser;
|
||||
/** Runtime-owned credential rotation inventory/cooldown coordinator. */
|
||||
credentialRotator?: CredentialInstanceRotator;
|
||||
/** Stuck task detector — monitors agent sessions for stagnation and triggers recovery. */
|
||||
stuckTaskDetector?: StuckTaskDetector;
|
||||
/** AgentStore for tracking spawned child agents. If not provided, spawning is disabled. */
|
||||
@@ -13323,6 +13326,62 @@ export class TaskExecutor {
|
||||
|
||||
let accumulatedStepTokenUsage = detail.tokenUsage;
|
||||
const tokenUsageRecordedSteps = new Set<number>();
|
||||
let stepRotationEvent: import("./credential-instance-rotation.js").RotationEvent | undefined;
|
||||
let stepRotationDeclined = false;
|
||||
let stepDispatchedRotation = false;
|
||||
const initialStepSessionModel = resolveExecutorSessionModel(
|
||||
detail.modelProvider,
|
||||
detail.modelId,
|
||||
settings,
|
||||
(stepIdentityAgent?.runtimeConfig ?? undefined) as Record<string, unknown> | undefined,
|
||||
detail.credentialInstanceId ?? undefined,
|
||||
);
|
||||
let activeStepInstanceRef: ProviderInstanceRef | undefined = initialStepSessionModel.provider
|
||||
? {
|
||||
providerId: initialStepSessionModel.provider,
|
||||
instanceId: initialStepSessionModel.credentialInstanceId ?? DEFAULT_PROVIDER_INSTANCE_ID,
|
||||
}
|
||||
: undefined;
|
||||
const stepExecutorRef: { current?: StepSessionExecutor } = {};
|
||||
const nextStepInstance = async (): Promise<ProviderInstanceRef | undefined> => {
|
||||
/*
|
||||
FNXC:CredentialInstanceRotation 2026-08-01-11:22:
|
||||
Executor-step retries refresh task and project pause state at the limit
|
||||
boundary, rather than trusting dispatch snapshots. A pause arriving while
|
||||
a session is in flight must prevent an autonomous billed-account switch.
|
||||
*/
|
||||
const [liveTask, liveSettings] = await Promise.all([
|
||||
this.store.getTask(task.id).catch(() => undefined),
|
||||
this.store.getSettings().catch(() => settings),
|
||||
]);
|
||||
if (stepRotationDeclined || this.pausedAborted.has(task.id) || !liveTask
|
||||
|| liveTask.userPaused === true || liveTask.autoMerge === false
|
||||
|| liveSettings.globalPause === true || liveSettings.enginePaused === true
|
||||
|| !activeStepInstanceRef?.providerId) return undefined;
|
||||
stepRotationEvent ??= await this.options.credentialRotator?.beginEvent({
|
||||
providerId: activeStepInstanceRef.providerId,
|
||||
startingInstanceId: activeStepInstanceRef.instanceId,
|
||||
lane: "executor-step",
|
||||
taskId: task.id,
|
||||
});
|
||||
if (!stepRotationEvent) { stepRotationDeclined = true; return undefined; }
|
||||
// FNXC:CredentialInstanceRotation 2026-08-01-11:34: beginEvent awaits credential inventory, so repeat the human-control check after it resolves. A pause that races this await must prevent cooldown writes and credential dispatch.
|
||||
const [postInventoryTask, postInventorySettings] = await Promise.all([
|
||||
this.store.getTask(task.id).catch(() => undefined),
|
||||
this.store.getSettings().catch(() => settings),
|
||||
]);
|
||||
if (this.pausedAborted.has(task.id) || !postInventoryTask
|
||||
|| postInventoryTask.userPaused === true || postInventoryTask.autoMerge === false
|
||||
|| postInventorySettings.globalPause === true || postInventorySettings.enginePaused === true) return undefined;
|
||||
this.options.credentialRotator?.markLimited(activeStepInstanceRef);
|
||||
if (stepDispatchedRotation) stepRotationEvent.recordOutcome("rotation-failed-limit");
|
||||
const next = await stepRotationEvent.next();
|
||||
if (!next) { stepRotationEvent.finishExhausted(); return undefined; }
|
||||
activeStepInstanceRef = next;
|
||||
stepDispatchedRotation = true;
|
||||
await stepExecutorRef.current?.retargetCredentialInstance(next);
|
||||
return next;
|
||||
};
|
||||
/*
|
||||
FNXC:WorkflowStepControl 2026-06-29-10:15:
|
||||
Graph-pinned step sessions are lifecycle-owned by the workflow graph, not by the legacy executor prompt/tools. Their callback projection must use source:"graph" so independent steps can finish out of index order and so duplicate graph runner writes do not trigger the legacy sequential fn_task_update guard.
|
||||
@@ -13348,20 +13407,7 @@ export class TaskExecutor {
|
||||
* effective column-agent runtime config used to create the session.
|
||||
*/
|
||||
credentialInstanceId: detail.credentialInstanceId,
|
||||
resolveCredentialInstanceRetarget: async () => {
|
||||
const liveDetail = await this.store.getTask(task.id);
|
||||
if (!liveDetail) return undefined;
|
||||
const resolvedModel = resolveExecutorSessionModel(
|
||||
liveDetail.modelProvider,
|
||||
liveDetail.modelId,
|
||||
settings,
|
||||
(stepIdentityAgent?.runtimeConfig ?? undefined) as Record<string, unknown> | undefined,
|
||||
liveDetail.credentialInstanceId ?? undefined,
|
||||
);
|
||||
return resolvedModel.provider && resolvedModel.credentialInstanceId
|
||||
? { providerId: resolvedModel.provider, instanceId: resolvedModel.credentialInstanceId }
|
||||
: undefined;
|
||||
},
|
||||
resolveCredentialInstanceRetarget: nextStepInstance,
|
||||
// Attribute the per-step run auditor to the column agent when it governs
|
||||
// (U4); absent → StepSessionExecutor falls back to assignedAgentId.
|
||||
effectiveAgentId: stepColumnAgent?.agent.id,
|
||||
@@ -13446,6 +13492,7 @@ export class TaskExecutor {
|
||||
});
|
||||
},
|
||||
});
|
||||
stepExecutorRef.current = stepExecutor;
|
||||
this.setActiveStepExecutor(task.id, stepExecutor, worktreePath, this.createSeenSteeringIds(detail));
|
||||
|
||||
const stepWork = async () => {
|
||||
@@ -13703,6 +13750,11 @@ export class TaskExecutor {
|
||||
};
|
||||
|
||||
const retryableStepWork = () => withRateLimitRetry(stepWork, {
|
||||
signal: this.activeWorkflowGraphAbortControllers.get(task.id)?.signal,
|
||||
rotation: this.options.credentialRotator && activeStepInstanceRef ? {
|
||||
providerId: activeStepInstanceRef.providerId,
|
||||
nextInstance: nextStepInstance,
|
||||
} : undefined,
|
||||
onRetry: (attempt, delayMs, error) => {
|
||||
const delaySec = Math.round(delayMs / 1000);
|
||||
executorLog.warn(`⏳ ${task.id} rate limited — retry ${attempt} in ${delaySec}s: ${error.message}`);
|
||||
@@ -13715,6 +13767,7 @@ export class TaskExecutor {
|
||||
|
||||
try {
|
||||
await this.runWithExecutorSemaphore(task.id, retryableStepWork);
|
||||
if (stepDispatchedRotation) stepRotationEvent?.recordOutcome("rotation-succeeded");
|
||||
} catch (err: unknown) {
|
||||
const { message: errorMessage, detail: errorDetail, stack: errorStack } = formatError(err);
|
||||
if (this.depAborted.has(task.id)) {
|
||||
@@ -14194,6 +14247,11 @@ export class TaskExecutor {
|
||||
},
|
||||
});
|
||||
|
||||
let agentRotationEvent: import("./credential-instance-rotation.js").RotationEvent | undefined;
|
||||
let agentRotationDeclined = false;
|
||||
let agentDispatchedRotation = false;
|
||||
let activeAgentInstanceRef: ProviderInstanceRef | undefined;
|
||||
|
||||
const agentWork = async () => {
|
||||
// Resolve model settings using canonical lane hierarchy:
|
||||
// 1. Task override pair (modelProvider + modelId)
|
||||
@@ -14214,9 +14272,12 @@ export class TaskExecutor {
|
||||
overrideColumnGovernsInitialSession ? undefined : detail.modelId,
|
||||
settings,
|
||||
(identityAgent?.runtimeConfig ?? undefined) as Record<string, unknown> | undefined,
|
||||
overrideColumnGovernsInitialSession ? undefined : detail.credentialInstanceId,
|
||||
overrideColumnGovernsInitialSession ? undefined : activeAgentInstanceRef?.instanceId ?? detail.credentialInstanceId,
|
||||
);
|
||||
const { provider: executorProvider, modelId: executorModelId } = executorSessionModel;
|
||||
activeAgentInstanceRef ??= executorProvider
|
||||
? { providerId: executorProvider, instanceId: executorSessionModel.credentialInstanceId ?? DEFAULT_PROVIDER_INSTANCE_ID }
|
||||
: undefined;
|
||||
const { provider: executorFallbackProvider, modelId: executorFallbackModelId } = resolveExecutorFallbackModel(settings);
|
||||
const executorSessionThinkingSource = this.graphSeamThinkingLevel.get(task.id) ?? detail.thinkingLevel;
|
||||
const executorThinkingLevel = resolveExecutorThinkingLevel(executorSessionThinkingSource, settings);
|
||||
@@ -14317,7 +14378,7 @@ export class TaskExecutor {
|
||||
onToolEnd: agentLogger.onToolEnd,
|
||||
defaultProvider: executorProvider,
|
||||
defaultModelId: executorModelId,
|
||||
...(executorSessionModel.credentialInstanceId ? { credentialInstanceId: executorSessionModel.credentialInstanceId } : {}),
|
||||
...(activeAgentInstanceRef ? { credentialInstanceId: activeAgentInstanceRef.instanceId } : {}),
|
||||
fallbackProvider: executorFallbackProvider,
|
||||
fallbackModelId: executorFallbackModelId,
|
||||
fallbackThinkingLevel: executorFallbackThinkingLevel,
|
||||
@@ -15026,6 +15087,51 @@ export class TaskExecutor {
|
||||
};
|
||||
|
||||
const retryableWork = () => withRateLimitRetry(agentWork, {
|
||||
signal: this.activeWorkflowGraphAbortControllers.get(task.id)?.signal,
|
||||
rotation: this.options.credentialRotator ? {
|
||||
providerId: activeAgentInstanceRef?.providerId ?? detail.modelProvider ?? "",
|
||||
nextInstance: async () => {
|
||||
/*
|
||||
FNXC:CredentialInstanceRotation 2026-08-01-11:05:
|
||||
Executor agent runs rotate only after the shared retry helper classifies a
|
||||
usage limit. Live task/settings reads and the executor pause-abort marker
|
||||
bail before opening an event, because a pause arriving mid-run cannot
|
||||
authorize changing the billed credential. A successful offer causes
|
||||
agentWork to construct a fresh session; a non-limit failure intentionally
|
||||
leaves its attempt without an outcome row.
|
||||
*/
|
||||
const [liveTask, liveSettings] = await Promise.all([
|
||||
this.store.getTask(task.id).catch(() => undefined),
|
||||
this.store.getSettings().catch(() => settings),
|
||||
]);
|
||||
if (agentRotationDeclined || this.pausedAborted.has(task.id) || !liveTask
|
||||
|| liveTask.userPaused === true || liveTask.autoMerge === false
|
||||
|| liveSettings.globalPause === true || liveSettings.enginePaused === true
|
||||
|| !activeAgentInstanceRef?.providerId) return undefined;
|
||||
agentRotationEvent ??= await this.options.credentialRotator!.beginEvent({
|
||||
providerId: activeAgentInstanceRef.providerId,
|
||||
startingInstanceId: activeAgentInstanceRef.instanceId,
|
||||
lane: "executor-agent",
|
||||
taskId: task.id,
|
||||
});
|
||||
if (!agentRotationEvent) { agentRotationDeclined = true; return undefined; }
|
||||
// FNXC:CredentialInstanceRotation 2026-08-01-11:34: Inventory lookup is asynchronous; re-check human control before this retry marks a credential limited or offers another billed account.
|
||||
const [postInventoryTask, postInventorySettings] = await Promise.all([
|
||||
this.store.getTask(task.id).catch(() => undefined),
|
||||
this.store.getSettings().catch(() => settings),
|
||||
]);
|
||||
if (this.pausedAborted.has(task.id) || !postInventoryTask
|
||||
|| postInventoryTask.userPaused === true || postInventoryTask.autoMerge === false
|
||||
|| postInventorySettings.globalPause === true || postInventorySettings.enginePaused === true) return undefined;
|
||||
this.options.credentialRotator!.markLimited(activeAgentInstanceRef);
|
||||
if (agentDispatchedRotation) agentRotationEvent.recordOutcome("rotation-failed-limit");
|
||||
const next = await agentRotationEvent.next();
|
||||
if (!next) { agentRotationEvent.finishExhausted(); return undefined; }
|
||||
activeAgentInstanceRef = next;
|
||||
agentDispatchedRotation = true;
|
||||
return next;
|
||||
},
|
||||
} : undefined,
|
||||
onRetry: (attempt, delayMs, error) => {
|
||||
const delaySec = Math.round(delayMs / 1000);
|
||||
executorLog.warn(`⏳ ${task.id} rate limited — retry ${attempt} in ${delaySec}s: ${error.message}`);
|
||||
@@ -15037,6 +15143,7 @@ export class TaskExecutor {
|
||||
});
|
||||
|
||||
await this.runWithExecutorSemaphore(task.id, retryableWork);
|
||||
if (agentDispatchedRotation) agentRotationEvent?.recordOutcome("rotation-succeeded");
|
||||
} catch (err: unknown) {
|
||||
const { message: errorMessage, detail: errorDetail, stack: errorStack } = formatError(err);
|
||||
if (this.depAborted.has(task.id)) {
|
||||
|
||||
@@ -723,6 +723,13 @@ export {
|
||||
} from "./worktrunk-failure-handler.js";
|
||||
export { isUsageLimitError, UsageLimitPauser } from "./usage-limit-detector.js";
|
||||
export { withRateLimitRetry } from "./rate-limit-retry.js";
|
||||
export {
|
||||
CredentialInstanceRotator,
|
||||
CREDENTIAL_INSTANCE_COOLDOWN_MS,
|
||||
createRotationPlan,
|
||||
type RotationEvent,
|
||||
type RotationLane,
|
||||
} from "./credential-instance-rotation.js";
|
||||
export { ResearchOrchestrator, type ResearchOrchestratorOptions, type ResearchOrchestratorStatus, type ResearchOrchestratorStartOptions } from "./research-orchestrator.js";
|
||||
export {
|
||||
ExperimentExecutor,
|
||||
|
||||
@@ -4072,6 +4072,10 @@ export class ProjectEngine {
|
||||
const agentStore = (this.runtime as any).agentStore;
|
||||
|
||||
const usageLimitPauser = (this.runtime as any).usageLimitPauser;
|
||||
// FNXC:CredentialInstanceRotation 2026-08-01-11:05:
|
||||
// Preserve the runtime-owned rotator identity in downstream option bags;
|
||||
// merger does not opt into rotation, so this is forwarding only.
|
||||
const credentialRotator = (this.runtime as any).credentialRotator;
|
||||
|
||||
const rawMerge = async () => {
|
||||
const abortSignal = this.claimActiveMerge(taskId);
|
||||
@@ -4083,6 +4087,7 @@ export class ProjectEngine {
|
||||
manual: hasManualResolver,
|
||||
pool,
|
||||
usageLimitPauser,
|
||||
credentialRotator,
|
||||
agentStore,
|
||||
pluginRunner: this.getPluginRunner(),
|
||||
signal: abortSignal,
|
||||
|
||||
@@ -23,6 +23,7 @@
|
||||
* (transient-error retry, failure marking, etc.) is unaffected.
|
||||
*/
|
||||
|
||||
import type { ProviderInstanceRef } from "@fusion/core";
|
||||
import { isUsageLimitError } from "./usage-limit-detector.js";
|
||||
import { isTransientAuthCredentialError } from "./transient-error-detector.js";
|
||||
|
||||
@@ -64,6 +65,16 @@ export interface RateLimitRetryOptions {
|
||||
* re-throws the last error immediately. Essential for paused / cancelled tasks.
|
||||
*/
|
||||
signal?: AbortSignal;
|
||||
/**
|
||||
* Optional credential-instance reroute. The finite RotationEvent behind
|
||||
* nextInstance owns boundedness; candidateCount is diagnostics only.
|
||||
*/
|
||||
rotation?: {
|
||||
providerId: string;
|
||||
currentInstanceId?: string;
|
||||
candidateCount?: number;
|
||||
nextInstance(error: unknown): Promise<ProviderInstanceRef | undefined>;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -100,6 +111,7 @@ export async function withRateLimitRetry<T>(
|
||||
maxDelayMs = 120_000,
|
||||
onRetry,
|
||||
signal,
|
||||
rotation,
|
||||
} = options;
|
||||
|
||||
let lastError: Error | undefined;
|
||||
@@ -140,17 +152,27 @@ export async function withRateLimitRetry<T>(
|
||||
continue;
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:CredentialInstanceRotation 2026-08-01-06:21:
|
||||
Abort remains the first usage-limit short-circuit: cancellation never asks a
|
||||
credential rotator to reroute or enters backoff. A successful rotation retries
|
||||
immediately and decrements the loop counter like transient-auth recovery, so
|
||||
using another configured account does not consume the existing rate-limit budget.
|
||||
RotationEvent owns the finite offered-once bound; this wrapper intentionally
|
||||
keeps no cap, and an always-undefined hook is indistinguishable from no hook.
|
||||
*/
|
||||
if (signal?.aborted) throw lastError;
|
||||
if (rotation && await rotation.nextInstance(error)) {
|
||||
attempt--;
|
||||
continue;
|
||||
}
|
||||
|
||||
// FNXC:ProviderRateLimitIsolation 2026-07-21-18:00: exhaustion parks only
|
||||
// the affected provider-routed task instead of stopping the project.
|
||||
if (attempt >= maxRetries) {
|
||||
throw lastError;
|
||||
}
|
||||
|
||||
// Check abort before sleeping
|
||||
if (signal?.aborted) {
|
||||
throw lastError;
|
||||
}
|
||||
|
||||
// Exponential backoff with ±10 % jitter
|
||||
const rawDelay = Math.min(baseDelayMs * 2 ** attempt, maxDelayMs);
|
||||
const jitter = rawDelay * 0.1 * (2 * Math.random() - 1); // ±10 %
|
||||
|
||||
@@ -53,6 +53,8 @@ import { runtimeLog } from "../logger.js";
|
||||
import { getActiveNotificationService } from "../notifier.js";
|
||||
import { StuckTaskDetector } from "../stuck-task-detector.js";
|
||||
import { UsageLimitPauser } from "../usage-limit-detector.js";
|
||||
import { CredentialInstanceRotator } from "../credential-instance-rotation.js";
|
||||
import { createFusionAuthStorage } from "../auth-storage.js";
|
||||
import { SelfHealingManager, VALIDATOR_RUN_STALE_MAX_AGE_MS } from "../self-healing.js";
|
||||
import { RestartRecoveryCoordinator } from "../restart-recovery-coordinator.js";
|
||||
import { MeshLeaseManager } from "../mesh-lease-manager.js";
|
||||
@@ -712,6 +714,8 @@ export class InProcessRuntime
|
||||
*/
|
||||
private cliAgentRuntime?: BootstrappedCliAgentRuntime;
|
||||
private usageLimitPauser?: UsageLimitPauser;
|
||||
/** One runtime-owned cooldown map keeps executor and recovery paths coherent. */
|
||||
private credentialRotator?: CredentialInstanceRotator;
|
||||
/** FNXC:PlanReviewLease 2026-07-26-20:42: cluster node id stamped onto review-gate leases; undefined until start() resolves it, or if resolution fails. */
|
||||
private localNodeId?: string;
|
||||
private selfHealingManager?: SelfHealingManager;
|
||||
@@ -846,7 +850,27 @@ export class InProcessRuntime
|
||||
FNXC:ProviderRateLimitIsolation 2026-07-19-19:10:
|
||||
Every project runtime owns one usage-limit coordinator and shares it across executor, triage, reviewer, and merger surfaces. Runtime isolation replaced the old dashboard-level construction site; constructing it here prevents a silently undefined pauser while keeping a provider outage local to the affected project/task.
|
||||
*/
|
||||
this.usageLimitPauser ??= new UsageLimitPauser(this.taskStore);
|
||||
this.credentialRotator ??= new CredentialInstanceRotator({
|
||||
instanceSource: createFusionAuthStorage(),
|
||||
// FNXC:CredentialInstanceRotation 2026-08-01-11:05:
|
||||
// Rotation evidence is emitted through the runtime-owned audit seam. Metadata
|
||||
// is supplied by the rotator as ids/counts/outcomes only; audit failures stay
|
||||
// non-fatal so an observability outage cannot prevent rate-limit recovery.
|
||||
recordRunAuditEvent: async (mutationType, metadata) => {
|
||||
await this.taskStore.recordRunAuditEvent?.({
|
||||
taskId: typeof metadata.taskId === "string" ? metadata.taskId : undefined,
|
||||
agentId: typeof metadata.agentId === "string" ? metadata.agentId : "runtime",
|
||||
runId: generateSyntheticRunId("credential-instance-rotation", typeof metadata.taskId === "string" ? metadata.taskId : String(metadata.providerId ?? "unknown")),
|
||||
domain: "database",
|
||||
mutationType,
|
||||
target: String(metadata.providerId ?? "unknown"),
|
||||
metadata,
|
||||
});
|
||||
},
|
||||
});
|
||||
this.usageLimitPauser ??= new UsageLimitPauser(this.taskStore, {
|
||||
credentialRotator: this.credentialRotator,
|
||||
});
|
||||
|
||||
// Initialize MessageStore early so TaskExecutor receives send_message capability.
|
||||
// FNXC:RuntimeSatelliteAsync 2026-06-24-12:45:
|
||||
@@ -1258,6 +1282,7 @@ export class InProcessRuntime
|
||||
getLocalNodeId: () => this.localNodeId,
|
||||
pool: this.worktreePool,
|
||||
usageLimitPauser: this.usageLimitPauser,
|
||||
credentialRotator: this.credentialRotator,
|
||||
stuckTaskDetector: this.stuckTaskDetector,
|
||||
cliAgentRuntime: this.cliAgentRuntime?.bundle,
|
||||
pluginRunner: this.pluginRunner,
|
||||
@@ -1388,6 +1413,7 @@ export class InProcessRuntime
|
||||
reflectionStore: reflectionStoreForService,
|
||||
reflectionService,
|
||||
selfImproveService,
|
||||
credentialRotator: this.credentialRotator,
|
||||
snapshotManager: autoClaimSnapshotManager,
|
||||
onMissed: (agentId, reason) => {
|
||||
runtimeLog.warn(`Agent ${agentId} missed heartbeat: ${reason}`);
|
||||
@@ -2602,6 +2628,7 @@ export class InProcessRuntime
|
||||
*/
|
||||
setUsageLimitPauser(pauser: UsageLimitPauser): void {
|
||||
this.usageLimitPauser = pauser;
|
||||
pauser.setCredentialRotator(this.credentialRotator);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -11,6 +11,7 @@
|
||||
*/
|
||||
|
||||
import type { Task, TaskStore } from "@fusion/core";
|
||||
import type { CredentialInstanceRotator } from "./credential-instance-rotation.js";
|
||||
// FNXC:WorkflowLifecycleColumns 2026-07-30-11:00: `agentType` is an AGENT ROLE, not a column.
|
||||
// The planner lane is named `triage` and keeps that name; only the COLUMN was removed by U11.
|
||||
import { PLANNER_AGENT_ROLE, resolveTaskLifecycleColumns, type WorkflowIr } from "@fusion/core";
|
||||
@@ -80,7 +81,15 @@ export function checkSessionError(session: { state: { errorMessage?: string; err
|
||||
}
|
||||
|
||||
export class UsageLimitPauser {
|
||||
constructor(private store: TaskStore) {}
|
||||
constructor(
|
||||
private store: TaskStore,
|
||||
private readonly options: { credentialRotator?: CredentialInstanceRotator } = {},
|
||||
) {}
|
||||
|
||||
/** Rebinds an externally supplied pauser to the runtime-owned cooldown map. */
|
||||
setCredentialRotator(credentialRotator: CredentialInstanceRotator | undefined): void {
|
||||
(this.options as { credentialRotator?: CredentialInstanceRotator }).credentialRotator = credentialRotator;
|
||||
}
|
||||
|
||||
private normalizeProviderId(provider: string): string {
|
||||
return provider.trim().toLowerCase().replace(/[^a-z0-9._-]+/g, "-").replace(/^-+|-+$/g, "");
|
||||
@@ -96,6 +105,13 @@ export class UsageLimitPauser {
|
||||
if (!providerId) return 0;
|
||||
|
||||
const pausedReason = `provider-rate-limit:${providerId}`;
|
||||
/*
|
||||
FNXC:CredentialInstanceRotation 2026-08-01-06:21:
|
||||
Provider recovery clears the runtime-shared, process-local cooldown hint before
|
||||
resuming existing rate-limit parks. Parking remains the fallback after rotation,
|
||||
and remains the first response for a single configured instance.
|
||||
*/
|
||||
this.options.credentialRotator?.clearCooldowns(providerId);
|
||||
// FNXC:ArchitectureHotPath 2026-07-22-17:20: listTasks() must be explicit about payload shape (architecture-hot-paths contract). Recovery only reads scalar pause fields, so request slim rows to avoid loading heavy log/steps/comments for every task.
|
||||
const tasks = await this.store.listTasks({ slim: true });
|
||||
const recoverableTasks = tasks.filter((task) =>
|
||||
|
||||
Reference in New Issue
Block a user