magpie/internal/usage/request_page.go

package usage

// The request page keeps compact per-source records, not a second full ledger.
// Unchanged sources are never decoded again. Aggregation streams those records;
// only the requested page is materialized/sorted. Export can still ask for the
// complete ledger explicitly. This cache is disposable and bounded.

import (
	"container/heap"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"io"
	"maps"
	"os"
	"path/filepath"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"
	"unsafe"

	"github.com/yetone/magpie/internal/catalog"
	"github.com/yetone/magpie/internal/provider"
	"github.com/yetone/magpie/internal/sessions"
	"github.com/yetone/magpie/internal/settings"
)

type RequestPage struct {
	Rows              []Row
	Sum               Totals
	Total             int
	Agents, Providers []string
	Purposes          []string // all purposes in the period, independent of the selected filter
	CallerKeys        []Group
	// Through and Direct are the rows the gateway served and the rows read
	// from the agents' own session files, which the gateway never saw. Sum
	// is the two together; the page shows the three beside each other so a
	// reader can tell what magpie carried from what it only read about.
	Through, Direct Totals
	// Accounts are the subscription accounts that answered calls in the
	// period, by provider and account, for the Account filter (#557)
	Accounts []Group
	Bucket   string
	Series   []SeriesPoint
	By       map[string][]Share
	ChartBy  map[string][]Share // the whole period's legend while a day is selected
	// Computers are the rows told apart by the computer they were made on,
	// ThisComputer's and each other's by id, of the rows without the
	// filter's computer, and Names what the others are called: none when no
	// other computer's calls were brought here by sync (#542)
	Computers []Share
	Names     map[string]string
	// Heat is the rows by day, as the chart's filter keeps them: only for
	// the heatmap's own period (HeatmapOf)
	Heat *Heatmap
}

type packedRow struct {
	Time                           time.Time
	Text                           [28]uint32
	Tokens                         [6]int64
	Millis, TTFT, FirstText, Order int64
	Sent, Flow                     int64
	RouteID                        int64
	Cost                           float64
	Status                         int32
	Flags                          uint8
}
type rowChunk struct {
	Archive   []byte
	Locations []*time.Location
	Rows      []packedRow
	Strings   []string
	dict      map[string]uint32
	Source    sessions.CallSource
	Bytes     int64
	Used      uint64
	Count     int
	Latest    time.Time
	// Computer is the other computer whose day this is (#542), by id
	Computer string
}

// rowMsg is the Text of a row's Claude message id, after rowText's
const rowMsg = 27

func rowText(r *Row) [27]*string {
	return [27]*string{&r.Agent, &r.Provider, &r.Host, &r.SessionProvider, &r.SessionAccount, &r.Model, &r.Requested, &r.Served, &r.Effort, &r.Error, &r.ErrType, &r.RequestID, &r.Endpoint, &r.Session, &r.NativeSession, &r.Kind, &r.Source, &r.Via, &r.ProviderKeyID, &r.ProviderKeyName, &r.CallerKeyID, &r.CallerKeyName, &r.Archive, &r.Operation, &r.ProviderAccount, &r.ResponseID, &r.Upstream}
}
func (c *rowChunk) add(r Row, msg string, order int64, failed bool) {
	if c.dict == nil {
		c.dict = map[string]uint32{"": 0}
		c.Strings = []string{""}
	}
	intern := func(s string) uint32 {
		if id, ok := c.dict[s]; ok {
			return id
		}
		id := uint32(len(c.Strings))
		c.dict[s] = id
		c.Strings = append(c.Strings, s)
		c.Bytes += int64(len(s) + 48)
		return id
	}
	p := packedRow{Time: r.Time, Tokens: [6]int64{int64(r.Input), int64(r.Output), int64(r.CacheRead), int64(r.CacheWrite), int64(r.Reasoning), int64(r.CacheWrite1h)}, Millis: r.Millis, TTFT: r.TTFT, FirstText: r.FirstText, Sent: r.Sent, Flow: r.Flow, Order: order, RouteID: r.RouteID, Cost: r.Cost, Status: int32(r.Status)}
	for i, s := range rowText(&r) {
		p.Text[i] = intern(*s)
	}
	p.Text[rowMsg] = intern(msg)
	if r.Priced {
		p.Flags |= 1
	}
	if r.Swapped {
		p.Flags |= 2
	}
	if r.Rejected {
		p.Flags |= 4
	}
	if r.SessionOfficialLogin {
		p.Flags |= 8
	}
	if r.Routed {
		p.Flags |= 32
	}
	if failed {
		p.Flags |= 16
	}
	c.Rows = append(c.Rows, p)
	c.Count = len(c.Rows)
	if r.Time.After(c.Latest) {
		c.Latest = r.Time
	}
	c.Bytes += int64(unsafe.Sizeof(packedRow{}))
}
func (c *rowChunk) row(i int) Row {
	p := &c.Rows[i]
	r := Row{Record: Record{RouteID: p.RouteID, Time: p.Time, Input: int(p.Tokens[0]), Output: int(p.Tokens[1]), CacheRead: int(p.Tokens[2]), CacheWrite: int(p.Tokens[3]), CacheWrite1h: int(p.Tokens[5]), Reasoning: int(p.Tokens[4]), Millis: p.Millis, TTFT: p.TTFT, FirstText: p.FirstText, Sent: p.Sent, Flow: p.Flow, Status: int(p.Status), Rejected: p.Flags&4 != 0, SessionOfficialLogin: p.Flags&8 != 0}, Cost: p.Cost, Priced: p.Flags&1 != 0, Swapped: p.Flags&2 != 0, Routed: p.Flags&32 != 0}
	for i, s := range rowText(&r) {
		*s = c.Strings[p.Text[i]]
	}
	if r.Swapped && (SameSpelled(r.Model, r.Served) || GeminiServing(r.Model, r.Served)) {
		r.Swapped = false // kept before a name spelled otherwise, or Google's serving name, was the same
	}
	r.Computer = c.Computer
	return r
}

