perf: engine reads plans and task indexes once per change

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.

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'
+```

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
+}

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)
+			}
+		})
+	}
+}

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]")
+			}
+		}
+	}
+}

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
 }