perf: index routing snapshots and quota usage

AUDIT.md

@@ -64,11 +64,17 @@ released agent, or reject a valid completion.
 
 ## P2 — performance
 
-- Cache/parallelise health probes; schedule from one task/worker snapshot.
-  Current routing repeatedly scans tasks and probes candidates per queued task.
-- Index active leases and quota windows. Do not scan the whole event log per
-  availability check or rewrite the full task snapshot after every event.
-- Add 1k/10k-task benchmarks with assignment and append p95 budgets.
+- **Closed 2026-07-30.** Each scheduling pass takes one atomic task/lease
+  snapshot, batches cached (TTL) reachability probes concurrently, and
+  evaluates candidate availability once. It no longer probes candidates or
+  scans active tasks once per queued task.
+- **Closed 2026-07-30.** Active leases and per-harness, time-ordered quota
+  receipts are projection indexes. Availability uses indexed rolling-window
+  sums rather than decoding the event log; the task snapshot is a one-time
+  disposable compatibility cache rather than a full rewrite on every append.
+- **Closed 2026-07-30.** `BenchmarkAssignPending{1K,10K}` and
+  `BenchmarkAppend{1K,10K}` report and enforce p95 budgets, with durable
+  fsync cost included in their respective paths.
 
 ## Delivery order
 

internal/registry/registry.go

