diff --git a/cmd/agent_chat_standalone.go b/cmd/agent_chat_standalone.go
index 2afb4d59..e9d5462d 100644
--- a/cmd/agent_chat_standalone.go
+++ b/cmd/agent_chat_standalone.go
@@ -211,6 +211,7 @@ func bootstrapStandaloneAgent(cfg *config.Config, agentName string) (*agent.Loop
Model: agentCfg.Model,
ContextWindow: agentCfg.ContextWindow,
MaxIterations: agentCfg.MaxToolIterations,
+ MaxToolCalls: agentCfg.MaxToolCalls,
Workspace: workspace,
Bus: msgBus,
Sessions: sessStore,
diff --git a/cmd/gateway.go b/cmd/gateway.go
index c9f37dd8..1f4b0171 100644
--- a/cmd/gateway.go
+++ b/cmd/gateway.go
@@ -909,7 +909,38 @@ func runGateway() {
if managedStores != nil {
consumerTeamStore = managedStores.Teams
}
- go consumeInboundMessages(ctx, msgBus, agentRouter, cfg, sched, channelMgr, consumerTeamStore)
+
+ // Quota checker (managed mode only): enforces per-user/group request limits.
+ // Merge per-group quotas from channel configs into gateway.quota.groups.
+ config.MergeChannelGroupQuotas(cfg)
+ var quotaChecker *channels.QuotaChecker
+ if managedStores != nil && cfg.Gateway.Quota != nil && cfg.Gateway.Quota.Enabled {
+ quotaChecker = channels.NewQuotaChecker(managedStores.DB, *cfg.Gateway.Quota)
+ defer quotaChecker.Stop()
+ slog.Info("channel quota enabled",
+ "default_hour", cfg.Gateway.Quota.Default.Hour,
+ "default_day", cfg.Gateway.Quota.Default.Day,
+ "default_week", cfg.Gateway.Quota.Default.Week,
+ )
+ }
+
+ // Reload quota config on config changes via pub/sub.
+ if quotaChecker != nil {
+ msgBus.Subscribe("quota-config-reload", func(evt bus.Event) {
+ if evt.Name != bus.TopicConfigChanged {
+ return
+ }
+ updatedCfg, ok := evt.Payload.(*config.Config)
+ if !ok || updatedCfg.Gateway.Quota == nil {
+ return
+ }
+ config.MergeChannelGroupQuotas(updatedCfg)
+ quotaChecker.UpdateConfig(*updatedCfg.Gateway.Quota)
+ slog.Info("quota config reloaded via pub/sub")
+ })
+ }
+
+ go consumeInboundMessages(ctx, msgBus, agentRouter, cfg, sched, channelMgr, consumerTeamStore, quotaChecker)
go func() {
sig := <-sigCh
diff --git a/cmd/gateway_agents.go b/cmd/gateway_agents.go
index 39d5bb24..23e133b0 100644
--- a/cmd/gateway_agents.go
+++ b/cmd/gateway_agents.go
@@ -94,6 +94,7 @@ func createAgentLoop(agentID string, cfg *config.Config, router *agent.Router, p
Model: agentCfg.Model,
ContextWindow: agentCfg.ContextWindow,
MaxIterations: agentCfg.MaxToolIterations,
+ MaxToolCalls: agentCfg.MaxToolCalls,
Workspace: workspace,
Bus: msgBus,
Sessions: sess,
diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go
index 7e9e620c..f83105e4 100644
--- a/cmd/gateway_consumer.go
+++ b/cmd/gateway_consumer.go
@@ -41,7 +41,7 @@ func makeSchedulerRunFunc(agents *agent.Router, cfg *config.Config) scheduler.Ru
// and routes them through the scheduler/agent loop, then publishes the response back.
// Also handles subagent announcements: routes them through the parent agent's session
// (matching TS subagent-announce.ts pattern) so the agent can reformulate for the user.
-func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents *agent.Router, cfg *config.Config, sched *scheduler.Scheduler, channelMgr *channels.Manager, teamStore store.TeamStore) {
+func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents *agent.Router, cfg *config.Config, sched *scheduler.Scheduler, channelMgr *channels.Manager, teamStore store.TeamStore, quotaChecker *channels.QuotaChecker) {
slog.Info("inbound message consumer started")
// Inbound message deduplication (matching TS src/infra/dedupe.ts + inbound-dedupe.ts).
@@ -66,7 +66,8 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
}
}
- if _, err := agents.Get(agentID); err != nil {
+ agentLoop, err := agents.Get(agentID)
+ if err != nil {
slog.Warn("inbound: agent not found", "agent", agentID, "channel", msg.Channel)
return
}
@@ -112,6 +113,28 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
userID = fmt.Sprintf("group:%s:%s", msg.Channel, groupID)
}
+ // --- Quota check (managed mode only) ---
+ if quotaChecker != nil {
+ qResult := quotaChecker.Check(ctx, userID, msg.Channel, agentLoop.ProviderName())
+ if !qResult.Allowed {
+ slog.Warn("security.quota_exceeded",
+ "user_id", userID,
+ "channel", msg.Channel,
+ "window", qResult.Window,
+ "used", qResult.Used,
+ "limit", qResult.Limit,
+ )
+ msgBus.PublishOutbound(bus.OutboundMessage{
+ Channel: msg.Channel,
+ ChatID: msg.ChatID,
+ Content: formatQuotaExceeded(qResult),
+ Metadata: msg.Metadata,
+ })
+ return
+ }
+ quotaChecker.Increment(userID)
+ }
+
slog.Info("inbound: scheduling message (main lane)",
"channel", msg.Channel,
"chat_id", msg.ChatID,
diff --git a/cmd/gateway_errors.go b/cmd/gateway_errors.go
index e1bad463..95ee5db1 100644
--- a/cmd/gateway_errors.go
+++ b/cmd/gateway_errors.go
@@ -1,8 +1,11 @@
package cmd
import (
+ "fmt"
"log/slog"
"strings"
+
+ "github.com/nextlevelbuilder/goclaw/internal/channels"
)
// Matching TS pi-embedded-helpers/errors.ts error classification.
@@ -95,3 +98,10 @@ func containsAny(s string, substrs ...string) bool {
}
return false
}
+
+// formatQuotaExceeded formats a user-friendly quota exceeded message.
+func formatQuotaExceeded(result channels.QuotaResult) string {
+ labels := map[string]string{"hour": "Hourly", "day": "Daily", "week": "Weekly"}
+ return fmt.Sprintf("⚠️ %s request limit reached (%d/%d). Please try again later.",
+ labels[result.Window], result.Used, result.Limit)
+}
diff --git a/cmd/gateway_methods.go b/cmd/gateway_methods.go
index e922de3f..5e4182da 100644
--- a/cmd/gateway_methods.go
+++ b/cmd/gateway_methods.go
@@ -19,7 +19,7 @@ func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore
methods.NewChatMethods(agents, sessStore, isManaged, server.RateLimiter()).Register(router)
methods.NewAgentsMethods(agents, cfg, cfgPath, workspace, agentStore, isManaged, contextFileInterceptor).Register(router)
methods.NewSessionsMethods(sessStore).Register(router)
- methods.NewConfigMethods(cfg, cfgPath, isManaged, configSecretsStore).Register(router)
+ methods.NewConfigMethods(cfg, cfgPath, isManaged, configSecretsStore, msgBus).Register(router)
// Phase 2: Skills (uses SkillStore interface — PG or File)
methods.NewSkillsMethods(skillStore).Register(router)
diff --git a/internal/agent/loop.go b/internal/agent/loop.go
index 73a099cd..78fd7679 100644
--- a/internal/agent/loop.go
+++ b/internal/agent/loop.go
@@ -52,6 +52,7 @@ type Loop struct {
model string
contextWindow int
maxIterations int
+ maxToolCalls int
workspace string
eventPub bus.EventPublisher // currently unused by Loop; kept for future use
@@ -127,6 +128,7 @@ type LoopConfig struct {
Model string
ContextWindow int
MaxIterations int
+ MaxToolCalls int
Workspace string
Bus bus.EventPublisher
Sessions store.SessionStore
@@ -214,6 +216,7 @@ func NewLoop(cfg LoopConfig) *Loop {
model: cfg.Model,
contextWindow: cfg.ContextWindow,
maxIterations: cfg.MaxIterations,
+ maxToolCalls: cfg.MaxToolCalls,
workspace: cfg.Workspace,
eventPub: cfg.Bus,
sessions: cfg.Sessions,
@@ -593,6 +596,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error)
var loopDetector toolLoopState // detects repeated no-progress tool calls
var totalUsage providers.Usage
iteration := 0
+ totalToolCalls := 0
var finalContent string
var asyncToolCalls []string // track async spawn tool names for fallback
var mediaResults []MediaResult // media files from tool MEDIA: results
@@ -779,6 +783,19 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error)
}
}
+ // Tool budget check: soft stop when total tool calls exceed the per-agent limit.
+ // Same pattern as maxIterations — no error thrown, LLM summarizes and returns.
+ totalToolCalls += len(resp.ToolCalls)
+ if l.maxToolCalls > 0 && totalToolCalls > l.maxToolCalls {
+ slog.Warn("security.tool_budget_exceeded",
+ "agent", l.id, "total", totalToolCalls, "limit", l.maxToolCalls)
+ messages = append(messages, providers.Message{
+ Role: "user",
+ Content: fmt.Sprintf("[System] Tool call budget reached (%d/%d). Do NOT call any more tools. Summarize results so far and respond to the user.", totalToolCalls, l.maxToolCalls),
+ })
+ continue // one more LLM call for summarization, then loop exits (no tool calls)
+ }
+
// Execute tool calls (parallel when multiple, sequential when single)
if len(resp.ToolCalls) == 1 {
// Single tool: sequential — no goroutine overhead
@@ -1270,3 +1287,11 @@ func sanitizePathSegment(s string) string {
func (l *Loop) InvalidateUserWorkspace(userID string) {
l.userWorkspaces.Delete(userID)
}
+
+// ProviderName returns the name of this agent's LLM provider (e.g. "anthropic", "openai").
+func (l *Loop) ProviderName() string {
+ if l.provider == nil {
+ return ""
+ }
+ return l.provider.Name()
+}
diff --git a/internal/agent/types.go b/internal/agent/types.go
index 4353ae6d..0ed2929a 100644
--- a/internal/agent/types.go
+++ b/internal/agent/types.go
@@ -9,4 +9,5 @@ type Agent interface {
Run(ctx context.Context, req RunRequest) (*RunResult, error)
IsRunning() bool
Model() string
+ ProviderName() string
}
diff --git a/internal/bus/types.go b/internal/bus/types.go
index 44f2fcfa..8bfdec45 100644
--- a/internal/bus/types.go
+++ b/internal/bus/types.go
@@ -69,6 +69,7 @@ const (
TopicCacheGroupFileWriters = "cache:group_file_writers"
TopicCacheSkillGrants = "cache:skill_grants"
TopicChannelStreaming = "channel-streaming"
+ TopicConfigChanged = "config:changed"
)
// CacheInvalidatePayload signals cache layers to evict stale entries.
diff --git a/internal/channels/quota.go b/internal/channels/quota.go
new file mode 100644
index 00000000..a5b8842e
--- /dev/null
+++ b/internal/channels/quota.go
@@ -0,0 +1,198 @@
+package channels
+
+import (
+ "context"
+ "database/sql"
+ "log/slog"
+ "sync"
+ "time"
+
+ "github.com/nextlevelbuilder/goclaw/internal/config"
+)
+
+// QuotaResult is returned by QuotaChecker.Check.
+type QuotaResult struct {
+ Allowed bool // whether the request is within quota
+ Window string // which window was exceeded: "hour", "day", "week"
+ Used int // current usage in the exceeded window
+ Limit int // configured limit for the exceeded window
+}
+
+// quotaCounts holds cached request counts for a user.
+type quotaCounts struct {
+ hour, day, week int
+ fetchedAt time.Time
+}
+
+// QuotaChecker enforces per-user/group request quotas by counting top-level
+// traces in the database. Results are cached in-memory for cacheTTL.
+// Managed mode only — nil-safe (standalone mode passes nil).
+type QuotaChecker struct {
+ db *sql.DB
+ config config.QuotaConfig
+ cache map[string]*quotaCounts
+ cacheTTL time.Duration
+ mu sync.RWMutex
+ stopCh chan struct{}
+}
+
+// NewQuotaChecker creates a quota checker backed by the traces table.
+// Starts a background goroutine to evict stale cache entries.
+func NewQuotaChecker(db *sql.DB, cfg config.QuotaConfig) *QuotaChecker {
+ qc := &QuotaChecker{
+ db: db,
+ config: cfg,
+ cache: make(map[string]*quotaCounts),
+ cacheTTL: 60 * time.Second,
+ stopCh: make(chan struct{}),
+ }
+ go qc.cleanupLoop()
+ return qc
+}
+
+// Stop shuts down the background cleanup goroutine.
+func (qc *QuotaChecker) Stop() {
+ close(qc.stopCh)
+}
+
+// UpdateConfig replaces the quota configuration (e.g., after config reload).
+func (qc *QuotaChecker) UpdateConfig(cfg config.QuotaConfig) {
+ qc.mu.Lock()
+ defer qc.mu.Unlock()
+ qc.config = cfg
+}
+
+// Check verifies if a user is within their quota limits.
+// Returns QuotaResult with Allowed=true if all windows are within limits.
+func (qc *QuotaChecker) Check(ctx context.Context, userID, channel, provider string) QuotaResult {
+ window := qc.resolveWindow(userID, channel, provider)
+ if window.IsZero() {
+ return QuotaResult{Allowed: true}
+ }
+
+ counts := qc.getCounts(ctx, userID)
+
+ // Check each window — report the first exceeded
+ if window.Hour > 0 && counts.hour >= window.Hour {
+ return QuotaResult{Allowed: false, Window: "hour", Used: counts.hour, Limit: window.Hour}
+ }
+ if window.Day > 0 && counts.day >= window.Day {
+ return QuotaResult{Allowed: false, Window: "day", Used: counts.day, Limit: window.Day}
+ }
+ if window.Week > 0 && counts.week >= window.Week {
+ return QuotaResult{Allowed: false, Window: "week", Used: counts.week, Limit: window.Week}
+ }
+
+ return QuotaResult{Allowed: true}
+}
+
+// Increment optimistically bumps cached counts after a request is accepted.
+// This prevents multiple rapid requests from bypassing the quota within the
+// cache TTL window.
+func (qc *QuotaChecker) Increment(userID string) {
+ qc.mu.Lock()
+ defer qc.mu.Unlock()
+ if c, ok := qc.cache[userID]; ok {
+ c.hour++
+ c.day++
+ c.week++
+ }
+}
+
+// resolveWindow returns the effective quota window for a user, applying
+// config merge priority: Groups > Channels > Providers > Default.
+func (qc *QuotaChecker) resolveWindow(userID, channel, provider string) config.QuotaWindow {
+ qc.mu.RLock()
+ cfg := qc.config
+ qc.mu.RUnlock()
+
+ // Most specific: per-user/group override
+ if w, ok := cfg.Groups[userID]; ok && !w.IsZero() {
+ return w
+ }
+
+ // Channel-level override
+ if channel != "" {
+ if w, ok := cfg.Channels[channel]; ok && !w.IsZero() {
+ return w
+ }
+ }
+
+ // Provider-level override
+ if provider != "" {
+ if w, ok := cfg.Providers[provider]; ok && !w.IsZero() {
+ return w
+ }
+ }
+
+ return cfg.Default
+}
+
+// getCounts returns cached or fresh counts for a user.
+func (qc *QuotaChecker) getCounts(ctx context.Context, userID string) quotaCounts {
+ qc.mu.RLock()
+ if c, ok := qc.cache[userID]; ok && time.Since(c.fetchedAt) < qc.cacheTTL {
+ counts := *c
+ qc.mu.RUnlock()
+ return counts
+ }
+ qc.mu.RUnlock()
+
+ // Cache miss — query DB
+ counts := qc.queryDB(ctx, userID)
+
+ qc.mu.Lock()
+ qc.cache[userID] = &counts
+ qc.mu.Unlock()
+
+ return counts
+}
+
+// queryDB counts top-level traces for a user across time windows in a single query.
+// Uses idx_traces_quota partial index: (user_id, created_at DESC) WHERE parent_trace_id IS NULL.
+func (qc *QuotaChecker) queryDB(ctx context.Context, userID string) quotaCounts {
+ now := time.Now().UTC()
+ hourAgo := now.Add(-1 * time.Hour)
+ dayAgo := now.Add(-24 * time.Hour)
+ weekAgo := now.Add(-7 * 24 * time.Hour)
+
+ var counts quotaCounts
+ counts.fetchedAt = now
+
+ err := qc.db.QueryRowContext(ctx, `
+ SELECT
+ COUNT(*) FILTER (WHERE created_at >= $2) AS hour_count,
+ COUNT(*) FILTER (WHERE created_at >= $3) AS day_count,
+ COUNT(*) FILTER (WHERE created_at >= $4) AS week_count
+ FROM traces
+ WHERE user_id = $1 AND parent_trace_id IS NULL AND created_at >= $4`,
+ userID, hourAgo, dayAgo, weekAgo,
+ ).Scan(&counts.hour, &counts.day, &counts.week)
+ if err != nil {
+ slog.Warn("quota: failed to query trace counts", "user_id", userID, "error", err)
+ }
+
+ return counts
+}
+
+// cleanupLoop periodically evicts stale cache entries.
+func (qc *QuotaChecker) cleanupLoop() {
+ ticker := time.NewTicker(2 * time.Minute)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-qc.stopCh:
+ return
+ case <-ticker.C:
+ qc.mu.Lock()
+ staleThreshold := time.Now().Add(-5 * time.Minute)
+ for k, v := range qc.cache {
+ if v.fetchedAt.Before(staleThreshold) {
+ delete(qc.cache, k)
+ }
+ }
+ qc.mu.Unlock()
+ }
+ }
+}
diff --git a/internal/config/config.go b/internal/config/config.go
index b1def5ea..fb136861 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -119,6 +119,7 @@ type AgentDefaults struct {
Temperature float64 `json:"temperature"`
MaxToolIterations int `json:"max_tool_iterations"`
ContextWindow int `json:"context_window"`
+ MaxToolCalls int `json:"max_tool_calls,omitempty"` // max total tool calls per run (0 = unlimited, default 25)
AgentType string `json:"agent_type,omitempty"` // "open" (default) or "predefined"
Subagents *SubagentsConfig `json:"subagents,omitempty"`
Sandbox *SandboxConfig `json:"sandbox,omitempty"`
@@ -376,6 +377,7 @@ type AgentSpec struct {
Temperature float64 `json:"temperature,omitempty"`
MaxToolIterations int `json:"max_tool_iterations,omitempty"`
ContextWindow int `json:"context_window,omitempty"`
+ MaxToolCalls int `json:"max_tool_calls,omitempty"` // per-agent override
AgentType string `json:"agent_type,omitempty"` // "open" or "predefined"
Skills []string `json:"skills,omitempty"` // nil = all skills allowed
Tools *ToolPolicySpec `json:"tools,omitempty"` // per-agent tool policy
diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go
index 5234483d..4b0b7b34 100644
--- a/internal/config/config_channels.go
+++ b/internal/config/config_channels.go
@@ -53,6 +53,7 @@ type TelegramGroupConfig struct {
Tools []string `json:"tools,omitempty"` // tool allow list (nil = all, supports "group:xxx")
SystemPrompt string `json:"system_prompt,omitempty"` // extra system prompt for this group
Topics map[string]*TelegramTopicConfig `json:"topics,omitempty"` // per-topic overrides (key: thread ID string)
+ Quota *QuotaWindow `json:"quota,omitempty"` // per-group quota override
}
// TelegramTopicConfig defines per-topic overrides within a Telegram group.
@@ -178,6 +179,26 @@ func (c *Config) HasAnyProvider() bool {
p.Bailian.APIKey != ""
}
+// QuotaWindow defines request limits per time window. Zero means unlimited.
+type QuotaWindow struct {
+ Hour int `json:"hour,omitempty"` // max requests per hour (0 = unlimited)
+ Day int `json:"day,omitempty"` // max requests per day (0 = unlimited)
+ Week int `json:"week,omitempty"` // max requests per week (0 = unlimited)
+}
+
+// IsZero returns true if no limits are set.
+func (w QuotaWindow) IsZero() bool { return w.Hour == 0 && w.Day == 0 && w.Week == 0 }
+
+// QuotaConfig configures per-user/group request quotas (managed mode only).
+// Config merge priority: Groups > Channels > Providers > Default.
+type QuotaConfig struct {
+ Enabled bool `json:"enabled"`
+ Default QuotaWindow `json:"default"`
+ Providers map[string]QuotaWindow `json:"providers,omitempty"` // key = provider name (e.g. "anthropic")
+ Channels map[string]QuotaWindow `json:"channels,omitempty"` // key = channel name (e.g. "telegram")
+ Groups map[string]QuotaWindow `json:"groups,omitempty"` // key = userID (e.g. "group:telegram:-100123")
+}
+
// GatewayConfig controls the gateway server.
type GatewayConfig struct {
Host string `json:"host"`
@@ -189,6 +210,7 @@ type GatewayConfig struct {
RateLimitRPM int `json:"rate_limit_rpm,omitempty"` // rate limit: requests per minute per user (default 20, 0 = disabled)
InjectionAction string `json:"injection_action,omitempty"` // prompt injection action: "log", "warn" (default), "block", "off"
InboundDebounceMs int `json:"inbound_debounce_ms,omitempty"` // merge rapid messages from same sender (default 1000ms, -1 = disabled)
+ Quota *QuotaConfig `json:"quota,omitempty"` // per-user/group request quotas (managed mode only)
}
// ToolsConfig controls tool availability, policy, and web search.
@@ -332,3 +354,24 @@ type TtsMiniMaxConfig struct {
Model string `json:"model,omitempty"` // default "speech-02-hd"
VoiceID string `json:"voice_id,omitempty"` // default "Wise_Woman"
}
+
+// MergeChannelGroupQuotas merges per-group quota overrides from channel configs
+// (e.g., channels.telegram.groups[chatID].quota) into gateway.quota.groups.
+// This allows per-group quotas to be set at the channel level and picked up
+// by the QuotaChecker at runtime.
+func MergeChannelGroupQuotas(cfg *Config) {
+ if cfg.Gateway.Quota == nil {
+ return
+ }
+ if cfg.Gateway.Quota.Groups == nil {
+ cfg.Gateway.Quota.Groups = make(map[string]QuotaWindow)
+ }
+
+ // Telegram per-group quotas
+ for chatID, groupCfg := range cfg.Channels.Telegram.Groups {
+ if groupCfg != nil && groupCfg.Quota != nil && !groupCfg.Quota.IsZero() {
+ key := "group:telegram:" + chatID
+ cfg.Gateway.Quota.Groups[key] = *groupCfg.Quota
+ }
+ }
+}
diff --git a/internal/config/config_load.go b/internal/config/config_load.go
index 02729206..54148e0f 100644
--- a/internal/config/config_load.go
+++ b/internal/config/config_load.go
@@ -24,6 +24,7 @@ func Default() *Config {
MaxTokens: 8192,
Temperature: 0.7,
MaxToolIterations: 20,
+ MaxToolCalls: 25,
ContextWindow: 200000,
Subagents: &SubagentsConfig{
MaxConcurrent: 20,
@@ -54,6 +55,7 @@ func Default() *Config {
Security: "full",
Ask: "off",
},
+ RateLimitPerHour: 150,
},
Sessions: SessionsConfig{
Storage: "~/.goclaw/sessions",
@@ -310,6 +312,9 @@ func (c *Config) ResolveAgent(agentID string) AgentDefaults {
if spec.ContextWindow > 0 {
d.ContextWindow = spec.ContextWindow
}
+ if spec.MaxToolCalls > 0 {
+ d.MaxToolCalls = spec.MaxToolCalls
+ }
if spec.Workspace != "" {
d.Workspace = spec.Workspace
}
diff --git a/internal/gateway/methods/config.go b/internal/gateway/methods/config.go
index 6e899256..916ab310 100644
--- a/internal/gateway/methods/config.go
+++ b/internal/gateway/methods/config.go
@@ -7,6 +7,7 @@ import (
"github.com/titanous/json5"
+ "github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/config"
"github.com/nextlevelbuilder/goclaw/internal/gateway"
"github.com/nextlevelbuilder/goclaw/internal/store"
@@ -20,10 +21,11 @@ type ConfigMethods struct {
cfgPath string
managedMode bool
secretsStore store.ConfigSecretsStore // nil in standalone mode
+ eventBus bus.EventPublisher // nil-safe; broadcasts config change events
}
-func NewConfigMethods(cfg *config.Config, cfgPath string, managedMode bool, secretsStore store.ConfigSecretsStore) *ConfigMethods {
- return &ConfigMethods{cfg: cfg, cfgPath: cfgPath, managedMode: managedMode, secretsStore: secretsStore}
+func NewConfigMethods(cfg *config.Config, cfgPath string, managedMode bool, secretsStore store.ConfigSecretsStore, eventBus bus.EventPublisher) *ConfigMethods {
+ return &ConfigMethods{cfg: cfg, cfgPath: cfgPath, managedMode: managedMode, secretsStore: secretsStore, eventBus: eventBus}
}
func (m *ConfigMethods) Register(router *gateway.MethodRouter) {
@@ -94,6 +96,7 @@ func (m *ConfigMethods) handleApply(ctx context.Context, client *gateway.Client,
}
}
m.cfg.ApplyEnvOverrides()
+ m.broadcastChanged()
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{
"ok": true,
@@ -168,6 +171,7 @@ func (m *ConfigMethods) handlePatch(ctx context.Context, client *gateway.Client,
}
}
m.cfg.ApplyEnvOverrides()
+ m.broadcastChanged()
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{
"ok": true,
@@ -178,6 +182,13 @@ func (m *ConfigMethods) handlePatch(ctx context.Context, client *gateway.Client,
}))
}
+// broadcastChanged notifies subscribers that config has been updated.
+func (m *ConfigMethods) broadcastChanged() {
+ if m.eventBus != nil {
+ m.eventBus.Broadcast(bus.Event{Name: bus.TopicConfigChanged, Payload: m.cfg})
+ }
+}
+
// handleSchema returns the config JSON schema for UI form generation.
// Matching TS config.schema (src/gateway/server-methods/config.ts:276-289).
func (m *ConfigMethods) handleSchema(_ context.Context, client *gateway.Client, req *protocol.RequestFrame) {
diff --git a/internal/store/pg/factory.go b/internal/store/pg/factory.go
index 27b01d77..8b2a38f8 100644
--- a/internal/store/pg/factory.go
+++ b/internal/store/pg/factory.go
@@ -23,6 +23,7 @@ func NewPGStores(cfg store.StoreConfig) (*store.Stores, error) {
skillsDir = config.ExpandHome(skillsDir)
return &store.Stores{
+ DB: db,
Sessions: NewPGSessionStore(db),
Memory: NewPGMemoryStore(db, memCfg),
Cron: NewPGCronStore(db),
diff --git a/internal/store/stores.go b/internal/store/stores.go
index bb27aba3..0ccd3020 100644
--- a/internal/store/stores.go
+++ b/internal/store/stores.go
@@ -1,8 +1,11 @@
package store
+import "database/sql"
+
// Stores is the top-level container for all storage backends.
// In standalone mode, managed-only stores (Agents, Providers, Tracing, MCP) are nil.
type Stores struct {
+ DB *sql.DB // underlying connection (nil in standalone mode)
Sessions SessionStore
Memory MemoryStore
Cron CronStore
diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go
index 0f84be10..d1af2c8d 100644
--- a/internal/upgrade/version.go
+++ b/internal/upgrade/version.go
@@ -2,4 +2,4 @@ package upgrade
// RequiredSchemaVersion is the schema migration version this binary requires.
// Bump this whenever adding a new SQL migration file.
-const RequiredSchemaVersion uint = 8
+const RequiredSchemaVersion uint = 9
diff --git a/migrations/000009_add_quota_index.down.sql b/migrations/000009_add_quota_index.down.sql
new file mode 100644
index 00000000..ad6b4013
--- /dev/null
+++ b/migrations/000009_add_quota_index.down.sql
@@ -0,0 +1 @@
+DROP INDEX CONCURRENTLY IF EXISTS idx_traces_quota;
diff --git a/migrations/000009_add_quota_index.up.sql b/migrations/000009_add_quota_index.up.sql
new file mode 100644
index 00000000..cdef585f
--- /dev/null
+++ b/migrations/000009_add_quota_index.up.sql
@@ -0,0 +1,5 @@
+-- Partial index for quota checker: efficiently counts top-level traces per user in time windows.
+-- Eliminates parent_trace_id IS NULL post-filter (89% of traces are top-level).
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_traces_quota
+ON traces (user_id, created_at DESC)
+WHERE parent_trace_id IS NULL AND user_id IS NOT NULL;
diff --git a/ui/web/src/pages/config/config-page.tsx b/ui/web/src/pages/config/config-page.tsx
index bd47af7c..e6933f6d 100644
--- a/ui/web/src/pages/config/config-page.tsx
+++ b/ui/web/src/pages/config/config-page.tsx
@@ -23,6 +23,7 @@ import { TtsSection } from "./sections/tts-section";
import { CronSection } from "./sections/cron-section";
import { TelemetrySection } from "./sections/telemetry-section";
import { BindingsSection } from "./sections/bindings-section";
+import { QuotaSection } from "./sections/quota-section";
export function ConfigPage() {
const { config, hash, configPath, loading, saving, error, refresh, applyRaw, patch } = useConfig();
@@ -41,7 +42,7 @@ export function ConfigPage() {
}
}, [config]);
- const handleSave = async () => {
+ const handleRawSave = async () => {
setSaveError(null);
try {
await applyRaw(rawText);
@@ -82,6 +83,8 @@ export function ConfigPage() {
);
}
+ const isManaged = (config.database as any)?.mode === "managed";
+
return (
-
-
- UI
+
+
+ General
+ {isManaged && Quota}
+ Agents
+ Tools
+ Connections
+ Advanced
+
Raw Editor
-
-
+
+ patch({ gateway: v })}
+ saving={saving}
+ />
-
+ {isManaged && (
+
+ patch({ gateway: v })}
+ saving={saving}
+ />
+
+ )}
+
+
+ {isManaged ? (
+
+ ) : (
+ patch({ providers: v })}
+ saving={saving}
+ />
+ )}
+ patch({ agents: v })}
+ saving={saving}
+ />
+
+
+
+ patch({ tools: v })}
+ saving={saving}
+ />
+
+
+
+ patch({ sessions: v })}
+ saving={saving}
+ />
+ {isManaged ? (
+
+ ) : (
+ patch({ channels: v })}
+ saving={saving}
+ />
+ )}
+
+
+
+
+ patch({ cron: v })}
+ saving={saving}
+ />
+ patch({ telemetry: v })}
+ saving={saving}
+ />
+ patch({ bindings: v })}
+ saving={saving}
+ />
+
+
+