feat(claude-do): Worker: UsageGate — Queue ab Schwelle pausieren
> **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
This commit is contained in:
@@ -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<T>()`. 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:
|
||||
|
||||
@@ -206,6 +206,7 @@ builder.Services.AddHttpClient<IUsageClient, ClaudeOAuthUsageClient>(client =>
|
||||
client.Timeout = TimeSpan.FromSeconds(5);
|
||||
});
|
||||
builder.Services.AddHostedService<UsageMonitorService>();
|
||||
builder.Services.AddSingleton<IUsageGate, UsageGate>();
|
||||
|
||||
// Loopback-only bind. Firewall is irrelevant for 127.0.0.1.
|
||||
builder.WebHost.UseUrls($"http://127.0.0.1:{cfg.SignalRPort}");
|
||||
|
||||
@@ -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<string, QueueSlotState> _queueSlots = new();
|
||||
private bool _usageGateBlocked;
|
||||
|
||||
public QueueService(
|
||||
IDbContextFactory<ClaudeDoDbContext> 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<int> GetMaxParallelAsync(CancellationToken ct)
|
||||
{
|
||||
try
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
namespace ClaudeDo.Worker.Usage.Interfaces;
|
||||
|
||||
public interface IUsageGate
|
||||
{
|
||||
Task<UsageGateDecision> EvaluateAsync(CancellationToken ct = default);
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public sealed class UsageGate : IUsageGate
|
||||
{
|
||||
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
|
||||
private readonly UsageState _state;
|
||||
private readonly ILogger<UsageGate> _logger;
|
||||
|
||||
public UsageGate(IDbContextFactory<ClaudeDoDbContext> dbFactory, UsageState state, ILogger<UsageGate> logger)
|
||||
{
|
||||
_dbFactory = dbFactory;
|
||||
_state = state;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public async Task<UsageGateDecision> 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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user