Merge pull request #1464 from connermo/feat/outbound-dispatch-sharding

perf(channels): shard outbound dispatch per conversation

docs/00-architecture-overview.md

@@ -389,6 +389,35 @@ flowchart TD
 | `team` | 100 | `GOCLAW_LANE_TEAM` | Agent team/delegation executions |
 | `cron` | 30 | `GOCLAW_LANE_CRON` | Scheduled cron jobs |
 
+### Scaling Beyond the Lane Defaults
+
+`main` bounds how many users can be answered at once, but raising it alone just
+moves the queue downstream. These knobs need to move with it:
+
+| Setting | Default | Env Override | Why it binds |
+|---------|:-------:|--------------|--------------|
+| Postgres max open conns | 25 | `GOCLAW_PG_MAX_OPEN_CONNS` | Held per query, not per run — undersizing shows up as burst latency when many runs load context at once |
+| Postgres max idle conns | 10 | `GOCLAW_PG_MAX_IDLE_CONNS` | Clamped to max open |
+| Provider idle conns | 100 | `GOCLAW_HTTP_MAX_IDLE_CONNS` | Total across all provider hosts |
+| Provider idle conns per host | 10 | `GOCLAW_HTTP_MAX_IDLE_CONNS_PER_HOST` | When one provider takes nearly all traffic, every request past this limit pays a fresh TCP+TLS handshake. Self-hosted/in-network providers should match `main`. |
+| Outbound dispatch shards | 8 | `GOCLAW_OUTBOUND_SHARDS` | See below |
+
+Channel-side rate limits are metered by the *remote* platform and are not
+covered by any of the above — DingTalk's AI Card API, for instance, meters per
+app, so `card_update_interval_ms` decides how many conversations can stream
+concurrently regardless of local capacity.
+
+### Outbound Dispatch
+
+Outbound delivery is sharded by `channel + ChatID`. A conversation always maps
+to the same shard and is delivered serially there, so the ordering a run
+depends on — block replies, retry notices, then the final answer — is
+preserved. Unrelated conversations run in parallel, so one slow send (a media
+upload of several seconds) no longer stalls every other reply in the process.
+
+Raising `GOCLAW_OUTBOUND_SHARDS` past the point where the remote API meters the
+gateway only moves queueing from this process into theirs.
+
 ### Session Queue Concurrency
 
 Per-session queues now support configurable `maxConcurrent`:

internal/channels/dispatch.go

