Merge claudedo/38394081d47048fea82317c6c52e01a5

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, 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

src/ClaudeDo.Worker/Runner/TaskRunner.cs

@@ -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)
@@ -364,8 +369,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())
@@ -397,6 +402,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)

src/ClaudeDo.Worker/Usage/Interfaces/ITranscriptUsageReader.cs

@@ -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);
 }

src/ClaudeDo.Worker/Usage/TranscriptUsageReader.cs

@@ -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)

src/ClaudeDo.Worker/Usage/UsageModels.cs

@@ -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);