fix(worker): record real raw token usage per run, not the uncached remainder
task_runs.tokens_in only ever held the API's uncached "input_tokens" field (off by a factor of ~400,000 on a resumed session), and tokens_out summed only the last result event instead of the whole session. TaskRunner now reads each run's cache-read/cache-write/input/output totals from the session transcript via a new ITranscriptUsageReader.ReadSessionTotalsAsync, storing the delta against prior runs on the same session so a --resume doesn't double-count. New task_runs.cache_read_tokens/cache_write_tokens columns; the Session tab now shows the raw total (what actually counts against the 5h/7d usage limit) with a breakdown tooltip.
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, broadcasts `HubBroadcaster.UsageUpdated` after every tick via `UsageSnapshotBuilder`), UsageSnapshotBuilder (builds the Hub-facing `UsageSnapshotDto` from `UsageState` + `IUsageGate` + `AppSettings` thresholds — the one place `WorkerHub.GetUsageSnapshot` and `UsageMonitorService` share the stale/threshold/gate logic), 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)
|
||||
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, broadcasts `HubBroadcaster.UsageUpdated` after every tick via `UsageSnapshotBuilder`), UsageSnapshotBuilder (builds the Hub-facing `UsageSnapshotDto` from `UsageState` + `IUsageGate` + `AppSettings` thresholds — the one place `WorkerHub.GetUsageSnapshot` and `UsageMonitorService` share the stale/threshold/gate logic), 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; `<synthetic>`-model lines are skipped everywhere, not real API calls; also exposes `ReadSessionTotalsAsync(sessionId)` — cumulative raw input/output/cache-read/cache-write totals for one session, located by `{sessionId}.jsonl`, used by `TaskRunner` to populate `task_runs`' per-run token fields), 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.
|
||||
@@ -180,7 +180,15 @@ A child that hits a roadblock (fails, or reports `CLAUDEDO_BLOCKED` roadblocks)
|
||||
## Execution History
|
||||
|
||||
Each CLI invocation is recorded in the `task_runs` table via `TaskRunRepository`:
|
||||
- Fields: `session_id`, input/output/cache token counts, turn count, `result` text, structured output JSON
|
||||
- Fields: `session_id`, turn count, `result` text, structured output JSON, and the four raw token
|
||||
fields (`tokens_in`/`tokens_out`/`cache_read_tokens`/`cache_write_tokens`) — **not** read from the
|
||||
stream-json "result" event's `usage.input_tokens` (that's only the uncached remainder of one API
|
||||
call and undercounts the real prompt size by orders of magnitude once caching kicks in). Instead
|
||||
`TaskRunner.ApplyUsageAsync` reads `ITranscriptUsageReader.ReadSessionTotalsAsync(sessionId)` —
|
||||
the session transcript's cumulative totals across every assistant message — and stores the
|
||||
**delta** against prior `task_runs` rows sharing the same `session_id`, so a `--resume`'d run
|
||||
doesn't double-count the turns already billed to an earlier run. A missing/unreadable transcript
|
||||
leaves all four fields `null`; it never fails the run.
|
||||
- Enables auto-retry on failure (resume last session) and multi-turn follow-up via `ContinueAsync`
|
||||
|
||||
## Multi-Turn / Continue
|
||||
|
||||
@@ -6,6 +6,8 @@ using ClaudeDo.Worker.Config;
|
||||
using ClaudeDo.Worker.Hub;
|
||||
using ClaudeDo.Worker.Skills;
|
||||
using ClaudeDo.Worker.State;
|
||||
using ClaudeDo.Worker.Usage;
|
||||
using ClaudeDo.Worker.Usage.Interfaces;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using TaskStatus = ClaudeDo.Data.Models.TaskStatus;
|
||||
|
||||
@@ -24,6 +26,7 @@ public sealed class TaskRunner
|
||||
private readonly TaskRunTokenRegistry _tokens;
|
||||
private readonly AttachmentStore _attachments;
|
||||
private readonly ISessionSkillSeeder _skillSeeder;
|
||||
private readonly ITranscriptUsageReader _usageReader;
|
||||
|
||||
public TaskRunner(
|
||||
IClaudeProcess claude,
|
||||
@@ -36,7 +39,8 @@ public sealed class TaskRunner
|
||||
ITaskStateService state,
|
||||
TaskRunTokenRegistry tokens,
|
||||
AttachmentStore attachments,
|
||||
ISessionSkillSeeder skillSeeder)
|
||||
ISessionSkillSeeder skillSeeder,
|
||||
ITranscriptUsageReader usageReader)
|
||||
{
|
||||
_claude = claude;
|
||||
_dbFactory = dbFactory;
|
||||
@@ -49,6 +53,7 @@ public sealed class TaskRunner
|
||||
_tokens = tokens;
|
||||
_attachments = attachments;
|
||||
_skillSeeder = skillSeeder;
|
||||
_usageReader = usageReader;
|
||||
}
|
||||
|
||||
public async Task RunAsync(TaskEntity task, string slot, CancellationToken ct, bool alreadyClaimed = false)
|
||||
@@ -361,8 +366,8 @@ public sealed class TaskRunner
|
||||
run.ErrorMarkdown = result.ErrorMarkdown;
|
||||
run.ExitCode = result.ExitCode;
|
||||
run.TurnCount = result.TurnCount;
|
||||
run.TokensIn = result.TokensIn;
|
||||
run.TokensOut = result.TokensOut;
|
||||
if (result.SessionId is not null)
|
||||
await ApplyUsageAsync(run, taskId, result.SessionId);
|
||||
run.FinishedAt = DateTime.UtcNow;
|
||||
|
||||
using (var context = _dbFactory.CreateDbContext())
|
||||
@@ -394,6 +399,40 @@ public sealed class TaskRunner
|
||||
}
|
||||
}
|
||||
|
||||
/// Populates the run's raw token fields from the session transcript (input, output,
|
||||
/// cache-read, cache-write — the API's "input_tokens" alone is only the uncached
|
||||
/// remainder and undercounts the real prompt size by orders of magnitude). A resumed
|
||||
/// session's transcript is cumulative, so the delta against prior runs sharing the same
|
||||
/// SessionId is stored, not the running total. Any failure here (missing/unreadable
|
||||
/// transcript) leaves the fields null and must never fail the run itself.
|
||||
private async Task ApplyUsageAsync(TaskRunEntity run, string taskId, string sessionId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var totals = await _usageReader.ReadSessionTotalsAsync(sessionId, CancellationToken.None);
|
||||
if (totals is null) return;
|
||||
|
||||
List<TaskRunEntity> priorRuns;
|
||||
using (var context = _dbFactory.CreateDbContext())
|
||||
priorRuns = await new TaskRunRepository(context).GetByTaskIdAsync(taskId, CancellationToken.None);
|
||||
|
||||
var sameSession = priorRuns.Where(r => r.Id != run.Id && r.SessionId == sessionId).ToList();
|
||||
var priorInput = sameSession.Sum(r => (long)(r.TokensIn ?? 0));
|
||||
var priorOutput = sameSession.Sum(r => (long)(r.TokensOut ?? 0));
|
||||
var priorCacheRead = sameSession.Sum(r => (long)(r.CacheReadTokens ?? 0));
|
||||
var priorCacheWrite = sameSession.Sum(r => (long)(r.CacheWriteTokens ?? 0));
|
||||
|
||||
run.TokensIn = (int)Math.Max(0, totals.InputTokens - priorInput);
|
||||
run.TokensOut = (int)Math.Max(0, totals.OutputTokens - priorOutput);
|
||||
run.CacheReadTokens = (int)Math.Max(0, totals.CacheReadTokens - priorCacheRead);
|
||||
run.CacheWriteTokens = (int)Math.Max(0, totals.CacheCreationTokens - priorCacheWrite);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to read session usage totals for task {TaskId}, session {SessionId}", taskId, sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleSuccess(TaskEntity task, ListEntity list, string slot, WorktreeContext? wtCtx, RunResult result, CancellationToken ct)
|
||||
{
|
||||
if (wtCtx is not null)
|
||||
|
||||
@@ -4,4 +4,9 @@ public interface ITranscriptUsageReader
|
||||
{
|
||||
Task<IReadOnlyList<UsageAggregateRow>> ReadAsync(
|
||||
DateOnly start, DateOnly end, CancellationToken ct = default);
|
||||
|
||||
/// <summary>Cumulative raw token totals for one session's transcript file
|
||||
/// (located by <c>{sessionId}.jsonl</c> under the projects root), or null when
|
||||
/// no matching transcript file can be found or read.</summary>
|
||||
Task<SessionUsageTotals?> ReadSessionTotalsAsync(string sessionId, CancellationToken ct = default);
|
||||
}
|
||||
|
||||
@@ -62,6 +62,33 @@ public sealed class TranscriptUsageReader : ITranscriptUsageReader
|
||||
return Task.FromResult<IReadOnlyList<UsageAggregateRow>>(rows);
|
||||
}
|
||||
|
||||
public Task<SessionUsageTotals?> ReadSessionTotalsAsync(string sessionId, CancellationToken ct = default)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(sessionId) || !Directory.Exists(_projectsRoot))
|
||||
return Task.FromResult<SessionUsageTotals?>(null);
|
||||
|
||||
var file = Directory
|
||||
.EnumerateFiles(_projectsRoot, $"{sessionId}.jsonl", SearchOption.AllDirectories)
|
||||
.FirstOrDefault();
|
||||
if (file is null) return Task.FromResult<SessionUsageTotals?>(null);
|
||||
|
||||
var seenKeys = new HashSet<string>();
|
||||
long input = 0, output = 0, cacheRead = 0, cacheCreation = 0;
|
||||
foreach (var record in GetOrReadFile(file))
|
||||
{
|
||||
ct.ThrowIfCancellationRequested();
|
||||
if (!seenKeys.Add(record.DedupeKey)) continue;
|
||||
|
||||
input += record.InputTokens;
|
||||
output += record.OutputTokens;
|
||||
cacheRead += record.CacheReadTokens;
|
||||
cacheCreation += record.CacheCreationTokens;
|
||||
}
|
||||
|
||||
return Task.FromResult<SessionUsageTotals?>(
|
||||
new SessionUsageTotals(input, output, cacheRead, cacheCreation));
|
||||
}
|
||||
|
||||
private List<UsageMessageRecord> GetOrReadFile(string file)
|
||||
{
|
||||
var info = new FileInfo(file);
|
||||
@@ -100,6 +127,7 @@ public sealed class TranscriptUsageReader : ITranscriptUsageReader
|
||||
|
||||
var date = DateOnly.FromDateTime(ts.LocalDateTime);
|
||||
var model = modelEl.GetString()!;
|
||||
if (model == "<synthetic>") continue;
|
||||
|
||||
long input = 0, output = 0, cacheRead = 0, cacheCreation = 0;
|
||||
if (msg.TryGetProperty("usage", out var usage) && usage.ValueKind == JsonValueKind.Object)
|
||||
|
||||
@@ -32,3 +32,11 @@ public sealed record UsageAggregateRow(
|
||||
long CacheReadTokens,
|
||||
long CacheCreationTokens,
|
||||
int Messages);
|
||||
|
||||
/// <summary>Cumulative raw token usage for one Claude Code session (all its transcript
|
||||
/// lines to date), synthetic messages excluded. Not a per-run delta.</summary>
|
||||
public sealed record SessionUsageTotals(
|
||||
long InputTokens,
|
||||
long OutputTokens,
|
||||
long CacheReadTokens,
|
||||
long CacheCreationTokens);
|
||||
|
||||
Reference in New Issue
Block a user