From 8f8c2a65b23f65b1f6d8af4e68f22b2a678876eb Mon Sep 17 00:00:00 2001 From: mika kuns Date: Wed, 5 Aug 2026 11:10:30 +0200 Subject: [PATCH] =?UTF-8?q?=EF=BB=BFfeat(claude-do):=20Worker:=20UsageGate?= =?UTF-8?q?=20=E2=80=94=20Queue=20ab=20Schwelle=20pausieren?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit > **Stand 2026-08-05 (List-Handler):** Der Roadblock aus dem letzten Lauf ist erledigt. Beide Voraussetzungen sind jetzt auf `main` gemerged: die `app_settings`-Schwellen `UsageGateFiveHourPct`/`UsageGateSevenDayPct` (Merge-Commit `b1efcdc`) und `UsageState`/`IUsageClient`/`UsageMonitorService` unter `src/ClaudeDo.Worker/Usage/` (Merge-Commit `b126a21`). Dein Worktree ist frisch von diesem `main` ClaudeDo-Task: 06a7cc32-6ab7-4758-98f4-bee77149b2bf --- src/ClaudeDo.Worker/CLAUDE.md | 4 +- src/ClaudeDo.Worker/Program.cs | 1 + src/ClaudeDo.Worker/Queue/QueueService.cs | 91 ++++++++++---- .../Usage/Interfaces/IUsageGate.cs | 6 + src/ClaudeDo.Worker/Usage/UsageGate.cs | 58 +++++++++ .../External/AddSubtaskToolTests.cs | 4 +- .../External/BatchMcpToolsTests.cs | 4 +- .../External/ExternalMcpServiceTests.cs | 4 +- .../Infrastructure/FakeUsageGate.cs | 11 ++ .../Services/QueueServiceSlotGuardTests.cs | 4 +- .../Services/QueueServiceTests.cs | 116 +++++++++++++++++- .../Usage/UsageGateTests.cs | 105 ++++++++++++++++ 12 files changed, 375 insertions(+), 33 deletions(-) create mode 100644 src/ClaudeDo.Worker/Usage/Interfaces/IUsageGate.cs create mode 100644 src/ClaudeDo.Worker/Usage/UsageGate.cs create mode 100644 tests/ClaudeDo.Worker.Tests/Infrastructure/FakeUsageGate.cs create mode 100644 tests/ClaudeDo.Worker.Tests/Usage/UsageGateTests.cs diff --git a/src/ClaudeDo.Worker/CLAUDE.md b/src/ClaudeDo.Worker/CLAUDE.md index 480252d8..746d450a 100644 --- a/src/ClaudeDo.Worker/CLAUDE.md +++ b/src/ClaudeDo.Worker/CLAUDE.md @@ -21,7 +21,7 @@ Worker/ Report/ — ClaudeHistoryReader, WeekReportPromptBuilder, WeekReportService; interfaces in Report/Interfaces/ Prime/ — daily-prep ("Prime Claude"): PrimeScheduler (BackgroundService), PrimeRunner (runs the daily prep), DailyPrepPrompt (fixed prompt + CLI args + LogPath() helper), NextDueCalculator, PrimeScheduleSignal; interfaces in Prime/Interfaces/ (IPrimeRunner, IPrimeClock, IPrimeScheduleSignal, IPrimeBroadcaster) Online/ — optional Online Inbox sync: OnlineInboxConfig (config record), Dtos (RemoteList/RemoteTask/MirrorTask), IOnlineInboxApi, OnlineInboxApiClient (typed HttpClient, bearer auth, HTTPS guard), OnlineTokenStore (DPAPI refresh-token store, Windows-only), StaticTokenAuthProvider (default/test IOnlineAuthProvider), ZitadelAuthProvider (OIDC discovery + refresh-token flow), OnlineSyncService (BackgroundService: reconcile loop), OnlineBacklog (Idle-backlog filter/query); interface in Online/Interfaces/ (IOnlineAuthProvider) - Usage/ — OAuth usage monitor: UsageModels (UsageBucket/UsageLimitRow/UsageSnapshot), ClaudeOAuthUsageClient (reads the access token Claude Code keeps fresh at `~/.claude/.credentials.json`, calls `GET https://api.anthropic.com/api/oauth/usage`; defensive parsing — missing/null buckets → null, missing `limits` → empty list; never logs the token), UsageState (threadsafe singleton; a failed poll never overwrites the last good snapshot, only sets `LastError`), UsageMonitorService (BackgroundService, polls on `usage_poll_interval_seconds`, one poll at startup, logs a failure at most once per distinct error message), TranscriptUsageReader (aggregates Claude Code transcript token usage from `~/.claude/projects/**/*.jsonl` by date/model/scope (ClaudeDo vs Other), deduped by requestId, with a per-file length+mtime cache); interfaces in Usage/Interfaces/ (IUsageClient, ITranscriptUsageReader) + Usage/ — OAuth usage monitor: UsageModels (UsageBucket/UsageLimitRow/UsageSnapshot), ClaudeOAuthUsageClient (reads the access token Claude Code keeps fresh at `~/.claude/.credentials.json`, calls `GET https://api.anthropic.com/api/oauth/usage`; defensive parsing — missing/null buckets → null, missing `limits` → empty list; never logs the token), UsageState (threadsafe singleton; a failed poll never overwrites the last good snapshot, only sets `LastError`), UsageMonitorService (BackgroundService, polls on `usage_poll_interval_seconds`, one poll at startup, logs a failure at most once per distinct error message), TranscriptUsageReader (aggregates Claude Code transcript token usage from `~/.claude/projects/**/*.jsonl` by date/model/scope (ClaudeDo vs Other), deduped by requestId, with a per-file length+mtime cache), UsageGate (reads `UsageState` + `AppSettings.UsageGateFiveHourPct`/`UsageGateSevenDayPct`, returns a `UsageGateDecision(IsBlocked, Reason)`; `Utilization` from `UsageBucket` is already a 0–100 percent, compared directly against the threshold with `>=`; threshold `0` = that bucket never gates; fail-open — no snapshot yet, a failed last poll, or a settings-read error all resolve to not-blocked); interfaces in Usage/Interfaces/ (IUsageClient, ITranscriptUsageReader, IUsageGate) ``` Interfaces (e.g. `IQueueWaker`, `IPrimeClock`, `ITaskStateService`) live in an `Interfaces/` subfolder within their area; the namespace stays the area namespace. @@ -30,7 +30,7 @@ Interfaces (e.g. `IQueueWaker`, `IPrimeClock`, `ITaskStateService`) live in an ` - **Program.cs** — loads config, inits schema, registers DI, configures SignalR on `/hub`, binds to `127.0.0.1:47821` - **TaskStateService** — only component that writes `Status`, `PlanningPhase`, `BlockedByTaskId`. All transitions return a `TransitionResult` (no exceptions on invalid moves). Wakes the queue and broadcasts `TaskUpdated` automatically; advances the planning chain on child terminal transitions. -- **IQueueWaker / IQueuePicker / QueueService** — waker is a singleton `SemaphoreSlim`; picker performs the atomic `Queued → Running` claim filtered by `BlockedByTaskId IS NULL`, `is_manual = 0` and schedule; QueueService is a thin `BackgroundService` that loops on the waker and dispatches via `TaskRunner`. +- **IQueueWaker / IQueuePicker / QueueService** — waker is a singleton `SemaphoreSlim`; picker performs the atomic `Queued → Running` claim filtered by `BlockedByTaskId IS NULL`, `is_manual = 0` and schedule; QueueService is a thin `BackgroundService` that loops on the waker and dispatches via `TaskRunner`. On each loop tick, after computing `maxParallel`, it also asks `IUsageGate.EvaluateAsync` — if blocked, the slot-fill loop is skipped entirely for that tick (already-running slots are untouched; `RunNow`/`ContinueTask`/interactive/planning/daily-prep all bypass the queue and are unaffected). A blocked↔free transition is logged/broadcast (`WorkerLog`, Warn on block / Info on resume) exactly once per change, not on every tick; the 30 s backstop timer re-evaluates the gate on its own even with no wake signal, so the queue self-recovers once usage drops back under the threshold. - **OverrideSlotService** — owns `RunNow` / `ContinueTask`; goes through `TaskStateService.StartRunningAsync` (caller-driven, serialized by slot lock). - **StaleTaskRecovery** — startup-only service; calls `TaskStateService.RecoverStaleRunningAsync` to flip orphaned `Running` rows to `Failed`. - **External/*** — always-on MCP tools for general Claude sessions, scoped to *starting* and *observing* sessions (no worktree/merge, multi-turn, planning, or app-settings writes). Auth via optional `X-ClaudeDo-Key` header. Registered explicitly in `Program.cs`'s external app via `.WithTools()`. Every optional/filter parameter across these tools must carry a C# default value (e.g. `string? status = null`) — the MCP schema only marks a parameter optional when it has one; nullability alone doesn't do it (`ExternalMcpToolSchemaTests` guards this by reflection). `ExternalMcpExceptionFilter.Wrap` is registered as a call-tool filter so `InvalidOperationException`/`ArgumentException` messages survive as `McpException` — otherwise the SDK's own catch-all replaces any non-`McpException` with a generic "An error occurred invoking 'X'." Organized by concern: diff --git a/src/ClaudeDo.Worker/Program.cs b/src/ClaudeDo.Worker/Program.cs index 5acfe65e..a88149e2 100644 --- a/src/ClaudeDo.Worker/Program.cs +++ b/src/ClaudeDo.Worker/Program.cs @@ -206,6 +206,7 @@ builder.Services.AddHttpClient(client => client.Timeout = TimeSpan.FromSeconds(5); }); builder.Services.AddHostedService(); +builder.Services.AddSingleton(); // Loopback-only bind. Firewall is irrelevant for 127.0.0.1. builder.WebHost.UseUrls($"http://127.0.0.1:{cfg.SignalRPort}"); diff --git a/src/ClaudeDo.Worker/Queue/QueueService.cs b/src/ClaudeDo.Worker/Queue/QueueService.cs index d3161c2f..ed63bc8e 100644 --- a/src/ClaudeDo.Worker/Queue/QueueService.cs +++ b/src/ClaudeDo.Worker/Queue/QueueService.cs @@ -2,8 +2,11 @@ using ClaudeDo.Data; using ClaudeDo.Data.Models; using ClaudeDo.Data.Repositories; using ClaudeDo.Worker.Config; +using ClaudeDo.Worker.Hub; using ClaudeDo.Worker.Runner; using ClaudeDo.Worker.State; +using ClaudeDo.Worker.Usage; +using ClaudeDo.Worker.Usage.Interfaces; using Microsoft.EntityFrameworkCore; using TaskStatus = ClaudeDo.Data.Models.TaskStatus; @@ -20,9 +23,13 @@ public sealed class QueueService : BackgroundService private readonly OverrideSlotService _override; private readonly ITaskStateService _state; private readonly RunCancellationRegistry _runCancels; + private readonly IUsageGate _usageGate; + private readonly UsageState _usageState; + private readonly HubBroadcaster _broadcaster; private readonly object _lock = new(); private readonly Dictionary _queueSlots = new(); + private bool _usageGateBlocked; public QueueService( IDbContextFactory dbFactory, @@ -33,7 +40,10 @@ public sealed class QueueService : BackgroundService IQueuePicker picker, OverrideSlotService overrideSlot, ITaskStateService state, - RunCancellationRegistry runCancels) + RunCancellationRegistry runCancels, + IUsageGate usageGate, + UsageState usageState, + HubBroadcaster broadcaster) { _dbFactory = dbFactory; _runner = runner; @@ -44,6 +54,9 @@ public sealed class QueueService : BackgroundService _override = overrideSlot; _state = state; _runCancels = runCancels; + _usageGate = usageGate; + _usageState = usageState; + _broadcaster = broadcaster; } public IReadOnlyList<(string slot, string taskId, DateTime startedAt)> GetActive() @@ -113,32 +126,40 @@ public sealed class QueueService : BackgroundService var maxParallel = await GetMaxParallelAsync(stoppingToken); - // Fill as many free slots as the limit allows. - while (!stoppingToken.IsCancellationRequested) + var gateDecision = await _usageGate.EvaluateAsync(stoppingToken); + await ReportUsageGateTransitionAsync(gateDecision); + + // Only queue refill is gated. Runs already in flight (RunNow, ContinueTask, + // interactive sessions, planning, daily prep) keep going regardless. + if (!gateDecision.IsBlocked) { - lock (_lock) + // Fill as many free slots as the limit allows. + while (!stoppingToken.IsCancellationRequested) { - if (_queueSlots.Count >= maxParallel) break; - } - - var task = await _picker.ClaimNextAsync(DateTime.UtcNow, stoppingToken); - if (task is null) break; - - lock (_lock) - { - var cts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken); - _queueSlots[task.Id] = new QueueSlotState { TaskId = task.Id, StartedAt = DateTime.UtcNow, Cts = cts }; - _runCancels.Register(task.Id, cts); - - _ = RunInSlotAsync(task.Id, cts.Token).ContinueWith(t => + lock (_lock) { - if (t.IsFaulted) - _logger.LogError(t.Exception, "RunInSlotAsync failed for task {TaskId} in queue slot", task.Id); - lock (_lock) { _queueSlots.Remove(task.Id); } - _runCancels.Unregister(task.Id, cts); - cts.Dispose(); - _waker.Wake(); // Check for next task immediately. - }, TaskScheduler.Default); + if (_queueSlots.Count >= maxParallel) break; + } + + var task = await _picker.ClaimNextAsync(DateTime.UtcNow, stoppingToken); + if (task is null) break; + + lock (_lock) + { + var cts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken); + _queueSlots[task.Id] = new QueueSlotState { TaskId = task.Id, StartedAt = DateTime.UtcNow, Cts = cts }; + _runCancels.Register(task.Id, cts); + + _ = RunInSlotAsync(task.Id, cts.Token).ContinueWith(t => + { + if (t.IsFaulted) + _logger.LogError(t.Exception, "RunInSlotAsync failed for task {TaskId} in queue slot", task.Id); + lock (_lock) { _queueSlots.Remove(task.Id); } + _runCancels.Unregister(task.Id, cts); + cts.Dispose(); + _waker.Wake(); // Check for next task immediately. + }, TaskScheduler.Default); + } } } } @@ -155,6 +176,28 @@ public sealed class QueueService : BackgroundService _logger.LogInformation("QueueService stopping"); } + private async Task ReportUsageGateTransitionAsync(UsageGateDecision decision) + { + if (decision.IsBlocked == _usageGateBlocked) return; + _usageGateBlocked = decision.IsBlocked; + + if (decision.IsBlocked) + { + _logger.LogInformation("QueueService: usage gate blocking queue refill ({Reason})", decision.Reason); + await _broadcaster.WorkerLog($"Queue pausiert: {decision.Reason}", WorkerLogLevel.Warn, DateTime.UtcNow); + } + else + { + var snapshot = _usageState.Snapshot; + var message = snapshot?.FiveHour is not null && snapshot.SevenDay is not null + ? $"Queue fortgesetzt: 5h {snapshot.FiveHour.Utilization:0}%, 7d {snapshot.SevenDay.Utilization:0}%" + : "Queue fortgesetzt"; + + _logger.LogInformation("QueueService: usage gate cleared, queue refill resumed"); + await _broadcaster.WorkerLog(message, WorkerLogLevel.Info, DateTime.UtcNow); + } + } + private async Task GetMaxParallelAsync(CancellationToken ct) { try diff --git a/src/ClaudeDo.Worker/Usage/Interfaces/IUsageGate.cs b/src/ClaudeDo.Worker/Usage/Interfaces/IUsageGate.cs new file mode 100644 index 00000000..e85d00de --- /dev/null +++ b/src/ClaudeDo.Worker/Usage/Interfaces/IUsageGate.cs @@ -0,0 +1,6 @@ +namespace ClaudeDo.Worker.Usage.Interfaces; + +public interface IUsageGate +{ + Task EvaluateAsync(CancellationToken ct = default); +} diff --git a/src/ClaudeDo.Worker/Usage/UsageGate.cs b/src/ClaudeDo.Worker/Usage/UsageGate.cs new file mode 100644 index 00000000..55923b6f --- /dev/null +++ b/src/ClaudeDo.Worker/Usage/UsageGate.cs @@ -0,0 +1,58 @@ +using ClaudeDo.Data; +using ClaudeDo.Data.Repositories; +using ClaudeDo.Worker.Usage.Interfaces; +using Microsoft.EntityFrameworkCore; + +namespace ClaudeDo.Worker.Usage; + +public sealed record UsageGateDecision(bool IsBlocked, string? Reason); + +/// +/// Gates autonomous queue refill on the 5h/7d Claude usage window. Fail-open: no snapshot yet, +/// a failed last poll, or a settings read error all resolve to "not blocked" — an undocumented +/// or unreachable usage API must never stall automation. Only queue refill is gated; runs already +/// in flight (RunNow, ContinueTask, interactive sessions, planning, daily prep) are untouched. +/// +public sealed class UsageGate : IUsageGate +{ + private readonly IDbContextFactory _dbFactory; + private readonly UsageState _state; + private readonly ILogger _logger; + + public UsageGate(IDbContextFactory dbFactory, UsageState state, ILogger logger) + { + _dbFactory = dbFactory; + _state = state; + _logger = logger; + } + + public async Task EvaluateAsync(CancellationToken ct = default) + { + var snapshot = _state.Snapshot; + if (snapshot is null || _state.LastError is not null) + return new UsageGateDecision(false, null); + + int fiveHourPct; + int sevenDayPct; + try + { + using var context = _dbFactory.CreateDbContext(); + var settings = await new AppSettingsRepository(context).GetAsync(ct); + fiveHourPct = settings.UsageGateFiveHourPct; + sevenDayPct = settings.UsageGateSevenDayPct; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "UsageGate: failed to read gate thresholds; not blocking"); + return new UsageGateDecision(false, null); + } + + if (fiveHourPct > 0 && snapshot.FiveHour is not null && snapshot.FiveHour.Utilization >= fiveHourPct) + return new UsageGateDecision(true, $"5h-Limit {snapshot.FiveHour.Utilization:0}% >= {fiveHourPct}%"); + + if (sevenDayPct > 0 && snapshot.SevenDay is not null && snapshot.SevenDay.Utilization >= sevenDayPct) + return new UsageGateDecision(true, $"7d-Limit {snapshot.SevenDay.Utilization:0}% >= {sevenDayPct}%"); + + return new UsageGateDecision(false, null); + } +} diff --git a/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs b/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs index d9eaf7ef..5de2bfa0 100644 --- a/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs +++ b/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs @@ -8,6 +8,7 @@ using ClaudeDo.Worker.Planning; using ClaudeDo.Worker.Queue; using ClaudeDo.Worker.Runner; using ClaudeDo.Worker.Tests.Infrastructure; +using ClaudeDo.Worker.Usage; using ClaudeDo.Worker.Worktrees; using ClaudeDo.Worker.Config; using Microsoft.Extensions.Logging.Abstractions; @@ -75,7 +76,8 @@ public sealed class AddSubtaskToolTests : IDisposable var picker = new ClaudeDo.Worker.Queue.QueuePicker(dbFactory); var runCancels = new RunCancellationRegistry(); var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, runCancels); - var queue = new QueueService(dbFactory, runner, cfg, NullLogger.Instance, waker, picker, overrideSlot, state, runCancels); + var queue = new QueueService(dbFactory, runner, cfg, NullLogger.Instance, waker, picker, overrideSlot, state, runCancels, + new FakeUsageGate(), new UsageState(), broadcaster); var maintenance = new WorktreeMaintenanceService(dbFactory, git, NullLogger.Instance); var merge = new TaskMergeService(dbFactory, git, broadcaster, state, NullLogger.Instance); var aggregator = new PlanningAggregator(dbFactory, git, NullLogger.Instance); diff --git a/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs b/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs index 66f17b16..2d044666 100644 --- a/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs +++ b/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs @@ -10,6 +10,7 @@ using ClaudeDo.Worker.Planning; using ClaudeDo.Worker.Queue; using ClaudeDo.Worker.Runner; using ClaudeDo.Worker.Tests.Infrastructure; +using ClaudeDo.Worker.Usage; using ClaudeDo.Worker.Worktrees; using Microsoft.Extensions.Logging.Abstractions; using TaskStatus = ClaudeDo.Data.Models.TaskStatus; @@ -96,7 +97,8 @@ public sealed class BatchMcpToolsTests : IDisposable var runCancels = new RunCancellationRegistry(); var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, runCancels); return new QueueService(dbFactory, runner, cfg, NullLogger.Instance, - new QueueWaker(), new QueuePicker(dbFactory), overrideSlot, state, runCancels); + new QueueWaker(), new QueuePicker(dbFactory), overrideSlot, state, runCancels, + new FakeUsageGate(), new UsageState(), broadcaster); } [Fact] diff --git a/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs b/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs index 38b490ec..7200b2c8 100644 --- a/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs +++ b/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs @@ -10,6 +10,7 @@ using ClaudeDo.Worker.Planning; using ClaudeDo.Worker.Queue; using ClaudeDo.Worker.Runner; using ClaudeDo.Worker.Tests.Infrastructure; +using ClaudeDo.Worker.Usage; using ClaudeDo.Worker.Worktrees; using Microsoft.AspNetCore.SignalR; using Microsoft.Extensions.Logging.Abstractions; @@ -161,7 +162,8 @@ public sealed class ExternalMcpServiceTests : IDisposable var picker = new ClaudeDo.Worker.Queue.QueuePicker(dbFactory); var runCancels = new RunCancellationRegistry(); var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, runCancels); - return new QueueService(dbFactory, runner, cfg, NullLogger.Instance, waker, picker, overrideSlot, state, runCancels); + return new QueueService(dbFactory, runner, cfg, NullLogger.Instance, waker, picker, overrideSlot, state, runCancels, + new FakeUsageGate(), new UsageState(), broadcaster); } [Fact] diff --git a/tests/ClaudeDo.Worker.Tests/Infrastructure/FakeUsageGate.cs b/tests/ClaudeDo.Worker.Tests/Infrastructure/FakeUsageGate.cs new file mode 100644 index 00000000..e06fdfbd --- /dev/null +++ b/tests/ClaudeDo.Worker.Tests/Infrastructure/FakeUsageGate.cs @@ -0,0 +1,11 @@ +using ClaudeDo.Worker.Usage; +using ClaudeDo.Worker.Usage.Interfaces; + +namespace ClaudeDo.Worker.Tests.Infrastructure; + +public sealed class FakeUsageGate : IUsageGate +{ + public UsageGateDecision Decision { get; set; } = new(false, null); + + public Task EvaluateAsync(CancellationToken ct = default) => Task.FromResult(Decision); +} diff --git a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotGuardTests.cs b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotGuardTests.cs index a9cbc2fd..e6fbaac8 100644 --- a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotGuardTests.cs +++ b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotGuardTests.cs @@ -6,6 +6,7 @@ using ClaudeDo.Worker.Hub; using ClaudeDo.Worker.Queue; using ClaudeDo.Worker.Runner; using ClaudeDo.Worker.Tests.Infrastructure; +using ClaudeDo.Worker.Usage; using Microsoft.Extensions.Logging.Abstractions; using TaskStatus = ClaudeDo.Data.Models.TaskStatus; @@ -59,7 +60,8 @@ public sealed class QueueServiceSlotGuardTests : IDisposable _waker = new QueueWaker(); var picker = new QueuePicker(dbFactory); var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, built.RunCancels); - var service = new QueueService(dbFactory, runner, _cfg, NullLogger.Instance, _waker, picker, overrideSlot, state, built.RunCancels); + var service = new QueueService(dbFactory, runner, _cfg, NullLogger.Instance, _waker, picker, overrideSlot, state, built.RunCancels, + new FakeUsageGate(), new UsageState(), broadcaster); return (service, fake); } diff --git a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceTests.cs b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceTests.cs index e276e6a1..5430a721 100644 --- a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceTests.cs +++ b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceTests.cs @@ -7,6 +7,7 @@ using ClaudeDo.Worker.Hub; using ClaudeDo.Worker.Queue; using ClaudeDo.Worker.Runner; using ClaudeDo.Worker.Tests.Infrastructure; +using ClaudeDo.Worker.Usage; using Microsoft.Extensions.Logging.Abstractions; using TaskStatus = ClaudeDo.Data.Models.TaskStatus; @@ -44,12 +45,16 @@ public sealed class QueueServiceTests : IDisposable } private QueueWaker _waker = null!; + private FakeUsageGate _usageGate = null!; + private CapturingHubContext _hubContext = null!; private (QueueService service, FakeClaudeProcess fakeProcess) CreateService( - Func, Func, CancellationToken, Task>? handler = null) + Func, Func, CancellationToken, Task>? handler = null, + FakeUsageGate? usageGate = null) { var fake = new FakeClaudeProcess(handler); - var broadcaster = new HubBroadcaster(new CapturingHubContext()); + _hubContext = new CapturingHubContext(); + var broadcaster = new HubBroadcaster(_hubContext); var dbFactory = _db.CreateFactory(); var wtManager = new WorktreeManager(new GitService(), dbFactory, _cfg, NullLogger.Instance); var argsBuilder = new ClaudeArgsBuilder(); @@ -60,7 +65,9 @@ public sealed class QueueServiceTests : IDisposable _waker = new QueueWaker(); var picker = new QueuePicker(dbFactory); var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, built.RunCancels); - var service = new QueueService(dbFactory, runner, _cfg, NullLogger.Instance, _waker, picker, overrideSlot, state, built.RunCancels); + _usageGate = usageGate ?? new FakeUsageGate(); + var service = new QueueService(dbFactory, runner, _cfg, NullLogger.Instance, _waker, picker, overrideSlot, state, built.RunCancels, + _usageGate, new UsageState(), broadcaster); return (service, fake); } @@ -342,4 +349,107 @@ public sealed class QueueServiceTests : IDisposable tcs.SetResult(new RunResult { ExitCode = 0, ResultMarkdown = "ok" }); } + + [Fact] + public async Task Blocked_UsageGate_Skips_Queue_Refill() + { + var listId = await SeedListAsync(); + await SeedQueuedTask(listId); + + var gate = new FakeUsageGate { Decision = new UsageGateDecision(true, "5h-Limit 90% >= 80%") }; + var (service, fake) = CreateService( + (_, _, _, _, _) => Task.FromResult(new RunResult { ExitCode = 0, ResultMarkdown = "ok" }), gate); + + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + _waker.Wake(); + await Task.Delay(200); + cts.Cancel(); + + Assert.Equal(0, fake.CallCount); + } + + [Fact] + public async Task UsageGate_Clearing_Resumes_QueueRefill_OnNextTick() + { + var listId = await SeedListAsync(); + await SeedQueuedTask(listId); + + var gate = new FakeUsageGate { Decision = new UsageGateDecision(true, "5h-Limit 90% >= 80%") }; + var done = new TaskCompletionSource(); + var (service, fake) = CreateService((_, _, _, _, _) => + { + done.TrySetResult(); + return Task.FromResult(new RunResult { ExitCode = 0, ResultMarkdown = "ok" }); + }, gate); + + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + _waker.Wake(); + await Task.Delay(150); + Assert.Equal(0, fake.CallCount); + + // Clear the gate; the 50ms backstop timer in this test's config picks it up. + gate.Decision = new UsageGateDecision(false, null); + await done.Task.WaitAsync(TimeSpan.FromSeconds(5)); + cts.Cancel(); + + Assert.Equal(1, fake.CallCount); + } + + [Fact] + public async Task Blocked_UsageGate_Does_Not_Cancel_AlreadyRunning_Slot() + { + var listId = await SeedListAsync(); + await SeedQueuedTask(listId); + + var running = new TaskCompletionSource(); + var cancelled = false; + var gate = new FakeUsageGate(); + var (service, _) = CreateService(async (_, _, _, _, ct) => + { + running.SetResult(); + try + { + await Task.Delay(Timeout.Infinite, ct); + } + catch (OperationCanceledException) + { + cancelled = true; + throw; + } + return new RunResult { ExitCode = 0, ResultMarkdown = "ok" }; + }, gate); + + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + _waker.Wake(); + await running.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + // Block after the slot is already running — several backstop ticks pass. + gate.Decision = new UsageGateDecision(true, "5h-Limit 90% >= 80%"); + await Task.Delay(200); + + Assert.False(cancelled); + cts.Cancel(); + } + + [Fact] + public async Task UsageGate_TransitionLogging_FiresOncePerChange() + { + var gate = new FakeUsageGate { Decision = new UsageGateDecision(true, "5h-Limit 90% >= 80%") }; + var (service, _) = CreateService( + (_, _, _, _, _) => Task.FromResult(new RunResult { ExitCode = 0, ResultMarkdown = "ok" }), gate); + + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + + // Several backstop ticks (50ms interval) all observe the same blocked state. + await Task.Delay(200); + cts.Cancel(); + + var warnCalls = _hubContext.Proxy.Calls + .Count(c => c.Method == "WorkerLog" && (WorkerLogLevel)c.Args[1]! == WorkerLogLevel.Warn); + Assert.Equal(1, warnCalls); + } } diff --git a/tests/ClaudeDo.Worker.Tests/Usage/UsageGateTests.cs b/tests/ClaudeDo.Worker.Tests/Usage/UsageGateTests.cs new file mode 100644 index 00000000..e5437f18 --- /dev/null +++ b/tests/ClaudeDo.Worker.Tests/Usage/UsageGateTests.cs @@ -0,0 +1,105 @@ +using ClaudeDo.Data.Repositories; +using ClaudeDo.Worker.Tests.Infrastructure; +using ClaudeDo.Worker.Usage; +using Microsoft.Extensions.Logging.Abstractions; + +namespace ClaudeDo.Worker.Tests.Usage; + +public sealed class UsageGateTests : IDisposable +{ + private readonly DbFixture _db = new(); + + public void Dispose() => _db.Dispose(); + + private async Task SetThresholdsAsync(int fiveHourPct, int sevenDayPct) + { + using var ctx = _db.CreateContext(); + var repo = new AppSettingsRepository(ctx); + var settings = await repo.GetAsync(); + settings.UsageGateFiveHourPct = fiveHourPct; + settings.UsageGateSevenDayPct = sevenDayPct; + await repo.UpdateAsync(settings); + } + + private UsageGate CreateGate(UsageState state) => + new(_db.CreateFactory(), state, NullLogger.Instance); + + private static UsageState StateWithSnapshot(double fiveHourPct, double sevenDayPct) + { + var state = new UsageState(); + state.ReportSuccess(new UsageSnapshot( + new UsageBucket(fiveHourPct, null), + new UsageBucket(sevenDayPct, null), + Array.Empty(), + DateTime.UtcNow)); + return state; + } + + [Fact] + public async Task BothUnderThreshold_NotBlocked() + { + await SetThresholdsAsync(80, 90); + var decision = await CreateGate(StateWithSnapshot(50, 60)).EvaluateAsync(); + Assert.False(decision.IsBlocked); + } + + [Fact] + public async Task FiveHourAtOrOverThreshold_Blocked() + { + await SetThresholdsAsync(80, 90); + var decision = await CreateGate(StateWithSnapshot(85, 60)).EvaluateAsync(); + Assert.True(decision.IsBlocked); + Assert.Contains("5h", decision.Reason); + } + + [Fact] + public async Task SevenDayAtOrOverThreshold_Blocked() + { + await SetThresholdsAsync(80, 90); + var decision = await CreateGate(StateWithSnapshot(50, 95)).EvaluateAsync(); + Assert.True(decision.IsBlocked); + Assert.Contains("7d", decision.Reason); + } + + [Fact] + public async Task ExactlyOnThreshold_Blocked() + { + await SetThresholdsAsync(80, 90); + var decision = await CreateGate(StateWithSnapshot(80, 60)).EvaluateAsync(); + Assert.True(decision.IsBlocked); + } + + [Fact] + public async Task ThresholdZero_ThatBucketNeverGates() + { + await SetThresholdsAsync(0, 90); + var decision = await CreateGate(StateWithSnapshot(100, 60)).EvaluateAsync(); + Assert.False(decision.IsBlocked); + } + + [Fact] + public async Task BothThresholdsZero_AlwaysFree() + { + await SetThresholdsAsync(0, 0); + var decision = await CreateGate(StateWithSnapshot(100, 100)).EvaluateAsync(); + Assert.False(decision.IsBlocked); + } + + [Fact] + public async Task NoSnapshotYet_FailsOpen() + { + await SetThresholdsAsync(80, 90); + var decision = await CreateGate(new UsageState()).EvaluateAsync(); + Assert.False(decision.IsBlocked); + } + + [Fact] + public async Task LastPollFailed_FailsOpen() + { + await SetThresholdsAsync(80, 90); + var state = StateWithSnapshot(95, 95); + state.ReportFailure("boom", DateTime.UtcNow); + var decision = await CreateGate(state).EvaluateAsync(); + Assert.False(decision.IsBlocked); + } +}