type rowRef struct {
	Chunk *rowChunk
	Index int
}

func (r rowRef) row() Row      { return r.Chunk.row(r.Index) }
func (r rowRef) at() time.Time { return r.Chunk.Rows[r.Index].Time }
func refNewer(a, b rowRef) bool {
	if !a.at().Equal(b.at()) {
		return a.at().After(b.at())
	}
	if (a.Chunk.Source.Path == "") != (b.Chunk.Source.Path == "") {
		return a.Chunk.Source.Path == ""
	}
	if a.Chunk.Source.Path != b.Chunk.Source.Path {
		return a.Chunk.Source.Path > b.Chunk.Source.Path
	}
	return a.Chunk.Rows[a.Index].Order > b.Chunk.Rows[b.Index].Order
}

type newestHeap []rowRef

func (h newestHeap) Len() int           { return len(h) }
func (h newestHeap) Less(i, j int) bool { return refNewer(h[j], h[i]) }
func (h newestHeap) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }
func (h *newestHeap) Push(v any)        { *h = append(*h, v.(rowRef)) }
func (h *newestHeap) Pop() any          { a := *h; v := a[len(a)-1]; *h = a[:len(a)-1]; return v }

type pageKey struct {
	Period        Period
	Filter        Filter
	Offset, Limit int
}
type requestIndex struct {
	sync.Mutex
	root, meta, key string
	chunks          map[string]*rowChunk
	gateways        map[*rowChunk]*rowChunk // immutable raw block -> priced block
	pages           map[pageKey]RequestPage
	tick            uint64
}

var requestCache requestIndex

var requestCacheBytes int64 = 24 << 20

func statKey(path string) string {
	s, e := os.Stat(path)
	if e != nil {
		return path + ":missing"
	}
	return fmt.Sprintf("%s:%d:%d", path, s.Size(), s.ModTime().UnixNano())
}
func recordHash(path string, n int64) string {
	f, e := os.Open(path)
	if e != nil {
		return ""
	}
	defer f.Close()
	h := sha256.New()
	if _, e = io.CopyN(h, f, n); e != nil {
		return ""
	}
	return fmt.Sprintf("%x", h.Sum(nil))
}

