Budgetende, Stufe 3: Nachholen nur im Spielraum des Tages, keine Kappung bei 500 (#716)
- TaskSchedule.OccurrencesBetween zaehlt, wie oft eine Aufgabe bis zu einem Zeitpunkt noch laeuft, einschliesslich eines jetzt faelligen Laufs. - BudgetHeadroom (ILateRunGate): Ein mitgenommener oder nachzuholender Lauf startet nur, wenn er in den Spielraum passt: Grenze minus Verbrauch minus Reservierung minus Bedarf der restlichen geplanten Laeufe bis Mitternacht minus 10 % Reserve. Das gilt je Agent und je Instanz, fuer Kosten und Tokens. Anderes Spaetwerk zaehlt nicht als Bedarf, sonst blockierten sich mitgenommene Aufgaben gegenseitig. - Der Scanner fragt den Spielraum nur fuer Spaetwerk und protokolliert, wenn es warten muss. Die Engine gibt ihren BudgetGuard heraus, damit Scanner und Engine dieselben Reservierungen sehen. - Ersetzt den pauschalen 25-%-Anteil aus dem Konzept (dort nachgezogen): Ein fester Anteil ist an vollen Tagen zu viel, an ruhigen zu wenig. Drei weitere stille Kappungen bei 500 beseitigt (ListAllAsync, ungekappt): Kostenprognose (Aufgaben ueber 500 fehlten), Waisenabgleich beim Import und task_find aus #674, wo Dubletten unter aelteren Aufgaben mit niedriger Prioritaet unsichtbar wurden. Tests: BudgetHeadroomTests (9), Scanner-Pforte (1), Dublette hinter 500 Aufgaben (1). 1.257 gruen. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
8a83a67fb4
commit
9cf9a2cfc2
@@ -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"/>)
|
||||
|
||||
@@ -270,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);
|
||||
|
||||
@@ -44,6 +44,15 @@ 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>Bis C1 kommt: der Markt gilt als immer offen.</summary>
|
||||
public sealed class AlwaysOpenMarketCalendar : IMarketCalendar
|
||||
{
|
||||
@@ -110,6 +119,11 @@ public sealed class TaskScanner : IAsyncDisposable
|
||||
private readonly Dictionary<string, (int Misses, int CarriedOver, int CatchUps)> _missesToday = new(StringComparer.OrdinalIgnoreCase);
|
||||
private DateOnly? _missDay;
|
||||
|
||||
private readonly ILateRunGate? _lateRunGate;
|
||||
|
||||
/// <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;
|
||||
|
||||
@@ -122,7 +136,8 @@ public sealed class TaskScanner : IAsyncDisposable
|
||||
TimeSpan? tick = null,
|
||||
TimeSpan? lease = null,
|
||||
int maxParallelRuns = DefaultMaxParallelRuns,
|
||||
Func<TaskItem, BudgetMissPolicy>? budgetMissPolicy = null)
|
||||
Func<TaskItem, BudgetMissPolicy>? budgetMissPolicy = null,
|
||||
ILateRunGate? lateRunGate = null)
|
||||
{
|
||||
_repo = repo;
|
||||
_dispatcher = dispatcher;
|
||||
@@ -133,6 +148,7 @@ public sealed class TaskScanner : IAsyncDisposable
|
||||
_lease = lease ?? TimeSpan.FromMinutes(15);
|
||||
_maxParallelRuns = Math.Max(1, maxParallelRuns);
|
||||
_missPolicy = budgetMissPolicy ?? (_ => BudgetMissPolicy.Expire);
|
||||
_lateRunGate = lateRunGate;
|
||||
}
|
||||
|
||||
/// <summary>How many scheduled runs may be in flight at once (#682).</summary>
|
||||
@@ -319,7 +335,8 @@ public sealed class TaskScanner : IAsyncDisposable
|
||||
|
||||
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.
|
||||
@@ -329,6 +346,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)));
|
||||
@@ -347,6 +372,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;
|
||||
}
|
||||
|
||||
|
||||
@@ -201,6 +201,70 @@ 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)
|
||||
{
|
||||
fromUtc = DateTime.SpecifyKind(fromUtc, DateTimeKind.Utc);
|
||||
toUtc = DateTime.SpecifyKind(toUtc, DateTimeKind.Utc);
|
||||
if (toUtc <= fromUtc)
|
||||
return 0;
|
||||
|
||||
var dueNow = DueOccurrence(task, fromUtc) is not null ? 1 : 0;
|
||||
|
||||
if (task.When is not { } when)
|
||||
return dueNow; // 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;
|
||||
}
|
||||
|
||||
case TaskWhenKind.Every:
|
||||
{
|
||||
if (!TryParseInterval(when.Value, out var interval) || interval <= TimeSpan.Zero)
|
||||
return dueNow;
|
||||
|
||||
var next = dueNow == 1
|
||||
? fromUtc + interval
|
||||
: NextOccurrenceUtc(task, fromUtc) ?? toUtc;
|
||||
var count = dueNow;
|
||||
while (count < cap && next < toUtc)
|
||||
{
|
||||
count++;
|
||||
next += interval;
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
case TaskWhenKind.Cron:
|
||||
{
|
||||
if (!TryPrepareCron(when, out var cron, out var tz))
|
||||
return 0;
|
||||
|
||||
var count = dueNow;
|
||||
var cursor = fromUtc;
|
||||
while (count < cap && NextCronUtc(cron, tz, cursor) is { } next && next < toUtc)
|
||||
{
|
||||
if (next > fromUtc)
|
||||
count++;
|
||||
cursor = next;
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
default:
|
||||
return dueNow;
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
Reference in New Issue
Block a user