fix: keep active Codex sessions current (#1308)

Long-running and resumed Codex sessions can stay stale while their rollout descriptor remains open, because recursive filesystem notifications may not expose each append. Activity intervals, session freshness, and daily usage then lag until the descriptor closes.

Add a capability-gated live-activity backstop that consumes identity-only Codex history hints and polls only the recently active rollout paths already indexed in SQLite. Poll-wide byte, decoder, retry, path-memory, and hot-set bounds keep recurring work independent of archive cardinality. If a shared budget ends partway through a source, the attempted work still consumes that poll's allowance while its cursor is restored so a later rotation can process the records without loss. A hinted session whose indexed source is briefly stale stays in the bounded lookup-retry window until its canonical path becomes visible.

Configured hint roots are processed through rotating windows of at most 256 sources per poll. Retained cursors are least-recently-used bounded to 256 entries and 2 MiB of source paths. Cursor state is metadata-only: rewrite validation stores a SHA-256 boundary digest and incomplete records store a file offset, with the line reread under existing limits after an append. Raw history or prompt bytes are never retained between polls. Cursor resets may replay older valid hints, so hot activity and retry timestamps only move forward; observed autonomous growth cannot be shortened by replay.

The normal watcher remains the low-latency path and exact-path synchronization remains the sole write path. Cancellation exits at each reader, lookup, stat, and sync boundary so daemon shutdown does not drain the remaining bounded workload. History-disabled installations and frontends without the evidenced TUI history producer retain their existing watcher/degraded-coverage behavior, while restart bootstrap is intentionally bounded to the newest 4 MiB and preceding 24 hours.


Co-authored-by: Wes McKinney <wesm@users.noreply.github.com>

Author: wesm

cmd/agentsview/live_activity.go

@@ -0,0 +1,138 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+
+	"go.kenn.io/agentsview/internal/config"
+	"go.kenn.io/agentsview/internal/db"
+	"go.kenn.io/agentsview/internal/parser"
+	"go.kenn.io/agentsview/internal/server"
+	agentsync "go.kenn.io/agentsview/internal/sync"
+)
+
+func collectLiveActivityTargets(
+	ctx context.Context,
+	cfg config.Config,
+) ([]agentsync.LiveActivityTarget, error) {
+	var targets []agentsync.LiveActivityTarget
+	var targetErrors []error
+	for _, factory := range parser.ProviderFactories() {
+		roots := cfg.ResolveDirs(factory.Definition().Type)
+		if len(roots) == 0 {
+			continue
+		}
+		provider := factory.NewProvider(parser.ProviderConfig{
+			Roots:   roots,
+			Machine: cfg.LocalMachineName,
+		})
+		hints, supported, err := parser.ResolveActivityHintProvider(provider)
+		if err != nil {
+			targetErrors = append(targetErrors, err)
+			continue
+		}
+		if !supported || hints == nil {
+			continue
+		}
+		sources, err := hints.ActivityHintSources(ctx)
+		if err != nil {
+			targetErrors = append(targetErrors, fmt.Errorf(
+				"%s activity hint sources: %w",
+				factory.Definition().Type, err,
+			))
+			continue
+		}
+		if len(sources) == 0 {
+			continue
+		}
+		targets = append(targets, agentsync.LiveActivityTarget{
+			Provider: provider,
+			Hints:    hints,
+			Sources:  sources,
+		})
+	}
+	return targets, errors.Join(targetErrors...)
+}
+
+func newLiveActivityLookup(database *db.DB) agentsync.LiveActivityLookup {
+	return func(
+		ctx context.Context,
+		fullSessionID string,
+	) (agentsync.LiveActivitySource, bool, error) {
+		session, err := database.GetSessionFull(ctx, fullSessionID)
+		if err != nil {
+			return agentsync.LiveActivitySource{}, false, err
+		}
+		if session == nil || session.FilePath == nil || *session.FilePath == "" {
+			return agentsync.LiveActivitySource{}, false, nil
+		}
+		source := agentsync.LiveActivitySource{Path: *session.FilePath}
+		if session.FileSize != nil && session.FileMtime != nil {
+			source.StoredSize = *session.FileSize
+			source.StoredMTimeNS = *session.FileMtime
+			source.HasStoredStat = true
+		}
+		if session.FileInode != nil && session.FileDevice != nil {
+			source.StoredInode = *session.FileInode
+			source.StoredDevice = *session.FileDevice
+			source.HasStoredIdentity = true
+		}
+		return source, true, nil
+	}
+}
+
+func trackLiveActivitySync(
+	idleTracker *server.IdleTracker,
+	syncPaths agentsync.LiveActivitySync,
+) agentsync.LiveActivitySync {
+	return func(ctx context.Context, paths []string) error {
+		done, ok := idleTracker.BeginWork()
+		if !ok {
+			return context.Canceled
+		}
+		defer done()
+		return syncPaths(ctx, paths)
+	}
+}
+
+func startLiveActivityRun(
+	ctx context.Context,
+	cancel context.CancelFunc,
+	poller *agentsync.LiveActivityPoller,
+) func() {
+	var workers sync.WaitGroup
+	workers.Go(func() {
+		poller.Run(ctx)
+	})
+	var once sync.Once
+	return func() {
+		once.Do(func() {
+			cancel()
+			workers.Wait()
+		})
+	}
+}
+
+func startLiveActivityPoller(
+	ctx context.Context,
+	cfg config.Config,
+	database *db.DB,
+	engine *agentsync.Engine,
+	idleTracker *server.IdleTracker,
+) func() {
+	runCtx, cancel := context.WithCancel(ctx)
+	targets, err := collectLiveActivityTargets(runCtx, cfg)
+	if err != nil {
+		log.Printf("live activity target discovery: %v", err)
+	}
+	poller := agentsync.NewLiveActivityPoller(
+		targets,
+		newLiveActivityLookup(database),
+		trackLiveActivitySync(idleTracker, engine.SyncPathsContext),
+		log.Printf,
+	)
+	return startLiveActivityRun(runCtx, cancel, poller)
+}