func QueryPage(p Period, f Filter, offset, limit int) RequestPage {
	return queryPage(p, f, offset, limit, sessions.ReadCallSource)
}
func queryPage(p Period, f Filter, offset, limit int, readSource func(sessions.CallSource) []sessions.Call) RequestPage {
	if limit <= 0 {
		limit = 100
	}
	limit = min(limit, 500)
	offset = max(0, offset)
	// The time is read once, so that the rows, the chart and the day the page
	// is kept for are of one day when midnight falls while it is worked out.
	now := Clock()
	// A replacement reader remains useful to callers supplying synthetic logs.
	if LogCalls != nil {
		return pageFromLedgerAt(p, f, offset, limit, LedgerOfAt(p, Filter{}, now), now)
	}
	sources := sessions.CallSources()
	slices.SortFunc(sources, func(a, b sessions.CallSource) int {
		if n := a.Modified.Compare(b.Modified); n != 0 {
			return n
		}
		return strings.Compare(a.Path, b.Path)
	})
	ids, _ := json.Marshal(provider.SessionIdentities(sessions.CodexDir()))
	renamed, _ := json.Marshal(provider.Renamed())
	cfg := settings.Load()
	prices, _ := json.Marshal(cfg.ModelPrices)
	wires, _ := json.Marshal(cfg.ModelWires)
	meta := fmt.Sprintf("%s|%s|%s|%s|%s|%s", ids, renamed, prices, wires, statKey(provider.Path()), statKey(catalog.CachePath()))
	h := sha256.New()
	for _, current := range provider.All() {
		fmt.Fprintf(h, "%q:%q:%v:%t:%t:%t;", current.ID, current.Where(), current.Catalogs(), current.IsPlugin(), current.Account == nil, current.Key != "")
		fmt.Fprint(h, statKey(catalog.LivePath(current.ID)))
	}
	for _, root := range sessions.DesktopDataDirs() {
		for _, kind := range []string{"local-agent-mode-sessions", "claude-code-sessions"} {
			files, _ := sessions.SessionGlob(filepath.Join(root, kind, "*", "*", "local_*.json"))
			for _, path := range files {
				fmt.Fprint(h, statKey(path))
			}
		}
	}
	meta += fmt.Sprintf("|%x|%s", h.Sum(nil), statKey(catalog.LivePath("antigravity")))
	h.Reset()
	snapshot := logSnapshotFor(true)
	version := snapshot.version
	fmt.Fprint(h, meta, version, now.Format("2006-01-02 MST"))
	for _, s := range sources {
		fmt.Fprintf(h, "%s:%d:%d;", s.Path, s.Size, s.Modified.UnixNano())
	}
	days, someShared := sharedSources(time.Time{})
	for _, s := range days {
		fmt.Fprintf(h, "%s:%d:%d;", s.path, s.size, s.mod.UnixNano())
	}
	key := fmt.Sprintf("%x", h.Sum(nil))
	root := Path() + "|" + catalog.CachePath()
	idx := &requestCache
	idx.Lock()

	if idx.root != root || idx.meta != meta {
		idx.root, idx.meta = root, meta
		idx.chunks = map[string]*rowChunk{}
		idx.gateways = map[*rowChunk]*rowChunk{}
		idx.pages = nil
		idx.key = ""
	}
	if idx.key != key {
		idx.pages = map[pageKey]RequestPage{}
		idx.key = key
	}
	q := pageKey{p, f, offset, limit}
	if page, ok := idx.pages[q]; ok {
		idx.Unlock()
		return page
	}
	idx.tick++
	// Copy only metadata while holding the lock. Published chunks are immutable;
	// reading, decoding, pricing and aggregation use a private snapshot.
	shared := idx
	idx = &requestIndex{root: idx.root, meta: idx.meta, key: idx.key, tick: idx.tick,
		chunks: maps.Clone(idx.chunks), gateways: maps.Clone(idx.gateways)}
	shared.Unlock()
	if snapshot.uncached && len(snapshot.blocks) == 0 {
		snapshot = readLogSnapshot()
	}
	price := pricer()
	priceRow := func(r Record, source string) Row {
		sent := r.Model
		if source == "" {
			sent = provider.SentNameIn(cfg.ModelWires, r.Provider, r.Model, r.Effort)
		}
		row := Row{Record: r, Source: source, Swapped: r.Served != "" && Swapped(sent, r.Served), Routed: GroupRouted(sent, r.Served)}
		if pr := price(r); pr != nil && r.Input+r.Output > 0 {
			row.Priced = true
			row.Cost = r.CostAt(*pr)
		}
		row.Agent = AgentOf(r.Agent)
		return row
	}
	since := p.Since(now)
	gatewaySince := since
	if !since.IsZero() {
		gatewaySince = since.Add(-24 * time.Hour)
	}
	var gateways []*rowChunk
	aliases := provider.Renamed()
	live := make(map[*rowChunk]bool, len(snapshot.blocks))
	for _, packed := range snapshot.blocks {
		live[packed] = true
		if !gatewaySince.IsZero() && packed.Latest.Before(gatewaySince) {
			continue
		}
		cached := idx.gateways[packed]
		c := cached.unpack()
		if c == nil {
			raw := packed.unpack()
			if raw == nil {
				continue
			}
			c = &rowChunk{}
			// Cache the entire block: another period can reuse the same prices.
			// Only the append tail changes; a tariff change clears all blocks.
			for i, pr := range raw.Rows {
				r := raw.row(i).Record
				if id, ok := aliases[r.Provider]; ok {
					r.Provider = id
				}
				c.add(priceRow(r, ""), "", pr.Order, false)
			}
			cached = c.freeze()
		}
		kept := *cached
		kept.Used = idx.tick
		idx.gateways[packed] = &kept
		gateways = append(gateways, c)
	}
	for raw := range idx.gateways {
		if !live[raw] {
			delete(idx.gateways, raw)
		}
	}
	on := map[string]bool{}
	var chunks []*rowChunk
	var resolver *sessionResolver
	read := prefetchSources(sources, func(s sessions.CallSource) bool {
		if !since.IsZero() && s.Modified.Before(since) {
			return false
		}
		c := idx.chunks[s.Path]
		return c == nil || c.Source.Size != s.Size || !c.Source.Modified.Equal(s.Modified)
	}, readSource)
	for i, s := range sources {
		on[s.Path] = true
		if !since.IsZero() && s.Modified.Before(since) {
			continue
		}
		cached := idx.chunks[s.Path]
		c := cached.unpack()
		reused := c != nil && c.Source.Size == s.Size && c.Source.Modified.Equal(s.Modified)
		if !reused {
			if resolver == nil {
				resolver = newSessionResolver([]sessions.Call{{}})
			}
			cs := read(i)
			c = &rowChunk{Source: s, Rows: make([]packedRow, 0, len(cs))}
			for _, call := range cs {
				r := logRecord(call)
				r.SessionAccount = resolver.resolve(call)
				r.SessionOfficialLogin = resolver.officialLogin(call, r.SessionAccount)
				r.SessionProvider = call.Upstream
				c.add(priceRow(r, "log"), call.Msg, call.To, call.Error != "")
			}
			c.Bytes -= int64(32 * len(c.Strings))
			c.dict = nil // local chunks are immutable until the source changes
			idx.chunks[s.Path] = c
		}
		next := *c
		c = &next
		c.Used = idx.tick
		if reused {
			kept := *cached
			kept.Used = idx.tick
			idx.chunks[s.Path] = &kept
		} else {
			idx.chunks[s.Path] = c.pack()
		}
		chunks = append(chunks, c)
	}
	// the other computers' days (#542), a chunk each, read again only when
	// sync brought a new copy
	var others []*rowChunk
	var renames map[string]string
	from := ""
	if !since.IsZero() {
		from = since.AddDate(0, 0, -1).Format(time.DateOnly)
	}
	for _, s := range days {
		on[s.path] = true
		if s.day < from {
			continue
		}
		cached := idx.chunks[s.path]
		c := cached.unpack()
		reused := c != nil && c.Source.Size == s.size && c.Source.Modified.Equal(s.mod)
		if !reused {
			d, _ := readShared(s.path)
			if renames == nil {
				renames = provider.Renamed()
			}
			c = &rowChunk{Source: sessions.CallSource{Path: s.path, Size: s.size, Modified: s.mod}, Computer: s.computer, Rows: make([]packedRow, 0, len(d.Calls))}
			for i, call := range d.Calls {
				r := call.Record
				r.Computer = ""
				if id, ok := renames[r.Provider]; ok {
					r.Provider = id
				}
				c.add(priceRow(r, call.Source), "", int64(i), false)
			}
			c.Bytes -= int64(32 * len(c.Strings))
			c.dict = nil
		}
		next := *c
		c = &next
		c.Used = idx.tick
		if reused {
			kept := *cached
			kept.Used = idx.tick
			idx.chunks[s.path] = &kept
		} else {
			idx.chunks[s.path] = c.pack()
		}
		others = append(others, c)
	}
	for path := range idx.chunks {
		if !on[path] {
			delete(idx.chunks, path)
		}
	}
	var names map[string]string
	if someShared {
		names = sharedNames()
	}
	page := buildRequestBlocks(p, f, offset, limit, now, gateways, chunks, others, names)
	shared.Lock()
	defer shared.Unlock()
	// A query for an older filesystem snapshot may finish after a newer one.
	if shared.root != idx.root || shared.meta != idx.meta || shared.key != idx.key || snapshot.version != version {
		return page
	}
	for path, c := range idx.chunks {
		if prev := shared.chunks[path]; prev == nil || prev.Used <= c.Used {
			shared.chunks[path] = c
		}
	}
	for path := range shared.chunks {
		if !on[path] {
			delete(shared.chunks, path)
		}
	}
	for raw, c := range idx.gateways {
		if prev := shared.gateways[raw]; prev == nil || prev.Used <= c.Used {
			shared.gateways[raw] = c
		}
	}
	for raw := range shared.gateways {
		if !live[raw] {
			delete(shared.gateways, raw)
		}
	}
	if len(shared.pages) >= 16 {
		shared.pages = map[pageKey]RequestPage{}
	}
	if !snapshot.uncached {
		shared.pages[q] = page
	}
	// Raw blocks are also retained by gateway-map keys: charge their bytes once.
	bytes := snapshot.bytes
	budget := int64(requestCacheBytes)
	if snapshot.uncached {
		budget = 0
		shared.pages = map[pageKey]RequestPage{}
	}

	type cachedChunk struct {
		path string
		raw  *rowChunk
		row  *rowChunk
	}
	var kept []cachedChunk
	for path, c := range shared.chunks {
		bytes += c.Bytes
		kept = append(kept, cachedChunk{path: path, row: c})
	}
	for raw, c := range shared.gateways {
		bytes += c.Bytes
		kept = append(kept, cachedChunk{raw: raw, row: c})
	}
	slices.SortFunc(kept, func(a, b cachedChunk) int {
		if a.row.Used < b.row.Used {
			return -1
		}
		if a.row.Used > b.row.Used {
			return 1
		}
		return strings.Compare(a.path, b.path)
	})
	for _, c := range kept {
		if bytes <= budget {
			break
		}
		if c.raw != nil {
			delete(shared.gateways, c.raw)
		} else {
			delete(shared.chunks, c.path)
		}
		bytes -= c.row.Bytes
	}
	return page
}

