fix(backend): server-bound retention fanout read to per-run cap

The retention fanout materialized the entire eligible-session backlog every
run (GetSessionsOlderThanAsync returned a full List<SessionSummary>) but only
advanced MaxEnqueuesPerTenantPerRun=100 per tenant. A large backlog was reread
and rematerialized each run, so maintenance got slower as the backlog grew
while only 100 sessions progressed.

Cap the read at the per-run dispatch limit: GetSessionsOlderThanAsync gains a
maxResults parameter that caps maxPerPage and stops enumerating once the cap is
reached. The fanout fetches MaxEnqueuesPerTenantPerRun+1 (the +1 is a cheap
"more remaining" probe) instead of the whole backlog. Read cost is now
proportional to the cap, not the backlog; per-run progress is unchanged.

The exact eligible count is no longer known without a full scan, so the audit
now records EligibleThisRun + a MoreRemaining flag instead of the exact total.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/Backend/AutopilotMonitor.Functions.Tests/OffboardingSessionEnumeratorTests.cs

@@ -109,7 +109,7 @@ public class OffboardingSessionEnumeratorTests
         public Task<RawPage<AuditLogEntry>> GetAuditLogsPageAsync(string tenantId, DateTime? dateFrom, DateTime? dateTo, int pageSize, string? continuation, bool excludeDeletions = false, AuditLogQueryFilters? filters = null) => throw new NotImplementedException();
         public Task<RawPage<AuditLogEntry>> GetAllAuditLogsPageAsync(DateTime? dateFrom, DateTime? dateTo, int pageSize, string? continuation, bool excludeDeletions = false, AuditLogQueryFilters? filters = null) => throw new NotImplementedException();
         public Task<int> DeleteAuditLogsOlderThanAsync(DateTime cutoffUtc) => throw new NotImplementedException();
-        public Task<List<SessionSummary>> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate) => throw new NotImplementedException();
+        public Task<List<SessionSummary>> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate, int maxResults = int.MaxValue) => throw new NotImplementedException();
         public Task<List<SessionSummary>> GetSessionsByDateRangeAsync(DateTime startDate, DateTime endDate, string? tenantId = null) => throw new NotImplementedException();
         public Task<List<SessionSummary>> GetStalledSessionsAsync(string tenantId, DateTime cutoffTime) => throw new NotImplementedException();
         public Task<List<SessionSummary>> GetAgentSilentSessionsAsync(string tenantId, DateTime silenceCutoff, DateTime hardCutoff) => throw new NotImplementedException();

src/Backend/AutopilotMonitor.Functions.Tests/SessionRetentionFanoutServiceTests.cs

@@ -50,9 +50,9 @@ public class SessionRetentionFanoutServiceTests
         harness.WithTenant(TenantB, retentionDays: 120, sessions: new[] { Old("b1", 45), Old("b2", 200) });
 
         // The repo only returns sessions older than each cutoff. Wire each tenant's mock to honor that.
-        harness.MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(TenantA, It.IsAny<DateTime>()))
+        harness.MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(TenantA, It.IsAny<DateTime>(), It.IsAny<int>()))
             .ReturnsAsync(new List<SessionSummary> { Summary(TenantA, "a1") });
-        harness.MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(TenantB, It.Is<DateTime>(d => d <= DateTime.UtcNow.AddDays(-120))))
+        harness.MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(TenantB, It.Is<DateTime>(d => d <= DateTime.UtcNow.AddDays(-120)), It.IsAny<int>()))
             .ReturnsAsync(new List<SessionSummary> { Summary(TenantB, "b2") });
 
         var result = await harness.Sut.RunAsync(CancellationToken.None);
@@ -84,6 +84,25 @@ public class SessionRetentionFanoutServiceTests
             Times.Never);
     }
 
