Import Codex cloud chats and record their token usage

Makefile

@@ -414,6 +414,8 @@ test-postgres: pricing-snapshot ensure-embed-dir postgres-up
 		-run 'TestRawSync.*CleanUploads|TestRawSyncBackfill|TestHostedRuntimeHealthRequiresStatusToken|TestReplicaRecipeQueryWithoutConfig' -count=1
 	TEST_PG_URL="postgres://agentsview_test:agentsview_test_password@localhost:5433/agentsview_test?sslmode=disable" \
 	CGO_ENABLED=1 go test -tags "fts5,pgtest" -v ./internal/postgres/... ./internal/activity/... -count=1 -timeout=20m
+	TEST_PG_URL="postgres://agentsview_test:agentsview_test_password@localhost:5433/agentsview_test?sslmode=disable" \
+	CGO_ENABLED=1 go test -tags "fts5,pgtest" -v ./internal/service -run '^TestPlanUsageCountsParity$$' -count=1
 
 # PostgreSQL integration tests for CI (postgres already running as service)
 POSTGRES_CORE_HEAVY_TEST := TestRawProjectionProofLookupsStayBounded
@@ -429,6 +431,7 @@ test-postgres-core-ci:
 
 test-postgres-core-fast-ci: pricing-snapshot ensure-embed-dir
 	CGO_ENABLED=1 go test -tags "fts5,pgtest" -v ./internal/postgres/... ./internal/activity/... -skip '^$(POSTGRES_CORE_HEAVY_TEST)$$' -count=1 -timeout=20m
+	CGO_ENABLED=1 go test -tags "fts5,pgtest" -v ./internal/service -run '^TestPlanUsageCountsParity$$' -count=1
 
 test-postgres-core-heavy-ci: pricing-snapshot ensure-embed-dir
 	CGO_ENABLED=1 go test -tags "fts5,pgtest" ./internal/postgres/... -list '^$(POSTGRES_CORE_HEAVY_TEST)$$' | grep -Fx '$(POSTGRES_CORE_HEAVY_TEST)'
@@ -448,10 +451,13 @@ test-clickhouse: pricing-snapshot ensure-embed-dir clickhouse-up
 	@sleep 2
 	TEST_CLICKHOUSE_URL="clickhouse://localhost:19000/default" \
 		CGO_ENABLED=1 go test -tags "fts5,chtest" -v ./internal/clickhouse/... ./internal/activity/... -count=1 -timeout=20m
+	TEST_CLICKHOUSE_URL="clickhouse://localhost:19000/default" \
+		CGO_ENABLED=1 go test -tags "fts5,chtest" -v ./internal/service -run '^TestPlanUsageCountsParity$$' -count=1
 
 # ClickHouse integration tests for CI (clickhouse already running as service)
 test-clickhouse-ci: pricing-snapshot ensure-embed-dir
 	CGO_ENABLED=1 go test -tags "fts5,chtest" -v ./internal/clickhouse/... ./internal/activity/... -count=1 -timeout=20m
+	CGO_ENABLED=1 go test -tags "fts5,chtest" -v ./internal/service -run '^TestPlanUsageCountsParity$$' -count=1
 
 # S3 discovery integration tests. testcontainers starts and tears down a
 # rustfs (S3-compatible) container automatically, so only a working Docker

README.md

@@ -388,6 +388,7 @@ local Amp thread JSON files.
 | OpenClaude            | `~/.openclaude/projects/`                                                                                                                                                                                                                            |
 | Claude Cowork         | `~/Library/Application Support/Claude/local-agent-mode-sessions/` (macOS)                                                                                                                                                                            |
 | Codex                 | `~/.codex/sessions/`                                                                                                                                                                                                                                 |
+| Codex Cloud           | Fetched from OpenAI (ChatGPT dot, Codex web and cloud tasks); off until `[agents.codex-cloud] enabled = true`                                                                                                                                        |
 | Copilot CLI           | `~/.copilot/`                                                                                                                                                                                                                                        |
 | Devin CLI             | `~/.local/share/devin/` (Linux), `~/Library/Application Support/devin/` (macOS); point `DEVIN_DIR` / `agents.devin.dirs` at the root that contains `cli/`                                                                                            |
 | Evener                | `~/.local/state/evener/projects/` (or `$XDG_STATE_HOME/evener/projects/`; semantic transcript v2)                                                                                                                                                    |

cmd/agentsview/archive_query_backend.go