// localRefs visits first occurrences of Claude message IDs. It stores only
// references for matches, never a full hydrated row for every call.
func visibleLocal(chunks []*rowChunk) map[rowRef]bool {
	duplicates := map[rowRef]bool{}
	seen := map[string]bool{}
	for _, c := range chunks {
		for i, p := range c.Rows {
			msg := c.Strings[p.Text[rowMsg]]
			if msg == "" {
				continue
			}
			if seen[msg] {
				duplicates[rowRef{c, i}] = true
			} else {
				seen[msg] = true
			}
		}
	}
	return duplicates
}

// matchKey narrows candidates before comparing their end times. Request IDs
// are consumed first; fallback matching still requires uniqueness both ways.
// The agent isn't in it: a session's id names the one conversation, and a
// client built on Codex (miyi-helper, say) is logged under its own name by
// the gateway while its Codex session file names Codex, so its calls were
// counted twice, once through magpie and once as not.
type matchKey struct {
	session       string
	tokens        [3]int64
	failed, hasID bool
}

// matchTokens is what a call's tokens are matched by: its input with what it
// wrote to the cache in it, its output and what it read from the cache. The
// two logs needn't split a cache write out of the input alike (#589): a
// Codex that names cache_write_input_tokens beside a gateway record that
// didn't, an older Codex that doesn't beside one that does; the sum is the
// same either way.
func matchTokens(t [6]int64) [3]int64 { return [3]int64{t[0] + t[3], t[1], t[2]} }

