From 8b8621f98c1077158f22a30a2b6d2efb5aeb610a Mon Sep 17 00:00:00 2001 From: Santosh Date: Sat, 24 Oct 2026 22:52:02 -0400 Subject: [PATCH 1/7] plandb: a read reloads the plan only when the database changed Every Store read (Show, Task, Notes, ...) reloaded the whole plan from SQLite, and each plandb.Open read it twice more. Listing one conversation's tasks with 800 run rows therefore read the plan 800 times over. refresh now records SQLite's data_version and the size, mtime and inode of the database and its log before it reads, and skips the read while all of them stand still. A write by this handle or another still moves the stamp, so the next read sees it. Open takes the stamp before its own load. Measured with a synthetic 200-task plan: 200 Show calls allocated 153 MB before and under 3 MB after (TestRepeatedReadsOfAnUnchangedPlanLoadItOnce). Co-Authored-By: Claude Opus 5.5 --- internal/plandb/delete.go | 2 +- internal/plandb/refresh_scale_test.go | 80 +++++++++++++++++++++++++++ internal/plandb/stamp_other.go | 15 +++++ internal/plandb/stamp_unix.go | 22 ++++++++ internal/plandb/store.go | 65 ++++++++++++++++++++-- 5 files changed, 179 insertions(+), 5 deletions(-) create mode 100644 internal/plandb/refresh_scale_test.go create mode 100644 internal/plandb/stamp_other.go create mode 100644 internal/plandb/stamp_unix.go diff --git a/internal/plandb/delete.go b/internal/plandb/delete.go index cfb0c3d3f7..dae325a26b 100644 --- a/internal/plandb/delete.go +++ b/internal/plandb/delete.go @@ -94,6 +94,6 @@ func (s *Store) DeleteSubtree(id string) ([]string, error) { if err = tx.Commit(); err != nil { return nil, err } - s.data = fresh + s.adopt(fresh) return ids, nil } diff --git a/internal/plandb/refresh_scale_test.go b/internal/plandb/refresh_scale_test.go new file mode 100644 index 0000000000..3ba71f8488 --- /dev/null +++ b/internal/plandb/refresh_scale_test.go @@ -0,0 +1,80 @@ +package plandb + +import ( + "fmt" + "path/filepath" + "runtime" + "testing" +) + +// A READ OF AN UNCHANGED PLAN DOES NOT LOAD IT AGAIN. Every read used to +// reload the whole plan from the database, so listing a run with hundreds of +// tasks one task at a time read the plan hundreds of times over: gigabytes of +// allocation for one list. A read now loads when the database has changed, and +// only then. +func TestRepeatedReadsOfAnUnchangedPlanLoadItOnce(t *testing.T) { + path := filepath.Join(t.TempDir(), "plandb.db") + store, err := Open(path, "program", "1", "root", "the root") + if err != nil { + t.Fatal(err) + } + defer store.Close() + const tasks = 200 + specs := make([]TaskSpec, tasks) + for i := range specs { + specs[i] = TaskSpec{ID: fmt.Sprintf("s%d", i), Title: fmt.Sprintf("task %d", i), Description: "a step with some words in it to give the row a size", ParentID: "1"} + } + added, err := store.AddMany(specs) + if err != nil { + t.Fatal(err) + } + + // One read to settle whatever the first load costs. + if _, err := store.Show(added[0].ID); err != nil { + t.Fatal(err) + } + var before, after runtime.MemStats + runtime.ReadMemStats(&before) + for _, task := range added { + if _, err := store.Show(task.ID); err != nil { + t.Fatal(err) + } + } + runtime.ReadMemStats(&after) + const budget = 3 << 20 + if spent := after.TotalAlloc - before.TotalAlloc; spent > budget { + t.Fatalf("%d reads of an unchanged plan allocated %d bytes, want under %d", tasks, spent, budget) + } +} + +// A WRITE BY ANOTHER HANDLE IS STILL SEEN BY THE NEXT READ, which is the whole +// reason a read refreshes at all. +func TestAReadSeesAnotherHandlesWriteAfterAnUnchangedRead(t *testing.T) { + path := filepath.Join(t.TempDir(), "plandb.db") + reader, err := Open(path, "program", "1", "root", "the root") + if err != nil { + t.Fatal(err) + } + defer reader.Close() + if _, err := reader.Show("1"); err != nil { + t.Fatal(err) + } + writer, err := Open(path, "program", "1", "root", "the root") + if err != nil { + t.Fatal(err) + } + defer writer.Close() + added, err := writer.AddMany([]TaskSpec{{ID: "later", Title: "later", Description: "added by the other handle", ParentID: "1"}}) + if err != nil { + t.Fatal(err) + } + if _, err := reader.Show(added[0].ID); err != nil { + t.Fatalf("a task another handle committed was not seen: %v", err) + } + if _, err := writer.AddNote(added[0].ID, "agent", "a note"); err != nil { + t.Fatal(err) + } + if notes := reader.Notes(added[0].ID, 10); len(notes) != 1 { + t.Fatalf("reader saw %d notes after another handle's note, want 1", len(notes)) + } +} diff --git a/internal/plandb/stamp_other.go b/internal/plandb/stamp_other.go new file mode 100644 index 0000000000..784b7305a9 --- /dev/null +++ b/internal/plandb/stamp_other.go @@ -0,0 +1,15 @@ +//go:build !unix + +package plandb + +import "os" + +// statStamp is the size and modification time of path, or the zero stamp for a +// file that is not there, which is a stamp too. +func statStamp(path string) fileStamp { + info, err := os.Stat(path) + if err != nil { + return fileStamp{} + } + return fileStamp{size: info.Size(), mtime: info.ModTime().UnixNano()} +} diff --git a/internal/plandb/stamp_unix.go b/internal/plandb/stamp_unix.go new file mode 100644 index 0000000000..a896e6540b --- /dev/null +++ b/internal/plandb/stamp_unix.go @@ -0,0 +1,22 @@ +//go:build unix + +package plandb + +import ( + "os" + "syscall" +) + +// statStamp is the size, modification time and inode of path, or the zero +// stamp for a file that is not there, which is a stamp too. +func statStamp(path string) fileStamp { + info, err := os.Stat(path) + if err != nil { + return fileStamp{} + } + stamp := fileStamp{size: info.Size(), mtime: info.ModTime().UnixNano()} + if raw, ok := info.Sys().(*syscall.Stat_t); ok { + stamp.inode = uint64(raw.Ino) + } + return stamp +} diff --git a/internal/plandb/store.go b/internal/plandb/store.go index b42ca36b15..a288a217f8 100644 --- a/internal/plandb/store.go +++ b/internal/plandb/store.go @@ -48,6 +48,9 @@ type Store struct { path string now func() time.Time data state + // seen is what the database looked like when data was last read from it, so + // a read of a plan nobody has written does not read it again (refresh). + seen readStamp // closed is set by Close under the lock. Every method asks it there before // it touches a handle Close may already have released — the nil `db` and // `rdb` below, which used to be what a late write found. @@ -109,6 +112,9 @@ func Open(path, project, rootID, rootTitle, rootDescription string, chat ...stri return nil, err } store.rdb = rdb + // The stamp is taken before the plan is read, so a store that had to seed + // its root (a write the stamp does not know) is read once more, never less. + seen := store.currentStamp() loaded, err := store.loadOrCreate(project, rootID, rootTitle, rootDescription, tag) if err != nil { _ = rdb.Close() @@ -116,6 +122,7 @@ func Open(path, project, rootID, rootTitle, rootDescription string, chat ...stri return nil, err } store.data = loaded + store.seen = seen return store, nil } @@ -1820,7 +1827,7 @@ func (s *Store) Archive(olderThan time.Duration) ([]*Task, error) { now := s.now().UTC() doomed, moved := archiveSelection(fresh, now.Add(-olderThan)) if len(moved) == 0 { - s.data = fresh + s.adopt(fresh) return nil, nil } for _, task := range moved { @@ -1842,7 +1849,7 @@ func (s *Store) Archive(olderThan time.Duration) ([]*Task, error) { if err := tx.Commit(); err != nil { return nil, err } - s.data = fresh + s.adopt(fresh) return moved, nil } @@ -2314,7 +2321,7 @@ func (s *Store) transact(change func(*state, time.Time) error) error { if err != nil { return err } - s.data = fresh + s.adopt(fresh) next := cloneState(fresh) if err := change(&next, s.now().UTC()); err != nil { if errors.Is(err, errNoChange) { @@ -2333,7 +2340,7 @@ func (s *Store) transact(change func(*state, time.Time) error) error { // error; this one wears [ErrCommitUncertain] and the original cause. return fmt.Errorf("%w: %w", ErrCommitUncertain, err) } - s.data = next + s.adopt(next) return nil } @@ -2378,11 +2385,61 @@ func (s *Store) refresh() { return } defer tx.Rollback() + // THE STAMP IS READ BEFORE THE PLAN, so a commit between the two leaves a + // stamp older than the data and costs one more read, never a stale answer. + stamp := s.stampOf(tx) + if stamp.known && stamp == s.seen { + return + } fresh, err := loadState(tx) if err != nil { return } s.data = fresh + s.seen = stamp +} + +// readStamp is what says the database has not changed since a read: SQLite's +// own counter of commits by other connections, and the size and modification +// time of the database and its log. Either moving is a change; both standing +// still across a write would need the counter and two clocks to agree on it. +type readStamp struct { + known bool + version int64 + db, wal fileStamp +} + +type fileStamp struct { + size int64 + mtime int64 + inode uint64 +} + +// currentStamp is the stamp of the database as it stands. +func (s *Store) currentStamp() readStamp { + tx, err := s.rdb.Begin() + if err != nil { + return readStamp{} + } + defer tx.Rollback() + return s.stampOf(tx) +} + +func (s *Store) stampOf(tx *sql.Tx) readStamp { + var stamp readStamp + if err := tx.QueryRow("PRAGMA data_version").Scan(&stamp.version); err != nil { + return readStamp{} + } + stamp.db, stamp.wal = statStamp(s.path), statStamp(s.path+"-wal") + stamp.known = true + return stamp +} + +// adopt takes a plan this handle has just written as what it holds. The next +// read looks at the database again, whatever it last saw. +func (s *Store) adopt(data state) { + s.data = data + s.seen = readStamp{} } // Close releases the store's database handles and marks the handle closed From c84ea7d5dea2d3aab7249ab5af34c3beb1e87d4e Mon Sep 17 00:00:00 2001 From: Santosh Date: Sat, 24 Oct 2026 22:52:10 -0400 Subject: [PATCH 2/7] session: listing a conversation's tasks reads each file once Three costs that grew with the size of one conversation's history: - AppTasks opened the plan store once per run row, and every open read the whole plan. It now opens the read handles once for the listing. 150 run rows over a 150-task plan: 235 MB allocated before, under 16 MB after (TestAppTasksOpensThePlanOnceForAllTheRunRows). - markPlanInterrupted copied every run row for every plan row. It now walks the rows where they are kept; the listing is no longer quadratic. - Every conversation in a project decoded the project's whole task index. The decoded rows of the last four indexes (up to 4 MiB of file each) are kept under the file's stamp and shared; lastPerNode builds the slice each caller then orders. 40 readings of a 1,500-row index: 180 MB before, under 64 MB after (TestAnUnchangedTaskIndexIsDecodedOnce). The journal record cache also forgets what nobody reads: a kept record not touched for 30 s is dropped (RecordCache.Expire, armed by a timer while anything is kept), and it keeps at most 8 records instead of 32. Paging a chat still hits within seconds; a hundred journals viewed in a morning are no longer held until the app quits. Co-Authored-By: Claude Opus 5.5 --- internal/session/plandb_tasks.go | 33 +++-- internal/session/recordcache.go | 45 ++++++- internal/session/recordcache_test.go | 38 ++++++ internal/session/task_app.go | 14 +- internal/session/task_app_runs.go | 19 ++- internal/session/task_app_runs_scale_test.go | 62 +++++++++ internal/session/task_delete.go | 3 +- internal/session/task_index.go | 127 +++++++++++++++---- internal/session/task_index_cache_test.go | 61 +++++++++ 9 files changed, 352 insertions(+), 50 deletions(-) create mode 100644 internal/session/task_app_runs_scale_test.go create mode 100644 internal/session/task_index_cache_test.go diff --git a/internal/session/plandb_tasks.go b/internal/session/plandb_tasks.go index 1a16ed36bf..11ab452709 100644 --- a/internal/session/plandb_tasks.go +++ b/internal/session/plandb_tasks.go @@ -517,22 +517,27 @@ func (a *Agent) markPlanInterrupted(row *PlanTaskRow) { } g.mu.Lock() defer g.mu.Unlock() - for _, notice := range g.runRowsLocked() { - if notice.PlanTask != row.ID { - continue - } - if notice.State == TaskInterrupted { - row.Interrupted = true - } - if notice.RecoveredCompletion { - row.Status = string(notice.State) - row.Ended = notice.EndedAt - if notice.CostUSD > row.USD { - row.USD = notice.CostUSD + // The rows are walked where they are kept: copying them all for every plan + // row made one listing of a conversation's tasks quadratic in its runs. + for _, root := range g.runRuns { + for i := range g.runs[root] { + notice := &g.runs[root][i] + if g.deleted[notice.ID] || g.deleted[notice.Parent] || notice.PlanTask != row.ID { + continue } - row.Note = notice.Report + if notice.State == TaskInterrupted { + row.Interrupted = true + } + if notice.RecoveredCompletion { + row.Status = string(notice.State) + row.Ended = notice.EndedAt + if notice.CostUSD > row.USD { + row.USD = notice.CostUSD + } + row.Note = notice.Report + } + return } - return } } diff --git a/internal/session/recordcache.go b/internal/session/recordcache.go index 345283141e..9cdeb0346a 100644 --- a/internal/session/recordcache.go +++ b/internal/session/recordcache.go @@ -117,6 +117,8 @@ type RecordCacheLimits struct { // Settle is how old a file's change time must be before its reading is // kept. Settle time.Duration + // Idle is how long a kept record may go unread before it is dropped. + Idle time.Duration // Now is the clock the change time is measured against. Now func() time.Time } @@ -126,9 +128,12 @@ type RecordCacheLimits struct { const ( recordCacheOperations = 2 recordCacheBytes = 256 << 20 - recordCacheEntries = 32 + recordCacheEntries = 8 recordCacheFileBytes = 16 << 20 recordCacheSettle = 2 * time.Second + // A page asked for again comes within seconds; a journal nobody has read + // for this long is read again for the price of one parse. + recordCacheIdle = 30 * time.Second ) // RecordCacheStats is what a cache holds and has done, for a test or a bench. @@ -155,6 +160,8 @@ type RecordCache struct { retained int64 reserved int64 stats RecordCacheStats + // sweeping is true while a timer is armed to look for idle records. + sweeping bool } // recordSlot is one kept reading. A slot evicted while copies of it are out is @@ -166,6 +173,8 @@ type recordSlot struct { cost int64 leases int dropped bool + // used is when the record was kept or last handed out. + used time.Time } // recordFlight is one parse that may be kept, and whether a reader waits on it. @@ -192,6 +201,9 @@ func NewRecordCache(limits RecordCacheLimits) *RecordCache { if limits.Settle <= 0 { limits.Settle = recordCacheSettle } + if limits.Idle <= 0 { + limits.Idle = recordCacheIdle + } if limits.Now == nil { limits.Now = time.Now } @@ -296,6 +308,7 @@ func (c *RecordCache) leaseLocked(element *list.Element) (Record, func(), error) return Record{}, nil, ErrRecordReadBusy } c.order.MoveToFront(element) + slot.used = c.limits.Now() c.stats.Hits++ c.mu.Unlock() record := cloneRecord(slot.record) @@ -347,9 +360,10 @@ func (c *RecordCache) parseLocked(path string, size int64, stamp recordStamp, st } if c.reserveLocked(cost + vectors) { c.reserved -= cost - slot := &recordSlot{path: path, stamp: stamp, record: record, cost: cost, leases: 1} + slot := &recordSlot{path: path, stamp: stamp, record: record, cost: cost, leases: 1, used: c.limits.Now()} c.kept[path] = c.order.PushFront(slot) c.retained += cost + c.watchIdleLocked() copied := cloneRecord(record) return copied, func() { c.mu.Lock() @@ -393,6 +407,33 @@ func (c *RecordCache) reserveLocked(n int64) bool { return true } +// Expire drops the kept records nobody has read for the idle time. A record a +// callback still holds stays until the next look. It runs by itself while +// anything is kept; calling it is only for a test. +func (c *RecordCache) Expire() { + c.mu.Lock() + defer c.mu.Unlock() + c.sweeping = false + now := c.limits.Now() + for element := c.order.Back(); element != nil; { + prev := element.Prev() + if slot := element.Value.(*recordSlot); slot.leases == 0 && now.Sub(slot.used) >= c.limits.Idle { + c.dropLocked(element) + } + element = prev + } + c.watchIdleLocked() +} + +// watchIdleLocked arms the next look for idle records while any are kept. +func (c *RecordCache) watchIdleLocked() { + if c.sweeping || c.order.Len() == 0 { + return + } + c.sweeping = true + time.AfterFunc(c.limits.Idle, c.Expire) +} + // dropLocked forgets one kept reading; its bytes stay counted while a copy of // it is out. func (c *RecordCache) dropLocked(element *list.Element) { diff --git a/internal/session/recordcache_test.go b/internal/session/recordcache_test.go index 84370bf288..91a208bab0 100644 --- a/internal/session/recordcache_test.go +++ b/internal/session/recordcache_test.go @@ -663,3 +663,41 @@ func BenchmarkRecordCache(b *testing.B) { }) } } + +// A KEPT RECORD THAT NOBODY ASKS FOR GOES BACK TO THE HEAP. Paging a long +// conversation wants the next page from the same parse, a few seconds later; +// nothing wants a journal's whole reading an hour after the last page, and +// eight of them held for the life of a desktop engine were most of its +// resident set. Reading it again later is one parse. +func TestAKeptRecordNobodyAsksForIsDroppedWhenItIdles(t *testing.T) { + clock := time.Now().Add(time.Hour) + dir := t.TempDir() + cache := NewRecordCache(RecordCacheLimits{Settle: time.Nanosecond, Now: func() time.Time { return clock }}) + var paths []string + for _, name := range []string{"a.jsonl", "b.jsonl"} { + path := filepath.Join(dir, name) + writeRecord(t, path, recordJournal(6, false)) + paths = append(paths, path) + read(t, cache, path) + } + if s := cache.Stats(); s.Entries != 2 || s.Retained == 0 { + t.Fatalf("stats = %+v, want both journals kept", s) + } + clock = clock.Add(recordCacheIdle / 2) + read(t, cache, paths[0]) // a hit: this one is in use again + clock = clock.Add(recordCacheIdle/2 + time.Second) + cache.Expire() + s := cache.Stats() + if s.Entries != 1 { + t.Fatalf("after the idle time %d records are kept, want only the one read since", s.Entries) + } + clock = clock.Add(recordCacheIdle) + cache.Expire() + if s := cache.Stats(); s.Entries != 0 || s.Retained != 0 { + t.Fatalf("stats = %+v, want an empty cache once nothing was read for the idle time", s) + } + read(t, cache, paths[1]) + if s := cache.Stats(); s.Parses != 3 { + t.Fatalf("parses = %d, want two first readings and the expired journal read afresh", s.Parses) + } +} diff --git a/internal/session/task_app.go b/internal/session/task_app.go index 9803fee4f0..250ad97cd4 100644 --- a/internal/session/task_app.go +++ b/internal/session/task_app.go @@ -13,6 +13,7 @@ import ( "github.com/Agent-Field/codeaf/internal/config" "github.com/Agent-Field/codeaf/internal/crewroute" + "github.com/Agent-Field/codeaf/internal/plandb" ) // AppTaskView is an engine read, independent of the app's JSON contract. @@ -263,14 +264,23 @@ func (a *Agent) AppTasks() ([]AppTaskView, error) { graphIDs := map[string]bool{} out = a.appGraphTasks(out, cap, graphIDs) // A HANDED-OVER TASK is a run's row, not a node (task_app_runs.go). - for _, row := range a.appRunRows() { + // The plan is opened once for every run row and closed before the plan's + // own rows below open it again: the handles hold the plan's gate. + rows := a.appRunRows() + var stores []*plandb.Store + closeStores := func() {} + if len(rows) > 0 { + stores, _, closeStores = a.openPlanReadHandles() + } + for _, row := range rows { id := strconv.FormatUint(row.ID, 10) if graphIDs[id] { continue } graphIDs[id] = true - out = append(out, a.appRunView(row, cap)) + out = append(out, a.appRunViewIn(row, cap, stores)) } + closeStores() for _, row := range a.PlanTasks() { if graphIDs[row.ID] { continue diff --git a/internal/session/task_app_runs.go b/internal/session/task_app_runs.go index f189e112e9..70e40ccd95 100644 --- a/internal/session/task_app_runs.go +++ b/internal/session/task_app_runs.go @@ -98,8 +98,16 @@ func (a *Agent) appRunRow(id uint64) (TaskNotice, bool) { return graph.runRowLocked(id) } -// appRunView is one run row as the app's task. +// appRunView is one run row as the app's task, opening the plan for itself. func (a *Agent) appRunView(row TaskNotice, cap float64) AppTaskView { + stores, _, closeStores := a.openPlanReadHandles() + defer closeStores() + return a.appRunViewIn(row, cap, stores) +} + +// appRunViewIn is one run row as the app's task, read from stores a listing +// has already opened. +func (a *Agent) appRunViewIn(row TaskNotice, cap float64, stores []*plandb.Store) AppTaskView { if _, held := a.appHeldRuns.Load(row.ID); held && !row.State.settled() { row.Paused = true } @@ -117,7 +125,7 @@ func (a *Agent) appRunView(row TaskNotice, cap float64) AppTaskView { for _, id := range row.DependsOn { view.Waits = append(view.Waits, strconv.FormatUint(id, 10)) } - a.appRunDetail(&view, appRunRowKey(row)) + a.appRunDetail(&view, appRunRowKey(row), stores) switch { case row.park != nil || a.taskRouting.isParked(row.ID): view.RoutingState = TaskRoutingBlocked @@ -135,9 +143,10 @@ func (a *Agent) appRunView(row TaskNotice, cap float64) AppTaskView { // appRunDetail fills what the row does not carry from the run's store: the // person's notes with the step each was read at, the worker's steps, and the // step count. -func (a *Agent) appRunDetail(view *AppTaskView, key string) { - stores, _, closeStores := a.openPlanReadHandles() - defer closeStores() +// +// The stores are the plan's read handles, opened once for the whole listing: +// opening one reads the whole plan, and a listing draws a row for every run. +func (a *Agent) appRunDetail(view *AppTaskView, key string, stores []*plandb.Store) { for _, store := range stores { task := store.Task(key) if task == nil { diff --git a/internal/session/task_app_runs_scale_test.go b/internal/session/task_app_runs_scale_test.go new file mode 100644 index 0000000000..d7cad87463 --- /dev/null +++ b/internal/session/task_app_runs_scale_test.go @@ -0,0 +1,62 @@ +package session + +import ( + "fmt" + "path/filepath" + "runtime" + "testing" + + "github.com/Agent-Field/codeaf/internal/plandb" +) + +// A LIST OF A CONVERSATION'S RUNS OPENS ITS PLAN ONCE. Each run row used to +// open the plan store for itself, and every open read the whole plan, so a +// conversation with a hundred and fifty rows read a hundred and fifty plans +// to draw one list — gigabytes of allocation, seconds of work, for every +// listing of one agent's tasks. +func TestAppTasksOpensThePlanOnceForAllTheRunRows(t *testing.T) { + const runs = 150 + dir := t.TempDir() + path := filepath.Join(dir, planStoreFilename) + specs := make([]plandb.TaskSpec, runs) + for i := range specs { + specs[i] = plandb.TaskSpec{ID: fmt.Sprintf("r%d", i), Title: fmt.Sprintf("Run step %d", i), Description: "a step with some words in it to give the row a size"} + } + seedPlanStore(t, path, "chat-a", specs...) + for i := range specs { + writePlanTrajectory(t, dir, specs[i].ID, `{"kind":"step","step":1,"command":"$ echo one","observation":"one"}`) + } + + agent, _ := newTestAgent(t, &scriptedCompleter{}, nil) + armPlanStore(t, agent, path, "chat-a") + graph := agent.graph() + graph.mu.Lock() + graph.runs = map[uint64][]TaskNotice{} + for i := range specs { + id := uint64(1000 + i) + graph.runs[id] = []TaskNotice{{ID: id, Title: specs[i].Title, State: TaskDone, PlanTask: specs[i].ID}} + graph.runRuns = append(graph.runRuns, id) + } + graph.mu.Unlock() + + var before, after runtime.MemStats + runtime.ReadMemStats(&before) + views, err := agent.AppTasks() + runtime.ReadMemStats(&after) + if err != nil { + t.Fatal(err) + } + withSteps := 0 + for _, view := range views { + if view.StepsKnown { + withSteps++ + } + } + if withSteps != runs { + t.Fatalf("%d of %d run rows carried their steps", withSteps, runs) + } + const budget = 16 << 20 + if spent := after.TotalAlloc - before.TotalAlloc; spent > budget { + t.Fatalf("listing %d run rows allocated %d bytes, want under %d", runs, spent, budget) + } +} diff --git a/internal/session/task_delete.go b/internal/session/task_delete.go index 9009e31f00..5660cf74a8 100644 --- a/internal/session/task_delete.go +++ b/internal/session/task_delete.go @@ -436,7 +436,8 @@ func filterDeletedTaskIndex(path string, rows []TaskIndexEntry) []TaskIndexEntry } deletions[row.SessionID] = ids } - kept := rows[:0:0] + // The caller owns rows (lastPerNode built them), so they are thinned in place. + kept := rows[:0] for _, row := range rows { if !gone[row.SessionID] && !deletions[row.SessionID][row.ID] { kept = append(kept, row) diff --git a/internal/session/task_index.go b/internal/session/task_index.go index f1459f4ab5..8d3a88ac24 100644 --- a/internal/session/task_index.go +++ b/internal/session/task_index.go @@ -60,9 +60,11 @@ package session import ( "bufio" + "bytes" "encoding/json" "os" "path/filepath" + "slices" "sort" "strconv" "strings" @@ -501,33 +503,10 @@ func readTaskIndexRows(path string, limit int, filter bool) []TaskIndexEntry { if strings.TrimSpace(path) == "" { return nil } - file, err := os.Open(path) - if err != nil { + rows, ok := taskIndexParsed.read(path) + if !ok { return nil } - defer file.Close() - - var rows []TaskIndexEntry - scanner := bufio.NewScanner(file) - // A row is small by construction, but an outcome plus two URIs plus a title - // can pass the scanner's default line budget on a pathological title. - scanner.Buffer(make([]byte, 0, 4<<10), 256<<10) - for scanner.Scan() { - line := strings.TrimSpace(scanner.Text()) - if line == "" { - continue - } - var entry TaskIndexEntry - if err := json.Unmarshal([]byte(line), &entry); err != nil { - continue - } - if strings.TrimSpace(entry.Title) == "" { - continue - } - rows = append(rows, entry) - } - // scanner.Err() is deliberately unread: a truncated tail is the same - // tolerated case as an unparseable line, and the rows before it are good. rows = lastPerNode(rows) if filter { rows = filterDeletedTaskIndex(path, rows) @@ -562,7 +541,7 @@ func lastPerNode(rows []TaskIndexEntry) []TaskIndexEntry { } last[row.SessionID+"\x00"+id] = index } - kept := rows[:0] + kept := make([]TaskIndexEntry, 0, len(rows)) for index, row := range rows { id := strings.TrimSpace(row.ID) if id == "" { @@ -1317,3 +1296,99 @@ func LookupTask(rows []TaskIndexEntry, token string) (TaskIndexEntry, bool) { } return TaskIndexEntry{}, false } + +// parseTaskIndex decodes every row of the index at path in file order, and +// whether the file could be opened at all. +func parseTaskIndex(path string) ([]TaskIndexEntry, bool) { + file, err := os.Open(path) + if err != nil { + return nil, false + } + defer file.Close() + + var rows []TaskIndexEntry + scanner := bufio.NewScanner(file) + // A row is small by construction, but an outcome plus two URIs plus a title + // can pass the scanner's default line budget on a pathological title. + scanner.Buffer(make([]byte, 0, 4<<10), 256<<10) + for scanner.Scan() { + line := bytes.TrimSpace(scanner.Bytes()) + if len(line) == 0 { + continue + } + var entry TaskIndexEntry + if err := json.Unmarshal(line, &entry); err != nil { + continue + } + if strings.TrimSpace(entry.Title) == "" { + continue + } + rows = append(rows, entry) + } + // scanner.Err() is deliberately unread: a truncated tail is the same + // tolerated case as an unparseable line, and the rows before it are good. + return rows, true +} + +// A PROJECT'S INDEX IS READ BY EVERY CONVERSATION IN IT, and a view of a +// hundred conversations used to decode the same file a hundred times. The +// decoded rows of the last few indexes are kept for as long as the file has +// the stamp they were read under; an append moves its size and time, and the +// next reading decodes it again. The kept rows are shared and never written: +// [lastPerNode] builds the slice every caller then orders and thins. +const ( + taskIndexKeptPaths = 4 + // taskIndexKeptBytes is the largest file whose decoding is kept, the + // decoded rows being several times the file. + taskIndexKeptBytes = 4 << 20 +) + +type taskIndexKept struct { + path string + stamp recordStamp + rows []TaskIndexEntry +} + +type taskIndexCache struct { + mu sync.Mutex + kept []taskIndexKept // most recently read last +} + +var taskIndexParsed taskIndexCache + +func (c *taskIndexCache) read(path string) ([]TaskIndexEntry, bool) { + info, err := os.Stat(path) + if err != nil || !info.Mode().IsRegular() { + return parseTaskIndex(path) + } + stamp, stamped := recordStampOf(info) + if !stamped || info.Size() > taskIndexKeptBytes { + return parseTaskIndex(path) + } + c.mu.Lock() + for i, entry := range c.kept { + if entry.path == path && entry.stamp == stamp { + c.kept = append(append(c.kept[:i:i], c.kept[i+1:]...), entry) + c.mu.Unlock() + return entry.rows, true + } + } + c.mu.Unlock() + rows, ok := parseTaskIndex(path) + if !ok { + return nil, false + } + // A FILE THAT CHANGED WHILE IT WAS READ IS NOT KEPT: the stamp taken + // before the reading must still be the file's after it. + if after, err := os.Stat(path); err != nil || !sameStamp(after, stamp) { + return rows, true + } + c.mu.Lock() + defer c.mu.Unlock() + c.kept = slices.DeleteFunc(c.kept, func(entry taskIndexKept) bool { return entry.path == path }) + c.kept = append(c.kept, taskIndexKept{path: path, stamp: stamp, rows: rows}) + if len(c.kept) > taskIndexKeptPaths { + c.kept = c.kept[len(c.kept)-taskIndexKeptPaths:] + } + return rows, true +} diff --git a/internal/session/task_index_cache_test.go b/internal/session/task_index_cache_test.go new file mode 100644 index 0000000000..36fbe0d028 --- /dev/null +++ b/internal/session/task_index_cache_test.go @@ -0,0 +1,61 @@ +package session + +import ( + "fmt" + "os" + "path/filepath" + "reflect" + "runtime" + "testing" + "time" +) + +func writeIndexRows(t *testing.T, path string, from, to int) { + t.Helper() + file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) + if err != nil { + t.Fatal(err) + } + defer file.Close() + for i := from; i < to; i++ { + row := fmt.Sprintf(`{"id":"%d","sessionId":"s%d","name":"task-%d","label":"task %d","title":"task number %d, groomed","status":"done","endedAt":%q}`+"\n", + i, i%40, i, i, i, time.Unix(1_700_000_000+int64(i), 0).UTC().Format(time.RFC3339)) + if _, err := file.WriteString(row); err != nil { + t.Fatal(err) + } + } +} + +// THE PROJECT'S INDEX IS SHARED BY EVERY CONVERSATION IN IT, so listing the +// tasks of a project's hundred conversations read and decoded the same file a +// hundred times. An index nobody has appended to since it was read is answered +// from that reading; a row appended after it is read. +func TestAnUnchangedTaskIndexIsDecodedOnce(t *testing.T) { + path := filepath.Join(t.TempDir(), taskIndexName) + writeIndexRows(t, path, 0, 1500) + want := ReadTaskIndex(path) + if len(want) != 1500 { + t.Fatalf("read %d rows, want 1500", len(want)) + } + + var before, after runtime.MemStats + runtime.ReadMemStats(&before) + for range 40 { + // Whoever gets a reading may reorder it; the next must not see that. + got := ReadTaskIndex(path) + got[0], got[len(got)-1] = got[len(got)-1], got[0] + } + runtime.ReadMemStats(&after) + const budget = 64 << 20 + if spent := after.TotalAlloc - before.TotalAlloc; spent > budget { + t.Fatalf("40 readings of an unchanged index allocated %d bytes, want under %d", spent, budget) + } + if got := ReadTaskIndex(path); !reflect.DeepEqual(got, want) { + t.Fatal("a reading changed what the next one saw") + } + + writeIndexRows(t, path, 1500, 1501) + if got := ReadTaskIndex(path); len(got) != 1501 { + t.Fatalf("after an append the index has %d rows, want 1501", len(got)) + } +} From 6b0976f54650119a0447c60f9422c2a23bce5608 Mon Sep 17 00:00:00 2001 From: Santosh Date: Sun, 25 Oct 2026 08:55:51 -0400 Subject: [PATCH 3/7] engine: settle task indexes and schedule the earliest journal expiry Port only the engine cache corrections from 531e2f1dc and c404aea7e: clear evicted index references, share the mtime/ctime settling guard, and target the earliest unleased journal deadline. Keep their deterministic regressions and log allocation fixture measurements. The project index already uses replacement-sensitive record stamps. No changes from 799d0baa2 are needed here. Document this limited subset and leave idle member release, app task-list disk caching, persistence write skipping and GC policy for follow-up work. Co-Authored-By: Claude Opus 5.5 --- .../unreleased/1933-engine-memory-safe.md | 20 ++++++ docs/perf/engine-memory-safe.md | 49 +++++++++++++++ internal/filememo/settled.go | 12 ++++ internal/filememo/settled_test.go | 28 +++++++++ internal/plandb/refresh_scale_test.go | 1 + internal/session/recordcache.go | 63 ++++++++++++++++--- internal/session/recordcache_test.go | 31 +++++++++ internal/session/task_app_runs_scale_test.go | 1 + internal/session/task_index.go | 12 +++- internal/session/task_index_cache_test.go | 60 ++++++++++++++++++ 10 files changed, 265 insertions(+), 12 deletions(-) create mode 100644 docs/changes/unreleased/1933-engine-memory-safe.md create mode 100644 docs/perf/engine-memory-safe.md create mode 100644 internal/filememo/settled.go create mode 100644 internal/filememo/settled_test.go diff --git a/docs/changes/unreleased/1933-engine-memory-safe.md b/docs/changes/unreleased/1933-engine-memory-safe.md new file mode 100644 index 0000000000..9965ba23a4 --- /dev/null +++ b/docs/changes/unreleased/1933-engine-memory-safe.md @@ -0,0 +1,20 @@ +--- +kind: internal +title: Reuse unchanged engine plan and task-index reads and expire idle journal records +pr: 1933 +surface: [engine, docs] +invalidates: + - Unchanged plan reads previously reloaded the whole plan; matching database stamps now reuse it while committed writes still refresh the next read. + - Journal caches previously kept up to 32 records without idle expiry; they now keep at most eight and expire unleased records after 30 seconds without a read. +--- + +Conversation task listings share plan handles across run rows and avoid copying +the run history per plan task. Project task indexes reuse decoded rows for up to +four files of at most 4 MiB each, with replacement-sensitive stamps and a shared +mtime/ctime settling guard. Eviction clears retained rows; journal expiry targets +the earliest unleased deadline. See `docs/perf/engine-memory-safe.md` for gates. + +Only the low-risk engine subset of the original memory work ships here. Idle +member release and the app task-list disk cache are follow-up work. Member +ownership, reader refcounts, persistence writes and GC policy retain their +base behavior. diff --git a/docs/perf/engine-memory-safe.md b/docs/perf/engine-memory-safe.md new file mode 100644 index 0000000000..301edada2a --- /dev/null +++ b/docs/perf/engine-memory-safe.md @@ -0,0 +1,49 @@ +# Engine memory reuse + +## Plan, task index and journal reads + +Unchanged plan reads reuse the loaded plan while SQLite's data version and the +database/WAL file stamps match. A write through this or another handle still +refreshes the next read. Conversation task listings share plan read handles +across their run rows and inspect interrupted run rows without copying the +whole run history for each plan task. + +Project task indexes reuse decoded rows for matching device, inode, size, mtime +and ctime stamps. Admission requires mtime and ctime to be at least two seconds +old; unsupported file identities bypass reuse. The cache keeps four paths, each +at most 4 MiB on disk, and clears evicted references. Callers get separate row +slices for ordering and deletion filtering; the cached rows remain immutable. + +Journal record caches now keep at most eight records, down from 32. Unleased +records expire 30 seconds after their last read, with the timer targeting the +earliest deadline. A callback's active record remains accounted for until it +returns. Journal and task-index admission share the timestamp settling guard. + +Idle member release and the app task-list disk cache remain follow-up work. +This branch changes neither member ownership nor persistence writes or GC policy. + +## Reproduce the allocation and correctness gates + +On Go 1.26.5, linux/arm64, the included fixtures gate 200 unchanged plan reads +below 3 MiB allocated, a listing of 150 runs below 16 MiB, and 40 readings of a +1,500-row index below 64 MiB. These are allocation budgets, not process RSS or +elapsed-time claims. Compare the same tests at the base and candidate revisions. + +With these fixtures and ordinary `-count=1` tests, a source overlay of the +`a71555ec0` base versus this branch measured: + +| Fixture | Base bytes allocated | Candidate bytes allocated | +| --- | ---: | ---: | +| 200 unchanged plan reads | 152,977,728 | 565,272 | +| Listing 150 runs | 503,060,592 | 14,893,176 | +| 40 readings of a 1,500-row index | 267,895,312 | 33,463,056 | + +The overlay substitutes base production files while keeping the allocation +fixtures; its index fixture predates the injected cache clock. Test logs report +allocations so the candidate measurement can be repeated directly: + +```sh +make test-focus PKGS=./internal/plandb RUN='^Test(RepeatedReadsOfAnUnchangedPlanLoadItOnce|AReadSeesAnotherHandlesWriteAfterAnUnchangedRead)$$' TEST_FLAGS='-race -count=3' +make test-focus PKGS=./internal/session RUN='^Test(AppTasksOpensThePlanOnceForAllTheRunRows|AnUnchangedTaskIndexIsDecodedOnce|TaskIndexEvictionClearsRetainedRows|TaskIndexRejectsUnsettledChangeTime|AKeptRecordNobodyAsksForIsDroppedWhenItIdles|RecordExpirySchedulesTheEarliestRemainingDeadline)$$' TEST_FLAGS='-race -count=3' +make test-focus PKGS=./internal/filememo RUN='^TestTimestampsSettled$$' TEST_FLAGS='-race -count=3' +``` diff --git a/internal/filememo/settled.go b/internal/filememo/settled.go new file mode 100644 index 0000000000..da80c1916d --- /dev/null +++ b/internal/filememo/settled.go @@ -0,0 +1,12 @@ +package filememo + +import "time" + +// TimestampSettle covers coarse filesystem timestamp ticks before caching. +const TimestampSettle = 2 * time.Second + +// TimestampsSettled rejects recent and future writes, including a writer that +// restored modification time while change time remains inside its coarse tick. +func TimestampsSettled(now, modified, changed time.Time, settle time.Duration) bool { + return now.Sub(modified) >= settle && now.Sub(changed) >= settle +} diff --git a/internal/filememo/settled_test.go b/internal/filememo/settled_test.go new file mode 100644 index 0000000000..87d9d4e5e7 --- /dev/null +++ b/internal/filememo/settled_test.go @@ -0,0 +1,28 @@ +package filememo + +import ( + "testing" + "time" +) + +func TestTimestampsSettled(t *testing.T) { + now := time.Unix(1700000000, 0) + old := now.Add(-TimestampSettle) + for _, test := range []struct { + name string + modified, changed time.Time + want bool + }{ + {"boundary", old, old, true}, + {"recent modification", now.Add(-time.Second), old, false}, + {"recent change", old, now.Add(-time.Second), false}, + {"future modification", now.Add(time.Second), old, false}, + {"future change", old, now.Add(time.Second), false}, + } { + t.Run(test.name, func(t *testing.T) { + if got := TimestampsSettled(now, test.modified, test.changed, TimestampSettle); got != test.want { + t.Fatalf("settled=%v, want %v", got, test.want) + } + }) + } +} diff --git a/internal/plandb/refresh_scale_test.go b/internal/plandb/refresh_scale_test.go index 3ba71f8488..b3048df5a0 100644 --- a/internal/plandb/refresh_scale_test.go +++ b/internal/plandb/refresh_scale_test.go @@ -41,6 +41,7 @@ func TestRepeatedReadsOfAnUnchangedPlanLoadItOnce(t *testing.T) { } } runtime.ReadMemStats(&after) + t.Logf("%d unchanged plan reads allocated %d bytes", tasks, after.TotalAlloc-before.TotalAlloc) const budget = 3 << 20 if spent := after.TotalAlloc - before.TotalAlloc; spent > budget { t.Fatalf("%d reads of an unchanged plan allocated %d bytes, want under %d", tasks, spent, budget) diff --git a/internal/session/recordcache.go b/internal/session/recordcache.go index 9cdeb0346a..6687b6dc5e 100644 --- a/internal/session/recordcache.go +++ b/internal/session/recordcache.go @@ -89,6 +89,7 @@ import ( "unsafe" "github.com/Agent-Field/codeaf/internal/factory" + "github.com/Agent-Field/codeaf/internal/filememo" ) // ErrRecordReadBusy is a reading refused because as many as may run at once @@ -130,7 +131,7 @@ const ( recordCacheBytes = 256 << 20 recordCacheEntries = 8 recordCacheFileBytes = 16 << 20 - recordCacheSettle = 2 * time.Second + recordCacheSettle = filememo.TimestampSettle // A page asked for again comes within seconds; a journal nobody has read // for this long is read again for the price of one parse. recordCacheIdle = 30 * time.Second @@ -160,8 +161,11 @@ type RecordCache struct { retained int64 reserved int64 stats RecordCacheStats - // sweeping is true while a timer is armed to look for idle records. - sweeping bool + // The timer always targets the earliest unleased record's deadline. + sweepAt time.Time + stopSweep func() + sweepGeneration uint64 + after func(time.Duration, func()) func() } // recordSlot is one kept reading. A slot evicted while copies of it are out is @@ -208,7 +212,11 @@ func NewRecordCache(limits RecordCacheLimits) *RecordCache { limits.Now = time.Now } return &RecordCache{ - limits: limits, + limits: limits, + after: func(delay time.Duration, f func()) func() { + timer := time.AfterFunc(delay, f) + return func() { timer.Stop() } + }, order: list.New(), kept: map[string]*list.Element{}, flights: map[string]*recordFlight{}, @@ -309,6 +317,7 @@ func (c *RecordCache) leaseLocked(element *list.Element) (Record, func(), error) } c.order.MoveToFront(element) slot.used = c.limits.Now() + c.watchIdleLocked() c.stats.Hits++ c.mu.Unlock() record := cloneRecord(slot.record) @@ -334,7 +343,7 @@ func (c *RecordCache) parseLocked(path string, size int64, stamp recordStamp, st } var flight *recordFlight if stamped && c.flights[path] == nil && size <= c.limits.FileBytes && estimate <= c.limits.Bytes && - c.limits.Now().Sub(time.Unix(0, stamp.ctime)) >= c.limits.Settle { + filememo.TimestampsSettled(c.limits.Now(), time.Unix(0, stamp.mtime), time.Unix(0, stamp.ctime), c.limits.Settle) { flight = &recordFlight{stamp: stamp, done: make(chan struct{})} c.flights[path] = flight } @@ -413,7 +422,11 @@ func (c *RecordCache) reserveLocked(n int64) bool { func (c *RecordCache) Expire() { c.mu.Lock() defer c.mu.Unlock() - c.sweeping = false + c.expireLocked() +} + +func (c *RecordCache) expireLocked() { + c.cancelSweepLocked() now := c.limits.Now() for element := c.order.Back(); element != nil; { prev := element.Prev() @@ -425,13 +438,42 @@ func (c *RecordCache) Expire() { c.watchIdleLocked() } -// watchIdleLocked arms the next look for idle records while any are kept. +func (c *RecordCache) cancelSweepLocked() { + if c.stopSweep != nil { + c.stopSweep() + c.stopSweep = nil + } + c.sweepAt = time.Time{} + c.sweepGeneration++ +} + +// watchIdleLocked schedules the earliest unleased deadline, including the +// remainder after a sweep. Leased records are reconsidered when returned. func (c *RecordCache) watchIdleLocked() { - if c.sweeping || c.order.Len() == 0 { + var deadline time.Time + for element := c.order.Back(); element != nil; element = element.Prev() { + slot := element.Value.(*recordSlot) + due := slot.used.Add(c.limits.Idle) + if slot.leases == 0 && (deadline.IsZero() || due.Before(deadline)) { + deadline = due + } + } + if deadline.Equal(c.sweepAt) { + return + } + c.cancelSweepLocked() + if deadline.IsZero() { return } - c.sweeping = true - time.AfterFunc(c.limits.Idle, c.Expire) + c.sweepAt = deadline + generation := c.sweepGeneration + c.stopSweep = c.after(max(0, deadline.Sub(c.limits.Now())), func() { + c.mu.Lock() + defer c.mu.Unlock() + if c.sweepGeneration == generation { + c.expireLocked() + } + }) } // dropLocked forgets one kept reading; its bytes stay counted while a copy of @@ -452,6 +494,7 @@ func (c *RecordCache) unleaseLocked(slot *recordSlot) { if slot.dropped && slot.leases == 0 { c.retained -= slot.cost } + c.watchIdleLocked() } // readSettled is the plain reading of the journal at path, through the one diff --git a/internal/session/recordcache_test.go b/internal/session/recordcache_test.go index 91a208bab0..6d3600e47a 100644 --- a/internal/session/recordcache_test.go +++ b/internal/session/recordcache_test.go @@ -701,3 +701,34 @@ func TestAKeptRecordNobodyAsksForIsDroppedWhenItIdles(t *testing.T) { t.Fatalf("parses = %d, want two first readings and the expired journal read afresh", s.Parses) } } + +func TestRecordExpirySchedulesTheEarliestRemainingDeadline(t *testing.T) { + first := filepath.Join(t.TempDir(), "first.jsonl") + second := filepath.Join(t.TempDir(), "second.jsonl") + writeRecord(t, first, recordJournal(1, false)) + writeRecord(t, second, recordJournal(1, false)) + now := time.Now().Add(time.Hour) + cache := NewRecordCache(RecordCacheLimits{Now: func() time.Time { return now }}) + var delay time.Duration + var fire func() + cache.after = func(d time.Duration, f func()) func() { delay, fire = d, f; return func() {} } + read(t, cache, first) + if delay != 30*time.Second { + t.Fatalf("first deadline in %s, want 30s", delay) + } + now = now.Add(time.Second) + read(t, cache, second) + now = now.Add(29 * time.Second) + fire() + if got := cache.Stats().Entries; got != 1 { + t.Fatalf("first deadline kept %d entries, want 1", got) + } + if delay != time.Second { + t.Fatalf("next expiry in %s, want the remaining 1s", delay) + } + now = now.Add(time.Second) + fire() + if got := cache.Stats().Entries; got != 0 { + t.Fatalf("second deadline kept %d entries, want 0", got) + } +} diff --git a/internal/session/task_app_runs_scale_test.go b/internal/session/task_app_runs_scale_test.go index d7cad87463..f29c50a1a1 100644 --- a/internal/session/task_app_runs_scale_test.go +++ b/internal/session/task_app_runs_scale_test.go @@ -43,6 +43,7 @@ func TestAppTasksOpensThePlanOnceForAllTheRunRows(t *testing.T) { runtime.ReadMemStats(&before) views, err := agent.AppTasks() runtime.ReadMemStats(&after) + t.Logf("listing %d run rows allocated %d bytes", runs, after.TotalAlloc-before.TotalAlloc) if err != nil { t.Fatal(err) } diff --git a/internal/session/task_index.go b/internal/session/task_index.go index 8d3a88ac24..e3ca89814a 100644 --- a/internal/session/task_index.go +++ b/internal/session/task_index.go @@ -73,6 +73,7 @@ import ( "github.com/Agent-Field/codeaf/internal/crewroute" "github.com/Agent-Field/codeaf/internal/delegate" + "github.com/Agent-Field/codeaf/internal/filememo" "github.com/Agent-Field/codeaf/internal/plandb" "github.com/Agent-Field/codeaf/internal/reportline" ) @@ -1350,6 +1351,7 @@ type taskIndexKept struct { } type taskIndexCache struct { + now func() time.Time mu sync.Mutex kept []taskIndexKept // most recently read last } @@ -1361,8 +1363,12 @@ func (c *taskIndexCache) read(path string) ([]TaskIndexEntry, bool) { if err != nil || !info.Mode().IsRegular() { return parseTaskIndex(path) } + now := time.Now() + if c.now != nil { + now = c.now() + } stamp, stamped := recordStampOf(info) - if !stamped || info.Size() > taskIndexKeptBytes { + if !stamped || info.Size() > taskIndexKeptBytes || !filememo.TimestampsSettled(now, info.ModTime(), time.Unix(0, stamp.ctime), recordCacheSettle) { return parseTaskIndex(path) } c.mu.Lock() @@ -1388,7 +1394,9 @@ func (c *taskIndexCache) read(path string) ([]TaskIndexEntry, bool) { c.kept = slices.DeleteFunc(c.kept, func(entry taskIndexKept) bool { return entry.path == path }) c.kept = append(c.kept, taskIndexKept{path: path, stamp: stamp, rows: rows}) if len(c.kept) > taskIndexKeptPaths { - c.kept = c.kept[len(c.kept)-taskIndexKeptPaths:] + discard := len(c.kept) - taskIndexKeptPaths + clear(c.kept[:discard]) + c.kept = c.kept[discard:] } return rows, true } diff --git a/internal/session/task_index_cache_test.go b/internal/session/task_index_cache_test.go index 36fbe0d028..16951da95a 100644 --- a/internal/session/task_index_cache_test.go +++ b/internal/session/task_index_cache_test.go @@ -17,6 +17,12 @@ func writeIndexRows(t *testing.T, path string, from, to int) { t.Fatal(err) } defer file.Close() + defer func() { + old := time.Now().Add(-time.Hour) + if err := os.Chtimes(path, old, old); err != nil { + t.Fatal(err) + } + }() for i := from; i < to; i++ { row := fmt.Sprintf(`{"id":"%d","sessionId":"s%d","name":"task-%d","label":"task %d","title":"task number %d, groomed","status":"done","endedAt":%q}`+"\n", i, i%40, i, i, i, time.Unix(1_700_000_000+int64(i), 0).UTC().Format(time.RFC3339)) @@ -31,6 +37,9 @@ func writeIndexRows(t *testing.T, path string, from, to int) { // hundred times. An index nobody has appended to since it was read is answered // from that reading; a row appended after it is read. func TestAnUnchangedTaskIndexIsDecodedOnce(t *testing.T) { + // Move only the cache clock; changing mtime cannot age the inode's ctime. + taskIndexParsed.now = func() time.Time { return time.Now().Add(time.Hour) } + t.Cleanup(func() { taskIndexParsed.now = nil }) path := filepath.Join(t.TempDir(), taskIndexName) writeIndexRows(t, path, 0, 1500) want := ReadTaskIndex(path) @@ -46,6 +55,7 @@ func TestAnUnchangedTaskIndexIsDecodedOnce(t *testing.T) { got[0], got[len(got)-1] = got[len(got)-1], got[0] } runtime.ReadMemStats(&after) + t.Logf("40 readings of a 1,500-row index allocated %d bytes", after.TotalAlloc-before.TotalAlloc) const budget = 64 << 20 if spent := after.TotalAlloc - before.TotalAlloc; spent > budget { t.Fatalf("40 readings of an unchanged index allocated %d bytes, want under %d", spent, budget) @@ -59,3 +69,53 @@ func TestAnUnchangedTaskIndexIsDecodedOnce(t *testing.T) { t.Fatalf("after an append the index has %d rows, want 1501", len(got)) } } + +func TestTaskIndexEvictionClearsRetainedRows(t *testing.T) { + cache := taskIndexCache{now: func() time.Time { return time.Now().Add(time.Hour) }, kept: make([]taskIndexKept, 0, 8)} + paths := make([]string, 8) + for i := range paths { + paths[i] = filepath.Join(t.TempDir(), taskIndexName) + writeIndexRows(t, paths[i], i, i+1) + } + for _, path := range paths[:4] { + cache.read(path) + } + backing := cache.kept[:cap(cache.kept)] + for _, path := range paths[4:] { + cache.read(path) + } + for _, entry := range backing[:4] { + if entry.rows != nil { + t.Fatal("evicted index still reachable through discarded backing array") + } + } +} + +func TestTaskIndexRejectsUnsettledChangeTime(t *testing.T) { + path := filepath.Join(t.TempDir(), taskIndexName) + writeIndexRows(t, path, 0, 1) + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + stamp, ok := recordStampOf(info) + if !ok { + t.Skip("file identity is unavailable") + } + now := time.Unix(0, stamp.ctime).Add(time.Second) + cache := taskIndexCache{now: func() time.Time { return now }} + for range 2 { + rows, ok := cache.read(path) + if !ok || len(rows) != 1 { + t.Fatalf("unsettled file was not read: %+v %v", rows, ok) + } + if len(cache.kept) != 0 { + t.Fatal("index with old mtime but unsettled ctime was cached") + } + } + now = now.Add(2 * time.Second) + cache.read(path) + if len(cache.kept) != 1 { + t.Fatal("settled index did not enter cache") + } +} From 6c54d92134a438f7fae3b3410675f47f0d967174 Mon Sep 17 00:00:00 2001 From: Santosh Date: Sun, 25 Oct 2026 09:20:12 -0400 Subject: [PATCH 4/7] engine: isolate cached task rows and share listing reads Returned task-index structs still shared nested files, crew and verdict storage with cached rows. Deep-copy those fields on return, including route prices, crew maps and nested slices. Concurrent misses decoded the same index independently. Coordinate in-flight reads by path and full file stamp, without admitting unsettled or oversized files. Changed stamps start separate reads and cannot be replaced by an old completion. AppTasks reopened the live plan for its plan projection and reread matching trajectories. Share handles and memoized trajectory projections within one listing. Keep only plan counts and requested app facts; release raw payloads before the next task and start fresh on the next listing. Deterministic red/green gates observed 16 -> 1 decodings and 2 -> 1 plan opens and matching trajectory reads. Added nested ownership, stamp independence, listing freshness and weak-reference payload-retention coverage. Existing allocation caps remain unchanged. Validation: 35 focused internal/session tests passed with -race -count=3 via make test-focus; six final regression tests also passed with those flags. An overlay restoring duplicate trajectory reads fails the read-count gate. git diff --check passes. The changed changelog entry validates individually; repository-wide changelog checking has the same 231 errors at the base. Co-Authored-By: Claude Opus 5.5 --- .../unreleased/1933-engine-memory-safe.md | 9 +- docs/perf/engine-memory-safe.md | 23 +++- internal/session/plandb_read.go | 42 ++++++ internal/session/plandb_tasks.go | 35 +++-- internal/session/run_lifecycle_rows_test.go | 2 +- internal/session/task_app.go | 18 +-- internal/session/task_app_runs.go | 15 ++- internal/session/task_app_runs_scale_test.go | 104 +++++++++++++++ internal/session/task_app_steps.go | 13 +- internal/session/task_index.go | 121 +++++++++++++---- internal/session/task_index_cache_test.go | 125 ++++++++++++++++++ internal/session/task_run_belt_test.go | 2 +- 12 files changed, 441 insertions(+), 68 deletions(-) create mode 100644 internal/session/plandb_read.go diff --git a/docs/changes/unreleased/1933-engine-memory-safe.md b/docs/changes/unreleased/1933-engine-memory-safe.md index 9965ba23a4..0ea919d79e 100644 --- a/docs/changes/unreleased/1933-engine-memory-safe.md +++ b/docs/changes/unreleased/1933-engine-memory-safe.md @@ -8,10 +8,13 @@ invalidates: - Journal caches previously kept up to 32 records without idle expiry; they now keep at most eight and expire unleased records after 30 seconds without a read. --- -Conversation task listings share plan handles across run rows and avoid copying -the run history per plan task. Project task indexes reuse decoded rows for up to +Conversation task listings share plan handles across run and plan projections, +read each trajectory file once per listing, and avoid copying the run history +per plan task. Project task indexes reuse decoded rows for up to four files of at most 4 MiB each, with replacement-sensitive stamps and a shared -mtime/ctime settling guard. Eviction clears retained rows; journal expiry targets +mtime/ctime settling guard. Concurrent readers coalesce decoding per path and +stamp, and returned rows own their nested files, crew and verdict evidence. +Eviction clears retained rows; journal expiry targets the earliest unleased deadline. See `docs/perf/engine-memory-safe.md` for gates. Only the low-risk engine subset of the original memory work ships here. Idle diff --git a/docs/perf/engine-memory-safe.md b/docs/perf/engine-memory-safe.md index 301edada2a..a691008b7d 100644 --- a/docs/perf/engine-memory-safe.md +++ b/docs/perf/engine-memory-safe.md @@ -5,14 +5,20 @@ Unchanged plan reads reuse the loaded plan while SQLite's data version and the database/WAL file stamps match. A write through this or another handle still refreshes the next read. Conversation task listings share plan read handles -across their run rows and inspect interrupted run rows without copying the -whole run history for each plan task. +across their run and plan projections. Each trajectory file is read at most once +per listing, including missing files; its parsed projections are reused within +that listing. The next listing reads fresh trajectory data. Interrupted run rows +are inspected without copying the whole run history for each plan task. Project task indexes reuse decoded rows for matching device, inode, size, mtime and ctime stamps. Admission requires mtime and ctime to be at least two seconds old; unsupported file identities bypass reuse. The cache keeps four paths, each at most 4 MiB on disk, and clears evicted references. Callers get separate row -slices for ordering and deletion filtering; the cached rows remain immutable. +slices for ordering and deletion filtering, with deep copies of files, crew +slices and maps, route prices, and verdict evidence. The cached rows remain +immutable. Concurrent misses share one in-flight decoding per path and stamp; +a changed stamp starts its own read. Unsettled and oversized files can share an +in-flight read but are never admitted to the retained cache. Journal record caches now keep at most eight records, down from 32. Unleased records expire 30 seconds after their last read, with the timer targeting the @@ -38,12 +44,21 @@ With these fixtures and ordinary `-count=1` tests, a source overlay of the | Listing 150 runs | 503,060,592 | 14,893,176 | | 40 readings of a 1,500-row index | 267,895,312 | 33,463,056 | +Review regression gates additionally assert that mutating a returned row's +nested fields leaves the next read unchanged, 16 concurrent readers of one stamp +decode once, and a listing opens its live plan once and reads each trajectory +once. A weak-reference gate checks that plan-only reads release each raw +trajectory payload before reading the next task; the memo keeps only plan step +counts and requested app facts. Before these corrections, the deterministic +fixtures observed 16 decodings, two plan opens and two reads per matching +trajectory. The allocation caps above remain unchanged. + The overlay substitutes base production files while keeping the allocation fixtures; its index fixture predates the injected cache clock. Test logs report allocations so the candidate measurement can be repeated directly: ```sh make test-focus PKGS=./internal/plandb RUN='^Test(RepeatedReadsOfAnUnchangedPlanLoadItOnce|AReadSeesAnotherHandlesWriteAfterAnUnchangedRead)$$' TEST_FLAGS='-race -count=3' -make test-focus PKGS=./internal/session RUN='^Test(AppTasksOpensThePlanOnceForAllTheRunRows|AnUnchangedTaskIndexIsDecodedOnce|TaskIndexEvictionClearsRetainedRows|TaskIndexRejectsUnsettledChangeTime|AKeptRecordNobodyAsksForIsDroppedWhenItIdles|RecordExpirySchedulesTheEarliestRemainingDeadline)$$' TEST_FLAGS='-race -count=3' +make test-focus PKGS=./internal/session RUN='^Test(AppTasksOpensThePlanOnceForAllTheRunRows|AnUnchangedTaskIndexIsDecodedOnce|TaskIndexEvictionClearsRetainedRows|TaskIndexRejectsUnsettledChangeTime|TaskIndexReadOwnsNestedFields|ConcurrentTaskIndexReadersDecodeOneStampOnce|TaskIndexChangedStampDoesNotJoinOlderRead|AppTaskListingRereadsChangedTrajectories|PlanTaskListingDoesNotRetainTrajectoryPayloads|AKeptRecordNobodyAsksForIsDroppedWhenItIdles|RecordExpirySchedulesTheEarliestRemainingDeadline)$$' TEST_FLAGS='-race -count=3' make test-focus PKGS=./internal/filememo RUN='^TestTimestampsSettled$$' TEST_FLAGS='-race -count=3' ``` diff --git a/internal/session/plandb_read.go b/internal/session/plandb_read.go new file mode 100644 index 0000000000..43aa5c6973 --- /dev/null +++ b/internal/session/plandb_read.go @@ -0,0 +1,42 @@ +package session + +import ( + "bytes" + + "github.com/Agent-Field/codeaf/internal/plandb" +) + +// planTaskRead shares handles and trajectory projections within one listing. +// Its caller owns the plan gate and closes the handles after both projections. +type planTaskRead struct { + stores []*plandb.Store + plan *planState + trajectories map[string]*planTrajectoryRead +} + +// trajectory reads each path once. App run rows are projected before plan rows, +// so their step facts are requested on the first read of any shared path. +func (r *planTaskRead) trajectory(path string, withAppSteps bool) *planTrajectoryRead { + if got, ok := r.trajectories[path]; ok { + return got + } + data, _ := readPlanTrajectoryFile(path) + got := &planTrajectoryRead{steps: len(parsePlanTrajectory(data))} + if withAppSteps { + got.app, got.reads = parseAppTrajectorySteps(bytes.NewReader(data), appTaskStepsMax) + } + if r.trajectories == nil { + r.trajectories = make(map[string]*planTrajectoryRead) + } + r.trajectories[path] = got + return got +} + +// Only counts and requested app facts survive a read; raw files and complete +// plan steps become collectible before the next task. Both parsers retain their +// distinct line limits and tolerated record formats. +type planTrajectoryRead struct { + steps int + app []AppTaskStep + reads map[string]int +} diff --git a/internal/session/plandb_tasks.go b/internal/session/plandb_tasks.go index 11ab452709..c3e7230649 100644 --- a/internal/session/plandb_tasks.go +++ b/internal/session/plandb_tasks.go @@ -297,6 +297,12 @@ const planTrajectoryFile = "trajectory.jsonl" func (a *Agent) PlanTasks() []PlanTaskRow { stores, plan, closeStores := a.openPlanReadHandles() defer closeStores() + return a.planTasksIn(&planTaskRead{stores: stores, plan: plan}) +} + +// planTasksIn projects plan rows using the handles and trajectories of one listing. +func (a *Agent) planTasksIn(read *planTaskRead) []PlanTaskRow { + stores, plan := read.stores, read.plan if len(stores) == 0 { return nil } @@ -326,7 +332,7 @@ func (a *Agent) PlanTasks() []PlanTaskRow { } linked := func(token string) (*TaskSource, bool) { source, ok := links[token]; return source, ok } for _, task := range tasks { - row := planTaskRow(store, dir, task, spend, live) + row := planTaskRow(store, dir, task, spend, live, read.trajectory(planTrajectoryPath(dir, task.ID), false).steps) row.Source = cloneSource(planRowSource(task.ID, func(id string) string { return parents[id] }, linked)) planCarriedRow(&row, carried[task.ID]) clocks.apply(&row, dir, task, root) @@ -418,7 +424,7 @@ func (a *Agent) planTaskPageFor(id string, withProgram bool) (PlanTaskPage, plan var children []PlanTaskRow depths := map[string]int{task.ID: -1} for _, child := range all { - row := planTaskRow(store, dir, child, spend, live) + row := planTaskRow(store, dir, child, spend, live, len(planTrajectory(dir, child.ID))) planCarriedRow(&row, carried[child.ID]) clocks.apply(&row, dir, child, store.RootID()) a.markPlanInterrupted(&row) @@ -541,6 +547,15 @@ func (a *Agent) markPlanInterrupted(row *PlanTaskRow) { } } +// These file seams let read-count gates exercise real stores and trajectories. +var openPlanReadStore = func(path string) *plandb.Store { + store, _ := plandb.Open(path, "", "", "", "") + return store +} + +var readPlanTrajectoryFile = os.ReadFile +var openPlanTrajectoryFile = os.Open + // openPlanReadHandles opens every ended run oldest-first and then the live run. // Ended handles stay cached because an ended store never changes; only the live // handle is reopened for each read so writes from another process are visible. @@ -561,8 +576,8 @@ func (a *Agent) openPlanReadHandles() ([]*plandb.Store, *planState, func()) { for _, path := range planArchivePaths(plan.path) { store := plan.archives[path] if store == nil { - opened, err := plandb.Open(path, "", "", "", "") - if err != nil { + opened := openPlanReadStore(path) + if opened == nil { continue } plan.archives[path] = opened @@ -570,7 +585,7 @@ func (a *Agent) openPlanReadHandles() ([]*plandb.Store, *planState, func()) { } stores = append(stores, store) } - live := plan.open() + live := openPlanReadStore(plan.path) if live != nil { stores = append(stores, live) } @@ -807,7 +822,7 @@ func applyPlanRootProgress(row *PlanTaskRow, tasks []*plandb.Task, root string) // not on the task: the dollars its spend rows carry, already summed, and its // steps, already read. The live step is the third: the store's live rows, read // whole in one pass by the caller, keyed by the task's own bare id. -func planTaskRow(store *plandb.Store, dir string, task *plandb.Task, spend map[string]float64, live map[string]plandb.LiveStep) PlanTaskRow { +func planTaskRow(store *plandb.Store, dir string, task *plandb.Task, spend map[string]float64, live map[string]plandb.LiveStep, steps int) PlanTaskRow { seat, _ := store.RoleOf(task.ID) // A HELD TASK WEARS THE HOLD'S OWN WORD. Pause is status-independent in the // store — a held task keeps the rung it reached — while the row says what a @@ -825,7 +840,7 @@ func planTaskRow(store *plandb.Store, dir string, task *plandb.Task, spend map[s Stopped: planTaskStopped(store, task), Interrupted: planTaskInterrupted(task), Seat: seat, - Steps: len(planTrajectory(dir, task.ID)), + Steps: steps, USD: spend[task.ID], Started: task.CreatedAt, Ended: task.CompletedAt, @@ -974,10 +989,14 @@ func planTrajectoryPath(dir, id string) string { // skipped — the record is about the steps, and an interrupted append leaves a // half-written last line behind. func planTrajectory(dir, id string) []PlanStep { - data, err := os.ReadFile(planTrajectoryPath(dir, id)) + data, err := readPlanTrajectoryFile(planTrajectoryPath(dir, id)) if err != nil { return nil } + return parsePlanTrajectory(data) +} + +func parsePlanTrajectory(data []byte) []PlanStep { var steps []PlanStep for _, line := range strings.Split(string(data), "\n") { line = strings.TrimSpace(line) diff --git a/internal/session/run_lifecycle_rows_test.go b/internal/session/run_lifecycle_rows_test.go index e7d32e5f8c..9676ac8215 100644 --- a/internal/session/run_lifecycle_rows_test.go +++ b/internal/session/run_lifecycle_rows_test.go @@ -32,7 +32,7 @@ func TestARunSetAsideAsInterruptedReadsInterruptedOnItsPage(t *testing.T) { } defer archived.Close() for _, id := range []string{"71", "part"} { - row := planTaskRow(archived, dir, archived.Task(id), nil, nil) + row := planTaskRow(archived, dir, archived.Task(id), nil, nil, len(planTrajectory(dir, id))) if !row.Interrupted || row.Stopped { t.Fatalf("row %s of a run set aside reads %+v, want interrupted and not stopped", id, row) } diff --git a/internal/session/task_app.go b/internal/session/task_app.go index 250ad97cd4..e34d291576 100644 --- a/internal/session/task_app.go +++ b/internal/session/task_app.go @@ -13,7 +13,6 @@ import ( "github.com/Agent-Field/codeaf/internal/config" "github.com/Agent-Field/codeaf/internal/crewroute" - "github.com/Agent-Field/codeaf/internal/plandb" ) // AppTaskView is an engine read, independent of the app's JSON contract. @@ -263,25 +262,20 @@ func (a *Agent) AppTasks() ([]AppTaskView, error) { cap, _ := config.CrewTaskSpendCap(a.config.ProfileDir) graphIDs := map[string]bool{} out = a.appGraphTasks(out, cap, graphIDs) - // A HANDED-OVER TASK is a run's row, not a node (task_app_runs.go). - // The plan is opened once for every run row and closed before the plan's - // own rows below open it again: the handles hold the plan's gate. + // Both projections share the plan gate and each trajectory for this listing. rows := a.appRunRows() - var stores []*plandb.Store - closeStores := func() {} - if len(rows) > 0 { - stores, _, closeStores = a.openPlanReadHandles() - } + stores, plan, closeStores := a.openPlanReadHandles() + defer closeStores() + read := &planTaskRead{stores: stores, plan: plan} for _, row := range rows { id := strconv.FormatUint(row.ID, 10) if graphIDs[id] { continue } graphIDs[id] = true - out = append(out, a.appRunViewIn(row, cap, stores)) + out = append(out, a.appRunViewIn(row, cap, read)) } - closeStores() - for _, row := range a.PlanTasks() { + for _, row := range a.planTasksIn(read) { if graphIDs[row.ID] { continue } diff --git a/internal/session/task_app_runs.go b/internal/session/task_app_runs.go index 70e40ccd95..661a416cad 100644 --- a/internal/session/task_app_runs.go +++ b/internal/session/task_app_runs.go @@ -100,14 +100,14 @@ func (a *Agent) appRunRow(id uint64) (TaskNotice, bool) { // appRunView is one run row as the app's task, opening the plan for itself. func (a *Agent) appRunView(row TaskNotice, cap float64) AppTaskView { - stores, _, closeStores := a.openPlanReadHandles() + stores, plan, closeStores := a.openPlanReadHandles() defer closeStores() - return a.appRunViewIn(row, cap, stores) + return a.appRunViewIn(row, cap, &planTaskRead{stores: stores, plan: plan}) } // appRunViewIn is one run row as the app's task, read from stores a listing // has already opened. -func (a *Agent) appRunViewIn(row TaskNotice, cap float64, stores []*plandb.Store) AppTaskView { +func (a *Agent) appRunViewIn(row TaskNotice, cap float64, read *planTaskRead) AppTaskView { if _, held := a.appHeldRuns.Load(row.ID); held && !row.State.settled() { row.Paused = true } @@ -125,7 +125,7 @@ func (a *Agent) appRunViewIn(row TaskNotice, cap float64, stores []*plandb.Store for _, id := range row.DependsOn { view.Waits = append(view.Waits, strconv.FormatUint(id, 10)) } - a.appRunDetail(&view, appRunRowKey(row), stores) + a.appRunDetail(&view, appRunRowKey(row), read) switch { case row.park != nil || a.taskRouting.isParked(row.ID): view.RoutingState = TaskRoutingBlocked @@ -146,14 +146,15 @@ func (a *Agent) appRunViewIn(row TaskNotice, cap float64, stores []*plandb.Store // // The stores are the plan's read handles, opened once for the whole listing: // opening one reads the whole plan, and a listing draws a row for every run. -func (a *Agent) appRunDetail(view *AppTaskView, key string, stores []*plandb.Store) { - for _, store := range stores { +func (a *Agent) appRunDetail(view *AppTaskView, key string, read *planTaskRead) { + for _, store := range read.stores { task := store.Task(key) if task == nil { continue } dir := filepath.Dir(store.Path()) - steps, reads := appTrajectorySteps(filepath.Join(plandb.TaskDir(dir, key), planTrajectoryFile), appTaskStepsMax) + trajectory := read.trajectory(planTrajectoryPath(dir, key), true) + steps, reads := trajectory.app, trajectory.reads view.Steps = steps view.WorkLog = appTrajectoryLines(steps) if len(steps) > 0 { diff --git a/internal/session/task_app_runs_scale_test.go b/internal/session/task_app_runs_scale_test.go index f29c50a1a1..d2feb90428 100644 --- a/internal/session/task_app_runs_scale_test.go +++ b/internal/session/task_app_runs_scale_test.go @@ -2,9 +2,11 @@ package session import ( "fmt" + "os" "path/filepath" "runtime" "testing" + "weak" "github.com/Agent-Field/codeaf/internal/plandb" ) @@ -39,6 +41,16 @@ func TestAppTasksOpensThePlanOnceForAllTheRunRows(t *testing.T) { } graph.mu.Unlock() + openStore, readFile, openFile := openPlanReadStore, readPlanTrajectoryFile, openPlanTrajectoryFile + t.Cleanup(func() { + openPlanReadStore, readPlanTrajectoryFile, openPlanTrajectoryFile = openStore, readFile, openFile + }) + opens := map[string]int{} + reads := map[string]int{} + openPlanReadStore = func(path string) *plandb.Store { opens[path]++; return openStore(path) } + readPlanTrajectoryFile = func(path string) ([]byte, error) { reads[path]++; return readFile(path) } + openPlanTrajectoryFile = func(path string) (*os.File, error) { reads[path]++; return openFile(path) } + var before, after runtime.MemStats runtime.ReadMemStats(&before) views, err := agent.AppTasks() @@ -56,8 +68,100 @@ func TestAppTasksOpensThePlanOnceForAllTheRunRows(t *testing.T) { if withSteps != runs { t.Fatalf("%d of %d run rows carried their steps", withSteps, runs) } + if got := opens[path]; got != 1 { + t.Errorf("listing opened the live plan %d times, want 1", got) + } + for _, spec := range specs { + path := planTrajectoryPath(dir, spec.ID) + if got := reads[path]; got != 1 { + t.Errorf("listing read %s %d times, want 1", spec.ID, got) + break + } + } + for path, count := range reads { + if count != 1 { + t.Errorf("trajectory %s was read %d times, want 1", path, count) + break + } + } const budget = 16 << 20 if spent := after.TotalAlloc - before.TotalAlloc; spent > budget { t.Fatalf("listing %d run rows allocated %d bytes, want under %d", runs, spent, budget) } } + +func TestAppTaskListingRereadsChangedTrajectories(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, planStoreFilename) + seedPlanStore(t, path, "chat-a", plandb.TaskSpec{ID: "work", Title: "Work"}) + writePlanTrajectory(t, dir, "work", `{"kind":"step","step":1,"command":"echo first"}`) + agent, _ := newTestAgent(t, &scriptedCompleter{}, nil) + armPlanStore(t, agent, path, "chat-a") + graph := agent.graph() + graph.mu.Lock() + graph.runs = map[uint64][]TaskNotice{7: {{ID: 7, Title: "Work", State: TaskDone, PlanTask: "work"}}} + graph.runRuns = []uint64{7} + graph.mu.Unlock() + for _, count := range []int{1, 2} { + views, err := agent.AppTasks() + if err != nil { + t.Fatal(err) + } + seen := 0 + for _, view := range views { + switch view.ID { + case "7", "t-work": + if view.Step != count { + t.Errorf("%s has %d steps, want %d", view.ID, view.Step, count) + } + seen++ + } + } + if seen != 2 { + t.Fatalf("saw %d work projections, want 2", seen) + } + writePlanTrajectory(t, dir, "work", `{"kind":"step","step":1,"command":"echo first"}`, `{"kind":"step","step":2,"command":"echo next"}`) + } +} + +func TestPlanTaskListingDoesNotRetainTrajectoryPayloads(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, planStoreFilename) + specs := []plandb.TaskSpec{{ID: "alpha", Title: "Alpha"}, {ID: "beta", Title: "Beta"}, {ID: "gamma", Title: "Gamma"}} + seedPlanStore(t, path, "chat-a", specs...) + for _, spec := range specs { + writePlanTrajectory(t, dir, spec.ID, `{"kind":"step","step":1,"command":"echo recorded"}`) + } + agent, _ := newTestAgent(t, &scriptedCompleter{}, nil) + armPlanStore(t, agent, path, "chat-a") + readFile := readPlanTrajectoryFile + t.Cleanup(func() { readPlanTrajectoryFile = readFile }) + var previous weak.Pointer[byte] + checked := 0 + readPlanTrajectoryFile = func(path string) ([]byte, error) { + runtime.GC() + if previous != (weak.Pointer[byte]{}) { + checked++ + if previous.Value() != nil { + t.Error("previous trajectory payload is retained while reading the next task") + } + } + data, err := os.ReadFile(path) + if len(data) > 0 { + previous = weak.Make(&data[0]) + } + return data, err + } + rows := agent.PlanTasks() + if len(rows) != 4 { + t.Fatalf("read %d plan rows, want 4", len(rows)) + } + for _, row := range rows[1:] { + if row.Steps != 1 { + t.Errorf("%s has %d steps, want 1", row.ID, row.Steps) + } + } + if checked != 2 { + t.Fatalf("checked %d prior payloads, want 2", checked) + } +} diff --git a/internal/session/task_app_steps.go b/internal/session/task_app_steps.go index aadb14c9a0..d1fbf05064 100644 --- a/internal/session/task_app_steps.go +++ b/internal/session/task_app_steps.go @@ -40,6 +40,7 @@ package session import ( "bufio" "encoding/json" + "io" "os" "sort" "strconv" @@ -489,14 +490,18 @@ func appStepOfEntry(item DisplayEntry, at int) AppTaskStep { // command — so a run's page lost which tool had been used, whether the step ran // at all, and what it exited with, on every read. func appTrajectorySteps(path string, limit int) ([]AppTaskStep, map[string]int) { - reads := map[string]int{} - file, err := os.Open(path) + file, err := openPlanTrajectoryFile(path) if err != nil { - return nil, reads + return nil, map[string]int{} } defer file.Close() + return parseAppTrajectorySteps(file, limit) +} + +func parseAppTrajectorySteps(reader io.Reader, limit int) ([]AppTaskStep, map[string]int) { + reads := map[string]int{} var steps []AppTaskStep - scanner := bufio.NewScanner(file) + scanner := bufio.NewScanner(reader) scanner.Buffer(make([]byte, 0, 64*1024), 4*1024*1024) for scanner.Scan() { var line struct { diff --git a/internal/session/task_index.go b/internal/session/task_index.go index e3ca89814a..b63d76c426 100644 --- a/internal/session/task_index.go +++ b/internal/session/task_index.go @@ -516,9 +516,45 @@ func readTaskIndexRows(path string, limit int, filter bool) []TaskIndexEntry { if limit > 0 && len(rows) > limit { rows = rows[:limit] } + for i := range rows { + rows[i] = cloneTaskIndexEntry(rows[i]) + } return rows } +// cloneTaskIndexEntry gives callers ownership of every mutable part of a row. +func cloneTaskIndexEntry(row TaskIndexEntry) TaskIndexEntry { + row.Files = slices.Clone(row.Files) + if row.Crew != nil { + crew := *row.Crew + clonePicks := func(picks []crewroute.Pick) []crewroute.Pick { + picks = slices.Clone(picks) + for i := range picks { + picks[i].Price = clonePrice(picks[i].Price) + } + return picks + } + crew.Crew = clonePicks(crew.Crew) + if crew.Ladder != nil { + crew.Ladder = make(map[crewroute.Seat][]crewroute.Pick, len(row.Crew.Ladder)) + for seat, picks := range row.Crew.Ladder { + crew.Ladder[seat] = clonePicks(picks) + } + } + crew.Retried = slices.Clone(crew.Retried) + crew.Rungs = slices.Clone(crew.Rungs) + crew.Warnings = slices.Clone(crew.Warnings) + row.Crew = &crew + } + if row.VerdictBasis != nil { + basis := *row.VerdictBasis + basis.Runs = slices.Clone(basis.Runs) + basis.Unobserved = slices.Clone(basis.Unobserved) + row.VerdictBasis = &basis + } + return row +} + // lastPerNode keeps ONE row per node: the last thing the file says about it. // // The file is append-only and a node can be written more than once — a run's @@ -1350,53 +1386,82 @@ type taskIndexKept struct { rows []TaskIndexEntry } +type taskIndexReadKey struct { + path string + stamp recordStamp +} + +type taskIndexRead struct { + done chan struct{} + rows []TaskIndexEntry + ok bool +} + type taskIndexCache struct { - now func() time.Time - mu sync.Mutex - kept []taskIndexKept // most recently read last + parse func(string) ([]TaskIndexEntry, bool) + now func() time.Time + mu sync.Mutex + kept []taskIndexKept // most recently read last + reading map[taskIndexReadKey]*taskIndexRead } var taskIndexParsed taskIndexCache func (c *taskIndexCache) read(path string) ([]TaskIndexEntry, bool) { + parse := c.parse + if parse == nil { + parse = parseTaskIndex + } info, err := os.Stat(path) if err != nil || !info.Mode().IsRegular() { - return parseTaskIndex(path) + return parse(path) + } + stamp, stamped := recordStampOf(info) + if !stamped { + return parse(path) } now := time.Now() if c.now != nil { now = c.now() } - stamp, stamped := recordStampOf(info) - if !stamped || info.Size() > taskIndexKeptBytes || !filememo.TimestampsSettled(now, info.ModTime(), time.Unix(0, stamp.ctime), recordCacheSettle) { - return parseTaskIndex(path) - } + keep := info.Size() <= taskIndexKeptBytes && filememo.TimestampsSettled(now, info.ModTime(), time.Unix(0, stamp.ctime), recordCacheSettle) + key := taskIndexReadKey{path: path, stamp: stamp} c.mu.Lock() - for i, entry := range c.kept { - if entry.path == path && entry.stamp == stamp { - c.kept = append(append(c.kept[:i:i], c.kept[i+1:]...), entry) - c.mu.Unlock() - return entry.rows, true + if keep { + for i, entry := range c.kept { + if entry.path == path && entry.stamp == stamp { + c.kept = append(append(c.kept[:i:i], c.kept[i+1:]...), entry) + c.mu.Unlock() + return entry.rows, true + } } } - c.mu.Unlock() - rows, ok := parseTaskIndex(path) - if !ok { - return nil, false + if read := c.reading[key]; read != nil { + c.mu.Unlock() + <-read.done + return read.rows, read.ok } - // A FILE THAT CHANGED WHILE IT WAS READ IS NOT KEPT: the stamp taken - // before the reading must still be the file's after it. - if after, err := os.Stat(path); err != nil || !sameStamp(after, stamp) { - return rows, true + read := &taskIndexRead{done: make(chan struct{})} + if c.reading == nil { + c.reading = make(map[taskIndexReadKey]*taskIndexRead) } + c.reading[key] = read + c.mu.Unlock() + rows, ok := parse(path) c.mu.Lock() defer c.mu.Unlock() - c.kept = slices.DeleteFunc(c.kept, func(entry taskIndexKept) bool { return entry.path == path }) - c.kept = append(c.kept, taskIndexKept{path: path, stamp: stamp, rows: rows}) - if len(c.kept) > taskIndexKeptPaths { - discard := len(c.kept) - taskIndexKeptPaths - clear(c.kept[:discard]) - c.kept = c.kept[discard:] + // A changed or unsettled file can serve its waiting readers, but is not kept. + if after, err := os.Stat(path); keep && ok && err == nil && sameStamp(after, stamp) { + c.kept = slices.DeleteFunc(c.kept, func(entry taskIndexKept) bool { return entry.path == path }) + c.kept = append(c.kept, taskIndexKept{path: path, stamp: stamp, rows: rows}) + if len(c.kept) > taskIndexKeptPaths { + discard := len(c.kept) - taskIndexKeptPaths + clear(c.kept[:discard]) + c.kept = c.kept[discard:] + } } - return rows, true + read.rows, read.ok = rows, ok + delete(c.reading, key) + close(read.done) + return rows, ok } diff --git a/internal/session/task_index_cache_test.go b/internal/session/task_index_cache_test.go index 16951da95a..acc61e255b 100644 --- a/internal/session/task_index_cache_test.go +++ b/internal/session/task_index_cache_test.go @@ -1,13 +1,20 @@ package session import ( + "encoding/json" "fmt" "os" "path/filepath" "reflect" "runtime" + "sync" + "sync/atomic" "testing" + "testing/synctest" "time" + + "github.com/Agent-Field/codeaf/internal/crewroute" + "github.com/Agent-Field/codeaf/internal/plandb" ) func writeIndexRows(t *testing.T, path string, from, to int) { @@ -119,3 +126,121 @@ func TestTaskIndexRejectsUnsettledChangeTime(t *testing.T) { t.Fatal("settled index did not enter cache") } } + +func TestTaskIndexReadOwnsNestedFields(t *testing.T) { + taskIndexParsed.now = func() time.Time { return time.Now().Add(time.Hour) } + t.Cleanup(func() { taskIndexParsed.now = nil }) + path := filepath.Join(t.TempDir(), taskIndexName) + entry := TaskIndexEntry{ID: "1", Title: "Isolated result", Files: []string{"original.go"}, + Crew: &crewroute.Decision{Why: "original", Crew: []crewroute.Pick{{Model: "original", Price: &crewroute.RoutePrice{Prompt: 1}}}, + Retried: []crewroute.Retry{{Why: "original"}}, Rungs: []crewroute.Retry{{Why: "original"}}, Warnings: []crewroute.SeatWarning{{Model: "original"}}}, + VerdictBasis: &plandb.VerdictBasis{Kind: "run", Runs: []plandb.VerdictRun{{Command: "go test", ExitCode: 0}}, Unobserved: []string{"original"}}, + } + data, err := json.Marshal(entry) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, append(data, '\n'), 0o600); err != nil { + t.Fatal(err) + } + // Admission and subsequent hits must both return independently owned fields. + for range 2 { + got := ReadTaskIndex(path) + if len(got) != 1 { + t.Fatalf("read %d rows, want one", len(got)) + } + got[0].Files[0] = "changed.go" + got[0].Crew.Why = "changed" + got[0].Crew.Crew[0].Model = "changed" + got[0].Crew.Crew[0].Price.Prompt = 2 + got[0].Crew.Retried[0].Why = "changed" + got[0].Crew.Rungs[0].Why = "changed" + got[0].Crew.Warnings[0].Model = "changed" + got[0].VerdictBasis.Kind = "reading" + got[0].VerdictBasis.Runs[0].ExitCode = 1 + got[0].VerdictBasis.Unobserved[0] = "changed" + if next := ReadTaskIndex(path); !reflect.DeepEqual(next, []TaskIndexEntry{entry}) { + t.Fatalf("nested mutation contaminated the next read: %+v", next) + } + } +} + +func TestConcurrentTaskIndexReadersDecodeOneStampOnce(t *testing.T) { + path := filepath.Join(t.TempDir(), taskIndexName) + writeIndexRows(t, path, 0, 1) + settledNow := time.Now().Add(time.Hour) + synctest.Test(t, func(t *testing.T) { + release := make(chan struct{}) + var decodes atomic.Int32 + taskIndexParsed.now = func() time.Time { return settledNow } + taskIndexParsed.parse = func(path string) ([]TaskIndexEntry, bool) { + decodes.Add(1) + <-release + return parseTaskIndex(path) + } + t.Cleanup(func() { taskIndexParsed.now = nil; taskIndexParsed.parse = nil }) + var readers sync.WaitGroup + for range 16 { + readers.Go(func() { + if rows := ReadTaskIndex(path); len(rows) != 1 || rows[0].Title != "task number 0, groomed" { + t.Errorf("concurrent read = %+v, want the indexed row", rows) + } + }) + } + // All readers are blocked in decoding or waiting for the same decoding. + synctest.Wait() + if got := decodes.Load(); got != 1 { + t.Errorf("decoded one unchanged stamp %d times, want 1", got) + } + close(release) + readers.Wait() + if got := decodes.Load(); got != 1 { + t.Errorf("decoded one unchanged stamp %d times after completion, want 1", got) + } + }) +} + +func TestTaskIndexChangedStampDoesNotJoinOlderRead(t *testing.T) { + path := filepath.Join(t.TempDir(), taskIndexName) + writeIndexRows(t, path, 0, 1) + settledNow := time.Now().Add(time.Hour) + synctest.Test(t, func(t *testing.T) { + release := make(chan struct{}) + var decodes atomic.Int32 + taskIndexParsed.now = func() time.Time { return settledNow } + taskIndexParsed.parse = func(path string) ([]TaskIndexEntry, bool) { + first := decodes.Add(1) == 1 + rows, ok := parseTaskIndex(path) + if first { + <-release + } + return rows, ok + } + t.Cleanup(func() { taskIndexParsed.now = nil; taskIndexParsed.parse = nil }) + old := make(chan []TaskIndexEntry, 1) + go func() { old <- ReadTaskIndex(path) }() + synctest.Wait() + writeIndexRows(t, path, 1, 2) + fresh := make(chan []TaskIndexEntry, 1) + go func() { fresh <- ReadTaskIndex(path) }() + synctest.Wait() + select { + case rows := <-fresh: + if len(rows) != 2 { + t.Errorf("new stamp returned %d rows, want 2", len(rows)) + } + default: + t.Error("new stamp waited for the old decoding") + } + close(release) + if rows := <-old; len(rows) != 1 { + t.Errorf("old stamp returned %d rows, want 1", len(rows)) + } + if rows := ReadTaskIndex(path); len(rows) != 2 { + t.Errorf("old completion replaced the new stamp: %+v", rows) + } + if got := decodes.Load(); got != 2 { + t.Errorf("decoded two stamps %d times, want 2", got) + } + }) +} diff --git a/internal/session/task_run_belt_test.go b/internal/session/task_run_belt_test.go index 6e6049f735..27f5c194a8 100644 --- a/internal/session/task_run_belt_test.go +++ b/internal/session/task_run_belt_test.go @@ -1405,7 +1405,7 @@ func TestPlanReadsWithoutArchiveStayIdentical(t *testing.T) { tasks := store.Tasks(plandb.Filter{Chat: "chat-a"}) wantRows := make([]PlanTaskRow, 0, len(tasks)) for _, task := range tasks { - row := planTaskRow(store, dir, task, planSpendByTask(path), store.LiveSteps()) + row := planTaskRow(store, dir, task, planSpendByTask(path), store.LiveSteps(), len(planTrajectory(dir, task.ID))) // The listing says where the run works, once per row, exactly as the // single-store read did before there was an archive to aggregate. row.Folder = agent.config.Workspace From e86438d65e730919469df16ba1c7f6775a2b1028 Mon Sep 17 00:00:00 2001 From: Santosh Date: Sun, 25 Oct 2026 09:32:32 -0400 Subject: [PATCH 5/7] fix: remove task-index singleflight and seal returned row capacity Remove per-path singleflight coordination and its injected parser and tests. Concurrent misses decode independently again; retained-cache admission still requires a settled, unchanged file stamp. This removes decoder-panic waiters and stale sharing across unsettled same-stamp rewrites. Keep deep copies and cap caller-visible row slices at their lengths. A new regression reslices results to capacity, mutates reachable files, crew and verdict evidence, and verifies ordinary reads remain unchanged after limiting, deduplication and tombstone filtering. Preserve shared listing reads and their read-count assertion. Update the existing change entry and performance gates to describe independent misses. Validation: focused internal/session tests passed with -race -count=3 via make test-focus, including the new regression and listing read-count gate. The regression failed before the fix for excess capacity and cache mutation. Existing allocation budgets pass; make fmt-check and git diff --check pass. The changed changelog entry validates; full changelog checking has the same 231 errors at the base commit. Standards review: 0 findings. Spec review: 0 findings. Co-Authored-By: Claude Opus 5.5 --- .../unreleased/1933-engine-memory-safe.md | 7 +- docs/perf/engine-memory-safe.md | 22 +-- internal/session/task_index.go | 99 +++++--------- internal/session/task_index_cache_test.go | 128 +++++++----------- 4 files changed, 102 insertions(+), 154 deletions(-) diff --git a/docs/changes/unreleased/1933-engine-memory-safe.md b/docs/changes/unreleased/1933-engine-memory-safe.md index 0ea919d79e..338d75690e 100644 --- a/docs/changes/unreleased/1933-engine-memory-safe.md +++ b/docs/changes/unreleased/1933-engine-memory-safe.md @@ -12,8 +12,11 @@ Conversation task listings share plan handles across run and plan projections, read each trajectory file once per listing, and avoid copying the run history per plan task. Project task indexes reuse decoded rows for up to four files of at most 4 MiB each, with replacement-sensitive stamps and a shared -mtime/ctime settling guard. Concurrent readers coalesce decoding per path and -stamp, and returned rows own their nested files, crew and verdict evidence. +mtime/ctime settling guard. Concurrent misses decode independently; per-path +singleflight was removed to avoid stranded waiters after decoder panics and +stale results shared across unsettled same-stamp rewrites. Returned rows own +their nested files, crew and verdict evidence, and their slice capacity equals +their length so hidden rows cannot expose cached fields after filtering or limiting. Eviction clears retained rows; journal expiry targets the earliest unleased deadline. See `docs/perf/engine-memory-safe.md` for gates. diff --git a/docs/perf/engine-memory-safe.md b/docs/perf/engine-memory-safe.md index a691008b7d..01afd6614f 100644 --- a/docs/perf/engine-memory-safe.md +++ b/docs/perf/engine-memory-safe.md @@ -15,10 +15,11 @@ and ctime stamps. Admission requires mtime and ctime to be at least two seconds old; unsupported file identities bypass reuse. The cache keeps four paths, each at most 4 MiB on disk, and clears evicted references. Callers get separate row slices for ordering and deletion filtering, with deep copies of files, crew -slices and maps, route prices, and verdict evidence. The cached rows remain -immutable. Concurrent misses share one in-flight decoding per path and stamp; -a changed stamp starts its own read. Unsettled and oversized files can share an -in-flight read but are never admitted to the retained cache. +slices and maps, route prices, and verdict evidence. Returned row slices have +capacity equal to length, so reslicing cannot expose un-cloned cached fields +after limiting, deduplication or tombstone filtering. The cached rows remain +immutable. Concurrent misses decode independently; there is no in-flight read +coordination. Unsettled and oversized files bypass the retained cache. Journal record caches now keep at most eight records, down from 32. Unleased records expire 30 seconds after their last read, with the timer targeting the @@ -45,13 +46,14 @@ With these fixtures and ordinary `-count=1` tests, a source overlay of the | 40 readings of a 1,500-row index | 267,895,312 | 33,463,056 | Review regression gates additionally assert that mutating a returned row's -nested fields leaves the next read unchanged, 16 concurrent readers of one stamp -decode once, and a listing opens its live plan once and reads each trajectory -once. A weak-reference gate checks that plan-only reads release each raw +nested fields, including every row reachable by reslicing to capacity, leaves +the next read unchanged, and a listing opens its live plan once and reads each +trajectory once. A weak-reference gate checks that plan-only reads release each raw trajectory payload before reading the next task; the memo keeps only plan step counts and requested app facts. Before these corrections, the deterministic -fixtures observed 16 decodings, two plan opens and two reads per matching -trajectory. The allocation caps above remain unchanged. +listing fixture observed two plan opens and two reads per matching trajectory. +The former concurrent-decoding gate was removed with singleflight coordination; +independent concurrent misses are intentional. The allocation caps above remain unchanged. The overlay substitutes base production files while keeping the allocation fixtures; its index fixture predates the injected cache clock. Test logs report @@ -59,6 +61,6 @@ allocations so the candidate measurement can be repeated directly: ```sh make test-focus PKGS=./internal/plandb RUN='^Test(RepeatedReadsOfAnUnchangedPlanLoadItOnce|AReadSeesAnotherHandlesWriteAfterAnUnchangedRead)$$' TEST_FLAGS='-race -count=3' -make test-focus PKGS=./internal/session RUN='^Test(AppTasksOpensThePlanOnceForAllTheRunRows|AnUnchangedTaskIndexIsDecodedOnce|TaskIndexEvictionClearsRetainedRows|TaskIndexRejectsUnsettledChangeTime|TaskIndexReadOwnsNestedFields|ConcurrentTaskIndexReadersDecodeOneStampOnce|TaskIndexChangedStampDoesNotJoinOlderRead|AppTaskListingRereadsChangedTrajectories|PlanTaskListingDoesNotRetainTrajectoryPayloads|AKeptRecordNobodyAsksForIsDroppedWhenItIdles|RecordExpirySchedulesTheEarliestRemainingDeadline)$$' TEST_FLAGS='-race -count=3' +make test-focus PKGS=./internal/session RUN='^Test(AppTasksOpensThePlanOnceForAllTheRunRows|AnUnchangedTaskIndexIsDecodedOnce|TaskIndexEvictionClearsRetainedRows|TaskIndexRejectsUnsettledChangeTime|TaskIndexReadOwnsNestedFields|TaskIndexReadCapacityCannotReachCachedRows|AppTaskListingRereadsChangedTrajectories|PlanTaskListingDoesNotRetainTrajectoryPayloads|AKeptRecordNobodyAsksForIsDroppedWhenItIdles|RecordExpirySchedulesTheEarliestRemainingDeadline)$$' TEST_FLAGS='-race -count=3' make test-focus PKGS=./internal/filememo RUN='^TestTimestampsSettled$$' TEST_FLAGS='-race -count=3' ``` diff --git a/internal/session/task_index.go b/internal/session/task_index.go index b63d76c426..0e719b1310 100644 --- a/internal/session/task_index.go +++ b/internal/session/task_index.go @@ -514,12 +514,12 @@ func readTaskIndexRows(path string, limit int, filter bool) []TaskIndexEntry { } sortTaskIndex(rows) if limit > 0 && len(rows) > limit { - rows = rows[:limit] + rows = rows[:limit:limit] } for i := range rows { rows[i] = cloneTaskIndexEntry(rows[i]) } - return rows + return rows[:len(rows):len(rows)] } // cloneTaskIndexEntry gives callers ownership of every mutable part of a row. @@ -635,7 +635,7 @@ func (a *Agent) TaskIndex() []TaskIndexEntry { rows := ReadTaskIndex(a.config.taskIndexFile()) live := a.liveTaskRows() if len(live) == 0 { - return rows + return rows[:len(rows):len(rows)] } // A live row REPLACES the file's row for the same node: a node that landed // and was recorded, and is somehow still in the graph, is described more @@ -653,7 +653,7 @@ func (a *Agent) TaskIndex() []TaskIndexEntry { merged = append(merged, entry) } sortTaskIndex(merged) - return merged + return merged[:len(merged):len(merged)] } // liveTaskRows is this session's graph as index rows. It reads the graph @@ -677,7 +677,7 @@ func (a *Agent) liveTaskRows() []TaskIndexEntry { } rows = append(rows, node.indexEntryLocked(session)) } - return rows + return rows[:len(rows):len(rows)] } // recordTaskIndex writes one landed node into the project's index. It is called @@ -1091,9 +1091,9 @@ func SearchTaskIndex(rows []TaskIndexEntry, query string, limit int) []TaskIndex needle := strings.ToLower(strings.TrimSpace(query)) if needle == "" { if len(rows) > limit { - return rows[:limit] + return rows[:limit:limit] } - return rows + return rows[:len(rows):len(rows)] } type ranked struct { entry TaskIndexEntry @@ -1386,82 +1386,53 @@ type taskIndexKept struct { rows []TaskIndexEntry } -type taskIndexReadKey struct { - path string - stamp recordStamp -} - -type taskIndexRead struct { - done chan struct{} - rows []TaskIndexEntry - ok bool -} - type taskIndexCache struct { - parse func(string) ([]TaskIndexEntry, bool) - now func() time.Time - mu sync.Mutex - kept []taskIndexKept // most recently read last - reading map[taskIndexReadKey]*taskIndexRead + now func() time.Time + mu sync.Mutex + kept []taskIndexKept // most recently read last } var taskIndexParsed taskIndexCache func (c *taskIndexCache) read(path string) ([]TaskIndexEntry, bool) { - parse := c.parse - if parse == nil { - parse = parseTaskIndex - } info, err := os.Stat(path) if err != nil || !info.Mode().IsRegular() { - return parse(path) - } - stamp, stamped := recordStampOf(info) - if !stamped { - return parse(path) + return parseTaskIndex(path) } now := time.Now() if c.now != nil { now = c.now() } - keep := info.Size() <= taskIndexKeptBytes && filememo.TimestampsSettled(now, info.ModTime(), time.Unix(0, stamp.ctime), recordCacheSettle) - key := taskIndexReadKey{path: path, stamp: stamp} + stamp, stamped := recordStampOf(info) + if !stamped || info.Size() > taskIndexKeptBytes || !filememo.TimestampsSettled(now, info.ModTime(), time.Unix(0, stamp.ctime), recordCacheSettle) { + return parseTaskIndex(path) + } c.mu.Lock() - if keep { - for i, entry := range c.kept { - if entry.path == path && entry.stamp == stamp { - c.kept = append(append(c.kept[:i:i], c.kept[i+1:]...), entry) - c.mu.Unlock() - return entry.rows, true - } + for i, entry := range c.kept { + if entry.path == path && entry.stamp == stamp { + c.kept = append(append(c.kept[:i:i], c.kept[i+1:]...), entry) + c.mu.Unlock() + return entry.rows, true } } - if read := c.reading[key]; read != nil { - c.mu.Unlock() - <-read.done - return read.rows, read.ok + c.mu.Unlock() + rows, ok := parseTaskIndex(path) + if !ok { + return nil, false } - read := &taskIndexRead{done: make(chan struct{})} - if c.reading == nil { - c.reading = make(map[taskIndexReadKey]*taskIndexRead) + // A FILE THAT CHANGED WHILE IT WAS READ IS NOT KEPT: the stamp taken + // before the reading must still be the file's after it. + if after, err := os.Stat(path); err != nil || !sameStamp(after, stamp) { + return rows, true } - c.reading[key] = read - c.mu.Unlock() - rows, ok := parse(path) c.mu.Lock() defer c.mu.Unlock() - // A changed or unsettled file can serve its waiting readers, but is not kept. - if after, err := os.Stat(path); keep && ok && err == nil && sameStamp(after, stamp) { - c.kept = slices.DeleteFunc(c.kept, func(entry taskIndexKept) bool { return entry.path == path }) - c.kept = append(c.kept, taskIndexKept{path: path, stamp: stamp, rows: rows}) - if len(c.kept) > taskIndexKeptPaths { - discard := len(c.kept) - taskIndexKeptPaths - clear(c.kept[:discard]) - c.kept = c.kept[discard:] - } + c.kept = slices.DeleteFunc(c.kept, func(entry taskIndexKept) bool { return entry.path == path }) + c.kept = append(c.kept, taskIndexKept{path: path, stamp: stamp, rows: rows}) + if len(c.kept) > taskIndexKeptPaths { + discard := len(c.kept) - taskIndexKeptPaths + clear(c.kept[:discard]) + c.kept = c.kept[discard:] } - read.rows, read.ok = rows, ok - delete(c.reading, key) - close(read.done) - return rows, ok + return rows, true } diff --git a/internal/session/task_index_cache_test.go b/internal/session/task_index_cache_test.go index acc61e255b..9bd98327c4 100644 --- a/internal/session/task_index_cache_test.go +++ b/internal/session/task_index_cache_test.go @@ -7,10 +7,7 @@ import ( "path/filepath" "reflect" "runtime" - "sync" - "sync/atomic" "testing" - "testing/synctest" "time" "github.com/Agent-Field/codeaf/internal/crewroute" @@ -165,82 +162,57 @@ func TestTaskIndexReadOwnsNestedFields(t *testing.T) { } } -func TestConcurrentTaskIndexReadersDecodeOneStampOnce(t *testing.T) { - path := filepath.Join(t.TempDir(), taskIndexName) - writeIndexRows(t, path, 0, 1) - settledNow := time.Now().Add(time.Hour) - synctest.Test(t, func(t *testing.T) { - release := make(chan struct{}) - var decodes atomic.Int32 - taskIndexParsed.now = func() time.Time { return settledNow } - taskIndexParsed.parse = func(path string) ([]TaskIndexEntry, bool) { - decodes.Add(1) - <-release - return parseTaskIndex(path) - } - t.Cleanup(func() { taskIndexParsed.now = nil; taskIndexParsed.parse = nil }) - var readers sync.WaitGroup - for range 16 { - readers.Go(func() { - if rows := ReadTaskIndex(path); len(rows) != 1 || rows[0].Title != "task number 0, groomed" { - t.Errorf("concurrent read = %+v, want the indexed row", rows) +func TestTaskIndexReadCapacityCannotReachCachedRows(t *testing.T) { + taskIndexParsed.now = func() time.Time { return time.Now().Add(time.Hour) } + t.Cleanup(func() { taskIndexParsed.now = nil }) + for _, scenario := range []string{"unlimited", "limited", "repeated-node", "tombstones"} { + t.Run(scenario, func(t *testing.T) { + path := filepath.Join(t.TempDir(), taskIndexName) + for i := 1; i <= 3; i++ { + id := fmt.Sprint(i) + if scenario == "repeated-node" && i == 1 { + id = "2" } - }) - } - // All readers are blocked in decoding or waiting for the same decoding. - synctest.Wait() - if got := decodes.Load(); got != 1 { - t.Errorf("decoded one unchanged stamp %d times, want 1", got) - } - close(release) - readers.Wait() - if got := decodes.Load(); got != 1 { - t.Errorf("decoded one unchanged stamp %d times after completion, want 1", got) - } - }) -} - -func TestTaskIndexChangedStampDoesNotJoinOlderRead(t *testing.T) { - path := filepath.Join(t.TempDir(), taskIndexName) - writeIndexRows(t, path, 0, 1) - settledNow := time.Now().Add(time.Hour) - synctest.Test(t, func(t *testing.T) { - release := make(chan struct{}) - var decodes atomic.Int32 - taskIndexParsed.now = func() time.Time { return settledNow } - taskIndexParsed.parse = func(path string) ([]TaskIndexEntry, bool) { - first := decodes.Add(1) == 1 - rows, ok := parseTaskIndex(path) - if first { - <-release + appendTaskIndex(path, TaskIndexEntry{ID: id, SessionID: "chat", Title: "Isolated result", + Files: []string{"original.go"}, + Crew: &crewroute.Decision{Why: "original"}, + VerdictBasis: &plandb.VerdictBasis{Kind: "run"}, + }) } - return rows, ok - } - t.Cleanup(func() { taskIndexParsed.now = nil; taskIndexParsed.parse = nil }) - old := make(chan []TaskIndexEntry, 1) - go func() { old <- ReadTaskIndex(path) }() - synctest.Wait() - writeIndexRows(t, path, 1, 2) - fresh := make(chan []TaskIndexEntry, 1) - go func() { fresh <- ReadTaskIndex(path) }() - synctest.Wait() - select { - case rows := <-fresh: - if len(rows) != 2 { - t.Errorf("new stamp returned %d rows, want 2", len(rows)) + if scenario == "tombstones" { + dir := filepath.Join(filepath.Dir(path), "chat") + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, taskDeletedFile), []byte(`{"1":true}`), 0o600); err != nil { + t.Fatal(err) + } } - default: - t.Error("new stamp waited for the old decoding") - } - close(release) - if rows := <-old; len(rows) != 1 { - t.Errorf("old stamp returned %d rows, want 1", len(rows)) - } - if rows := ReadTaskIndex(path); len(rows) != 2 { - t.Errorf("old completion replaced the new stamp: %+v", rows) - } - if got := decodes.Load(); got != 2 { - t.Errorf("decoded two stamps %d times, want 2", got) - } - }) + want := ReadTaskIndex(path) + got := ReadTaskIndex(path) + if scenario == "limited" { + got = readTaskIndex(path, 1) + } + if len(got) == 0 { + t.Fatal("read no rows, want mutable results") + } + if cap(got) != len(got) { + t.Errorf("returned %d rows with capacity %d; reslicing can expose hidden rows", len(got), cap(got)) + } + for _, row := range got[:cap(got)] { + if len(row.Files) > 0 { + row.Files[0] = "changed.go" + } + if row.Crew != nil { + row.Crew.Why = "changed" + } + if row.VerdictBasis != nil { + row.VerdictBasis.Kind = "reading" + } + } + if next := ReadTaskIndex(path); !reflect.DeepEqual(next, want) { + t.Fatal("mutating rows reachable through capacity contaminated the next ordinary read") + } + }) + } } From 68ae8977f8e62d671364f90a769474c42f647a49 Mon Sep 17 00:00:00 2001 From: Santosh Date: Sun, 25 Oct 2026 09:47:44 -0400 Subject: [PATCH 6/7] fix: isolate returned plan tasks from reused state Deep-copy and seal every mutable Task slice so caller mutations cannot corrupt unchanged plan reads. Add recursive reflection coverage for future mutable fields and sequential and concurrent read isolation tests. Co-Authored-By: Claude Opus 5.5 --- internal/plandb/clone_test.go | 187 ++++++++++++++++++++++++++++++++++ internal/plandb/store.go | 20 ++-- 2 files changed, 199 insertions(+), 8 deletions(-) create mode 100644 internal/plandb/clone_test.go diff --git a/internal/plandb/clone_test.go b/internal/plandb/clone_test.go new file mode 100644 index 0000000000..45f0200093 --- /dev/null +++ b/internal/plandb/clone_test.go @@ -0,0 +1,187 @@ +package plandb + +import ( + "reflect" + "testing" + "time" +) + +func TestCloneTaskCopiesEveryMutableField(t *testing.T) { + var source Task + planTaskFields(t, reflect.ValueOf(&source).Elem(), false) + cloned := cloneTask(&source) + if !reflect.DeepEqual(cloned, &source) { + t.Fatal("clone changed task values") + } + planAssertTaskCopy(t, reflect.ValueOf(&source), reflect.ValueOf(cloned), "Task") + planTaskFields(t, reflect.ValueOf(cloned).Elem(), true) + var want Task + planTaskFields(t, reflect.ValueOf(&want).Elem(), false) + if !reflect.DeepEqual(source, want) { + t.Fatal("mutating clone changed source task") + } +} + +func TestTaskReadIsolatesEveryNestedField(t *testing.T) { + store := planOpen(t, "") + t.Cleanup(func() { _ = store.Close() }) + planAdd(t, store, planSpec("upstream", "Upstream"), TaskSpec{ + ID: "leaf", Title: "Leaf", + Dependencies: []Dependency{{TaskID: "upstream", Kind: DepSuggests}}, + Capabilities: []string{"read", "write", "verify"}, + Resources: []ResourceClaim{{URI: "repo/file", Mode: "read"}}, + ContextInputs: []string{"context"}, Deliverables: []string{"result"}, + EvidenceRequirements: []string{"proof"}, Checks: []string{"go test ./internal/plandb"}, + }) + if _, err := store.Claim("leaf", "worker"); err != nil { + t.Fatal(err) + } + if _, err := store.Done("leaf", "worker", "done", []string{"artifact"}, []string{"evidence"}); err != nil { + t.Fatal(err) + } + if _, err := store.SetVerdictBasis("leaf", VerdictBasis{ + Kind: "run", Runs: []VerdictRun{{Command: "go test ./internal/plandb", ExitCode: 0}}, + Unobserved: []string{"go vet ./internal/plandb"}, + }); err != nil { + t.Fatal(err) + } + // A second handle supplies an independent expectation, without sharing a clone's arrays. + reopened := planReopen(t, store.Path()) + t.Cleanup(func() { _ = reopened.Close() }) + want := reopened.Task("leaf") + got := store.Task("leaf") + if got == nil || !reflect.DeepEqual(got, want) { + t.Fatal("initial task read differs from persisted task") + } + planTaskFields(t, reflect.ValueOf(got).Elem(), true) + for range 3 { + if next := store.Task("leaf"); !reflect.DeepEqual(next, want) { + t.Fatalf("caller mutation changed unchanged task read:\n got: %#v\nwant: %#v", next, want) + } + } + got = store.Task("leaf") + mutated := make(chan struct{}) + go func() { + defer close(mutated) + for range 50 { + planTaskFields(t, reflect.ValueOf(got).Elem(), true) + } + }() + for range 50 { + if next := store.Task("leaf"); !reflect.DeepEqual(next, want) { + t.Error("concurrent caller mutation changed unchanged task read") + break + } + } + <-mutated +} + +// Reflection populates new fields automatically, so an omitted clone cannot hide behind a nil fixture. +// time.Time is an immutable value; its unexported location pointer is safe to share. +func planTaskFields(t *testing.T, value reflect.Value, mutate bool) { + t.Helper() + if value.Type() == reflect.TypeFor[time.Time]() { + moment := time.Unix(1, 0).UTC() + if mutate { + moment = moment.Add(time.Second) + } + value.Set(reflect.ValueOf(moment)) + return + } + switch value.Kind() { + case reflect.Struct: + for i := 0; i < value.NumField(); i++ { + planTaskFields(t, value.Field(i), mutate) + } + case reflect.Slice: + if !mutate { + value.Set(reflect.MakeSlice(value.Type(), 3, 4)) + } + for i := 0; i < value.Len(); i++ { + planTaskFields(t, value.Index(i), mutate) + } + case reflect.Array: + for i := 0; i < value.Len(); i++ { + planTaskFields(t, value.Index(i), mutate) + } + case reflect.Map: + if !mutate { + value.Set(reflect.MakeMap(value.Type())) + key := reflect.New(value.Type().Key()).Elem() + planTaskFields(t, key, false) + entry := reflect.New(value.Type().Elem()).Elem() + planTaskFields(t, entry, false) + value.SetMapIndex(key, entry) + } else { + for _, key := range value.MapKeys() { + entry := reflect.New(value.Type().Elem()).Elem() + entry.Set(value.MapIndex(key)) + planTaskFields(t, entry, true) + value.SetMapIndex(key, entry) + } + } + case reflect.Pointer: + if !mutate { + value.Set(reflect.New(value.Type().Elem())) + } + if !value.IsNil() { + planTaskFields(t, value.Elem(), mutate) + } + case reflect.String: + if mutate { + value.SetString(value.String() + " changed") + } else { + value.SetString("original") + } + case reflect.Bool: + value.SetBool(!value.Bool()) + case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64: + value.SetInt(value.Int() + 1) + case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uintptr: + value.SetUint(value.Uint() + 1) + case reflect.Float32, reflect.Float64: + value.SetFloat(value.Float() + 1) + default: + t.Fatalf("add task field coverage for %s", value.Type()) + } +} + +func planAssertTaskCopy(t *testing.T, source, cloned reflect.Value, path string) { + t.Helper() + if source.Type() == reflect.TypeFor[time.Time]() { + return + } + switch source.Kind() { + case reflect.Struct: + for i := 0; i < source.NumField(); i++ { + planAssertTaskCopy(t, source.Field(i), cloned.Field(i), path+"."+source.Type().Field(i).Name) + } + case reflect.Slice: + if cloned.Cap() != cloned.Len() { + t.Errorf("%s: clone capacity = %d, want length %d", path, cloned.Cap(), cloned.Len()) + } + if source.Pointer() == cloned.Pointer() { + t.Errorf("%s: clone shares slice backing array", path) + } + for i := 0; i < source.Len(); i++ { + planAssertTaskCopy(t, source.Index(i), cloned.Index(i), path+"[]") + } + case reflect.Array: + for i := 0; i < source.Len(); i++ { + planAssertTaskCopy(t, source.Index(i), cloned.Index(i), path+"[]") + } + case reflect.Pointer, reflect.Map: + if source.UnsafePointer() == cloned.UnsafePointer() { + t.Errorf("%s: clone shares %s", path, source.Kind()) + } + if source.Kind() == reflect.Pointer { + planAssertTaskCopy(t, source.Elem(), cloned.Elem(), path+"*") + } else { + for _, key := range source.MapKeys() { + // Mutable map keys need explicit coverage rather than silently sharing their state. + planAssertTaskCopy(t, key, key, path+"[key]") + planAssertTaskCopy(t, source.MapIndex(key), cloned.MapIndex(key), path+"[value]") + } + } + } +} diff --git a/internal/plandb/store.go b/internal/plandb/store.go index a288a217f8..c639f87e7e 100644 --- a/internal/plandb/store.go +++ b/internal/plandb/store.go @@ -12,6 +12,7 @@ import ( "path" "path/filepath" "regexp" + "slices" "sort" "strings" "sync" @@ -3316,14 +3317,17 @@ func cloneTask(task *Task) *Task { return nil } copyTask := *task - copyTask.Dependencies = append([]Dependency(nil), task.Dependencies...) - copyTask.Capabilities = append([]string(nil), task.Capabilities...) - copyTask.Resources = append([]ResourceClaim(nil), task.Resources...) - copyTask.ContextInputs = append([]string(nil), task.ContextInputs...) - copyTask.Deliverables = append([]string(nil), task.Deliverables...) - copyTask.EvidenceRequirements = append([]string(nil), task.EvidenceRequirements...) - copyTask.Artifacts = append([]string(nil), task.Artifacts...) - copyTask.Evidence = append([]string(nil), task.Evidence...) + copyTask.Dependencies = slices.Clip(slices.Clone(task.Dependencies)) + copyTask.Capabilities = slices.Clip(slices.Clone(task.Capabilities)) + copyTask.Resources = slices.Clip(slices.Clone(task.Resources)) + copyTask.ContextInputs = slices.Clip(slices.Clone(task.ContextInputs)) + copyTask.Deliverables = slices.Clip(slices.Clone(task.Deliverables)) + copyTask.EvidenceRequirements = slices.Clip(slices.Clone(task.EvidenceRequirements)) + copyTask.Checks = slices.Clip(slices.Clone(task.Checks)) + copyTask.Artifacts = slices.Clip(slices.Clone(task.Artifacts)) + copyTask.Evidence = slices.Clip(slices.Clone(task.Evidence)) + copyTask.VerdictBasis.Runs = slices.Clip(slices.Clone(task.VerdictBasis.Runs)) + copyTask.VerdictBasis.Unobserved = slices.Clip(slices.Clone(task.VerdictBasis.Unobserved)) return ©Task } From a70b23e09ea4261f40225d0b88dd91a3dca62a44 Mon Sep 17 00:00:00 2001 From: Santosh Date: Sun, 25 Oct 2026 10:05:54 -0400 Subject: [PATCH 7/7] changes: number the engine memory entry Co-Authored-By: Claude Opus 5.5 --- .../{1933-engine-memory-safe.md => 1936-engine-memory-safe.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename docs/changes/unreleased/{1933-engine-memory-safe.md => 1936-engine-memory-safe.md} (100%) diff --git a/docs/changes/unreleased/1933-engine-memory-safe.md b/docs/changes/unreleased/1936-engine-memory-safe.md similarity index 100% rename from docs/changes/unreleased/1933-engine-memory-safe.md rename to docs/changes/unreleased/1936-engine-memory-safe.md