diff --git a/internal/cli/app.go b/internal/cli/app.go
index e9e8c27..0fc3d98 100644
--- a/internal/cli/app.go
+++ b/internal/cli/app.go
@@ -7,6 +7,8 @@ import (
 	"io"
 	"log/slog"
 	"os"
+	"strings"
+	"time"
 
 	"github.com/labstack/driga/internal/cleanup"
 	"github.com/labstack/driga/internal/config"
@@ -21,8 +23,10 @@ import (
 
 // AppCmd opens the desktop app, or focuses the running one.
 type AppCmd struct {
-	NotificationStdin bool   `hidden:"" help:"Read a notification target from standard input."`
-	Notification      string `hidden:"" help:"Desktop notification event target."`
+	CPUProfile        string        `name:"cpu-profile" type:"path" help:"Write a CPU profile of the app to this file."`
+	ProfileFor        time.Duration `name:"profile-for" default:"2m" help:"How long --cpu-profile records before it stops."`
+	NotificationStdin bool          `hidden:"" help:"Read a notification target from standard input."`
+	Notification      string        `hidden:"" help:"Desktop notification event target."`
 }
 
 var (
@@ -47,16 +51,24 @@ var (
 func AppLaunch(args []string) bool {
 	login := len(args) == 0 && os.Getenv("DRIGA_APP") != "1" && openedAtLogin()
 	app := appLaunch(os.Getenv, login, args)
-	systemLaunch = app && (len(args) == 0 || len(args) > 1 && (args[1] == "--notification" || args[1] == "--notification-stdin") && os.Getenv("DRIGA_APP") == "1")
+	systemLaunch = app && (len(args) == 0 || os.Getenv("DRIGA_APP") == "1")
 	_ = os.Unsetenv("DRIGA_APP")
 	return app
 }
 
 // appLaunch decides app mode: `driga app`, or a bare launch by the system.
 func appLaunch(env func(string) string, openedAtLogin bool, args []string) bool {
-	if len(args) >= 1 && args[0] == "app" && (len(args) == 1 || (args[1] == "--notification" || args[1] == "--notification-stdin")) {
-		return true
+	if len(args) > 0 && args[0] == "app" {
+		if len(args) == 1 {
+			return true
+		}
+		flag, _, _ := strings.Cut(args[1], "=")
+		switch flag {
+		case "--notification", "--notification-stdin", "--cpu-profile", "--profile-for":
+			return true
+		}
 	}
+
 	return len(args) == 0 && (env("DRIGA_APP") == "1" || openedAtLogin)
 }
 
@@ -104,9 +116,23 @@ func (c AppCmd) Run(g *Globals) error {
 		}
 		args = []string{"app", "--notification", c.Notification}
 	}
+	if c.CPUProfile != "" {
+		if len(args) == 0 {
+			args = []string{"app"}
+		}
+		args = append(args, "--cpu-profile", c.CPUProfile, "--profile-for", c.ProfileFor.String())
+	}
 	if handed, err := handOffApp(systemLaunch || c.NotificationStdin, args...); handed || err != nil {
 		return err
 	}
+	if c.CPUProfile != "" {
+		stopProfile, err := startCPUProfile(c.CPUProfile, c.ProfileFor)
+		if err != nil {
+			return err
+		}
+		defer stopProfile()
+	}
+
 	restore := appLogging()
 	defer restore()
 	// Opened from Finder or at login, an error here would reach no one: the
diff --git a/internal/cli/app_test.go b/internal/cli/app_test.go
index e6b52fa..2cf4bae 100644
--- a/internal/cli/app_test.go
+++ b/internal/cli/app_test.go
@@ -37,6 +37,8 @@ func TestAppLaunch(t *testing.T) {
 		{"command", "1", false, []string{"top"}, false},
 		{"command at login", "", true, []string{"top"}, false},
 		{"app", "", false, []string{"app"}, true},
+		{"profile equals", "", false, []string{"app", "--cpu-profile=/tmp/app.pprof"}, true},
+		{"profile", "", false, []string{"app", "--cpu-profile", "/tmp/app.pprof"}, true},
 		{"app help", "", false, []string{"app", "--help"}, false},
 		{"notification", "", false, []string{"app", "--notification", `{}`}, true},
 		{"other value", "0", false, nil, false},
diff --git a/internal/cli/collect_profile.go b/internal/cli/collect_profile.go
index 456749c..59ca166 100644
--- a/internal/cli/collect_profile.go
+++ b/internal/cli/collect_profile.go
@@ -1,14 +1,17 @@
 package cli
 
 import (
+	"encoding/json"
 	"os"
+	"runtime"
 	"runtime/pprof"
 	"sync"
 	"time"
 )
 
-// startCPUProfile records the collector's CPU use into path for d and writes
-// allocations to path + ".alloc", so the live service can be measured in place.
+// startCPUProfile records CPU use into path for d, cumulative allocations to
+// path + ".alloc", a final collected heap to ".heap", and memory/goroutine
+// samples every five seconds to ".mem.jsonl".
 // The returned stop is idempotent and flushes early on shutdown.
 func startCPUProfile(path string, d time.Duration) (func(), error) {
 	f, err := os.Create(path)
@@ -25,13 +28,54 @@ func startCPUProfile(path string, d time.Duration) (func(), error) {
 		alloc.Close()
 		return nil, err
 	}
+	memory, err := os.Create(path + ".mem.jsonl")
+	if err != nil {
+		pprof.StopCPUProfile()
+		f.Close()
+		alloc.Close()
+		return nil, err
+	}
+	done, sampled := make(chan struct{}), make(chan struct{})
+	go func() {
+		defer close(sampled)
+		defer memory.Close()
+		ticker := time.NewTicker(5 * time.Second)
+		defer ticker.Stop()
+		sample := func() {
+			var m runtime.MemStats
+			runtime.ReadMemStats(&m)
+			_ = json.NewEncoder(memory).Encode(struct {
+				At                                      time.Time
+				HeapAlloc, HeapObjects, TotalAlloc, Sys uint64
+				NumGC                                   uint32
+				Goroutines                              int
+			}{time.Now(), m.HeapAlloc, m.HeapObjects, m.TotalAlloc, m.Sys, m.NumGC, runtime.NumGoroutine()})
+		}
+		sample()
+		for {
+			select {
+			case <-ticker.C:
+				sample()
+			case <-done:
+				sample()
+				return
+			}
+		}
+	}()
 	var once sync.Once
 	stop := func() {
 		once.Do(func() {
 			pprof.StopCPUProfile()
+			close(done)
+			<-sampled
 			f.Close()
 			pprof.Lookup("allocs").WriteTo(alloc, 0)
 			alloc.Close()
+			if heap, err := os.Create(path + ".heap"); err == nil {
+				runtime.GC()
+				_ = pprof.WriteHeapProfile(heap)
+				_ = heap.Close()
+			}
 		})
 	}
 	t := time.AfterFunc(d, stop)
diff --git a/internal/cli/collect_profile_test.go b/internal/cli/collect_profile_test.go
index a9cbfb5..984fe66 100644
--- a/internal/cli/collect_profile_test.go
+++ b/internal/cli/collect_profile_test.go
@@ -33,6 +33,12 @@ func TestCPUProfileStopsAfterDuration(t *testing.T) {
 		}
 		time.Sleep(10 * time.Millisecond)
 	}
+	stop()
+	for _, suffix := range []string{".heap", ".mem.jsonl"} {
+		if fi, err := os.Stat(path + suffix); err != nil || fi.Size() == 0 {
+			t.Fatalf("missing %s: %v", suffix, err)
+		}
+	}
 	// A second profile can start, so the first one released the profiler.
 	again, err := startCPUProfile(filepath.Join(t.TempDir(), "again.pprof"), time.Hour)
 	if err != nil {
diff --git a/internal/cli/command_tree_test.go b/internal/cli/command_tree_test.go
index b6f46db..4005926 100644
--- a/internal/cli/command_tree_test.go
+++ b/internal/cli/command_tree_test.go
@@ -22,7 +22,7 @@ func parseCommand(t *testing.T, args ...string) (*Root, *kong.Context, error) {
 }
 func TestCommandTree(t *testing.T) {
 	paths := []string{
-		"", "summary", "flow", "flow 12", "agents", "agents list", "agents show 12", "agents jump 12", "agents kill 12", "projects", "projects list", "projects show app", "projects summarize app other", "projects summarize --all", "inbox", "inbox list", "inbox ack item", "inbox ack --all", "inbox jump item", "orphans", "orphans list", "orphans kill item", "orphans kill --all", "machines", "machines list", "machines show", "machines show local", "machines reset-id", "cleanup", "cleanup list", "cleanup remove item", "cleanup remove --all --kind worktree", "top", "app", "ask what needs me", "mcp", "doctor", "init", "config path", "config show", "config edit", "service run", "service run --serve --listen 127.0.0.1:7411", "service run --scheduled --service-data-dir fixture --service-env DRIGA_CAPTURE_TEXT=titles", "service install", "service start", "service stop", "service restart", "service status", "service logs -f", "service uninstall", "hooks install", "hooks install claude", "hooks uninstall codex", "hooks status", "hooks status all", "fleet peers list", "fleet peers add host:7411 --device-id device -y", "fleet peers remove peer", "fleet viewers list", "fleet viewers revoke viewer", "fleet enroll --allow-viewer viewer --allow-ack -y", "fleet rekey", "fleet id", "fleet status", "version", "hook claude Stop", "feed tail --after-seq 1 -f", "dev agy-schema", "mcp --chat", "mcp --snapshot",
+		"", "summary", "flow", "flow 12", "agents", "agents list", "agents show 12", "agents jump 12", "agents kill 12", "projects", "projects list", "projects show app", "projects summarize app other", "projects summarize --all", "inbox", "inbox list", "inbox ack item", "inbox ack --all", "inbox jump item", "orphans", "orphans list", "orphans kill item", "orphans kill --all", "machines", "machines list", "machines show", "machines show local", "machines reset-id", "cleanup", "cleanup list", "cleanup remove item", "cleanup remove --all --kind worktree", "top", "app", "app --cpu-profile app.pprof --profile-for 1m", "dev serve --cpu-profile app.pprof --profile-for 1m", "ask what needs me", "mcp", "doctor", "init", "config path", "config show", "config edit", "service run", "service run --serve --listen 127.0.0.1:7411", "service run --scheduled --service-data-dir fixture --service-env DRIGA_CAPTURE_TEXT=titles", "service install", "service start", "service stop", "service restart", "service status", "service logs -f", "service uninstall", "hooks install", "hooks install claude", "hooks uninstall codex", "hooks status", "hooks status all", "fleet peers list", "fleet peers add host:7411 --device-id device -y", "fleet peers remove peer", "fleet viewers list", "fleet viewers revoke viewer", "fleet enroll --allow-viewer viewer --allow-ack -y", "fleet rekey", "fleet id", "fleet status", "version", "hook claude Stop", "feed tail --after-seq 1 -f", "dev agy-schema", "mcp --chat", "mcp --snapshot",
 	}
 	for _, path := range paths {
 		t.Run(path, func(t *testing.T) {
diff --git a/internal/cli/dev_serve.go b/internal/cli/dev_serve.go
index eae5cfb..9292c73 100644
--- a/internal/cli/dev_serve.go
+++ b/internal/cli/dev_serve.go
@@ -19,10 +19,20 @@ import (
 // dev). It is the app without its window: the same page, data, Chat and
 // actions, but no native dialogs, menus, tray or notifications.
 type DevServeCmd struct {
-	Addr string `default:"127.0.0.1:47801" help:"Loopback address to listen on."`
+	CPUProfile string        `name:"cpu-profile" type:"path" help:"Write a CPU profile of the app to this file."`
+	ProfileFor time.Duration `name:"profile-for" default:"2m" help:"How long --cpu-profile records before it stops."`
+	Addr       string        `default:"127.0.0.1:47801" help:"Loopback address to listen on."`
 }
 
 func (c DevServeCmd) Run(g *Globals) error {
+	if c.CPUProfile != "" {
+		stopProfile, err := startCPUProfile(c.CPUProfile, c.ProfileFor)
+		if err != nil {
+			return err
+		}
+		defer stopProfile()
+	}
+
 	host, _, err := net.SplitHostPort(c.Addr)
 	if err != nil {
 		return invalidUsage("--addr must be host:port")
diff --git a/internal/collector/pass.go b/internal/collector/pass.go
index d52574d..7d6a72c 100644
--- a/internal/collector/pass.go
+++ b/internal/collector/pass.go
@@ -373,10 +373,12 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err
 			return err
 		}
 		// Inbox grouping needs parent IDs before labels consume the unseen items.
-		parents := derive.LinkParents(ss, idx)
-		if err := explicitParents(tx, ss, parents); err != nil {
+		resolveParents, err := explicitParentResolver(tx, ss)
+		if err != nil {
 			return err
 		}
+		parents := linkPassParents(ss, idx, now)
+		resolveParents(parents)
 		if err := syncInbox(tx, d, ss, parents, hooks, baseline); err != nil {
 			return err
 		}
@@ -415,10 +417,8 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err
 				return err
 			}
 		}
-		parents = derive.LinkParents(ss, idx)
-		if err := explicitParents(tx, ss, parents); err != nil {
-			return err
-		}
+		parents = linkPassParents(ss, idx, now)
+		resolveParents(parents)
 		for i := range ss {
 			ss[i].ParentID = nil
 			if id, ok := parents[ss[i].ID]; ok {
@@ -500,11 +500,10 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err
 			}
 			work.procs = slices.Clone(procs)
 		}
-		for _, s := range ss {
+		for i, s := range ss {
 			before := original[s.ID]
-			current := s
-			current.Completions, current.FilesTouched, current.Events = nil, nil, nil
-			if reflect.DeepEqual(before, current) {
+			if sameSessionRow(before, s) {
+				ss[i] = before
 				continue
 			}
 			if _, err := tx.UpsertSession(s); err != nil {
@@ -534,6 +533,12 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err
 				return err
 			}
 		}
+		work.sessions = make([]model.Session, len(ss))
+		for i, s := range ss {
+			// Keep only row data. Transcript batches must not accumulate in run state.
+			s.Events, s.Completions, s.FilesTouched = nil, nil, nil
+			work.sessions[i] = s
+		}
 		r.Sessions, r.Procs = len(ss), len(procs)
 		return tx.SetMeta("last_pass_at", now.Format(time.RFC3339Nano))
 	})
@@ -546,9 +551,18 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err
 		state.procs = work.procs
 		state.logs = work.logs
 		state.panelSeen = work.panelSeen
+		state.sessions, state.sessionStore, state.sessionVersion = work.sessions, work.sessionStore, work.sessionVersion
+		state.sessionRevision = st.Revision()
 	}
 	if err == nil {
+		revision := st.Revision()
 		err = alertInbox(ctx, d, st)
+		// Alert planning commits one session-neutral metadata transaction. Do not
+		// invalidate our row index for that transaction, but keep invalidation if
+		// another writer (including a notifier callback) committed as well.
+		if err == nil && d.Notifier != nil && !d.snapshot && state.sessionRevision == revision && st.Revision() == revision+1 {
+			state.sessionRevision = revision + 1
+		}
 	}
 	return r, err
 }
@@ -558,52 +572,53 @@ func sessionKey(a, id string) string { return a + "\x00" + id }
 // Keep unresolved adapter references in transactional metadata so a parent
 // first seen after a collector restart can still be linked. Session row IDs
 // are resolved only through the local machine's store view.
-func explicitParents(tx *store.Tx, ss []model.Session, parents map[int64]int64) error {
-	stored, err := tx.Sessions(store.SessionFilter{IncludeEnded: true})
+func explicitParentResolver(tx *store.Tx, ss []model.Session) (func(map[int64]int64), error) {
+	refs, err := tx.MetaPrefix("session-parent:")
 	if err != nil {
-		return err
+		return nil, err
 	}
-	ids := make(map[string]int64, len(stored))
-	for _, s := range stored {
+	ids := make(map[string]int64, len(ss))
+	for _, s := range ss {
 		ids[sessionKey(s.Adapter, s.ExternalID)] = s.ID
 	}
+	type edge struct{ child, parent int64 }
+	var edges []edge
 	for _, s := range ss {
-		key := "session-parent:" + sessionKey(s.Adapter, s.ExternalID)
+		key := sessionKey(s.Adapter, s.ExternalID)
 		ref := s.ParentExternalID
-		if ref != "" {
-			if err := tx.SetMeta(key, ref); err != nil {
-				return err
-			}
-		} else {
-			ref, err = tx.Meta(key)
-			if err != nil && !errors.Is(err, store.ErrNotFound) {
-				return err
-			}
-		}
 		if ref == "" {
-			continue
-		}
-		// Explicit evidence also suppresses an unrelated inferred parent while
-		// the referenced row is still missing.
-		delete(parents, s.ID)
-		id := ids[sessionKey(s.Adapter, ref)]
-		if id == 0 || id == s.ID {
-			continue
-		}
-		seen := map[int64]bool{s.ID: true}
-		cycle := false
-		for p := id; p != 0; p = parents[p] {
-			if seen[p] {
-				cycle = true
-				break
+			ref = refs[key]
+		} else if refs[key] != ref {
+			if err := tx.SetMeta("session-parent:"+key, ref); err != nil {
+				return nil, err
 			}
-			seen[p] = true
 		}
-		if !cycle {
-			parents[s.ID] = id
+		if ref != "" {
+			edges = append(edges, edge{s.ID, ids[sessionKey(s.Adapter, ref)]})
 		}
 	}
-	return nil
+	// Only children with explicit evidence need resolution; both label phases
+	// reuse this index. Pending historical references still resolve on arrival.
+	return func(parents map[int64]int64) {
+		for _, e := range edges {
+			delete(parents, e.child)
+			if e.parent == 0 || e.parent == e.child {
+				continue
+			}
+			seen := map[int64]bool{e.child: true}
+			cycle := false
+			for p := e.parent; p != 0; p = parents[p] {
+				if seen[p] {
+					cycle = true
+					break
+				}
+				seen[p] = true
+			}
+			if !cycle {
+				parents[e.child] = e.parent
+			}
+		}
+	}, nil
 }
 
 func collectFeedEnabled(d Deps, dir string) (bool, error) {
@@ -1188,3 +1203,46 @@ func spoolOffsetPrefix(d Deps) string {
 	}
 	return "spool/"
 }
+
+// Reconsider live and recently ended children. Historical links remain stable;
+// explicit references are resolved against the complete local identity index.
+func linkPassParents(ss []model.Session, idx derive.ProcIndex, now time.Time) map[int64]int64 {
+	candidates := make([]model.Session, 0, 32)
+	historical := make(map[int64]int64)
+	for _, s := range ss {
+		if s.EndedAt == nil || !s.EndedAt.Before(now.Add(-5*time.Minute)) {
+			candidates = append(candidates, s)
+		} else if s.ParentID != nil {
+			historical[s.ID] = *s.ParentID
+		}
+	}
+	for child, parent := range derive.LinkParents(candidates, idx) {
+		historical[child] = parent
+	}
+	return historical
+}
+
+// Compare persisted facts, excluding adapter-only batches and clock-derived
+// mixer presentation. A changed verdict, support, input or CPU sample still
+// writes immediately; elapsed time alone does not create a new observation.
+func sameSessionRow(a, b model.Session) bool {
+	normalize := func(s model.Session) model.Session {
+		s.MachineID = ""
+		s.ParentExternalID = ""
+		s.Events, s.Completions, s.FilesTouched = nil, nil, nil
+		s.Evidence = evidenceDuration.ReplaceAllString(s.Evidence, "<age>")
+		if s.Mixer != nil {
+			m := *s.Mixer
+			m.ObservedAt = time.Time{}
+			m.Evidence = evidenceDuration.ReplaceAllString(m.Evidence, "<age>")
+			m.Rows = slices.Clone(m.Rows)
+			for i := range m.Rows {
+				m.Rows[i].Age = 0
+				m.Rows[i].Strength = 0
+			}
+			s.Mixer = &m
+		}
+		return s
+	}
+	return reflect.DeepEqual(normalize(a), normalize(b))
+}
diff --git a/internal/collector/pass_bench_test.go b/internal/collector/pass_bench_test.go
new file mode 100644
index 0000000..1760fbb
--- /dev/null
+++ b/internal/collector/pass_bench_test.go
@@ -0,0 +1,64 @@
+package collector
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/labstack/driga/internal/adapter"
+	"github.com/labstack/driga/internal/config"
+	"github.com/labstack/driga/internal/model"
+	"github.com/labstack/driga/internal/signatures"
+	"github.com/labstack/driga/internal/store"
+)
+
+func BenchmarkCollectorPass(b *testing.B) {
+	now := time.Date(2026, 10, 30, 12, 0, 0, 0, time.UTC)
+	st, err := store.Open("", store.Memory)
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer st.Close()
+	p := &fakePlatform{}
+	a := &fakeAdapter{name: "claude"}
+	err = st.Tx(context.Background(), func(tx *store.Tx) error {
+		for i := 0; i < 1600; i++ {
+			obs := model.SessionObs{Adapter: "claude", ExternalID: fmt.Sprint(i), Cwd: "/synthetic", StartedAt: now.Add(-24 * time.Hour), LastActivityAt: now.Add(-time.Hour), ConversationClaimed: true}
+			s := model.Session{SessionObs: obs, Label: "ended"}
+			ended := now.Add(-time.Hour)
+			s.EndedAt = &ended
+			if i < 25 {
+				s.EndedAt = nil
+				s.Label = "working"
+				obs.PID = int32(100 + i)
+				obs.ProcStart = obs.StartedAt
+				obs.RegistryStatus = "busy"
+				s.SessionObs = obs
+				a.obs = append(a.obs, obs)
+				p.procs = append(p.procs, model.Proc{PID: obs.PID, StartTime: obs.ProcStart, Cmdline: "claude", Argv: []string{"claude"}, Cwd: obs.Cwd})
+			}
+			if _, err := tx.UpsertSession(s); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		b.Fatal(err)
+	}
+	d := Deps{Cfg: config.Default(), Plat: p, Adapters: []adapter.Adapter{a}, Sigs: signatures.Builtin(), SpoolDir: b.TempDir(), State: NewState(), Now: func() time.Time { return now }}
+	for range 2 {
+		if _, err := Pass(context.Background(), d, st, Plan{Procs: true}); err != nil {
+			b.Fatal(err)
+		}
+	}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for range b.N {
+		now = now.Add(time.Second)
+		if _, err := Pass(context.Background(), d, st, Plan{Procs: true}); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
diff --git a/internal/collector/session_write_changes_test.go b/internal/collector/session_write_changes_test.go
index c194e1e..6c73300 100644
--- a/internal/collector/session_write_changes_test.go
+++ b/internal/collector/session_write_changes_test.go
@@ -1,15 +1,19 @@
 package collector
 
 import (
+	"context"
 	"database/sql"
 	"path/filepath"
 	"testing"
+	"time"
 
+	"github.com/labstack/driga/internal/model"
 	"github.com/labstack/driga/internal/store"
 )
 
 func TestPassUpsertsOnlyChangedSessions(t *testing.T) {
 	d, _, _, a := setup(t)
+	d.Notifier = &fakeNotifier{}
 	path := filepath.Join(t.TempDir(), "db")
 	st, err := store.Open(path, store.ReadWrite)
 	if err != nil {
@@ -29,6 +33,7 @@ func TestPassUpsertsOnlyChangedSessions(t *testing.T) {
 	if _, err = db.Exec("UPDATE writes SET n=0"); err != nil {
 		t.Fatal(err)
 	}
+	d.Now = func() time.Time { return time.Date(2026, 10, 18, 12, 0, 1, 0, time.UTC) }
 	pass(t, d, st, Plan{})
 	var n int
 	if err := db.QueryRow("SELECT n FROM writes").Scan(&n); err != nil {
@@ -37,6 +42,9 @@ func TestPassUpsertsOnlyChangedSessions(t *testing.T) {
 	if n != 0 {
 		t.Fatalf("unchanged session updated %d times", n)
 	}
+	if d.State.sessionRevision != st.Revision() {
+		t.Fatal("session-neutral alert transaction invalidated session cache")
+	}
 	a.obs[0].Title = "changed"
 	pass(t, d, st, Plan{})
 	if err := db.QueryRow("SELECT n FROM writes").Scan(&n); err != nil {
@@ -46,3 +54,83 @@ func TestPassUpsertsOnlyChangedSessions(t *testing.T) {
 		t.Fatalf("changed session writes=%d", n)
 	}
 }
+
+func TestPassSessionCacheSeesLocalAndExternalWrites(t *testing.T) {
+	d, _, _, a := setup(t)
+	path := filepath.Join(t.TempDir(), "db")
+	st, err := store.Open(path, store.ReadWrite)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	pass(t, d, st, Plan{Procs: true})
+	a.obs = nil
+	for _, external := range []bool{false, true} {
+		writer := st
+		if external {
+			writer, err = store.Open(path, store.ReadWrite)
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+		err = writer.Tx(context.Background(), func(tx *store.Tx) error {
+			ss, err := tx.Sessions(store.SessionFilter{IncludeEnded: true})
+			if err != nil {
+				return err
+			}
+			ss[0].Title = "written outside pass"
+			if external {
+				ss[0].Title = "external connection"
+			}
+			_, err = tx.UpsertSession(ss[0])
+			return err
+		})
+		if external {
+			writer.Close()
+		}
+		if err != nil {
+			t.Fatal(err)
+		}
+		pass(t, d, st, Plan{})
+		want := "written outside pass"
+		if external {
+			want = "external connection"
+		}
+		if len(d.State.sessions) != 1 || d.State.sessions[0].Title != want {
+			t.Fatalf("cache failed to invalidate: %+v", d.State.sessions)
+		}
+	}
+}
+
+func TestSessionRowComparisonKeepsFacts(t *testing.T) {
+	at := time.Date(2026, 10, 30, 12, 0, 0, 0, time.UTC)
+	base := model.Session{ID: 1, SessionObs: model.SessionObs{Adapter: "claude", ExternalID: "one"}, Mixer: &model.StateMixer{ObservedAt: at, Input: &model.StateInput{}, Verdict: "idle"}}
+	later := base
+	m := *base.Mixer
+	m.ObservedAt = at.Add(time.Second)
+	later.Mixer = &m
+	if !sameSessionRow(base, later) {
+		t.Fatal("clock alone changed row")
+	}
+	for _, mutate := range []func(*model.Session){
+		func(s *model.Session) { s.Title = "changed" },
+		func(s *model.Session) { s.RegistryUpdatedAt = at },
+		func(s *model.Session) { s.LastPrompt = "changed" },
+		func(s *model.Session) { s.Label = "stuck" },
+		func(s *model.Session) { id := int64(2); s.ParentID = &id },
+		func(s *model.Session) {
+			m := *s.Mixer
+			in := *m.Input
+			in.Proc = &model.Proc{PID: 42, CPUPercent: 7}
+			m.Input = &in
+			s.Mixer = &m
+		},
+		func(s *model.Session) { m := *s.Mixer; m.Agreement++; s.Mixer = &m },
+	} {
+		changed := later
+		mutate(&changed)
+		if sameSessionRow(base, changed) {
+			t.Fatalf("changed fact suppressed: %+v", changed)
+		}
+	}
+}
diff --git a/internal/collector/sources.go b/internal/collector/sources.go
index e93ada3..6cc7710 100644
--- a/internal/collector/sources.go
+++ b/internal/collector/sources.go
@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"log/slog"
 	"path/filepath"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -101,7 +102,17 @@ func collectSources(ctx context.Context, d Deps, st *store.Store, p Plan, state
 	if err != nil {
 		return r, err
 	}
-	ss, err := st.Sessions(store.SessionFilter{IncludeEnded: true})
+	version, err := st.DataVersion(ctx)
+	if err != nil {
+		return r, err
+	}
+	var ss []model.Session
+	if state.sessions != nil && state.sessionStore == st && state.sessionVersion == version && state.sessionRevision == st.Revision() {
+		ss = slices.Clone(state.sessions)
+	} else {
+		ss, err = st.Sessions(store.SessionFilter{IncludeEnded: true})
+	}
+	state.sessionStore, state.sessionVersion = st, version
 	if err != nil {
 		return r, err
 	}
diff --git a/internal/collector/state.go b/internal/collector/state.go
index 77a4af6..738b73b 100644
--- a/internal/collector/state.go
+++ b/internal/collector/state.go
@@ -6,20 +6,25 @@ import (
 	"time"
 
 	"github.com/labstack/driga/internal/model"
+	"github.com/labstack/driga/internal/store"
 )
 
 // State retains sampling and process ownership evidence between passes. It is
 // private to a collector run and is never written to disk.
 type State struct {
-	mu          sync.Mutex
-	prev        map[int32]model.Proc
-	prevAt      time.Time
-	prevMachine model.MachineSample
-	cwds        map[int32]cwdResult
-	cwdsAt      time.Time
-	cwdNext     int
-	spawn       map[string]int64
-	gitFullAt   time.Time
+	mu              sync.Mutex
+	sessions        []model.Session
+	sessionStore    *store.Store
+	sessionVersion  int64
+	sessionRevision uint64
+	prev            map[int32]model.Proc
+	prevAt          time.Time
+	prevMachine     model.MachineSample
+	cwds            map[int32]cwdResult
+	cwdsAt          time.Time
+	cwdNext         int
+	spawn           map[string]int64
+	gitFullAt       time.Time
 	// procs is the process table this run last wrote, or nil before the first
 	// successful scan, when the stored rows are the only source.
 	procs     []model.Proc
@@ -42,7 +47,7 @@ func NewState() *State {
 // replaces prev, spawn, procs and logs wholesale and never writes into them, so
 // they are shared; only the cwd cache is updated in place and is copied.
 func (s *State) copy() *State {
-	n := &State{prev: s.prev, spawn: s.spawn, procs: s.procs, logs: s.logs, cwds: make(map[int32]cwdResult, len(s.cwds))}
+	n := &State{sessions: s.sessions, sessionStore: s.sessionStore, sessionVersion: s.sessionVersion, sessionRevision: s.sessionRevision, prev: s.prev, spawn: s.spawn, procs: s.procs, logs: s.logs, cwds: make(map[int32]cwdResult, len(s.cwds))}
 	n.prevAt, n.cwdsAt, n.gitFullAt = s.prevAt, s.cwdsAt, s.gitFullAt
 	n.cwdNext = s.cwdNext
 	n.prevMachine = s.prevMachine
diff --git a/internal/store/mcp_history.go b/internal/store/mcp_history.go
index aad85de..9a2f46d 100644
--- a/internal/store/mcp_history.go
+++ b/internal/store/mcp_history.go
@@ -128,12 +128,20 @@ func (tx *Tx) ObserveMCP(at time.Time, samples []model.MCPObservation) error {
 // Endpoints are chronological within each group; groups can interleave.
 func (q *querier) ScanMCPHistory(since, until time.Time, visit func(model.MCPObservation) error) error {
 	since = maxTime(since, until.Add(-30*24*time.Hour)).Truncate(time.Minute)
-	rows, err := q.conn.QueryContext(q.ctx, `SELECT `+mcpStateColumns+`,
- CASE WHEN s.resolution=60 THEN (SELECT MIN(at) FROM mcp_clock c WHERE c.machine_id=s.machine_id AND c.at>=MAX(s.first_at,?) AND c.at<=MIN(s.last_at,?)) ELSE s.last_at END,
- CASE WHEN s.resolution=60 THEN (SELECT MAX(at) FROM mcp_clock c WHERE c.machine_id=s.machine_id AND c.at>=MAX(s.first_at,?) AND c.at<=MIN(s.last_at,?)) ELSE s.last_at END
+	// Resolve run endpoints against one bounded shared clock in memory. Two
+	// correlated MIN/MAX subqueries per state repeatedly reread the same pages.
+	clock, err := readMany(q, `SELECT at FROM mcp_clock WHERE machine_id=? AND at>=? AND at<=? ORDER BY at`, []any{q.machineID, unix(since), unix(until)}, func(s scanner) (int64, error) {
+		var at int64
+		err := s.Scan(&at)
+		return at, err
+	})
+	if err != nil {
+		return err
+	}
+	rows, err := q.conn.QueryContext(q.ctx, `SELECT `+mcpStateColumns+`
  FROM mcp_states s JOIN mcp_groups g ON g.id=s.group_id JOIN mcp_labels l ON l.id=s.labels_id
  WHERE s.machine_id=? AND s.first_at>=? AND s.first_at<=? AND s.last_at>=?
- ORDER BY s.first_at,s.group_id`, unix(since), unix(until), unix(since), unix(until), q.machineID, unix(since.Truncate(time.Hour)), unix(until), unix(since))
+ ORDER BY s.first_at,s.group_id`, q.machineID, unix(since.Truncate(time.Hour)), unix(until), unix(since))
 	if err != nil {
 		return err
 	}
@@ -141,24 +149,33 @@ func (q *querier) ScanMCPHistory(since, until time.Time, visit func(model.MCPObs
 	for rows.Next() {
 		var r mcpState
 		var start int64
-		var first, last *int64
-		err := rows.Scan(&r.id, &r.first, &r.last, &r.sample.Host.PID, &start, &r.sample.Key, &r.sample.Host.Name, &r.sample.Host.Exe, &r.sample.Host.Cwd, &r.sample.Name, &r.sample.Count, &r.sample.Footprint, &r.members, &r.sample.ResolutionSeconds, &first, &last)
+		var first, last int64
+		err := rows.Scan(&r.id, &r.first, &r.last, &r.sample.Host.PID, &start, &r.sample.Key, &r.sample.Host.Name, &r.sample.Host.Exe, &r.sample.Host.Cwd, &r.sample.Name, &r.sample.Count, &r.sample.Footprint, &r.members, &r.sample.ResolutionSeconds)
 		if err != nil {
 			return err
 		}
-		if first == nil || last == nil || *last > unix(until) {
+		first, last = r.last, r.last
+		if r.sample.ResolutionSeconds == 60 {
+			lo := sort.Search(len(clock), func(i int) bool { return clock[i] >= r.first })
+			hi := sort.Search(len(clock), func(i int) bool { return clock[i] > r.last })
+			if lo >= hi {
+				continue
+			}
+			first, last = clock[lo], clock[hi-1]
+		}
+		if last > unix(until) {
 			continue
 		}
 		r.sample.Host.StartedAt = fromUnix(start)
 		if err := json.Unmarshal([]byte(r.members), &r.sample.Members); err != nil {
 			return err
 		}
-		r.sample.At = fromUnix(*first)
+		r.sample.At = fromUnix(first)
 		if err := visit(r.sample); err != nil {
 			return err
 		}
-		if *last != *first {
-			r.sample.At = fromUnix(*last)
+		if last != first {
+			r.sample.At = fromUnix(last)
 			if err := visit(r.sample); err != nil {
 				return err
 			}
diff --git a/internal/store/store.go b/internal/store/store.go
index 90f3d6d..3a9096f 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -16,6 +16,7 @@ import (
 	"path/filepath"
 	"runtime"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/google/uuid"
@@ -58,7 +59,8 @@ type querier struct {
 // there would wait for the connection held by the transaction and deadlock.
 type Store struct {
 	querier
-	db *sql.DB
+	db       *sql.DB
+	revision *atomic.Uint64
 }
 
 // Tx provides writes and the same reads as Store, on the active transaction.
@@ -204,7 +206,7 @@ func Open(path string, mode Mode) (_ *Store, retErr error) {
 	if err := ValidateMachineID(id); err != nil {
 		return nil, err
 	}
-	return &Store{querier: querier{conn: db, ctx: ctx, machineID: id, localID: id, dataDir: dataDir}, db: db}, nil
+	return &Store{revision: new(atomic.Uint64), querier: querier{conn: db, ctx: ctx, machineID: id, localID: id, dataDir: dataDir}, db: db}, nil
 }
 
 func randomID() (string, error) {
@@ -252,7 +254,11 @@ func (s *Store) Tx(ctx context.Context, fn func(*Tx) error) error {
 	if err := fn(&Tx{querier{conn: tx, ctx: ctx, machineID: s.machineID, localID: s.localID, dataDir: s.dataDir}}); err != nil {
 		return err
 	}
-	return tx.Commit()
+	if err := tx.Commit(); err != nil {
+		return err
+	}
+	s.revision.Add(1)
+	return nil
 }
 
 // DataVersion changes when another connection commits a write. Compare values
@@ -428,3 +434,7 @@ func checkSchema(ctx context.Context, db *sql.DB, path string) (have, want int64
 	}
 	return have, want, nil
 }
+
+// Revision counts successful transactions on this connection, including local writes
+// which SQLite data_version deliberately excludes.
+func (s *Store) Revision() uint64 { return s.revision.Load() }
diff --git a/internal/web/overview_cache_test.go b/internal/web/overview_cache_test.go
index a001c7d..fe32090 100644
--- a/internal/web/overview_cache_test.go
+++ b/internal/web/overview_cache_test.go
@@ -290,3 +290,48 @@ func TestCleanupSizesReachHeaderWithoutRelisting(t *testing.T) {
 		t.Fatalf("removed candidate still counted: %v", v)
 	}
 }
+
+func TestSnapshotCacheDropsSupersededVersions(t *testing.T) {
+	version := int64(1)
+	s := &Server{source: cacheSource{version: func(context.Context) (int64, error) { return version, nil }, load: func(context.Context, store.Scope) (viewer.Snapshot, error) { return viewer.Snapshot{}, nil }}}
+	for range 40 {
+		if _, err := s.cachedLoad(context.Background(), store.Scope{}, false); err != nil {
+			t.Fatal(err)
+		}
+		if len(s.cache) != 1 {
+			t.Fatalf("retained %d snapshot versions", len(s.cache))
+		}
+		version++
+	}
+}
+
+func TestDashboardPollSharesSnapshotAndInvalidatesOnCommit(t *testing.T) {
+	base, st, _, _ := fixture(t)
+	source := &countingSource{Source: base.source}
+	s, err := New(source, Options{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	poll := func() {
+		t.Helper()
+		for _, spec := range []panel.Dashboard{panel.Header(), panel.Home(),
+			{Name: "Inbox", Panels: []panel.Panel{{ID: "inbox", Title: "Inbox", Viz: "inbox", Source: "inbox"}}},
+			{Name: "Fleet", Panels: []panel.Panel{{ID: "machines", Title: "Machines", Viz: "table", Source: "machines"}, {ID: "sessions", Title: "Sessions", Viz: "table", Source: "sessions"}, {ID: "flow", Title: "Flow", Viz: "table", Source: "flow"}}},
+		} {
+			if _, err := s.runPanels(context.Background(), panel.RunRequest{Dashboard: spec}); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+	poll()
+	if source.loads != 1 {
+		t.Fatalf("poll built %d snapshots", source.loads)
+	}
+	if err := st.Tx(context.Background(), func(tx *store.Tx) error { return tx.SetMeta("cache-test", "changed") }); err != nil {
+		t.Fatal(err)
+	}
+	poll()
+	if source.loads != 2 || len(s.cache) != 1 {
+		t.Fatalf("commit: loads=%d retained=%d", source.loads, len(s.cache))
+	}
+}
diff --git a/internal/web/snapshot.go b/internal/web/snapshot.go
index 9a8f2f0..844fafb 100644
--- a/internal/web/snapshot.go
+++ b/internal/web/snapshot.go
@@ -75,6 +75,13 @@ func (s *Server) cachedLoad(ctx context.Context, scope store.Scope, all bool) (v
 		s.cacheMu.Lock()
 		delete(s.snapshotLoads, key)
 		if load.err == nil {
+			// A new store version invalidates every old overview. Keeping those
+			// snapshots until the size cap retained entire process/event histories.
+			for old, cached := range s.cache {
+				if old.version != version || time.Since(cached.at) >= 15*time.Second {
+					delete(s.cache, old)
+				}
+			}
 			if s.cache == nil || len(s.cache) >= 32 {
 				s.cache = map[snapshotKey]cachedSnapshot{}
 			}
diff --git a/site/src/content/docs/reference/cli.mdx b/site/src/content/docs/reference/cli.mdx
index 14922db..b54579f 100644
--- a/site/src/content/docs/reference/cli.mdx
+++ b/site/src/content/docs/reference/cli.mdx
@@ -480,9 +480,14 @@ driga top [flags]
 Open the Driga desktop app, or focus it.
 
 ```sh
-driga app
+driga app [flags]
 ```
 
+| Flag | Default | Description |
+|---|---|---|
+| `--cpu-profile=<cpu-profile>` |  | Write a CPU profile of the app to this file. |
+| `--profile-for=<profile-for>` | `2m` | How long --cpu-profile records before it stops. |
+
 ## driga ask
 
 Ask a one-shot Chat question through Claude and driga's read-only tools.