type matchEnd struct {
	at    time.Time
	index rowRef
}
type matchGroup struct {
	ends   []matchEnd
	counts []int
}

func matchedLocal(gateway *rowChunk, chunks []*rowChunk, skip map[rowRef]bool, since time.Time) map[rowRef]bool {
	return matchedBlocks([]*rowChunk{gateway}, chunks, skip, since, true)
}

func matchedBlocks(gateways []*rowChunk, chunks []*rowChunk, skip map[rowRef]bool, since time.Time, newestIDs bool) map[rowRef]bool {
	gatewaySince := since
	if !since.IsZero() {
		gatewaySince = since.Add(-24 * time.Hour)
	}
	byID := map[string][]rowRef{}
	for _, gateway := range gateways {
		for i, p := range gateway.Rows {
			if (p.Flags&4 != 0 || p.Text[1] == 0 && p.Status >= 400) || p.Time.Before(gatewaySince) {
				continue
			}
			if id := gateway.Strings[p.Text[11]]; id != "" {
				byID[id] = append(byID[id], rowRef{gateway, i})
			}
		}
	}
	byRequest := map[string][]rowRef{}
	for _, c := range chunks {
		for i, p := range c.Rows {
			ref := rowRef{c, i}
			if skip[ref] || p.Time.Before(since) {
				continue
			}
			if id := c.Strings[p.Text[11]]; len(byID[id]) > 0 {
				byRequest[id] = append(byRequest[id], ref)
			}
		}
	}
	matched, used := map[rowRef]bool{}, map[rowRef]bool{}
	for id, refs := range byRequest {
		if newestIDs {
			slices.SortFunc(refs, func(a, b rowRef) int {
				if refNewer(a, b) {
					return -1
				}
				if refNewer(b, a) {
					return 1
				}
				return 0
			})
		}
		for j, ref := range refs {
			if j >= len(byID[id]) {
				break
			}
			matched[ref], used[byID[id][j]] = true, true
		}
	}
	groups := map[matchKey]*matchGroup{}
	for _, gateway := range gateways {
		for i, p := range gateway.Rows {
			if used[rowRef{gateway, i}] || (p.Flags&4 != 0 || p.Text[1] == 0 && p.Status >= 400) || p.Time.Before(gatewaySince) {
				continue
			}
			session := gateway.Strings[p.Text[14]]
			if session == "" {
				session = gateway.Strings[p.Text[13]]
			}
			if session == "" {
				continue
			}
			key := matchKey{session, matchTokens(p.Tokens), p.Status >= 400 || p.Text[9] != 0, p.Text[11] != 0}
			g := groups[key]
			if g == nil {
				g = &matchGroup{}
				groups[key] = g
			}
			g.ends = append(g.ends, matchEnd{p.Time.Add(time.Duration(p.Millis) * time.Millisecond), rowRef{gateway, i}})
		}
	}
	for _, g := range groups {
		slices.SortFunc(g.ends, func(a, b matchEnd) int { return a.at.Compare(b.at) })
		g.counts = make([]int, len(g.ends)+1)
	}
	candidates := map[rowRef]rowRef{}
	for _, c := range chunks {
		for j, p := range c.Rows {
			ref := rowRef{c, j}
			if skip[ref] || matched[ref] || p.Time.Before(since) {
				continue
			}
			failed := p.Flags&16 != 0
			if p.Tokens[0]+p.Tokens[1]+p.Tokens[2]+p.Tokens[3] == 0 && !failed {
				continue
			}
			key := matchKey{session: c.Strings[p.Text[13]], tokens: matchTokens(p.Tokens), failed: failed}
			count, candidate := 0, (rowRef{})
			// Without a local ID either gateway partition can match. With an ID,
			// only an unnamed gateway call can match (different IDs stay distinct).
			for _, hasID := range []bool{false, true} {
				if hasID && p.Text[11] != 0 {
					continue
				}
				key.hasID = hasID
				g := groups[key]
				if g == nil {
					continue
				}
				lo := sort.Search(len(g.ends), func(i int) bool { return !g.ends[i].at.Before(p.Time.Add(-2 * time.Second)) })
				hi := sort.Search(len(g.ends), func(i int) bool { return g.ends[i].at.After(p.Time.Add(2 * time.Second)) })
				if hi > lo {
					count += hi - lo
					candidate = g.ends[lo].index
					// Range counts preserve ambiguity even for thousands of identical
					// retries, without enumerating their Cartesian product.
					g.counts[lo]++
					g.counts[hi]--
				}
			}
			if count == 1 {
				candidates[ref] = candidate
			}
		}
	}
	unique := map[rowRef]bool{}
	for _, g := range groups {
		count := 0
		for i, end := range g.ends {
			count += g.counts[i]
			if count == 1 {
				unique[end.index] = true
			}
		}
	}
	for ref, i := range candidates {
		if unique[i] {
			matched[ref] = true
		}
	}
	return matched
}
func sharesOf(m map[string]*Share) []Share {
	out := make([]Share, 0, len(m))
	for _, s := range m {
		out = append(out, *s)
	}
	slices.SortFunc(out, func(a, b Share) int {
		if a.AllTokens() != b.AllTokens() {
			return b.AllTokens() - a.AllTokens()
		}
		if a.Calls != b.Calls {
			return b.Calls - a.Calls
		}
		return strings.Compare(a.ID, b.ID)
	})
	return out
}
func buildRequestPage(p Period, f Filter, offset, limit int, gateway *rowChunk, chunks []*rowChunk) RequestPage {
	return buildRequestBlocks(p, f, offset, limit, Clock(), []*rowChunk{gateway}, chunks, nil, nil)
}

