Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -1216,6 +1216,16 @@ The other top-level views are:
minutes receives `CHECK SESSION`: a deliberately uncertain prompt that can
also mean a long-running local tool. Subsequent activity from the root or any
linked agent clears the uncertain group-level warning.
Fallback activity detection uses owned event timestamps and observed file-size
changes as well as modification times, since Windows can retain an old
modification time while a rollout writer remains open. Plain and compressed
histories share these signals; identical restoration, unchanged polls and
repeated reads of an incomplete record do not invent new activity.
Startup examines the last 256 KiB, extending backward by at most 4 MiB to
include an overlapping record. Records larger than the existing 4 MiB baseline
limit, incomplete records, future timestamps and inherited parent events are
conservatively excluded from timestamp-based freshness; later complete records
are still checked. These cancellable, bounded activity scans do not recount tokens.
Confirmed failed turns instead enter **ERRORS**, with an amber **MODEL BUSY**
badge for server-reported model capacity and **TURN FAILED** for other failures.
The row, full detail and attention pill retain the observed error message;
Expand Down
38 changes: 24 additions & 14 deletions internal/codex/live_usage.go
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,7 @@ type LiveTurnTiming struct {
// sessions. It also extracts bounded display-only replies and request context;
// reasoning and arbitrary tool output are never retained.
type LiveUsageReader struct {
now func() time.Time
knownUsageCoverage map[string]string
localFeatureChecks map[string]localFeatureCheck
discoveryCoverageIssue string
Expand Down Expand Up @@ -173,6 +174,9 @@ type rolloutCursor struct {
totalTokens int64
observedTokens int64
lastModified time.Time
lastSize int64
lastEventAt time.Time
lastGrowthAt time.Time
threadID string
parentThreadID string
workingDirectory string
Expand Down Expand Up @@ -379,7 +383,7 @@ func (r *LiveUsageReader) fetchTokenUsage(ctx context.Context, forceFullDiscover
return LiveUsageSnapshot{}, err
}

now := time.Now()
now := r.activityNow()
if !r.initialized {
if err := r.initialize(ctx, now); err != nil {
return LiveUsageSnapshot{}, err
Expand Down Expand Up @@ -422,7 +426,7 @@ func (r *LiveUsageReader) fetchTokenUsage(ctx context.Context, forceFullDiscover
continue
}
if err == nil {
err = r.restoreCompressedCursor(path, info, cursor)
err = r.restoreCompressedCursor(ctx, path, info, cursor)
}
if err != nil {
if firstReadErr == nil {
Expand All @@ -432,7 +436,7 @@ func (r *LiveUsageReader) fetchTokenUsage(ctx context.Context, forceFullDiscover
}
cursor = r.files[path]
}
if now.Sub(cursor.lastModified) > activeRolloutHorizon {
if now.Sub(cursor.activityTime()) > activeRolloutHorizon {
continue
}
if err := r.consume(path, cursor); err != nil {
Expand Down Expand Up @@ -508,7 +512,7 @@ func (r *LiveUsageReader) observedThreadIDs(now time.Time) []string {
seen := make(map[string]struct{}, len(r.files))
threadIDs := make([]string, 0, len(r.files))
for _, cursor := range r.files {
if cursor.threadID == "" || now.Sub(cursor.lastModified) > activeRolloutHorizon {
if cursor.threadID == "" || now.Sub(cursor.activityTime()) > activeRolloutHorizon {
continue
}
if _, ok := seen[cursor.threadID]; ok {
Expand Down Expand Up @@ -581,12 +585,12 @@ func (r *LiveUsageReader) discover(ctx context.Context, now time.Time, full bool
}
if cursor, exists := r.files[path]; exists {
if cursor.compressed {
return r.restoreCompressedCursor(path, info, cursor)
return r.restoreCompressedCursor(ctx, path, info, cursor)
}
cursor.lastModified = info.ModTime()
cursor.observeFile(info, r.activityNow())
return nil
}
if err := r.addFile(path, info); err != nil {
if err := r.addFile(ctx, path, info); err != nil {
if firstFileErr == nil {
firstFileErr = err
}
Expand All @@ -607,28 +611,32 @@ func (r *LiveUsageReader) discover(ctx context.Context, now time.Time, full bool
return nil
}

func (r *LiveUsageReader) addFile(path string, info os.FileInfo) error {
func (r *LiveUsageReader) addFile(ctx context.Context, path string, info os.FileInfo) error {
metadata, err := readRolloutMetadata(path)
if err != nil {
return err
}
cursor := &rolloutCursor{
lastModified: info.ModTime(),
lastSize: info.Size(),
threadID: metadata.ID,
parentThreadID: metadata.ParentThreadID,
workingDirectory: metadata.WorkingDirectory,
startedAt: metadata.StartedAt,
nonRoot: metadata.NonRoot,
subagentHistoryStartOrdinal: metadata.SubagentHistoryStartOrdinal,
}
if err := cursor.bootstrapActivity(ctx, path, info.Size(), r.activityNow()); err != nil {
return err
}
// Both bootstrap scans reject late failures from older turns. Establish
// their turn scope first, including when discovery finds a new rollout.
turnID, err := latestRolloutTurnID(path, cursor)
if err != nil {
return err
}
cursor.currentTurnID = turnID
if time.Since(info.ModTime()) <= activeRolloutHorizon {
if r.activityNow().Sub(cursor.activityTime()) <= activeRolloutHorizon {
attention, err := latestSessionAttention(path, cursor)
if err != nil {
return err
Expand Down Expand Up @@ -671,6 +679,7 @@ func (r *LiveUsageReader) addFile(path string, info os.FileInfo) error {
// already observed by this process. Keep it only for the same thread.
if previous := r.files[path]; previous != nil && cursor.threadID != "" && previous.threadID == cursor.threadID {
cursor.observedTokens, cursor.modelCalls, cursor.turnTimings = previous.observedTokens, previous.modelCalls, previous.turnTimings
cursor.lastGrowthAt = previous.lastGrowthAt
}
r.files[path] = cursor
return nil
Expand All @@ -686,7 +695,7 @@ func (r *LiveUsageReader) consume(path string, cursor *rolloutCursor) error {
if err != nil {
return err
}
cursor.lastModified = info.ModTime()
cursor.observeFile(info, r.activityNow())
if info.Size() < cursor.offset {
total, latestErr := latestTokenTotal(path)
if latestErr != nil {
Expand Down Expand Up @@ -736,6 +745,7 @@ func (r *LiveUsageReader) consume(path string, cursor *rolloutCursor) error {
line, readErr := reader.ReadBytes('\n')
if len(line) > 0 && line[len(line)-1] == '\n' {
cursor.offset += int64(len(line))
cursor.observeEvent(line, r.activityNow())
if settings, ordinal, at, ok := rolloutSessionSettingsRecord(line, cursor.modelSettings); ok &&
tokenRecordIsOwned(ordinal, cursor.subagentHistoryStartOrdinal, at, cursor.nonRoot, cursor.startedAt) {
cursor.modelSettings = settings
Expand Down Expand Up @@ -1420,7 +1430,7 @@ func (r *LiveUsageReader) sessionSnapshots(now time.Time, liveWriters map[string
for _, cursor := range r.files {
if cursor.threadID != "" {
previous := byID[cursor.threadID]
if previous == nil || cursor.lastModified.After(previous.lastModified) {
if previous == nil || cursor.activityTime().After(previous.activityTime()) {
byID[cursor.threadID] = cursor
}
}
Expand All @@ -1430,7 +1440,7 @@ func (r *LiveUsageReader) sessionSnapshots(now time.Time, liveWriters map[string
groupHasFreshActivity := make(map[string]bool)
groupWorking := make(map[string]bool)
for _, cursor := range r.files {
quietFor := now.Sub(cursor.lastModified)
quietFor := now.Sub(cursor.activityTime())
_, writerActive := liveWriters[cursor.threadID]
exactStatus, exact := exactStatuses[cursor.threadID]
// A currently loaded, ready daemon thread is still active even without
Expand Down Expand Up @@ -1500,8 +1510,8 @@ func (r *LiveUsageReader) sessionSnapshots(now time.Time, liveWriters map[string
// root is demonstrably active, even if a linked child was left quiet
// overnight. Definite input and approval signals still propagate.
groupHasFreshActivity[rootID] = groupHasFreshActivity[rootID] || quietFor < fallbackAttentionAfter || exactWorking
if cursor.lastModified.After(group.LastActivity) {
group.LastActivity = cursor.lastModified
if cursor.activityTime().After(group.LastActivity) {
group.LastActivity = cursor.activityTime()
}
if group.WorkingDirectory == "" {
group.WorkingDirectory = cursor.workingDirectory
Expand Down
21 changes: 15 additions & 6 deletions internal/codex/live_usage_compressed.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (
"io"
"os"
"strings"
"time"

"github.com/klauspost/compress/zstd"
)
Expand All @@ -36,12 +37,13 @@ func (r *LiveUsageReader) addCompressedFile(ctx context.Context, path string, in
if old != nil && old.compressed && old.compressedInfo.Size() == info.Size() && old.compressedInfo.ModTime() == info.ModTime() {
return nil
}
cursor, err := compressedRolloutBaseline(ctx, path, info)
cursor, err := compressedRolloutBaseline(ctx, path, info, r.activityNow())
if err != nil {
return fmt.Errorf("read compressed rollout: %w", err)
}
if old != nil && old.threadID == cursor.threadID {
cursor.observedTokens, cursor.modelCalls, cursor.turnTimings = old.observedTokens, old.modelCalls, old.turnTimings
cursor.lastGrowthAt = old.lastGrowthAt
if old.offset != cursor.offset {
r.apiEqAccountingAdjustments++
}
Expand All @@ -50,7 +52,7 @@ func (r *LiveUsageReader) addCompressedFile(ctx context.Context, path string, in
return nil
}

func compressedRolloutBaseline(ctx context.Context, path string, info os.FileInfo) (*rolloutCursor, error) {
func compressedRolloutBaseline(ctx context.Context, path string, info os.FileInfo, now time.Time) (*rolloutCursor, error) {
if info.Size() > compressedHistoryLimit {
return nil, errors.New("compressed history exceeds 256 MiB source limit")
}
Expand All @@ -67,6 +69,7 @@ func compressedRolloutBaseline(ctx context.Context, path string, info os.FileInf
reader := bufio.NewReaderSize(io.LimitReader(decoder, compressedHistoryLimit+1), 64<<10)
cursor := &rolloutCursor{lastModified: info.ModTime(), compressed: true, compressedInfo: info}
var line, tail []byte
var activityTail rolloutActivityTail
digest := sha256.New()
var decoded int64
oversized := false
Expand All @@ -80,6 +83,7 @@ func compressedRolloutBaseline(ctx context.Context, path string, info os.FileInf
if decoded > compressedHistoryLimit {
return nil, errors.New("decoded history exceeds 256 MiB baseline limit")
}
activityTail.append(chunk)
tail = append(tail, chunk...)
if len(tail) > 4096 {
tail = append(tail[:0], tail[len(tail)-4096:]...)
Expand Down Expand Up @@ -112,9 +116,14 @@ func compressedRolloutBaseline(ctx context.Context, path string, info os.FileInf
}
}
cursor.exactUsageUncertain = cursor.exactUsageUncertain && !cursor.exactUsage
cursor.lastSize = decoded // logical JSONL bytes, including an incomplete tail, not archive size
if cursor.threadID == "" {
return nil, errors.New("compressed history omitted session metadata")
}
size := int64(len(activityTail.data))
if err := cursor.observeActivityTail(ctx, &activityTail, size, decoded > size, now); err != nil {
return nil, err
}
return cursor, nil
}

Expand Down Expand Up @@ -156,7 +165,7 @@ func baselineRolloutRecord(c *rolloutCursor, line []byte) {
}
}

func (r *LiveUsageReader) restoreCompressedCursor(path string, info os.FileInfo, cursor *rolloutCursor) error {
func (r *LiveUsageReader) restoreCompressedCursor(ctx context.Context, path string, info os.FileInfo, cursor *rolloutCursor) error {
file, err := os.Open(path)
if err != nil {
return err
Expand All @@ -169,7 +178,7 @@ func (r *LiveUsageReader) restoreCompressedCursor(path string, info os.FileInfo,
matches := false
if readErr == nil && n == len(tail) && bytes.Equal(tail, cursor.compressedTail) && info.Size() >= cursor.offset {
digest := sha256.New()
_, err = io.Copy(digest, io.NewSectionReader(file, 0, cursor.offset))
_, err = io.Copy(digest, rolloutContextReader{ctx, io.NewSectionReader(file, 0, cursor.offset)})
if err != nil {
file.Close()
return err
Expand All @@ -181,10 +190,10 @@ func (r *LiveUsageReader) restoreCompressedCursor(path string, info os.FileInfo,
cursor.compressed = false
cursor.compressedInfo = nil
cursor.compressedTail = nil
cursor.lastModified = info.ModTime()
cursor.observeFile(info, r.activityNow())
return nil
}
// A changed/truncated restoration cannot reuse an offset from different
// history. Existing-thread discovery rebases it rather than billing copies.
return r.addFile(path, info)
return r.addFile(ctx, path, info)
}
2 changes: 1 addition & 1 deletion internal/codex/live_usage_compressed_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -243,7 +243,7 @@ func TestReplacingRolloutWithDifferentThreadDoesNotInheritActivity(t *testing.T)
if err != nil {
t.Fatal(err)
}
if err := r.addFile(path, info); err != nil {
if err := r.addFile(context.Background(), path, info); err != nil {
t.Fatal(err)
}
cursor := r.files[path]
Expand Down
4 changes: 2 additions & 2 deletions internal/codex/live_usage_exact_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -202,7 +202,7 @@ func TestExactDiscoveryDoesNotBaselineConcurrentAppendOrPartialRecord(t *testing
appendRollout(t, path, line)
}
delete(r.files, path)
if err := r.addFile(path, file); err != nil {
if err := r.addFile(context.Background(), path, file); err != nil {
t.Fatal(err)
}
got, err := r.FetchTokenUsage(context.Background())
Expand All @@ -226,7 +226,7 @@ func TestExactReplacementRolloutDoesNotBillCopiedThreadHistory(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if err := r.addFile(replacement, info); err != nil {
if err := r.addFile(context.Background(), replacement, info); err != nil {
t.Fatal(err)
}
got, err := r.FetchTokenUsage(context.Background())
Expand Down
10 changes: 5 additions & 5 deletions internal/codex/live_usage_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -182,26 +182,26 @@ func TestFreshLiveUsageReadNoticesAResumedDormantSession(t *testing.T) {

func TestLiveUsageReaderExpiresInactiveSessionCount(t *testing.T) {
home := t.TempDir()
now := time.Now()
now := time.Date(2026, time.October, 23, 9, 0, 0, 0, time.UTC)
path := testRolloutPath(t, home, now.Add(-time.Hour), "inactive")
writeRollout(t, path, tokenCountLine(now.Add(-time.Hour), 100)+"\n")
reader, err := NewLiveUsageReader(home)
if err != nil {
t.Fatal(err)
}
freezeRolloutTime(t, path, now)
reader.now = func() time.Time { return now }
if _, err := reader.FetchTokenUsage(context.Background()); err != nil {
t.Fatal(err)
}
appendRollout(t, path, tokenCountLine(now, 125)+"\n")
freezeRolloutTime(t, path, now)
usage, err := reader.FetchTokenUsage(context.Background())
if err != nil || usage.SessionCount != 1 {
t.Fatalf("active session count = %#v, %v; want one", usage, err)
}

inactiveAt := now.Add(-6 * time.Minute)
if err := os.Chtimes(path, inactiveAt, inactiveAt); err != nil {
t.Fatal(err)
}
now = now.Add(6 * time.Minute)
usage, err = reader.FetchTokenUsage(context.Background())
if err != nil || usage.SessionCount != 0 {
t.Fatalf("inactive session count = %#v, %v; want zero", usage, err)
Expand Down
Loading
Loading