+    [Fact]
+    public async Task RunAsync_reads_session_backlog_server_bounded_to_cap_plus_one()
+    {
+        // Codex HIGH fix: the fanout must NOT materialize the whole backlog. It reads at most
+        // MaxEnqueuesPerTenantPerRun+1 (the +1 is the "more remaining" probe) so a large backlog
+        // is no longer reread+rematerialized every run while only 100 sessions advance.
+        var harness = new Harness();
+        harness.WithTenant(TenantA, retentionDays: 30, sessions: new[] { Old("s1", 60) });
+
+        await harness.Sut.RunAsync(CancellationToken.None);
+
+        harness.MaintenanceRepo.Verify(
+            m => m.GetSessionsOlderThanAsync(
+                TenantA,
+                It.IsAny<DateTime>(),
+                SessionRetentionFanoutService.MaxEnqueuesPerTenantPerRun + 1),
+            Times.Once);
+    }
+
     [Fact]
     public async Task RunAsync_skips_tenant_with_DataRetentionDays_zero()
     {
@@ -94,7 +113,7 @@ public class SessionRetentionFanoutServiceTests
 
         Assert.Equal(0, result.SessionsEnqueued);
         // GetSessionsOlderThanAsync must not have been called for that tenant.
-        harness.MaintenanceRepo.Verify(m => m.GetSessionsOlderThanAsync(TenantA, It.IsAny<DateTime>()), Times.Never);
+        harness.MaintenanceRepo.Verify(m => m.GetSessionsOlderThanAsync(TenantA, It.IsAny<DateTime>(), It.IsAny<int>()), Times.Never);
     }
 
     // ────────────────────────────────────────────────────────────────────────── PR6 follow-up F2 ─
@@ -228,7 +247,7 @@ public class SessionRetentionFanoutServiceTests
         harness.Enqueuer.Verify(e => e.EnqueueAsync(TenantB, It.IsAny<string>(), It.IsAny<string>(),
             It.IsAny<DeletionActor>(), It.IsAny<DeletionRetentionContext?>(), It.IsAny<CancellationToken>()),
             Times.Never);
-        harness.MaintenanceRepo.Verify(m => m.GetSessionsOlderThanAsync(TenantB, It.IsAny<DateTime>()), Times.Never);
+        harness.MaintenanceRepo.Verify(m => m.GetSessionsOlderThanAsync(TenantB, It.IsAny<DateTime>(), It.IsAny<int>()), Times.Never);
     }
 
     // ───────────────────────────────────────────────────────────────────────────────────────────────
@@ -322,7 +341,7 @@ public class SessionRetentionFanoutServiceTests
             _tenantIds.Add(tenantId);
             TenantConfig.Setup(t => t.GetConfigurationAsync(tenantId))
                 .ReturnsAsync(new TenantConfiguration { TenantId = tenantId, DataRetentionDays = retentionDays });
-            MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(tenantId, It.IsAny<DateTime>()))
+            MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(tenantId, It.IsAny<DateTime>(), It.IsAny<int>()))
                 .ReturnsAsync(new List<SessionSummary>(WithTenantId(tenantId, sessions)));
         }
 
@@ -332,7 +351,7 @@ public class SessionRetentionFanoutServiceTests
             TenantConfig.Setup(t => t.GetConfigurationAsync(tenantId))
                 .ReturnsAsync(new TenantConfiguration { TenantId = tenantId, DataRetentionDays = retentionDays });
             foreach (var s in sessions) s.TenantId = tenantId;
-            MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(tenantId, It.IsAny<DateTime>())).ReturnsAsync(sessions);
+            MaintenanceRepo.Setup(m => m.GetSessionsOlderThanAsync(tenantId, It.IsAny<DateTime>(), It.IsAny<int>())).ReturnsAsync(sessions);
         }
 
         private static IEnumerable<SessionSummary> WithTenantId(string tenantId, IEnumerable<SessionSummary> sessions)

src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableMaintenanceRepository.cs

@@ -51,8 +51,8 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
         public Task<int> DeleteAuditLogsOlderThanAsync(DateTime cutoffUtc)
             => _storage.DeleteAuditLogsOlderThanAsync(cutoffUtc);
 