// Shared days are never matched to this computer's gateway or session calls.
// now is the moment the page is asked at: its period and its chart's hours or
// days are read from it.
func buildRequestBlocks(p Period, f Filter, offset, limit int, now time.Time, gateways, chunks, others []*rowChunk, names map[string]string) RequestPage {
	skip := visibleLocal(chunks)
	since := p.Since(now)
	matched := matchedBlocks(gateways, chunks, skip, since, true)
	all := append(append(slices.Clone(gateways), chunks...), others...)
	visit := func(fn func(rowRef, Row)) {
		for _, c := range all {
			for i, pr := range c.Rows {
				ref := rowRef{c, i}
				if pr.Time.Before(since) || skip[ref] || matched[ref] {
					continue
				}
				fn(ref, c.row(i))
			}
		}
	}
	out := RequestPage{Rows: []Row{}, Agents: []string{}, Providers: []string{}, Purposes: []string{}, By: map[string][]Share{}}
	agents, providers := map[string]bool{}, map[string]bool{}
	purposes := map[string]bool{}
	callers := map[string]*Group{}
	accounts := map[string]*Group{}
	computers := map[string]*Share{}
	groups := map[string]map[string]*Share{}
	chartGroups := map[string]map[string]*Share{}
	seriesGroups := map[string]map[string]*Share{}
	keys := keyer{}
	for _, d := range Dimensions {
		groups[d] = map[string]*Share{}
		chartGroups[d] = map[string]*Share{}
		seriesGroups[d] = map[string]*Share{}
	}
	// Bound the requested prefix before adding limit: external offsets can
	// reach MaxInt, and a page beyond all sources needs no heap at all.
	maxRows := 0
	for _, c := range all {
		maxRows += len(c.Rows)
	}
	take := 0
	if offset < maxRows {
		take = offset + min(limit, maxRows-offset)
	}
	selected := newestHeap{}
	chartFilter := f
	chartFilter.Day = ""
	// the two source cells' own numbers, counted without the source pick
	viaFilter := f
	viaFilter.Through = ""
	var first time.Time
	visit(func(ref rowRef, r Row) {
		agents[r.Agent] = true
		purposes[PurposeOf(r.Kind)] = true
		if r.Provider != "" {
			providers[r.Provider] = true
		}
		addCallerRow(callers, r)
		addAccountRow(accounts, r)
		keep := f.keepsRow(r)
		if keep {
			out.Total++
			if take > 0 && len(selected) < take {
				heap.Push(&selected, ref)
			} else if take > 0 && refNewer(ref, selected[0]) {
				selected[0] = ref
				heap.Fix(&selected, 0)
			}
			if !r.IsRejected() {
				out.Sum.addRow(r)
			}
		}
		// The two sources are counted with the Via pick cleared, as the
		// dimension groups above are: the cells are what the reader
		// switches between, so each must go on saying its own number
		// while another is picked, or one of them would read 0 and
		// there would be no way back.
		if !r.IsRejected() && viaFilter.keepsRow(r) {
			if r.Source == "log" {
				out.Direct.addRow(r)
			} else {
				out.Through.addRow(r)
			}
		}
		if r.IsRejected() {
			return
		}
		if keep || f.Day != "" && chartFilter.keepsRow(r) {
			for _, d := range Dimensions {
				k := keys.key(r, d)
				s := seriesGroups[d][k]
				if s == nil {
					s = &Share{ID: k}
					seriesGroups[d][k] = s
				}
				s.addRow(r)
			}
			if first.IsZero() || r.Time.Before(first) {
				first = r.Time
			}
		}
		if g := f; names != nil {
			g.Computer = ""
			g.Day = ""
			if g.keepsRow(r) {
				k := r.key("computer")
				if computers[k] == nil {
					computers[k] = &Share{ID: k}
				}
				computers[k].addRow(r)
			}
		}
		for _, d := range Dimensions {
			g := f
			if d == "provider" {
				g.Provider = ""
			}
			if d == "agent" {
				g.Agent = ""
			}
			if d == "model" || d == "modelAt" {
				g.Model = ""
			}
			if d == "modelAt" {
				g.Provider = ""
			}
			if f.Day != "" {
				g.Day = ""
				if g.keepsRow(r) {
					k := keys.key(r, d)
					if chartGroups[d][k] == nil {
						chartGroups[d][k] = &Share{ID: k}
					}
					chartGroups[d][k].addRow(r)
				}
				g.Day = f.Day
			}
			if !g.keepsRow(r) {
				continue
			}
			k := keys.key(r, d)
			s := groups[d][k]
			if s == nil {
				s = &Share{ID: k}
				groups[d][k] = s
			}
			s.addRow(r)
		}
	})
	slices.SortFunc(selected, func(a, b rowRef) int {
		if refNewer(a, b) {
			return -1
		}
		if refNewer(b, a) {
			return 1
		}
		return 0
	})
	for _, ref := range selected[min(offset, len(selected)):] {
		out.Rows = append(out.Rows, ref.row())
	}
	for a := range agents {
		out.Agents = append(out.Agents, a)
	}
	slices.Sort(out.Agents)
	for a := range providers {
		out.Providers = append(out.Providers, a)
	}
	slices.Sort(out.Providers)
	for purpose := range purposes {
		out.Purposes = append(out.Purposes, purpose)
	}
	slices.Sort(out.Purposes)
	out.CallerKeys = callerGroups(callers)
	out.Accounts = callerGroups(accounts)
	for _, d := range Dimensions {
		out.By[d] = sharesOf(groups[d])
	}
	if f.Day != "" {
		out.ChartBy = map[string][]Share{}
		for _, d := range Dimensions {
			out.ChartBy[d] = sharesOf(chartGroups[d])
		}
	}
	out.Computers, out.Names = computerShares(computers, names)
	var base []Point
	chartSince := since
	chartSince, out.Bucket, base = timeline(p, now, first)
	out.Series = make([]SeriesPoint, len(base))
	for i := range base {
		out.Series[i] = SeriesPoint{Point: base[i], By: map[string]map[string]Part{}}
		for _, d := range Dimensions {
			out.Series[i].By[d] = map[string]Part{}
		}
	}
	visit(func(_ rowRef, r Row) {
		if !chartFilter.keepsRow(r) || r.IsRejected() {
			return
		}
		t := r.Time.In(time.Local)
		if t.Before(chartSince) {
			return
		}
		i := bucketIndex(out.Bucket, chartSince, t)
		if i < 0 || i >= len(out.Series) {
			return
		}
		pt := &out.Series[i]
		pt.addRow(r)
		for _, d := range Dimensions {
			k := keys.key(r, d)
			part := pt.By[d][k]
			part.add(r)
			pt.By[d][k] = part
		}
	})
	if p == heatmapPeriod {
		out.Heat = heatmapOf(since, now, func(add func(Row)) {
			visit(func(_ rowRef, r Row) {
				if chartFilter.keepsRow(r) {
					add(r)
				}
			})
		})
	}
	// Match LedgerSeries's top-24 selection on the filtered data, not facets.
	for _, d := range Dimensions {
		kept := map[string]bool{}
		for i, s := range sharesOf(seriesGroups[d]) {
			if i < seriesKeep {
				kept[s.ID] = true
			}
		}
		for i := range out.Series {
			for k := range out.Series[i].By[d] {
				if !kept[k] {
					delete(out.Series[i].By[d], k)
				}
			}
		}
	}
	return out
}
func pageFromLedger(p Period, f Filter, offset, limit int, all Ledgered) RequestPage {
	return pageFromLedgerAt(p, f, offset, limit, all, Clock())
}

