From 8a5ea3edf772cee69f9b8b96ab43cc7d74332e4c Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 21 Aug 2026 19:31:19 -0700 Subject: [PATCH 1/2] Add tests for boot-restore failure quarantine Covers the accumulation bug: a record that fails restore forever gets no counter, no marker, nothing that ever stops the per-boot warning. Unit tests exercise recordWorkflowDeploymentRestoreFailure, clearWorkflowDeploymentRestoreFailure, and isWorkflowDeploymentRestoreQuarantined directly; the integration test proves the actual behavior a boot cares about: a permanently unrestorable record (address/directory mismatch) stops being attempted after RESTORE_QUARANTINE_THRESHOLD consecutive boots, while a transiently unbuildable inference provider is retried every boot indefinitely and never quarantines. --- .../src/workflow-deployment-record.test.ts | 144 +++++++++++ .../test/workflow-restore-quarantine.test.ts | 236 ++++++++++++++++++ 2 files changed, 380 insertions(+) create mode 100644 apps/sidecar/test/workflow-restore-quarantine.test.ts diff --git a/apps/sidecar/src/workflow-deployment-record.test.ts b/apps/sidecar/src/workflow-deployment-record.test.ts index a7e0deb64..5c7187648 100644 --- a/apps/sidecar/src/workflow-deployment-record.test.ts +++ b/apps/sidecar/src/workflow-deployment-record.test.ts @@ -4,8 +4,12 @@ import os from "node:os"; import path from "node:path"; import { + RESTORE_QUARANTINE_THRESHOLD, + clearWorkflowDeploymentRestoreFailure, + isWorkflowDeploymentRestoreQuarantined, markWorkflowDeploymentRecordParked, readWorkflowDeploymentRecord, + recordWorkflowDeploymentRestoreFailure, scanWorkflowDeploymentRecords, writeWorkflowDeploymentRecord, type WorkflowDeploymentRecord, @@ -143,3 +147,143 @@ describe("scanWorkflowDeploymentRecords reaps pre-cutover records", () => { await fs.access(filePath); }); }); + +describe("recordWorkflowDeploymentRestoreFailure", () => { + test("starts a kind's counter at 1 and persists reason/timestamp", async () => { + const dataDir = await makeDataDir(); + await writeWorkflowDeploymentRecord(dataDir, "dep_1", baseRecord); + + const updated = await recordWorkflowDeploymentRestoreFailure( + dataDir, + "dep_1", + baseRecord, + { kind: "permanent", reason: "address derives a different slug" }, + ); + + expect(updated.restoreFailure).toEqual({ + kind: "permanent", + attempts: 1, + reason: "address derives a different slug", + lastAttemptAt: updated.restoreFailure?.lastAttemptAt ?? "", + }); + const onDisk = await readWorkflowDeploymentRecord(dataDir, "dep_1"); + expect(onDisk?.restoreFailure?.attempts).toBe(1); + }); + + test("increments the counter across repeated failures of the same kind", async () => { + const dataDir = await makeDataDir(); + await writeWorkflowDeploymentRecord(dataDir, "dep_1", baseRecord); + + let record = baseRecord; + for (let i = 0; i < 3; i++) { + record = await recordWorkflowDeploymentRestoreFailure( + dataDir, + "dep_1", + record, + { kind: "permanent", reason: "still malformed" }, + ); + } + + expect(record.restoreFailure?.attempts).toBe(3); + expect(record.restoreFailure?.kind).toBe("permanent"); + }); + + test("a kind change resets the counter rather than adding to the other kind's count", async () => { + const dataDir = await makeDataDir(); + await writeWorkflowDeploymentRecord(dataDir, "dep_1", baseRecord); + + let record = await recordWorkflowDeploymentRestoreFailure( + dataDir, + "dep_1", + baseRecord, + { kind: "permanent", reason: "malformed" }, + ); + record = await recordWorkflowDeploymentRestoreFailure( + dataDir, + "dep_1", + record, + { kind: "permanent", reason: "still malformed" }, + ); + expect(record.restoreFailure?.attempts).toBe(2); + + record = await recordWorkflowDeploymentRestoreFailure( + dataDir, + "dep_1", + record, + { kind: "transient", reason: "provider not registered" }, + ); + + expect(record.restoreFailure?.kind).toBe("transient"); + expect(record.restoreFailure?.attempts).toBe(1); + }); +}); + +describe("clearWorkflowDeploymentRestoreFailure", () => { + test("drops restoreFailure after a successful restore", async () => { + const dataDir = await makeDataDir(); + await writeWorkflowDeploymentRecord(dataDir, "dep_1", baseRecord); + const failed = await recordWorkflowDeploymentRestoreFailure( + dataDir, + "dep_1", + baseRecord, + { kind: "transient", reason: "provider not registered" }, + ); + + await clearWorkflowDeploymentRestoreFailure(dataDir, "dep_1", failed); + + const onDisk = await readWorkflowDeploymentRecord(dataDir, "dep_1"); + expect(onDisk?.restoreFailure).toBeUndefined(); + }); + + test("is a no-op when there is no failure to clear", async () => { + const dataDir = await makeDataDir(); + await writeWorkflowDeploymentRecord(dataDir, "dep_1", baseRecord); + + await clearWorkflowDeploymentRestoreFailure(dataDir, "dep_1", baseRecord); + + const onDisk = await readWorkflowDeploymentRecord(dataDir, "dep_1"); + expect(onDisk?.agentAddress).toBe(baseRecord.agentAddress); + }); +}); + +describe("isWorkflowDeploymentRestoreQuarantined", () => { + test("is false below the threshold and true at or above it, for permanent failures only", () => { + const belowThreshold: WorkflowDeploymentRecord = { + ...baseRecord, + restoreFailure: { + kind: "permanent", + attempts: RESTORE_QUARANTINE_THRESHOLD - 1, + reason: "malformed", + lastAttemptAt: new Date().toISOString(), + }, + }; + const atThreshold: WorkflowDeploymentRecord = { + ...baseRecord, + restoreFailure: { + kind: "permanent", + attempts: RESTORE_QUARANTINE_THRESHOLD, + reason: "malformed", + lastAttemptAt: new Date().toISOString(), + }, + }; + const transientAtSameCount: WorkflowDeploymentRecord = { + ...baseRecord, + restoreFailure: { + kind: "transient", + attempts: RESTORE_QUARANTINE_THRESHOLD + 5, + reason: "provider not registered", + lastAttemptAt: new Date().toISOString(), + }, + }; + + expect(isWorkflowDeploymentRestoreQuarantined(belowThreshold)).toBe(false); + expect(isWorkflowDeploymentRestoreQuarantined(atThreshold)).toBe(true); + expect(isWorkflowDeploymentRestoreQuarantined(baseRecord)).toBe(false); + // A transient failure never quarantines, no matter how high its own + // counter climbs -- it is tracked on a separate counter from the + // permanent one. + expect(isWorkflowDeploymentRestoreQuarantined(transientAtSameCount)).toBe( + false, + ); + }); +}); diff --git a/apps/sidecar/test/workflow-restore-quarantine.test.ts b/apps/sidecar/test/workflow-restore-quarantine.test.ts new file mode 100644 index 000000000..1086126c6 --- /dev/null +++ b/apps/sidecar/test/workflow-restore-quarantine.test.ts @@ -0,0 +1,236 @@ +// Boot-time restore's failure accounting: a record whose failure is +// PERMANENT (deterministic, intrinsic to the record's own persisted +// bytes -- here, an address that derives a different slug than its +// on-disk directory) stops being attempted after +// RESTORE_QUARANTINE_THRESHOLD consecutive boots and collapses into a +// one-line summary; a record whose failure is TRANSIENT (this boot's +// environment -- here, an inference provider the sidecar cannot build) +// is retried every boot forever, however many times it has failed. Real +// child spawning is out of scope here (see workflow-deploy-lifecycle.test.ts); +// this only exercises the restore loop's classification and skip logic. + +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, expect, test } from "bun:test"; +import { + createEd25519Crypto, + generateKeyPair, + signEd25519, + verifySSHSignature, +} from "@intx/crypto"; +import { + createAgentKeyStore, + createAgentRepoStore as createSidecarSideRepoStore, + createSessionManager, +} from "@intx/hub-agent"; +import { createAgentRepoStore } from "@intx/hub-sessions"; +import { createInMemoryTransport } from "@intx/mail-memory"; +import type { InferenceSource } from "@intx/types/runtime"; +import type { SubprocessSpawner } from "@intx/workflow-host"; +import { defineWorkflow, step, type WorkflowDefinition } from "@intx/workflow"; +import { buildSingleStepAgentDefinition } from "@intx/workflow-deploy"; +import { + createSidecarDeployRouter, + deriveDeploymentId, + type SidecarDeployRouter, +} from "../src/workflow-host-wiring"; +import { + RESTORE_QUARANTINE_THRESHOLD, + readWorkflowDeploymentRecord, + writeWorkflowDeploymentRecord, + type WorkflowDeploymentRecord, +} from "../src/workflow-deployment-record"; + +const tempDirs: string[] = []; + +afterEach(async () => { + await Promise.all( + tempDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true })), + ); +}); + +async function makeDataDir(): Promise { + const dir = await mkdtemp(path.join(tmpdir(), "sidecar-restore-")); + tempDirs.push(dir); + return dir; +} + +const closureDefinitions = new Map(); + +async function makeRouter( + dataDir: string, + opts: { assertSourceBuildable: (source: InferenceSource) => void }, +): Promise { + const signingKey = await generateKeyPair(); + const substrate = createAgentRepoStore({ dataDir, signingKey }); + const repoStore = createSidecarSideRepoStore({ dataDir }); + const sessions = createSessionManager({ repoStore }); + const keyStore = createAgentKeyStore({ + dataDir, + generateKeyPair, + signEd25519, + verifySSHSig: verifySSHSignature, + }); + const recordingSpawner: SubprocessSpawner = () => { + throw new Error("test spawner refuses to launch a real child"); + }; + return createSidecarDeployRouter({ + sessions, + keyStore, + transport: createInMemoryTransport(), + repoStore: substrate.repoStore, + signingKeySeed: signingKey.privateKey, + createAgentCrypto: createEd25519Crypto, + assertSourceBuildable: opts.assertSourceBuildable, + registerDeployment: () => undefined, + unregisterDeployment: () => undefined, + multistepSubstrateEnv: { SIDECAR_DATA_DIR: dataDir }, + materializeDeploymentClosure: ({ deploymentId }) => { + const definition = closureDefinitions.get(deploymentId); + if (definition === undefined) { + throw new Error( + `test closure materializer: no definition registered for ${deploymentId}`, + ); + } + return Promise.resolve({ + definition, + packageDir: path.join(dataDir, "closure-package", deploymentId), + deployDir: path.join(dataDir, "closure-deploy", deploymentId), + }); + }, + multistepSubprocessSpawner: recordingSpawner, + multistepBinaryPath: path.join(dataDir, "workflow-child-sentinel"), + }); +} + +function stageClosureDefinition(deploymentId: string): void { + closureDefinitions.set( + deploymentId, + defineWorkflow({ + id: "definition-1", + trigger: { type: "mail", to: "definition-1@example.com" }, + steps: { + "step-1": step({ + agent: buildSingleStepAgentDefinition({ + id: "step-1", + systemPrompt: "", + inferencePreferences: [], + toolFactories: [], + }), + triggers: "unbounded", + }), + }, + }), + ); +} + +function makeSource(provider: string): InferenceSource { + return { + id: `source-${provider}`, + provider, + baseURL: "https://inference.example.com", + apiKey: "key", + model: "model-1", + }; +} + +const SOURCE_REF: WorkflowDeploymentRecord["sourceRef"] = { + source: { kind: "registry", registry: "npm" }, + closure: { schemaVersion: "1", topLevel: [], entries: [] }, +}; + +test("a permanently unrestorable record is quarantined after RESTORE_QUARANTINE_THRESHOLD boots and stops being attempted", async () => { + const dataDir = await makeDataDir(); + const router = await makeRouter(dataDir, { + assertSourceBuildable: () => undefined, + }); + + // A record filed under a directory that does not match what its own + // agentAddress derives to -- corrupt/misplaced, and deterministically so: + // no future boot ever makes `deriveDeploymentId` agree with the directory. + const mismatchedDeploymentId = "dep_mismatched-directory"; + const record: WorkflowDeploymentRecord = { + version: 1, + agentAddress: "run_permanent-case@example.com", + definitionId: "def_1", + sources: {}, + approvedWireHash: "d".repeat(64), + sourceRef: SOURCE_REF, + }; + await writeWorkflowDeploymentRecord(dataDir, mismatchedDeploymentId, record); + + for (let boot = 1; boot <= RESTORE_QUARANTINE_THRESHOLD; boot++) { + await router.restoreWorkflowDeployments(); + const onDisk = await readWorkflowDeploymentRecord( + dataDir, + mismatchedDeploymentId, + ); + expect(onDisk?.restoreFailure?.kind).toBe("permanent"); + expect(onDisk?.restoreFailure?.attempts).toBe(boot); + } + + const quarantined = await readWorkflowDeploymentRecord( + dataDir, + mismatchedDeploymentId, + ); + expect(quarantined?.restoreFailure?.attempts).toBe( + RESTORE_QUARANTINE_THRESHOLD, + ); + + // One more boot: quarantine means the attempt count does NOT move -- + // the record is skipped entirely rather than attempted and re-failed. + await router.restoreWorkflowDeployments(); + const afterQuarantineBoot = await readWorkflowDeploymentRecord( + dataDir, + mismatchedDeploymentId, + ); + expect(afterQuarantineBoot?.restoreFailure?.attempts).toBe( + RESTORE_QUARANTINE_THRESHOLD, + ); + + // The record itself is never deleted -- an operator can still undeploy + // the address to reclaim it. + expect(afterQuarantineBoot?.agentAddress).toBe(record.agentAddress); +}); + +test("a transiently unbuildable provider is retried every boot and never quarantines", async () => { + const dataDir = await makeDataDir(); + let assertCalls = 0; + const router = await makeRouter(dataDir, { + assertSourceBuildable: (source) => { + assertCalls += 1; + if (source.provider === "unbuildable") { + throw new Error( + `Source provider "${source.provider}" is not registered`, + ); + } + }, + }); + + const agentAddress = "run_transient-case@example.com"; + const deploymentId = deriveDeploymentId(agentAddress); + stageClosureDefinition(deploymentId); + const record: WorkflowDeploymentRecord = { + version: 1, + agentAddress, + definitionId: "def_1", + sources: { "step-1": [makeSource("unbuildable")] }, + approvedWireHash: "d".repeat(64), + sourceRef: SOURCE_REF, + }; + await writeWorkflowDeploymentRecord(dataDir, deploymentId, record); + + const bootCount = RESTORE_QUARANTINE_THRESHOLD + 2; + for (let boot = 1; boot <= bootCount; boot++) { + await router.restoreWorkflowDeployments(); + } + + // Every boot actually reached the source-admission gate -- nothing was + // ever skipped, unlike the permanent case above. + expect(assertCalls).toBe(bootCount); + + const onDisk = await readWorkflowDeploymentRecord(dataDir, deploymentId); + expect(onDisk?.restoreFailure?.kind).toBe("transient"); + expect(onDisk?.restoreFailure?.attempts).toBe(bootCount); +}); From ade4666e126bb0d839a69f4b20775847d87d0d2b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 21 Aug 2026 19:31:32 -0700 Subject: [PATCH 2/2] Quarantine permanently unrestorable workflow deployment records The boot-time restore loop replays every persisted deployment record serially before hubLink.connect(), so a pile of dead records gates the hub handshake at every boot. The root cause: a record that fails to restore was never marked, so a permanently broken one warned on every boot forever with nothing to stop it. workflow-deployment-record.ts now tracks a restoreFailure marker (kind, attempts, reason, lastAttemptAt) on the record itself, plus a WorkflowRestoreFailure error class the restore path throws to say which kind a failure is. Only two gates are deterministic given solely the record's own persisted bytes -- an address that derives a different slug than its directory, and a closure-derived definition that fails structural validation -- so only those throw "permanent". Everything else (an unbuildable inference provider, a closure-store miss, a spawn-core failure) depends on this boot's environment, not the record's content, so it is treated as "transient" by default and retried every boot for as long as the record exists; it is deliberately never on the same counter as a permanent failure, so it can never trip the quarantine. After RESTORE_QUARANTINE_THRESHOLD (3) consecutive permanent failures, the boot loop stops attempting the record and folds it into one summary line instead of a per-record warning. The record itself is never deleted -- an operator reclaims it by undeploying the address -- extending the existing legacy-address pruning rather than replacing it. --- .../sidecar/src/workflow-deployment-record.ts | 118 +++++++++++++ .../sidecar/src/workflow-host-wiring/index.ts | 156 ++++++++++++++---- 2 files changed, 240 insertions(+), 34 deletions(-) diff --git a/apps/sidecar/src/workflow-deployment-record.ts b/apps/sidecar/src/workflow-deployment-record.ts index 50f3b65b5..b2e030008 100644 --- a/apps/sidecar/src/workflow-deployment-record.ts +++ b/apps/sidecar/src/workflow-deployment-record.ts @@ -67,9 +67,62 @@ export const WorkflowDeploymentRecord = type({ // and the boot scan both treat an absent marker as "live", so an old // on-disk record keeps loading and restoring exactly as before. "parkedAt?": "string > 0", + // Populated only after a failed boot-time restore attempt; absent for a + // record that has never failed restore, and dropped again the moment a + // restore succeeds (`clearWorkflowDeploymentRestoreFailure`). `kind` + // mirrors `WorkflowRestoreFailure.kind`: "permanent" counts toward + // `RESTORE_QUARANTINE_THRESHOLD`, "transient" is recorded for visibility + // only and never quarantines. `attempts` counts CONSECUTIVE failures of + // the SAME kind -- a kind change resets it to 1, so a transient failure + // streak can never inflate the permanent counter, or vice versa. + "restoreFailure?": { + kind: "'permanent' | 'transient'", + attempts: "number > 0", + reason: "string > 0", + lastAttemptAt: "string > 0", + }, }); export type WorkflowDeploymentRecord = typeof WorkflowDeploymentRecord.infer; +export type RestoreFailureKind = NonNullable< + WorkflowDeploymentRecord["restoreFailure"] +>["kind"]; + +/** + * Thrown from inside the restore path to mark WHY a specific attempt + * failed. `kind` distinguishes a failure intrinsic to the record's own + * persisted bytes -- deterministic, will fail identically on every future + * boot too (a corrupt/misplaced record, a closure-derived definition that + * fails structural validation) -- from one that depends on this boot's + * environment and may clear on its own (an inference provider the sidecar + * cannot currently build). Only "permanent" failures count toward + * `RESTORE_QUARANTINE_THRESHOLD`; an ordinary (unwrapped) `Error` thrown + * anywhere else in the restore path is treated as "transient" by the boot + * loop -- the safe default, since misclassifying a recoverable failure as + * permanent stops future restore attempts for it. + */ +export class WorkflowRestoreFailure extends Error { + readonly kind: RestoreFailureKind; + constructor(kind: RestoreFailureKind, message: string) { + super(message); + this.name = "WorkflowRestoreFailure"; + this.kind = kind; + } +} + +/** + * Consecutive PERMANENT restore failures before a record stops being + * attempted and collapses into the boot scan's one-line summary instead + * of an individual per-record warning every boot. 3: permanent failures + * are deterministic (derived only from the record's own immutable + * closure pin and address, never from runtime environment), so they do + * not actually need repeated confirmation to be believed -- but a small + * buffer costs nothing and guards against a misclassification, while + * still bounding the "warn every boot forever" failure mode this + * quarantine exists to fix to a handful of boots. + */ +export const RESTORE_QUARANTINE_THRESHOLD = 3; + function recordPath(dataDir: string, deploymentId: string): string { return pathJoin(dataDir, "workflow-runs", deploymentId, RECORD_FILENAME); } @@ -169,6 +222,71 @@ export async function markWorkflowDeploymentRecordParked( }); } +/** + * Record a failed boot-time restore attempt on an existing record: bump + * `attempts` (reset to 1 when `failure.kind` differs from the previously + * recorded kind, so a transient streak can never inflate the permanent + * counter or vice versa), stamp `reason`/`lastAttemptAt`, and persist. + * Returns the updated record so the caller can check `isWorkflowDeploymentRestoreQuarantined` + * without a re-read. Takes the caller's in-memory `record` rather than + * re-reading it, matching `markWorkflowDeploymentRecordParked`'s sibling + * shape but avoiding a redundant read on the hot boot-restore path. + */ +export async function recordWorkflowDeploymentRestoreFailure( + dataDir: string, + deploymentId: string, + record: WorkflowDeploymentRecord, + failure: { kind: RestoreFailureKind; reason: string }, +): Promise { + const previous = record.restoreFailure; + const attempts = + previous !== undefined && previous.kind === failure.kind + ? previous.attempts + 1 + : 1; + const updated: WorkflowDeploymentRecord = { + ...record, + restoreFailure: { + kind: failure.kind, + attempts, + reason: failure.reason, + lastAttemptAt: new Date().toISOString(), + }, + }; + await writeWorkflowDeploymentRecord(dataDir, deploymentId, updated); + return updated; +} + +/** + * Clear a record's failure tracking after a restore succeeds. A no-op + * (no write) when there is nothing to clear, so a caller can call this + * unconditionally after every successful restore. + */ +export async function clearWorkflowDeploymentRestoreFailure( + dataDir: string, + deploymentId: string, + record: WorkflowDeploymentRecord, +): Promise { + if (record.restoreFailure === undefined) return; + const { restoreFailure, ...rest } = record; + await writeWorkflowDeploymentRecord(dataDir, deploymentId, rest); +} + +/** + * Whether a record's next boot-time restore attempt should be skipped + * entirely: a "permanent"-kind failure that has reached + * `RESTORE_QUARANTINE_THRESHOLD` consecutive attempts. A "transient"-kind + * streak never quarantines, however high its own counter climbs -- see + * `WorkflowRestoreFailure`. + */ +export function isWorkflowDeploymentRestoreQuarantined( + record: WorkflowDeploymentRecord, +): boolean { + return ( + record.restoreFailure?.kind === "permanent" && + record.restoreFailure.attempts >= RESTORE_QUARANTINE_THRESHOLD + ); +} + /** * Remove a deployment record. Called on undeploy and on a soft-failed * deploy so a torn-down or never-completed deployment is not restored on diff --git a/apps/sidecar/src/workflow-host-wiring/index.ts b/apps/sidecar/src/workflow-host-wiring/index.ts index 8fff6fcc5..515181ccf 100644 --- a/apps/sidecar/src/workflow-host-wiring/index.ts +++ b/apps/sidecar/src/workflow-host-wiring/index.ts @@ -51,8 +51,12 @@ import type { MultistepSourcesRouter, } from "../workflow-run-pack-client"; import { + WorkflowRestoreFailure, + clearWorkflowDeploymentRestoreFailure, deleteWorkflowDeploymentRecord, + isWorkflowDeploymentRestoreQuarantined, markWorkflowDeploymentRecordParked, + recordWorkflowDeploymentRestoreFailure, scanWorkflowDeploymentRecords, writeWorkflowDeploymentRecord, type WorkflowDeploymentRecord, @@ -174,8 +178,17 @@ export interface SidecarDeployRouter extends DeployRouter { * substrate. Runs once at boot, before `hubLink.connect()`, so a single-step * head's mailbox/transport registration is live before the hub routes to it. * Soft-fails per deployment: a record that cannot be restored (unbuildable - * provider, corrupt `workflow.json`, spawn failure) is logged and left on - * disk for a later boot to retry -- it is never deleted here. + * provider, corrupt `deployment.json`, spawn failure) is logged and left on + * disk for a later boot to retry -- it is never deleted here. A failure + * that is deterministic given only the record's own bytes (a + * `WorkflowRestoreFailure("permanent", ...)`, e.g. a corrupt/misplaced + * record or a closure-derived definition that fails validation) is + * tracked on the record and, after `RESTORE_QUARANTINE_THRESHOLD` + * consecutive such failures, quarantined: later boots skip it entirely + * rather than re-attempting and re-warning. A failure that depends on + * this boot's environment (an unbuildable inference provider, and + * anything else not explicitly classified permanent) never quarantines + * and is retried every boot for as long as the record exists. */ restoreWorkflowDeployments(): Promise; /** @@ -1378,10 +1391,18 @@ export function createSidecarDeployRouter(deps: { * shared core of the boot-time restore loop and the CL-5477 idle-reap * wake path. Applies exactly the gates the live deploy path applies * (address integrity, wire arktype, tool-metadata-equivalent structural - * projection, source admission). Soft-skips (corrupt record, failed - * validation, unbuildable provider) log and return without spawning, - * matching the boot scan's existing posture; the record is never deleted - * here. Throws only where the spawn core itself throws. + * projection, source admission). Every failure throws so the caller's + * failure-accounting (`recordWorkflowDeploymentRestoreFailure`) sees it; + * the record is never deleted here. A failure that is intrinsic to the + * record's own persisted bytes -- deterministic, will recur identically + * on every future boot -- throws a `WorkflowRestoreFailure("permanent", + * ...)`: address-derivation mismatch and closure-derived-definition + * validation are the only two such gates below. Every other failure + * (an unbuildable inference provider, a closure-materialization miss, a + * spawn-core throw) is a plain `Error`, which the caller treats as + * "transient" -- it depends on this boot's environment, not the + * record's content, and may clear on a later boot with no change to the + * record at all. */ async function restoreDeploymentFromRecord( dataDir: string, @@ -1389,12 +1410,15 @@ export function createSidecarDeployRouter(deps: { record: WorkflowDeploymentRecord, ): Promise { // Integrity: the stored address must re-derive to its own directory - // name. A mismatch means a corrupt or misplaced record; skip it - // rather than restore a deployment under the wrong slug. + // name. A mismatch means a corrupt or misplaced record -- permanent, + // since re-deriving the same address on a later boot yields the same + // mismatch every time. const derived = deriveDeploymentId(record.agentAddress); if (derived !== deploymentId) { - logger.warn`skipping workflow deployment restore: ${record.agentAddress} derives slug ${derived}, not its directory ${deploymentId}`; - return; + throw new WorkflowRestoreFailure( + "permanent", + `${record.agentAddress} derives slug ${derived}, not its directory ${deploymentId}`, + ); } // A record whose address the platform's own parser rejects is @@ -1428,25 +1452,47 @@ export function createSidecarDeployRouter(deps: { projectLiveToInert(applied.definition), ); if (validatedDefinition instanceof type.errors) { - logger.warn`skipping workflow deployment restore for ${record.agentAddress}: workflow definition loaded from the frozen closure failed projection validation: ${validatedDefinition.summary}`; - return; + // Permanent: the closure is an immutable pin, so re-evaluating it on + // a later boot yields the identical (broken) projection every time. + throw new WorkflowRestoreFailure( + "permanent", + `workflow definition loaded from the frozen closure failed projection validation for ${record.agentAddress}: ${validatedDefinition.summary}`, + ); } const definition: WorkflowProjectionDefinition = validatedDefinition; // Structural invariants the wire arktype does not cover. The closure eval // skips the deploy frame's coverage narrow, so this is where the - // definition-vs-sources coverage is checked. - validateWorkflowProjection({ definition, sources: record.sources }); + // definition-vs-sources coverage is checked. Also permanent, for the + // same reason as the projection check above: both the definition and + // `record.sources` are immutable once persisted. + try { + validateWorkflowProjection({ definition, sources: record.sources }); + } catch (cause) { + const reason = cause instanceof Error ? cause.message : String(cause); + throw new WorkflowRestoreFailure("permanent", reason); + } // Re-run the source-admission gate: refuse to restore a deployment // whose pinned provider this sidecar can no longer build. Every // source in a step's failover chain must be buildable, so this - // iterates the whole list. The record is KEPT (not deleted) so a - // later boot with the provider restored retries it. + // iterates the whole list. Transient, not permanent: the buildable- + // provider set is this boot's adapter registry, not anything the + // record itself carries, so a later boot with the provider restored + // retries and can succeed with the SAME record unchanged. The record + // is KEPT (not deleted) either way. for (const stepId of definition.stepOrder) { const chain = record.sources[stepId]; if (chain !== undefined) { - for (const source of chain) deps.assertSourceBuildable(source); + for (const source of chain) { + try { + deps.assertSourceBuildable(source); + } catch (cause) { + const reason = + cause instanceof Error ? cause.message : String(cause); + throw new WorkflowRestoreFailure("transient", reason); + } + } } } @@ -1472,10 +1518,12 @@ export function createSidecarDeployRouter(deps: { // the spawn, release on failure. Unlike deploy's soft-fail, restore // does NOT delete the record and does NOT re-materialize the step grants // or the onTrigger body sources -- both are already on disk from the - // original deploy. A failed restore just warns and - // leaves the record for the next boot; there is deliberately no GC - // of a permanently-unrestorable record here (an operator reclaims it - // by undeploying the address). + // original deploy. A failed restore's caller (the boot loop, below) + // tracks the failure on the record and always leaves the record itself + // in place for the next boot; there is deliberately no GC of the record + // FILE here (an operator reclaims it by undeploying the address) -- + // only of the ATTENTION a permanently-unrestorable one demands, via + // the quarantine threshold below. // // Release only a slug THIS pass newly claimed: if the address is // already live (its slug still held by the running deployment), the @@ -1628,34 +1676,74 @@ export function createSidecarDeployRouter(deps: { // Report parked-vs-live counts from the `parkedAt` marker // (`markWorkflowDeploymentRecordParked`, written on the hibernate // teardown) before restoring anything. This is REPORT-ONLY: every - // record below still restores LIVE regardless of this count, exactly - // as before this marker existed. Making the boot scan actually skip - // (or defer) the parked ones is CL-6282 -- a separate change, once its - // design pass lands -- and needs a signal this scan cannot see on its - // own before that cutover is safe: whether the hub still wants a - // parked deployment running is a hub-side decision, not something a - // sidecar with no hub connection yet can determine at boot. + // non-quarantined record below still restores LIVE regardless of + // this count, exactly as before this marker existed. Making the + // boot scan actually skip (or defer) the parked ones is CL-6282 -- a + // separate change, once its design pass lands -- and needs a signal + // this scan cannot see on its own before that cutover is safe: + // whether the hub still wants a parked deployment running is a + // hub-side decision, not something a sidecar with no hub connection + // yet can determine at boot. const parkedCount = scanned.filter( ({ record }) => record.parkedAt !== undefined, ).length; - logger.info`Boot scan found ${scanned.length} deployment record(s): ${parkedCount} parked, ${scanned.length - parkedCount} live; restoring all of them (parked-aware boot restore is CL-6282)`; + const alreadyQuarantinedCount = scanned.filter(({ record }) => + isWorkflowDeploymentRestoreQuarantined(record), + ).length; + logger.info`Boot scan found ${scanned.length} deployment record(s): ${parkedCount} parked, ${alreadyQuarantinedCount} quarantined, ${scanned.length - parkedCount - alreadyQuarantinedCount} to restore (parked-aware boot restore is CL-6282)`; // Restore serially, not in parallel: deterministic boot-log ordering, // one isolable warning per failed record, and no concurrent // child-spawn / transport-register storm. Restore runs before // `hubLink.connect()`, so there are no concurrent deploys to contend // with. Each record's failure is caught so one bad deployment cannot - // strand the rest. Every persisted record restores LIVE: a deployment - // that was previously torn down as a state-preserving hibernate is - // relaunched by the hub's own reap-and-relaunch flow, not by this - // boot scan guessing at staleness. + // strand the rest. Every non-quarantined persisted record restores + // LIVE: a deployment that was previously torn down as a + // state-preserving hibernate is relaunched by the hub's own + // reap-and-relaunch flow, not by this boot scan guessing at + // staleness. + let skippedQuarantinedCount = 0; for (const { deploymentId, record } of scanned) { + // Skip WITHOUT attempting: a permanently unrestorable record that + // has already crossed RESTORE_QUARANTINE_THRESHOLD gets neither a + // spawn attempt nor a per-record warning this boot -- both are + // pointless for a deterministic failure that has already been + // reported that many times. It still counts toward the one + // summary line below, and the record itself is untouched (an + // operator reclaims it by undeploying the address). + if (isWorkflowDeploymentRestoreQuarantined(record)) { + skippedQuarantinedCount += 1; + continue; + } try { await restoreDeploymentFromRecord(dataDir, deploymentId, record); + if (record.restoreFailure !== undefined) { + await clearWorkflowDeploymentRestoreFailure( + dataDir, + deploymentId, + record, + ); + } } catch (cause) { const reason = cause instanceof Error ? cause.message : String(cause); - logger.warn`Failed to restore workflow deployment ${deploymentId}: ${reason}`; + const kind = + cause instanceof WorkflowRestoreFailure ? cause.kind : "transient"; + const updated = await recordWorkflowDeploymentRestoreFailure( + dataDir, + deploymentId, + record, + { kind, reason }, + ); + if (isWorkflowDeploymentRestoreQuarantined(updated)) { + const attempts = updated.restoreFailure?.attempts ?? 0; + logger.warn`Workflow deployment ${deploymentId} failed to restore ${attempts} consecutive times and is now quarantined -- it will not be retried again until the address is undeployed. Last failure: ${reason}`; + } else { + logger.warn`Failed to restore workflow deployment ${deploymentId}: ${reason}`; + } } } + if (skippedQuarantinedCount > 0) { + logger.warn`Skipped ${skippedQuarantinedCount} quarantined workflow deployment record(s) (permanent restore failures, already reported); undeploy an address to clear its record`; + } }, activeAddresses(): string[] { // `activeSupervisors` holds exactly the deployments with a live