-        public Task<List<SessionSummary>> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate)
-            => _storage.GetSessionsOlderThanAsync(tenantId, cutoffDate);
+        public Task<List<SessionSummary>> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate, int maxResults = int.MaxValue)
+            => _storage.GetSessionsOlderThanAsync(tenantId, cutoffDate, maxResults);
 
         public Task<List<SessionSummary>> GetSessionsByDateRangeAsync(DateTime startDate, DateTime endDate, string? tenantId = null)
             => _storage.GetSessionsByDateRangeAsync(startDate, endDate, tenantId);

src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionRetentionFanoutService.cs

@@ -145,7 +145,15 @@ namespace AutopilotMonitor.Functions.Services.Deletion
             }
 
             var cutoffUtc = DateTime.UtcNow.AddDays(-retentionDays);
-            var oldSessions = await _maintenanceRepo.GetSessionsOlderThanAsync(tenantId, cutoffUtc).ConfigureAwait(false);
+
+            // Server-bounded read: fetch one more than the per-run dispatch cap. The loop below
+            // only ever advances MaxEnqueuesPerTenantPerRun sessions, so loading the whole backlog
+            // every run is wasted I/O that grows with the backlog. The "+1" is a cheap probe that
+            // lets us tell "exactly cap eligible, done" from "cap+ eligible, more deferred" without
+            // scanning the rest — it is observability only and is never dispatched.
+            const int fetchLimit = MaxEnqueuesPerTenantPerRun + 1;
+            var oldSessions = await _maintenanceRepo.GetSessionsOlderThanAsync(tenantId, cutoffUtc, fetchLimit).ConfigureAwait(false);
+            bool moreRemaining = oldSessions.Count > MaxEnqueuesPerTenantPerRun;
 
             if (oldSessions.Count == 0)
             {
@@ -163,9 +171,12 @@ namespace AutopilotMonitor.Functions.Services.Deletion
 
                 if (processed >= MaxEnqueuesPerTenantPerRun)
                 {
+                    // We only fetched cap+1, so we know "more remain" but not the exact backlog
+                    // size (that would require the full scan this change removes). Report it as a
+                    // floor — the next run drains the following batch.
                     _logger.LogInformation(
-                        "Tenant {TenantId}: hit rate limit ({Limit}/run) — deferring remaining {Remaining} sessions to next run",
-                        tenantId, MaxEnqueuesPerTenantPerRun, oldSessions.Count - processed);
+                        "Tenant {TenantId}: hit rate limit ({Limit}/run) — at least one more batch deferred to next run",
+                        tenantId, MaxEnqueuesPerTenantPerRun);
                     result.RateLimitedTenants++;
                     break;
                 }
@@ -223,15 +234,18 @@ namespace AutopilotMonitor.Functions.Services.Deletion
                 {
                     { "RetentionDays", retentionDays.ToString() },
                     { "CutoffUtc", cutoffUtc.ToString("o") },
-                    { "Eligible", oldSessions.Count.ToString() },
+                    // Read is capped at MaxEnqueuesPerTenantPerRun+1, so this is a floor, not the
+                    // exact backlog size. MoreRemaining=true ⇒ at least one further batch is pending.
+                    { "EligibleThisRun", Math.Min(oldSessions.Count, MaxEnqueuesPerTenantPerRun).ToString() },
+                    { "MoreRemaining", moreRemaining.ToString() },
                     { "Enqueued", enqueued.ToString() },
                     { "Skipped", skipped.ToString() },
                     { "AbortedByKillSwitch", result.AbortedByKillSwitch.ToString() },
                 }).ConfigureAwait(false);
 
             _logger.LogInformation(
-                "Tenant {TenantId}: retention fanout — cutoff={Cutoff:o} eligible={Eligible} enqueued={Enqueued} skipped={Skipped}",
-                tenantId, cutoffUtc, oldSessions.Count, enqueued, skipped);
+                "Tenant {TenantId}: retention fanout — cutoff={Cutoff:o} eligibleThisRun={Eligible} moreRemaining={More} enqueued={Enqueued} skipped={Skipped}",
+                tenantId, cutoffUtc, Math.Min(oldSessions.Count, MaxEnqueuesPerTenantPerRun), moreRemaining, enqueued, skipped);
         }
 
         /// <summary>