// pageFromLedgerAt is the page of a ledger read at now, its chart over the
// period read at the same moment.
func pageFromLedgerAt(p Period, f Filter, offset, limit int, all Ledgered, now time.Time) RequestPage {
	l := all.Filtered(f)
	out := RequestPage{Sum: l.Sum, Total: len(l.Rows), Agents: l.Agents, Providers: l.Providers, Purposes: []string{}, By: map[string][]Share{}}
	// the same split the streaming path makes, so both pages of the ledger
	// agree: "log" is a call read from an agent's own session file, which
	// the gateway never saw; the rest went through it. Counted with the source
	// pick cleared, as the streaming path does, so the three cells stay
	// switchable whichever one is picked.
	viaFilter := f
	viaFilter.Through = ""
	for _, r := range all.Rows {
		if r.IsRejected() || !viaFilter.keepsRow(r) {
			continue
		}
		if r.Source == "log" {
			out.Direct.addRow(r)
		} else {
			out.Through.addRow(r)
		}
	}
	callers, accounts := map[string]*Group{}, map[string]*Group{}
	purposes := map[string]bool{}
	for i := len(all.Rows) - 1; i >= 0; i-- {
		purposes[PurposeOf(all.Rows[i].Kind)] = true
		addCallerRow(callers, all.Rows[i])
		addAccountRow(accounts, all.Rows[i])
	}
	for purpose := range purposes {
		out.Purposes = append(out.Purposes, purpose)
	}
	slices.Sort(out.Purposes)
	out.CallerKeys = callerGroups(callers)
	out.Accounts = callerGroups(accounts)
	offset = min(offset, len(l.Rows))
	out.Rows = l.Rows[offset:min(len(l.Rows), offset+limit)]
	chartRows := l.Rows
	if f.Day != "" {
		chartFilter := f
		chartFilter.Day = ""
		chartRows = all.Filtered(chartFilter).Rows
		out.ChartBy = map[string][]Share{}
	}
	out.Bucket, out.Series = ledgerSeriesAt(p, chartRows, now)
	if p == heatmapPeriod {
		out.Heat = heatmapOf(p.Since(now), now, func(add func(Row)) {
			for _, r := range chartRows {
				add(r)
			}
		})
	}
	for _, d := range Dimensions {
		g := f
		if d == "provider" {
			g.Provider = ""
		}
		if d == "agent" {
			g.Agent = ""
		}
		if d == "model" || d == "modelAt" {
			g.Model = ""
		}
		if d == "modelAt" {
			g.Provider = ""
		}
		out.By[d] = Breakdown(all.Filtered(g).Rows, d)
		if f.Day != "" {
			g.Day = ""
			out.ChartBy[d] = Breakdown(all.Filtered(g).Rows, d)
		}
	}
	if names := sharedNames(); len(names) > 0 {
		g := f
		g.Computer = ""
		g.Day = ""
		computers := map[string]*Share{}
		for _, s := range Breakdown(all.Filtered(g).Rows, "computer") {
			computers[s.ID] = &s
		}
		out.Computers, out.Names = computerShares(computers, names)
	}
	return out
}

