From 5c8e0b7af27b69bd35981ba9a4f8b3a37377fe7d Mon Sep 17 00:00:00 2001 From: viettranx Date: Sat, 28 Mar 2026 22:48:27 +0700 Subject: [PATCH] fix(heartbeat): apply provider override from heartbeat config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Heartbeat's ProviderID was stored in DB but never used during execution — the agent's default provider always won. Add ProviderOverride to RunRequest, resolve hb.ProviderID in the ticker, and use the override in the agent loop. --- cmd/gateway.go | 14 ++++---- internal/agent/loop.go | 12 ++++--- internal/agent/loop_types.go | 5 +-- internal/heartbeat/ticker.go | 62 +++++++++++++++++++++++++----------- 4 files changed, 62 insertions(+), 31 deletions(-) diff --git a/cmd/gateway.go b/cmd/gateway.go index e571a662..5a6789ac 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -872,12 +872,14 @@ func runGateway() { // Start heartbeat ticker (routes through scheduler's cron lane) heartbeatTicker := heartbeat.NewTicker(heartbeat.TickerConfig{ - Store: pgStores.Heartbeats, - Agents: pgStores.Agents, - Sessions: pgStores.Sessions, - MsgBus: msgBus, - Sched: sched, - RunAgent: makeHeartbeatRunFn(sched), + Store: pgStores.Heartbeats, + Agents: pgStores.Agents, + Sessions: pgStores.Sessions, + ProviderStore: pgStores.Providers, + ProviderReg: providerRegistry, + MsgBus: msgBus, + Sched: sched, + RunAgent: makeHeartbeatRunFn(sched), }) heartbeatTicker.SetOnEvent(func(event store.HeartbeatEvent) { server.BroadcastEvent(*protocol.NewEvent(protocol.EventHeartbeat, event)) diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 785a9841..6b1d2169 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -205,11 +205,15 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) var allowedTools map[string]bool toolDefs, allowedTools, messages = l.buildFilteredTools(&req, hadBootstrap, rs.iteration, maxIter, messages) - // Use per-request model override if set (e.g. heartbeat uses cheaper model). + // Use per-request overrides if set (e.g. heartbeat uses cheaper provider/model). model := l.model + provider := l.provider if req.ModelOverride != "" { model = req.ModelOverride } + if req.ProviderOverride != nil { + provider = req.ProviderOverride + } chatReq := providers.ChatRequest{ Messages: messages, @@ -231,11 +235,11 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) chatReq.Options[providers.OptTenantID] = tid.String() } if l.thinkingLevel != "" && l.thinkingLevel != "off" { - if tc, ok := l.provider.(providers.ThinkingCapable); ok && tc.SupportsThinking() { + if tc, ok := provider.(providers.ThinkingCapable); ok && tc.SupportsThinking() { chatReq.Options[providers.OptThinkingLevel] = l.thinkingLevel } else { slog.Debug("thinking_level ignored: provider does not support thinking", - "provider", l.provider.Name(), "level", l.thinkingLevel) + "provider", provider.Name(), "level", l.thinkingLevel) } } @@ -248,7 +252,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) llmSpanID := l.emitLLMSpanStart(callCtx, llmSpanStart, rs.iteration, messages) if req.Stream { - resp, err = l.provider.ChatStream(callCtx, chatReq, func(chunk providers.StreamChunk) { + resp, err = provider.ChatStream(callCtx, chatReq, func(chunk providers.StreamChunk) { if chunk.Thinking != "" { emitRun(AgentEvent{ Type: protocol.ChatEventThinking, diff --git a/internal/agent/loop_types.go b/internal/agent/loop_types.go index 550f20d6..5eeb54da 100644 --- a/internal/agent/loop_types.go +++ b/internal/agent/loop_types.go @@ -427,8 +427,9 @@ type RunRequest struct { TraceName string // override trace name (default: "chat ") TraceTags []string // additional tags for the trace (e.g. "cron") MaxIterations int // per-request override (0 = use agent default, must be lower) - ModelOverride string // per-request model override (heartbeat uses cheaper model) - LightContext bool // skip loading context files (only inject ExtraSystemPrompt) + ModelOverride string // per-request model override (heartbeat uses cheaper model) + ProviderOverride providers.Provider // per-request provider override (heartbeat uses different provider) + LightContext bool // skip loading context files (only inject ExtraSystemPrompt) // Run classification RunKind string // "delegation", "announce" — empty for user-initiated runs diff --git a/internal/heartbeat/ticker.go b/internal/heartbeat/ticker.go index 885f7f06..e44fd776 100644 --- a/internal/heartbeat/ticker.go +++ b/internal/heartbeat/ticker.go @@ -15,6 +15,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/agent" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/scheduler" "github.com/nextlevelbuilder/goclaw/internal/sessions" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -28,23 +29,27 @@ const ( // TickerConfig holds dependencies for the heartbeat ticker. type TickerConfig struct { - Store store.HeartbeatStore - Agents store.AgentStore - Sessions store.SessionStore // optional: for cleaning up isolated heartbeat sessions - MsgBus *bus.MessageBus - Sched *scheduler.Scheduler - RunAgent func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome + Store store.HeartbeatStore + Agents store.AgentStore + Sessions store.SessionStore // optional: for cleaning up isolated heartbeat sessions + ProviderStore store.ProviderStore + ProviderReg *providers.Registry + MsgBus *bus.MessageBus + Sched *scheduler.Scheduler + RunAgent func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome } // Ticker polls for due heartbeats and runs them through the agent loop. type Ticker struct { - store store.HeartbeatStore - agents store.AgentStore - sessions store.SessionStore - msgBus *bus.MessageBus - sched *scheduler.Scheduler - runAgent func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome - onEvent func(store.HeartbeatEvent) + store store.HeartbeatStore + agents store.AgentStore + sessions store.SessionStore + providerStore store.ProviderStore + providerReg *providers.Registry + msgBus *bus.MessageBus + sched *scheduler.Scheduler + runAgent func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome + onEvent func(store.HeartbeatEvent) wakeCh chan uuid.UUID stopCh chan struct{} @@ -54,12 +59,14 @@ type Ticker struct { // NewTicker creates a new heartbeat ticker. func NewTicker(cfg TickerConfig) *Ticker { return &Ticker{ - store: cfg.Store, - agents: cfg.Agents, - sessions: cfg.Sessions, - msgBus: cfg.MsgBus, - sched: cfg.Sched, - runAgent: cfg.RunAgent, + store: cfg.Store, + agents: cfg.Agents, + sessions: cfg.Sessions, + providerStore: cfg.ProviderStore, + providerReg: cfg.ProviderReg, + msgBus: cfg.MsgBus, + sched: cfg.Sched, + runAgent: cfg.RunAgent, wakeCh: make(chan uuid.UUID, 16), stopCh: make(chan struct{}), } @@ -239,6 +246,22 @@ func (t *Ticker) runOne(ctx context.Context, hb store.AgentHeartbeat) { modelOverride = *hb.Model } + // Provider override: use heartbeat-specific provider if configured. + var providerOverride providers.Provider + if hb.ProviderID != nil && t.providerStore != nil && t.providerReg != nil { + if provData, err := t.providerStore.GetProvider(ctx, *hb.ProviderID); err == nil { + if prov, err := t.providerReg.GetForTenant(ag.TenantID, provData.Name); err == nil { + providerOverride = prov + } else { + slog.Warn("heartbeat.provider_not_in_registry", + "agent", agentKey, "provider_id", hb.ProviderID, "error", err) + } + } else { + slog.Warn("heartbeat.provider_not_found", + "agent", agentKey, "provider_id", hb.ProviderID, "error", err) + } + } + for attempt := range maxAttempts { outCh := t.runAgent(ctx, agent.RunRequest{ SessionKey: sessionKey, @@ -249,6 +272,7 @@ func (t *Ticker) runOne(ctx context.Context, hb store.AgentHeartbeat) { Stream: false, ExtraSystemPrompt: extraSystem, ModelOverride: modelOverride, + ProviderOverride: providerOverride, LightContext: hb.LightContext, TraceName: fmt.Sprintf("Heartbeat [%s]", agentKey), TraceTags: []string{"heartbeat"},