[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8,"_9":10,"_11":12},"actionData","errors","root",{"_229":-5,"_232":233,"_234":235},"routes/shell",{"_42":43,"_228":-5,"_229":-5,"_230":231},"routes/repo",{"_33":-7,"_34":35,"_36":37,"_38":39,"_40":41,"_42":43,"_44":31,"_45":31,"_46":31,"_47":31,"_48":31,"_49":31,"_50":51,"_52":51,"_53":31,"_54":-7,"_55":-7,"_56":57,"_58":59,"_60":61,"_62":-5,"_63":64,"_65":66,"_67":68,"_69":70},"routes/repo.blob",{"_13":14,"_15":16,"_17":18,"_19":20,"_21":14,"_22":23,"_24":-5,"_25":26,"_27":28,"_28":29,"_30":31,"_32":14},"ref","90a2f99e37773b779f75498b1a4e3dfe8ca1ff64","path","apps/server/src/jobs/runner.ts","oid","3d4c8291eaf964d34812abce8fbcf92275e82286","size",28231,"commit","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","import { StorageError } from \"../storage/errors\";\nimport { JobExecutionError } from \"./registry\";\n\nimport type { BackgroundJob, BackgroundJobCursor, JsonValue, Lease } from \"../storage/model\";\nimport type { StorageAdapter } from \"../storage/port\";\nimport type { JobDefinition, JobExecutionCredential, JobRegistry } from \"./registry\";\nimport type { JobService, JobView } from \"./service\";\n\nexport type ResolvedJobCredential = {\n\tcredential: JobExecutionCredential;\n\townerKey: string | undefined;\n\tbinding: JsonValue | undefined;\n\tsignal?: AbortSignal;\n\tactive: () => Promise;\n\trelease: () => void;\n};\n\nexport type RunnerScheduler = {\n\tnow: () => Date;\n\tafter: (delayMs: number, action: () => void) => () => void;\n};\n\nexport type JobRunnerOptions = {\n\tstorage: StorageAdapter;\n\tservice: JobService;\n\tregistry: JobRegistry;\n\tlease: () => Lease;\n\tresolveActivePlanner: (\n\t\tjob: Readonly,\n\t\tsignal: AbortSignal,\n\t) => Promise;\n\tenabled: boolean;\n\tglobalConcurrency: number;\n\townerConcurrency: number;\n\tpollMs?: number;\n\tclaimTtlMs?: number;\n\theartbeatMs?: number;\n\tretryBaseMs?: number;\n\tretryMaxMs?: number;\n\tshutdownGraceMs?: number;\n\tid?: () => string;\n\tscheduler?: RunnerScheduler;\n\tchanged?: (channelId: string) => void | Promise;\n\tattemptFailed?: (job: Readonly, err: unknown) => void | Promise;\n\tfatal?: (err: unknown) => void;\n};\n\ntype Attempt = {\n\tjob: BackgroundJob;\n\tcontroller: AbortController;\n\taccepting: boolean;\n\tmutation: Promise;\n\theartbeat?: () => void;\n\tcredential?: ResolvedJobCredential;\n\treleaseOwner?: () => void;\n\texecution?: Promise;\n\trecovery?: Promise;\n\tdone: Promise;\n};\n\nfunction positive(value: number, field: string, maximum = Number.MAX_SAFE_INTEGER): void {\n\tif (!Number.isSafeInteger(value) || value < 1 || value > maximum) {\n\t\tthrow new Error(`Background job runner ${field} must be an integer between 1 and ${maximum}.`);\n\t}\n}\n\nfunction defaultScheduler(): RunnerScheduler {\n\treturn {\n\t\tnow: () => new Date(),\n\t\tafter(delayMs, action) {\n\t\t\tlet timer = setTimeout(action, delayMs);\n\t\t\treturn () => clearTimeout(timer);\n\t\t},\n\t};\n}\n\nfunction retryDelay(attempt: number, base: number, maximum: number): number {\n\treturn Math.min(maximum, base * 2 ** Math.min(Math.max(0, attempt - 1), 18));\n}\n\nfunction expected(err: unknown): boolean {\n\treturn err instanceof StorageError && (err.failure === \"conflict\" || err.failure === \"missing\");\n}\n\nfunction noop(): void {}\n\n/** Claims and executes registered work without sharing Planner sessions or transcript state. */\nexport class JobRunner {\n\t#options:\n\t\t& Required<\n\t\t\tPick<\n\t\t\t\tJobRunnerOptions,\n\t\t\t\t| \"claimTtlMs\"\n\t\t\t\t| \"globalConcurrency\"\n\t\t\t\t| \"heartbeatMs\"\n\t\t\t\t| \"ownerConcurrency\"\n\t\t\t\t| \"pollMs\"\n\t\t\t\t| \"retryBaseMs\"\n\t\t\t\t| \"retryMaxMs\"\n\t\t\t\t| \"shutdownGraceMs\"\n\t\t\t>\n\t\t>\n\t\t& JobRunnerOptions;\n\t#scheduler: RunnerScheduler;\n\t#claimOwner: string;\n\t#active = new Map();\n\t#attempts = new Set();\n\t#owners = new Map();\n\t#started = false;\n\t#stopped = false;\n\t#dirty = false;\n\t#pump?: Promise;\n\t#poll?: () => void;\n\t#ownerRecovery = new Map void>();\n\t#blockedChannels = new Set();\n\t#shutdown?: Promise;\n\t#pollFailures = 0;\n\n\tconstructor(options: JobRunnerOptions) {\n\t\tlet values = {\n\t\t\tpollMs: options.pollMs ?? 1_000,\n\t\t\tclaimTtlMs: options.claimTtlMs ?? 30_000,\n\t\t\theartbeatMs: options.heartbeatMs ?? 10_000,\n\t\t\tretryBaseMs: options.retryBaseMs ?? 1_000,\n\t\t\tretryMaxMs: options.retryMaxMs ?? 300_000,\n\t\t\tshutdownGraceMs: options.shutdownGraceMs ?? 10_000,\n\t\t\tglobalConcurrency: options.globalConcurrency,\n\t\t\townerConcurrency: options.ownerConcurrency,\n\t\t};\n\t\tpositive(values.globalConcurrency, \"global concurrency\", 100);\n\t\tpositive(values.ownerConcurrency, \"owner concurrency\", values.globalConcurrency);\n\t\tpositive(values.pollMs, \"poll interval\");\n\t\tpositive(values.claimTtlMs, \"claim ttl\");\n\t\tpositive(values.heartbeatMs, \"heartbeat interval\");\n\t\tpositive(values.retryBaseMs, \"retry base\");\n\t\tpositive(values.retryMaxMs, \"retry maximum\");\n\t\tpositive(values.shutdownGraceMs, \"shutdown grace\");\n\t\tif (values.heartbeatMs * 2 >= values.claimTtlMs) {\n\t\t\tthrow new Error(\"Background job runner heartbeat must be less than half the claim ttl.\");\n\t\t}\n\t\tif (values.shutdownGraceMs + values.heartbeatMs >= values.claimTtlMs) {\n\t\t\tthrow new Error(\n\t\t\t\t\"Background job runner shutdown grace plus heartbeat must be less than the claim ttl.\",\n\t\t\t);\n\t\t}\n\t\tthis.#options = { ...options, ...values };\n\t\tthis.#scheduler = options.scheduler ?? defaultScheduler();\n\t\tthis.#claimOwner = `jobs:${options.id?.() ?? crypto.randomUUID()}`;\n\t}\n\n\tstart(): void {\n\t\tif (this.#started || this.#stopped || !this.#options.enabled) return;\n\t\tthis.#started = true;\n\t\tthis.wake();\n\t}\n\n\twake(): void {\n\t\tif (!this.#started || this.#stopped) return;\n\t\tthis.#dirty = true;\n\t\tthis.#poll?.();\n\t\tthis.#poll = undefined;\n\t\tif (this.#pump) return;\n\t\tqueueMicrotask(() => {\n\t\t\tif (!this.#pump && this.#dirty && !this.#stopped) void this.#drain();\n\t\t});\n\t}\n\n\tnotify(job: JobView): void {\n\t\tif (job.state === \"pending\") this.wake();\n\t\tfor (let attempt of this.#active.values()) {\n\t\t\tlet same = attempt.job.id === job.id;\n\t\t\tlet superseded = attempt.job.channelId === job.channelId\n\t\t\t\t&& attempt.job.targetKey === job.targetKey\n\t\t\t\t&& attempt.job.targetGeneration < job.targetGeneration;\n\t\t\tif (same && job.state !== \"running\" || superseded) this.#detach(attempt, \"job-changed\");\n\t\t}\n\t}\n\n\tasync ownerRevoked(ownerSessionId: string): Promise {\n\t\tawait Promise.all(\n\t\t\t[...this.#active.values()].filter(attempt =>\n\t\t\t\tattempt.credential?.credential.kind === \"active-planner\"\n\t\t\t\t&& attempt.credential.credential.ownerSessionId === ownerSessionId\n\t\t\t).map(attempt => this.#pause(attempt, \"owner-unavailable\")),\n\t\t);\n\t}\n\n\tasync credentialsWillRotate(ownerSessionId: string, revision: number): Promise {\n\t\tawait Promise.all(\n\t\t\t[...this.#active.values()].filter(attempt =>\n\t\t\t\tattempt.credential?.credential.kind === \"active-planner\"\n\t\t\t\t&& attempt.credential.credential.ownerSessionId === ownerSessionId\n\t\t\t\t&& attempt.credential.credential.credentialRevision === revision\n\t\t\t).map(attempt => this.#retry(attempt, \"credential-rotated\", false)),\n\t\t);\n\t}\n\n\tasync channelOwnerReset(channelId: string): Promise {\n\t\tawait Promise.all(\n\t\t\t[...this.#active.values()].filter(attempt =>\n\t\t\t\tattempt.job.channelId === channelId\n\t\t\t\t&& attempt.credential?.credential.kind === \"active-planner\"\n\t\t\t).map(attempt => this.#pause(attempt, \"owner-unavailable\")),\n\t\t);\n\t}\n\n\t/** Fence and cancel every job owned by a document before that document is deleted. */\n\tasync cancelChannel(channelId: string): Promise {\n\t\tthis.#blockedChannels.add(channelId);\n\t\tthis.#ownerRecovery.get(channelId)?.();\n\t\tthis.#ownerRecovery.delete(channelId);\n\t\tlet attempts = [...this.#attempts].filter(attempt => attempt.job.channelId === channelId);\n\t\tfor (let attempt of attempts) this.#detach(attempt, \"document-deleted\");\n\n\t\tlet cursor: BackgroundJobCursor | undefined;\n\t\tdo {\n\t\t\tlet page = await this.#options.service.list(channelId, 100, cursor);\n\t\t\tif (!page) break;\n\t\t\tfor (let job of page.jobs) {\n\t\t\t\tif (job.state !== \"pending\" && job.state !== \"paused\" && job.state !== \"running\") continue;\n\t\t\t\ttry {\n\t\t\t\t\tawait this.#options.service.cancel({ channelId, jobId: job.id }, true);\n\t\t\t\t} catch (err) {\n\t\t\t\t\tif (!expected(err)) throw err;\n\t\t\t\t}\n\t\t\t}\n\t\t\tcursor = page.next;\n\t\t} while (cursor);\n\t\tawait this.#withinGrace(Promise.allSettled(attempts.map(attempt => attempt.done)));\n\t}\n\n\t/** Used only when deletion failed before the channel row was removed. */\n\tunblockChannel(channelId: string): void {\n\t\tthis.#blockedChannels.delete(channelId);\n\t\tthis.wake();\n\t}\n\n\tasync ownerAvailable(channelId: string): Promise {\n\t\tthis.#ownerRecovery.get(channelId)?.();\n\t\tthis.#ownerRecovery.delete(channelId);\n\t\ttry {\n\t\t\tlet cursor: BackgroundJobCursor | undefined;\n\t\t\tdo {\n\t\t\t\tlet page = await this.#options.service.list(channelId, 100, cursor);\n\t\t\t\tif (!page) return;\n\t\t\t\tfor (let job of page.jobs) {\n\t\t\t\t\tlet definition = this.#options.registry.get(job.type, job.version);\n\t\t\t\t\tif (\n\t\t\t\t\t\tjob.state !== \"paused\"\n\t\t\t\t\t\t|| (job.reason !== \"owner-unavailable\" && job.reason !== \"unregistered-type\")\n\t\t\t\t\t\t|| definition?.credential !== \"active-planner\"\n\t\t\t\t\t) continue;\n\t\t\t\t\ttry {\n\t\t\t\t\t\tawait this.#options.service.resume({\n\t\t\t\t\t\t\tchannelId,\n\t\t\t\t\t\t\tjobId: job.id,\n\t\t\t\t\t\t\texpectedRevision: job.revision,\n\t\t\t\t\t\t});\n\t\t\t\t\t} catch (err) {\n\t\t\t\t\t\tif (!expected(err)) throw err;\n\t\t\t\t\t}\n\t\t\t\t}\n\t\t\t\tcursor = page.next;\n\t\t\t} while (cursor);\n\t\t\tthis.wake();\n\t\t} catch (err) {\n\t\t\tthis.#options.fatal?.(err);\n\t\t\tif (!this.#stopped && !this.#ownerRecovery.has(channelId)) {\n\t\t\t\tlet cancel = this.#scheduler.after(this.#options.pollMs, () => {\n\t\t\t\t\tthis.#ownerRecovery.delete(channelId);\n\t\t\t\t\tvoid this.ownerAvailable(channelId);\n\t\t\t\t});\n\t\t\t\tthis.#ownerRecovery.set(channelId, cancel);\n\t\t\t}\n\t\t}\n\t}\n\n\tshutdown(): Promise {\n\t\tif (this.#shutdown) return this.#shutdown;\n\t\tthis.#stopped = true;\n\t\tthis.#dirty = false;\n\t\tthis.#poll?.();\n\t\tthis.#poll = undefined;\n\t\tfor (let cancel of this.#ownerRecovery.values()) cancel();\n\t\tthis.#ownerRecovery.clear();\n\t\tlet attempts = [...this.#attempts];\n\t\tlet requeue = attempts.filter(attempt => attempt.accepting);\n\t\tfor (let attempt of attempts) {\n\t\t\tattempt.accepting = false;\n\t\t\tattempt.heartbeat?.();\n\t\t\tattempt.controller.abort(new Error(\"runner-shutdown\"));\n\t\t}\n\t\treturn this.#shutdown = (async () => {\n\t\t\tawait this.#pump?.catch(() => {});\n\t\t\tawait Promise.all(requeue.map(attempt => this.#requeueStopped(attempt, \"runner-shutdown\")));\n\t\t\tawait Promise.allSettled(\n\t\t\t\tattempts.flatMap(attempt => attempt.recovery ? [attempt.recovery] : []),\n\t\t\t);\n\t\t\tawait this.#withinGrace(Promise.allSettled(attempts.map(attempt => attempt.done)));\n\t\t})();\n\t}\n\n\tasync #drain(): Promise {\n\t\tif (this.#pump || this.#stopped) return;\n\t\tthis.#pump = (async () => {\n\t\t\twhile (this.#dirty && !this.#stopped) {\n\t\t\t\tthis.#dirty = false;\n\t\t\t\tlet capacity = this.#options.globalConcurrency - this.#attempts.size;\n\t\t\t\tif (capacity < 1) continue;\n\t\t\t\ttry {\n\t\t\t\t\tlet claimed = await this.#options.storage.jobs.claim({\n\t\t\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\t\t\tcount: capacity,\n\t\t\t\t\t\tttlMs: this.#options.claimTtlMs,\n\t\t\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\t\t\tlease: this.#options.lease(),\n\t\t\t\t\t});\n\t\t\t\t\tthis.#pollFailures = 0;\n\t\t\t\t\tfor (let channelId of new Set(claimed.map(job => job.channelId))) {\n\t\t\t\t\t\tawait this.#changed(channelId);\n\t\t\t\t\t}\n\t\t\t\t\tif (this.#stopped) {\n\t\t\t\t\t\tawait Promise.all(claimed.map(job => this.#requeueClaim(job, \"runner-shutdown\")));\n\t\t\t\t\t} else {\n\t\t\t\t\t\tlet duplicates = claimed.map(job => this.#start(job)).filter(value => !!value);\n\t\t\t\t\t\tawait Promise.all(duplicates);\n\t\t\t\t\t}\n\t\t\t\t} catch (err) {\n\t\t\t\t\tthis.#pollFailures++;\n\t\t\t\t\tthis.#options.fatal?.(err);\n\t\t\t\t}\n\t\t\t}\n\t\t})().finally(() => {\n\t\t\tthis.#pump = undefined;\n\t\t\tif (this.#dirty && !this.#stopped) this.wake();\n\t\t\telse if (!this.#stopped) {\n\t\t\t\tlet delay = this.#pollFailures\n\t\t\t\t\t? Math.min(30_000, this.#options.pollMs * 2 ** Math.min(this.#pollFailures - 1, 5))\n\t\t\t\t\t: this.#options.pollMs;\n\t\t\t\tthis.#poll = this.#scheduler.after(delay, () => {\n\t\t\t\t\tthis.#poll = undefined;\n\t\t\t\t\tthis.wake();\n\t\t\t\t});\n\t\t\t}\n\t\t});\n\t\tawait this.#pump;\n\t}\n\n\t#start(job: BackgroundJob): Promise | undefined {\n\t\tif (this.#blockedChannels.has(job.channelId)) {\n\t\t\treturn this.#options.service.cancel({ channelId: job.channelId, jobId: job.id }, true)\n\t\t\t\t.then(() => {}, err => {\n\t\t\t\t\tif (!expected(err)) this.#options.fatal?.(err);\n\t\t\t\t});\n\t\t}\n\t\tlet previous = this.#active.get(job.id);\n\t\tif (previous) {\n\t\t\tthis.#detach(previous, \"claim-reclaimed\");\n\t\t\treturn this.#requeueClaim(job, \"claim-reclaimed\");\n\t\t}\n\t\tlet attempt: Attempt = {\n\t\t\tjob,\n\t\t\tcontroller: new AbortController(),\n\t\t\taccepting: true,\n\t\t\tmutation: Promise.resolve(),\n\t\t\tdone: Promise.resolve(),\n\t\t};\n\t\tthis.#active.set(job.id, attempt);\n\t\tthis.#attempts.add(attempt);\n\t\tattempt.done = this.#execute(attempt).catch(err => {\n\t\t\tthis.#options.fatal?.(err);\n\t\t}).finally(async () => {\n\t\t\tawait attempt.recovery;\n\t\t\tawait attempt.mutation;\n\t\t\tattempt.heartbeat?.();\n\t\t\tattempt.credential?.release();\n\t\t\tattempt.releaseOwner?.();\n\t\t\tthis.#attempts.delete(attempt);\n\t\t\tif (this.#active.get(job.id) === attempt) this.#active.delete(job.id);\n\t\t\tif (!this.#stopped) this.wake();\n\t\t});\n\t}\n\n\tasync #execute(attempt: Attempt): Promise {\n\t\tlet definition = this.#options.registry.get(attempt.job.type, attempt.job.version);\n\t\tif (!definition) {\n\t\t\tawait this.#pause(attempt, \"unregistered-type\");\n\t\t\treturn;\n\t\t}\n\t\tlet input: JsonValue;\n\t\ttry {\n\t\t\tinput = definition.input.parse(structuredClone(attempt.job.input));\n\t\t} catch {\n\t\t\tawait this.#fail(attempt, \"invalid-persisted-input\");\n\t\t\treturn;\n\t\t}\n\t\tthis.#heartbeat(attempt);\n\t\tlet running: Promise | undefined;\n\t\ttry {\n\t\t\tlet credential = definition.credential === \"none\"\n\t\t\t\t? {\n\t\t\t\t\tcredential: { kind: \"none\" } as const,\n\t\t\t\t\townerKey: undefined,\n\t\t\t\t\tbinding: undefined,\n\t\t\t\t\tactive: async () => true,\n\t\t\t\t\trelease: () => {},\n\t\t\t\t}\n\t\t\t\t: await this.#options.resolveActivePlanner(attempt.job, attempt.controller.signal);\n\t\t\tif (!attempt.accepting || attempt.controller.signal.aborted) {\n\t\t\t\tcredential?.release();\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tif (!credential) {\n\t\t\t\tawait this.#pause(attempt, \"owner-unavailable\");\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tattempt.credential = credential;\n\t\t\tif (credential.signal) {\n\t\t\t\tif (credential.signal.aborted) {\n\t\t\t\t\tawait this.#pause(attempt, \"owner-unavailable\");\n\t\t\t\t\treturn;\n\t\t\t\t}\n\t\t\t\tcredential.signal.addEventListener(\"abort\", () => {\n\t\t\t\t\tvoid this.#pause(attempt, \"owner-unavailable\");\n\t\t\t\t}, { once: true });\n\t\t\t}\n\t\t\tawait this.#serial(attempt, async () => {\n\t\t\t\tattempt.job = await this.#options.storage.jobs.renew({\n\t\t\t\t\tchannelId: attempt.job.channelId,\n\t\t\t\t\tjobId: attempt.job.id,\n\t\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\t\tclaimGeneration: attempt.job.claimGeneration,\n\t\t\t\t\texpectedRevision: attempt.job.revision,\n\t\t\t\t\tclaimBinding: credential.binding,\n\t\t\t\t\tttlMs: this.#options.claimTtlMs + this.#options.heartbeatMs + 1,\n\t\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\t\tlease: this.#options.lease(),\n\t\t\t\t});\n\t\t\t});\n\t\t\tthis.#heartbeat(attempt);\n\t\t\tif (!this.#accepted(attempt, credential)) return;\n\t\t\tattempt.releaseOwner = await this.#owner(credential.ownerKey);\n\t\t\tif (!attempt.releaseOwner) {\n\t\t\t\tawait this.#retry(attempt, \"owner-capacity\", false);\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tif (!this.#accepted(attempt, credential)) return;\n\t\t\tlet started = this.#scheduler.now();\n\t\t\tlet deadline = new Date(started.getTime() + definition.limits.timeoutMs);\n\t\t\tlet timeout: () => void = noop;\n\t\t\tlet timedOut = new Promise((_, reject) => {\n\t\t\t\ttimeout = this.#scheduler.after(definition.limits.timeoutMs, () => {\n\t\t\t\t\treject(new Error(\"attempt-timeout\"));\n\t\t\t\t});\n\t\t\t});\n\t\t\trunning = Promise.resolve().then(() =>\n\t\t\t\tdefinition.execute({\n\t\t\t\t\tjob: structuredClone(attempt.job),\n\t\t\t\t\tinput,\n\t\t\t\t\tcredential: credential.credential,\n\t\t\t\t\tsignal: attempt.controller.signal,\n\t\t\t\t\tdeadline,\n\t\t\t\t\tprogress: (stage, state) => this.#progress(attempt, definition, stage, state),\n\t\t\t\t})\n\t\t\t);\n\t\t\tattempt.execution = running;\n\t\t\tvoid running.catch(() => {});\n\t\t\tlet artifact = await Promise.race([running, timedOut]).finally(timeout);\n\t\t\tif (!this.#accepted(attempt, credential)) return;\n\t\t\tif (!await credential.active()) {\n\t\t\t\tawait this.#pause(attempt, \"owner-unavailable\");\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tif (!this.#accepted(attempt, credential)) return;\n\t\t\tawait this.#options.service.settle({\n\t\t\t\tchannelId: attempt.job.channelId,\n\t\t\t\tjobId: attempt.job.id,\n\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\tclaimGeneration: attempt.job.claimGeneration,\n\t\t\t\tartifact,\n\t\t\t\tguard: async () => {\n\t\t\t\t\tif (!this.#accepted(attempt, credential)) return false;\n\t\t\t\t\tlet active = await credential.active();\n\t\t\t\t\treturn active && this.#accepted(attempt, credential);\n\t\t\t\t},\n\t\t\t});\n\t\t\tattempt.accepting = false;\n\t\t} catch (err) {\n\t\t\tif (!attempt.accepting) return;\n\t\t\tif (err instanceof StorageError && err.failure === \"unavailable\") {\n\t\t\t\tthis.#detach(attempt, \"storage-unavailable\");\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tif (expected(err)) {\n\t\t\t\tthis.#detach(attempt, \"claim-lost\");\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tlet timeout = err instanceof Error && err.message === \"attempt-timeout\";\n\t\t\tif (timeout && running) {\n\t\t\t\tawait this.#timeout(attempt, running, definition.limits.maxAttempts);\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tif (attempt.credential && !await attempt.credential.active().catch(() => false)) {\n\t\t\t\tawait this.#pause(attempt, \"owner-unavailable\");\n\t\t\t\treturn;\n\t\t\t}\n\t\t\ttry {\n\t\t\t\tlet diagnostic = this.#options.attemptFailed?.(structuredClone(attempt.job), err);\n\t\t\t\tvoid Promise.resolve(diagnostic).catch(noop);\n\t\t\t} catch {\n\t\t\t\t// Diagnostics must not change durable retry behavior.\n\t\t\t}\n\t\t\tlet reason = err instanceof JobExecutionError ? err.progressReason : \"attempt-error\";\n\t\t\tawait this.#retry(attempt, timeout ? \"attempt-timeout\" : reason, true);\n\t\t}\n\t}\n\n\tasync #progress(\n\t\tattempt: Attempt,\n\t\tdefinition: JobDefinition,\n\t\tstage: string,\n\t\tstate: \"started\" | \"completed\",\n\t): Promise {\n\t\tlet label = definition.progress?.[stage];\n\t\tif (!label) {\n\t\t\tthrow new Error(`Background job ${definition.type} reported an unknown progress stage.`);\n\t\t}\n\t\tif (!attempt.accepting || attempt.controller.signal.aborted) {\n\t\t\tthrow new Error(\"Background job progress is no longer accepted.\");\n\t\t}\n\t\tawait this.#serial(attempt, async () => {\n\t\t\tif (!attempt.accepting || attempt.controller.signal.aborted) {\n\t\t\t\tthrow new Error(\"Background job progress is no longer accepted.\");\n\t\t\t}\n\t\t\tattempt.job = await this.#options.storage.jobs.appendProgress({\n\t\t\t\tchannelId: attempt.job.channelId,\n\t\t\t\tjobId: attempt.job.id,\n\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\tclaimGeneration: attempt.job.claimGeneration,\n\t\t\t\tstage,\n\t\t\t\tlabel,\n\t\t\t\tstate,\n\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\tlease: this.#options.lease(),\n\t\t\t});\n\t\t});\n\t\tawait this.#changed(attempt.job.channelId);\n\t}\n\n\t#heartbeat(attempt: Attempt): void {\n\t\tattempt.heartbeat?.();\n\t\tattempt.heartbeat = this.#scheduler.after(this.#options.heartbeatMs, () => {\n\t\t\tattempt.heartbeat = undefined;\n\t\t\tif (!attempt.accepting || this.#stopped) return;\n\t\t\tvoid this.#serial(attempt, async () => {\n\t\t\t\tattempt.job = await this.#options.storage.jobs.renew({\n\t\t\t\t\tchannelId: attempt.job.channelId,\n\t\t\t\t\tjobId: attempt.job.id,\n\t\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\t\tclaimGeneration: attempt.job.claimGeneration,\n\t\t\t\t\texpectedRevision: attempt.job.revision,\n\t\t\t\t\tclaimBinding: attempt.credential?.binding,\n\t\t\t\t\tttlMs: this.#options.claimTtlMs + this.#options.heartbeatMs + 1,\n\t\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\t\tlease: this.#options.lease(),\n\t\t\t\t});\n\t\t\t}).then(async () => {\n\t\t\t\tif (!attempt.accepting) return;\n\t\t\t\ttry {\n\t\t\t\t\tif (\n\t\t\t\t\t\tattempt.credential\n\t\t\t\t\t\t&& !await this.#withinHeartbeat(attempt.credential.active())\n\t\t\t\t\t) {\n\t\t\t\t\t\tawait this.#pause(attempt, \"owner-unavailable\");\n\t\t\t\t\t\treturn;\n\t\t\t\t\t}\n\t\t\t\t} catch {\n\t\t\t\t\tthis.#recoverHeartbeat(attempt);\n\t\t\t\t\treturn;\n\t\t\t\t}\n\t\t\t\tif (attempt.accepting) this.#heartbeat(attempt);\n\t\t\t}, () => {\n\t\t\t\tthis.#recoverHeartbeat(attempt);\n\t\t\t});\n\t\t});\n\t}\n\n\tasync #heartbeatLost(attempt: Attempt): Promise {\n\t\tif (!attempt.accepting) return;\n\t\tattempt.accepting = false;\n\t\tattempt.heartbeat?.();\n\t\tattempt.controller.abort(new Error(\"heartbeat-lost\"));\n\t\tlet settled = !attempt.execution || await this.#settledWithinGrace(attempt.execution);\n\t\tawait this.#interrupt(attempt, \"heartbeat-lost\");\n\t\tlet definition = this.#options.registry.get(attempt.job.type, attempt.job.version);\n\t\tif (!settled || !definition || attempt.job.failures + 1 >= definition.limits.maxAttempts) {\n\t\t\tawait this.#failStopped(attempt, \"attempts-exhausted:heartbeat-lost\");\n\t\t} else await this.#requeueStopped(attempt, \"heartbeat-lost\", true);\n\t}\n\n\t#recoverHeartbeat(attempt: Attempt): void {\n\t\tif (attempt.recovery) return;\n\t\tattempt.recovery = this.#heartbeatLost(attempt).catch(err => {\n\t\t\tthis.#options.fatal?.(err);\n\t\t});\n\t}\n\n\tasync #interrupt(attempt: Attempt, reason: string): Promise {\n\t\tlet latest = new Map();\n\t\tfor (let entry of attempt.job.progress) {\n\t\t\tif (entry.attempt === attempt.job.attempts) latest.set(entry.stage, entry);\n\t\t}\n\t\tlet active = [...latest.values()].findLast(entry => entry.state === \"started\");\n\t\tif (!active) return;\n\t\ttry {\n\t\t\tawait this.#serial(attempt, async () => {\n\t\t\t\tattempt.job = await this.#options.storage.jobs.appendProgress({\n\t\t\t\t\tchannelId: attempt.job.channelId,\n\t\t\t\t\tjobId: attempt.job.id,\n\t\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\t\tclaimGeneration: attempt.job.claimGeneration,\n\t\t\t\t\tstage: active.stage,\n\t\t\t\t\tlabel: active.label,\n\t\t\t\t\tstate: \"interrupted\",\n\t\t\t\t\treason,\n\t\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\t\tlease: this.#options.lease(),\n\t\t\t\t});\n\t\t\t});\n\t\t} catch (err) {\n\t\t\tif (!expected(err)) this.#options.fatal?.(err);\n\t\t}\n\t}\n\n\t#serial(attempt: Attempt, action: () => Promise): Promise {\n\t\tlet result = attempt.mutation.then(action);\n\t\tattempt.mutation = result.then(() => {}, () => {});\n\t\treturn result;\n\t}\n\n\t#accepted(attempt: Attempt, credential?: ResolvedJobCredential): boolean {\n\t\treturn attempt.accepting\n\t\t\t&& !attempt.controller.signal.aborted\n\t\t\t&& !credential?.signal?.aborted\n\t\t\t&& !this.#stopped;\n\t}\n\n\tasync #pause(attempt: Attempt, reason: string): Promise {\n\t\tif (!attempt.accepting) return;\n\t\tattempt.accepting = false;\n\t\tattempt.heartbeat?.();\n\t\tattempt.controller.abort(new Error(reason));\n\t\tawait this.#interrupt(attempt, reason);\n\t\ttry {\n\t\t\tattempt.job = await this.#serial(attempt, () =>\n\t\t\t\tthis.#options.storage.jobs.pause({\n\t\t\t\t\tchannelId: attempt.job.channelId,\n\t\t\t\t\tjobId: attempt.job.id,\n\t\t\t\t\texpectedRevision: attempt.job.revision,\n\t\t\t\t\treason,\n\t\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\t\tlease: this.#options.lease(),\n\t\t\t\t}));\n\t\t\tawait this.#changed(attempt.job.channelId);\n\t\t} catch (err) {\n\t\t\tif (!expected(err)) this.#options.fatal?.(err);\n\t\t}\n\t}\n\n\tasync #fail(attempt: Attempt, reason: string): Promise {\n\t\tif (!attempt.accepting) return;\n\t\tattempt.accepting = false;\n\t\tattempt.heartbeat?.();\n\t\ttry {\n\t\t\tattempt.job = await this.#serial(attempt, () =>\n\t\t\t\tthis.#options.storage.jobs.fail({\n\t\t\t\t\tchannelId: attempt.job.channelId,\n\t\t\t\t\tjobId: attempt.job.id,\n\t\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\t\tclaimGeneration: attempt.job.claimGeneration,\n\t\t\t\t\treason,\n\t\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\t\tlease: this.#options.lease(),\n\t\t\t\t}));\n\t\t\tawait this.#changed(attempt.job.channelId);\n\t\t} catch (err) {\n\t\t\tif (!expected(err)) this.#options.fatal?.(err);\n\t\t} finally {\n\t\t\tattempt.controller.abort(new Error(reason));\n\t\t}\n\t}\n\n\tasync #retry(attempt: Attempt, reason: string, countFailure: boolean): Promise {\n\t\tif (!attempt.accepting) return;\n\t\tlet definition = this.#options.registry.get(attempt.job.type, attempt.job.version);\n\t\tlet failures = attempt.job.failures + (countFailure ? 1 : 0);\n\t\tattempt.accepting = false;\n\t\tattempt.heartbeat?.();\n\t\tattempt.controller.abort(new Error(reason));\n\t\tawait this.#interrupt(attempt, reason);\n\t\tif (countFailure && definition && failures >= definition.limits.maxAttempts) {\n\t\t\tawait this.#failStopped(attempt, `attempts-exhausted:${reason}`);\n\t\t\treturn;\n\t\t}\n\t\ttry {\n\t\t\tlet availableAt = new Date(\n\t\t\t\tthis.#scheduler.now().getTime()\n\t\t\t\t\t+ (countFailure\n\t\t\t\t\t\t? retryDelay(failures, this.#options.retryBaseMs, this.#options.retryMaxMs)\n\t\t\t\t\t\t: this.#options.retryBaseMs),\n\t\t\t);\n\t\t\tattempt.job = await this.#serial(attempt, () =>\n\t\t\t\tthis.#options.storage.jobs.requeue({\n\t\t\t\t\tchannelId: attempt.job.channelId,\n\t\t\t\t\tjobId: attempt.job.id,\n\t\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\t\tclaimGeneration: attempt.job.claimGeneration,\n\t\t\t\t\tavailableAt,\n\t\t\t\t\treason,\n\t\t\t\t\tcountFailure,\n\t\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\t\tlease: this.#options.lease(),\n\t\t\t\t}));\n\t\t\tawait this.#changed(attempt.job.channelId);\n\t\t} catch (err) {\n\t\t\tif (!expected(err)) this.#options.fatal?.(err);\n\t\t}\n\t}\n\n\tasync #requeueStopped(attempt: Attempt, reason: string, countFailure = false): Promise {\n\t\ttry {\n\t\t\tattempt.job = await this.#serial(attempt, () =>\n\t\t\t\tthis.#options.storage.jobs.requeue({\n\t\t\t\t\tchannelId: attempt.job.channelId,\n\t\t\t\t\tjobId: attempt.job.id,\n\t\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\t\tclaimGeneration: attempt.job.claimGeneration,\n\t\t\t\t\tavailableAt: new Date(this.#scheduler.now().getTime() + this.#options.retryBaseMs),\n\t\t\t\t\treason,\n\t\t\t\t\tcountFailure,\n\t\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\t\tlease: this.#options.lease(),\n\t\t\t\t}));\n\t\t\tawait this.#changed(attempt.job.channelId);\n\t\t} catch (err) {\n\t\t\tif (!expected(err)) this.#options.fatal?.(err);\n\t\t}\n\t}\n\n\tasync #failStopped(attempt: Attempt, reason: string): Promise {\n\t\ttry {\n\t\t\tattempt.job = await this.#serial(attempt, () =>\n\t\t\t\tthis.#options.storage.jobs.fail({\n\t\t\t\t\tchannelId: attempt.job.channelId,\n\t\t\t\t\tjobId: attempt.job.id,\n\t\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\t\tclaimGeneration: attempt.job.claimGeneration,\n\t\t\t\t\treason,\n\t\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\t\tlease: this.#options.lease(),\n\t\t\t\t}));\n\t\t\tawait this.#changed(attempt.job.channelId);\n\t\t} catch (err) {\n\t\t\tif (!expected(err)) this.#options.fatal?.(err);\n\t\t}\n\t}\n\n\tasync #timeout(\n\t\tattempt: Attempt,\n\t\texecution: Promise,\n\t\tmaxAttempts: number,\n\t): Promise {\n\t\tif (!attempt.accepting) return;\n\t\tattempt.accepting = false;\n\t\tattempt.heartbeat?.();\n\t\tattempt.controller.abort(new Error(\"attempt-timeout\"));\n\t\tlet cleanup = this.#withinGrace(execution.then(() => {}, () => {}));\n\t\tawait this.#interrupt(attempt, \"attempt-timeout\");\n\t\tif (attempt.job.failures + 1 >= maxAttempts) {\n\t\t\tawait this.#failStopped(attempt, \"attempts-exhausted:attempt-timeout\");\n\t\t} else {\n\t\t\tawait this.#requeueStopped(attempt, \"attempt-timeout\", true);\n\t\t}\n\t\tawait cleanup;\n\t}\n\n\tasync #requeueClaim(job: BackgroundJob, reason: string): Promise {\n\t\ttry {\n\t\t\tawait this.#options.storage.jobs.requeue({\n\t\t\t\tchannelId: job.channelId,\n\t\t\t\tjobId: job.id,\n\t\t\t\tclaimOwner: this.#claimOwner,\n\t\t\t\tclaimGeneration: job.claimGeneration,\n\t\t\t\tavailableAt: new Date(this.#scheduler.now().getTime() + this.#options.retryBaseMs),\n\t\t\t\treason,\n\t\t\t\tcountFailure: false,\n\t\t\t\tnow: this.#scheduler.now(),\n\t\t\t\tlease: this.#options.lease(),\n\t\t\t});\n\t\t\tawait this.#changed(job.channelId);\n\t\t} catch (err) {\n\t\t\tif (!expected(err)) this.#options.fatal?.(err);\n\t\t}\n\t}\n\n\t#detach(attempt: Attempt, reason: string): void {\n\t\tif (!attempt.accepting) return;\n\t\tattempt.accepting = false;\n\t\tattempt.heartbeat?.();\n\t\tattempt.controller.abort(new Error(reason));\n\t}\n\n\t#withinGrace(operation: Promise): Promise {\n\t\treturn new Promise(resolve => {\n\t\t\tlet cancel = this.#scheduler.after(this.#options.shutdownGraceMs, resolve);\n\t\t\toperation.then(\n\t\t\t\t() => {\n\t\t\t\t\tcancel();\n\t\t\t\t\tresolve();\n\t\t\t\t},\n\t\t\t\t() => {\n\t\t\t\t\tcancel();\n\t\t\t\t\tresolve();\n\t\t\t\t},\n\t\t\t);\n\t\t});\n\t}\n\n\tasync #settledWithinGrace(operation: Promise): Promise {\n\t\tlet settled = false;\n\t\tlet observed = operation.then(\n\t\t\t() => settled = true,\n\t\t\t() => settled = true,\n\t\t);\n\t\tawait this.#withinGrace(observed);\n\t\treturn settled;\n\t}\n\n\t#withinHeartbeat(operation: Promise): Promise {\n\t\treturn new Promise((resolve, reject) => {\n\t\t\tlet cancel = this.#scheduler.after(this.#options.heartbeatMs, () => {\n\t\t\t\treject(new Error(\"owner revalidation timed out\"));\n\t\t\t});\n\t\t\toperation.then(\n\t\t\t\tvalue => {\n\t\t\t\t\tcancel();\n\t\t\t\t\tresolve(value);\n\t\t\t\t},\n\t\t\t\terr => {\n\t\t\t\t\tcancel();\n\t\t\t\t\treject(err);\n\t\t\t\t},\n\t\t\t);\n\t\t});\n\t}\n\n\t#owner(key: string | undefined): Promise<(() => void) | undefined> {\n\t\tif (!key) return Promise.resolve(() => {});\n\t\tlet active = this.#owners.get(key) ?? 0;\n\t\tif (active >= this.#options.ownerConcurrency) return Promise.resolve(undefined);\n\t\tthis.#owners.set(key, active + 1);\n\t\tlet released = false;\n\t\treturn Promise.resolve(() => {\n\t\t\tif (released) return;\n\t\t\treleased = true;\n\t\t\tthis.#releaseOwner(key);\n\t\t});\n\t}\n\n\t#releaseOwner(key: string): void {\n\t\tlet active = Math.max(0, (this.#owners.get(key) ?? 1) - 1);\n\t\tif (active === 0) this.#owners.delete(key);\n\t\telse this.#owners.set(key, active);\n\t}\n\n\tasync #changed(channelId: string): Promise {\n\t\ttry {\n\t\t\tawait this.#options.changed?.(channelId);\n\t\t} catch (err) {\n\t\t\tthis.#options.fatal?.(err);\n\t\t}\n\t}\n}\n","editable",false,"branch","documentNavigation","repo","githubnext/chopin","repoId","7d1a8ac3-35b0-4dea-85d4-06a980f3bada","ownerId","org:89615882","organization",true,"site","nexthub","private","admin","settings","maintainer","triage","starred","stars",0,"watchers","watching","pinned","notify","views",[184,185,186,187,188,189,190,191,192,193,194,195,196],"navigation",{"_56":57,"_141":142,"_143":144,"_145":146,"_147":148,"_149":150,"_151":152},"sources",[],"viewer","meta",{"_133":134,"_135":136,"_137":138,"_44":31,"_139":128,"_140":31,"_50":51,"_52":51},"branches",[75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99,100,101,102,103,104,105,106,107,108,109,110,111,112,113,114,115,116,117,118,119,120,121,122,123,124,125,126,127,128,129,130,131,132],"tags",[71,72,73,74],"cloneUrl","https://git.nexthub.sh/githubnext/chopin.git","color-controls-slice-1","color-controls-slice-2","color-controls-slice-3","color-controls-slice-4","Krzysztof-Cieslak-patch-1","Krzysztof-Cieslak-patch-2","Maggie/app-preview-toolkit","Maggie/ci-playwright-install-timeout","Maggie/collaborative-visual-decision","Maggie/design-audit","Maggie/generic-visual-decision-runtime","Maggie/impeccable-chopin-metadata","Maggie/jev-chat-product","Maggie/jev-composed-assessment","Maggie/native-document-layouts","Maggie/openui-composition-product","Maggie/openui-composition-trial","Maggie/planner-visual-authoring","Maggie/planner-visual-requests","Maggie/visual-selection-eval","artifacts/pr-419-live-preview-20261009","automation/pr-maintenance-state","automation/use-effect/callout-disabled-picker-8cf9e49ba5c09471","automation/use-effect/collaboration-options-ref-4e3a5a08a879a5ae","automation/use-effect/room-workspace-effective-access-a57d40866c71bed0","chore/bun-1.4.2","color-controls","color-controls-plan","comments","design-pr-assets","design/archived-decisions-badge","design/channel-route-remount","design/chat-decision-entry","design/conversation-analysis-markers","design/decision-evidence","design/settled-line-icon","dev/integration","diagram-step-controls","feat/liveapp-frontend-pilot","feat/local-experiments","feat/local-launcher-tracer","fix/autolink-mdx-crash","github-pills/2-pills","github-pills/3-hover-card","maggie/app-preview-toolkit","maggie/ci-playwright-install-timeout","maggie/collaborative-visual-decision","maggie/design-audit","maggie/generic-visual-decision-runtime","maggie/impeccable-chopin-metadata","maggie/jev-composed-assessment","maggie/native-document-layouts","maggie/openui-composition-product","maggie/openui-composition-trial","maggie/planner-visual-authoring","maggie/planner-visual-requests","maggie/visual-selection-eval","main","plain-text-fence-chrome","tokens/09-copy","wireframes/2-renderer","wireframes/3-planner","description","Let's compose rich plans together","owner","githubnext","section","","head","empty","pins",[167,168,169],"documents",{"_162":163,"_164":165},"code",{"_158":159,"_160":161},"automations",{"_155":156,"_157":-5},"delivery",{"_153":51,"_154":-5},"tag","04zvKQ8QOrDUF8TO1Ovz0Zy8_TT","releases","applications","workflows",3,"deployments","commits",1000,"changes",13,"counts",{"_166":165},"total",11,"issue",{"_170":160,"_172":182,"_15":160,"_174":183,"_54":41,"_176":31,"_177":41,"_178":161},{"_170":179,"_172":180,"_15":179,"_174":181,"_54":41,"_176":31,"_177":41,"_178":51},{"_170":171,"_172":173,"_15":171,"_174":175,"_54":41,"_176":31,"_177":41,"_178":165},"id","issues","name","Issues","filters",{},"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_170":160,"_172":182,"_15":160,"_174":227,"_54":41,"_176":31,"_177":41},{"_170":179,"_172":180,"_15":179,"_174":226,"_54":41,"_176":31,"_177":41},{"_170":171,"_172":173,"_15":171,"_174":225,"_54":41,"_176":31,"_177":41},{"_170":222,"_172":223,"_15":222,"_174":224,"_54":31,"_176":31,"_177":41},{"_170":219,"_172":220,"_15":219,"_174":221,"_54":31,"_176":31,"_177":41},{"_170":155,"_172":217,"_15":147,"_174":218,"_54":31,"_176":31,"_177":41},{"_170":157,"_172":214,"_15":215,"_174":216,"_54":31,"_176":31,"_177":41},{"_170":153,"_172":212,"_15":153,"_174":213,"_54":31,"_176":31,"_177":41},{"_170":208,"_172":209,"_15":210,"_174":211,"_54":31,"_176":31,"_177":41},{"_170":143,"_172":206,"_15":143,"_174":207,"_54":31,"_176":31,"_177":31},{"_170":203,"_172":204,"_15":203,"_174":205,"_54":31,"_176":31,"_177":31},{"_170":200,"_172":201,"_15":200,"_174":202,"_54":31,"_176":31,"_177":31},{"_170":197,"_172":198,"_15":197,"_174":199,"_54":31,"_176":31,"_177":31},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794612223357,"sidebar","expanded"]