cmd/agentsview/live_activity_test.go

@@ -0,0 +1,216 @@
+package main
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.kenn.io/agentsview/internal/config"
+	"go.kenn.io/agentsview/internal/db"
+	"go.kenn.io/agentsview/internal/dbtest"
+	"go.kenn.io/agentsview/internal/parser"
+	"go.kenn.io/agentsview/internal/server"
+	agentsync "go.kenn.io/agentsview/internal/sync"
+)
+
+func TestCollectLiveActivityTargetsUsesOnlyConfiguredHintProviders(t *testing.T) {
+	base := t.TempDir()
+	custom := filepath.Join(t.TempDir(), "custom")
+	cfg := config.Config{
+		LocalMachineName: "local",
+		AgentDirs: map[parser.AgentType][]string{
+			parser.AgentCodex: {
+				filepath.Join(base, "sessions"),
+				filepath.Join(base, "archived_sessions"),
+				filepath.Join(custom, "sessions"),
+				"s3://bucket/archive/sessions",
+			},
+			parser.AgentClaude: {filepath.Join(t.TempDir(), "claude")},
+		},
+	}
+
+	targets, err := collectLiveActivityTargets(t.Context(), cfg)
+
+	require.NoError(t, err)
+	require.Len(t, targets, 1)
+	assert.Equal(t, parser.AgentCodex, targets[0].Provider.Definition().Type)
+	assert.Equal(t, []parser.ActivityHintSource{
+		{Path: filepath.Join(base, "history.jsonl")},
+		{Path: filepath.Join(custom, "history.jsonl")},
+	}, targets[0].Sources)
+}
+
+func TestCollectLiveActivityTargetsDoesNotRequireExistingRoots(t *testing.T) {
+	base := t.TempDir()
+	missing := filepath.Join(base, "missing", "sessions")
+	cfg := config.Config{
+		AgentDirs: map[parser.AgentType][]string{
+			parser.AgentCodex: {missing},
+		},
+	}
+
+	targets, err := collectLiveActivityTargets(t.Context(), cfg)
+
+	require.NoError(t, err)
+	require.Len(t, targets, 1)
+	assert.Equal(t, filepath.Join(base, "missing", "history.jsonl"),
+		targets[0].Sources[0].Path)
+	_, statErr := os.Stat(missing)
+	assert.ErrorIs(t, statErr, os.ErrNotExist,
+		"target collection must not create or discover rollout roots")
+}
+
+func TestLiveActivityIndexedLookupReturnsExactStoredMetadata(t *testing.T) {
+	database := dbtest.OpenTestDB(t)
+	path := filepath.Join(t.TempDir(), "rollout.jsonl")
+	size := int64(123)
+	mtime := int64(456)
+	inode := int64(789)
+	device := int64(1011)
+	require.NoError(t, database.UpsertSession(db.Session{
+		ID:         "codex:exact-id",
+		Project:    "project",
+		Machine:    "local",
+		Agent:      string(parser.AgentCodex),
+		FilePath:   &path,
+		FileSize:   &size,
+		FileMtime:  &mtime,
+		FileInode:  &inode,
+		FileDevice: &device,
+	}))
+	lookup := newLiveActivityLookup(database)
+
+	got, found, err := lookup(t.Context(), "codex:exact-id")
+
+	require.NoError(t, err)
+	assert.True(t, found)
+	assert.Equal(t, agentsync.LiveActivitySource{
+		Path:              path,
+		StoredSize:        size,
+		StoredMTimeNS:     mtime,
+		StoredInode:       inode,
+		StoredDevice:      device,
+		HasStoredStat:     true,
+		HasStoredIdentity: true,
+	}, got)
+
+	_, found, err = lookup(t.Context(), "codex:missing-id")
+	require.NoError(t, err)
+	assert.False(t, found)
+}
+
+func TestLiveActivityIndexedLookupSchedulesRowsWithoutCompleteStat(t *testing.T) {
+	database := dbtest.OpenTestDB(t)
+	path := filepath.Join(t.TempDir(), "rollout.jsonl")
+	require.NoError(t, database.UpsertSession(db.Session{
+		ID:       "codex:no-stat",
+		Project:  "project",
+		Machine:  "local",
+		Agent:    string(parser.AgentCodex),
+		FilePath: &path,
+	}))
+
+	got, found, err := newLiveActivityLookup(database)(
+		t.Context(), "codex:no-stat",
+	)
+
+	require.NoError(t, err)
+	assert.True(t, found)
+	assert.Equal(t, path, got.Path)
+	assert.False(t, got.HasStoredStat)
+}
+
+func TestStartLiveActivityRunTracksSyncAndWaitsForStop(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	base := t.TempDir()
+	sessions := filepath.Join(base, "sessions")
+	history := filepath.Join(base, "history.jsonl")
+	rollout := filepath.Join(base, "rollout.jsonl")
+	id := "019f0000-0000-7000-8000-000000000002"
+	now := time.Now()
+	require.NoError(t, os.WriteFile(history, []byte(
+		`{"session_id":"`+id+`","ts":`+
+			strconv.FormatInt(now.Unix(), 10)+
+			`,"text":"private prompt sentinel"}`+"\n",
+	), 0o644))
+	require.NoError(t, os.WriteFile(rollout, []byte("changed"), 0o644))
+	provider, ok := parser.NewProvider(parser.AgentCodex, parser.ProviderConfig{
+		Roots: []string{sessions},
+	})
+	require.True(t, ok)
+	hints, ok, err := parser.ResolveActivityHintProvider(provider)
+	require.NoError(t, err)
+	require.True(t, ok)
+	sources, err := hints.ActivityHintSources(t.Context())
+	require.NoError(t, err)
+
+	idled := make(chan struct{}, 1)
+	idle := server.NewIdleTracker(20*time.Millisecond, func() {
+		idled <- struct{}{}
+	})
+	go idle.Run(ctx)
+
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	finished := make(chan struct{})
+	trackedSync := trackLiveActivitySync(idle,
+		func(context.Context, []string) error {
+			close(entered)
+			<-release
+			close(finished)
+			return nil
+		})
+	runCtx, runCancel := context.WithCancel(ctx)
+	poller := agentsync.NewLiveActivityPoller(
+		[]agentsync.LiveActivityTarget{{
+			Provider: provider,
+			Hints:    hints,
+			Sources:  sources,
+		}},
+		func(context.Context, string) (agentsync.LiveActivitySource, bool, error) {
+			return agentsync.LiveActivitySource{Path: rollout}, true, nil
+		},
+		trackedSync,
+		nil,
+	)
+	stop := startLiveActivityRun(runCtx, runCancel, poller)
+
+	select {
+	case <-entered:
+	case <-time.After(time.Second):
+		require.FailNow(t, "tracked sync did not start")
+	}
+	select {
+	case <-idled:
+		require.FailNow(t, "idle callback fired while sync work was active")
+	case <-time.After(3 * 20 * time.Millisecond):
+	}
+
+	stopped := make(chan struct{})
+	go func() {
+		stop()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+		require.FailNow(t, "stop returned before active sync work completed")
+	case <-time.After(20 * time.Millisecond):
+	}
+	close(release)
+	select {
+	case <-finished:
+	case <-time.After(time.Second):
+		require.FailNow(t, "tracked sync did not finish")
+	}
+	select {
+	case <-stopped:
+	case <-time.After(time.Second):
+		require.FailNow(t, "stop did not join the poller goroutine")
+	}
+}

cmd/agentsview/main.go

@@ -418,6 +418,10 @@ func runServe(cfg config.Config, opts serveOptions) {
 			log.Printf("warning: remote_hosts config invalid, skipping periodic remote sync: %v", err)
 			validRemotes = false
 		}
+		stopLiveActivity := startLiveActivityPoller(
+			ctx, cfg, database, engine, idleTracker,
+		)
+		defer stopLiveActivity()
 		go startPeriodicSync(
 			ctx, cfg, engine, database, writeLock, idleTracker, validRemotes, emitter,
 		)