feat: bound scheduler catch-up dispatch (AEG-V15-034)

Implements LATEST_ONLY, SKIP_MISSED, and ALL_WITH_LIMIT dispatch plans anchored to scheduledFor. Evidence: targeted Release tests 4/4 passed; TRX SHA256 DC28BE4F2FCF511D5859B9FC3A0ADDF8CE3A566262C9848F05B06D825EA944AD. Schedules remain disabled; DEC-083 is not resolved.

src/KArtSell.Modules.ModelOperations/Application/ModelOperationsContracts.cs

@@ -54,6 +54,13 @@ public interface IModelScheduleRepository
         DateTimeOffset nextDueAt,
         CancellationToken cancellationToken);
 
+    Task AdvanceWithoutDispatchAsync(
+        Guid scheduleId,
+        string leaseOwner,
+        DateTimeOffset advancedAt,
+        DateTimeOffset nextDueAt,
+        CancellationToken cancellationToken);
+
     Task ReleaseAsync(
         Guid scheduleId,
         string leaseOwner,

src/KArtSell.Modules.ModelOperations/Domain/ScheduleOccurrencePlanner.cs

@@ -2,29 +2,56 @@ namespace KArtSell.Modules.ModelOperations.Domain;
 
 public sealed class ScheduleOccurrencePlanner
 {
+    public sealed record CatchUpPlan(
+        IReadOnlyList<DateTimeOffset> OccurrencesToDispatch,
+        DateTimeOffset NextDueAt);
+
     public static DateTimeOffset GetNextDueAt(
         DateTimeOffset scheduledFor,
         string cadence,
         string catchUpPolicy,
         int maxCatchUp,
         DateTimeOffset now)
+        => Plan(scheduledFor, cadence, catchUpPolicy, maxCatchUp, now).NextDueAt;
+
+    public static CatchUpPlan Plan(
+        DateTimeOffset scheduledFor,
+        string cadence,
+        string catchUpPolicy,
+        int maxCatchUp,
+        DateTimeOffset now)
     {
         if (maxCatchUp is < 0 or > 31) throw new ArgumentOutOfRangeException(nameof(maxCatchUp));
-        var next = Advance(scheduledFor, cadence);
+
+        var due = GetDueOccurrences(scheduledFor, cadence, now);
+        var nextDueAt = due.Count == 0 ? scheduledFor : Advance(due[^1], cadence);
+
+        if (catchUpPolicy.Equals("LATEST_ONLY", StringComparison.OrdinalIgnoreCase))
+            return new CatchUpPlan(due.Count == 0 ? [scheduledFor] : [due[^1]], nextDueAt);
+
+        if (catchUpPolicy.Equals("SKIP_MISSED", StringComparison.OrdinalIgnoreCase))
+            return new CatchUpPlan(scheduledFor < now ? [] : [scheduledFor], nextDueAt);
+
         if (catchUpPolicy.Equals("ALL_WITH_LIMIT", StringComparison.OrdinalIgnoreCase))
-            return next;
+            return new CatchUpPlan(due.Count <= maxCatchUp ? due : due.TakeLast(maxCatchUp).ToArray(), nextDueAt);
 
-        if (!catchUpPolicy.Equals("LATEST_ONLY", StringComparison.OrdinalIgnoreCase)
-            && !catchUpPolicy.Equals("SKIP_MISSED", StringComparison.OrdinalIgnoreCase))
-            throw new InvalidOperationException($"Unsupported catch-up policy '{catchUpPolicy}'.");
+        throw new InvalidOperationException($"Unsupported catch-up policy '{catchUpPolicy}'.");
+    }
 
+    private static IReadOnlyList<DateTimeOffset> GetDueOccurrences(DateTimeOffset scheduledFor, string cadence, DateTimeOffset now)
+    {
+        if (scheduledFor > now) return [];
+
+        var due = new List<DateTimeOffset>();
+        var occurrence = scheduledFor;
         var guard = 0;
-        while (next <= now)
+        while (occurrence <= now)
         {
-            next = Advance(next, cadence);
+            due.Add(occurrence);
+            occurrence = Advance(occurrence, cadence);
             if (++guard > 5000) throw new InvalidOperationException("Schedule catch-up exceeded safety guard.");
         }
-        return next;
+        return due;
     }
 
     private static DateTimeOffset Advance(DateTimeOffset value, string cadence) => cadence.ToUpperInvariant() switch

src/KArtSell.Modules.ModelOperations/Infrastructure/DapperModelScheduleRepository.cs

@@ -59,6 +59,16 @@ public sealed class DapperModelScheduleRepository(IDbConnectionFactory connectio
         where schedule_id = @ScheduleId and lease_owner = @LeaseOwner;
         """;
 
+    private const string AdvanceSql = """
+        update evaluation.model_operation_schedule
+        set next_due_at = @NextDueAt,
+            lease_owner = null,
+            lease_until = null,
+            last_error_code = null,
+            updated_at = @AdvancedAt
+        where schedule_id = @ScheduleId and lease_owner = @LeaseOwner;
+        """;
+
     public async Task<IReadOnlyList<DueModelOperation>> AcquireDueAsync(DateTimeOffset now, string leaseOwner, TimeSpan leaseDuration, int limit, CancellationToken cancellationToken)
     {
         await using var connection = await connectionFactory.OpenAsync(cancellationToken);
@@ -73,6 +83,13 @@ public sealed class DapperModelScheduleRepository(IDbConnectionFactory connectio
         if (affected != 1) throw new InvalidOperationException("Schedule lease was lost before dispatch completion.");
     }
 
+    public async Task AdvanceWithoutDispatchAsync(Guid scheduleId, string leaseOwner, DateTimeOffset advancedAt, DateTimeOffset nextDueAt, CancellationToken cancellationToken)
+    {
+        await using var connection = await connectionFactory.OpenAsync(cancellationToken);
+        var affected = await connection.ExecuteAsync(new CommandDefinition(AdvanceSql, new { ScheduleId = scheduleId, LeaseOwner = leaseOwner, AdvancedAt = advancedAt, NextDueAt = nextDueAt }, cancellationToken: cancellationToken));
+        if (affected != 1) throw new InvalidOperationException("Schedule lease was lost before advance completion.");
+    }
+
     public async Task ReleaseAsync(Guid scheduleId, string leaseOwner, string reasonCode, DateTimeOffset releasedAt, CancellationToken cancellationToken)
     {
         await using var connection = await connectionFactory.OpenAsync(cancellationToken);

src/KArtSell.Modules.ModelOperations/Scheduling/ModelOperationsDispatcherJob.cs

@@ -33,21 +33,32 @@ public sealed class ModelOperationsDispatcherJob(
         {
             try
             {
-                var job = Job.FromExpression<ScheduledModelOperationJob>(handler => handler.ExecuteAsync(
-                    item.ScheduleId,
-                    item.OperationCode,
-                    item.ScopeKey,
-                    item.AutomationMode,
-                    item.IdempotencyKey));
-                var backgroundJobId = jobs.Create(job, new EnqueuedState(item.Queue));
+                var plan = ScheduleOccurrencePlanner.Plan(item.ScheduledFor, item.Cadence, item.CatchUpPolicy, item.MaxCatchUp, now);
+                if (plan.OccurrencesToDispatch.Count == 0)
+                {
+                    await schedules.AdvanceWithoutDispatchAsync(item.ScheduleId, leaseOwner, now, plan.NextDueAt, CancellationToken.None);
+                    continue;
+                }
+
+                var backgroundJobId = string.Empty;
+                foreach (var occurrence in plan.OccurrencesToDispatch)
+                {
+                    var idempotencyKey = $"{item.OperationCode}:{item.ScopeKey}:{item.ScheduleVersion}:{occurrence.UtcDateTime:yyyyMMddHHmmss}Z";
+                    var job = Job.FromExpression<ScheduledModelOperationJob>(handler => handler.ExecuteAsync(
+                        item.ScheduleId,
+                        item.OperationCode,
+                        item.ScopeKey,
+                        item.AutomationMode,
+                        idempotencyKey));
+                    backgroundJobId = jobs.Create(job, new EnqueuedState(item.Queue));
+                }
 
-                var nextDueAt = ScheduleOccurrencePlanner.GetNextDueAt(item.ScheduledFor, item.Cadence, item.CatchUpPolicy, item.MaxCatchUp, now);
                 await schedules.MarkDispatchedAsync(
                     item.ScheduleId,
                     leaseOwner,
                     backgroundJobId,
                     now,
-                    nextDueAt,
+                    plan.NextDueAt,
                     CancellationToken.None);
             }
             catch (Exception exception)