diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index d79043b3e4..92dc2f280b 100644 --- a/api/server/controllers/agents/request.js +++ b/api/server/controllers/agents/request.js @@ -414,12 +414,12 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit status: 'interrupted', conversationId: streamId, }); - // If the outcome write failed (transient Mongo across its retries), preserve - // the job so the reconciler can finalize this run instead of deleting the - // only evidence while the run row is still `started` (which would leave a - // deleted schedule draining until the orphan cutoff). Mirrors the other - // scheduled terminal paths. - await GenerationJobManager.completeJob(streamId, undefined, { + // Terminalize as ABORTED, not complete: if the outcome write failed a + // preserved `complete` job would be mapped to `success` by the schedules + // reconciler, mislabeling this pre-start abort as a successful run. abortJob + // stores it as `aborted` (reconcile -> interrupted), and preserves it for + // reconcile only when the outcome write failed so the evidence survives. + await GenerationJobManager.abortJob(streamId, { preserveForReconcile: !outcomeRecorded, }).catch(() => undefined); return res.json({ streamId, conversationId, status: 'aborted' }); diff --git a/client/src/components/SidePanel/Schedules/cadence.ts b/client/src/components/SidePanel/Schedules/cadence.ts index 3f381f553e..cbcb3dde22 100644 --- a/client/src/components/SidePanel/Schedules/cadence.ts +++ b/client/src/components/SidePanel/Schedules/cadence.ts @@ -5,6 +5,10 @@ export type Meridiem = 'AM' | 'PM'; const DAY_MS = 24 * 60 * 60 * 1000; +/** Mirrors the server's default weekly day (Monday) when a weekly cadence omits + * daysOfWeek, so an API-created/migrated `frequency: 'weekly'` renders as weekly. */ +const WEEKLY_DEFAULT_DAY = 1; + /** August 1st, 2021 was a Sunday; anchors day-of-week indices 0-6 to real dates */ const SUNDAY_UTC = Date.UTC(2021, 7, 1); @@ -42,8 +46,12 @@ export const describeCadence = ( if (frequency === 'weekdays') { return localize('com_ui_schedule_runs_weekdays', { time }); } - if (frequency === 'weekly' && daysOfWeek != null && daysOfWeek.length > 0) { - const days = daysOfWeek.map((day) => formatScheduleDay(day, locale)).join(', '); + if (frequency === 'weekly') { + // A weekly cadence with no daysOfWeek is valid — the server fires it on the + // default weekly day — so render it as weekly (not daily) using that same day. + const effectiveDays = + daysOfWeek != null && daysOfWeek.length > 0 ? daysOfWeek : [WEEKLY_DEFAULT_DAY]; + const days = effectiveDays.map((day) => formatScheduleDay(day, locale)).join(', '); return localize('com_ui_schedule_runs_weekly', { days, time }); } return localize('com_ui_schedule_runs_daily', { time }); diff --git a/packages/api/src/schedules/fire.spec.ts b/packages/api/src/schedules/fire.spec.ts index 69f43fb358..d387643b9d 100644 --- a/packages/api/src/schedules/fire.spec.ts +++ b/packages/api/src/schedules/fire.spec.ts @@ -90,6 +90,7 @@ function makeMethods() { ), revalidateClaim: jest.fn(async () => true), holdsLease: jest.fn(async () => true), + releaseLeaseByHolder: jest.fn(async () => undefined), deleteScheduleRun: jest.fn(async (id: string, when: Date, _status?: string) => { runs.delete(key(id, when)); }), diff --git a/packages/api/src/schedules/fire.ts b/packages/api/src/schedules/fire.ts index 10baa88a56..eee0e37730 100644 --- a/packages/api/src/schedules/fire.ts +++ b/packages/api/src/schedules/fire.ts @@ -316,6 +316,15 @@ export async function fireSchedule( !(await methods.revalidateClaim(schedule.id, claimToken, !options?.manual)) ) { await rollbackReservation(); + // This fire is superseded (owner edit/delete). advance() is fenced on the OLD + // claim token, which the edit rotated, so it no-ops and would leave this + // worker's lease held until its TTL — reporting the edited schedule / Run now + // as "already in progress" though no run was dispatched. Release our own lease + // by holder (leaseBy) so it's immediately re-claimable; a takeover changed + // leaseBy, so this correctly no-ops there and never strips the new holder's lease. + if (schedule.leaseBy != null) { + await methods.releaseLeaseByHolder(schedule.id, schedule.leaseBy); + } await advance(); return { fired: false, skipped: 'superseded' as const }; } diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 8c756fbc99..9150ad1fc8 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -476,18 +476,22 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer if (schedule == null) { return 'gone'; } + const when = new Date(scheduledFor); // Overlap = a DIFFERENT occurrence is currently `started`. Exclude this run's own // row: if its pause bookkeeping failed transiently the row can still be `started`, // and treating that as an overlap would wrongly reject resuming THIS same occurrence. - if (await methods.hasOtherActiveRun(scheduleId, new Date(scheduledFor))) { + if (await methods.hasOtherActiveRun(scheduleId, when)) { return 'overlap'; } // Read-only capacity gate BEFORE promoting, so we never mutate a row a concurrent - // same-pause resume may already be driving (no rollback path exists). + // same-pause resume may already be driving (no rollback path exists). Discount + // this occurrence's OWN `started` row when present (a transient pause-bookkeeping + // failure): resuming it adds no new active run, so the global count already + // includes it and must not block the resume. const owner = await engineDeps.getUserContext(schedule.user); const limits = await getLimits(owner ?? undefined); - const active = await engineDeps.countActiveRunsGlobal(); - if (active >= limits.fireConcurrency) { + const selfActive = await methods.isOccurrenceStarted(scheduleId, when); + if (!selfActive && (await engineDeps.countActiveRunsGlobal()) >= limits.fireConcurrency) { return 'capacity'; } // Reserve the single active slot. If a different occurrence won the slot since @@ -496,7 +500,7 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer // occurrence with the paused row still `requires_action` (which overlap/capacity // accounting would miss). 'missing' means a concurrent same-pause resume already // promoted it — proceed. Never rolled back. - const promoted = await methods.promoteRunToStarted(scheduleId, new Date(scheduledFor)); + const promoted = await methods.promoteRunToStarted(scheduleId, when); if (promoted === 'overlap') { return 'overlap'; } diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index 4da901532a..f5df9a86b8 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -945,6 +945,24 @@ describe('deletion quiescing (soft-delete, drain, erase)', () => { expect(await methods.claimDueSchedule({ instanceId: 'w1', leaseMs: 60_000 })).toBeNull(); }); + it('keeps a live lease and does not erase until it is released (worker mid-claim)', async () => { + const schedule = await methods.createScheduleWithSlot(scheduleData(), 10); + const sched = schedule as ISchedule; + // A worker has CLAIMED the schedule (live lease) but not yet inserted a run row. + const claim = await methods.claimDueSchedule({ instanceId: 'w1', leaseMs: 60_000 }); + expect(claim?.id).toBe(sched.id); + const marked = await methods.markScheduleDeleting(sched.id, sched.user); + // The lease is PRESERVED so the worker can prove ownership on its rollback. + expect(marked?.leaseBy).toBe('w1'); + // Erase is blocked while the lease is live, even with no active run row. + expect(await methods.eraseScheduleIfDrained(sched.id)).toBe(false); + expect(await Schedule.findOne({ id: sched.id }).lean()).not.toBeNull(); + // Once the worker releases its own lease (by holder), it drains and erases. + await methods.releaseLeaseByHolder(sched.id, 'w1'); + expect(await methods.eraseScheduleIfDrained(sched.id)).toBe(true); + expect(await Schedule.findOne({ id: sched.id }).lean()).toBeNull(); + }); + it('frees the slot immediately on soft-delete so a new create can take it under the cap', async () => { const user = new mongoose.Types.ObjectId(); const a = (await methods.createScheduleWithSlot(scheduleData({ user }), 1)) as ISchedule; diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 2c48168cf7..f61b91090c 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -95,9 +95,11 @@ export type ScheduleMethods = { leaseMs: number, ) => Promise; releaseLease: (id: string, expectedClaimToken?: string) => Promise; + releaseLeaseByHolder: (id: string, leaseBy: string) => Promise; revalidateClaim: (id: string, claimToken: string, requireEnabled?: boolean) => Promise; holdsLease: (id: string, leaseBy: string) => Promise; hasOtherActiveRun: (scheduleId: string, scheduledFor: Date) => Promise; + isOccurrenceStarted: (scheduleId: string, scheduledFor: Date) => Promise; advanceSchedule: ( id: string, nextRunAt: Date | null, @@ -357,6 +359,18 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche await Schedule().updateOne(filter, { $unset: { leaseUntil: 1, leaseBy: 1 } }); } + /** + * Releases a lease fenced on the lease HOLDER (`leaseBy`) rather than the claim + * token. Used when a fire is superseded by an owner edit that rotated the token + * (so a token-fenced release would no-op): the worker still owns the lease, so it + * must clear it — otherwise the edited schedule (and Run now) is reported "already + * in progress" until the lease TTL, even though no run was dispatched. A takeover + * changed `leaseBy`, so this correctly no-ops and never strips the new holder's lease. + */ + async function releaseLeaseByHolder(id: string, leaseBy: string): Promise { + await Schedule().updateOne({ id, leaseBy }, { $unset: { leaseUntil: 1, leaseBy: 1 } }); + } + /** * Whether the caller still holds an authoritative claim on the schedule: it is * not being deleted, its claim token is unchanged, and its lease has not expired @@ -419,6 +433,20 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche return row != null; } + /** + * Whether THIS occurrence's own run row is already `started`. Lets the HITL resume + * capacity gate discount a self-active row (e.g. one whose pause bookkeeping failed + * transiently): resuming it adds no new active run, so it must not be blocked by a + * global count that already includes it. + */ + async function isOccurrenceStarted(scheduleId: string, scheduledFor: Date): Promise { + const row = await ScheduleRun() + .findOne({ scheduleId, scheduledFor, status: 'started' }) + .select('_id') + .lean(); + return row != null; + } + /** * Advances past a fired (or skipped) occurrence and releases the lease. When * `expectedNextRunAt` is given, the update is predicated on the schedule still @@ -807,12 +835,18 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche id: string, userId: string | Types.ObjectId, ): Promise { + // Keep leaseUntil/leaseBy: a fire that already leased/reserved this occurrence + // must be able to prove (holdsLease) it still owns the lease so it can roll back + // its own unposted `started` row on the superseded revalidation. Unsetting the + // lease here would fail that check and strand a ghost `started` row. Only clear + // nextRunAt (belt-and-suspenders atop enabled:false to stop new claims); the + // lease releases itself when the fire finishes its rollback, or via TTL. return Schedule() .findOneAndUpdate( { id, user: userId, deleting: { $ne: true } }, { $set: { enabled: false, deleting: true, claimToken: randomUUID() }, - $unset: { leaseUntil: 1, leaseBy: 1, nextRunAt: 1 }, + $unset: { nextRunAt: 1 }, }, { new: true }, ) @@ -854,12 +888,27 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche } /** - * Erases a soft-deleted schedule and its runs ONLY once no active run remains, - * so a live loopback generation's evidence is never destroyed out from under - * it. The schedule is already disabled + `deleting`, so no new run can start; - * once none are active none will become active. Returns whether it erased. + * Erases a soft-deleted schedule and its runs ONLY once it has fully drained, so a + * live loopback generation's evidence is never destroyed out from under it. Drained + * means BOTH: (a) no run is active, and (b) no LIVE lease is held. The lease check + * is essential — a worker can have CLAIMED the schedule but not yet inserted its + * `started` reservation (or be mid-rollback of one); erasing in that window would + * let the worker then insert a ghost row against a gone schedule that it can no + * longer prove it owns. Returns whether it erased. */ async function eraseScheduleIfDrained(id: string): Promise { + // A live lease (leaseUntil > $$NOW) means a worker still holds the claim. + const leased = await Schedule() + .findOne({ + id, + deleting: true, + $expr: { $gt: [{ $ifNull: ['$leaseUntil', new Date(0)] }, '$$NOW'] }, + }) + .select('_id') + .lean(); + if (leased != null) { + return false; + } const active = await ScheduleRun() .findOne({ scheduleId: id, status: { $in: ACTIVE_RUN_STATUSES } }) .select('_id') @@ -907,9 +956,11 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche claimDueSchedule, acquireManualRunLease, releaseLease, + releaseLeaseByHolder, revalidateClaim, holdsLease, hasOtherActiveRun, + isOccurrenceStarted, advanceSchedule, disableSchedule, insertScheduleRun,