diff --git a/docs/changes/unreleased/1936-engine-memory-safe.md b/docs/changes/unreleased/1936-engine-memory-safe.md new file mode 100644 index 0000000000..338d75690e --- /dev/null +++ b/docs/changes/unreleased/1936-engine-memory-safe.md @@ -0,0 +1,26 @@ +--- +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 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 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. + +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..01afd6614f --- /dev/null +++ b/docs/perf/engine-memory-safe.md @@ -0,0 +1,66 @@ +# 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 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, with deep copies of files, crew +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 +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 | + +Review regression gates additionally assert that mutating a returned row's +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 +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 +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|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/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/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/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..b3048df5a0 --- /dev/null +++ b/internal/plandb/refresh_scale_test.go @@ -0,0 +1,81 @@ +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) + 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) + } +} + +// 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..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" @@ -48,6 +49,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 +113,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 +123,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 +1828,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 +1850,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 +2322,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 +2341,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 +2386,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 @@ -3259,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 } 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 1a16ed36bf..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) @@ -517,25 +523,39 @@ 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 } } +// 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. @@ -556,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 @@ -565,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) } @@ -802,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 @@ -820,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, @@ -969,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/recordcache.go b/internal/session/recordcache.go index 345283141e..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 @@ -117,6 +118,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 +129,12 @@ type RecordCacheLimits struct { const ( recordCacheOperations = 2 recordCacheBytes = 256 << 20 - recordCacheEntries = 32 + 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 ) // RecordCacheStats is what a cache holds and has done, for a test or a bench. @@ -155,6 +161,11 @@ type RecordCache struct { retained int64 reserved int64 stats RecordCacheStats + // 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 @@ -166,6 +177,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,11 +205,18 @@ 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 } 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{}, @@ -296,6 +316,8 @@ func (c *RecordCache) leaseLocked(element *list.Element) (Record, func(), error) return Record{}, nil, ErrRecordReadBusy } c.order.MoveToFront(element) + slot.used = c.limits.Now() + c.watchIdleLocked() c.stats.Hits++ c.mu.Unlock() record := cloneRecord(slot.record) @@ -321,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 } @@ -347,9 +369,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 +416,66 @@ 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.expireLocked() +} + +func (c *RecordCache) expireLocked() { + c.cancelSweepLocked() + 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() +} + +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() { + 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.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 // it is out. func (c *RecordCache) dropLocked(element *list.Element) { @@ -411,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 84370bf288..6d3600e47a 100644 --- a/internal/session/recordcache_test.go +++ b/internal/session/recordcache_test.go @@ -663,3 +663,72 @@ 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) + } +} + +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/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 9803fee4f0..e34d291576 100644 --- a/internal/session/task_app.go +++ b/internal/session/task_app.go @@ -262,16 +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). - for _, row := range a.appRunRows() { + // Both projections share the plan gate and each trajectory for this listing. + rows := a.appRunRows() + 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.appRunView(row, cap)) + out = append(out, a.appRunViewIn(row, cap, read)) } - 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 f189e112e9..661a416cad 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, plan, closeStores := a.openPlanReadHandles() + defer closeStores() + 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, read *planTaskRead) 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), read) switch { case row.park != nil || a.taskRouting.isParked(row.ID): view.RoutingState = TaskRoutingBlocked @@ -135,16 +143,18 @@ 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() - for _, store := range stores { +// +// 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, 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 new file mode 100644 index 0000000000..d2feb90428 --- /dev/null +++ b/internal/session/task_app_runs_scale_test.go @@ -0,0 +1,167 @@ +package session + +import ( + "fmt" + "os" + "path/filepath" + "runtime" + "testing" + "weak" + + "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() + + 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() + runtime.ReadMemStats(&after) + t.Logf("listing %d run rows allocated %d bytes", runs, after.TotalAlloc-before.TotalAlloc) + 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) + } + 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_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..0e719b1310 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" @@ -71,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" ) @@ -501,42 +504,55 @@ 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) } sortTaskIndex(rows) if limit > 0 && len(rows) > limit { - rows = rows[:limit] + rows = rows[:limit:limit] } - return rows + for i := range rows { + rows[i] = cloneTaskIndexEntry(rows[i]) + } + return rows[:len(rows):len(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. @@ -562,7 +578,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 == "" { @@ -619,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 @@ -637,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 @@ -661,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 @@ -1075,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 @@ -1317,3 +1333,106 @@ 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 { + now func() time.Time + 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) + } + 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) + } + 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 { + 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 new file mode 100644 index 0000000000..9bd98327c4 --- /dev/null +++ b/internal/session/task_index_cache_test.go @@ -0,0 +1,218 @@ +package session + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "reflect" + "runtime" + "testing" + "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) { + 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() + 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)) + 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) { + // 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) + 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) + 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) + } + 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)) + } +} + +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") + } +} + +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 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" + } + appendTaskIndex(path, TaskIndexEntry{ID: id, SessionID: "chat", Title: "Isolated result", + Files: []string{"original.go"}, + Crew: &crewroute.Decision{Why: "original"}, + VerdictBasis: &plandb.VerdictBasis{Kind: "run"}, + }) + } + 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) + } + } + 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") + } + }) + } +} 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