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

## Hazard

`storage.Manager.StreamActivities` walked the activity bucket inside a
BBolt read transaction and sent every matching record into a channel
buffered at 100, with no context or done channel. Any consumer that
stopped reading early — the 24h profile stats loop breaking on
`ctx.Err()`, an activity export whose client disconnected mid-stream —
left the producer blocked on the send for the life of the process,
holding the read transaction and the mmap read lock: `Stats().OpenTxN`
stayed at 1, later writes could not remap the file, and `db.Close` hung
on shutdown. #1718 worked around it in its own consumer with a drain
goroutine (`drainActivities`); the two older consumers still leaked.

## Fix

- `StreamActivities(ctx, filter)` at the source: the producer checks
`ctx.Err()` before every send and selects on `ctx.Done()` around it, so
cancellation returns from `db.View` promptly and closes the channel.
Cancellation is not logged as a stream error. The signature changed
everywhere (storage → runtime → server → httpapi controller interface
and its test doubles); no external callers.
- Consumers pass their request/loop context: the httpapi export cancels
a per-request context on its write-error return, summary and scoped
usage pass `r.Context()`, `Stats24h` and `Suggest` pass their `ctx`.
`Stats24h` no longer caches a truncated rollup after cancellation;
`Suggest` returns the context error after the loop instead of a partial
window (a pre-cancelled context now yields zero records, so the in-loop
check alone would have returned a clean empty result).
- `drainActivities` and its test are removed.

## Tests

- `internal/storage/activity_stream_cancel_test.go`: abandon the stream
after one record with 300 matching records, assert `OpenTxN` returns to
0 and `Close` returns; the no-cancel variant proves the transaction
really is pinned without cancellation (guards the first test against
passing vacuously); pre-cancelled context yields nothing; live context
streams all. The first test hung `Close` for the full timeout before the
fix.
- `TestProfilesService_Stats24hDoesNotCacheTruncatedRollup`,
`TestSuggest_CancelledContextReturnsWithRecordsLeft` (updated).
- Green under `-race`: `internal/storage`, `internal/httpapi`,
`internal/runtime`. `golangci-lint` (CI config, bare and `--build-tags
server`) reports only pre-existing findings in untouched lines.

Author: Dumbris

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}