diff --git a/README.md b/README.md index ddf1471..691c408 100644 --- a/README.md +++ b/README.md @@ -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; diff --git a/internal/codex/live_usage.go b/internal/codex/live_usage.go index c598f73..651e57f 100644 --- a/internal/codex/live_usage.go +++ b/internal/codex/live_usage.go @@ -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 @@ -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 @@ -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 @@ -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 { @@ -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 { @@ -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 { @@ -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 } @@ -607,13 +611,14 @@ 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, @@ -621,6 +626,9 @@ func (r *LiveUsageReader) addFile(path string, info os.FileInfo) error { 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) @@ -628,7 +636,7 @@ func (r *LiveUsageReader) addFile(path string, info os.FileInfo) error { 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 @@ -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 @@ -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 { @@ -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 @@ -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 } } @@ -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 @@ -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 diff --git a/internal/codex/live_usage_compressed.go b/internal/codex/live_usage_compressed.go index 34df86c..c6112f8 100644 --- a/internal/codex/live_usage_compressed.go +++ b/internal/codex/live_usage_compressed.go @@ -10,6 +10,7 @@ import ( "io" "os" "strings" + "time" "github.com/klauspost/compress/zstd" ) @@ -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++ } @@ -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") } @@ -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 @@ -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:]...) @@ -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 } @@ -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 @@ -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 @@ -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) } diff --git a/internal/codex/live_usage_compressed_test.go b/internal/codex/live_usage_compressed_test.go index 80002a0..445d58f 100644 --- a/internal/codex/live_usage_compressed_test.go +++ b/internal/codex/live_usage_compressed_test.go @@ -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] diff --git a/internal/codex/live_usage_exact_test.go b/internal/codex/live_usage_exact_test.go index 696348c..e03377a 100644 --- a/internal/codex/live_usage_exact_test.go +++ b/internal/codex/live_usage_exact_test.go @@ -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()) @@ -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()) diff --git a/internal/codex/live_usage_test.go b/internal/codex/live_usage_test.go index 8cb8148..be104b2 100644 --- a/internal/codex/live_usage_test.go +++ b/internal/codex/live_usage_test.go @@ -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) diff --git a/internal/codex/rollout_activity.go b/internal/codex/rollout_activity.go new file mode 100644 index 0000000..eda29ee --- /dev/null +++ b/internal/codex/rollout_activity.go @@ -0,0 +1,196 @@ +package codex + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "io" + "os" + "time" +) + +// Match the context bootstrap target. Extend through an overlapping record up +// to the existing 4 MiB baseline-record bound, never an unbounded history scan. +const rolloutActivityTailBytes = 256 * 1024 + +// Keep only the decoded activity window while the accounting scan traverses +// compressed history. A ring avoids copying the whole window on every chunk. +type rolloutActivityTail struct { + data []byte + next int +} + +func (t *rolloutActivityTail) append(p []byte) { + const limit = rolloutActivityTailBytes + rolloutBaselineLineLimit + if len(p) >= limit { + t.data = append(t.data[:0], p[len(p)-limit:]...) + t.next = 0 + return + } + if len(t.data) < limit { + n := min(limit-len(t.data), len(p)) + t.data = append(t.data, p[:n]...) + p = p[n:] + } + if len(p) == 0 { + return + } + n := copy(t.data[t.next:], p) + copy(t.data, p[n:]) + t.next = (t.next + len(p)) % limit +} + +func (t *rolloutActivityTail) ReadAt(p []byte, offset int64) (int, error) { + if len(p) == 0 { + return 0, nil + } + if offset < 0 { + return 0, os.ErrInvalid + } + if offset >= int64(len(t.data)) { + return 0, io.EOF + } + n := min(len(p), len(t.data)-int(offset)) + start := (t.next + int(offset)) % len(t.data) + first := copy(p[:n], t.data[start:]) + copy(p[first:n], t.data[:n-first]) + if n < len(p) { + return n, io.EOF + } + return n, nil +} + +type rolloutContextReader struct { + ctx context.Context + io.Reader +} + +func (r rolloutContextReader) Read(p []byte) (int, error) { + if err := r.ctx.Err(); err != nil { + return 0, err + } + return r.Reader.Read(p) +} + +func (r *LiveUsageReader) activityNow() time.Time { + if r.now != nil { + return r.now() + } + return time.Now() +} + +// Windows can retain an old modification time while the writer handle is open. +// Keep that metadata separate from observed growth and persisted event time. +func (c *rolloutCursor) activityTime() time.Time { + latest := c.lastModified + if c.lastEventAt.After(latest) { + latest = c.lastEventAt + } + if c.lastGrowthAt.After(latest) { + latest = c.lastGrowthAt + } + return latest +} + +func (c *rolloutCursor) observeFile(info os.FileInfo, now time.Time) { + c.lastModified = info.ModTime() + if info.Size() != c.lastSize { + c.lastGrowthAt = now + c.lastSize = info.Size() + } +} + +func (c *rolloutCursor) observeEvent(line []byte, now time.Time) { + var event struct { + Timestamp time.Time `json:"timestamp"` + Ordinal *uint64 `json:"ordinal"` + Type string `json:"type"` + } + if json.Unmarshal(line, &event) != nil || event.Type == "" || event.Type == "session_meta" || + event.Timestamp.IsZero() || event.Timestamp.After(now) || + !tokenRecordIsOwned(event.Ordinal, c.subagentHistoryStartOrdinal, event.Timestamp, c.nonRoot, c.startedAt) { + return + } + if event.Timestamp.After(c.lastEventAt) { + c.lastEventAt = event.Timestamp + } +} + +// Existing records establish activity age, never new token usage. Ignore an +// incomplete final record and future timestamps rather than keeping old logs +// active indefinitely. Only headers are decoded; no content is retained. +func (c *rolloutCursor) bootstrapActivity(ctx context.Context, path string, size int64, now time.Time) error { + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + return c.observeActivityTail(ctx, f, size, false, now) +} + +func (c *rolloutCursor) observeActivityTail(ctx context.Context, source io.ReaderAt, size int64, clipped bool, now time.Time) error { + start := max(size-rolloutActivityTailBytes, 0) + start, partial, err := rolloutActivityRecordStart(ctx, source, start) + if err != nil { + return err + } + // A bounded decoded window may begin in the middle of a record. Never + // treat that clipped prefix as a complete event when no boundary was found. + partial = partial || clipped && start == 0 + reader := bufio.NewReaderSize(io.NewSectionReader(source, start, size-start), 64*1024) + var line []byte + for { + if err := ctx.Err(); err != nil { + return err + } + chunk, readErr := reader.ReadSlice('\n') + if !partial && len(line)+len(chunk) <= rolloutBaselineLineLimit { + line = append(line, chunk...) + } else { + partial = true + line = nil + } + if len(chunk) > 0 && chunk[len(chunk)-1] == '\n' { + if !partial { + c.observeEvent(line, now) + } + line = line[:0] + partial = false + } + if readErr == bufio.ErrBufferFull { + continue + } + if readErr != nil { + if readErr == io.EOF { + return nil // An incomplete final record is not an event. + } + return readErr + } + } +} + +// Find the beginning of the record overlapping the target tail. Fixed chunks +// avoid repeatedly copying a growing carry buffer. Read at most that record, +// not earlier history. If the overlap exceeds the bound, skip that record +// conservatively but still inspect complete records later in the tail. +func rolloutActivityRecordStart(ctx context.Context, source io.ReaderAt, start int64) (int64, bool, error) { + const chunkBytes = 64 * 1024 + buffer := make([]byte, chunkBytes) + floor := max(start-rolloutBaselineLineLimit, 0) + for start > floor { + if err := ctx.Err(); err != nil { + return 0, false, err + } + size := min(start-floor, int64(len(buffer))) + start -= size + n, err := source.ReadAt(buffer[:size], start) + if err != nil && err != io.EOF { + return 0, false, err + } + if index := bytes.LastIndexByte(buffer[:n], '\n'); index >= 0 { + return start + int64(index) + 1, false, nil + } + } + return start, start > 0, nil +} diff --git a/internal/codex/rollout_activity_compressed_test.go b/internal/codex/rollout_activity_compressed_test.go new file mode 100644 index 0000000..8977c58 --- /dev/null +++ b/internal/codex/rollout_activity_compressed_test.go @@ -0,0 +1,281 @@ +package codex + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + "strings" + "testing" + "time" +) + +func TestCompressedRolloutActivityUsesOwnedEventTimeAndRestorationIsNotGrowth(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + old := now.Add(-72 * time.Hour) + home := t.TempDir() + path := testRolloutPath(t, home, old, "growing") + writeRollout(t, path, sessionMetaLine("growing", `"cli"`, "/work", nil)+"\n"+ + attentionEventLine(now, "task_started", nil)+"\n"+responseItemLine(now, "reasoning")+"\n") + plain := compressTestRollout(t, path, true) + freezeRolloutTime(t, path+".zst", old) + r, err := NewLiveUsageReader(home) + if err != nil { + t.Fatal(err) + } + r.now = func() time.Time { return now } + r.statusProvider = stubSessionStatusProvider{unavailable: true} + u, err := r.FetchTokenUsage(context.Background()) + if err != nil || u.TotalTokens != 0 || u.SessionCount != 1 || !r.files[path].lastEventAt.Equal(now) { + t.Fatal("compressed startup missed fresh events or counted history", u, err) + } + assertRolloutAttention(t, r, now, SessionAttentionNone, true) + now = now.Add(fallbackAttentionAfter + time.Second) + if err := os.WriteFile(path, plain, 0600); err != nil { + t.Fatal(err) + } + freezeRolloutTime(t, path, old) + u, err = r.FetchTokenUsageFresh(context.Background()) + if err != nil || u.TotalTokens != 0 || r.files[path].compressed || !r.files[path].lastGrowthAt.IsZero() { + t.Fatal("identical restoration invented activity or tokens", u, err) + } + assertRolloutAttention(t, r, now, SessionAttentionCheck, true) + appendRollout(t, path, responseItemLine(now, "reasoning")+"\n") + freezeRolloutTime(t, path, old) + u, err = r.FetchTokenUsage(context.Background()) + if err != nil || u.TotalTokens != 0 { + t.Fatal("fresh append failed or counted tokens", u, err) + } + assertRolloutAttention(t, r, now, SessionAttentionNone, true) +} + +func TestRolloutActivityCompressionPreservesGrowthAndPartialBaseline(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + old := now.Add(-time.Hour) + home := t.TempDir() + path := testRolloutPath(t, home, now, "growing") + writeRollout(t, path, sessionMetaLine("growing", `"cli"`, "/work", nil)+"\n"+attentionEventLine(old, "task_started", nil)+"\n") + freezeRolloutTime(t, path, old) + r, err := NewLiveUsageReader(home) + if err != nil { + t.Fatal(err) + } + r.now = func() time.Time { return now } + r.statusProvider = stubSessionStatusProvider{unavailable: true} + if _, err := r.FetchTokenUsage(context.Background()); err != nil { + t.Fatal(err) + } + appendRollout(t, path, `{"timestamp":`) + freezeRolloutTime(t, path, old) + if _, err := r.FetchTokenUsage(context.Background()); err != nil { + t.Fatal(err) + } + assertRolloutAttention(t, r, now, SessionAttentionNone, true) + plain := compressTestRollout(t, path, true) + freezeRolloutTime(t, path+".zst", old) + if _, err := r.FetchTokenUsageFresh(context.Background()); err != nil { + t.Fatal(err) + } + if !r.files[path].lastGrowthAt.Equal(now) || r.files[path].lastSize != int64(len(plain)) { + t.Fatal("compression lost observed growth or partial bytes", r.files[path]) + } + assertRolloutAttention(t, r, now, SessionAttentionNone, true) + now = now.Add(fallbackAttentionAfter + time.Second) + if err := os.WriteFile(path, plain, 0600); err != nil { + t.Fatal(err) + } + freezeRolloutTime(t, path, old) + if _, err := r.FetchTokenUsageFresh(context.Background()); err != nil { + t.Fatal(err) + } + assertRolloutAttention(t, r, now, SessionAttentionCheck, true) +} + +func TestActivityBootstrapBoundsOverlapAndStillReadsLaterEvents(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + for _, suffix := range []string{"\n", "\n" + responseItemLine(now, "reasoning") + "\n"} { + path := testRolloutPath(t, t.TempDir(), now, "huge") + // The giant record advertises a timestamp but is too large to validate. + // It must not establish freshness; a subsequent small valid record can. + data := `{"timestamp":"` + now.Format(time.RFC3339) + `","type":"response_item","payload":"` + + strings.Repeat("x", rolloutBaselineLineLimit+rolloutActivityTailBytes) + `"}` + suffix + writeRollout(t, path, data) + f, err := os.Open(path) + if err != nil { + t.Fatal(err) + } + target := int64(len(data) - rolloutActivityTailBytes) + start, partial, err := rolloutActivityRecordStart(context.Background(), f, target) + f.Close() + if err != nil || !partial || start != target-rolloutBaselineLineLimit { + t.Fatal("overlap scan exceeded its bound", start, partial, err) + } + c := &rolloutCursor{} + if err := c.bootstrapActivity(context.Background(), path, int64(len(data)), now); err != nil { + t.Fatal(err) + } + if suffix == "\n" && !c.lastEventAt.IsZero() || suffix != "\n" && !c.lastEventAt.Equal(now) { + t.Fatal("oversized record was trusted or hid later event", c.lastEventAt) + } + } +} + +func TestActivityBootstrapAndRestorationHonourCancellation(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + path := testRolloutPath(t, t.TempDir(), now, "cancel") + writeRollout(t, path, sessionMetaLine("s", `"cli"`, "/work", nil)+"\n"+responseItemLine(now, "reasoning")+"\n") + plain := compressTestRollout(t, path, true) + info, err := os.Stat(path + ".zst") + if err != nil { + t.Fatal(err) + } + c, err := compressedRolloutBaseline(context.Background(), path+".zst", info, now) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, plain, 0600); err != nil { + t.Fatal(err) + } + info, err = os.Stat(path) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := c.bootstrapActivity(ctx, path, info.Size(), now); !errors.Is(err, context.Canceled) { + t.Fatal("bootstrap ignored cancellation", err) + } + r := &LiveUsageReader{} + if err := r.restoreCompressedCursor(ctx, path, info, c); !errors.Is(err, context.Canceled) || !c.compressed { + t.Fatal("restoration ignored cancellation or committed partial state", err) + } +} + +func TestCompressedRolloutActivityRejectsInheritedAndFutureEvents(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + boundary := uint64(10) + for _, tc := range []struct { + name string + ordinal uint64 + at time.Time + want bool + }{{"inherited", 9, now, false}, {"future", 10, now.Add(time.Hour), false}, {"owned", 10, now, true}} { + t.Run(tc.name, func(t *testing.T) { + path := testRolloutPath(t, t.TempDir(), now, "child") + line := fmt.Sprintf(`{"timestamp":%q,"ordinal":%d,"type":"response_item","payload":{"type":"reasoning"}}`, tc.at.Format(time.RFC3339), tc.ordinal) + writeRollout(t, path, sessionMetaLine("child", threadSpawnSource("parent"), "/work", &boundary)+"\n"+line+"\n") + compressTestRollout(t, path, true) + info, err := os.Stat(path + ".zst") + if err != nil { + t.Fatal(err) + } + c, err := compressedRolloutBaseline(context.Background(), path+".zst", info, now) + if err != nil || tc.want && !c.lastEventAt.Equal(now) || !tc.want && !c.lastEventAt.IsZero() { + t.Fatal("wrong compressed ownership/future-time decision", tc.name, c, err) + } + }) + } +} + +func TestRebasedRolloutPreservesPreviouslyObservedGrowth(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + home := t.TempDir() + path := testRolloutPath(t, home, now, "growing") + writeRollout(t, path, sessionMetaLine("growing", `"cli"`, "/work", nil)+"\n") + freezeRolloutTime(t, path, now.Add(-time.Hour)) + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + growth := now.Add(-time.Minute) + r := &LiveUsageReader{initialized: true, startedAt: now.Add(-time.Hour), now: func() time.Time { return now }, files: map[string]*rolloutCursor{ + path: {threadID: "growing", lastGrowthAt: growth}, + }} + if err := r.addFile(context.Background(), path, info); err != nil || !r.files[path].lastGrowthAt.Equal(growth) { + t.Fatal("rebasing discarded real observed activity", r.files[path], err) + } +} + +func TestCompressedActivityUsesSameBoundedTailAsPlainHistory(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + old := now.Add(-time.Hour) + // The fresh timestamp is outside even the permitted overlap. The full + // accounting pass still needs the token baseline from this older region. + filler := `{"timestamp":"` + old.Format(time.RFC3339) + `","type":"response_item","payload":{"type":"reasoning","text":"` + + strings.Repeat("x", 3*1024*1024) + `"}}` + "\n" + data := sessionMetaLine("growing", `"cli"`, "/work", nil) + "\n" + + attentionEventLine(now, "task_started", nil) + "\n" + tokenCountLine(now, 900) + "\n" + filler + filler + for _, compressed := range []bool{false, true} { + home := t.TempDir() + path := testRolloutPath(t, home, old, "growing") + writeRollout(t, path, data) + freezeRolloutTime(t, path, old) + if compressed { + compressTestRollout(t, path, true) + freezeRolloutTime(t, path+".zst", old) + } + r, err := NewLiveUsageReader(home) + if err != nil { + t.Fatal(err) + } + r.now = func() time.Time { return now } + r.statusProvider = stubSessionStatusProvider{unavailable: true} + u, err := r.FetchTokenUsage(context.Background()) + if err != nil { + t.Fatal(err) + } + c := r.files[path] + if !c.lastEventAt.Equal(old) || !c.activityTime().Equal(old) || u.SessionCount != 0 { + t.Fatal("out-of-window timestamp restored freshness", compressed, c.lastEventAt, c.activityTime(), u.SessionCount) + } + if c.totalTokens != 900 || u.TotalTokens != 0 { + t.Fatal("bounded activity scan changed the full accounting baseline", compressed, c.totalTokens, u.TotalTokens) + } + } +} + +func TestRolloutActivityTailRetainsOnlyNewestBytesAcrossWraps(t *testing.T) { + limit := rolloutActivityTailBytes + rolloutBaselineLineLimit + data := make([]byte, 2*limit+1600) + for i := range data { + data[i] = byte((i*37 + i/251 + i/65521) % 251) + } + for _, chunks := range [][]int{{1, 65535, 123, limit, 17}, {limit + 10, 0, 33}, {limit - 1, 2, limit - 1, 100}} { + var tail rolloutActivityTail + used := 0 + for _, n := range chunks { + tail.append(data[used : used+n]) + used += n + if len(tail.data) != min(used, limit) { + t.Fatal("tail exceeded limit or dropped recent bytes", used, len(tail.data)) + } + want := data[max(used-limit, 0):used] + got, err := io.ReadAll(io.NewSectionReader(&tail, 0, int64(len(tail.data)))) + if err != nil || !bytes.Equal(got, want) { + t.Fatal("ring lost byte order", chunks, n, err) + } + // Match ReaderAt behaviour for reads that cross the physical wrap, + // end of data, and the EOF boundary, not just whole-window reads. + for _, offset := range []int64{0, int64(len(want) / 2), int64(len(want) - 1), int64(len(want))} { + got, expected := make([]byte, 73), make([]byte, 73) + n, err := tail.ReadAt(got, offset) + wn, werr := bytes.NewReader(want).ReadAt(expected, offset) + if n != wn || !errors.Is(err, werr) || !bytes.Equal(got, expected) { + t.Fatal("tail ReaderAt disagrees with ordered bytes", chunks, offset, n, wn, err, werr) + } + } + } + } +} + +func TestClippedActivityTailCannotTrustItsInitialPartialRecord(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + old := now.Add(-time.Hour) + data := responseItemLine(now, "reasoning") + "\n" + responseItemLine(old, "reasoning") + "\n" + c := &rolloutCursor{} + if err := c.observeActivityTail(context.Background(), strings.NewReader(data), int64(len(data)), true, now); err != nil || !c.lastEventAt.Equal(old) { + t.Fatal("clipped prefix established freshness or hid later complete record", c.lastEventAt, err) + } +} diff --git a/internal/codex/rollout_activity_smoke_test.go b/internal/codex/rollout_activity_smoke_test.go new file mode 100644 index 0000000..a4d2f57 --- /dev/null +++ b/internal/codex/rollout_activity_smoke_test.go @@ -0,0 +1,49 @@ +package codex + +import ( + "context" + "os" + "testing" +) + +// Explicit opt-in, read-only check against an already running local session. +// It never starts an app-server, a model turn, or a tool in that session. +func TestLiveRolloutActivitySmoke(t *testing.T) { + home, thread := os.Getenv("CODEXOMETER_ACTIVITY_SMOKE_HOME"), os.Getenv("CODEXOMETER_ACTIVITY_SMOKE_THREAD") + if home == "" || thread == "" { + t.Skip("set both CODEXOMETER_ACTIVITY_SMOKE_HOME and CODEXOMETER_ACTIVITY_SMOKE_THREAD") + } + r, err := NewLiveUsageReader(home) + if err != nil { + t.Fatal(err) + } + r.statusProvider = stubSessionStatusProvider{unavailable: true} + if _, err := r.FetchTokenUsage(context.Background()); err != nil { + t.Fatal(err) + } + for _, c := range r.files { + if c.threadID != thread { + continue + } + now := r.activityNow() + t.Logf("filesystem=%s event=%s detected=%s age=%s", c.lastModified, c.lastEventAt, c.activityTime(), now.Sub(c.activityTime())) + if now.Sub(c.lastModified) < fallbackAttentionAfter { + t.Skip("filesystem timestamp is fresh; stale-timestamp case not present") + } + if now.Sub(c.lastEventAt) >= fallbackAttentionAfter { + t.Skip("no fresh persisted event available in this session") + } + sessions, _, _ := r.sessionSnapshots(now, map[string]struct{}{thread: {}}, true, nil) + for _, s := range sessions { + if s.ID == thread { + t.Logf("session=%s active=%v working=%v attention=%v", thread, s.Active, s.Working, s.Attention) + if s.Attention == SessionAttentionCheck { + t.Fatal("fresh actual events still produced CHECK SESSION") + } + return + } + } + t.Fatal("session was hidden despite fresh events") + } + t.Fatal("requested session log not found") +} diff --git a/internal/codex/rollout_activity_test.go b/internal/codex/rollout_activity_test.go new file mode 100644 index 0000000..23e2b98 --- /dev/null +++ b/internal/codex/rollout_activity_test.go @@ -0,0 +1,264 @@ +package codex + +import ( + "context" + "encoding/json" + "os" + "strings" + "testing" + "time" +) + +func TestRolloutActivityBootstrapUsesEventTimeWithStaleModTime(t *testing.T) { + now := time.Date(2026, time.October, 23, 9, 0, 0, 0, time.UTC) + old := now.Add(-72 * time.Hour) + home := t.TempDir() + path := testRolloutPath(t, home, old, "growing") + writeRollout(t, path, sessionMetaLine("growing", `"cli"`, "/work", nil)+"\n"+ + attentionEventLine(now.Add(-time.Minute), "task_started", nil)+"\n"+ + responseItemLine(now.Add(-time.Second), "reasoning")+"\n") + freezeRolloutTime(t, path, old) + r, err := NewLiveUsageReader(home) + if err != nil { + t.Fatal(err) + } + r.now = func() time.Time { return now } + r.statusProvider = stubSessionStatusProvider{unavailable: true} + usage, err := r.FetchTokenUsage(context.Background()) + if err != nil || usage.TotalTokens != 0 { + t.Fatalf("bootstrap counted tokens or failed: %+v, %v", usage, err) + } + assertRolloutAttention(t, r, now, SessionAttentionNone, true) + if !r.files[path].activityTime().Equal(now.Add(-time.Second)) { + t.Fatalf("bootstrap activity = %v", r.files[path].activityTime()) + } +} + +func TestRolloutActivityBootstrapOversizedLatestRecord(t *testing.T) { + now := time.Date(2026, time.October, 23, 9, 0, 0, 0, time.UTC) + for _, kind := range []string{"function_call_output", "reasoning"} { + t.Run(kind, func(t *testing.T) { + old := now.Add(-72 * time.Hour) + home := t.TempDir() + path := testRolloutPath(t, home, old, "growing") + payload := map[string]any{"type": kind, "output": strings.Repeat("x", rolloutActivityTailBytes+1)} + line, err := json.Marshal(map[string]any{"timestamp": now, "type": "response_item", "payload": payload}) + if err != nil { + t.Fatal(err) + } + writeRollout(t, path, sessionMetaLine("growing", `"cli"`, "/work", nil)+"\n"+ + attentionEventLine(old, "task_started", nil)+"\n"+string(line)+"\n") + freezeRolloutTime(t, path, old) + r, err := NewLiveUsageReader(home) + if err != nil { + t.Fatal(err) + } + r.now = func() time.Time { return now } + r.statusProvider = stubSessionStatusProvider{unavailable: true} + usage, err := r.FetchTokenUsage(context.Background()) + if err != nil || usage.TotalTokens != 0 { + t.Fatalf("bootstrap = %+v, %v", usage, err) + } + if !r.files[path].lastEventAt.Equal(now) { + t.Fatalf("latest owned event lost: %v", r.files[path].lastEventAt) + } + assertRolloutAttention(t, r, now, SessionAttentionNone, true) + }) + } +} + +func TestRolloutActivityBootstrapKeepsRecordAtExactTailBoundary(t *testing.T) { + now := time.Date(2026, time.October, 23, 9, 0, 0, 0, time.UTC) + path := testRolloutPath(t, t.TempDir(), now, "boundary") + recent := responseItemLine(now, "reasoning") + "\n" + // The first recent record begins exactly at size-tailTarget. The rest of + // the tail is an unfinished record and must not hide the complete one. + data := "{}\n" + recent + strings.Repeat(" ", rolloutActivityTailBytes-len(recent)) + writeRollout(t, path, data) + c := &rolloutCursor{} + if err := c.bootstrapActivity(context.Background(), path, int64(len(data)), now); err != nil { + t.Fatal(err) + } + if !c.lastEventAt.Equal(now) { + t.Fatalf("boundary event dropped: %v", c.lastEventAt) + } +} + +func TestRolloutActivityBootstrapOversizedPartialRecord(t *testing.T) { + now := time.Date(2026, time.October, 23, 9, 0, 0, 0, time.UTC) + path := testRolloutPath(t, t.TempDir(), now, "partial") + line, err := json.Marshal(map[string]any{"timestamp": now, "type": "response_item", "payload": map[string]any{ + "type": "reasoning", "text": strings.Repeat("x", rolloutActivityTailBytes+1), + }}) + if err != nil { + t.Fatal(err) + } + writeRollout(t, path, string(line)) + c := &rolloutCursor{} + if err := c.bootstrapActivity(context.Background(), path, int64(len(line)), now); err != nil { + t.Fatal(err) + } + if !c.lastEventAt.IsZero() { + t.Fatalf("unterminated oversized event accepted: %v", c.lastEventAt) + } +} + +func TestRolloutActivityBootstrapRespectsOrdinalBoundary(t *testing.T) { + now := time.Date(2026, time.October, 23, 9, 0, 0, 0, time.UTC) + path := testRolloutPath(t, t.TempDir(), now, "ordinal") + line := `{"timestamp":"2026-10-23T09:00:00Z","ordinal":9,"type":"response_item","payload":{"type":"reasoning"}}` + "\n" + writeRollout(t, path, line) + boundary := uint64(10) + c := &rolloutCursor{nonRoot: true, subagentHistoryStartOrdinal: &boundary} + if err := c.bootstrapActivity(context.Background(), path, int64(len(line)), now); err != nil { + t.Fatal(err) + } + if !c.lastEventAt.IsZero() { + t.Fatalf("inherited ordinal accepted: %v", c.lastEventAt) + } +} + +func TestRolloutActivityGrowthClearsCheckAndUnchangedPollsExpire(t *testing.T) { + now := time.Date(2026, time.October, 23, 9, 0, 0, 0, time.UTC) + old := now.Add(-72 * time.Hour) + home := t.TempDir() + path := testRolloutPath(t, home, old, "growing") + writeRollout(t, path, sessionMetaLine("growing", `"cli"`, "/work", nil)+"\n"+ + attentionEventLine(old, "task_started", nil)+"\n") + freezeRolloutTime(t, path, old) + r, err := NewLiveUsageReader(home) + if err != nil { + t.Fatal(err) + } + r.now = func() time.Time { return now } + r.statusProvider = stubSessionStatusProvider{unavailable: true} + if _, err := r.FetchTokenUsage(context.Background()); err != nil { + t.Fatal(err) + } + // A real open writer would produce CHECK while its old log is quiet. + r.files[path].attention = sessionAttentionWorking + assertRolloutAttention(t, r, now, SessionAttentionCheck, true) + // A complete event arrives with no token-count update or mtime change. + appendRollout(t, path, responseItemLine(now, "reasoning")+"\n") + freezeRolloutTime(t, path, old) + usage, err := r.FetchTokenUsageFresh(context.Background()) + if err != nil || usage.TotalTokens != 0 { + t.Fatalf("growth counted tokens or failed: %+v, %v", usage, err) + } + assertRolloutAttention(t, r, now, SessionAttentionNone, true) + // Re-reading/discovery of unchanged bytes must not keep the log fresh. + now = now.Add(fallbackAttentionAfter + time.Second) + if _, err := r.FetchTokenUsageFresh(context.Background()); err != nil { + t.Fatal(err) + } + assertRolloutAttention(t, r, now, SessionAttentionCheck, true) +} + +func TestRolloutActivityPartialRecordRefreshesOnlyWhenSizeChanges(t *testing.T) { + now := time.Date(2026, time.October, 23, 9, 0, 0, 0, time.UTC) + old := now.Add(-time.Hour) + home := t.TempDir() + path := testRolloutPath(t, home, old, "growing") + writeRollout(t, path, sessionMetaLine("growing", `"cli"`, "/work", nil)+"\n") + freezeRolloutTime(t, path, old) + r, err := NewLiveUsageReader(home) + if err != nil { + t.Fatal(err) + } + r.now = func() time.Time { return now } + r.statusProvider = stubSessionStatusProvider{unavailable: true} + if _, err := r.FetchTokenUsage(context.Background()); err != nil { + t.Fatal(err) + } + r.files[path].attention = sessionAttentionWorking + appendRollout(t, path, `{"timestamp":`) + freezeRolloutTime(t, path, old) + if _, err := r.FetchTokenUsage(context.Background()); err != nil { + t.Fatal(err) + } + assertRolloutAttention(t, r, now, SessionAttentionNone, true) + now = now.Add(fallbackAttentionAfter + time.Second) + if _, err := r.FetchTokenUsageFresh(context.Background()); err != nil { + t.Fatal(err) + } + assertRolloutAttention(t, r, now, SessionAttentionCheck, true) +} + +func TestRolloutActivityPreservesDefiniteRequests(t *testing.T) { + now := time.Date(2026, time.October, 23, 9, 0, 0, 0, time.UTC) + for _, tc := range []struct { + kind string + want SessionAttention + }{ + {"request_user_input", SessionAttentionInput}, + {"exec_approval_request", SessionAttentionApproval}, + } { + t.Run(tc.kind, func(t *testing.T) { + home := t.TempDir() + old := now.Add(-time.Hour) + path := testRolloutPath(t, home, old, "growing") + writeRollout(t, path, sessionMetaLine("growing", `"cli"`, "/work", nil)+"\n"+ + attentionEventLine(old, tc.kind, nil)+"\n") + freezeRolloutTime(t, path, old) + r, err := NewLiveUsageReader(home) + if err != nil { + t.Fatal(err) + } + r.now = func() time.Time { return now } + r.statusProvider = stubSessionStatusProvider{unavailable: true} + if _, err := r.FetchTokenUsage(context.Background()); err != nil { + t.Fatal(err) + } + appendRollout(t, path, responseItemLine(now, "reasoning")+"\n") + freezeRolloutTime(t, path, old) + if _, err := r.FetchTokenUsage(context.Background()); err != nil { + t.Fatal(err) + } + assertRolloutAttention(t, r, now, tc.want, false) + }) + } +} + +func TestRolloutActivityIgnoresFutureInheritedAndPartialEvents(t *testing.T) { + now := time.Date(2026, time.October, 23, 9, 0, 0, 0, time.UTC) + for _, tc := range []struct { + name, data string + child bool + }{ + {"future", responseItemLine(now.Add(time.Hour), "reasoning") + "\n", false}, + {"inherited", responseItemLine(now.Add(-time.Hour), "reasoning") + "\n", true}, + {"partial", responseItemLine(now, "reasoning"), false}, + {"malformed", "not JSON\n", false}, + } { + t.Run(tc.name, func(t *testing.T) { + path := testRolloutPath(t, t.TempDir(), now, tc.name) + writeRollout(t, path, tc.data) + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + c := &rolloutCursor{nonRoot: tc.child, startedAt: now} + if err := c.bootstrapActivity(context.Background(), path, info.Size(), now); err != nil { + t.Fatal(err) + } + if !c.lastEventAt.IsZero() { + t.Fatalf("unreliable activity accepted: %v", c.lastEventAt) + } + }) + } +} + +func freezeRolloutTime(t *testing.T, path string, at time.Time) { + t.Helper() + if err := os.Chtimes(path, at, at); err != nil { + t.Fatal(err) + } +} + +func assertRolloutAttention(t *testing.T, r *LiveUsageReader, now time.Time, want SessionAttention, working bool) { + t.Helper() + sessions, _, _ := r.sessionSnapshots(now, map[string]struct{}{"growing": {}}, true, nil) + if len(sessions) != 1 || sessions[0].Attention != want || sessions[0].Working != working { + t.Fatalf("sessions = %+v; want attention %v, working %v", sessions, want, working) + } +}