@@ -3,11 +3,14 @@ package channels
 import (
 	"context"
 	"fmt"
+	"hash/fnv"
 	"log/slog"
 	"net/http"
 	"os"
 	"regexp"
+	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/google/uuid"
 	"github.com/nextlevelbuilder/goclaw/internal/bus"
@@ -20,85 +23,205 @@ type WebhookRoute struct {
 	Handler http.Handler
 }
 
-// dispatchOutbound consumes outbound messages from the bus and routes them
-// to the appropriate channel. Internal channels are silently skipped.
+// Outbound dispatch is sharded by conversation.
+//
+// A single dispatch goroutine made every reply in the process — every channel,
+// every user — wait behind the slowest Send. One media upload of a few seconds
+// froze all outbound traffic for that long. The ordering the old loop actually
+// protected is per-conversation (a run emits block replies, retry notices and
+// the final answer to one ChatID, and those must arrive in that order); it
+// never needed a global order.
+//
+// Sharding on channel+ChatID keeps that guarantee — a conversation always maps
+// to the same shard and is processed serially there — while letting unrelated
+// conversations proceed in parallel.
+const (
+	defaultOutboundShards = 8
+
+	// outboundShardQueue buffers each shard so a brief stall on one Send does
+	// not immediately push back on the reader (which would re-serialize the
+	// shards). The bus itself remains the real backpressure boundary.
+	outboundShardQueue = 256
+)
+
+// outboundShardCount resolves the dispatch worker count.
+//
+//	GOCLAW_OUTBOUND_SHARDS=8
+//
+// Raising it past the point where the *remote* API meters us (DingTalk's
+// per-app card QPS, a provider's rate limit) only moves queueing from our
+// process into theirs.
+func outboundShardCount() int {
+	if v := os.Getenv("GOCLAW_OUTBOUND_SHARDS"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			return n
+		}
+	}
+	return defaultOutboundShards
+}
+
+// outboundShardIndex maps a conversation to a stable shard.
+//
+// The channel name is part of the key because ChatIDs are only unique within a
+// channel — two channels can legitimately use the same conversation id.
+func outboundShardIndex(channel, chatID string, shards int) int {
+	h := fnv.New32a()
+	_, _ = h.Write([]byte(channel))
+	_, _ = h.Write([]byte{0})
+	_, _ = h.Write([]byte(chatID))
+	return int(h.Sum32() % uint32(shards))
+}
+
+// dispatchOutbound consumes outbound messages from the bus and fans them out to
+// per-conversation shard workers. Internal channels are silently skipped.
 func (m *Manager) dispatchOutbound(ctx context.Context) {
-	slog.Info("outbound dispatcher started")
+	shards := outboundShardCount()
+
+	queues := make([]chan bus.OutboundMessage, shards)
+	var wg sync.WaitGroup
+	for i := range queues {
+		queues[i] = make(chan bus.OutboundMessage, outboundShardQueue)
+		wg.Add(1)
+		go func(id int, q <-chan bus.OutboundMessage) {
+			defer wg.Done()
+			m.dispatchShard(ctx, id, q)
+		}(i, queues[i])
+	}
+
+	slog.Info("outbound dispatcher started", "shards", shards)
+
+	defer func() {
+		// Closing the queues drains whatever each shard already accepted, so a
+		// reply that made it out of the bus is not dropped on shutdown.
+		for _, q := range queues {
+			close(q)
+		}
+		wg.Wait()
+		slog.Info("outbound dispatcher stopped")
+	}()
 
 	for {
 		select {
 		case <-ctx.Done():
-			slog.Info("outbound dispatcher stopped")
 			return
 		default:
-			msg, ok := m.bus.SubscribeOutbound(ctx)
-			if !ok {
-				continue
-			}
-
-			// Skip internal channels
-			if IsInternalChannel(msg.Channel) {
-				continue
-			}
-
-			m.mu.RLock()
-			channel, exists := m.channels[msg.Channel]
-			m.mu.RUnlock()
-
-			if !exists {
-				slog.Warn("unknown channel for outbound message", "channel", msg.Channel)
-				continue
-			}
-
-			// Filter out temp media files that no longer exist (already sent by another dispatch).
-			if len(msg.Media) > 0 {
-				tmpDir := os.TempDir()
-				filtered := msg.Media[:0]
-				for _, media := range msg.Media {
-					if media.URL != "" && strings.HasPrefix(media.URL, tmpDir) {
-						if _, err := os.Stat(media.URL); err != nil {
-							slog.Debug("skipping already-delivered temp media", "path", media.URL)
-							continue
-						}
-					}
-					filtered = append(filtered, media)
-				}
-				msg.Media = filtered
-				// If only media was in this message and all files are gone, skip entirely.
-				if len(msg.Media) == 0 && msg.Content == "" {
-					continue
-				}
-			}
-
-			// Add tenant context for per-tenant TTS auto-apply
-			sendCtx := ctx
-			if msg.TenantID != uuid.Nil {
-				sendCtx = store.WithTenantID(ctx, msg.TenantID)
-			}
-
-			// Add agent audio context for per-agent TTS voice override
-			if msg.AgentID != uuid.Nil && len(msg.AgentOtherConfig) > 0 {
-				sendCtx = store.WithAgentAudio(sendCtx, store.AgentAudioSnapshot{
-					AgentID:     msg.AgentID,
-					OtherConfig: msg.AgentOtherConfig,
-				})
-			}
-
-			if err := channel.Send(sendCtx, msg); err != nil {
-				m.handleSendFailure(sendCtx, channel, msg, err)
-			}
-
-			// Clean up temp media files only. Workspace-generated files are preserved
-			// so they remain accessible via workspace/web UI after delivery.
-			tmpDir := os.TempDir()
-			for _, media := range msg.Media {
-				if media.URL != "" && strings.HasPrefix(media.URL, tmpDir) {
-					if err := os.Remove(media.URL); err != nil {
-						slog.Debug("failed to clean up media file", "path", media.URL, "error", err)
-					}
-				}
-			}
 		}
+
+		msg, ok := m.bus.SubscribeOutbound(ctx)
+		if !ok {
+			return
+		}
+		if IsInternalChannel(msg.Channel) {
+			continue
+		}
+
+		idx := outboundShardIndex(msg.Channel, msg.ChatID, shards)
+		select {
+		case queues[idx] <- msg:
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// dispatchShard delivers one shard's messages, strictly in order.
+func (m *Manager) dispatchShard(ctx context.Context, id int, q <-chan bus.OutboundMessage) {
+	for msg := range q {
+		m.deliverOutbound(ctx, msg)
+	}
+	slog.Debug("outbound shard stopped", "shard", id)
+}
+
+// deliverOutbound sends one message and cleans up its temp media.
+func (m *Manager) deliverOutbound(ctx context.Context, msg bus.OutboundMessage) {
+	m.mu.RLock()
+	channel, exists := m.channels[msg.Channel]
+	m.mu.RUnlock()
+
+	if !exists {
+		slog.Warn("unknown channel for outbound message", "channel", msg.Channel)
+		return
+	}
+
+	kept, claimed := m.claimTempMedia(msg.Media)
+	defer m.releaseTempMedia(claimed)
+	msg.Media = kept
+
+	// If only media was in this message and every file is gone, skip entirely.
+	if len(msg.Media) == 0 && msg.Content == "" {
+		return
+	}
+
+	// Add tenant context for per-tenant TTS auto-apply
+	sendCtx := ctx
+	if msg.TenantID != uuid.Nil {
+		sendCtx = store.WithTenantID(ctx, msg.TenantID)
+	}
+
+	// Add agent audio context for per-agent TTS voice override
+	if msg.AgentID != uuid.Nil && len(msg.AgentOtherConfig) > 0 {
+		sendCtx = store.WithAgentAudio(sendCtx, store.AgentAudioSnapshot{
+			AgentID:     msg.AgentID,
+			OtherConfig: msg.AgentOtherConfig,
+		})
+	}
+
+	if err := channel.Send(sendCtx, msg); err != nil {
+		m.handleSendFailure(sendCtx, channel, msg, err)
+	}
+}
+
+// claimTempMedia filters msg.Media down to the files this dispatch owns, and
+// returns the temp paths it claimed so they can be released after the send.
+//
+// Serial dispatch deduped temp media implicitly: the first delivery removed the
+// file, and a later message carrying the same path found os.Stat failing and
+// skipped it. Across concurrent shards that check alone is a TOCTOU race — two
+// shards can both stat a live file and upload it twice.
+//
+// The claim set closes the window. Entries live only for the duration of a
+// send, so the set stays bounded, and a message that arrives after the owner
+// finished still finds the file gone and skips it exactly as before.
+//
+// Non-temp media (workspace files) is passed through untouched: it is never
+// removed after delivery, so it needs no claim.
+func (m *Manager) claimTempMedia(media []bus.MediaAttachment) (kept []bus.MediaAttachment, claimed []string) {
+	if len(media) == 0 {
+		return nil, nil
+	}
+
+	tmpDir := os.TempDir()
+	kept = make([]bus.MediaAttachment, 0, len(media))
+
+	for _, mf := range media {
+		if mf.URL == "" || !strings.HasPrefix(mf.URL, tmpDir) {
+			kept = append(kept, mf)
+			continue
+		}
+		if _, err := os.Stat(mf.URL); err != nil {
+			slog.Debug("skipping already-delivered temp media", "path", mf.URL)
+			continue
+		}
+		if _, loaded := m.mediaClaims.LoadOrStore(mf.URL, struct{}{}); loaded {
+			slog.Debug("skipping temp media already claimed by another shard", "path", mf.URL)
+			continue
+		}
+		kept = append(kept, mf)
+		claimed = append(claimed, mf.URL)
+	}
+	return kept, claimed
+}
+
+// releaseTempMedia removes delivered temp files and drops their claims.
+//
+// Workspace-generated files are preserved so they remain accessible via the
+// workspace/web UI after delivery; only os.TempDir() entries reach here.
+func (m *Manager) releaseTempMedia(claimed []string) {
+	for _, path := range claimed {
+		if err := os.Remove(path); err != nil {
+			slog.Debug("failed to clean up media file", "path", path, "error", err)
+		}
+		m.mediaClaims.Delete(path)
 	}
 }
 

internal/channels/dispatch_shard_test.go

@@ -0,0 +1,316 @@
+package channels
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nextlevelbuilder/goclaw/internal/bus"
+)
+
+// recorderChannel records every delivered message in arrival order, and can
+// block inside Send to simulate a slow upload.
+type recorderChannel struct {
+	BaseChannel
+
+	mu       sync.Mutex
+	received []bus.OutboundMessage
+
+	// blockOn holds Send hostage for messages whose ChatID matches, until
+	// release is closed. Empty means never block.
+	blockOn string
+	release chan struct{}
+}
+
+func newRecorderChannel(name string) *recorderChannel {
+	rc := &recorderChannel{release: make(chan struct{})}
+	rc.BaseChannel = BaseChannel{name: name}
+	return rc
+}
+
+func (r *recorderChannel) Type() string                  { return TypeTelegram }
+func (r *recorderChannel) Start(_ context.Context) error { return nil }
+func (r *recorderChannel) Stop(_ context.Context) error  { return nil }
+func (r *recorderChannel) IsRunning() bool               { return true }
+func (r *recorderChannel) IsAllowed(_ string) bool       { return true }
+
+func (r *recorderChannel) Send(_ context.Context, msg bus.OutboundMessage) error {
+	if r.blockOn != "" && msg.ChatID == r.blockOn {
+		<-r.release
+	}
+	r.mu.Lock()
+	r.received = append(r.received, msg)
+	r.mu.Unlock()
+	return nil
+}
+
+func (r *recorderChannel) contentsFor(chatID string) []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	var out []string
+	for _, m := range r.received {
+		if m.ChatID == chatID {
+			out = append(out, m.Content)
+		}
+	}
+	return out
+}
+
+func (r *recorderChannel) count() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return len(r.received)
+}
+
+// waitFor polls cond until it holds or the deadline passes.
+func waitFor(t *testing.T, timeout time.Duration, cond func() bool) bool {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return true
+		}
+		time.Sleep(2 * time.Millisecond)
+	}
+	return cond()
+}
+
+// distinctShardChats returns two chat IDs that hash to different shards, so a
+// test can prove two conversations really do run on separate workers.
+func distinctShardChats(t *testing.T, channel string, shards int) (string, string) {
+	t.Helper()
+	base := "chat-0"
+	baseIdx := outboundShardIndex(channel, base, shards)
+	for i := 1; i < 500; i++ {
+		other := fmt.Sprintf("chat-%d", i)
+		if outboundShardIndex(channel, other, shards) != baseIdx {
+			return base, other
+		}
+	}
+	t.Fatalf("no two chat ids landed on different shards out of %d", shards)
+	return "", ""
+}
+
+// A conversation must always map to the same shard — that mapping is the only
+// thing preserving per-chat ordering once dispatch runs in parallel.
+func TestOutboundShardIndex_StableAndBounded(t *testing.T) {
+	t.Parallel()
+
+	const shards = 8
+	for i := range 200 {
+		chatID := fmt.Sprintf("chat-%d", i)
+		got := outboundShardIndex("telegram-test", chatID, shards)
+		if got < 0 || got >= shards {
+			t.Fatalf("shard index %d out of range [0,%d)", got, shards)
+		}
+		if again := outboundShardIndex("telegram-test", chatID, shards); again != got {
+			t.Fatalf("shard index for %q not stable: %d then %d", chatID, got, again)
+		}
+	}
+}
+
+// ChatIDs are only unique within a channel, so the channel name must be part
+// of the shard key — otherwise two channels sharing a conversation id would
+// have their messages interleaved on one worker.
+func TestOutboundShardIndex_ChannelIsPartOfKey(t *testing.T) {
+	t.Parallel()
+
+	const shards = 64
+	same := 0
+	for i := range 100 {
+		chatID := fmt.Sprintf("chat-%d", i)
+		if outboundShardIndex("alpha", chatID, shards) == outboundShardIndex("beta", chatID, shards) {
+			same++
+		}
+	}
+	// With 64 shards, identical placement for every id would mean the channel
+	// name was ignored entirely.
+	if same == 100 {
+		t.Fatal("channel name does not affect shard placement")
+	}
+}
+
+// The ordering guarantee dispatch actually owes callers: a run emits block
+// replies, retry notices and the final answer to one ChatID, and they must
+// arrive in that order.
+func TestDispatchOutbound_PreservesPerChatOrder(t *testing.T) {
+	t.Parallel()
+
+	mb := bus.New()
+	mgr := NewManager(mb)
+	ch := newRecorderChannel("telegram-test")
+	mgr.channels["telegram-test"] = ch
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go mgr.dispatchOutbound(ctx)
+
+	const n = 50
+	for i := range n {
+		mb.PublishOutbound(bus.OutboundMessage{
+			Channel: "telegram-test",
+			ChatID:  "chat-order",
+			Content: fmt.Sprintf("msg-%d", i),
+		})
+	}
+
+	if !waitFor(t, 5*time.Second, func() bool { return ch.count() >= n }) {
+		t.Fatalf("only %d/%d messages delivered", ch.count(), n)
+	}
+
+	got := ch.contentsFor("chat-order")
+	for i := range n {
+		want := fmt.Sprintf("msg-%d", i)
+		if got[i] != want {
+			t.Fatalf("message %d out of order: got %q, want %q", i, got[i], want)
+		}
+	}
+}
+
+// The regression this whole change exists for: one stalled conversation must
+// not hold up every other conversation in the process.
+func TestDispatchOutbound_SlowChatDoesNotBlockOthers(t *testing.T) {
+	t.Parallel()
+
+	shards := outboundShardCount()
+	slowChat, fastChat := distinctShardChats(t, "telegram-test", shards)
+
+	mb := bus.New()
+	mgr := NewManager(mb)
+	ch := newRecorderChannel("telegram-test")
+	ch.blockOn = slowChat
+	mgr.channels["telegram-test"] = ch
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go mgr.dispatchOutbound(ctx)
+
+	// The stuck conversation goes first, exactly as it would when a media
+	// upload lands ahead of everyone else's replies.
+	mb.PublishOutbound(bus.OutboundMessage{
+		Channel: "telegram-test",
+		ChatID:  slowChat,
+		Content: "slow-upload",
+	})
+	mb.PublishOutbound(bus.OutboundMessage{
+		Channel: "telegram-test",
+		ChatID:  fastChat,
+		Content: "should-not-wait",
+	})
+
+	if !waitFor(t, 5*time.Second, func() bool { return len(ch.contentsFor(fastChat)) == 1 }) {
+		close(ch.release)
+		t.Fatal("a reply on an unrelated conversation was blocked behind a slow send")
+	}
+
+	// The slow one still completes once it is unblocked.
+	close(ch.release)
+	if !waitFor(t, 5*time.Second, func() bool { return len(ch.contentsFor(slowChat)) == 1 }) {
+		t.Fatal("the slow conversation never completed after release")
+	}
+}
+
+// Temp media was deduped by the fact that serial dispatch removed the file
+// before the next message was looked at. Concurrent shards make that a TOCTOU
+// race, so the claim must be exclusive.
+func TestClaimTempMedia_ExclusiveUnderConcurrency(t *testing.T) {
+	t.Parallel()
+
+	mgr := NewManager(bus.New())
+
+	path := filepath.Join(t.TempDir(), "shared.jpg")
+	if err := os.WriteFile(path, []byte("x"), 0o600); err != nil {
+		t.Fatalf("write temp media: %v", err)
+	}
+	// claimTempMedia only guards files under os.TempDir(); t.TempDir() lives
+	// there, so this mirrors the real path.
+	if !isUnderTempDir(path) {
+		t.Skipf("t.TempDir() %q is not under os.TempDir() %q on this platform", path, os.TempDir())
+	}
+
+	media := []bus.MediaAttachment{{URL: path, ContentType: "image/jpeg"}}
+
+	const racers = 16
+	var wg sync.WaitGroup
+	winners := make(chan []string, racers)
+	for range racers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			kept, claimed := mgr.claimTempMedia(media)
+			if len(kept) > 0 {
+				winners <- claimed
+			}
+		}()
+	}
+	wg.Wait()
+	close(winners)
+
+	if n := len(winners); n != 1 {
+		t.Fatalf("%d dispatchers claimed the same temp file; want exactly 1", n)
+	}
+}
+
+// Non-temp media (workspace files) is never removed after delivery, so it must
+// pass through unclaimed — otherwise a second message referencing the same
+// workspace file would silently lose it.
+func TestClaimTempMedia_PassesThroughNonTempAndSkipsMissing(t *testing.T) {
+	t.Parallel()
+
+	mgr := NewManager(bus.New())
+
+	missing := filepath.Join(os.TempDir(), "goclaw-does-not-exist-12345.bin")
+	workspace := "/workspace/report.pdf"
+
+	kept, claimed := mgr.claimTempMedia([]bus.MediaAttachment{
+		{URL: workspace},
+		{URL: missing},
+	})
+
+	if len(kept) != 1 || kept[0].URL != workspace {
+		t.Fatalf("kept = %+v, want only the workspace file", kept)
+	}
+	if len(claimed) != 0 {
+		t.Fatalf("claimed = %v, want no claims on non-temp media", claimed)
+	}
+}
+
+// releaseTempMedia must drop the claim as well as the file, so the claim set
+// stays bounded across a long-running process.
+func TestReleaseTempMedia_ClearsClaim(t *testing.T) {
+	t.Parallel()
+
+	mgr := NewManager(bus.New())
+
+	path := filepath.Join(t.TempDir(), "once.jpg")
+	if err := os.WriteFile(path, []byte("x"), 0o600); err != nil {
+		t.Fatalf("write temp media: %v", err)
+	}
+	if !isUnderTempDir(path) {
+		t.Skipf("t.TempDir() %q is not under os.TempDir() %q on this platform", path, os.TempDir())
+	}
+
+	media := []bus.MediaAttachment{{URL: path}}
+
+	kept, claimed := mgr.claimTempMedia(media)
+	if len(kept) != 1 || len(claimed) != 1 {
+		t.Fatalf("first claim failed: kept=%d claimed=%d", len(kept), len(claimed))
+	}
+	mgr.releaseTempMedia(claimed)
+
+	if _, stillHeld := mgr.mediaClaims.Load(path); stillHeld {
+		t.Fatal("claim survived release; the claim set would grow without bound")
+	}
+	if _, err := os.Stat(path); err == nil {
+		t.Fatal("temp file survived release")
+	}
+}
+
+func isUnderTempDir(path string) bool {
+	rel, err := filepath.Rel(os.TempDir(), path)
+	return err == nil && !filepath.IsAbs(rel) && rel != ".." && len(rel) > 0 && rel[0] != '.'
+}