@@ -227,6 +227,20 @@ func (TCPReachability) Reachable(address string, timeout time.Duration) bool {
 	return true
 }
 func (r Registry) Candidates(project string, check Reachability, timeout time.Duration) ([]Herdr, error) {
+	return r.candidates(project, func(h Herdr) bool {
+		address := r.Endpoint(h)
+		return check == nil || check.Reachable(address, timeout)
+	})
+}
+
+// CandidatesWithHealth filters candidates using a health snapshot gathered by
+// the router. Keeping probe execution outside this method lets a scheduling
+// pass probe each herdr once, in parallel, instead of once per queued task.
+func (r Registry) CandidatesWithHealth(project string, healthy map[string]bool) ([]Herdr, error) {
+	return r.candidates(project, func(h Herdr) bool { return healthy[h.ID] })
+}
+
+func (r Registry) candidates(project string, include func(Herdr) bool) ([]Herdr, error) {
 	p, ok := r.projects[project]
 	if !ok {
 		return nil, ErrUnknownProject
@@ -237,20 +251,25 @@ func (r Registry) Candidates(project string, check Reachability, timeout time.Du
 	}
 	out := []Herdr{}
 	for _, h := range r.herdrs {
-		if !allowed[h.MachineID] {
+		if !allowed[h.MachineID] || !include(h) {
 			continue
 		}
-		addr := h.Address
-		if addr == "" {
-			addr = r.machines[h.MachineID].Address
-			if host, _, err := net.SplitHostPort(addr); err == nil {
-				addr = net.JoinHostPort(host, defaultHerdrPort)
-			}
-		}
-		if check == nil || check.Reachable(addr, timeout) {
-			out = append(out, h)
-		}
+		out = append(out, h)
 	}
 	sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
 	return out, nil
 }
+
+// Endpoint resolves the herdr-specific address or its machine's default
+// herdr endpoint. It is exposed so health checks can be batched independently
+// from project routing.
+func (r Registry) Endpoint(h Herdr) string {
+	if h.Address != "" {
+		return h.Address
+	}
+	address := r.machines[h.MachineID].Address
+	if host, _, err := net.SplitHostPort(address); err == nil {
+		return net.JoinHostPort(host, defaultHerdrPort)
+	}
+	return address
+}

internal/router/benchmark_test.go

@@ -0,0 +1,82 @@
+package router
+
+import (
+	"encoding/json"
+	"fmt"
+	"orchestra/internal/authz"
+	"orchestra/internal/domain"
+	"orchestra/internal/registry"
+	"orchestra/internal/store"
+	"sort"
+	"testing"
+	"time"
+)
+
+// These are p95 budgets for a complete scheduling pass on an otherwise idle
+// coordinator. The benchmark deliberately excludes fixture construction and
+// disk population, which is measured separately by store's append benchmark.
+const (
+	// Assignment retains one fsynced TaskLeased event per task; these budgets
+	// include that durability cost on CI-backed storage.
+	assignmentP95Budget1K  = 6 * time.Second
+	assignmentP95Budget10K = time.Minute
+)
+
+func BenchmarkAssignPending1K(b *testing.B) { benchmarkAssignPending(b, 1_000, assignmentP95Budget1K) }
+func BenchmarkAssignPending10K(b *testing.B) {
+	benchmarkAssignPending(b, 10_000, assignmentP95Budget10K)
+}
+
+func benchmarkAssignPending(b *testing.B, tasks int, budget time.Duration) {
+	b.Helper()
+	samples := make([]time.Duration, 0, b.N)
+	b.ResetTimer()
+	for sample := 0; sample < b.N; sample++ {
+		b.StopTimer()
+		s, err := store.Open(b.TempDir())
+		if err != nil {
+			b.Fatal(err)
+		}
+		for i := 0; i < tasks; i++ {
+			id := fmt.Sprintf("task-%d", i)
+			payload, _ := json.Marshal(map[string]any{"source": "benchmark", "external_id": id, "project": "p"})
+			if err := s.Append(domain.Event{ID: id, Type: "TaskCreated", TaskID: id, Version: 1, Payload: payload, Surface: string(authz.System)}); err != nil {
+				b.Fatal(err)
+			}
+		}
+		r, err := registry.New(registry.Config{
+			Projects: []registry.Project{{ID: "p", MachineAffinity: []string{"m"}}},
+			Machines: []registry.Machine{{ID: "m", Address: "unused"}},
+			Herdrs:   []registry.Herdr{{ID: "h", MachineID: "m"}}, // unlimited concurrency
+		})
+		if err != nil {
+			b.Fatal(err)
+		}
+		router := Router{Store: s, Registry: r}
+		b.StartTimer()
+		started := time.Now()
+		assigned, err := router.AssignPending()
+		samples = append(samples, time.Since(started))
+		b.StopTimer()
+		if err != nil {
+			b.Fatal(err)
+		}
+		if len(assigned) != tasks {
+			b.Fatalf("assigned %d tasks, want %d", len(assigned), tasks)
+		}
+	}
+	p95 := durationP95(samples)
+	b.ReportMetric(float64(p95.Microseconds()), "assign_p95_us")
+	if p95 > budget {
+		b.Fatalf("assignment p95 %s exceeds budget %s for %d tasks", p95, budget, tasks)
+	}
+}
+
+func durationP95(samples []time.Duration) time.Duration {
+	if len(samples) == 0 {
+		return 0
+	}
+	sorted := append([]time.Duration(nil), samples...)
+	sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
+	return sorted[(len(sorted)*95+99)/100-1]
+}

internal/router/router.go

@@ -10,6 +10,7 @@ import (
 	"orchestra/internal/store"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -54,26 +55,7 @@ type QuotaAvailability struct {
 }
 
 func (q QuotaAvailability) sumSince(harnessID string, since time.Time) (float64, bool) {
-	var consumed float64
-	known := false
-	for _, e := range q.Store.Events(0) {
-		if e.Type != "QuotaReported" || e.At.Before(since) {
-			continue
-		}
-		var p struct {
-			HarnessID string  `json:"harness_id"`
-			Consumed  float64 `json:"consumed"`
-			Known     *bool   `json:"known"`
-		}
-		if json.Unmarshal(e.Payload, &p) == nil && p.HarnessID == harnessID && p.Consumed >= 0 {
-			if p.Known != nil && !*p.Known {
-				return 0, false
-			}
-			known = true
-			consumed += p.Consumed
-		}
-	}
-	return consumed, known
+	return q.Store.QuotaSince(harnessID, since)
 }
 
 func (q QuotaAvailability) Available(h registry.Herdr) bool {
@@ -116,6 +98,18 @@ type Router struct {
 	Retry        RetryPolicy
 	Now          func() time.Time
 	OnLease      func(domain.Event) error
+	// HealthTTL bounds re-use of a successful or failed reachability probe.
+	// A scheduling pass always has a coherent health snapshot; this cache also
+	// prevents bursts of TaskCreated events from repeatedly dialing the same
+	// herdr between passes.
+	HealthTTL time.Duration
+	healthMu  sync.Mutex
+	health    map[string]healthProbe
+}
+
+type healthProbe struct {
+	reachable bool
+	until     time.Time
 }
 
 func (r *Router) init() {
@@ -125,6 +119,9 @@ func (r *Router) init() {
 	if r.Now == nil {
 		r.Now = time.Now
 	}
+	if r.HealthTTL <= 0 {
+		r.HealthTTL = 5 * time.Second
+	}
 }
 
 // HandleEvent evaluates the sink after creation and after a lease is freed.
@@ -141,13 +138,22 @@ func (r *Router) AssignPending() ([]domain.Event, error) {
 	if r.Store == nil {
 		return nil, errors.New("router: store required")
 	}
+	now := r.Now()
+	snapshot := r.Store.SchedulingSnapshot()
 	var queued []domain.Task
-	for _, t := range r.Store.Tasks() {
-		if t.State == domain.StateQueued && (t.NextRetryAt.IsZero() || !r.Now().Before(t.NextRetryAt)) {
+	for _, t := range snapshot.Tasks {
+		if t.State == domain.StateQueued && (t.NextRetryAt.IsZero() || !now.Before(t.NextRetryAt)) {
 			queued = append(queued, t)
 		}
 	}
-	sort.SliceStable(queued, func(i, j int) bool { return importance(queued[i], r.Now()).Before(importance(queued[j], r.Now())) })
+	sort.SliceStable(queued, func(i, j int) bool { return importance(queued[i], now).Before(importance(queued[j], now)) })
+	health := r.healthSnapshot(now)
+	candidates := make(map[string][]registry.Herdr)
+	availability := make(map[string]bool)
+	availabilityKnown := make(map[string]bool)
+	projectSupport := make(map[string]bool)
+	projectSupportKnown := make(map[string]bool)
+	activeLeases := snapshot.ActiveLeases
 	var out []domain.Event
 	for _, t := range queued {
 		if r.Retry.MaxAttempts > 0 && t.Attempt >= r.Retry.MaxAttempts {
@@ -158,16 +164,34 @@ func (r *Router) AssignPending() ([]domain.Event, error) {
 			out = append(out, e)
 			continue
 		}
-		cs, err := r.Registry.Candidates(t.Project, r.Reachability, r.Timeout)
-		if err != nil {
-			continue
+		cs, found := candidates[t.Project]
+		if !found {
+			var err error
+			cs, err = r.Registry.CandidatesWithHealth(t.Project, health)
+			if err != nil {
+				continue
+			}
+			candidates[t.Project] = cs
 		}
 		for _, h := range cs {
-			projectOK := true
-			if projects, ok := r.Availability.(ProjectAvailability); ok {
-				projectOK = projects.Supports(h, t.Project)
+			projectKey := h.ID + "\x00" + t.Project
+			projectOK, checked := projectSupport[projectKey], projectSupportKnown[projectKey]
+			if !checked {
+				projectOK = true
+				if projects, ok := r.Availability.(ProjectAvailability); ok {
+					projectOK = projects.Supports(h, t.Project)
+				}
+				projectSupport[projectKey], projectSupportKnown[projectKey] = projectOK, true
 			}
-			if !projectOK || !matches(t.Capability, h.Capabilities) || !r.Availability.Available(h) || occupied(r.Store, h.ID, h.Concurrency) {
+			if !projectOK || !matches(t.Capability, h.Capabilities) {
+				continue
+			}
+			available, checked := availability[h.ID], availabilityKnown[h.ID]
+			if !checked {
+				available = r.Availability.Available(h)
+				availability[h.ID], availabilityKnown[h.ID] = available, true
+			}
+			if !available || occupiedCount(activeLeases[h.ID], h.Concurrency) {
 				continue
 			}
 			e, err := r.Store.Lease(t.ID, h.ID, 30*time.Minute)
@@ -175,6 +199,7 @@ func (r *Router) AssignPending() ([]domain.Event, error) {
 				continue
 			}
 			out = append(out, e)
+			activeLeases[h.ID]++
 			if r.OnLease != nil {
 				if err := r.OnLease(e); err != nil {
 					return out, err
@@ -198,17 +223,60 @@ func matches(need, have []string) bool {
 	}
 	return true
 }
-func occupied(s *store.Store, id string, limit int) bool {
-	if limit <= 0 {
-		return false
-	}
-	n := 0
-	for _, t := range s.Tasks() {
-		if (t.State == domain.StateLeased || t.State == domain.StateNeedsAttention) && t.Lease != nil && t.Lease.HarnessID == id {
-			n++
+func occupiedCount(count, limit int) bool {
+	return limit > 0 && count >= limit
+}
+
+func (r *Router) healthSnapshot(now time.Time) map[string]bool {
+	herdrs := r.Registry.Herdrs()
+	health := make(map[string]bool, len(herdrs))
+	if r.Reachability == nil {
+		for _, h := range herdrs {
+			health[h.ID] = true
 		}
+		return health
 	}
-	return n >= limit
+	type target struct {
+		key     string
+		id      string
+		address string
+	}
+	var probes []target
+	r.healthMu.Lock()
+	if r.health == nil {
+		r.health = make(map[string]healthProbe)
+	}
+	for _, h := range herdrs {
+		key := h.ID + "\x00" + r.Registry.Endpoint(h)
+		if cached, ok := r.health[key]; ok && now.Before(cached.until) {
+			health[h.ID] = cached.reachable
+			continue
+		}
+		probes = append(probes, target{key: key, id: h.ID, address: r.Registry.Endpoint(h)})
+	}
+	r.healthMu.Unlock()
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	for _, target := range probes {
+		target := target
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			reachable := r.Reachability.Reachable(target.address, r.Timeout)
+			mu.Lock()
+			health[target.id] = reachable
+			mu.Unlock()
+		}()
+	}
+	wg.Wait()
+	if len(probes) > 0 {
+		r.healthMu.Lock()
+		for _, target := range probes {
+			r.health[target.key] = healthProbe{reachable: health[target.id], until: now.Add(r.HealthTTL)}
+		}
+		r.healthMu.Unlock()
+	}
+	return health
 }
 func importance(t domain.Task, now time.Time) time.Time {
 	if t.Due != nil {

internal/router/router_test.go

@@ -6,6 +6,7 @@ import (
 	"orchestra/internal/domain"
 	"orchestra/internal/registry"
 	"orchestra/internal/store"
+	"sync/atomic"
 	"testing"
 	"time"
 )
@@ -14,6 +15,26 @@ type reachable struct{}
 
 func (reachable) Reachable(string, time.Duration) bool { return true }
 
+type countedReachability struct {
+	calls atomic.Int32
+	delay time.Duration
+}
+
+func (r *countedReachability) Reachable(string, time.Duration) bool {
+	r.calls.Add(1)
+	if r.delay > 0 {
+		time.Sleep(r.delay)
+	}
+	return true
+}
+
+type countedAvailability struct{ calls atomic.Int32 }
+
+func (a *countedAvailability) Available(registry.Herdr) bool {
+	a.calls.Add(1)
+	return true
+}
+
 type projectAvailability struct{ projects map[string]bool }
 
 func (p projectAvailability) Available(registry.Herdr) bool { return true }
@@ -63,6 +84,79 @@ func TestAssignsByAffinityCapabilityAndConcurrency(t *testing.T) {
 	}
 }
 
+func TestAssignPendingSnapshotsAndCachesHealth(t *testing.T) {
+	s, err := store.Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	r, err := registry.New(registry.Config{
+		Projects: []registry.Project{{ID: "p", MachineAffinity: []string{"m"}}},
+		Machines: []registry.Machine{{ID: "m", Address: "unused"}},
+		Herdrs: []registry.Herdr{
+			{ID: "h1", MachineID: "m", Address: "h1:1"},
+			{ID: "h2", MachineID: "m", Address: "h2:1"},
+			{ID: "h3", MachineID: "m", Address: "h3:1"},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 10; i++ {
+		id := domain.NewID()
+		payload, _ := json.Marshal(map[string]any{"source": "test", "external_id": id, "project": "p"})
+		if err := s.Append(domain.Event{ID: id, TaskID: id, Type: "TaskCreated", Version: 1, Payload: payload, Surface: string(authz.System)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	reach := &countedReachability{}
+	availability := &countedAvailability{}
+	rt := Router{Store: s, Registry: r, Reachability: reach, Availability: availability, HealthTTL: time.Minute}
+	if got, err := rt.AssignPending(); err != nil || len(got) != 10 {
+		t.Fatalf("assigned=%d, err=%v", len(got), err)
+	}
+	if got := reach.calls.Load(); got != 3 {
+		t.Fatalf("health probes=%d, want one per herdr rather than one per task", got)
+	}
+	if got := availability.calls.Load(); got != 1 {
+		t.Fatalf("availability checks=%d, want one for selected harness", got)
+	}
+	// The next pass has no eligible work, but still obtains a health snapshot.
+	// It must use the TTL cache rather than re-dial all three herdrs.
+	if _, err := rt.AssignPending(); err != nil {
+		t.Fatal(err)
+	}
+	if got := reach.calls.Load(); got != 3 {
+		t.Fatalf("health cache missed: probes=%d, want 3", got)
+	}
+}
+
+func TestHealthProbesRunInParallel(t *testing.T) {
+	s, err := store.Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	r, err := registry.New(registry.Config{
+		Projects: []registry.Project{{ID: "p", MachineAffinity: []string{"m"}}},
+		Machines: []registry.Machine{{ID: "m", Address: "unused"}},
+		Herdrs:   []registry.Herdr{{ID: "h1", MachineID: "m", Address: "h1:1"}, {ID: "h2", MachineID: "m", Address: "h2:1"}, {ID: "h3", MachineID: "m", Address: "h3:1"}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	payload, _ := json.Marshal(map[string]any{"source": "test", "external_id": "parallel", "project": "p"})
+	if err := s.Append(domain.Event{ID: "parallel", TaskID: "parallel", Type: "TaskCreated", Version: 1, Payload: payload, Surface: string(authz.System)}); err != nil {
+		t.Fatal(err)
+	}
+	reach := &countedReachability{delay: 100 * time.Millisecond}
+	started := time.Now()
+	if _, err := (&Router{Store: s, Registry: r, Reachability: reach}).AssignPending(); err != nil {
+		t.Fatal(err)
+	}
+	if elapsed := time.Since(started); elapsed > 250*time.Millisecond {
+		t.Fatalf("health probes took %s; expected parallel probes, not ~300ms serial", elapsed)
+	}
+}
+
 func TestAssignPendingRequiresWorkerProjectSupport(t *testing.T) {
 	s, err := store.Open(t.TempDir())
 	if err != nil {

internal/store/benchmark_test.go

@@ -0,0 +1,60 @@
+package store
+
+import (
+	"encoding/json"
+	"fmt"
+	"orchestra/internal/authz"
+	"orchestra/internal/domain"
+	"sort"
+	"testing"
+	"time"
+)
+
+// Append includes the fsynced event-log write and sparse projection-cache
+// checkpoint. The budgets catch accidental return to an O(tasks) snapshot
+// rewrite on every event while remaining practical on CI-backed storage.
+const (
+	appendP95Budget1K  = 10 * time.Second
+	appendP95Budget10K = 90 * time.Second
+)
+
+func BenchmarkAppend1K(b *testing.B)  { benchmarkAppend(b, 1_000, appendP95Budget1K) }
+func BenchmarkAppend10K(b *testing.B) { benchmarkAppend(b, 10_000, appendP95Budget10K) }
+
+func benchmarkAppend(b *testing.B, events int, budget time.Duration) {
+	b.Helper()
+	samples := make([]time.Duration, 0, b.N)
+	b.ResetTimer()
+	for sample := 0; sample < b.N; sample++ {
+		b.StopTimer()
+		s, err := Open(b.TempDir())
+		if err != nil {
+			b.Fatal(err)
+		}
+		b.StartTimer()
+		started := time.Now()
+		for i := 0; i < events; i++ {
+			id := fmt.Sprintf("task-%d", i)
+			payload, _ := json.Marshal(map[string]any{"source": "benchmark", "external_id": id, "project": "p"})
+			if err := s.Append(domain.Event{ID: id, Type: "TaskCreated", TaskID: id, Version: 1, Payload: payload, Surface: string(authz.System)}); err != nil {
+				b.Fatal(err)
+			}
+		}
+		samples = append(samples, time.Since(started))
+		b.StopTimer()
+	}
+	p95 := appendDurationP95(samples)
+	b.ReportMetric(float64(p95.Microseconds()), "append_p95_us")
+	if p95 > budget {
+		b.Fatalf("append p95 %s exceeds budget %s for %d events", p95, budget, events)
+	}
+}
+
+func appendDurationP95(samples []time.Duration) time.Duration {
+	if len(samples) == 0 {
+		return 0
+	}
+	sorted := append([]time.Duration(nil), samples...)
+	sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
+	return sorted[(len(sorted)*95+99)/100-1]
+}

internal/store/store.go

@@ -9,10 +9,26 @@ import (
 	"orchestra/internal/domain"
 	"os"
 	"path/filepath"
+	"sort"
 	"sync"
 	"time"
 )
 
+// QuotaUsage is an indexed native-usage receipt. It is deliberately kept
+// separate from the event log: event history remains authoritative, while
+// routing must not decode every historical event for each availability check.
+type QuotaUsage struct {
+	At       time.Time
+	Consumed float64
+	Known    bool
+}
+
+type quotaIndex struct {
+	records       []QuotaUsage // ordered by At
+	prefix        []float64
+	unknownPrefix []int
+}
+
 type Store struct {
 	mu       sync.Mutex
 	path     string
@@ -20,15 +36,19 @@ type Store struct {
 	events   []domain.Event
 	tasks    map[string]domain.Task
 	external map[string]string
-	snapshot string
-	seq      uint64
+	// activeLeases is the routing occupancy index. Needs-attention retains a
+	// fenced owner, so it counts as active until that lease is released.
+	activeLeases map[string]map[string]struct{}
+	quota        map[string]quotaIndex
+	snapshot     string
+	seq          uint64
 }
 
 func Open(dir string) (*Store, error) {
 	if err := os.MkdirAll(dir, 0755); err != nil {
 		return nil, err
 	}
-	s := &Store{path: filepath.Join(dir, "events.jsonl"), cas: filepath.Join(dir, "cas"), snapshot: filepath.Join(dir, "snapshot.json"), tasks: map[string]domain.Task{}, external: map[string]string{}}
+	s := &Store{path: filepath.Join(dir, "events.jsonl"), cas: filepath.Join(dir, "cas"), snapshot: filepath.Join(dir, "snapshot.json"), tasks: map[string]domain.Task{}, external: map[string]string{}, activeLeases: map[string]map[string]struct{}{}, quota: map[string]quotaIndex{}}
 	if err := os.MkdirAll(s.cas, 0755); err != nil {
 		return nil, err
 	}
@@ -58,7 +78,8 @@ func Open(dir string) (*Store, error) {
 			if t, ok := s.tasks[e.TaskID]; ok && e.Version != t.Version+1 {
 				return nil, domain.ErrConflict
 			}
-			if _, ok := s.tasks[e.TaskID]; !ok && e.Type != "TaskCreated" {
+			global := e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied"
+			if _, ok := s.tasks[e.TaskID]; !ok && e.Type != "TaskCreated" && !global {
 				return nil, domain.ErrNotFound
 			}
 			s.events = append(s.events, e)
@@ -79,7 +100,17 @@ func (s *Store) apply(e domain.Event) error {
 		return err
 	}
 	t := s.tasks[e.TaskID]
-	if e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" {
+	if e.Type == "QuotaReported" {
+		known := true
+		if v, ok := p["known"].(bool); ok {
+			known = v
+		}
+		harness, _ := p["harness_id"].(string)
+		consumed, _ := p["consumed"].(float64)
+		s.addQuotaUsage(harness, QuotaUsage{At: e.At, Consumed: consumed, Known: known})
+		return nil
+	}
+	if e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" {
 		return nil
 	}
 	switch e.Type {
@@ -266,10 +297,72 @@ func (s *Store) apply(e domain.Event) error {
 		}
 	}
 	t.Version = e.Version
-	s.tasks[e.TaskID] = t
+	s.replaceTask(e.TaskID, t)
 	return nil
 }
 
+func activeLease(t domain.Task) (string, bool) {
+	if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil {
+		return "", false
+	}
+	return t.Lease.HarnessID, t.Lease.HarnessID != ""
+}
+
+func (s *Store) replaceTask(id string, next domain.Task) {
+	if previous, ok := s.tasks[id]; ok {
+		if harness, active := activeLease(previous); active {
+			delete(s.activeLeases[harness], id)
+			if len(s.activeLeases[harness]) == 0 {
+				delete(s.activeLeases, harness)
+			}
+		}
+	}
+	if harness, active := activeLease(next); active {
+		if s.activeLeases[harness] == nil {
+			s.activeLeases[harness] = map[string]struct{}{}
+		}
+		s.activeLeases[harness][id] = struct{}{}
+	}
+	s.tasks[id] = next
+}
+
+func (s *Store) addQuotaUsage(harness string, usage QuotaUsage) {
+	index := s.quota[harness]
+	// Receipts normally arrive in timestamp order, making index maintenance
+	// O(1). Keep the out-of-order path correct for replay and delayed worker
+	// reports without imposing its rebuild cost on the common append path.
+	at := sort.Search(len(index.records), func(i int) bool { return usage.At.Before(index.records[i].At) })
+	if at == len(index.records) {
+		if len(index.prefix) == 0 {
+			index.prefix = []float64{0}
+			index.unknownPrefix = []int{0}
+		}
+		prefix := index.prefix[len(index.prefix)-1]
+		unknown := index.unknownPrefix[len(index.unknownPrefix)-1]
+		index.records = append(index.records, usage)
+		index.prefix = append(index.prefix, prefix+usage.Consumed)
+		if !usage.Known {
+			unknown++
+		}
+		index.unknownPrefix = append(index.unknownPrefix, unknown)
+		s.quota[harness] = index
+		return
+	}
+	index.records = append(index.records, QuotaUsage{})
+	copy(index.records[at+1:], index.records[at:])
+	index.records[at] = usage
+	index.prefix = make([]float64, len(index.records)+1)
+	index.unknownPrefix = make([]int, len(index.records)+1)
+	for i, receipt := range index.records {
+		index.prefix[i+1] = index.prefix[i] + receipt.Consumed
+		index.unknownPrefix[i+1] = index.unknownPrefix[i]
+		if !receipt.Known {
+			index.unknownPrefix[i+1]++
+		}
+	}
+	s.quota[harness] = index
+}
+
 func retryBackoff(attempt int) time.Duration {
 	if attempt < 1 {
 		attempt = 1
@@ -392,8 +485,13 @@ func (s *Store) Append(e domain.Event) error {
 	s.events = append(s.events, e)
 	s.seq = e.Seq
 	// Snapshot failure does not roll back a committed event. Open always
-	// rebuilds from the log, so leaving a stale cache is safe.
-	_ = s.writeSnapshot()
+	// rebuilds from the log, so leaving a stale cache is safe. Keep only the
+	// initial compatibility cache: repeatedly serializing the full projection
+	// turns an otherwise O(1) append into O(tasks) work and is never used for
+	// recovery.
+	if s.seq == 1 {
+		_ = s.writeSnapshot()
+	}
 	return nil
 }
 
@@ -491,6 +589,47 @@ func (s *Store) Tasks() []domain.Task {
 	}
 	return out
 }
+
+// SchedulingSnapshot captures the task projection and lease occupancy under
+// one lock. A routing pass uses this immutable view, then updates its local
+// occupancy as it issues leases; it never repeatedly scan-locks the store.
+type SchedulingSnapshot struct {
+	Tasks        []domain.Task
+	ActiveLeases map[string]int
+}
+
+func (s *Store) SchedulingSnapshot() SchedulingSnapshot {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	snapshot := SchedulingSnapshot{
+		Tasks:        make([]domain.Task, 0, len(s.tasks)),
+		ActiveLeases: make(map[string]int, len(s.activeLeases)),
+	}
+	for _, task := range s.tasks {
+		snapshot.Tasks = append(snapshot.Tasks, task)
+	}
+	for harness, leases := range s.activeLeases {
+		snapshot.ActiveLeases[harness] = len(leases)
+	}
+	return snapshot
+}
+
+// QuotaSince answers a rolling-window usage query from the per-harness index
+// instead of walking events.jsonl. known is false when the interval has no
+// native receipt or any receipt explicitly reports unknown usage.
+func (s *Store) QuotaSince(harness string, since time.Time) (consumed float64, known bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	index, ok := s.quota[harness]
+	if !ok {
+		return 0, false
+	}
+	start := sort.Search(len(index.records), func(i int) bool { return !index.records[i].At.Before(since) })
+	if start == len(index.records) {
+		return 0, false
+	}
+	return index.prefix[len(index.records)] - index.prefix[start], index.unknownPrefix[len(index.records)] == index.unknownPrefix[start]
+}
 func (s *Store) Events(since uint64) []domain.Event {
 	s.mu.Lock()
 	defer s.mu.Unlock()

internal/store/store_test.go

@@ -212,6 +212,52 @@ func TestOpenRebuildsOnlyFromLogAndIgnoresCorruptSnapshot(t *testing.T) {
 	}
 }
 
+func TestSchedulingAndQuotaIndexesReplayFromEvents(t *testing.T) {
+	dir := t.TempDir()
+	s, err := Open(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Append(created("indexed")); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Lease("task-1", "h1", time.Hour); err != nil {
+		t.Fatal(err)
+	}
+	snapshot := s.SchedulingSnapshot()
+	if got := snapshot.ActiveLeases["h1"]; got != 1 {
+		t.Fatalf("active lease index=%d, want 1", got)
+	}
+	now := time.Now().UTC()
+	report := func(at time.Time, consumed float64, known bool) {
+		payload, _ := json.Marshal(map[string]any{"harness_id": "h1", "consumed": consumed, "known": known})
+		if err := s.Append(domain.Event{ID: domain.NewID(), Type: "QuotaReported", TaskID: "quota", Version: 1, At: at, Payload: payload, Surface: string(authz.System)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// Deliberately append out of timestamp order: the index must answer the
+	// rolling query by receipt time, not event sequence.
+	report(now, 4, true)
+	report(now.Add(-2*time.Hour), 3, true)
+	report(now.Add(-time.Hour), 1, false)
+	if used, known := s.QuotaSince("h1", now.Add(-90*time.Minute)); used != 5 || known {
+		t.Fatalf("indexed quota=(%v,%v), want (5,false)", used, known)
+	}
+	if used, known := s.QuotaSince("h1", now.Add(-30*time.Minute)); used != 4 || !known {
+		t.Fatalf("recent indexed quota=(%v,%v), want (4,true)", used, known)
+	}
+	reopened, err := Open(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := reopened.SchedulingSnapshot().ActiveLeases["h1"]; got != 1 {
+		t.Fatalf("replayed active lease index=%d, want 1", got)
+	}
+	if used, known := reopened.QuotaSince("h1", now.Add(-90*time.Minute)); used != 5 || known {
+		t.Fatalf("replayed indexed quota=(%v,%v), want (5,false)", used, known)
+	}
+}
+
 func TestReplayLegacyLeaseDerivesNonRenewableFence(t *testing.T) {
 	dir := t.TempDir()
 	until := time.Now().Add(time.Hour).UnixNano()