// computerShares are the computers' shares, this one and every other kept
// here among them though it made no call in the period; none without names.
func computerShares(m map[string]*Share, names map[string]string) ([]Share, map[string]string) {
	if names == nil {
		return nil, nil
	}
	for _, id := range append([]string{ThisComputer}, slices.Collect(maps.Keys(names))...) {
		if m[id] == nil {
			m[id] = &Share{ID: id}
		}
	}
	return sharesOf(m), names
}

// Caller choices cover the period, not just the current page or selected key.
func addCallerRow(groups map[string]*Group, r Row) {
	if r.CallerKeyID == "" || r.IsRejected() {
		return
	}
	g := groups[r.CallerKeyID]
	if g == nil {
		g = &Group{ID: r.CallerKeyID, CallerKeyID: r.CallerKeyID}
		groups[r.CallerKeyID] = g
	}
	g.CallerKeyName = r.CallerKeyName
	g.addRow(r)
}

// Account choices cover the period too: each account that answered a call,
// by provider, for the Account filter, whatever else is picked.
func addAccountRow(groups map[string]*Group, r Row) {
	who := r.Account()
	if who == "" || r.IsRejected() {
		return
	}
	id := r.Provider + "@" + who
	g := groups[id]
	if g == nil {
		g = &Group{ID: id, Provider: r.Provider, Account: who}
		groups[id] = g
	}
	g.addRow(r)
}

func callerGroups(groups map[string]*Group) []Group {
	out := make([]Group, 0, len(groups))
	for _, g := range groups {
		out = append(out, *g)
	}
	slices.SortFunc(out, func(a, b Group) int {
		if a.Tokens() != b.Tokens() {
			return b.Tokens() - a.Tokens()
		}
		if a.Calls != b.Calls {
			return b.Calls - a.Calls
		}
		return strings.Compare(a.ID, b.ID)
	})
	return out
}

// prefetchSources reads the session files a query has to parse a few at a
// time, ahead of the loop that takes them in order. The first All after
// magpie starts parses every one of them, and one by one that was a wait on
// the disk for each: Requests stayed a skeleton for 40 s over 12k Codex
// sessions. read(i) gives source i's calls; at most a window of them waits
// to be taken, so the history is never all held at once. A source that
// wasn't wanted is read when it is asked for.
func prefetchSources(sources []sessions.CallSource, want func(sessions.CallSource) bool, readSource func(sessions.CallSource) []sessions.Call) func(int) []sessions.Call {
	type slot struct {
		done  chan struct{}
		calls []sessions.Call
	}
	slots := map[int]*slot{}
	var order []int
	for i, s := range sources {
		if want(s) {
			slots[i] = &slot{done: make(chan struct{})}
			order = append(order, i)
		}
	}
	if len(order) < 2 {
		return func(i int) []sessions.Call { return readSource(sources[i]) }
	}
	workers := min(8, len(order))
	window := make(chan struct{}, 4*workers)
	next := make(chan int)
	go func() {
		for _, i := range order {
			window <- struct{}{}
			next <- i
		}
		close(next)
	}()
	for range workers {
		go func() {
			for i := range next {
				s := slots[i]
				s.calls = readSource(sources[i])
				close(s.done)
			}
		}()
	}
	return func(i int) []sessions.Call {
		s := slots[i]
		if s == nil {
			return readSource(sources[i])
		}
		<-s.done
		<-window
		cs := s.calls
		s.calls = nil
		return cs
	}
}