internal/channels/manager.go

@@ -76,6 +76,7 @@ type Manager struct {
 	health           map[string]ChannelHealth
 	bus              *bus.MessageBus
 	runs             sync.Map // runID string → *RunContext
+	mediaClaims      sync.Map // temp media path → struct{}, in-flight dispatch claims
 	dispatchTask     *asyncTask
 	mu               sync.RWMutex
 	contactCollector *store.ContactCollector

internal/providers/defaults.go

@@ -2,6 +2,8 @@ package providers
 
 import (
 	"net/http"
+	"os"
+	"strconv"
 	"time"
 )
 
@@ -21,6 +23,32 @@ const (
 	StdioScanBufMax  = 10 * 1024 * 1024 // 10MB max for large protocol messages
 )
 
+// Idle connection pooling, overridable for deployments that raise agent-run
+// concurrency:
+//
+//	GOCLAW_HTTP_MAX_IDLE_CONNS=100
+//	GOCLAW_HTTP_MAX_IDLE_CONNS_PER_HOST=10
+//
+// When nearly all traffic goes to one provider host, the per-host limit is the
+// one that binds: every concurrent request past it gets a fresh TCP+TLS
+// handshake, and its connection is closed rather than pooled on completion.
+// A self-hosted or in-network provider has no reason to be throttled this way,
+// so those deployments should raise it to match LaneMain.
+const (
+	defaultMaxIdleConns        = 100
+	defaultMaxIdleConnsPerHost = 10
+)
+
+// transportEnv reads a positive int from an env var, falling back to defaultVal.
+func transportEnv(key string, defaultVal int) int {
+	if v := os.Getenv(key); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			return n
+		}
+	}
+	return defaultVal
+}
+
 // NewDefaultTransport returns an http.Transport with per-stage timeouts but no
 // overall deadline. The absence of Client.Timeout allows LLM streaming responses
 // (extended thinking, long completions) to run indefinitely once the response has
