diff --git a/src/ClawdDotNet.App/AppHost.cs b/src/ClawdDotNet.App/AppHost.cs index 3268f28..c9430b2 100644 --- a/src/ClawdDotNet.App/AppHost.cs +++ b/src/ClawdDotNet.App/AppHost.cs @@ -726,6 +726,10 @@ public sealed class AppHost : IAsyncDisposable var dispatcher = new EngineTaskDispatcher( engine, () => instance.Agents, instance.InstanceId, tools, stateStore, loggerFactory); + var headroom = engine.BudgetGuard is { } guard + ? new BudgetHeadroom(guard, () => instance.Agents, () => engine.InstanceBudget) + : null; + var scanner = new TaskScanner(taskRepository, dispatcher, loggerFactory, maxParallelRuns: instance.MaxParallelTaskRuns, budgetMissPolicy: task => @@ -736,10 +740,10 @@ public sealed class AppHost : IAsyncDisposable .FirstOrDefault(a => string.Equals(a.AgentId, agentId, StringComparison.OrdinalIgnoreCase)) ?.Budget.MissPolicy ?? BudgetMissPolicy.Expire; }, - // #716: Late work waits for headroom the rest of the day does not need. - lateRunGate: engine.BudgetGuard is { } guard - ? new BudgetHeadroom(guard, () => instance.Agents, () => engine.InstanceBudget) - : null); + // #716: Late work waits for headroom the rest of the day does not need, and a + // day the budget will not last is named while there is still time to act. + lateRunGate: headroom, + budgetOutlook: headroom); // Reconciliation nicht blockierend: Der Start soll nicht auf das Dateisystem warten. _ = Task.Run(async () => diff --git a/src/ClawdDotNet.Core/Budget/BudgetHeadroom.cs b/src/ClawdDotNet.Core/Budget/BudgetHeadroom.cs index d871ee5..18b9b8a 100644 --- a/src/ClawdDotNet.Core/Budget/BudgetHeadroom.cs +++ b/src/ClawdDotNet.Core/Budget/BudgetHeadroom.cs @@ -22,7 +22,7 @@ namespace ClawdDotNet.Core.Budget; /// then only the margin keeps a reserve. That errs towards letting late work through; /// the budget check at the start of the run still refuses once the limit is reached. /// -public sealed class BudgetHeadroom : ILateRunGate +public sealed class BudgetHeadroom : ILateRunGate, IBudgetOutlook { /// Share of each limit kept free as a reserve for surprises. public const decimal SafetyShare = 0.10m; @@ -108,6 +108,82 @@ public sealed class BudgetHeadroom : ILateRunGate return true; } + /// + /// #716: Walks through the rest of today's schedule in time order and finds, per agent + /// with a limit and for the instance, the first run the budget would no longer cover. + /// Late work is left out: it only ever runs in headroom, so it cannot be what makes the + /// day's own runs fail. + /// + public async Task> ShortfallsAsync( + IReadOnlyList runnable, DateTime nowUtc, CancellationToken ct) + { + var agents = _agents(); + var instanceBudget = _instanceBudget(); + var endOfDay = StartOfNextLocalDay(nowUtc); + var result = new List(); + + // Every planned run until midnight, with whose it is. + var planned = new List<(DateTime At, string AgentId)>(); + foreach (var task in runnable.Where(t => !IsLate(t))) + { + var agent = FindAgent(agents, task); + if (agent is null) + continue; + foreach (var at in TaskSchedule.OccurrenceTimesBetween(task, nowUtc, endOfDay)) + planned.Add((at, agent.AgentId)); + } + + planned.Sort((a, b) => a.At.CompareTo(b.At)); + + var estimates = new Dictionary(StringComparer.OrdinalIgnoreCase); + foreach (var agentId in planned.Select(p => p.AgentId).Distinct(StringComparer.OrdinalIgnoreCase)) + estimates[agentId] = await _guard.EstimateRunAsync(agentId, ct); + + foreach (var agent in agents.Where(a => a.Budget.HasAnyLimit)) + { + var runs = planned.Where(p => string.Equals(p.AgentId, agent.AgentId, StringComparison.OrdinalIgnoreCase)).ToList(); + var used = await _guard.GetUsedTodayAsync(agent.AgentId, ct); + var reserved = _guard.Reserved(agent.AgentId); + if (FirstRefused(runs, estimates, agent.Budget.DailyCostUsd, agent.Budget.DailyTokens, used, reserved) is { } index) + result.Add(new BudgetShortfall(agent.AgentId, runs[index].At, runs.Count - index, runs.Count)); + } + + if (instanceBudget.HasAnyLimit) + { + var used = await _guard.GetUsedTodayAsync("", ct); + var reserved = _guard.Reserved(""); + if (FirstRefused(planned, estimates, instanceBudget.DailyCostUsd, instanceBudget.DailyTokens, used, reserved) is { } index) + result.Add(new BudgetShortfall("", planned[index].At, planned.Count - index, planned.Count)); + } + + return result; + } + + /// + /// The index of the first run that would find the budget used up, or null. + /// Mirrors the check at the start of a run: a run is refused once usage has reached + /// the limit, not merely because its own estimate would overshoot. + /// + private static int? FirstRefused( + List<(DateTime At, string AgentId)> runs, Dictionary estimates, + decimal costLimit, long tokenLimit, Accounting.DailyUsage used, RunEstimate reserved) + { + var cost = used.CostUsd + reserved.CostUsd; + long tokens = used.TotalTokens + reserved.Tokens; + + for (var i = 0; i < runs.Count; i++) + { + if ((costLimit > 0 && cost >= costLimit) || (tokenLimit > 0 && tokens >= tokenLimit)) + return i; + + var estimate = estimates.GetValueOrDefault(runs[i].AgentId); + cost += estimate.CostUsd; + tokens += estimate.Tokens; + } + + return null; + } + /// Carried-over work and pending catch-ups are the late runs. public static bool IsLate(TaskItem task) => task.IsCarriedOver || task.CatchUpPending; diff --git a/src/ClawdDotNet.Core/Tasks/TaskScanner.cs b/src/ClawdDotNet.Core/Tasks/TaskScanner.cs index da5f36e..f6f349a 100644 --- a/src/ClawdDotNet.Core/Tasks/TaskScanner.cs +++ b/src/ClawdDotNet.Core/Tasks/TaskScanner.cs @@ -53,6 +53,21 @@ public interface ILateRunGate Task HasRoomAsync(TaskItem lateTask, IReadOnlyList runnable, DateTime nowUtc, CancellationToken ct); } +/// +/// #716: The day ahead — where the budget will not reach the end of today's schedule. +/// +public interface IBudgetOutlook +{ + Task> ShortfallsAsync(IReadOnlyList runnable, DateTime nowUtc, CancellationToken ct); +} + +/// +/// #716: An agent whose budget runs out before today's schedule does — from when on, and +/// how many planned runs will be refused. is empty for the instance +/// limit. +/// +public sealed record BudgetShortfall(string AgentId, DateTime ExhaustsAtUtc, int RunsRefused, int RunsPlanned); + /// Bis C1 kommt: der Markt gilt als immer offen. public sealed class AlwaysOpenMarketCalendar : IMarketCalendar { @@ -121,6 +136,16 @@ public sealed class TaskScanner : IAsyncDisposable private readonly ILateRunGate? _lateRunGate; + private readonly IBudgetOutlook? _outlook; + + /// How often the day ahead is looked at — a forecast, not a watchdog. + public static readonly TimeSpan OutlookInterval = TimeSpan.FromMinutes(30); + + private DateTime _lastOutlookAt = DateTime.MinValue; + + /// Agents already warned about today, with the day — once per agent and day. + private readonly Dictionary _shortfallWarned = new(StringComparer.OrdinalIgnoreCase); + /// How many late runs waited for budget headroom at the last tick. private int _lastWaitingForHeadroom; @@ -137,7 +162,8 @@ public sealed class TaskScanner : IAsyncDisposable TimeSpan? lease = null, int maxParallelRuns = DefaultMaxParallelRuns, Func? budgetMissPolicy = null, - ILateRunGate? lateRunGate = null) + ILateRunGate? lateRunGate = null, + IBudgetOutlook? budgetOutlook = null) { _repo = repo; _dispatcher = dispatcher; @@ -149,6 +175,7 @@ public sealed class TaskScanner : IAsyncDisposable _maxParallelRuns = Math.Max(1, maxParallelRuns); _missPolicy = budgetMissPolicy ?? (_ => BudgetMissPolicy.Expire); _lateRunGate = lateRunGate; + _outlook = budgetOutlook; } /// How many scheduled runs may be in flight at once (#682). @@ -257,6 +284,7 @@ public sealed class TaskScanner : IAsyncDisposable var candidates = await _repo.ListRunnableAsync(ct); SummarizeMissesIfNewDay(now); + await WarnAboutShortfallsAsync(candidates, now, ct); var due = new List<(TaskItem Task, string Occurrence, bool Late)>(); foreach (var task in candidates) @@ -563,6 +591,46 @@ public sealed class TaskScanner : IAsyncDisposable : "occurrence lapsed (policy: expire)"); } + /// + /// #716: Prevent rather than catch up. Every half hour the day ahead is checked; an + /// agent whose budget will not last until midnight is named once a day — with the time + /// it runs out and how many planned runs that costs — while there is still time to + /// raise the budget or decide what can go. + /// + private async Task WarnAboutShortfallsAsync(IReadOnlyList runnable, DateTime nowUtc, CancellationToken ct) + { + if (_outlook is null || nowUtc - _lastOutlookAt < OutlookInterval) + return; + _lastOutlookAt = nowUtc; + + IReadOnlyList shortfalls; + try + { + shortfalls = await _outlook.ShortfallsAsync(runnable, nowUtc, ct); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _logger.LogWarning(ex, "Budget outlook failed — no forecast this time"); + return; + } + + var today = DateOnly.FromDateTime(TimeZoneInfo.ConvertTimeFromUtc(nowUtc, _clock.LocalTimeZone)); + foreach (var shortfall in shortfalls) + { + if (_shortfallWarned.TryGetValue(shortfall.AgentId, out var warnedOn) && warnedOn == today) + continue; + _shortfallWarned[shortfall.AgentId] = today; + + var local = TimeZoneInfo.ConvertTimeFromUtc(shortfall.ExhaustsAtUtc, _clock.LocalTimeZone); + _logger.LogWarning( + "Budget forecast: the daily budget of {Who} will run out around {Time:HH:mm} — " + + "{Refused} of {Planned} runs planned for the rest of today would be refused. " + + "Raise the budget or decide what can wait.", + shortfall.AgentId.Length == 0 ? "the instance" : $"agent '{shortfall.AgentId}'", + local, shortfall.RunsRefused, shortfall.RunsPlanned); + } + } + /// /// #716: On the first tick of a new local day, one line per agent about what the /// budget refused the day before. The details are on the tasks themselves. diff --git a/src/ClawdDotNet.Core/Tasks/TaskSchedule.cs b/src/ClawdDotNet.Core/Tasks/TaskSchedule.cs index 58bee19..c5d01c3 100644 --- a/src/ClawdDotNet.Core/Tasks/TaskSchedule.cs +++ b/src/ClawdDotNet.Core/Tasks/TaskSchedule.cs @@ -207,61 +207,70 @@ public static class TaskSchedule /// basis for "how much budget does the rest of the day still need". /// public static int OccurrencesBetween(TaskItem task, DateTime fromUtc, DateTime toUtc, int cap = 1_440) + => OccurrenceTimesBetween(task, fromUtc, toUtc, cap).Count; + + /// + /// #716: When the task will run from up to (excluding) + /// . A run due right now is listed at . + /// + public static IReadOnlyList OccurrenceTimesBetween(TaskItem task, DateTime fromUtc, DateTime toUtc, int cap = 1_440) { fromUtc = DateTime.SpecifyKind(fromUtc, DateTimeKind.Utc); toUtc = DateTime.SpecifyKind(toUtc, DateTimeKind.Utc); + var times = new List(); if (toUtc <= fromUtc) - return 0; + return times; - var dueNow = DueOccurrence(task, fromUtc) is not null ? 1 : 0; + var dueNow = DueOccurrence(task, fromUtc) is not null; + if (dueNow) + times.Add(fromUtc); if (task.When is not { } when) - return dueNow; // one-time without a schedule: due until it ran + return times; // one-time without a schedule: due until it ran switch (when.Kind) { case TaskWhenKind.At: { - if (dueNow == 1 || task.LastOccurrence is not null) - return dueNow; - return TryResolveInstant(when.Value, when.TimeZone, out var at) && at < toUtc ? 1 : 0; + if (!dueNow && task.LastOccurrence is null + && TryResolveInstant(when.Value, when.TimeZone, out var at) && at < toUtc) + times.Add(at); + return times; } case TaskWhenKind.Every: { if (!TryParseInterval(when.Value, out var interval) || interval <= TimeSpan.Zero) - return dueNow; + return times; - var next = dueNow == 1 + var next = dueNow ? fromUtc + interval : NextOccurrenceUtc(task, fromUtc) ?? toUtc; - var count = dueNow; - while (count < cap && next < toUtc) + while (times.Count < cap && next < toUtc) { - count++; + times.Add(next); next += interval; } - return count; + return times; } case TaskWhenKind.Cron: { if (!TryPrepareCron(when, out var cron, out var tz)) - return 0; + return times; - var count = dueNow; var cursor = fromUtc; - while (count < cap && NextCronUtc(cron, tz, cursor) is { } next && next < toUtc) + while (times.Count < cap && NextCronUtc(cron, tz, cursor) is { } next && next < toUtc) { if (next > fromUtc) - count++; + times.Add(next); cursor = next; } - return count; + return times; } default: - return dueNow; + return times; } } diff --git a/tests/ClawdDotNet.Core.Tests/Budget/BudgetHeadroomTests.cs b/tests/ClawdDotNet.Core.Tests/Budget/BudgetHeadroomTests.cs index d0fa360..99c6d90 100644 --- a/tests/ClawdDotNet.Core.Tests/Budget/BudgetHeadroomTests.cs +++ b/tests/ClawdDotNet.Core.Tests/Budget/BudgetHeadroomTests.cs @@ -141,6 +141,54 @@ public sealed class BudgetHeadroomTests (await Headroom(used: 99m, Agent(dailyCost: 0)).HasRoomAsync(late, [late], Now, default)).ShouldBeTrue(); } + // ═══════════════════ the day ahead ═══════════════════ + + [Fact] + public async Task The_outlook_names_the_hour_the_budget_runs_out() + { + // Limit 1.00, used 0.60, runs at 14:00 (due now), 15:00 … 23:00 à 0.05. Before the + // 22:00 run 0.60 + 8 × 0.05 = 1.00 is used — from there on everything is refused. + var hourly = Cron("t-hourly", "0 * * * *", lastOccurrence: "2027-03-10T13:00:00.0000000Z"); + + var shortfalls = await Headroom(used: 0.60m, Agent()).ShortfallsAsync([hourly], Now, default); + + var shortfall = shortfalls.ShouldHaveSingleItem(); + shortfall.AgentId.ShouldBe("worker"); + shortfall.ExhaustsAtUtc.ShouldBe(new DateTime(2027, 3, 10, 22, 0, 0, DateTimeKind.Utc)); + shortfall.RunsRefused.ShouldBe(2); + shortfall.RunsPlanned.ShouldBe(10); + } + + [Fact] + public async Task A_day_that_fits_has_no_shortfall() + { + var hourly = Cron("t-hourly", "0 * * * *", lastOccurrence: "2027-03-10T13:00:00.0000000Z"); + + (await Headroom(used: 0.30m, Agent()).ShortfallsAsync([hourly], Now, default)).ShouldBeEmpty(); + } + + [Fact] + public async Task Late_work_does_not_make_the_day_fall_short() + { + // Fifty carried-over tasks only ever run in headroom — they cannot be the reason the + // day's own runs are refused. + var hourly = Cron("t-hourly", "0 * * * *", lastOccurrence: "2027-03-10T13:00:00.0000000Z"); + var late = Enumerable.Range(0, 50).Select(i => Carried($"t-late-{i}")).ToList(); + + (await Headroom(used: 0.30m, Agent()).ShortfallsAsync([hourly, .. late], Now, default)).ShouldBeEmpty(); + } + + [Fact] + public async Task The_instance_limit_gets_its_own_forecast() + { + var hourly = Cron("t-hourly", "0 * * * *", lastOccurrence: "2027-03-10T13:00:00.0000000Z"); + var instance = new InstanceBudget { DailyCostUsd = 1.00m }; + + var shortfalls = await Headroom(used: 0.60m, Agent(dailyCost: 0), instance).ShortfallsAsync([hourly], Now, default); + + shortfalls.ShouldHaveSingleItem().AgentId.ShouldBe("", "empty means: the instance"); + } + // ═══════════════════ test doubles ═══════════════════ private sealed class UtcClock(DateTime now) : TimeProvider