From f46423235e9fe5ee054fcfdbf1b981a3d3d136ee Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 19:03:25 +0000 Subject: [PATCH] =?UTF-8?q?fix(schedules):=20Wave=20C.1=20=E2=80=94=20reco?= =?UTF-8?q?ncile=20durable=20trigger=20delivery=20with=20the=20reservation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Threads r3803826192 (manual limiter) and r3804255924 (PII/moderation), plus the independently-found long-Retry-After race. fireSchedule reserves a `started` run and a global capacity slot BEFORE the durable trigger delivery reaches the chat route, where an interactive limiter (manual Run Now), PII, or moderation can reject it before any generation job exists — dead-lettering the delivery while the run sat `started` until the 30-minute orphan sweep mislabeled it interrupted. And a valid delivery deferred by Retry-After (up to 24h) could be orphan-settled and have its capacity released, then fire anyway. Translate durable delivery state into the schedule outcome: - Store the deterministic trigger deliveryKey on the ScheduleRun reservation (computed from the envelope BEFORE enqueue, so an ambiguous commit still has it). - Add a getTriggerDelivery engine dep (wired to the merged trigger service's getDelivery) that reads the durable delivery by key. - Schedule reconciliation, for a jobless `started` run: staging/pending/leased → admission is live, never orphan; dead → record `error` from the durable lastError and release capacity promptly (no 30-minute wait), through the ordinary outcome/auto-disable path; succeeded or no record → the existing legacy orphan policy (interrupted only past the cutoff); a delivery lookup failure defers rather than orphaning a possibly-live delivery. Limiter/PII/ moderation middleware writes no schedule state. Adds reconcile state-mapping tests (dead/pending/leased/staging/succeeded/none/ lookup-failure) and a fire test that the reservation's deliveryKey equals the enqueued delivery's idempotency key. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq --- api/server/services/Schedules/index.js | 8 +- packages/api/src/schedules/engine.spec.ts | 79 +++++++++++++++++++ packages/api/src/schedules/engine.ts | 56 +++++++++++-- packages/api/src/schedules/fire.spec.ts | 26 ++++++ packages/api/src/schedules/fire.ts | 70 ++++++++++++---- packages/api/src/schedules/gate.spec.ts | 1 + packages/api/src/schedules/service.spec.ts | 4 + packages/api/src/schedules/service.ts | 5 ++ packages/api/src/schedules/types.ts | 15 +++- .../data-schemas/src/schema/scheduleRun.ts | 6 ++ packages/data-schemas/src/types/schedule.ts | 5 ++ 11 files changed, 250 insertions(+), 25 deletions(-) diff --git a/api/server/services/Schedules/index.js b/api/server/services/Schedules/index.js index 5944ad2c1b..72f76fa7dc 100644 --- a/api/server/services/Schedules/index.js +++ b/api/server/services/Schedules/index.js @@ -13,7 +13,10 @@ function getService() { const mongoose = require('mongoose'); const { createSchedulesService } = require('@librechat/api'); const { getAppConfig } = require('~/server/services/Config/app'); - const { enqueueAgentTrigger } = require('~/server/services/Agents/triggers'); + const { + enqueueAgentTrigger, + getAgentTriggerDelivery, + } = require('~/server/services/Agents/triggers'); const { resolveAgentFireAccess } = require('./access'); const methods = require('~/models'); const isUserDeleting = async (userId) => !(await methods.isAgentTriggerPrincipalActive(userId)); @@ -45,6 +48,9 @@ function getService() { { new: true }, ).lean(), enqueueAgentTrigger, + // Reconciliation reads the durable delivery to tell a still-live admission (a + // deferred Retry-After) or a dead-letter apart from a genuinely orphaned run. + getTriggerDelivery: getAgentTriggerDelivery, resolveAgentFireAccess, // Reuse the merged trigger/deletion admission fence. A schedule fire and every // generic trigger now fail closed on the same durable principal state. diff --git a/packages/api/src/schedules/engine.spec.ts b/packages/api/src/schedules/engine.spec.ts index 05e6d2ba4d..3865bffbf1 100644 --- a/packages/api/src/schedules/engine.spec.ts +++ b/packages/api/src/schedules/engine.spec.ts @@ -81,6 +81,7 @@ function makeDeps( hasScheduleAccess: async () => true, resolveFiles: async () => [], enqueueTrigger: jest.fn(async () => undefined), + getTriggerDelivery: async () => null, runInTenantContext: (_user, fn) => fn(), getJobStatus: async () => null, abortScheduledJob: async () => undefined, @@ -224,6 +225,84 @@ describe('runTick error handling', () => { }); }); +describe('reconciliation consults the durable trigger delivery', () => { + const YOUNG = () => new Date(Date.now() - 5 * 60_000); // past reconcile min, before orphan cutoff + const OLD = () => new Date(Date.now() - 60 * 60_000); // past the 30-minute orphan cutoff + + function joblessRun(firedAt: Date, deliveryKey: string | undefined = 'dk1') { + return { + scheduleId: 's1', + scheduledFor: new Date('2026-01-01T00:00:00.000Z'), + user: 'u1', + status: 'started', + conversationId: 'conv-1', + firedAt, + ...(deliveryKey != null ? { deliveryKey } : {}), + }; + } + + async function runReconcile( + run: ReturnType, + getTriggerDelivery: ScheduleEngineDeps['getTriggerDelivery'], + ) { + const methods = makeMethods(makeClaimedSchedule()); + (methods.getRunsForReconciliation as jest.Mock).mockResolvedValue([run]); + await tickOnce(makeDeps(methods, { getJobStatus: async () => null, getTriggerDelivery })); + return methods; + } + + it('settles a DEAD delivery as error promptly, before the 30-minute orphan age', async () => { + const methods = await runReconcile(joblessRun(YOUNG()), async () => ({ + status: 'dead', + lastError: 'rate limited', + })); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ scheduleId: 's1', status: 'error', error: 'rate limited' }), + ); + }); + + it('does NOT orphan a PENDING delivery past the cutoff (Retry-After may still fire it)', async () => { + const methods = await runReconcile(joblessRun(OLD()), async () => ({ status: 'pending' })); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('does NOT orphan a LEASED delivery past the cutoff', async () => { + const methods = await runReconcile(joblessRun(OLD()), async () => ({ status: 'leased' })); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('does NOT orphan a STAGING delivery past the cutoff', async () => { + const methods = await runReconcile(joblessRun(OLD()), async () => ({ status: 'staging' })); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('falls back to the legacy interrupted orphan for a SUCCEEDED delivery past the cutoff', async () => { + const methods = await runReconcile(joblessRun(OLD()), async () => ({ status: 'succeeded' })); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ scheduleId: 's1', status: 'interrupted' }), + ); + }); + + it('does not prematurely orphan a SUCCEEDED delivery before the cutoff', async () => { + const methods = await runReconcile(joblessRun(YOUNG()), async () => ({ status: 'succeeded' })); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('uses the legacy orphan policy when the delivery record is gone (null)', async () => { + const methods = await runReconcile(joblessRun(OLD()), async () => null); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ status: 'interrupted' }), + ); + }); + + it('DEFERS (stays reconcilable) when the delivery lookup fails', async () => { + const methods = await runReconcile(joblessRun(OLD()), async () => { + throw new Error('delivery store unavailable'); + }); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); +}); + describe('reconciliation is isolated per row', () => { /** * getRunsForReconciliation returns the OLDEST rows first, so a row that always throws diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 91e4253ca0..4a7680b895 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -218,20 +218,60 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { await clearRetainedJob(); continue; } - // A `started` run whose job is gone (jobStatus null) is an orphan. Every run - // records its conversationId up front, so getJobStatus above already - // liveness-checked it: a live long-running fire reads as `running` and is left - // alone. v1 is single-process, so a null job genuinely means gone. A fresh - // abort also DELETES the job (account-deletion quiesce), so absence carries - // the same abort fence as `aborted` above. + // A `started` run whose job is gone (jobStatus null). Every run records its + // conversationId up front, so getJobStatus above already liveness-checked it: a + // live long-running fire reads as `running` and is left alone. A fresh abort + // also DELETES the job (account-deletion quiesce), so absence carries the same + // abort fence as `aborted` above. + // + // But a jobless run is NOT automatically an orphan: fireSchedule reserves the + // run and its capacity BEFORE the durable trigger delivery reaches the chat + // route, and that delivery can be deferred (a `Retry-After` up to 24h, far past + // the 30-minute cutoff) or dead-lettered by a pre-generation rejection (an + // interactive limiter, PII, or moderation). Consult the durable delivery keyed + // by the reservation before deciding. if ( run.status === 'started' && jobStatus == null && - ageMs > ORPHAN_RUN_AGE_MS && !hasAbortInFlight(run, Date.now()) && !hasResumeHandoffInFlight(run, Date.now()) ) { - await finalize('interrupted'); + let delivery: Awaited> | null = null; + try { + delivery = run.deliveryKey ? await deps.getTriggerDelivery(run.deliveryKey) : null; + } catch (deliveryError) { + // Unknown is not gone: defer rather than orphan a possibly-live delivery. + logger.warn( + `[schedules] delivery lookup failed for ${run.scheduleId}@${run.scheduledFor?.toISOString?.() ?? run.scheduledFor}; deferring:`, + deliveryError, + ); + continue; + } + const deliveryStatus = delivery?.status; + if ( + deliveryStatus === 'staging' || + deliveryStatus === 'pending' || + deliveryStatus === 'leased' + ) { + // Admission is still live; the delivery will fire or dead-letter later. + continue; + } + if (deliveryStatus === 'dead') { + // Definitively failed before a generation ever existed. Record the error + // PROMPTLY from the durable lastError and release capacity through the + // ordinary outcome path — no need to wait the 30-minute orphan age. + await finalize( + 'error', + delivery?.lastError ?? 'Scheduled delivery failed before running', + ); + continue; + } + // `succeeded` (a generation ran; its own job/outcome evidence is the + // authority) or NO delivery record (a legacy fire): only a genuinely aged + // jobless run is an orphan. + if (ageMs > ORPHAN_RUN_AGE_MS) { + await finalize('interrupted'); + } continue; } if ( diff --git a/packages/api/src/schedules/fire.spec.ts b/packages/api/src/schedules/fire.spec.ts index 83f1d68ae8..18618fbdba 100644 --- a/packages/api/src/schedules/fire.spec.ts +++ b/packages/api/src/schedules/fire.spec.ts @@ -2,6 +2,7 @@ import type { ScheduleEngineDeps, ScheduleLimits, ScheduleUserContext } from './ import type { FireableSchedule } from './types'; import { AgentTriggerServiceUnavailableError } from '../agents/triggers/service'; import { AgentTriggerDeliveryError } from '../agents/triggers/delivery'; +import { getAgentTriggerIdempotencyKey } from '../agents/triggers/envelope'; import { buildFireClientRequestId, fireSchedule } from './fire'; import { withCapacitySlot } from './capacity'; @@ -86,6 +87,7 @@ function makeMethods() { scheduledFor: Date; conversationId?: string; capacitySlot?: number; + deliveryKey?: string; }) => { const k = key(data.scheduleId, data.scheduledFor); if (runs.has(k)) { @@ -284,6 +286,30 @@ describe('fireSchedule', () => { expect(methods.releaseLeaseByHolder).toHaveBeenCalledWith('sched-1', 'claim-holder-1'); }); + it('stamps the deterministic deliveryKey on the reservation (== the enqueued key)', async () => { + const { methods } = makeMethods(); + const enqueueTrigger = jest.fn< + ReturnType, + Parameters + >(async () => undefined); + + const result = await fireSchedule( + makeDeps(methods, { enqueueTrigger }), + makeSchedule(), + LIMITS, + dueAt(), + ); + + expect(result.fired).toBe(true); + const reservation = (methods.reserveStartedRun as jest.Mock).mock.calls[0][0]; + const envelope = enqueueTrigger.mock.calls[0][0]; + // Stored BEFORE enqueue and equal to the enqueued delivery's idempotency key, so + // reconciliation can read the durable delivery for THIS occurrence even if the enqueue + // committed ambiguously. + expect(reservation.deliveryKey).toEqual(expect.any(String)); + expect(reservation.deliveryKey).toBe(getAgentTriggerIdempotencyKey(envelope)); + }); + it('marks a Run Now trigger as manual for downstream limiter policy', async () => { const { methods } = makeMethods(); const enqueueTrigger = jest.fn< diff --git a/packages/api/src/schedules/fire.ts b/packages/api/src/schedules/fire.ts index cf05d10b9c..0799444f60 100644 --- a/packages/api/src/schedules/fire.ts +++ b/packages/api/src/schedules/fire.ts @@ -2,7 +2,12 @@ import { randomUUID } from 'node:crypto'; import { logger } from '@librechat/data-schemas'; import type { ScheduleEngineDeps, ScheduleLimits, FireResult, FireableSchedule } from './types'; import type { JsonValue } from '../agents/json'; -import { AgentTriggerEnvelopeError, createAgentTriggerEnvelope } from '../agents/triggers/envelope'; +import type { AgentTriggerEnvelope } from '../agents/triggers/envelope'; +import { + AgentTriggerEnvelopeError, + createAgentTriggerEnvelope, + getAgentTriggerIdempotencyKey, +} from '../agents/triggers/envelope'; import { AgentTriggerServiceUnavailableError } from '../agents/triggers/service'; import { AgentTriggerDeliveryError } from '../agents/triggers/delivery'; import { computeNextRunAt, cadenceIntervalMinutes } from './cadence'; @@ -32,15 +37,21 @@ class ScheduleFireError extends Error { } } -async function enqueueScheduleTrigger( - deps: ScheduleEngineDeps, +/** + * Builds the durable trigger envelope for one occurrence. Pure and deterministic in the + * fields that {@link getAgentTriggerIdempotencyKey} hashes (only `requestId`/`receivedAt` + * vary), so the `deliveryKey` derived from it is stable across retries and is stored on + * the reservation BEFORE enqueue — that lets schedule reconciliation read the durable + * delivery state even for an ambiguously-committed enqueue. + */ +function buildScheduleTriggerEnvelope( schedule: FireableSchedule, user: NonNullable>>, scheduledFor: Date, files: Awaited>, conversationId: string, manual: boolean, -): Promise { +): AgentTriggerEnvelope { const occurrenceId = buildFireClientRequestId(schedule.id, scheduledFor); const triggerFiles: JsonValue[] = files.map((file) => { const value: { [key: string]: JsonValue } = { file_id: file.file_id }; @@ -52,7 +63,7 @@ async function enqueueScheduleTrigger( if (file.source != null) value.source = file.source; return value; }); - const envelope = createAgentTriggerEnvelope({ + return createAgentTriggerEnvelope({ mode: 'fire', requestId: randomUUID(), deliveryId: occurrenceId, @@ -82,8 +93,15 @@ async function enqueueScheduleTrigger( }, }, }); +} + +async function enqueueScheduleTrigger( + deps: ScheduleEngineDeps, + envelope: AgentTriggerEnvelope, + orderingKey: string, +): Promise { try { - await deps.enqueueTrigger(envelope, { orderingKey: schedule.id }); + await deps.enqueueTrigger(envelope, { orderingKey }); } catch (error) { const message = error instanceof Error ? error.message : String(error); const definite = @@ -374,6 +392,35 @@ export async function fireSchedule( // rejects a second `started` run for the schedule, so a concurrent occurrence // surfaces as 'overlap' with no read-then-insert race. const conversationId = randomUUID(); + // Build the durable trigger envelope NOW so the deterministic deliveryKey can be + // stored on the reservation below — before enqueue. That way schedule reconciliation + // can read the delivery state (live vs dead) even when the enqueue commits ambiguously, + // instead of orphaning a still-live delivery after the 30-minute cutoff. A schedule + // whose envelope cannot be built can never fire, so skip WITHOUT reserving a phantom + // `started` run (a definite, retry-invariant failure). + let triggerEnvelope: AgentTriggerEnvelope; + try { + triggerEnvelope = buildScheduleTriggerEnvelope( + schedule, + user, + scheduledFor, + files, + conversationId, + options?.manual === true, + ); + } catch (envelopeError) { + logger.error(`[schedules] trigger envelope build failed for ${schedule.id}:`, envelopeError); + if (options?.manual) { + await releaseManualLease(); + } else { + await advance(); + } + return { + fired: false, + error: envelopeError instanceof Error ? envelopeError.message : String(envelopeError), + }; + } + const deliveryKey = getAgentTriggerIdempotencyKey(triggerEnvelope); // The GLOBAL fireConcurrency cap is enforced by claiming a unique capacity slot in // the SAME insert that reserves the run, so it is decided by the DB rather than by // a count read before the write. The allocator advances to the next free slot when @@ -395,6 +442,7 @@ export async function fireSchedule( conversationId, firedAt: new Date(), capacitySlot, + deliveryKey, ...(typeof schedule.configRevision === 'number' ? { configRevision: schedule.configRevision } : {}), @@ -500,15 +548,7 @@ export async function fireSchedule( } try { - await enqueueScheduleTrigger( - deps, - schedule, - user, - scheduledFor, - files, - conversationId, - options?.manual === true, - ); + await enqueueScheduleTrigger(deps, triggerEnvelope, schedule.id); } catch (error) { const message = error instanceof Error ? error.message : String(error); const ambiguous = error instanceof ScheduleFireError && error.ambiguous; diff --git a/packages/api/src/schedules/gate.spec.ts b/packages/api/src/schedules/gate.spec.ts index c4ab51301a..6d2199707f 100644 --- a/packages/api/src/schedules/gate.spec.ts +++ b/packages/api/src/schedules/gate.spec.ts @@ -34,6 +34,7 @@ function makeService(base: Cfg, merged: Cfg = base) { resolveAgentFireAccess: jest.fn(async () => 'ok' as const), isUserDeleting: jest.fn(async () => false), enqueueAgentTrigger: jest.fn(async () => undefined), + getTriggerDelivery: jest.fn(async () => null), } as unknown as SchedulesServiceDeps; return createSchedulesService(deps); } diff --git a/packages/api/src/schedules/service.spec.ts b/packages/api/src/schedules/service.spec.ts index eea132bf3b..59b230be46 100644 --- a/packages/api/src/schedules/service.spec.ts +++ b/packages/api/src/schedules/service.spec.ts @@ -65,6 +65,7 @@ function makeService( resolveAgentFireAccess: jest.fn(async () => 'ok' as const), isUserDeleting: jest.fn(async () => false), enqueueAgentTrigger, + getTriggerDelivery: jest.fn(async () => null), } as unknown as SchedulesServiceDeps; // Short bounded waits so fail-closed drain paths resolve in test time. return createSchedulesService(deps, { @@ -166,6 +167,7 @@ describe('balance initialization', () => { resolveAgentFireAccess: jest.fn(async () => 'ok' as const), isUserDeleting: jest.fn(async () => false), enqueueAgentTrigger: jest.fn(async () => undefined), + getTriggerDelivery: jest.fn(async () => null), } as unknown as SchedulesServiceDeps); return { service, upsertBalance, initializeNullBalance, findBalance }; } @@ -276,6 +278,7 @@ describe('balance initialization', () => { resolveAgentFireAccess: jest.fn(async () => 'ok' as const), isUserDeleting: jest.fn(async () => false), enqueueAgentTrigger: jest.fn(async () => undefined), + getTriggerDelivery: jest.fn(async () => null), } as unknown as SchedulesServiceDeps); await service.engineDeps.isOutOfBalance({ id: 'user-1' } as never); @@ -1210,6 +1213,7 @@ describe('scheduled resume capacity', () => { resolveAgentFireAccess: jest.fn(async () => 'ok' as const), isUserDeleting: jest.fn(async () => false), enqueueAgentTrigger: jest.fn(async () => undefined), + getTriggerDelivery: jest.fn(async () => null), } as unknown as SchedulesServiceDeps); return { service, methods }; } diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 11384da6f0..8a4572ea0c 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -164,6 +164,9 @@ export interface SchedulesServiceDeps { isUserDeleting: (userId: string) => Promise; /** Shared durable trigger admission from the merged agent-trigger service. */ enqueueAgentTrigger: ScheduleEngineDeps['enqueueTrigger']; + /** Reads a durable trigger delivery by key from the merged agent-trigger service, for + * reconciliation's live-vs-dead delivery check. */ + getTriggerDelivery: ScheduleEngineDeps['getTriggerDelivery']; } export interface SchedulesService { @@ -292,6 +295,7 @@ export function createSchedulesService( 'resolveAgentFireAccess', 'isUserDeleting', 'enqueueAgentTrigger', + 'getTriggerDelivery', ]; for (const key of REQUIRED_DEPS) { if (deps[key] == null) { @@ -501,6 +505,7 @@ export function createSchedulesService( })); }, enqueueTrigger: deps.enqueueAgentTrigger, + getTriggerDelivery: deps.getTriggerDelivery, runInTenantContext: (user, fn) => tenantStorage.run({ tenantId: user.tenantId, userId: user.id }, fn), getJobStatus: async (conversationId) => { diff --git a/packages/api/src/schedules/types.ts b/packages/api/src/schedules/types.ts index 1e81a24c98..b3aa088f66 100644 --- a/packages/api/src/schedules/types.ts +++ b/packages/api/src/schedules/types.ts @@ -1,4 +1,8 @@ -import type { ScheduleMethods, ISchedule } from '@librechat/data-schemas'; +import type { + ScheduleMethods, + ISchedule, + AgentTriggerDeliveryStatus, +} from '@librechat/data-schemas'; import type { Types } from 'mongoose'; import type { AgentTriggerEnqueueOptions, AgentTriggerEnvelope } from '../agents/triggers'; import type { SlotClaimResult } from './capacity'; @@ -137,6 +141,15 @@ export interface ScheduleEngineDeps { envelope: AgentTriggerEnvelope, options?: AgentTriggerEnqueueOptions, ) => Promise; + /** + * Reads the durable trigger delivery for a reservation's `deliveryKey`, so + * reconciliation can tell a still-live admission (`staging`/`pending`/`leased`, which a + * `Retry-After` can defer up to 24h) or a dead-letter (`dead`, with its `lastError`) + * apart from a genuinely orphaned jobless run. Null when no delivery record exists. + */ + getTriggerDelivery: ( + deliveryKey: string, + ) => Promise<{ status: AgentTriggerDeliveryStatus; lastError?: string } | null>; /** Runs fn inside the owner's tenant ALS context. */ runInTenantContext: (user: ScheduleUserContext, fn: () => Promise) => Promise; /** diff --git a/packages/data-schemas/src/schema/scheduleRun.ts b/packages/data-schemas/src/schema/scheduleRun.ts index b6e92f7901..cf5e51a9a0 100644 --- a/packages/data-schemas/src/schema/scheduleRun.ts +++ b/packages/data-schemas/src/schema/scheduleRun.ts @@ -29,6 +29,12 @@ const scheduleRunSchema: Schema = new Schema( conversationId: { type: String, }, + /** Deterministic durable-trigger delivery key for this occurrence, stamped at + * reservation before enqueue so reconciliation can read the delivery's live/dead + * state rather than orphan-settling a deferred or dead-lettered run. */ + deliveryKey: { + type: String, + }, /** Fresh while a RESUME of this paused run is mid-flight; a re-pause hand-off's * writes are still landing, so quiesce must not settle on the paused job state. * Cleared by the pause record (the hand-off completion) or aged out. */ diff --git a/packages/data-schemas/src/types/schedule.ts b/packages/data-schemas/src/types/schedule.ts index fa3e578751..e806a82a69 100644 --- a/packages/data-schemas/src/types/schedule.ts +++ b/packages/data-schemas/src/types/schedule.ts @@ -83,6 +83,11 @@ export interface IScheduleRun { conversationId?: string; status: ScheduleRunStatus; error?: string; + /** Deterministic durable-trigger delivery key for this occurrence, stamped at + * reservation (before enqueue). Lets reconciliation read the delivery's live/dead + * state instead of orphan-settling a jobless run that is merely deferred (Retry-After) + * or that dead-lettered before a generation ever started. */ + deliveryKey?: string; droppedFileIds?: string[]; durationMs?: number; bookkept?: boolean;