fix(storage): make StreamActivities cancel-safe at the source

internal/httpapi/activity.go

@@ -747,8 +747,12 @@ func (s *Server) handleExportActivity(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Disposition", "attachment; filename=\""+filename+"\"")
 	w.Header().Set("Cache-Control", "no-cache")
 
-	// Stream activities
-	activityCh := s.controller.StreamActivities(filter)
+	// Stream activities. The context is cancelled on return, so a write
+	// error (client gone) releases the producer's read transaction instead of
+	// leaving it parked on a full channel for the life of the process.
+	streamCtx, cancelStream := context.WithCancel(r.Context())
+	defer cancelStream()
+	activityCh := s.controller.StreamActivities(streamCtx, filter)
 
 	// Write CSV header if format is CSV
 	if format == "csv" {
@@ -939,9 +943,9 @@ func (s *Server) handleActivitySummary(w http.ResponseWriter, r *http.Request) {
 	toolCounts := make(map[string]int)
 	perServer := make(map[string]*perServerAccumulator)
 
-	// The stream holds a read transaction open until the channel is drained or
-	// closed, so this loop must always run to completion.
-	for a := range s.controller.StreamActivities(filter) {
+	// The producer releases its read transaction when the request context is
+	// cancelled, so a client that disconnects mid-scan does not pin the store.
+	for a := range s.controller.StreamActivities(r.Context(), filter) {
 		totalCount++
 
 		// "How many rows" (totalCount) and "how many calls" (callCount) are
@@ -1398,7 +1402,7 @@ func (s *Server) handleActivityUsage(w http.ResponseWriter, r *http.Request) {
 
 	var resp *contracts.UsageAggregateResponse
 	if params.scopeFiltered() {
-		resp = s.scopedUsageResponse(params, time.Now().UTC())
+		resp = s.scopedUsageResponse(r.Context(), params, time.Now().UTC())
 	} else {
 		snap := s.controller.UsageSnapshot()
 		tokens, _ := s.controller.GetTokenSavings()
@@ -1416,7 +1420,7 @@ func (s *Server) handleActivityUsage(w http.ResponseWriter, r *http.Request) {
 // admission rule instead. Per-tool metrics are therefore window-bounded here
 // (lifetime-cumulative on the unfiltered path), and the global tokens-saved
 // headline is omitted because it cannot be re-derived for a subset.
-func (s *Server) scopedUsageResponse(p usageParams, now time.Time) *contracts.UsageAggregateResponse {
+func (s *Server) scopedUsageResponse(ctx context.Context, p usageParams, now time.Time) *contracts.UsageAggregateResponse {
 	filter := storage.ActivityFilter{
 		// Every record Apply may admit, including the call_tool_* mirrors it
 		// drops itself: the population a cold-start rebuild scans.
@@ -1433,9 +1437,9 @@ func (s *Server) scopedUsageResponse(p usageParams, now time.Time) *contracts.Us
 		filter.AllowedServers = p.allowed
 	}
 	agg := internalRuntime.NewUsageAggregate()
-	// The stream holds a read transaction open until drained, so it always runs
-	// to completion.
-	for rec := range s.controller.StreamActivities(filter) {
+	// The producer releases its read transaction when the request context is
+	// cancelled, so a client that disconnects mid-scan does not pin the store.
+	for rec := range s.controller.StreamActivities(ctx, filter) {
 		agg.Apply(rec)
 	}
 	agg.UpdatedAt = now

internal/httpapi/activity_auth_identity_test.go

@@ -1,6 +1,7 @@
 package httpapi
 
 import (
+	"context"
 	"encoding/json"
 	"net/http"
 	"testing"
@@ -166,7 +167,7 @@ type streamingActivityController struct {
 	mockActivityController
 }
 
-func (m *streamingActivityController) StreamActivities(filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
+func (m *streamingActivityController) StreamActivities(_ context.Context, filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
 	ch := make(chan *storage.ActivityRecord, len(m.activities))
 	for _, a := range m.activities {
 		if filter.Matches(a) {

internal/httpapi/activity_call_parity_test.go

@@ -1,6 +1,7 @@
 package httpapi
 
 import (
+	"context"
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
@@ -49,7 +50,7 @@ func (m *callParityController) GetCurrentConfig() *config.Config {
 	return &config.Config{APIKey: "test-key", Observability: config.DefaultObservabilityConfig()}
 }
 
-func (m *callParityController) StreamActivities(filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
+func (m *callParityController) StreamActivities(_ context.Context, filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
 	ch := make(chan *storage.ActivityRecord)
 	go func() {
 		defer close(ch)

internal/httpapi/activity_scope_params_test.go

@@ -1,6 +1,7 @@
 package httpapi
 
 import (
+	"context"
 	"encoding/csv"
 	"encoding/json"
 	"net/http"
@@ -52,7 +53,7 @@ func (c *scopeParamsController) ListActivities(f storage.ActivityFilter) ([]*sto
 	return out, total, nil
 }
 
-func (c *scopeParamsController) StreamActivities(f storage.ActivityFilter) <-chan *storage.ActivityRecord {
+func (c *scopeParamsController) StreamActivities(_ context.Context, f storage.ActivityFilter) <-chan *storage.ActivityRecord {
 	c.lastFilter = f
 	ch := make(chan *storage.ActivityRecord, len(c.records))
 	for _, r := range c.records {

internal/httpapi/activity_summary_scope_test.go

@@ -1,6 +1,7 @@
 package httpapi
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -49,7 +50,7 @@ func (m *summaryScopeController) ListActivities(filter storage.ActivityFilter) (
 	return matched, total, nil
 }
 
-func (m *summaryScopeController) StreamActivities(filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
+func (m *summaryScopeController) StreamActivities(_ context.Context, filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
 	ch := make(chan *storage.ActivityRecord)
 	go func() {
 		defer close(ch)

internal/httpapi/contracts_test.go

@@ -302,7 +302,7 @@ func (m *MockServerController) AggregateToolUsage(_ time.Time) (map[string]stora
 	return map[string]storage.ToolUsageStat{}, nil
 }
 func (m *MockServerController) UsageSnapshot() *internalRuntime.UsageAggregate { return nil }
-func (m *MockServerController) StreamActivities(_ storage.ActivityFilter) <-chan *storage.ActivityRecord {
+func (m *MockServerController) StreamActivities(_ context.Context, _ storage.ActivityFilter) <-chan *storage.ActivityRecord {
 	ch := make(chan *storage.ActivityRecord)
 	close(ch)
 	return ch

internal/httpapi/scope_round9_harness_test.go

@@ -1,6 +1,7 @@
 package httpapi
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -93,7 +94,7 @@ func (c *scopeController) ListActivities(filter storage.ActivityFilter) ([]*stor
 	return matched, total, nil
 }
 
-func (c *scopeController) StreamActivities(filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
+func (c *scopeController) StreamActivities(_ context.Context, filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
 	ch := make(chan *storage.ActivityRecord, len(scopeActivities()))
 	for _, rec := range scopeActivities() {
 		if filter.Matches(rec) {

internal/httpapi/security_test.go

@@ -352,7 +352,7 @@ func (m *baseController) AggregateToolUsage(_ time.Time) (map[string]storage.Too
 	return map[string]storage.ToolUsageStat{}, nil
 }
 func (m *baseController) UsageSnapshot() *runtime.UsageAggregate { return nil }
-func (m *baseController) StreamActivities(_ storage.ActivityFilter) <-chan *storage.ActivityRecord {
+func (m *baseController) StreamActivities(_ context.Context, _ storage.ActivityFilter) <-chan *storage.ActivityRecord {
 	ch := make(chan *storage.ActivityRecord)
 	close(ch)
 	return ch

internal/httpapi/server.go

@@ -264,7 +264,7 @@ type ServerController interface {
 	// Activity logging (RFC-003)
 	ListActivities(filter storage.ActivityFilter) ([]*storage.ActivityRecord, int, error)
 	GetActivity(id string) (*storage.ActivityRecord, error)
-	StreamActivities(filter storage.ActivityFilter) <-chan *storage.ActivityRecord
+	StreamActivities(ctx context.Context, filter storage.ActivityFilter) <-chan *storage.ActivityRecord
 	// AggregateToolUsage rolls up tool_call activity per (server,tool) since the
 	// given time. Backs the global tools page usage columns (spec 050).
 	AggregateToolUsage(since time.Time) (map[string]storage.ToolUsageStat, error)

internal/runtime/profiles_service.go

@@ -316,10 +316,7 @@ func (s *ProfilesService) Stats24h(ctx context.Context, viewer ViewerScope) Acti
 		StartTime:      now.Add(-24 * time.Hour),
 		AllowedServers: filterServers,
 	}
-	for rec := range s.rt.StreamActivities(filter) {
-		if ctx.Err() != nil {
-			break
-		}
+	for rec := range s.rt.StreamActivities(ctx, filter) {
 		blocked := rec.Status == "blocked"
 		isCall := rec.Type == storage.ActivityTypeToolCall
 		if !isCall && !blocked {
@@ -341,6 +338,10 @@ func (s *ProfilesService) Stats24h(ctx context.Context, viewer ViewerScope) Acti
 		add(stats.ByProfile, rec.EffectiveProfile())
 		add(stats.ByClient, rec.ClientID)
 	}
+	if ctx.Err() != nil {
+		// The stream closed early: a truncated rollup is not worth caching.
+		return stats
+	}
 
 	s.statsMu.Lock()
 	s.statsCache[key] = cachedStats{at: now, stats: stats}

internal/runtime/profiles_service_test.go

@@ -551,6 +551,26 @@ func TestProfilesService_Stats24hCountsByProfileAndClient(t *testing.T) {
 	assert.Equal(t, Counter{Calls: 2, Blocked: 2}, stats.ByClient["cursor"])
 }
 
+// A cancelled caller gets whatever was counted so far, but that truncated
+// rollup must not be cached for the next 30 s of callers (the stream closes
+// early on cancellation, so the loop ends as if it had seen everything).
+func TestProfilesService_Stats24hDoesNotCacheTruncatedRollup(t *testing.T) {
+	h := newProfilesHarness(t)
+	sm := h.rt.StorageManager()
+	now := time.Now()
+	for i := 0; i < 250; i++ { // more than the stream's channel buffer
+		require.NoError(t, sm.SaveActivity(&storage.ActivityRecord{Type: storage.ActivityTypeToolCall, Status: "success", Profile: "ro", ClientID: "cursor", Timestamp: now}))
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	truncated := h.svc.Stats24h(ctx, ViewerScope{})
+	assert.Less(t, truncated.ByProfile["ro"].Calls, 250, "a cancelled context ends the stream early")
+
+	stats := h.svc.Stats24h(context.Background(), ViewerScope{})
+	assert.Equal(t, 250, stats.ByProfile["ro"].Calls, "the truncated rollup must not have been cached")
+}
+
 // A save surfaces one warning per stale tools.classify entry, with its reason
 // (#1446): the editors previously only learned of staleness from a later
 // effective-tools read.

internal/runtime/profiles_suggest.go

@@ -198,12 +198,7 @@ func (s *ProfilesService) Suggest(ctx context.Context, req SuggestRequest) (*Acc
 		Profile:   req.Profile,
 		TokenName: req.Token,
 	}
-	stream := s.rt.StreamActivities(filter)
-	for rec := range stream {
-		if ctx.Err() != nil {
-			drainActivities(stream)
-			return nil, ctx.Err()
-		}
+	for rec := range s.rt.StreamActivities(ctx, filter) {
 		reason := profile.BlockReason(rec.EffectiveBlockReason())
 		if !profileBlockReasons[reason] {
 			out.NonProfileBlocked++
@@ -250,6 +245,10 @@ func (s *ProfilesService) Suggest(ctx context.Context, req SuggestRequest) (*Acc
 			g.clients[rec.ClientID] = true
 		}
 	}
+	if err := ctx.Err(); err != nil {
+		// The stream closed early; a partial window would read as a complete one.
+		return nil, err
+	}
 
 	for _, g := range groups {
 		keep, err := s.finishSuggestion(ctx, ev, cfg, g, out)
@@ -378,18 +377,6 @@ func deniedByRuleNow(exp *AccessExplanation) bool {
 	return true
 }
 
-// drainActivities lets a StreamActivities producer run to its end after the
-// consumer stops early. The producer sends into a bounded channel from inside
-// a BBolt read transaction, so a consumer that walks away with records left
-// would leave that transaction open for the life of the process, and an open
-// read transaction keeps the database from remapping on a later write.
-func drainActivities(ch <-chan *storage.ActivityRecord) {
-	go func() {
-		for range ch { //nolint:revive // discard the remainder
-		}
-	}()
-}
-
 // buildPatches applies every apply-ready suggestion (allow, add_server) to a
 // copy of the stored profile. Classify needs a tier and is left out.
 func buildPatches(cfg *config.Config, sugg []AccessSuggestion) []ProfilePatch {

internal/runtime/profiles_suggest_test.go

@@ -343,39 +343,12 @@ func TestSuggest_CancelledContextReturnsWithRecordsLeft(t *testing.T) {
 	_, err := r.svc.Suggest(ctx, SuggestRequest{})
 	require.ErrorIs(t, err, context.Canceled)
 
-	// The producer was left behind with records to send; it must be drained so
-	// its read transaction ends and the store keeps accepting writes.
+	// The producer was left behind with records to send; cancellation must have
+	// released its read transaction so the store keeps accepting writes.
 	for i := 0; i < 50; i++ {
 		r.block("research", "github", "create_issue", profile.BlockReasonTier, "ci-bot", time.Minute, fmt.Sprintf("w%d", i))
 	}
 	res, err := r.svc.Suggest(context.Background(), SuggestRequest{})
 	require.NoError(t, err)
 	require.NotNil(t, r.find(res, "github:create_issue", profile.BlockReasonTier))
 }
-
-func TestDrainActivities_LetsTheProducerFinish(t *testing.T) {
-	ch := make(chan *storage.ActivityRecord, 100)
-	finished := make(chan struct{})
-	go func() {
-		defer close(finished)
-		defer close(ch)
-		for i := 0; i < 250; i++ { // more than the buffer: the producer blocks without a reader
-			ch <- &storage.ActivityRecord{RequestID: fmt.Sprintf("r%d", i)}
-		}
-	}()
-	// The consumer takes a few records and walks away, as Suggest does on cancel.
-	<-ch
-	<-ch
-	select {
-	case <-finished:
-		t.Fatal("the producer cannot have finished with no reader")
-	case <-time.After(50 * time.Millisecond):
-	}
-
-	drainActivities(ch)
-	select {
-	case <-finished:
-	case <-time.After(5 * time.Second):
-		t.Fatal("drainActivities did not let the producer finish")
-	}
-}

internal/runtime/runtime.go

@@ -4349,14 +4349,16 @@ func (r *Runtime) UsageSnapshot() *UsageAggregate {
 	return r.activityService.UsageSnapshot()
 }
 
-// StreamActivities returns a channel that yields activity records matching the filter.
-func (r *Runtime) StreamActivities(filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
+// StreamActivities returns a channel that yields activity records matching the
+// filter. A consumer that stops reading early must cancel ctx so the producer
+// releases its read transaction (see storage.Manager.StreamActivities).
+func (r *Runtime) StreamActivities(ctx context.Context, filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
 	if r.storageManager == nil {
 		ch := make(chan *storage.ActivityRecord)
 		close(ch)
 		return ch
 	}
-	return r.storageManager.StreamActivities(filter)
+	return r.storageManager.StreamActivities(ctx, filter)
 }
 
 // ListToolApprovals returns all tool approval records for a server (Spec 032).

internal/server/server.go

@@ -4586,8 +4586,8 @@ func (s *Server) GetActivity(id string) (*storage.ActivityRecord, error) {
 }
 
 // StreamActivities returns a channel that yields activity records matching the filter.
-func (s *Server) StreamActivities(filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
-	return s.runtime.StreamActivities(filter)
+func (s *Server) StreamActivities(ctx context.Context, filter storage.ActivityFilter) <-chan *storage.ActivityRecord {
+	return s.runtime.StreamActivities(ctx, filter)
 }
 
 // AggregateToolUsage rolls up tool_call activity per (server,tool) since the

internal/storage/activity.go

@@ -1,7 +1,9 @@
 package storage
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"time"
 
@@ -288,10 +290,17 @@ func (m *Manager) AggregateToolUsage(since time.Time) (map[string]ToolUsageStat,
 }
 
 // StreamActivities returns a channel that yields activity records matching the filter.
-// The channel is closed when all matching records have been sent.
-// This is useful for streaming large exports without loading all records into memory.
-// Respects filter.Limit and filter.Offset for bounded streaming.
-func (m *Manager) StreamActivities(filter ActivityFilter) <-chan *ActivityRecord {
+// The channel is closed when all matching records have been sent, or as soon as
+// ctx is done. This is useful for streaming large exports without loading all
+// records into memory. Respects filter.Limit and filter.Offset for bounded streaming.
+//
+// The producer walks the activity bucket inside a BBolt read transaction and
+// sends into a bounded channel. A consumer that stops reading early MUST cancel
+// ctx: that is what lets the producer give up its pending send, return from
+// db.View and release the read transaction (an open read transaction pins the
+// mmap and blocks db.Close and remaps on later writes). Without cancellation an
+// abandoned stream holds that transaction for the life of the process.
+func (m *Manager) StreamActivities(ctx context.Context, filter ActivityFilter) <-chan *ActivityRecord {
 	ch := make(chan *ActivityRecord, 100)
 
 	go func() {
@@ -328,14 +337,23 @@ func (m *Manager) StreamActivities(filter ActivityFilter) <-chan *ActivityRecord
 					return nil
 				}
 
-				ch <- &record
-				sent++
+				// A done context wins even when the buffer has room: select
+				// picks a ready case at random, so check first.
+				if err := ctx.Err(); err != nil {
+					return err
+				}
+				select {
+				case ch <- &record:
+					sent++
+				case <-ctx.Done():
+					return ctx.Err()
+				}
 			}
 
 			return nil
 		})
 
-		if err != nil {
+		if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
 			m.logger.Errorw("Error streaming activities", "error", err)
 		}
 	}()