diff --git a/api/server/controllers/UserController.js b/api/server/controllers/UserController.js index 44b94ce1c8..0dfd9f8cef 100644 --- a/api/server/controllers/UserController.js +++ b/api/server/controllers/UserController.js @@ -356,9 +356,21 @@ const deleteUserController = async (req, res) => { // scheduling admission consults this durable user-level flag instead. Raising it // also invalidates the auth user-doc cache, without which the barrier would only be // as strong as the shortest cache TTL. - await markUserDeleting(user.id).catch((error) => - logger.error('[deleteUserController] Failed to raise the deletion barrier', error), - ); + let barrierRaised = true; + try { + await markUserDeleting(user.id); + } catch (error) { + barrierRaised = false; + logger.error('[deleteUserController] Failed to raise the deletion barrier', error); + } + if (!barrierRaised) { + // FAIL CLOSED. Without the barrier we cannot promise that admission is refused for + // the rest of the cascade, so destroying now could let concurrently-created work + // outlive the account. Refuse rather than proceed on an unenforced guarantee. + return res.status(503).json({ + message: 'Could not start account deletion. Please retry.', + }); + } const quiesced = await quiesceUserSchedules(user.id).catch((error) => { logger.error('[deleteUserController] Failed to quiesce scheduled chats', error); diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index 95c09736a3..ee201f8d02 100644 --- a/api/server/controllers/agents/request.js +++ b/api/server/controllers/agents/request.js @@ -866,6 +866,11 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit scheduledFor, status: 'requires_action', conversationId: streamId, + // Segment 0: this generation came from the fire, which creates the run + // with no resumeSeq. The approval card is emitted BEFORE this write, so a + // fast resume can promote the run (bumping the epoch) first — without this + // token the stale pause would demote the run that resume just started. + expectResumeSeq: 0, }); } logger.debug( diff --git a/api/server/index.js b/api/server/index.js index 59f21d570d..65672e49f2 100644 --- a/api/server/index.js +++ b/api/server/index.js @@ -368,12 +368,13 @@ const startServer = async () => { // via SCHEDULES_CLUSTERED (or use USE_REDIS_STREAMS for a shared store). Without // that, a peer replica's in-memory job would be misread as gone and its live run // wrongly interrupted after the orphan cutoff. - const scheduleEngine = await initializeScheduleEngine({ - clustered: isEnabled(process.env.SCHEDULES_CLUSTERED), - }); + const clustered = isEnabled(process.env.SCHEDULES_CLUSTERED); + const scheduleEngine = await initializeScheduleEngine({ clustered }); // Only accept schedule writes once the engine confirmed its unique idempotency // + TTL indexes exist. If index creation failed the engine is left undefined and // schedule writes keep returning 503 (the app otherwise runs normally). + // The engine refuses to arm on an unsafe clustered topology, so a null engine + // already means "not scheduling here"; the write gate follows it. schedulesReady = scheduleEngine != null; if (!schedulesReady) { logger.warn( diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 130b927962..0f6f92b8c1 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -96,7 +96,38 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { conversationId: run.conversationId, error, autoDisableAfterFailures: runLimits.autoDisableAfterFailures, + // Fence the reconciler's own pause on the epoch it OBSERVED: several awaits + // separate the read from this write, and a resume landing in between must + // not be demoted by a sweep that is now looking at a stale segment. + ...(status === 'requires_action' ? { expectResumeSeq: run.resumeSeq ?? 0 } : {}), + // Terminal bookkeeping is fenced on the config the run started under, so a + // reconciled outcome cannot auto-disable a schedule the owner has edited. + ...(run.configRevision != null ? { expectConfigRevision: run.configRevision } : {}), }); + // RESUME-LEASE RECOVERY. A resume that died between consuming the approval and + // reconstructing its generation leaves the row `started` with a live-looking + // lease, and the `running` skip below would step over it forever. Once the + // lease deadline passes, decide by phase: + // post-claim (resumeClaimedAt set) -> the approval is spent, so it can never + // be re-offered; roll FORWARD by terminalizing as interrupted. + // pre-claim -> the approval was never consumed, so roll BACK to + // requires_action (releaseResumeLease frees the slot) and let the user retry. + if ( + run.resumeHolder != null && + run.resumeExpiresAt != null && + run.resumeExpiresAt.getTime() < Date.now() + ) { + if (run.resumeClaimedAt != null) { + await finalize('interrupted', 'Resume did not complete'); + } else { + await deps.methods.releaseResumeLease( + run.scheduleId, + run.scheduledFor, + run.resumeHolder, + ); + } + continue; + } if (jobStatus === 'running') { continue; } diff --git a/packages/api/src/schedules/service.spec.ts b/packages/api/src/schedules/service.spec.ts index 33ac1770e7..c7e0addda7 100644 --- a/packages/api/src/schedules/service.spec.ts +++ b/packages/api/src/schedules/service.spec.ts @@ -22,6 +22,7 @@ function makeService( disableUserSchedulesForDeletion: jest.fn(async () => undefined), getActiveRunsForUser, countActiveRuns: jest.fn(async () => 0), + requestRunAbort: jest.fn(async () => true), }; const deps = { methods, diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 75c38766b4..0be5ec5efc 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -424,11 +424,17 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer // another worker, and orphan reaping is disabled. This is unsupported for // scheduled chats — warn the operator to enable USE_REDIS_STREAMS. if (clustered && !GenerationJobManager.isRedis) { - logger.warn( + // FAIL CLOSED: do not arm the engine at all. Gating only API writes still left + // EXISTING schedules being claimed and fired on private per-worker stores, where + // a peer's run is unreachable for abort/quiesce and orphan recovery cannot see it. + // Returning undefined also keeps the write gate shut (it keys on the engine). + logger.error( '[schedules] clustered deployment without a shared stream store (USE_REDIS_STREAMS): ' + 'scheduled-run peer aborts (deletion/account-deletion quiescing) and cross-worker ' + - 'orphan recovery are NOT available. Enable USE_REDIS_STREAMS for safe multi-worker scheduling.', + 'orphan recovery are NOT available. The scheduler is DISABLED for this process. ' + + 'Enable USE_REDIS_STREAMS for safe multi-worker scheduling.', ); + return undefined; } // Explicitly build the Schedule/ScheduleRun indexes first — the unique // idempotency index and TTL retention index would otherwise never exist when @@ -459,6 +465,11 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer schedule: FireableSchedule, limits: ScheduleLimits, ): Promise { + // The global stop means STOP: a manual run dispatches the same billed generation as + // an automatic one, so gating only the engine tick would leave Run Now wide open. + if (await engineDeps.isGloballyDisabled()) { + return { fired: false, skipped: 'disabled' as const }; + } const leased = await methods.acquireManualRunLease( schedule.id, schedule.user, @@ -581,6 +592,11 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer if (!scheduleId || !scheduledFor) { return { outcome: 'not-scheduled' }; } + // A resume restarts a billed generation, so the global stop applies here too. + // Reported as 'gone' so the approval is left unconsumed and the caller defers. + if (await engineDeps.isGloballyDisabled()) { + return { outcome: 'gone' }; + } // getScheduleById hides deleted/soft-deleted schedules, so a null here means the // owner already deleted the schedule — its paused run must not be resumable even // if the delete's best-effort abort raced. Reject before touching the run. @@ -665,6 +681,13 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer if (!run.conversationId) { return false; } + // Record the abort REQUEST before signalling. This keeps the run holding its global + // capacity slot until its generation owner writes a terminal outcome (settlement), + // so an abort that has been asked for but not yet honored cannot free capacity for a + // new run while the old generation is still alive. + await methods + .requestRunAbort(run.scheduleId, run.scheduledFor) + .catch((err) => logger.warn('[schedules] failed to record abort request:', err)); return engineDeps .abortScheduledJob( run.conversationId, diff --git a/packages/api/src/stream/implementations/InMemoryJobStore.ts b/packages/api/src/stream/implementations/InMemoryJobStore.ts index 0dfa05e096..5e0f9d82de 100644 --- a/packages/api/src/stream/implementations/InMemoryJobStore.ts +++ b/packages/api/src/stream/implementations/InMemoryJobStore.ts @@ -225,6 +225,8 @@ export class InMemoryJobStore implements IJobStore { async deleteJob(streamId: string, expectedCreatedAt?: number): Promise { // Generation-fenced: an omitted guard behaves exactly as before. Read-check-write // in one synchronous block, so no replacement can land in between. + // Guard BEFORE any teardown: a refused delete must leave the replacement + // generation's state completely untouched. if (expectedCreatedAt != null && this.jobs.get(streamId)?.createdAt !== expectedCreatedAt) { return false; } diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index ed135531f2..0ab08542f4 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -193,6 +193,18 @@ const JOB_DELETE_LUA = 'redis.call("DEL", KEYS[3]) redis.call("DEL", KEYS[4]) ' + 'return 1'; +/** + * Allocates a STRICTLY monotonic generation stamp for a stream, atomically. Reading the + * prior createdAt in TS and computing the next value in the client is a race: two + * concurrent creates observe the same prior and mint the same stamp, which is exactly + * the collision the stamp exists to prevent. ARGV[1] is the caller's wall clock. + */ +const JOB_STAMP_LUA = + 'local prev = redis.call("HGET", KEYS[1], "createdAt") ' + + 'local now = tonumber(ARGV[1]) ' + + 'if prev and tonumber(prev) and tonumber(prev) >= now then return tonumber(prev) + 1 end ' + + 'return now'; + const STEER_DRAIN_LUA = 'if ARGV[1] ~= "" and redis.call("HGET", KEYS[1], "createdAt") ~= ARGV[1] then return {} end ' + 'local items = redis.call("LRANGE", KEYS[2], 0, -1) ' + @@ -439,21 +451,18 @@ export class RedisJobStore implements IJobStore { conversationId?: string, tenantId?: string, ): Promise { - // STRICTLY monotonic per streamId: createdAt is the generation fence and a streamId - // is reused across generations, so two creates in the same millisecond would mint - // identical tokens and let a stale caller's guard pass against the replacement. - const previousCreatedAt = await this.redis - .hget(KEYS.job(streamId), 'createdAt') - .then((value) => (value == null ? undefined : parseInt(value, 10))) - .catch(() => undefined); + // STRICTLY monotonic per streamId, allocated ATOMICALLY inside Redis: createdAt is + // the generation fence and a streamId is reused across generations, so a read-then- + // compute in the client would let two concurrent creates mint the same stamp. + const allocated = await this.redis + .eval(JOB_STAMP_LUA, 1, KEYS.job(streamId), String(Date.now())) + .catch(() => null); const job: SerializableJobData = { streamId, userId, ...(tenantId && { tenantId }), status: 'running', - createdAt: nextGenerationStamp( - Number.isFinite(previousCreatedAt) ? previousCreatedAt : undefined, - ), + createdAt: typeof allocated === 'number' ? allocated : nextGenerationStamp(), conversationId, syncSent: false, }; @@ -785,9 +794,9 @@ export class RedisJobStore implements IJobStore { } async deleteJob(streamId: string, expectedCreatedAt?: number): Promise { - this.localGraphCache.delete(streamId); - this.localContentParts.delete(streamId); - this.localCollectedUsageCache.delete(streamId); + // Local caches are dropped only AFTER the guarded delete fires (see + // deleteJobInternal): clearing them up front would wipe the REPLACEMENT + // generation's cached graph/content when the guard refuses this delete. const job = await this.getJob(streamId); // Cheap pre-check so a mismatched generation costs no round trip; the Lua guard // below is what actually makes it atomic. @@ -803,10 +812,6 @@ export class RedisJobStore implements IJobStore { userJobsKey: string | null, expectedCreatedAt?: number, ): Promise { - this.localGraphCache.delete(streamId); - this.localContentParts.delete(streamId); - this.localCollectedUsageCache.delete(streamId); - // The four per-stream keys share the {streamId} hash tag, so the guarded delete is // one atomic single-slot script (safe on Cluster). The three SREMs are cross-slot // and stay OUTSIDE the script, running only when the script actually deleted — @@ -824,6 +829,10 @@ export class RedisJobStore implements IJobStore { logger.debug(`[RedisJobStore] Delete skipped (generation mismatch): ${streamId}`); return false; } + // Safe now: this delete owned the generation. + this.localGraphCache.delete(streamId); + this.localContentParts.delete(streamId); + this.localCollectedUsageCache.delete(streamId); if (this.isCluster) { await this.redis.srem(KEYS.runningJobs, streamId); await this.redis.srem(KEYS.requiresActionJobs, streamId); diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index 211cd6e52b..1ef36a6c3e 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -1095,13 +1095,15 @@ describe('lifecycle barriers: stale pause vs resume, and concurrent resumes at c const schedule = await methods.createSchedule(scheduleData()); await methods.insertScheduleRun(runData(schedule, { scheduledFor })); - // Segment 1 pauses. Its writer observed epoch 0 (no resume yet). + // Segment 0 pauses. This mirrors the INITIAL pause exactly as request.js writes it: + // expectResumeSeq 0, against a row that has no resumeSeq field at all. await methods.recordRunOutcome({ scheduleId: schedule.id, scheduledFor, status: 'requires_action', conversationId: 'convo-1', autoDisableAfterFailures: 3, + expectResumeSeq: 0, }); expect((await getRun(schedule.id, scheduledFor)).status).toBe('requires_action'); @@ -1196,3 +1198,127 @@ describe('lifecycle barriers: stale pause vs resume, and concurrent resumes at c expect(await ScheduleRun.countDocuments({ status: 'started', capacitySlot: 0 })).toBe(0); }); }); + +describe('resume lease: duplicate attempts and crash recovery', () => { + const scheduledFor = new Date('2026-07-20T12:00:00Z'); + + async function pausedRun() { + const schedule = await methods.createSchedule(scheduleData()); + await methods.insertScheduleRun(runData(schedule, { scheduledFor, status: 'requires_action' })); + return schedule; + } + + it('a delayed duplicate resume cannot demote the winner after it committed', async () => { + const schedule = await pausedRun(); + + // Winner acquires, consumes the approval, reconstructs, commits. Commit clears the + // holder, so the row is an ordinary `started` run again. + const winner = await methods.acquireResumeLease({ + scheduleId: schedule.id, + scheduledFor, + holder: 'winner', + ttlMs: 60_000, + capacitySlot: 0, + }); + expect(winner.outcome).toBe('acquired'); + await methods.markResumeClaimed(schedule.id, scheduledFor, 'winner', 60_000); + expect(await methods.commitResumeLease(schedule.id, scheduledFor, 'winner')).toBe(true); + + // A delayed duplicate submit for the SAME action now arrives. It can still adopt the + // holderless row, but it will LOSE approvals.resolve and then release. + const loser = await methods.acquireResumeLease({ + scheduleId: schedule.id, + scheduledFor, + holder: 'loser', + ttlMs: 60_000, + capacitySlot: 0, + }); + expect(loser.outcome).toBe('acquired'); + await methods.releaseResumeLease(schedule.id, scheduledFor, 'loser'); + + // The winner's run must still be live. Demoting here would knock a running + // generation back to requires_action and free its capacity slot underneath it. + const row = await getRun(schedule.id, scheduledFor); + expect(row.status).toBe('started'); + expect(row.capacitySlot).toBe(0); + expect(row.resumeHolder).toBeUndefined(); + }); + + it('still rolls back a resume that genuinely promoted a paused run', async () => { + const schedule = await pausedRun(); + const lease = await methods.acquireResumeLease({ + scheduleId: schedule.id, + scheduledFor, + holder: 'solo', + ttlMs: 60_000, + capacitySlot: 0, + }); + expect(lease.outcome).toBe('acquired'); + + // Pre-claim failure: this attempt promoted the row, so release MUST demote it and + // free the slot, leaving the approval actionable. + expect(await methods.releaseResumeLease(schedule.id, scheduledFor, 'solo')).toBe(true); + const row = await getRun(schedule.id, scheduledFor); + expect(row.status).toBe('requires_action'); + expect(row.capacitySlot).toBeUndefined(); + }); + + it('a second live attempt cannot steal an unexpired lease', async () => { + const schedule = await pausedRun(); + await methods.acquireResumeLease({ + scheduleId: schedule.id, + scheduledFor, + holder: 'first', + ttlMs: 60_000, + capacitySlot: 0, + }); + const second = await methods.acquireResumeLease({ + scheduleId: schedule.id, + scheduledFor, + holder: 'second', + ttlMs: 60_000, + capacitySlot: 1, + }); + expect(second.outcome).toBe('held'); + }); + + it('an EXPIRED pre-claim lease is adoptable; a post-claim one is not', async () => { + const preClaim = await pausedRun(); + await methods.acquireResumeLease({ + scheduleId: preClaim.id, + scheduledFor, + holder: 'crashed', + ttlMs: -1_000, // already expired + capacitySlot: 0, + }); + // Never consumed the approval, so a retry may take it over. + const adopted = await methods.acquireResumeLease({ + scheduleId: preClaim.id, + scheduledFor, + holder: 'retry', + ttlMs: 60_000, + capacitySlot: 0, + }); + expect(adopted.outcome).toBe('acquired'); + + const postClaim = await pausedRun(); + await methods.acquireResumeLease({ + scheduleId: postClaim.id, + scheduledFor, + holder: 'crashed-2', + ttlMs: -1_000, + capacitySlot: 1, + }); + // The approval WAS consumed, so this must roll forward (reconciler terminalizes it) + // rather than be re-run by another attempt. + await methods.markResumeClaimed(postClaim.id, scheduledFor, 'crashed-2', -1_000); + const stolen = await methods.acquireResumeLease({ + scheduleId: postClaim.id, + scheduledFor, + holder: 'retry-2', + ttlMs: 60_000, + capacitySlot: 2, + }); + expect(stolen.outcome).toBe('held'); + }); +}); diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 616aa82051..63a56b0778 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -673,6 +673,13 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche resumeHolder: holder, resumeExpiresAt: { $add: ['$$NOW', ttlMs] }, capacitySlot, + // Whether this lease ADOPTED a row that was already `started` rather than + // promoting a paused one. Expressions in a pipeline $set see the PRE-stage + // document, so this observes the status before the line above changes it. + // Release must never demote an adopted row: a committed resume clears its + // holder, so a late duplicate attempt can otherwise adopt a LIVE running + // run and roll it back to requires_action. + resumeAdopted: { $eq: ['$status', 'started'] }, }, }, { $unset: ['resumeClaimedAt'] }, @@ -736,7 +743,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche ): Promise { const result = await ScheduleRun().updateOne( { scheduleId, scheduledFor, resumeHolder: holder }, - { $unset: { resumeHolder: 1, resumeExpiresAt: 1, resumeClaimedAt: 1 } }, + { $unset: { resumeHolder: 1, resumeExpiresAt: 1, resumeClaimedAt: 1, resumeAdopted: 1 } }, ); return (result.matchedCount ?? 0) > 0; } @@ -749,19 +756,38 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche scheduledFor: Date, holder: string, ): Promise { - const result = await ScheduleRun().updateOne( + // Demote ONLY a row this holder actually promoted out of `requires_action`. An + // ADOPTED row was already `started` (a committed resume, or one whose pause + // bookkeeping never landed), and demoting it would knock a live generation back to + // requires_action and free its capacity slot underneath it. + const promoted = await ScheduleRun().updateOne( + { + scheduleId, + scheduledFor, + resumeHolder: holder, + resumeClaimedAt: { $exists: false }, + resumeAdopted: { $ne: true }, + }, + { + $set: { status: 'requires_action' }, + $unset: { resumeHolder: 1, resumeExpiresAt: 1, capacitySlot: 1, resumeAdopted: 1 }, + }, + ); + if ((promoted.matchedCount ?? 0) > 0) { + return true; + } + // Adopted row: drop only OUR lease so another attempt can take it, leaving the run + // (and its slot) exactly as we found it. + const released = await ScheduleRun().updateOne( { scheduleId, scheduledFor, resumeHolder: holder, resumeClaimedAt: { $exists: false }, }, - { - $set: { status: 'requires_action' }, - $unset: { resumeHolder: 1, resumeExpiresAt: 1, capacitySlot: 1 }, - }, + { $unset: { resumeHolder: 1, resumeExpiresAt: 1, resumeAdopted: 1 } }, ); - return (result.matchedCount ?? 0) > 0; + return (released.matchedCount ?? 0) > 0; } /** Records that an abort was requested WITHOUT freeing the capacity slot: the run @@ -877,10 +903,15 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche // EPOCH CAS: a pause may only land on the segment that produced it. A resume // bumps resumeSeq, so a stale requires_action callback from the PREVIOUS segment // no longer matches and cannot demote the run the resume just promoted. - const epochFilter = - params.expectResumeSeq != null - ? { resumeSeq: params.expectResumeSeq } - : ({} as Record); + // `{field: null}` matches null OR missing in Mongo, so epoch 0 must be expressed + // as $in [0, null] — a never-resumed row has no resumeSeq at all, and a plain + // equality on 0 would never match it (silently disabling the fence). + let epochFilter: Record = {}; + if (params.expectResumeSeq === 0) { + epochFilter = { resumeSeq: { $in: [0, null] } }; + } else if (params.expectResumeSeq != null) { + epochFilter = { resumeSeq: params.expectResumeSeq }; + } const activeRun = await ScheduleRun() .findOne({ scheduleId: params.scheduleId, @@ -946,6 +977,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche resumeHolder: 1, resumeExpiresAt: 1, resumeClaimedAt: 1, + resumeAdopted: 1, }, }, ); diff --git a/packages/data-schemas/src/schema/scheduleRun.ts b/packages/data-schemas/src/schema/scheduleRun.ts index 26348fe221..68f4da4146 100644 --- a/packages/data-schemas/src/schema/scheduleRun.ts +++ b/packages/data-schemas/src/schema/scheduleRun.ts @@ -80,6 +80,11 @@ const scheduleRunSchema: Schema = new Schema( resumeClaimedAt: { type: Date, }, + /** True when the lease ADOPTED an already-`started` row instead of promoting a + * paused one. Release must not demote an adopted row (it may be a live run). */ + resumeAdopted: { + type: Boolean, + }, /** Global concurrency slot held while `started`. The unique partial index below * turns fireConcurrency into a DB-enforced bound instead of a racy count. */ capacitySlot: { diff --git a/packages/data-schemas/src/types/schedule.ts b/packages/data-schemas/src/types/schedule.ts index 28f2203d47..26c998c0b3 100644 --- a/packages/data-schemas/src/types/schedule.ts +++ b/packages/data-schemas/src/types/schedule.ts @@ -66,6 +66,8 @@ export interface IScheduleRun { resumeExpiresAt?: Date; /** Set once the approval claim succeeded (pre-claim vs post-claim recovery). */ resumeClaimedAt?: Date; + /** True when the lease adopted an already-`started` row rather than promoting a paused one. */ + resumeAdopted?: boolean; /** Global concurrency slot held while `started`. */ capacitySlot?: number; /** When an abort was requested; capacity is held until settlement is confirmed. */