Merge feat/716-budgetende: Ausfall vermerken, Reservierung, Spielraum, Vorschau (#716)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Richard
2026-09-24 12:12:44 +02:00
co-authored by Claude Opus 5.5
31 changed files with 1709 additions and 38 deletions
+8 -3
View File
@@ -954,9 +954,14 @@ scheitert:
| `carry_over` | wird mitgenommen | einmalige Aufgaben: echte Arbeit, die nicht verschwinden darf |
Dazu gilt: **Der neue Tag hat Vorfahrt.** Nachholläufe kommen nach den regulär fälligen,
unterhalb jeder Priorität, und dürfen höchstens einen Anteil des neuen Tagesbudgets
verbrauchen (Vorgabe etwa 25 %). Ohne diese Regel schiebt ein einziger überzogener Tag
seinen Ausfall Tag für Tag vor sich her, statt ihn aufzulösen. Eine einmalige Aufgabe, die
unterhalb jeder Priorität, und sie nutzen nur den **Spielraum**, den der Rest des Tages
nicht braucht: Grenze minus Verbrauch, minus die Reservierung laufender Läufe, minus der
Bedarf der noch geplanten Läufe bis Mitternacht, minus 10 % Reserve. Ohne diese Regel
schiebt ein einziger überzogener Tag seinen Ausfall Tag für Tag vor sich her, statt ihn
aufzulösen. *Beim Bau am 2026-09-24 geändert:* Das Konzept sah hier zunächst einen
pauschalen Anteil von 25 % vor. Ein fester Anteil ist an einem vollen Tag zu viel und an
einem ruhigen unnötig wenig; der Spielraum folgt dem tatsächlichen Plan, und dieselbe
Rechnung trägt später die Füllarbeit. Eine einmalige Aufgabe, die
nach einigen Tagen immer noch nicht drankam, geht an den Menschen: Was nie läuft, darf
nicht unbemerkt liegen bleiben.
+18 -1
View File
@@ -3,6 +3,7 @@ using ClawdDotNet.App.Settings;
using ClawdDotNet.Core.Accounting;
using ClawdDotNet.Core.Api;
using ClawdDotNet.Core.Audit;
using ClawdDotNet.Core.Budget;
using ClawdDotNet.Core.Config;
using ClawdDotNet.Core.Engine;
using ClawdDotNet.Core.Logging;
@@ -725,8 +726,24 @@ 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);
maxParallelRuns: instance.MaxParallelTaskRuns,
budgetMissPolicy: task =>
{
// #716: The agent's rule. Bare '@new' names no agent — then the default applies.
var agentId = TaskAssignee.AgentId(task.Assignee);
return instance.Agents
.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, 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 () =>
@@ -187,7 +187,8 @@ public sealed class TaskAdministration(
if (board is null)
return null;
var aufgaben = await board.ListAsync(new TaskQuery { Limit = 500 }, ct);
// #716: ungekappt - vorher fehlten ab 500 Aufgaben die übrigen in der Prognose.
var aufgaben = await board.ListAllAsync(ct);
var belege = new Dictionary<string, IReadOnlyList<RunReceipt>>(StringComparer.Ordinal);
if (audit is not null)
@@ -125,6 +125,29 @@ public sealed class SqliteUsageRepository : IUsageRepository
return results;
}
public async Task<IReadOnlyList<RunUsage>> GetRecentForAgentAsync(string agentId, int limit, CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
using var cmd = conn.CreateCommand();
cmd.CommandText = """
SELECT Id, AgentId, Model, PromptTokens, CompletionTokens, CachedTokens,
CostUsd, CostIsKnown, Status, StepCount, DurationMs, OccurredAt
FROM RunUsage
WHERE AgentId = @agent COLLATE NOCASE
ORDER BY OccurredAt DESC
LIMIT @limit
""";
cmd.Parameters.AddWithValue("@agent", agentId);
cmd.Parameters.AddWithValue("@limit", Math.Clamp(limit, 1, 1000));
var results = new List<RunUsage>();
await using var reader = await cmd.ExecuteReaderAsync(ct);
while (await reader.ReadAsync(ct))
results.Add(Read(reader));
return results;
}
public Task<int> PurgeBeforeAsync(DateOnly date, CancellationToken ct)
=> _storage.WriteAsync(async conn =>
{
@@ -53,6 +53,14 @@ public interface IUsageRepository
Task<IReadOnlyList<RunUsage>> GetRecentAsync(int limit, CancellationToken ct);
/// <summary>
/// #716: The most recent runs of one agent — the basis for estimating what a run of
/// it will cost. Without an implementation there is no estimate, and a reservation
/// counts nothing.
/// </summary>
Task<IReadOnlyList<RunUsage>> GetRecentForAgentAsync(string agentId, int limit, CancellationToken ct)
=> Task.FromResult<IReadOnlyList<RunUsage>>([]);
/// <summary>Entfernt Einträge älter als das angegebene Datum. Gibt die Anzahl zurück.</summary>
Task<int> PurgeBeforeAsync(DateOnly date, CancellationToken ct);
}
+146 -7
View File
@@ -13,6 +13,13 @@ public enum BudgetLimit
InstanceTokens
}
/// <summary>#716: What a run is expected to use — the agent's average over recent runs.</summary>
public readonly record struct RunEstimate(decimal CostUsd, long Tokens)
{
public static RunEstimate operator +(RunEstimate a, RunEstimate b) => new(a.CostUsd + b.CostUsd, a.Tokens + b.Tokens);
public static RunEstimate operator -(RunEstimate a, RunEstimate b) => new(a.CostUsd - b.CostUsd, a.Tokens - b.Tokens);
}
public sealed record BudgetStatus(bool IsAllowed, BudgetLimit Limit, string? Reason)
{
public static BudgetStatus Allowed { get; } = new(true, BudgetLimit.None, null);
@@ -34,6 +41,19 @@ public sealed class BudgetGuard
private readonly IUsageRepository _usage;
private readonly Func<DateTime> _now;
/// <summary>How many recent runs of an agent make up its per-run estimate (#716).</summary>
public const int EstimateSample = 20;
/// <summary>
/// #716: Check and reserve happen under this lock. Without it, two runs starting at
/// the same moment would both read the same "used" figure and both pass.
/// </summary>
private readonly SemaphoreSlim _reserveGate = new(1, 1);
private readonly Lock _reservedLock = new();
private readonly Dictionary<string, RunEstimate> _reservedByAgent = new(StringComparer.OrdinalIgnoreCase);
private RunEstimate _reservedInstance;
public BudgetGuard(IUsageRepository usage, Func<DateTime>? now = null)
{
_usage = usage;
@@ -43,8 +63,116 @@ public sealed class BudgetGuard
/// <summary>
/// Prüft, ob der Agent heute noch laufen darf.
/// </summary>
public async Task<BudgetStatus> CheckAsync(
public Task<BudgetStatus> CheckAsync(
AgentConfig agent, InstanceBudget instanceBudget, CancellationToken ct)
=> CheckCoreAsync(agent, instanceBudget, default, default, ct);
/// <summary>
/// #716: Checks like <see cref="CheckAsync"/> and, if the run may start, reserves its
/// estimated usage until the returned handle is disposed — i.e. until the run's usage
/// is recorded.
///
/// <para><b>What changes.</b> Before, each run was checked on its own against what was
/// already recorded. Runs in flight were invisible, so several starting together all
/// passed and exceeded the budget together (#682 capped how many; this closes it).
/// Now the runs in flight count with their estimate.</para>
///
/// <para><b>What does not change.</b> A single run is still refused only when the
/// budget is already used up — its own estimate is not added. Otherwise a small task
/// would be refused because the agent's average run is large.</para>
/// </summary>
public async Task<(BudgetStatus Status, IDisposable? Reservation)> ReserveAsync(
AgentConfig agent, InstanceBudget instanceBudget, CancellationToken ct)
{
if (!agent.Budget.HasAnyLimit && !instanceBudget.HasAnyLimit)
return (BudgetStatus.Allowed, null);
var estimate = await EstimateRunAsync(agent.AgentId, ct);
await _reserveGate.WaitAsync(ct);
try
{
RunEstimate agentReserved, instanceReserved;
lock (_reservedLock)
{
agentReserved = _reservedByAgent.GetValueOrDefault(agent.AgentId);
instanceReserved = _reservedInstance;
}
var status = await CheckCoreAsync(agent, instanceBudget, agentReserved, instanceReserved, ct);
if (!status.IsAllowed)
return (status, null);
lock (_reservedLock)
{
_reservedByAgent[agent.AgentId] = _reservedByAgent.GetValueOrDefault(agent.AgentId) + estimate;
_reservedInstance += estimate;
}
return (status, new Reservation(this, agent.AgentId, estimate));
}
finally
{
_reserveGate.Release();
}
}
/// <summary>Today's usage of an agent, or with an empty id of the whole instance.</summary>
public Task<DailyUsage> GetUsedTodayAsync(string agentId, CancellationToken ct)
=> _usage.GetDailyAsync(DateOnly.FromDateTime(_now()), agentId, ct);
/// <summary>What is currently reserved by runs in flight — for the agent, or with an
/// empty id for the whole instance.</summary>
public RunEstimate Reserved(string agentId)
{
lock (_reservedLock)
return agentId.Length == 0 ? _reservedInstance : _reservedByAgent.GetValueOrDefault(agentId);
}
/// <summary>
/// The average of the agent's recent runs. Runs without usage (refusals, aborted
/// before the first call) do not count — they would drag the estimate towards zero.
/// </summary>
public async Task<RunEstimate> EstimateRunAsync(string agentId, CancellationToken ct)
{
var recent = await _usage.GetRecentForAgentAsync(agentId, EstimateSample, ct);
var runs = recent.Where(r => r.TotalTokens > 0).ToList();
if (runs.Count == 0)
return default;
return new RunEstimate(
runs.Average(r => r.CostUsd),
(long)Math.Ceiling(runs.Average(r => (double)r.TotalTokens)));
}
private void Release(string agentId, RunEstimate estimate)
{
lock (_reservedLock)
{
var left = _reservedByAgent.GetValueOrDefault(agentId) - estimate;
if (left.CostUsd <= 0 && left.Tokens <= 0)
_reservedByAgent.Remove(agentId);
else
_reservedByAgent[agentId] = left;
_reservedInstance -= estimate;
}
}
private sealed class Reservation(BudgetGuard guard, string agentId, RunEstimate estimate) : IDisposable
{
private int _disposed;
public void Dispose()
{
if (Interlocked.Exchange(ref _disposed, 1) == 0)
guard.Release(agentId, estimate);
}
}
private async Task<BudgetStatus> CheckCoreAsync(
AgentConfig agent, InstanceBudget instanceBudget,
RunEstimate agentReserved, RunEstimate instanceReserved, CancellationToken ct)
{
var today = DateOnly.FromDateTime(_now());
@@ -54,19 +182,21 @@ public sealed class BudgetGuard
{
var used = await _usage.GetDailyAsync(today, agent.AgentId, ct);
if (Exceeds(agentBudget.DailyCostUsd, used.CostUsd))
if (Exceeds(agentBudget.DailyCostUsd, used.CostUsd + agentReserved.CostUsd))
{
return new BudgetStatus(false, BudgetLimit.AgentCost,
$"Tagesbudget des Agenten erschöpft: {used.CostUsd:F4} von maximal " +
$"{agentBudget.DailyCostUsd:F2} USD verbraucht." +
ReservedHint(agentReserved.CostUsd) +
UnpricedHint(used));
}
if (Exceeds(agentBudget.DailyTokens, used.TotalTokens))
if (Exceeds(agentBudget.DailyTokens, used.TotalTokens + agentReserved.Tokens))
{
return new BudgetStatus(false, BudgetLimit.AgentTokens,
$"Token-Tageslimit des Agenten erschöpft: {used.TotalTokens:N0} von maximal " +
$"{agentBudget.DailyTokens:N0}.");
$"{agentBudget.DailyTokens:N0}." +
ReservedTokensHint(agentReserved.Tokens));
}
}
@@ -75,19 +205,21 @@ public sealed class BudgetGuard
{
var used = await _usage.GetDailyAsync(today, agentId: "", ct);
if (Exceeds(instanceBudget.DailyCostUsd, used.CostUsd))
if (Exceeds(instanceBudget.DailyCostUsd, used.CostUsd + instanceReserved.CostUsd))
{
return new BudgetStatus(false, BudgetLimit.InstanceCost,
$"Tagesbudget der Instanz erschöpft: {used.CostUsd:F4} von maximal " +
$"{instanceBudget.DailyCostUsd:F2} USD verbraucht." +
ReservedHint(instanceReserved.CostUsd) +
UnpricedHint(used));
}
if (Exceeds(instanceBudget.DailyTokens, used.TotalTokens))
if (Exceeds(instanceBudget.DailyTokens, used.TotalTokens + instanceReserved.Tokens))
{
return new BudgetStatus(false, BudgetLimit.InstanceTokens,
$"Token-Tageslimit der Instanz erschöpft: {used.TotalTokens:N0} von maximal " +
$"{instanceBudget.DailyTokens:N0}.");
$"{instanceBudget.DailyTokens:N0}." +
ReservedTokensHint(instanceReserved.Tokens));
}
}
@@ -109,6 +241,13 @@ public sealed class BudgetGuard
private static bool Exceeds(long limit, long used) => limit > 0 && used >= limit;
/// <summary>#716: Without this, a refusal at "3.20 of 5.00 USD used" would look wrong.</summary>
private static string ReservedHint(decimal reserved)
=> reserved > 0 ? $" Weitere {reserved:F4} USD sind für laufende Läufe reserviert." : "";
private static string ReservedTokensHint(long reserved)
=> reserved > 0 ? $" Weitere {reserved:N0} Tokens sind für laufende Läufe reserviert." : "";
/// <summary>
/// Ohne Preisangaben ist die verbrauchte Summe unvollständig — sonst wirkte ein
/// niedriger Wert wie ein noch offener Spielraum.
@@ -0,0 +1,220 @@
using ClawdDotNet.Core.Config;
using ClawdDotNet.Core.Tasks;
namespace ClawdDotNet.Core.Budget;
/// <summary>
/// #716: Decides whether a late run — carried-over work or a catch-up — may start now,
/// without taking budget the rest of the day still needs.
///
/// <para><b>The rule: the new day has right of way.</b> Headroom is the limit minus what
/// was used, minus what runs in flight have reserved, minus what the day's remaining
/// scheduled runs will need, minus a safety margin. A late run starts only if its
/// estimate fits into that. Otherwise one overdrawn day would push its shortfall into the
/// next, and the next, instead of resolving it.</para>
///
/// <para>This replaces the flat "at most 25 % of the new day" from the concept: a fixed
/// share is either too much on a busy day or needlessly little on a quiet one. The same
/// calculation is what fill work (#717) will run on.</para>
///
/// <para><b>Estimates, honestly.</b> A run is estimated from the agent's recent runs
/// (<see cref="BudgetGuard.EstimateRunAsync"/>). Without history the estimate is zero —
/// 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.</para>
/// </summary>
public sealed class BudgetHeadroom : ILateRunGate, IBudgetOutlook
{
/// <summary>Share of each limit kept free as a reserve for surprises.</summary>
public const decimal SafetyShare = 0.10m;
private readonly BudgetGuard _guard;
private readonly Func<IReadOnlyList<AgentConfig>> _agents;
private readonly Func<InstanceBudget> _instanceBudget;
private readonly TimeProvider _clock;
public BudgetHeadroom(
BudgetGuard guard,
Func<IReadOnlyList<AgentConfig>> agents,
Func<InstanceBudget> instanceBudget,
TimeProvider? clock = null)
{
_guard = guard;
_agents = agents;
_instanceBudget = instanceBudget;
_clock = clock ?? TimeProvider.System;
}
public async Task<bool> HasRoomAsync(
TaskItem lateTask, IReadOnlyList<TaskItem> runnable, DateTime nowUtc, CancellationToken ct)
{
var agents = _agents();
var agent = FindAgent(agents, lateTask);
var instanceBudget = _instanceBudget();
// Without a known agent there is nothing to calculate against; the budget check at
// the start of the run still applies.
if (agent is null || (!agent.Budget.HasAnyLimit && !instanceBudget.HasAnyLimit))
return true;
var endOfDay = StartOfNextLocalDay(nowUtc);
var estimates = new Dictionary<string, RunEstimate>(StringComparer.OrdinalIgnoreCase);
async Task<RunEstimate> EstimateFor(string agentId)
{
if (!estimates.TryGetValue(agentId, out var estimate))
estimates[agentId] = estimate = await _guard.EstimateRunAsync(agentId, ct);
return estimate;
}
var ownEstimate = await EstimateFor(agent.AgentId);
// Regular work of the rest of the day: everything scheduled except late work
// itself — that is what has to wait for headroom, not what headroom is kept for.
var regular = runnable.Where(t => t.Id != lateTask.Id && !IsLate(t)).ToList();
if (agent.Budget.HasAnyLimit)
{
var demand = default(RunEstimate);
foreach (var task in regular.Where(t => IsOf(t, agent.AgentId)))
demand += Times(ownEstimate, TaskSchedule.OccurrencesBetween(task, nowUtc, endOfDay));
var used = await _guard.GetUsedTodayAsync(agent.AgentId, ct);
var reserved = _guard.Reserved(agent.AgentId);
if (!Fits(agent.Budget.DailyCostUsd, used.CostUsd, reserved.CostUsd, demand.CostUsd, ownEstimate.CostUsd)
|| !Fits(agent.Budget.DailyTokens, used.TotalTokens, reserved.Tokens, demand.Tokens, ownEstimate.Tokens))
return false;
}
if (instanceBudget.HasAnyLimit)
{
var demand = default(RunEstimate);
foreach (var task in regular)
{
var agentId = TaskAssignee.AgentId(task.Assignee);
if (agentId.Length == 0)
continue;
demand += Times(await EstimateFor(agentId), TaskSchedule.OccurrencesBetween(task, nowUtc, endOfDay));
}
var used = await _guard.GetUsedTodayAsync("", ct);
var reserved = _guard.Reserved("");
if (!Fits(instanceBudget.DailyCostUsd, used.CostUsd, reserved.CostUsd, demand.CostUsd, ownEstimate.CostUsd)
|| !Fits(instanceBudget.DailyTokens, used.TotalTokens, reserved.Tokens, demand.Tokens, ownEstimate.Tokens))
return false;
}
return true;
}
/// <summary>
/// #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.
/// </summary>
public async Task<IReadOnlyList<BudgetShortfall>> ShortfallsAsync(
IReadOnlyList<TaskItem> runnable, DateTime nowUtc, CancellationToken ct)
{
var agents = _agents();
var instanceBudget = _instanceBudget();
var endOfDay = StartOfNextLocalDay(nowUtc);
var result = new List<BudgetShortfall>();
// 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<string, RunEstimate>(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;
}
/// <summary>
/// The index of the first run that would find the budget used up, or <c>null</c>.
/// 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.
/// </summary>
private static int? FirstRefused(
List<(DateTime At, string AgentId)> runs, Dictionary<string, RunEstimate> 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;
}
/// <summary>Carried-over work and pending catch-ups are the late runs.</summary>
public static bool IsLate(TaskItem task) => task.IsCarriedOver || task.CatchUpPending;
/// <summary>0 or less means: no limit of this kind.</summary>
private static bool Fits(decimal limit, decimal used, decimal reserved, decimal demand, decimal estimate)
=> limit <= 0 || limit - used - reserved - demand - limit * SafetyShare >= estimate;
private static bool Fits(long limit, long used, long reserved, long demand, long estimate)
=> limit <= 0 || limit - used - reserved - demand - (long)(limit * SafetyShare) >= estimate;
private static RunEstimate Times(RunEstimate estimate, int runs)
=> new(estimate.CostUsd * runs, estimate.Tokens * runs);
private static bool IsOf(TaskItem task, string agentId)
=> string.Equals(TaskAssignee.AgentId(task.Assignee), agentId, StringComparison.OrdinalIgnoreCase);
private static AgentConfig? FindAgent(IReadOnlyList<AgentConfig> agents, TaskItem task)
{
var agentId = TaskAssignee.AgentId(task.Assignee);
return agentId.Length == 0
? (agents.Count == 1 ? agents[0] : null)
: agents.FirstOrDefault(a => string.Equals(a.AgentId, agentId, StringComparison.OrdinalIgnoreCase));
}
private DateTime StartOfNextLocalDay(DateTime nowUtc)
{
var zone = _clock.LocalTimeZone;
var local = TimeZoneInfo.ConvertTimeFromUtc(DateTime.SpecifyKind(nowUtc, DateTimeKind.Utc), zone);
var midnight = DateTime.SpecifyKind(local.Date.AddDays(1), DateTimeKind.Unspecified);
while (zone.IsInvalidTime(midnight))
midnight = midnight.AddMinutes(1);
return TimeZoneInfo.ConvertTimeToUtc(midnight, zone);
}
}
@@ -183,6 +183,21 @@ public sealed class AgentBudget
[JsonPropertyName("dailyTokens")]
public long DailyTokens { get; set; }
/// <summary>
/// #716: What happens to a run of a recurring task when the budget is exhausted —
/// <c>expire</c> (default) or <c>catchUpOnce</c>. One-time tasks are always carried
/// over to the next day. Kept as text so an unknown value falls back instead of
/// failing the whole configuration.
/// </summary>
[JsonPropertyName("whenExhausted")]
public string WhenExhausted { get; set; } = "expire";
[JsonIgnore]
public Tasks.BudgetMissPolicy MissPolicy
=> string.Equals(WhenExhausted?.Trim(), "catchUpOnce", StringComparison.OrdinalIgnoreCase)
? Tasks.BudgetMissPolicy.CatchUpOnce
: Tasks.BudgetMissPolicy.Expire;
[JsonIgnore]
public bool HasAnyLimit => DailyCostUsd > 0 || DailyTokens > 0;
}
+19 -8
View File
@@ -35,6 +35,12 @@ public sealed class AgentEngine : IAgentMessageRouter, Staging.IFrozenCallExecut
/// <summary>Tagesgrenzen der Instanz. Wird vom Host gesetzt.</summary>
public InstanceBudget InstanceBudget { get; set; } = InstanceBudget.Unlimited;
/// <summary>
/// #716: The guard that checks and reserves every run — the scanner's headroom
/// calculation has to see the same reservations. <c>null</c> without usage recording.
/// </summary>
public BudgetGuard? BudgetGuard => _budgetGuard;
/// <summary>
/// Wie viele Weitergaben eine Delegationskette höchstens haben darf (B8).
/// A→B ist die Tiefe 1, A→B→C die Tiefe 2.
@@ -163,8 +169,11 @@ public sealed class AgentEngine : IAgentMessageRouter, Staging.IFrozenCallExecut
return zu;
// Vor der ersten Anfrage prüfen — ein erschöpftes Budget soll gar nichts kosten.
if (await CheckBudgetAsync(agentConfig, source ?? AuditSource.Direct, taskId, externalCt) is { } denied)
// #716: Die Reservierung gilt, bis der Verbrauch erfasst ist (Ende der Methode).
var (denied, reservation) = await CheckBudgetAsync(agentConfig, source ?? AuditSource.Direct, taskId, externalCt);
if (denied is not null)
return denied;
using var budgetReservation = reservation;
using var lauf = TrackRun();
@@ -444,8 +453,10 @@ public sealed class AgentEngine : IAgentMessageRouter, Staging.IFrozenCallExecut
if (RefuseWhileShuttingDown(agentConfig) is { } zu)
return zu;
if (await CheckBudgetAsync(agentConfig, source, taskId, externalCt) is { } denied)
var (denied, reservation) = await CheckBudgetAsync(agentConfig, source, taskId, externalCt);
if (denied is not null)
return denied;
using var budgetReservation = reservation;
using var lauf = TrackRun();
@@ -1638,17 +1649,17 @@ public sealed class AgentEngine : IAgentMessageRouter, Staging.IFrozenCallExecut
/// <summary>
/// Prüft das Tagesbudget. Gibt ein Ergebnis zurück, wenn der Lauf nicht stattfinden
/// darf — sonst null.
/// darf — sonst null und, bei gesetzter Grenze, die Reservierung des Laufs (#716).
/// </summary>
private async Task<AgentRunResult?> CheckBudgetAsync(
private async Task<(AgentRunResult? Denied, IDisposable? Reservation)> CheckBudgetAsync(
AgentConfig agentConfig, string? source, string? taskId, CancellationToken ct)
{
if (_budgetGuard is null)
return null;
return (null, null);
var status = await _budgetGuard.CheckAsync(agentConfig, InstanceBudget, ct);
var (status, reservation) = await _budgetGuard.ReserveAsync(agentConfig, InstanceBudget, ct);
if (status.IsAllowed)
return null;
return (null, reservation);
// #608: Ist ein Budget erschoepft, lehnt es jeden weiteren Lauf des Tages ab. Einmal je
// Agent, Grenze und Tag als Warnung genuegt - danach steht jede Ablehnung als Beleg da.
@@ -1676,7 +1687,7 @@ public sealed class AgentEngine : IAgentMessageRouter, Staging.IFrozenCallExecut
await RecordReceiptAsync(Guid.NewGuid().ToString("N"), agentConfig, result, source, taskId);
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
return (result, null);
}
/// <summary>Welche Budgetgrenze je Agent zuletzt an welchem Tag als Warnung gemeldet wurde.</summary>
@@ -339,6 +339,12 @@ public sealed class SqliteStorage
// #554: Paart den "angefangen"-Eintrag mit seinem Abschluss.
SpalteErgaenzen(connection, "AuditLog", "CallId", "TEXT NOT NULL DEFAULT ''");
// #716: What happened when the budget refused a run.
SpalteErgaenzen(connection, "Tasks", "BudgetMissCount", "INTEGER NOT NULL DEFAULT 0");
SpalteErgaenzen(connection, "Tasks", "LastBudgetMissAt", "TEXT NULL");
SpalteErgaenzen(connection, "Tasks", "DeferredUntil", "TEXT NULL");
SpalteErgaenzen(connection, "Tasks", "CatchUpPending", "INTEGER NOT NULL DEFAULT 0");
// Findet offene Aufrufe beim Start, ohne die ganze Tabelle zu lesen.
using var index = connection.CreateCommand();
index.CommandText = """
@@ -98,7 +98,8 @@ public sealed class EngineTaskDispatcher : ITaskDispatcher
return result.Status switch
{
AgentRunStatus.Completed => TaskLauf.Erledigt,
AgentRunStatus.BudgetExceeded or AgentRunStatus.ShuttingDown => TaskLauf.NichtGelaufen,
AgentRunStatus.BudgetExceeded => TaskLauf.KeinBudget,
AgentRunStatus.ShuttingDown => TaskLauf.NichtGelaufen,
_ => TaskLauf.Gescheitert
};
}
@@ -29,6 +29,13 @@ public interface ITaskRepository
/// </summary>
Task<IReadOnlyList<TaskItem>> ListRunnableAsync(CancellationToken ct);
/// <summary>
/// Every task, without a cap — for the few callers that must see the whole board: the
/// cost forecast, duplicate detection, the import's orphan sweep. Before #716 they used
/// <see cref="ListAsync"/> with its limit of 500 and silently saw only part of it.
/// </summary>
Task<IReadOnlyList<TaskItem>> ListAllAsync(bool includeArchived, CancellationToken ct);
/// <summary>
/// Beansprucht einen fälligen Termin atomar. Genau ein gleichzeitiger Aufruf gewinnt;
/// jeder weitere erhält <c>false</c>. Setzt den Marker (<paramref name="occurrenceKey"/>)
@@ -65,6 +72,28 @@ public interface ITaskRepository
/// <summary>Ändert den Status direkt (Tool-Aktionen, Auto-Dispatch, Eskalation).</summary>
Task<bool> SetStatusAsync(string id, TaskItemStatus status, DateTime now, CancellationToken ct);
/// <summary>
/// #716: The budget refused a run of a one-time task. Releases the claim, restores the
/// occurrence (the task is still due) and defers it to <paramref name="deferredUntil"/>.
/// </summary>
Task<bool> CarryOverAsync(
string id, string claimToken, string? previousOccurrence, DateTime deferredUntil,
DateTime now, CancellationToken ct);
/// <summary>
/// #716: The budget refused a run of a recurring task. Releases the claim back to
/// <c>todo</c>; the occurrence stays consumed. With <paramref name="catchUpUntil"/> a
/// single catch-up run is scheduled for then — an already pending one is kept.
/// </summary>
Task<bool> RecordBudgetMissAsync(
string id, string claimToken, DateTime? catchUpUntil, DateTime now, CancellationToken ct);
/// <summary>
/// #716: A run took place — clears the miss counter, a pending catch-up and the
/// deferral. <see cref="TaskItem.LastBudgetMissAt"/> stays as history.
/// </summary>
Task<bool> ClearBudgetMissAsync(string id, DateTime now, CancellationToken ct);
/// <summary>
/// Reconciliation beim Start: verwaiste Claims (älter als <paramref name="leaseCutoff"/>)
/// lösen und die betroffenen Aufgaben von <see cref="TaskItemStatus.InProgress"/> zurück
@@ -193,6 +193,68 @@ public sealed class SqliteTaskRepository : ITaskRepository
return await cmd.ExecuteNonQueryAsync(ct) > 0;
}, ct);
public Task<bool> CarryOverAsync(
string id, string claimToken, string? previousOccurrence, DateTime deferredUntil,
DateTime now, CancellationToken ct)
=> _storage.WriteAsync(async conn =>
{
using var cmd = conn.CreateCommand();
cmd.CommandText = """
UPDATE Tasks
SET Status = 'todo', ClaimToken = NULL, ClaimedAt = NULL,
LastOccurrence = @occ, UpdatedAt = @now,
BudgetMissCount = BudgetMissCount + 1, LastBudgetMissAt = @now,
DeferredUntil = @deferred
WHERE Id = @id AND ClaimToken = @token
""";
cmd.Parameters.AddWithValue("@occ", (object?)previousOccurrence ?? DBNull.Value);
cmd.Parameters.AddWithValue("@now", Format(now));
cmd.Parameters.AddWithValue("@deferred", Format(deferredUntil));
cmd.Parameters.AddWithValue("@id", id);
cmd.Parameters.AddWithValue("@token", claimToken);
return await cmd.ExecuteNonQueryAsync(ct) > 0;
}, ct);
public Task<bool> RecordBudgetMissAsync(
string id, string claimToken, DateTime? catchUpUntil, DateTime now, CancellationToken ct)
=> _storage.WriteAsync(async conn =>
{
using var cmd = conn.CreateCommand();
// A pending catch-up keeps its date: several misses still mean one catch-up run.
cmd.CommandText = """
UPDATE Tasks
SET Status = 'todo', ClaimToken = NULL, ClaimedAt = NULL, UpdatedAt = @now,
BudgetMissCount = BudgetMissCount + 1, LastBudgetMissAt = @now,
CatchUpPending = CASE WHEN @catchUp = 1 THEN 1 ELSE CatchUpPending END,
DeferredUntil = CASE WHEN @catchUp = 1 AND CatchUpPending = 0 THEN @deferred
ELSE DeferredUntil END
WHERE Id = @id AND ClaimToken = @token
""";
cmd.Parameters.AddWithValue("@now", Format(now));
cmd.Parameters.AddWithValue("@catchUp", catchUpUntil is null ? 0 : 1);
cmd.Parameters.AddWithValue("@deferred", catchUpUntil is { } until ? Format(until) : DBNull.Value);
cmd.Parameters.AddWithValue("@id", id);
cmd.Parameters.AddWithValue("@token", claimToken);
return await cmd.ExecuteNonQueryAsync(ct) > 0;
}, ct);
public Task<bool> ClearBudgetMissAsync(string id, DateTime now, CancellationToken ct)
=> _storage.WriteAsync(async conn =>
{
using var cmd = conn.CreateCommand();
cmd.CommandText = """
UPDATE Tasks
SET BudgetMissCount = 0, CatchUpPending = 0, DeferredUntil = NULL, UpdatedAt = @now
WHERE Id = @id
""";
cmd.Parameters.AddWithValue("@now", Format(now));
cmd.Parameters.AddWithValue("@id", id);
return await cmd.ExecuteNonQueryAsync(ct) > 0;
}, ct);
public Task<int> ReleaseStaleClaimsAsync(DateTime leaseCutoff, DateTime now, CancellationToken ct)
=> _storage.WriteAsync(async conn =>
{
@@ -208,6 +270,22 @@ public sealed class SqliteTaskRepository : ITaskRepository
return await cmd.ExecuteNonQueryAsync(ct);
}, ct);
public async Task<IReadOnlyList<TaskItem>> ListAllAsync(bool includeArchived, CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT " + Columns + " FROM Tasks"
+ (includeArchived ? "" : " WHERE Status <> 'archived'")
+ " ORDER BY Priority DESC, UpdatedAt DESC";
var results = new List<TaskItem>();
await using var reader = await cmd.ExecuteReaderAsync(ct);
while (await reader.ReadAsync(ct))
results.Add(Read(reader));
return results;
}
public async Task<IReadOnlyList<TaskItem>> ListRunnableAsync(CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
@@ -252,7 +330,8 @@ public sealed class SqliteTaskRepository : ITaskRepository
private const string Columns =
"Id, Title, Status, Type, Priority, Assignee, WhenKind, WhenValue, WhenTz, " +
"RequireApproval, Acceptance, BlockedBy, OnlyWhenMarketOpen, Body, FileName, " +
"ToolName, JobTypeId, LastOccurrence, ClaimToken, ClaimedAt, CreatedAt, UpdatedAt";
"ToolName, JobTypeId, LastOccurrence, ClaimToken, ClaimedAt, CreatedAt, UpdatedAt, "
+ "BudgetMissCount, LastBudgetMissAt, DeferredUntil, CatchUpPending";
private static void BindDefinition(SqliteCommand cmd, TaskItem task)
{
@@ -318,7 +397,11 @@ public sealed class SqliteTaskRepository : ITaskRepository
ClaimToken = r.IsDBNull(18) ? null : r.GetString(18),
ClaimedAt = r.IsDBNull(19) ? null : Parse(r.GetString(19)),
CreatedAt = Parse(r.GetString(20)),
UpdatedAt = Parse(r.GetString(21))
UpdatedAt = Parse(r.GetString(21)),
BudgetMissCount = r.GetInt32(22),
LastBudgetMissAt = r.IsDBNull(23) ? null : Parse(r.GetString(23)),
DeferredUntil = r.IsDBNull(24) ? null : Parse(r.GetString(24)),
CatchUpPending = r.GetInt32(25) != 0
};
}
+39
View File
@@ -205,6 +205,32 @@ public sealed record TaskItem
public DateTime CreatedAt { get; init; }
public DateTime UpdatedAt { get; init; }
// ─── Budget misses (#716) ───
/// <summary>
/// How many times in a row a run of this task was refused because the budget was
/// exhausted. Reset by the next run that actually takes place.
/// </summary>
public int BudgetMissCount { get; init; }
/// <summary>When the last budget refusal happened (UTC). Kept after a reset, as history.</summary>
public DateTime? LastBudgetMissAt { get; init; }
/// <summary>
/// Not before this point (UTC). Set after a budget refusal to the start of the next
/// local day: retrying within the same day only produces further refusals.
/// </summary>
public DateTime? DeferredUntil { get; init; }
/// <summary>
/// A recurring task under <see cref="BudgetMissPolicy.CatchUpOnce"/> missed at least
/// one occurrence; all of them are caught up by a single run.
/// </summary>
public bool CatchUpPending { get; init; }
/// <summary>A one-time task that was refused for lack of budget and carried over.</summary>
public bool IsCarriedOver => !IsRecurring && BudgetMissCount > 0;
/// <summary>
/// Wiederkehrend? Solche Tasks kehren nach dem Feuern auf <c>todo</c> zurück statt auf
/// <c>done</c> — sonst würde ein Cron-Task nur ein einziges Mal laufen. Ein
@@ -215,6 +241,19 @@ public sealed record TaskItem
|| When is { Kind: TaskWhenKind.Cron or TaskWhenKind.Every };
}
/// <summary>
/// #716: What happens to a run of a recurring task that the budget refused. One-time
/// tasks are always carried over — they are real work that must not disappear.
/// </summary>
public enum BudgetMissPolicy
{
/// <summary>The occurrence lapses and is recorded; the next regular one comes anyway.</summary>
Expire,
/// <summary>All missed occurrences are caught up by one run on the next day.</summary>
CatchUpOnce
}
/// <summary>Suchkriterien für <see cref="ITaskRepository.ListAsync"/>.</summary>
public sealed record TaskQuery
{
+246 -8
View File
@@ -12,10 +12,17 @@ public enum TaskLauf
Gescheitert,
/// <summary>
/// Der Lauf fand gar nicht statt — Budget erschöpft, Instanz im Herunterfahren (#643).
/// Der Lauf fand gar nicht statt — Instanz im Herunterfahren (#643).
/// Eine einmalige Aufgabe bleibt dann fällig, statt als gelaufen liegenzubleiben.
/// </summary>
NichtGelaufen
NichtGelaufen,
/// <summary>
/// #716: The run did not take place because the budget is exhausted. Unlike a
/// shutdown, retrying the same day is pointless — the scanner defers to the next day
/// and applies the agent's <see cref="BudgetMissPolicy"/>.
/// </summary>
KeinBudget
}
/// <summary>
@@ -37,6 +44,30 @@ public interface IMarketCalendar
bool IsOpen(DateTime nowUtc);
}
/// <summary>
/// #716: Decides whether a late run — carried-over work or a catch-up — may start now.
/// Without one, late runs only wait for the regular ones of the same tick.
/// </summary>
public interface ILateRunGate
{
Task<bool> HasRoomAsync(TaskItem lateTask, IReadOnlyList<TaskItem> runnable, DateTime nowUtc, CancellationToken ct);
}
/// <summary>
/// #716: The day ahead — where the budget will not reach the end of today's schedule.
/// </summary>
public interface IBudgetOutlook
{
Task<IReadOnlyList<BudgetShortfall>> ShortfallsAsync(IReadOnlyList<TaskItem> runnable, DateTime nowUtc, CancellationToken ct);
}
/// <summary>
/// #716: An agent whose budget runs out before today's schedule does — from when on, and
/// how many planned runs will be refused. <see cref="AgentId"/> is empty for the instance
/// limit.
/// </summary>
public sealed record BudgetShortfall(string AgentId, DateTime ExhaustsAtUtc, int RunsRefused, int RunsPlanned);
/// <summary>Bis C1 kommt: der Markt gilt als immer offen.</summary>
public sealed class AlwaysOpenMarketCalendar : IMarketCalendar
{
@@ -98,6 +129,26 @@ public sealed class TaskScanner : IAsyncDisposable
/// <summary>How many due tasks had to wait at the last tick — logged only on change.</summary>
private int _lastWaiting;
/// <summary>#716: What the budget refused, per agent, for the day summary.</summary>
private readonly Func<TaskItem, BudgetMissPolicy> _missPolicy;
private readonly Dictionary<string, (int Misses, int CarriedOver, int CatchUps)> _missesToday = new(StringComparer.OrdinalIgnoreCase);
private DateOnly? _missDay;
private readonly ILateRunGate? _lateRunGate;
private readonly IBudgetOutlook? _outlook;
/// <summary>How often the day ahead is looked at — a forecast, not a watchdog.</summary>
public static readonly TimeSpan OutlookInterval = TimeSpan.FromMinutes(30);
private DateTime _lastOutlookAt = DateTime.MinValue;
/// <summary>Agents already warned about today, with the day — once per agent and day.</summary>
private readonly Dictionary<string, DateOnly> _shortfallWarned = new(StringComparer.OrdinalIgnoreCase);
/// <summary>How many late runs waited for budget headroom at the last tick.</summary>
private int _lastWaitingForHeadroom;
private readonly CancellationTokenSource _cts = new();
private Task? _loop;
@@ -109,7 +160,10 @@ public sealed class TaskScanner : IAsyncDisposable
TimeProvider? clock = null,
TimeSpan? tick = null,
TimeSpan? lease = null,
int maxParallelRuns = DefaultMaxParallelRuns)
int maxParallelRuns = DefaultMaxParallelRuns,
Func<TaskItem, BudgetMissPolicy>? budgetMissPolicy = null,
ILateRunGate? lateRunGate = null,
IBudgetOutlook? budgetOutlook = null)
{
_repo = repo;
_dispatcher = dispatcher;
@@ -119,6 +173,9 @@ public sealed class TaskScanner : IAsyncDisposable
_tick = tick ?? TimeSpan.FromSeconds(60);
_lease = lease ?? TimeSpan.FromMinutes(15);
_maxParallelRuns = Math.Max(1, maxParallelRuns);
_missPolicy = budgetMissPolicy ?? (_ => BudgetMissPolicy.Expire);
_lateRunGate = lateRunGate;
_outlook = budgetOutlook;
}
/// <summary>How many scheduled runs may be in flight at once (#682).</summary>
@@ -226,7 +283,10 @@ public sealed class TaskScanner : IAsyncDisposable
// #682: ungekappt. Vorher ListAsync mit Limit 500 — darüber fielen Termine lautlos aus.
var candidates = await _repo.ListRunnableAsync(ct);
var due = new List<(TaskItem Task, string Occurrence)>();
SummarizeMissesIfNewDay(now);
await WarnAboutShortfallsAsync(candidates, now, ct);
var due = new List<(TaskItem Task, string Occurrence, bool Late)>();
foreach (var task in candidates)
{
if (task.Status != TaskItemStatus.Todo)
@@ -269,26 +329,42 @@ public sealed class TaskScanner : IAsyncDisposable
if (_zurueckgestellt.TryGetValue(task.Id, out var ab) && now < ab)
continue;
var occ = TaskSchedule.DueOccurrence(task, now);
if (occ is null)
// #716: Refused for lack of budget — not before the next day. For a recurring
// task this only holds back the catch-up; its regular occurrences keep coming.
var deferred = task.DeferredUntil is { } until && now < until;
if (deferred && !task.IsRecurring)
continue;
if (task.OnlyWhenMarketOpen && !_market.IsOpen(now))
continue;
due.Add((task, occ));
var occ = TaskSchedule.DueOccurrence(task, now);
if (occ is not null)
{
// A regular occurrence also covers a pending catch-up: one run, not two.
due.Add((task, occ, Late: task.IsCarriedOver));
continue;
}
if (task.CatchUpPending && !deferred)
due.Add((task, Iso(now), Late: true));
}
// #716: The new day has right of way — carried-over and catch-up runs come after
// everything regularly due. Within each group: priority, then the oldest occurrence.
// Occurrence keys are ISO-8601 UTC ("O"), so ordinal order is time order.
due.Sort((a, b) =>
{
if (a.Late != b.Late)
return a.Late ? 1 : -1;
var byPriority = b.Task.Priority.CompareTo(a.Task.Priority);
return byPriority != 0 ? byPriority : string.CompareOrdinal(a.Occurrence, b.Occurrence);
});
var launched = new List<Task>();
var waiting = 0;
foreach (var (task, occ) in due)
var waitingForHeadroom = 0;
foreach (var (task, occ, late) in due)
{
// OpenCount also counts runs started by hand (RunTaskNowAsync) — they take a
// slot, even though they did not wait for one.
@@ -298,6 +374,14 @@ public sealed class TaskScanner : IAsyncDisposable
continue;
}
// #716: Late work only uses what the rest of the day does not need.
if (late && _lateRunGate is not null
&& !await _lateRunGate.HasRoomAsync(task, candidates, now, ct))
{
waitingForHeadroom++;
continue;
}
var token = Guid.NewGuid().ToString("N");
if (await _repo.TryClaimAsync(task.Id, occ, token, now, leaseCutoff, ct))
launched.Add(Track(RunLoggedAsync(task, token, ct)));
@@ -316,6 +400,16 @@ public sealed class TaskScanner : IAsyncDisposable
_lastWaiting = waiting;
}
if (waitingForHeadroom != _lastWaitingForHeadroom)
{
if (waitingForHeadroom > 0)
_logger.LogInformation(
"{Waiting} carried-over or catch-up run(s) waiting for budget headroom — "
+ "the rest of today's schedule comes first",
waitingForHeadroom);
_lastWaitingForHeadroom = waitingForHeadroom;
}
return launched;
}
@@ -413,6 +507,12 @@ public sealed class TaskScanner : IAsyncDisposable
var now = _clock.GetUtcNow().UtcDateTime;
var ok = lauf == TaskLauf.Erledigt;
if (lauf == TaskLauf.KeinBudget)
{
await HandleBudgetMissAsync(task, token, now, ct);
return;
}
// #643: Fand der Lauf einer einmaligen Aufgabe gar nicht statt, gilt ihr Termin als
// nicht verbraucht. Vorher blieb sie mit gesetztem Marker für immer auf todo liegen.
if (lauf == TaskLauf.NichtGelaufen && !task.IsRecurring)
@@ -443,10 +543,148 @@ public sealed class TaskScanner : IAsyncDisposable
await _repo.CompleteClaimAsync(task.Id, token, final, now, ct);
// #716: A run took place — earlier budget misses are settled.
if (lauf != TaskLauf.NichtGelaufen
&& (task.BudgetMissCount > 0 || task.CatchUpPending || task.DeferredUntil is not null))
await _repo.ClearBudgetMissAsync(task.Id, now, ct);
if (ok && final == TaskItemStatus.Done)
await UnblockDependentsAsync(task.Id, now, ct);
}
/// <summary>
/// #716: The budget refused the run. A one-time task is carried over to the next day
/// with its occurrence intact; a recurring one follows the agent's policy — the
/// occurrence lapses, and under <see cref="BudgetMissPolicy.CatchUpOnce"/> one
/// catch-up run is scheduled for the next day.
/// </summary>
private async Task HandleBudgetMissAsync(TaskItem task, string token, DateTime now, CancellationToken ct)
{
var nextDay = StartOfNextLocalDay(now);
var agent = TaskAssignee.AgentId(task.Assignee);
var carried = false;
var catchUp = false;
if (!task.IsRecurring)
{
carried = await _repo.CarryOverAsync(task.Id, token, task.LastOccurrence, nextDay, now, ct);
}
else
{
catchUp = _missPolicy(task) == BudgetMissPolicy.CatchUpOnce;
await _repo.RecordBudgetMissAsync(task.Id, token, catchUp ? nextDay : null, now, ct);
}
lock (_missesToday)
{
var entry = _missesToday.GetValueOrDefault(agent);
_missesToday[agent] = (entry.Misses + 1,
entry.CarriedOver + (carried ? 1 : 0),
entry.CatchUps + (catchUp && !task.CatchUpPending ? 1 : 0));
}
_logger.LogInformation(
"Task {TaskId} refused for lack of budget — {Outcome}",
task.Id,
carried ? $"carried over, due again from {nextDay:yyyy-MM-dd HH:mm} UTC"
: catchUp ? "occurrence lapsed, one catch-up run scheduled for the next day"
: "occurrence lapsed (policy: expire)");
}
/// <summary>
/// #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.
/// </summary>
private async Task WarnAboutShortfallsAsync(IReadOnlyList<TaskItem> runnable, DateTime nowUtc, CancellationToken ct)
{
if (_outlook is null || nowUtc - _lastOutlookAt < OutlookInterval)
return;
_lastOutlookAt = nowUtc;
IReadOnlyList<BudgetShortfall> 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);
}
}
/// <summary>
/// #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.
/// </summary>
private void SummarizeMissesIfNewDay(DateTime nowUtc)
{
var today = DateOnly.FromDateTime(TimeZoneInfo.ConvertTimeFromUtc(nowUtc, _clock.LocalTimeZone));
List<KeyValuePair<string, (int Misses, int CarriedOver, int CatchUps)>> summary;
DateOnly? day;
lock (_missesToday)
{
if (_missDay is null)
{
_missDay = today;
return;
}
if (_missDay == today)
return;
day = _missDay;
summary = [.. _missesToday];
_missesToday.Clear();
_missDay = today;
}
foreach (var (agent, counts) in summary)
{
_logger.LogWarning(
"Budget summary {Day:yyyy-MM-dd} for agent {AgentId}: {Misses} run(s) refused for lack of "
+ "budget, {Carried} task(s) carried over, {CatchUps} catch-up run(s) scheduled",
day, agent.Length == 0 ? "-" : agent, counts.Misses, counts.CarriedOver, counts.CatchUps);
}
}
/// <summary>
/// Local midnight after <paramref name="nowUtc"/>, in UTC. The budget day follows the
/// machine's local date, like <see cref="Budget.BudgetGuard"/>.
/// </summary>
private DateTime StartOfNextLocalDay(DateTime nowUtc)
{
var zone = _clock.LocalTimeZone;
var local = TimeZoneInfo.ConvertTimeFromUtc(nowUtc, zone);
var midnight = DateTime.SpecifyKind(local.Date.AddDays(1), DateTimeKind.Unspecified);
// A midnight skipped by a DST change does not exist; take the next valid minute.
while (zone.IsInvalidTime(midnight))
midnight = midnight.AddMinutes(1);
return TimeZoneInfo.ConvertTimeToUtc(midnight, zone);
}
private static string Iso(DateTime utc) => utc.ToUniversalTime().ToString("O");
/// <summary>
/// Auto-Dispatch: Wird eine Aufgabe fertig, werden die auf sie wartenden Aufgaben
/// freigegeben, sobald <b>alle</b> ihre Blocker erledigt sind.
@@ -201,6 +201,79 @@ public static class TaskSchedule
}
}
/// <summary>
/// #716: How many runs the task will still trigger from <paramref name="fromUtc"/> up to
/// (excluding) <paramref name="toUtc"/> — including one that is due right now. The
/// basis for "how much budget does the rest of the day still need".
/// </summary>
public static int OccurrencesBetween(TaskItem task, DateTime fromUtc, DateTime toUtc, int cap = 1_440)
=> OccurrenceTimesBetween(task, fromUtc, toUtc, cap).Count;
/// <summary>
/// #716: When the task will run from <paramref name="fromUtc"/> up to (excluding)
/// <paramref name="toUtc"/>. A run due right now is listed at <paramref name="fromUtc"/>.
/// </summary>
public static IReadOnlyList<DateTime> 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<DateTime>();
if (toUtc <= fromUtc)
return times;
var dueNow = DueOccurrence(task, fromUtc) is not null;
if (dueNow)
times.Add(fromUtc);
if (task.When is not { } when)
return times; // one-time without a schedule: due until it ran
switch (when.Kind)
{
case TaskWhenKind.At:
{
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 times;
var next = dueNow
? fromUtc + interval
: NextOccurrenceUtc(task, fromUtc) ?? toUtc;
while (times.Count < cap && next < toUtc)
{
times.Add(next);
next += interval;
}
return times;
}
case TaskWhenKind.Cron:
{
if (!TryPrepareCron(when, out var cron, out var tz))
return times;
var cursor = fromUtc;
while (times.Count < cap && NextCronUtc(cron, tz, cursor) is { } next && next < toUtc)
{
if (next > fromUtc)
times.Add(next);
cursor = next;
}
return times;
}
default:
return times;
}
}
private static bool TryPrepareCron(TaskWhen when, out CronExpression cron, out TimeZoneInfo tz)
{
cron = null!;
@@ -31,6 +31,10 @@ public sealed class TaskboardService
public Task<IReadOnlyList<TaskItem>> ListAsync(TaskQuery query, CancellationToken ct)
=> _repo.ListAsync(query, ct);
/// <summary>The whole board, without the cap of <see cref="ListAsync"/>.</summary>
public Task<IReadOnlyList<TaskItem>> ListAllAsync(CancellationToken ct, bool includeArchived = false)
=> _repo.ListAllAsync(includeArchived, ct);
// ─── Anlegen ───
/// <summary>
@@ -104,9 +108,9 @@ public sealed class TaskboardService
if (task is not null) { seen.Add(task.Id); imported++; }
}
// Verwaiste Zeilen archivieren. Der Deckel von 500 ist für den Start unkritisch;
// wächst das Board darüber hinaus, gehört ohnehin das Aufräumen (siehe Konzept) her.
var known = await _repo.ListAsync(new TaskQuery { IncludeArchived = true, Limit = 500 }, ct);
// Verwaiste Zeilen archivieren. Ungekappt (#716): Vorher sah dieser Abgleich nur die
// ersten 500 Zeilen, und darüber blieben verwaiste Zeilen still stehen.
var known = await _repo.ListAllAsync(includeArchived: true, ct);
foreach (var task in known)
if (task.Status != TaskItemStatus.Archived && !seen.Contains(task.Id))
await _repo.SetStatusAsync(task.Id, TaskItemStatus.Archived, DateTime.UtcNow, ct);
@@ -1,3 +1,4 @@
using ClawdDotNet.Core.Tasks;
using System.Collections.ObjectModel;
using Avalonia.Threading;
using ClawdDotNet.App;
@@ -77,6 +78,13 @@ public sealed partial class AgentItemViewModel : ViewModelBase
[ObservableProperty]
private long _dailyTokens;
/// <summary>#716: What happens to a recurring task's run when the budget is exhausted.</summary>
[ObservableProperty]
private string _whenExhausted = "expire";
/// <summary>The values <see cref="AgentBudget.WhenExhausted"/> understands.</summary>
public IReadOnlyList<string> WhenExhaustedOptions { get; } = ["expire", "catchUpOnce"];
public ObservableCollection<ToolItemViewModel> AvailableTools { get; } = [];
/// <summary>
@@ -101,6 +109,7 @@ public sealed partial class AgentItemViewModel : ViewModelBase
_timeoutSeconds = config.LoopGuard.TimeoutSeconds;
_dailyCostUsd = config.Budget.DailyCostUsd;
_dailyTokens = config.Budget.DailyTokens;
_whenExhausted = config.Budget.MissPolicy == BudgetMissPolicy.CatchUpOnce ? "catchUpOnce" : "expire";
foreach (var tool in KnownTools)
{
@@ -137,6 +146,7 @@ public sealed partial class AgentItemViewModel : ViewModelBase
Config.LoopGuard.TimeoutSeconds = TimeoutSeconds;
Config.Budget.DailyCostUsd = DailyCostUsd;
Config.Budget.DailyTokens = DailyTokens;
Config.Budget.WhenExhausted = WhenExhausted;
}
}
@@ -208,6 +208,15 @@
<TextBox Text="{Binding DailyTokens}" />
</StackPanel>
</Grid>
<StackPanel>
<TextBlock Classes="label" Text="{Binding L[desktop.agents.whenExhausted]}" />
<ComboBox HorizontalAlignment="Stretch"
ItemsSource="{Binding WhenExhaustedOptions}"
SelectedItem="{Binding WhenExhausted}" />
<TextBlock Classes="caption" TextWrapping="Wrap"
Text="{Binding L[desktop.agents.whenExhaustedHint]}" />
</StackPanel>
</StackPanel>
</StackPanel>
+2
View File
@@ -64,6 +64,8 @@
"desktop.agents.maxToolResultChars": "Max. Tool-Ergebnis (Zeichen)",
"desktop.agents.dailyCost": "Tagesbudget ($ USD)",
"desktop.agents.dailyTokens": "Tagesbudget (Tokens)",
"desktop.agents.whenExhausted": "Wenn das Budget erschöpft ist (wiederkehrende Aufgaben)",
"desktop.agents.whenExhaustedHint": "expire: Der verpasste Lauf verfällt und wird an der Aufgabe vermerkt. catchUpOnce: Alle verpassten Läufe werden am nächsten Tag durch einen Lauf nachgeholt, nach der regulären Arbeit des Tages. Einmalige Aufgaben werden immer auf den nächsten Tag mitgenommen.",
"desktop.agents.allowedToolsFor": "ZUGELASSENE WERKZEUGE FÜR {0}",
"desktop.agents.thTool": "WERKZEUG",
"desktop.agents.thDescription": "BESCHREIBUNG",
+2
View File
@@ -64,6 +64,8 @@
"desktop.agents.maxToolResultChars": "Max Tool Result (chars)",
"desktop.agents.dailyCost": "Daily Budget ($ USD)",
"desktop.agents.dailyTokens": "Daily Budget (Tokens)",
"desktop.agents.whenExhausted": "When the budget is exhausted (recurring tasks)",
"desktop.agents.whenExhaustedHint": "expire: the missed run lapses and is recorded on the task. catchUpOnce: all missed runs are caught up by one run the next day, after the day's regular work. One-time tasks are always carried over to the next day.",
"desktop.agents.allowedToolsFor": "ALLOWED TOOLS FOR {0}",
"desktop.agents.thTool": "TOOL",
"desktop.agents.thDescription": "DESCRIPTION",
@@ -192,6 +192,12 @@
"tasks.saved": "Aufgabe gespeichert.",
"tasks.deleted": "Aufgabe gelöscht.",
"tasks.runTriggered": "Lauf ausgelöst.",
"tasks.nextRun": "Nächster",
"tasks.lastRun": "Letzter",
"tasks.budgetMissed": "Budget erschöpft: {n} Lauf/Läufe abgelehnt",
"tasks.budgetMissedLast": "zuletzt {time}",
"tasks.budgetCarriedOver": "mitgenommen, wieder fällig ab {time}",
"tasks.budgetCatchUp": "ein Nachholauf ab {time}",
"services.title": "Dienste",
"services.newService": "Neuer Dienst",
@@ -192,6 +192,12 @@
"tasks.saved": "Task saved.",
"tasks.deleted": "Task deleted.",
"tasks.runTriggered": "Run triggered.",
"tasks.nextRun": "Next",
"tasks.lastRun": "Last",
"tasks.budgetMissed": "Budget exhausted: {n} run(s) refused",
"tasks.budgetMissedLast": "last at {time}",
"tasks.budgetCarriedOver": "carried over, due again from {time}",
"tasks.budgetCatchUp": "one catch-up run from {time}",
"services.title": "Services",
"services.newService": "New Service",
@@ -163,11 +163,11 @@ function erstelleAufgabenKarte(task) {
zeilen.push(
erstelleElement("p", {
klasse: "gedaempft text-klein",
text: `Next: ${task.nextRunUtc ? formatiereDatumZeit(task.nextRunUtc) : "—"}`
text: `${t("tasks.nextRun")}: ${task.nextRunUtc ? formatiereDatumZeit(task.nextRunUtc) : "—"}`
})
);
const letzter = erstelleElement("p", { klasse: "gedaempft text-klein" }, "Last: ");
const letzter = erstelleElement("p", { klasse: "gedaempft text-klein" }, `${t("tasks.lastRun")}: `);
if (task.lastRun) {
letzter.append(
`${formatiereDatumZeit(task.lastRun.occurredAt)} `,
@@ -184,6 +184,19 @@ function erstelleAufgabenKarte(task) {
zeilen.push(erstelleElement("p", { klasse: "text-klein", text: `⚠ ${task.hinweis}` }));
}
// #716: What the budget refused, and what happens next.
const ausfall = task.budgetMiss;
if (ausfall) {
const teile = [t("tasks.budgetMissed", { n: ausfall.count })];
if (ausfall.lastAt) teile.push(t("tasks.budgetMissedLast", { time: formatiereDatumZeit(ausfall.lastAt) }));
if (ausfall.carriedOver && ausfall.deferredUntil) {
teile.push(t("tasks.budgetCarriedOver", { time: formatiereDatumZeit(ausfall.deferredUntil) }));
} else if (ausfall.catchUpPending) {
teile.push(t("tasks.budgetCatchUp", { time: ausfall.deferredUntil ? formatiereDatumZeit(ausfall.deferredUntil) : "—" }));
}
zeilen.push(erstelleElement("p", { klasse: "text-klein", text: `⚠ ${teile.join(" · ")}` }));
}
const aktionen = erstelleElement("div", {
klasse: "gruppe-horizontal",
stil: "margin-top: 8px; justify-content: flex-end;"
@@ -210,7 +210,9 @@ public sealed class TaskboardTool : IAgentTool
private static async Task<List<(TaskItem Task, TaskTitleMatch Match)>> SucheAehnlicheAsync(
TaskboardService board, string title, string? exceptId, CancellationToken ct)
{
var alle = await board.ListAsync(new TaskQuery { Limit = 500 }, ct);
// Uncapped (#716): with a limit of 500, a duplicate among older, low-priority tasks
// stayed invisible once the board grew beyond that.
var alle = await board.ListAllAsync(ct);
return TaskSimilarity.FindOpen(alle, title, exceptId);
}
+8 -1
View File
@@ -116,7 +116,14 @@ public sealed record TaskDto(
TaskWhenDto? When = null,
DateTime? NextRunUtc = null,
TaskLastRunDto? LastRun = null,
string? Hinweis = null);
string? Hinweis = null,
TaskBudgetMissDto? BudgetMiss = null);
/// <summary>
/// #716: Runs of this task the budget refused — only present while something is open
/// (misses since the last run that took place, a pending catch-up or a deferral).
/// </summary>
public sealed record TaskBudgetMissDto(int Count, DateTime? LastAt, DateTime? DeferredUntil, bool CatchUpPending, bool CarriedOver);
/// <summary>Der letzte Lauf einer Aufgabe (#594) - aus dem Beleg, samt Kosten.</summary>
public sealed record TaskLastRunDto(DateTime OccurredAt, string Status, decimal CostUsd, bool CostIsKnown, string ResultRef);
+4
View File
@@ -129,6 +129,10 @@ internal static class Mapping
task.UpdatedAt,
task.When is { } when
? new TaskWhenDto(when.Kind.ToString().ToLowerInvariant(), when.Value, when.TimeZone)
: null,
BudgetMiss: task.BudgetMissCount > 0 || task.CatchUpPending || task.DeferredUntil is not null
? new TaskBudgetMissDto(task.BudgetMissCount, task.LastBudgetMissAt, task.DeferredUntil,
task.CatchUpPending, task.IsCarriedOver)
: null);
/// <summary>
@@ -0,0 +1,214 @@
using ClawdDotNet.Core.Accounting;
using ClawdDotNet.Core.Budget;
using ClawdDotNet.Core.Config;
using ClawdDotNet.Core.Tasks;
using Shouldly;
namespace ClawdDotNet.Core.Tests.Budget;
/// <summary>
/// #716, stage 3: late work — carried-over tasks and catch-ups — only uses budget the rest
/// of the day does not need. The new day has right of way.
/// </summary>
public sealed class BudgetHeadroomTests
{
// 2027-03-10 14:00 UTC; the test clock's local zone is UTC, so the day ends at 24:00 UTC.
private static readonly DateTime Now = new(2027, 3, 10, 14, 0, 0, DateTimeKind.Utc);
private static readonly DateTime CreatedEarlier = new(2027, 3, 1, 0, 0, 0, DateTimeKind.Utc);
private static AgentConfig Agent(decimal dailyCost = 1.00m)
=> new() { AgentId = "worker", Budget = new AgentBudget { DailyCostUsd = dailyCost } };
/// <summary>Runs averaging 0.05 USD.</summary>
private static List<RunUsage> History(decimal cost = 0.05m)
=> [.. Enumerable.Range(0, 5).Select(_ => new RunUsage { AgentId = "worker", PromptTokens = 1_000, CostUsd = cost, CostIsKnown = true })];
private static TaskItem Cron(string id, string expression, string? lastOccurrence = null) => new()
{
Id = id, Title = id, Assignee = "@worker", CreatedAt = CreatedEarlier, LastOccurrence = lastOccurrence,
When = new TaskWhen { Kind = TaskWhenKind.Cron, Value = expression, TimeZone = "UTC" }
};
private static TaskItem Carried(string id) => new()
{
Id = id, Title = id, Assignee = "@worker", CreatedAt = CreatedEarlier, BudgetMissCount = 1
};
private static BudgetHeadroom Headroom(decimal used, AgentConfig agent, InstanceBudget? instance = null, decimal perRun = 0.05m)
=> new(new BudgetGuard(new FakeUsage(used, History(perRun))),
() => [agent], () => instance ?? InstanceBudget.Unlimited, new UtcClock(Now));
// ═══════════════════ counting the rest of the day ═══════════════════
[Fact]
public void Hourly_from_14_30_until_midnight_is_nine_more_runs()
{
// 15:00 … 23:00. The 14:00 run is done (marker), so nothing is due right now.
var task = Cron("t", "0 * * * *", lastOccurrence: "2027-03-10T14:00:00.0000000Z");
TaskSchedule.OccurrencesBetween(task, Now.AddMinutes(30), Now.Date.AddDays(1)).ShouldBe(9);
}
[Fact]
public void A_run_due_right_now_counts_as_well()
{
// Last run 13:00, so the 14:00 run is due now; then 15:00 … 23:00.
var task = Cron("t", "0 * * * *", lastOccurrence: "2027-03-10T13:00:00.0000000Z");
TaskSchedule.OccurrencesBetween(task, Now.AddMinutes(1), Now.Date.AddDays(1)).ShouldBe(10);
}
[Fact]
public void A_one_time_task_counts_once_until_it_ran()
{
var open = new TaskItem { Id = "t", CreatedAt = CreatedEarlier };
var done = open with { LastOccurrence = "2027-03-10T09:00:00.0000000Z" };
TaskSchedule.OccurrencesBetween(open, Now, Now.AddHours(1)).ShouldBe(1);
TaskSchedule.OccurrencesBetween(done, Now, Now.AddHours(1)).ShouldBe(0);
}
[Fact]
public void An_interval_counts_its_remaining_ticks()
{
var task = new TaskItem
{
Id = "t", CreatedAt = CreatedEarlier, LastOccurrence = "2027-03-10T13:45:00.0000000Z",
When = new TaskWhen { Kind = TaskWhenKind.Every, Value = "30m", TimeZone = "UTC" }
};
// 14:15, 14:45, 15:15, 15:45 before 16:00.
TaskSchedule.OccurrencesBetween(task, Now, Now.AddHours(2)).ShouldBe(4);
}
// ═══════════════════ headroom ═══════════════════
[Fact]
public async Task Late_work_fits_when_the_rest_of_the_day_leaves_room()
{
// Limit 1.00, used 0.30, margin 0.10. Rest of the day: 10 hourly runs à 0.05 = 0.50.
// Left: 1.00 - 0.30 - 0.50 - 0.10 = 0.10 ≥ 0.05 for the late run.
var hourly = Cron("t-hourly", "0 * * * *", lastOccurrence: "2027-03-10T13:00:00.0000000Z");
var late = Carried("t-late");
var room = await Headroom(used: 0.30m, Agent()).HasRoomAsync(late, [hourly, late], Now, default);
room.ShouldBeTrue();
}
[Fact]
public async Task Late_work_waits_when_the_rest_of_the_day_needs_the_budget()
{
// Same, but 0.40 used: 1.00 - 0.40 - 0.50 - 0.10 = 0.00 < 0.05.
var hourly = Cron("t-hourly", "0 * * * *", lastOccurrence: "2027-03-10T13:00:00.0000000Z");
var late = Carried("t-late");
var room = await Headroom(used: 0.40m, Agent()).HasRoomAsync(late, [hourly, late], Now, default);
room.ShouldBeFalse("the day's own schedule has right of way");
}
[Fact]
public async Task Other_late_work_does_not_count_as_the_days_demand()
{
// Ten carried-over tasks do not reserve budget against each other — they are what
// waits for headroom, not what headroom is kept for.
var late = Carried("t-late");
var others = Enumerable.Range(0, 10).Select(i => Carried($"t-other-{i}")).ToList();
var room = await Headroom(used: 0.50m, Agent()).HasRoomAsync(late, [late, .. others], Now, default);
room.ShouldBeTrue();
}
[Fact]
public async Task The_instance_limit_counts_the_schedules_of_all_agents()
{
var agent = Agent(dailyCost: 0); // no limit of its own
var instance = new InstanceBudget { DailyCostUsd = 1.00m };
var hourly = Cron("t-hourly", "0 * * * *", lastOccurrence: "2027-03-10T13:00:00.0000000Z");
var late = Carried("t-late");
(await Headroom(used: 0.30m, agent, instance).HasRoomAsync(late, [hourly, late], Now, default)).ShouldBeTrue();
(await Headroom(used: 0.40m, agent, instance).HasRoomAsync(late, [hourly, late], Now, default)).ShouldBeFalse();
}
[Fact]
public async Task Without_any_limit_there_is_always_room()
{
var late = Carried("t-late");
(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
{
public override DateTimeOffset GetUtcNow() => new(now, TimeSpan.Zero);
public override TimeZoneInfo LocalTimeZone => TimeZoneInfo.Utc;
}
private sealed class FakeUsage(decimal usedCost, IReadOnlyList<RunUsage> recent) : IUsageRepository
{
public Task RecordAsync(RunUsage usage, CancellationToken ct) => Task.CompletedTask;
public Task<DailyUsage> GetDailyAsync(DateOnly date, string agentId, CancellationToken ct)
=> Task.FromResult(new DailyUsage(date, 0, 0, usedCost, true, 0));
public Task<IReadOnlyList<RunUsage>> GetRecentAsync(int limit, CancellationToken ct) => Task.FromResult(recent);
public Task<IReadOnlyList<RunUsage>> GetRecentForAgentAsync(string agentId, int limit, CancellationToken ct)
=> Task.FromResult(recent);
public Task<int> PurgeBeforeAsync(DateOnly date, CancellationToken ct) => Task.FromResult(0);
}
}
@@ -0,0 +1,173 @@
using ClawdDotNet.Core.Accounting;
using ClawdDotNet.Core.Budget;
using ClawdDotNet.Core.Config;
using Shouldly;
namespace ClawdDotNet.Core.Tests.Budget;
/// <summary>
/// #716 (the rest of #682): Runs in flight count against the budget with their estimate.
///
/// Before, every run was checked on its own against what was already recorded, so runs
/// starting together all saw the same figure and all passed. #682 capped how many could
/// start at once; the reservation closes the gap.
/// </summary>
public sealed class BudgetReservationTests
{
private static AgentConfig Agent(decimal dailyCost = 1.00m, long dailyTokens = 0)
=> new() { AgentId = "worker", Budget = new AgentBudget { DailyCostUsd = dailyCost, DailyTokens = dailyTokens } };
/// <summary>Recent runs averaging 0.30 USD and 3,000 tokens.</summary>
private static List<RunUsage> History(decimal cost = 0.30m, int tokens = 3_000)
=> [.. Enumerable.Range(0, 5).Select(_ => new RunUsage
{
AgentId = "worker", PromptTokens = tokens, CostUsd = cost, CostIsKnown = true, Status = "Completed"
})];
[Fact]
public async Task Runs_in_flight_count_against_the_budget()
{
var guard = new BudgetGuard(new FakeUsage(usedCost: 0.60m, History()));
var (first, a) = await guard.ReserveAsync(Agent(), InstanceBudget.Unlimited, default);
var (second, b) = await guard.ReserveAsync(Agent(), InstanceBudget.Unlimited, default);
var (third, c) = await guard.ReserveAsync(Agent(), InstanceBudget.Unlimited, default);
first.IsAllowed.ShouldBeTrue("0.60 used, nothing in flight");
second.IsAllowed.ShouldBeTrue("0.60 used + 0.30 in flight");
third.IsAllowed.ShouldBeFalse("0.60 used + 0.60 in flight — before #716 this one passed too");
third.Reason.ShouldNotBeNull().ShouldContain("reserviert");
c.ShouldBeNull();
a!.Dispose();
b!.Dispose();
}
[Fact]
public async Task A_finished_run_frees_its_reservation()
{
var guard = new BudgetGuard(new FakeUsage(usedCost: 0.60m, History()));
var (_, a) = await guard.ReserveAsync(Agent(), InstanceBudget.Unlimited, default);
var (_, b) = await guard.ReserveAsync(Agent(), InstanceBudget.Unlimited, default);
a!.Dispose();
a.Dispose(); // twice is harmless
var (again, c) = await guard.ReserveAsync(Agent(), InstanceBudget.Unlimited, default);
again.IsAllowed.ShouldBeTrue();
b!.Dispose();
c!.Dispose();
guard.Reserved("worker").ShouldBe(default(RunEstimate));
guard.Reserved("").ShouldBe(default(RunEstimate));
}
[Fact]
public async Task A_single_run_is_not_refused_because_of_its_own_estimate()
{
// 0.95 of 1.00 used, an average run costs 0.30: a small task may still fit, and
// refusing it on the average would be guessing.
var guard = new BudgetGuard(new FakeUsage(usedCost: 0.95m, History()));
var (status, reservation) = await guard.ReserveAsync(Agent(), InstanceBudget.Unlimited, default);
status.IsAllowed.ShouldBeTrue();
reservation!.Dispose();
}
[Fact]
public async Task Parallel_starts_cannot_all_pass_together()
{
var guard = new BudgetGuard(new FakeUsage(usedCost: 0m, History()));
var results = await Task.WhenAll(Enumerable.Range(0, 10)
.Select(_ => guard.ReserveAsync(Agent(), InstanceBudget.Unlimited, default)));
// 0.00, 0.30, 0.60, 0.90 in flight pass; from 1.20 on the budget is spoken for.
results.Count(r => r.Status.IsAllowed).ShouldBe(4);
foreach (var (_, reservation) in results)
reservation?.Dispose();
}
[Fact]
public async Task The_token_limit_counts_reservations_too()
{
var guard = new BudgetGuard(new FakeUsage(usedCost: 0m, History(cost: 0m, tokens: 4_000)));
var agent = Agent(dailyCost: 0, dailyTokens: 10_000);
var (a, ra) = await guard.ReserveAsync(agent, InstanceBudget.Unlimited, default);
var (b, rb) = await guard.ReserveAsync(agent, InstanceBudget.Unlimited, default);
var (c, _) = await guard.ReserveAsync(agent, InstanceBudget.Unlimited, default);
var (d, _) = await guard.ReserveAsync(agent, InstanceBudget.Unlimited, default);
a.IsAllowed.ShouldBeTrue();
b.IsAllowed.ShouldBeTrue();
c.IsAllowed.ShouldBeTrue("8,000 of 10,000 reserved");
d.IsAllowed.ShouldBeFalse("12,000 reserved");
d.Limit.ShouldBe(BudgetLimit.AgentTokens);
ra!.Dispose();
rb!.Dispose();
}
[Fact]
public async Task The_instance_limit_sees_the_runs_of_all_agents()
{
var guard = new BudgetGuard(new FakeUsage(usedCost: 0m, History()));
var instance = new InstanceBudget { DailyCostUsd = 0.50m };
var unlimitedAgent = Agent(dailyCost: 0);
var (a, _) = await guard.ReserveAsync(unlimitedAgent, instance, default);
var (b, _) = await guard.ReserveAsync(unlimitedAgent, instance, default);
var (c, _) = await guard.ReserveAsync(unlimitedAgent, instance, default);
a.IsAllowed.ShouldBeTrue();
b.IsAllowed.ShouldBeTrue("0.30 in flight");
c.IsAllowed.ShouldBeFalse("0.60 in flight over an instance limit of 0.50");
c.Limit.ShouldBe(BudgetLimit.InstanceCost);
}
[Fact]
public async Task Without_any_limit_nothing_is_reserved()
{
var guard = new BudgetGuard(new FakeUsage(usedCost: 0m, History()));
var (status, reservation) = await guard.ReserveAsync(Agent(dailyCost: 0), InstanceBudget.Unlimited, default);
status.IsAllowed.ShouldBeTrue();
reservation.ShouldBeNull();
}
[Fact]
public async Task Runs_without_usage_do_not_drag_the_estimate_down()
{
var history = History();
history.Add(new RunUsage { AgentId = "worker", Status = "BudgetExceeded" }); // 0 tokens
var guard = new BudgetGuard(new FakeUsage(usedCost: 0m, history));
var estimate = await guard.EstimateRunAsync("worker", default);
estimate.CostUsd.ShouldBe(0.30m);
estimate.Tokens.ShouldBe(3_000);
}
private sealed class FakeUsage(decimal usedCost, IReadOnlyList<RunUsage> recent) : IUsageRepository
{
public Task RecordAsync(RunUsage usage, CancellationToken ct) => Task.CompletedTask;
public async Task<DailyUsage> GetDailyAsync(DateOnly date, string agentId, CancellationToken ct)
{
await Task.Yield(); // like the database: the caller does not get the answer synchronously
return new DailyUsage(date, 0, 0, usedCost, true, 0);
}
public Task<IReadOnlyList<RunUsage>> GetRecentAsync(int limit, CancellationToken ct)
=> Task.FromResult(recent);
public Task<IReadOnlyList<RunUsage>> GetRecentForAgentAsync(string agentId, int limit, CancellationToken ct)
=> Task.FromResult(recent);
public Task<int> PurgeBeforeAsync(DateOnly date, CancellationToken ct) => Task.FromResult(0);
}
}
@@ -0,0 +1,293 @@
using ClawdDotNet.Core.Storage;
using ClawdDotNet.Core.Tasks;
using ClawdDotNet.Core.Tests.Infrastructure;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
namespace ClawdDotNet.Core.Tests.Tasks;
/// <summary>
/// #716: What the scanner does with a run the budget refused.
///
/// Before, a one-time task was retried every 30 minutes for the rest of the day — each
/// attempt another refusal and another receipt — and then ran after midnight at the
/// expense of the new day, without anyone having decided that. A recurring task's
/// occurrence lapsed silently.
/// </summary>
public sealed class BudgetMissTests : IDisposable
{
private readonly string _directory;
private readonly SqliteStorage _storage;
private readonly SqliteTaskRepository _repo;
// 2027: a cron schedule never looks back before the task's creation date.
private readonly FakeTimeProvider _clock = new(new DateTimeOffset(2027, 3, 10, 12, 0, 0, TimeSpan.Zero));
public BudgetMissTests()
{
_directory = Path.Combine(Path.GetTempPath(), "clawd-tests", Guid.NewGuid().ToString("N"));
_storage = new SqliteStorage(Path.Combine(_directory, "state.db"));
_repo = new SqliteTaskRepository(_storage);
}
public void Dispose()
{
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
try { Directory.Delete(_directory, recursive: true); }
catch { /* cleanup is secondary */ }
}
private TaskScanner Scanner(ITaskDispatcher dispatcher, BudgetMissPolicy policy = BudgetMissPolicy.Expire)
=> new(_repo, dispatcher, NullLoggerFactory.Instance, clock: _clock, budgetMissPolicy: _ => policy);
private Task<TaskItem> SeedOneTime(string id, int priority = 3)
=> _repo.UpsertAsync(new TaskItem { Id = id, Title = id, Assignee = "@worker", Priority = priority }, default);
private Task<TaskItem> SeedHourly(string id)
=> _repo.UpsertAsync(new TaskItem
{
Id = id, Title = id, Assignee = "@worker",
When = new TaskWhen { Kind = TaskWhenKind.Cron, Value = "0 * * * *", TimeZone = "UTC" }
}, default);
private async Task<TaskItem> Get(string id) => (await _repo.GetAsync(id, default))!;
/// <summary>The next local midnight, in UTC — the budget day follows the local date.</summary>
private DateTime NextLocalMidnightUtc()
{
var zone = _clock.LocalTimeZone;
var local = TimeZoneInfo.ConvertTimeFromUtc(_clock.GetUtcNow().UtcDateTime, zone);
return TimeZoneInfo.ConvertTimeToUtc(DateTime.SpecifyKind(local.Date.AddDays(1), DateTimeKind.Unspecified), zone);
}
private void MoveTo(DateTime utc) => _clock.SetUtcNow(new DateTimeOffset(utc, TimeSpan.Zero));
// ═══════════════════ one-time tasks: carried over ═══════════════════
[Fact]
public async Task A_refused_one_time_task_is_carried_over_to_the_next_day()
{
await SeedOneTime("t-report");
var dispatcher = new BudgetDispatcher { HasBudget = false };
var scanner = Scanner(dispatcher);
var midnight = NextLocalMidnightUtc();
await scanner.ScanOnceAsync(default);
var task = await Get("t-report");
task.Status.ShouldBe(TaskItemStatus.Todo, "it is still work to be done");
task.BudgetMissCount.ShouldBe(1);
task.LastBudgetMissAt.ShouldNotBeNull();
task.DeferredUntil.ShouldBe(midnight);
task.IsCarriedOver.ShouldBeTrue();
}
[Fact]
public async Task A_carried_over_task_is_not_retried_the_same_day()
{
await SeedOneTime("t-report");
var dispatcher = new BudgetDispatcher { HasBudget = false };
var scanner = Scanner(dispatcher);
await scanner.ScanOnceAsync(default);
// Before #716: again after 30 minutes, and again, until midnight.
MoveTo(NextLocalMidnightUtc().AddMinutes(-1));
(await scanner.ScanOnceAsync(default)).ShouldBe(0);
dispatcher.Calls.ShouldBe(1, "one refusal, not one every half hour");
}
[Fact]
public async Task On_the_next_day_it_runs_and_the_miss_is_settled()
{
await SeedOneTime("t-report");
var dispatcher = new BudgetDispatcher { HasBudget = false };
var scanner = Scanner(dispatcher);
await scanner.ScanOnceAsync(default);
MoveTo(NextLocalMidnightUtc().AddMinutes(1));
dispatcher.HasBudget = true;
(await scanner.ScanOnceAsync(default)).ShouldBe(1);
var task = await Get("t-report");
task.Status.ShouldBe(TaskItemStatus.Done);
task.BudgetMissCount.ShouldBe(0);
task.DeferredUntil.ShouldBeNull();
task.LastBudgetMissAt.ShouldNotBeNull("kept as history");
}
[Fact]
public async Task The_new_day_has_right_of_way_over_carried_over_work()
{
await SeedOneTime("t-yesterday", priority: 5);
var dispatcher = new BudgetDispatcher { HasBudget = false };
var scanner = Scanner(dispatcher);
await scanner.ScanOnceAsync(default);
MoveTo(NextLocalMidnightUtc().AddMinutes(1));
await SeedOneTime("t-today", priority: 1);
dispatcher.HasBudget = true;
dispatcher.Started.Clear();
await scanner.ScanOnceAsync(default);
dispatcher.Started.ShouldBe(["t-today", "t-yesterday"],
"carried-over work comes after the day's own, even with a higher priority");
}
// ═══════════════════ recurring tasks: the agent's policy ═══════════════════
[Fact]
public async Task Under_expire_a_refused_occurrence_lapses_and_is_recorded()
{
await SeedHourly("t-hourly");
var dispatcher = new BudgetDispatcher { HasBudget = false };
var scanner = Scanner(dispatcher, BudgetMissPolicy.Expire);
await scanner.ScanOnceAsync(default);
var task = await Get("t-hourly");
task.Status.ShouldBe(TaskItemStatus.Todo);
task.BudgetMissCount.ShouldBe(1);
task.CatchUpPending.ShouldBeFalse();
// Nothing is caught up the next day beyond the regular occurrences.
MoveTo(NextLocalMidnightUtc().AddMinutes(30));
dispatcher.HasBudget = true;
dispatcher.Started.Clear();
(await scanner.ScanOnceAsync(default)).ShouldBe(1, "only the regular occurrence of this hour");
}
[Fact]
public async Task Under_catch_up_once_several_misses_leave_one_pending_catch_up()
{
await SeedHourly("t-hourly");
var dispatcher = new BudgetDispatcher { HasBudget = false };
var scanner = Scanner(dispatcher, BudgetMissPolicy.CatchUpOnce);
var midnight = NextLocalMidnightUtc();
// Three hours in a row refused.
for (var hour = 0; hour < 3; hour++)
{
await scanner.ScanOnceAsync(default);
_clock.Advance(TimeSpan.FromHours(1));
}
var task = await Get("t-hourly");
task.BudgetMissCount.ShouldBe(3);
task.CatchUpPending.ShouldBeTrue();
task.DeferredUntil.ShouldBe(midnight, "set by the first miss, not pushed on by the later ones");
}
[Fact]
public async Task The_catch_up_runs_exactly_once_on_the_next_day()
{
// A daily report at 22:30: its next regular occurrence is 24 hours away, so here
// the catch-up is the only run — the case catchUpOnce exists for.
MoveTo(new DateTime(2027, 3, 10, 22, 30, 0, DateTimeKind.Utc));
await _repo.UpsertAsync(new TaskItem
{
Id = "t-daily", Title = "t-daily", Assignee = "@worker",
When = new TaskWhen { Kind = TaskWhenKind.Cron, Value = "30 22 * * *", TimeZone = "UTC" }
}, default);
var dispatcher = new BudgetDispatcher { HasBudget = false };
var scanner = Scanner(dispatcher, BudgetMissPolicy.CatchUpOnce);
(await scanner.ScanOnceAsync(default)).ShouldBe(1, "the regular 22:30 run, refused");
MoveTo(NextLocalMidnightUtc().AddMinutes(1));
dispatcher.HasBudget = true;
dispatcher.Started.Clear();
(await scanner.ScanOnceAsync(default)).ShouldBe(1, "the catch-up");
_clock.Advance(TimeSpan.FromMinutes(1));
(await scanner.ScanOnceAsync(default)).ShouldBe(0, "once, not on every tick");
dispatcher.Started.ShouldBe(["t-daily"]);
var task = await Get("t-daily");
task.CatchUpPending.ShouldBeFalse();
task.BudgetMissCount.ShouldBe(0);
task.Status.ShouldBe(TaskItemStatus.Todo, "still recurring");
}
[Fact]
public async Task A_pending_catch_up_does_not_run_on_the_same_day()
{
await SeedHourly("t-hourly");
var dispatcher = new BudgetDispatcher { HasBudget = false };
var scanner = Scanner(dispatcher, BudgetMissPolicy.CatchUpOnce);
await scanner.ScanOnceAsync(default);
// Budget raised by the human in the evening: the regular occurrence runs, and it
// covers the pending catch-up — one run, not two.
dispatcher.HasBudget = true;
dispatcher.Started.Clear();
_clock.Advance(TimeSpan.FromHours(1));
(await scanner.ScanOnceAsync(default)).ShouldBe(1);
(await Get("t-hourly")).CatchUpPending.ShouldBeFalse();
}
[Fact]
public async Task A_shutdown_is_not_a_budget_miss()
{
await SeedOneTime("t-report");
var dispatcher = new BudgetDispatcher { Result = TaskLauf.NichtGelaufen };
await Scanner(dispatcher).ScanOnceAsync(default);
var task = await Get("t-report");
task.BudgetMissCount.ShouldBe(0);
task.DeferredUntil.ShouldBeNull("a shutdown keeps the old 30-minute retry");
}
[Fact]
public async Task Late_work_waits_while_the_headroom_gate_says_no()
{
await SeedOneTime("t-yesterday");
var dispatcher = new BudgetDispatcher { HasBudget = false };
var gate = new FixedGate { Room = false };
var scanner = new TaskScanner(_repo, dispatcher, NullLoggerFactory.Instance,
clock: _clock, lateRunGate: gate);
await scanner.ScanOnceAsync(default);
MoveTo(NextLocalMidnightUtc().AddMinutes(1));
await SeedOneTime("t-today");
dispatcher.HasBudget = true;
dispatcher.Started.Clear();
await scanner.ScanOnceAsync(default);
dispatcher.Started.ShouldBe(["t-today"], "regular work never asks the gate");
gate.Asked.ShouldBe(["t-yesterday"]);
gate.Room = true;
await scanner.ScanOnceAsync(default);
dispatcher.Started.ShouldBe(["t-today", "t-yesterday"]);
}
private sealed class FixedGate : ILateRunGate
{
public bool Room { get; set; }
public List<string> Asked { get; } = [];
public Task<bool> HasRoomAsync(TaskItem lateTask, IReadOnlyList<TaskItem> runnable, DateTime nowUtc, CancellationToken ct)
{
Asked.Add(lateTask.Id);
return Task.FromResult(Room);
}
}
// ═══════════════════ test double ═══════════════════
private sealed class BudgetDispatcher : ITaskDispatcher
{
public bool HasBudget { get; set; } = true;
public TaskLauf? Result { get; set; }
public int Calls { get; private set; }
public List<string> Started { get; } = [];
public Task<TaskLauf> DispatchAsync(TaskItem task, CancellationToken ct)
{
Calls++;
Started.Add(task.Id);
return Task.FromResult(Result ?? (HasBudget ? TaskLauf.Erledigt : TaskLauf.KeinBudget));
}
}
}
@@ -143,6 +143,24 @@ public sealed class DublettenTests : IDisposable
(await _board.ListAsync(new TaskQuery { Limit = 50 }, default)).Count.ShouldBe(2);
}
[Fact]
public async Task Die_Dublette_wird_auch_hinter_500_anderen_Aufgaben_gefunden()
{
// #716: Die Suche las ListAsync mit Limit 500, sortiert nach Prioritaet. Eine alte
// Aufgabe mit niedriger Prioritaet fiel damit aus dem Blick, sobald das Brett wuchs.
var alt = await _board.CreateAsync(new TaskItem
{
Title = "Betriebsluecke: Eur Kurse nicht aktuell", Assignee = "@human", Priority = 1
}, default);
for (var i = 0; i < 510; i++)
await _repo.UpsertAsync(new TaskItem { Id = $"t-fill-{i}", Title = $"Anderes Thema {i}", Priority = 5 }, default);
var result = await Run("""{"action":"task_create","title":"Betriebsluecke: Kurse EUR nicht aktuell","assignee":"@human"}""");
result.Success.ShouldBeFalse("die alte Aufgabe meint dasselbe");
result.ErrorMessage.ShouldNotBeNull().ShouldContain(alt.Id);
}
[Fact]
public async Task Eine_erledigte_Aufgabe_blockiert_den_naechsten_Befund_nicht()
{