@@ -57,8 +85,8 @@ func NewDefaultTransport() *http.Transport {
 		IdleConnTimeout:       90 * time.Second, // close idle keep-alive connections
 		TLSHandshakeTimeout:   10 * time.Second,
 		ExpectContinueTimeout: 1 * time.Second,
-		MaxIdleConns:          100,
-		MaxIdleConnsPerHost:   10,
+		MaxIdleConns:          transportEnv("GOCLAW_HTTP_MAX_IDLE_CONNS", defaultMaxIdleConns),
+		MaxIdleConnsPerHost:   transportEnv("GOCLAW_HTTP_MAX_IDLE_CONNS_PER_HOST", defaultMaxIdleConnsPerHost),
 	}
 }
 

internal/scheduler/lanes.go

@@ -154,6 +154,11 @@ func NewLaneManager(configs []LaneConfig) *LaneManager {
 //	GOCLAW_LANE_SUBAGENT=50
 //	GOCLAW_LANE_TEAM=100
 //	GOCLAW_LANE_CRON=30
+//
+// LaneMain bounds concurrent agent runs across every channel, so on a bot
+// shared by a whole org it is the ceiling on how many people can be answered
+// at once. A run is overwhelmingly blocked on the LLM, not on CPU, so raising
+// it trades provider load for wait time, not local work.
 func DefaultLanes() []LaneConfig {
 	return []LaneConfig{
 		{Name: LaneMain, Concurrency: laneEnv("GOCLAW_LANE_MAIN", 30)},

internal/store/pg/pool.go

@@ -4,6 +4,8 @@ import (
 	"database/sql"
 	"fmt"
 	"log/slog"
+	"os"
+	"strconv"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/stdlib"
@@ -14,6 +16,34 @@ import (
 // the gateway's own connections apart from external clients (issue #1338).
 const PoolApplicationName = "goclaw"
 
+// Connection pool sizing, overridable for deployments that raise LaneMain:
+//
+//	GOCLAW_PG_MAX_OPEN_CONNS=25
+//	GOCLAW_PG_MAX_IDLE_CONNS=10
+//
+// A connection is held per query, not per agent run, so the pool does not cap
+// concurrency directly — it caps how many queries can be in flight. The cost of
+// undersizing shows up as latency at burst, when many runs start together and
+// each loads context and history at the same moment, which is why the ceiling
+// wants to move with LaneMain rather than stay fixed below it.
+//
+// Postgres' own max_connections is the real limit: keep the sum across all
+// gateway processes under it.
+const (
+	defaultMaxOpenConns = 25
+	defaultMaxIdleConns = 10
+)
+
+// poolEnv reads a positive int from an env var, falling back to defaultVal.
+func poolEnv(key string, defaultVal int) int {
+	if v := os.Getenv(key); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			return n
+		}
+	}
+	return defaultVal
+}
+
 // OpenDB creates a database/sql connection to Postgres using pgx driver.
 func OpenDB(dsn string) (*sql.DB, error) {
 	config, err := pgx.ParseConfig(dsn)
@@ -29,14 +59,19 @@ func OpenDB(dsn string) (*sql.DB, error) {
 	}
 	db := stdlib.OpenDB(*config)
 
-	db.SetMaxOpenConns(25)
-	db.SetMaxIdleConns(10)
+	maxOpen := poolEnv("GOCLAW_PG_MAX_OPEN_CONNS", defaultMaxOpenConns)
+	maxIdle := poolEnv("GOCLAW_PG_MAX_IDLE_CONNS", defaultMaxIdleConns)
+	if maxIdle > maxOpen {
+		maxIdle = maxOpen
+	}
+	db.SetMaxOpenConns(maxOpen)
+	db.SetMaxIdleConns(maxIdle)
 
 	if err := db.Ping(); err != nil {
 		db.Close()
 		return nil, fmt.Errorf("ping postgres: %w", err)
 	}
 
-	slog.Info("postgres connected", "dsn_len", len(dsn))
+	slog.Info("postgres connected", "dsn_len", len(dsn), "max_open", maxOpen, "max_idle", maxIdle)
 	return db, nil
 }