src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Maintenance.cs

@@ -439,27 +439,38 @@ namespace AutopilotMonitor.Functions.Services
         // ===== DATA RETENTION METHODS =====
 
         /// <summary>
-        /// Gets sessions older than a specific date for a tenant
+        /// Gets sessions older than a specific date for a tenant, capped at <paramref name="maxResults"/>.
+        /// The retention fanout only advances a fixed number of sessions per run, so the read is
+        /// server-bounded to that cap instead of materializing the whole backlog every run.
         /// </summary>
-        public async Task<List<SessionSummary>> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate)
+        public async Task<List<SessionSummary>> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate, int maxResults = int.MaxValue)
         {
             SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+            if (maxResults <= 0) return new List<SessionSummary>();
 
             try
             {
                 var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions);
 
-                // Query sessions for this tenant older than cutoff date
+                // Query sessions for this tenant older than cutoff date.
                 var filter = $"PartitionKey eq '{tenantId}' and StartedAt lt datetime'{cutoffDate:yyyy-MM-ddTHH:mm:ss}Z'";
-                var query = tableClient.QueryAsync<TableEntity>(filter: filter);
+
+                // Bound the server read: cap page size to maxResults and stop enumerating once we
+                // have that many. Azure Tables has no server-side OrderBy, so this returns the first
+                // maxResults rows in RowKey order — the same ordering as the previous full scan, it
+                // just stops reading once the cap is reached. Unbounded callers pass int.MaxValue.
+                int? maxPerPage = maxResults == int.MaxValue ? (int?)null : maxResults;
+                var query = tableClient.QueryAsync<TableEntity>(filter: filter, maxPerPage: maxPerPage);
 
                 var sessions = new List<SessionSummary>();
                 await foreach (var entity in query)
                 {
                     sessions.Add(MapToSessionSummary(entity));
+                    if (sessions.Count >= maxResults)
+                        break;
                 }
 
-                _logger.LogInformation($"Found {sessions.Count} sessions older than {cutoffDate:yyyy-MM-dd} for tenant {tenantId}");
+                _logger.LogInformation($"Found {sessions.Count} sessions older than {cutoffDate:yyyy-MM-dd} for tenant {tenantId} (cap={maxResults})");
                 return sessions;
             }
             catch (Exception ex)

src/Shared/AutopilotMonitor.Shared/DataAccess/IMaintenanceRepository.cs

@@ -57,7 +57,12 @@ namespace AutopilotMonitor.Shared.DataAccess
         Task<int> DeleteAuditLogsOlderThanAsync(DateTime cutoffUtc);
 
         // --- Data Retention Queries ---
-        Task<List<SessionSummary>> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate);
+        /// <summary>
+        /// Sessions older than <paramref name="cutoffDate"/> for a tenant, capped at
+        /// <paramref name="maxResults"/> (server-bounded read). The retention fanout passes its
+        /// per-run dispatch cap so it never materializes a backlog it cannot process in one run.
+        /// </summary>
+        Task<List<SessionSummary>> GetSessionsOlderThanAsync(string tenantId, DateTime cutoffDate, int maxResults = int.MaxValue);
         Task<List<SessionSummary>> GetSessionsByDateRangeAsync(DateTime startDate, DateTime endDate, string? tenantId = null);
         Task<List<SessionSummary>> GetStalledSessionsAsync(string tenantId, DateTime cutoffTime);
         Task<List<SessionSummary>> GetAgentSilentSessionsAsync(string tenantId, DateTime silenceCutoff, DateTime hardCutoff);