From ea4c345712e312c088efbcf9e8c5f8e0cc29b48c Mon Sep 17 00:00:00 2001 From: Peace7 Date: Fri, 23 Oct 2026 13:28:58 +0300 Subject: [PATCH 1/3] FIX: detect rollout activity despite stale Windows timestamps Use owned event timestamps and observed size changes alongside filesystem mtime. Bootstrap complete tail records, including oversized records, without recounting tokens. Preserve definite requests and real inactivity warnings. --- README.md | 5 + internal/codex/live_usage.go | 29 +- internal/codex/live_usage_test.go | 10 +- internal/codex/rollout_activity.go | 103 +++++++ internal/codex/rollout_activity_smoke_test.go | 49 ++++ internal/codex/rollout_activity_test.go | 264 ++++++++++++++++++ 6 files changed, 445 insertions(+), 15 deletions(-) create mode 100644 internal/codex/rollout_activity.go create mode 100644 internal/codex/rollout_activity_smoke_test.go create mode 100644 internal/codex/rollout_activity_test.go diff --git a/README.md b/README.md index 6156f81..b534550 100644 --- a/README.md +++ b/README.md @@ -1173,6 +1173,11 @@ 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. + Activity detection uses recorded 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. Startup examines a + recent tail, extending through its overlapping JSONL record so large tool + results and reasoning items still establish activity without recounting tokens. Codexometer never guesses `APPROVAL NEEDED` from inactivity. Closing the CLI releases its per-thread writer lock and clears every attention badge. A session already included in the current Sessions recording can remain as an diff --git a/internal/codex/live_usage.go b/internal/codex/live_usage.go index b58a9c6..7b837e2 100644 --- a/internal/codex/live_usage.go +++ b/internal/codex/live_usage.go @@ -126,6 +126,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 sessionNames map[string]string nameIndexInfo os.FileInfo daemonContexts map[string]SessionContext @@ -167,6 +168,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 @@ -365,7 +369,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 @@ -400,7 +404,7 @@ func (r *LiveUsageReader) fetchTokenUsage(ctx context.Context, forceFullDiscover if err := contextErr(ctx); err != nil { return LiveUsageSnapshot{}, err } - if now.Sub(cursor.lastModified) > activeRolloutHorizon { + if now.Sub(cursor.activityTime()) > activeRolloutHorizon { continue } if err := r.consume(path, cursor); err != nil { @@ -462,7 +466,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 { @@ -522,7 +526,7 @@ func (r *LiveUsageReader) discover(ctx context.Context, now time.Time, full bool return nil } if cursor, exists := r.files[path]; exists { - cursor.lastModified = info.ModTime() + cursor.observeFile(info, r.activityNow()) return nil } if err := r.addFile(path, info); err != nil { @@ -553,6 +557,7 @@ func (r *LiveUsageReader) addFile(path string, info os.FileInfo) error { } cursor := &rolloutCursor{ lastModified: info.ModTime(), + lastSize: info.Size(), threadID: metadata.ID, parentThreadID: metadata.ParentThreadID, workingDirectory: metadata.WorkingDirectory, @@ -560,7 +565,10 @@ func (r *LiveUsageReader) addFile(path string, info os.FileInfo) error { nonRoot: metadata.NonRoot, subagentHistoryStartOrdinal: metadata.SubagentHistoryStartOrdinal, } - if time.Since(info.ModTime()) <= activeRolloutHorizon { + if err := cursor.bootstrapActivity(path, info.Size(), r.activityNow()); err != nil { + return err + } + if r.activityNow().Sub(cursor.activityTime()) <= activeRolloutHorizon { attention, err := latestSessionAttention(path, cursor) if err != nil { return err @@ -618,7 +626,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 { @@ -668,6 +676,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 @@ -1329,7 +1338,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 } } @@ -1339,7 +1348,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 @@ -1404,8 +1413,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_test.go b/internal/codex/live_usage_test.go index d254bfd..90c4adb 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..521398b --- /dev/null +++ b/internal/codex/rollout_activity.go @@ -0,0 +1,103 @@ +package codex + +import ( + "bytes" + "encoding/json" + "io" + "os" + "time" +) + +// Match the context bootstrap target. Extend through one overlapping record +// when needed, so a large tool result or reasoning item is not discarded. +const rolloutActivityTailBytes = 256 * 1024 + +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(path string, size int64, now time.Time) error { + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + start := max(size-rolloutActivityTailBytes, 0) + start, err = rolloutActivityRecordStart(f, start) + if err != nil { + return err + } + data, err := io.ReadAll(io.NewSectionReader(f, start, size-start)) + if err != nil { + return err + } + lines := bytes.Split(data, []byte{'\n'}) + for _, line := range lines[:max(len(lines)-1, 0)] { + c.observeEvent(line, now) + } + return nil +} + +// 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; memory for decoding follows the existing JSONL reader's +// largest-record behavior rather than imposing a truncating record limit. +func rolloutActivityRecordStart(f *os.File, start int64) (int64, error) { + const chunkBytes = 64 * 1024 + buffer := make([]byte, chunkBytes) + for start > 0 { + size := min(start, int64(len(buffer))) + start -= size + n, err := f.ReadAt(buffer[:size], start) + if err != nil && err != io.EOF { + return 0, err + } + if index := bytes.LastIndexByte(buffer[:n], '\n'); index >= 0 { + return start + int64(index) + 1, nil + } + } + return 0, nil +} 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..dcde489 --- /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(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(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(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(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) + } +} From 8112a30f0a4dc4399dedc26db5c46f7a5a65e65c Mon Sep 17 00:00:00 2001 From: merefield Date: Tue, 27 Oct 2026 17:42:51 +0000 Subject: [PATCH 2/3] FIX: harden rollout activity across compressed history --- README.md | 13 +- internal/codex/live_usage.go | 11 +- internal/codex/live_usage_compressed.go | 16 +- internal/codex/live_usage_compressed_test.go | 2 +- internal/codex/live_usage_exact_test.go | 4 +- internal/codex/rollout_activity.go | 79 +++++-- .../codex/rollout_activity_compressed_test.go | 197 ++++++++++++++++++ internal/codex/rollout_activity_test.go | 8 +- 8 files changed, 288 insertions(+), 42 deletions(-) create mode 100644 internal/codex/rollout_activity_compressed_test.go diff --git a/README.md b/README.md index 5bc82f5..691c408 100644 --- a/README.md +++ b/README.md @@ -1216,11 +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. - Activity detection uses recorded event timestamps and observed file-size + 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. Startup examines a - recent tail, extending through its overlapping JSONL record so large tool - results and reasoning items still establish activity without recounting tokens. + 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 520a94e..651e57f 100644 --- a/internal/codex/live_usage.go +++ b/internal/codex/live_usage.go @@ -426,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 { @@ -585,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.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 } @@ -611,7 +611,7 @@ 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 @@ -626,7 +626,7 @@ func (r *LiveUsageReader) addFile(path string, info os.FileInfo) error { nonRoot: metadata.NonRoot, subagentHistoryStartOrdinal: metadata.SubagentHistoryStartOrdinal, } - if err := cursor.bootstrapActivity(path, info.Size(), r.activityNow()); err != nil { + if err := cursor.bootstrapActivity(ctx, path, info.Size(), r.activityNow()); err != nil { return err } // Both bootstrap scans reject late failures from older turns. Establish @@ -679,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 diff --git a/internal/codex/live_usage_compressed.go b/internal/codex/live_usage_compressed.go index 34df86c..194ff33 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") } @@ -100,6 +102,7 @@ func compressedRolloutBaseline(ctx context.Context, path string, info os.FileInf cursor.compressedTail = append(cursor.compressedTail[:0], tail...) if !oversized { baselineRolloutRecord(cursor, line) + cursor.observeEvent(line, now) } line = line[:0] oversized = false @@ -112,6 +115,7 @@ 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") } @@ -156,7 +160,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 +173,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 +185,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/rollout_activity.go b/internal/codex/rollout_activity.go index 521398b..f533743 100644 --- a/internal/codex/rollout_activity.go +++ b/internal/codex/rollout_activity.go @@ -1,17 +1,31 @@ package codex import ( + "bufio" "bytes" + "context" "encoding/json" "io" "os" "time" ) -// Match the context bootstrap target. Extend through one overlapping record -// when needed, so a large tool result or reasoning item is not discarded. +// 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 +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() @@ -59,45 +73,70 @@ func (c *rolloutCursor) observeEvent(line []byte, now time.Time) { // 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(path string, size int64, now time.Time) error { +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() start := max(size-rolloutActivityTailBytes, 0) - start, err = rolloutActivityRecordStart(f, start) - if err != nil { - return err - } - data, err := io.ReadAll(io.NewSectionReader(f, start, size-start)) + start, partial, err := rolloutActivityRecordStart(ctx, f, start) if err != nil { return err } - lines := bytes.Split(data, []byte{'\n'}) - for _, line := range lines[:max(len(lines)-1, 0)] { - c.observeEvent(line, now) + reader := bufio.NewReaderSize(io.NewSectionReader(f, 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 + } } - return nil } // 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; memory for decoding follows the existing JSONL reader's -// largest-record behavior rather than imposing a truncating record limit. -func rolloutActivityRecordStart(f *os.File, start int64) (int64, error) { +// 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, f *os.File, start int64) (int64, bool, error) { const chunkBytes = 64 * 1024 buffer := make([]byte, chunkBytes) - for start > 0 { - size := min(start, int64(len(buffer))) + 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 := f.ReadAt(buffer[:size], start) if err != nil && err != io.EOF { - return 0, err + return 0, false, err } if index := bytes.LastIndexByte(buffer[:n], '\n'); index >= 0 { - return start + int64(index) + 1, nil + return start + int64(index) + 1, false, nil } } - return 0, 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..f9b84c7 --- /dev/null +++ b/internal/codex/rollout_activity_compressed_test.go @@ -0,0 +1,197 @@ +package codex + +import ( + "context" + "errors" + "fmt" + "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) + } +} diff --git a/internal/codex/rollout_activity_test.go b/internal/codex/rollout_activity_test.go index dcde489..23e2b98 100644 --- a/internal/codex/rollout_activity_test.go +++ b/internal/codex/rollout_activity_test.go @@ -76,7 +76,7 @@ func TestRolloutActivityBootstrapKeepsRecordAtExactTailBoundary(t *testing.T) { data := "{}\n" + recent + strings.Repeat(" ", rolloutActivityTailBytes-len(recent)) writeRollout(t, path, data) c := &rolloutCursor{} - if err := c.bootstrapActivity(path, int64(len(data)), now); err != nil { + if err := c.bootstrapActivity(context.Background(), path, int64(len(data)), now); err != nil { t.Fatal(err) } if !c.lastEventAt.Equal(now) { @@ -95,7 +95,7 @@ func TestRolloutActivityBootstrapOversizedPartialRecord(t *testing.T) { } writeRollout(t, path, string(line)) c := &rolloutCursor{} - if err := c.bootstrapActivity(path, int64(len(line)), now); err != nil { + if err := c.bootstrapActivity(context.Background(), path, int64(len(line)), now); err != nil { t.Fatal(err) } if !c.lastEventAt.IsZero() { @@ -110,7 +110,7 @@ func TestRolloutActivityBootstrapRespectsOrdinalBoundary(t *testing.T) { writeRollout(t, path, line) boundary := uint64(10) c := &rolloutCursor{nonRoot: true, subagentHistoryStartOrdinal: &boundary} - if err := c.bootstrapActivity(path, int64(len(line)), now); err != nil { + if err := c.bootstrapActivity(context.Background(), path, int64(len(line)), now); err != nil { t.Fatal(err) } if !c.lastEventAt.IsZero() { @@ -238,7 +238,7 @@ func TestRolloutActivityIgnoresFutureInheritedAndPartialEvents(t *testing.T) { t.Fatal(err) } c := &rolloutCursor{nonRoot: tc.child, startedAt: now} - if err := c.bootstrapActivity(path, info.Size(), now); err != nil { + if err := c.bootstrapActivity(context.Background(), path, info.Size(), now); err != nil { t.Fatal(err) } if !c.lastEventAt.IsZero() { From f1b5812985225f35b93056821d709a81e906d0da Mon Sep 17 00:00:00 2001 From: merefield Date: Tue, 27 Oct 2026 18:03:16 +0000 Subject: [PATCH 3/3] FIX: bound compressed activity observation to the decoded tail --- internal/codex/live_usage_compressed.go | 7 +- internal/codex/rollout_activity.go | 62 +++++++++++++- .../codex/rollout_activity_compressed_test.go | 84 +++++++++++++++++++ 3 files changed, 148 insertions(+), 5 deletions(-) diff --git a/internal/codex/live_usage_compressed.go b/internal/codex/live_usage_compressed.go index 194ff33..c6112f8 100644 --- a/internal/codex/live_usage_compressed.go +++ b/internal/codex/live_usage_compressed.go @@ -69,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 @@ -82,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:]...) @@ -102,7 +104,6 @@ func compressedRolloutBaseline(ctx context.Context, path string, info os.FileInf cursor.compressedTail = append(cursor.compressedTail[:0], tail...) if !oversized { baselineRolloutRecord(cursor, line) - cursor.observeEvent(line, now) } line = line[:0] oversized = false @@ -119,6 +120,10 @@ func compressedRolloutBaseline(ctx context.Context, path string, info os.FileInf 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 } diff --git a/internal/codex/rollout_activity.go b/internal/codex/rollout_activity.go index f533743..eda29ee 100644 --- a/internal/codex/rollout_activity.go +++ b/internal/codex/rollout_activity.go @@ -14,6 +14,53 @@ import ( // 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 @@ -79,12 +126,19 @@ func (c *rolloutCursor) bootstrapActivity(ctx context.Context, path string, size 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, f, start) + start, partial, err := rolloutActivityRecordStart(ctx, source, start) if err != nil { return err } - reader := bufio.NewReaderSize(io.NewSectionReader(f, start, size-start), 64*1024) + // 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 { @@ -120,7 +174,7 @@ func (c *rolloutCursor) bootstrapActivity(ctx context.Context, path string, size // 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, f *os.File, start int64) (int64, bool, error) { +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) @@ -130,7 +184,7 @@ func rolloutActivityRecordStart(ctx context.Context, f *os.File, start int64) (i } size := min(start-floor, int64(len(buffer))) start -= size - n, err := f.ReadAt(buffer[:size], start) + n, err := source.ReadAt(buffer[:size], start) if err != nil && err != io.EOF { return 0, false, err } diff --git a/internal/codex/rollout_activity_compressed_test.go b/internal/codex/rollout_activity_compressed_test.go index f9b84c7..8977c58 100644 --- a/internal/codex/rollout_activity_compressed_test.go +++ b/internal/codex/rollout_activity_compressed_test.go @@ -1,9 +1,11 @@ package codex import ( + "bytes" "context" "errors" "fmt" + "io" "os" "strings" "testing" @@ -195,3 +197,85 @@ func TestRebasedRolloutPreservesPreviouslyObservedGrowth(t *testing.T) { 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) + } +}