@@ -39,7 +39,7 @@ type archiveQueryPolicy struct {
 
 type archiveQueryBackend interface {
 	ActivityReport(context.Context, ActivityReportConfig) (activity.Report, error)
-	DailyUsage(context.Context, dailyUsageQuery) (db.DailyUsageResult, error)
+	DailyUsage(context.Context, dailyUsageQuery) (dailyUsageReport, error)
 	SessionUsage(context.Context, sessionUsageQuery) (*sessionUsageOutput, int, error)
 	MachineLabels(context.Context) (service.MachineLabelCatalog, error)
 }
@@ -54,6 +54,15 @@ type sessionUsageQuery struct {
 	NoSync    bool
 }
 
+// dailyUsageReport is a daily usage result plus the sessions its token and
+// cost totals leave out as billed to plan usage and the sessions whose
+// token counts miss turns live capture did not see.
+type dailyUsageReport struct {
+	db.DailyUsageResult
+	PlanOnlySessions       int
+	PartlyCapturedSessions int
+}
+
 type dailyUsageQuery struct {
 	Progress       func(string)
 	Filter         db.UsageFilter
@@ -217,7 +226,7 @@ func (b daemonArchiveQueryBackend) ActivityReport(
 func (b daemonArchiveQueryBackend) DailyUsage(
 	ctx context.Context,
 	query dailyUsageQuery,
-) (db.DailyUsageResult, error) {
+) (dailyUsageReport, error) {
 	return fetchHTTPDailyUsage(ctx, b.tr, b.authToken, query)
 }
 
@@ -257,7 +266,7 @@ func (b localArchiveQueryBackend) ActivityReport(
 func (b localArchiveQueryBackend) DailyUsage(
 	ctx context.Context,
 	query dailyUsageQuery,
-) (db.DailyUsageResult, error) {
+) (dailyUsageReport, error) {
 	ensureFreshData(ctx, b.cfg, b.database, b.skipFreshData)
 	ensureUsagePricing(
 		b.database, b.offline, b.cfg.CustomModelPricing,
@@ -266,9 +275,21 @@ func (b localArchiveQueryBackend) DailyUsage(
 	var err error
 	filter.Machine, err = db.ResolveMachineFilter(ctx, b.database, filter.Machine)
 	if err != nil {
-		return db.DailyUsageResult{}, err
+		return dailyUsageReport{}, err
+	}
+	result, err := b.database.GetDailyUsage(ctx, filter)
+	if err != nil {
+		return dailyUsageReport{}, err
+	}
+	planUsage, err := b.database.GetPlanUsageSessionCounts(ctx, filter)
+	if err != nil {
+		return dailyUsageReport{}, err
 	}
-	return b.database.GetDailyUsage(ctx, filter)
+	return dailyUsageReport{
+		DailyUsageResult:       result,
+		PlanOnlySessions:       planUsage.PlanOnly,
+		PartlyCapturedSessions: planUsage.PartlyCaptured,
+	}, nil
 }
 
 func (b localArchiveQueryBackend) MachineLabels(
@@ -386,11 +407,9 @@ func (b localArchiveQueryBackend) SessionUsage(
 			u.Agent = string(def.Type)
 		}
 	}
-	return &sessionUsageOutput{
-		SessionUsage:      *u,
-		ServerRunning:     false,
-		RefreshIncomplete: refreshIncomplete,
-	}, usageExitCode(u), nil
+	out := newStoreSessionUsageOutput(u)
+	out.RefreshIncomplete = refreshIncomplete
+	return out, usageExitCode(out), nil
 }
 
 func closeArchiveQueryBackend(cleanup func()) {

cmd/agentsview/codex_cloud_jobs.go

@@ -0,0 +1,136 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"log"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"time"
+
+	"go.kenn.io/agentsview/internal/codexcloud"
+	"go.kenn.io/agentsview/internal/config"
+	"go.kenn.io/agentsview/internal/poller"
+	"go.kenn.io/agentsview/internal/server"
+)
+
+const (
+	codexCloudSyncJobName  = "codex-cloud-sync"
+	codexCloudSyncInterval = 5 * time.Minute
+	codexCloudSyncJitter   = 30 * time.Second
+	codexCloudLiveJobName  = "codex-cloud-live"
+	codexCloudLiveInterval = 30 * time.Second
+	codexCloudLiveJitter   = 5 * time.Second
+)
+
+// newCodexCloudSyncer builds the thread syncer. load re-reads config.toml
+// before each run, so a hand edit to [agents.codex-cloud], disabled_agents,
+// or [agents.codex] homes takes effect without a restart. cfg.NoSync is
+// fixed at start: it is set by --no-sync and by an archive-only archive.
+func newCodexCloudSyncer(
+	cfg config.Config, load func() (config.Config, error), status *codexcloud.StatusBoard,
+) *codexcloud.Syncer {
+	userHome, err := os.UserHomeDir()
+	if err != nil {
+		// DiscoverHomes then falls back to CODEX_HOME and configured homes.
+		log.Printf("codex-cloud: home directory: %v", err)
+	}
+	return &codexcloud.Syncer{
+		Store:  codexcloud.NewStore(filepath.Join(cfg.DataDir, "sources", "codex-cloud")),
+		Waits:  codexcloud.NewWaits(time.Now),
+		Status: status,
+		Now:    time.Now,
+		Dial:   codexcloud.Dial,
+		Gates: codexcloud.NewGateSource(
+			filepath.Join(cfg.DataDir, "config.toml"), cfg, load, os.Getenv, userHome,
+		),
+		NoSync:    cfg.NoSync,
+		ReadLogin: codexcloud.ReadLogin,
+		Version:   version,
+		Log:       log.Default(),
+	}
+}
+
+// codexCloudSyncJob registers each pass as tracked work, so an idle
+// daemon waits for a running import instead of cutting it off.
+func codexCloudSyncJob(sync *codexcloud.Syncer, idle *server.IdleTracker) poller.Job {
+	return poller.Job{
+		Name:       codexCloudSyncJobName,
+		Interval:   codexCloudSyncInterval,
+		Jitter:     codexCloudSyncJitter,
+		RunAtStart: true,
+		Run: func(ctx context.Context) error {
+			done, ok := idle.BeginWork()
+			if !ok {
+				return nil
+			}
+			defer done()
+			return sync.Run(ctx)
+		},
+	}
+}
+
+// newCodexCloudMonitor builds the live token monitor. It shares the
+// syncer's state store, whose per-account lock keeps the two jobs' writes
+// apart, and its waits, which both jobs honor. trigger queues a sync run
+// after a joined turn completes. Each joined thread holds an idle work
+// lease until it leaves, so an idle daemon does not exit mid-capture.
+func newCodexCloudMonitor(
+	sync *codexcloud.Syncer, trigger *codexCloudTrigger, idle *server.IdleTracker,
+) *codexcloud.Monitor {
+	return &codexcloud.Monitor{
+		Store:     sync.Store,
+		Waits:     sync.Waits,
+		Status:    sync.Status,
+		Now:       sync.Now,
+		Dial:      sync.Dial,
+		Gates:     sync.Gates,
+		NoSync:    sync.NoSync,
+		ReadLogin: sync.ReadLogin,
+		Trigger:   trigger.Trigger,
+		BeginWork: idle.BeginWork,
+		Version:   sync.Version,
+		Log:       sync.Log,
+	}
+}
+
+// codexCloudLiveJob finds running threads for the monitor. Joined threads
+// stay open after a run returns; the monitor owns them and their work
+// leases.
+func codexCloudLiveJob(m *codexcloud.Monitor, idle *server.IdleTracker) poller.Job {
+	return poller.Job{
+		Name:       codexCloudLiveJobName,
+		Interval:   codexCloudLiveInterval,
+		Jitter:     codexCloudLiveJitter,
+		RunAtStart: true,
+		Run: func(ctx context.Context) error {
+			done, ok := idle.BeginWork()
+			if !ok {
+				return nil
+			}
+			defer done()
+			return m.Discover(ctx)
+		},
+	}
+}
+
+// codexCloudTrigger queues a sync run on a scheduler set after the jobs
+// start, since the monitor is built before the scheduler exists.
+type codexCloudTrigger struct {
+	scheduler atomic.Pointer[poller.Scheduler]
+}
+
+// Trigger queues one sync run. A run already queued covers this request,
+// so several turns completing together add at most one run.
+func (t *codexCloudTrigger) Trigger() error {
+	s := t.scheduler.Load()
+	if s == nil {
+		return nil
+	}
+	err := s.TriggerNow(codexCloudSyncJobName)
+	if errors.Is(err, poller.ErrTriggerPending) {
+		return nil
+	}
+	return err
+}