mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-17 06:14:48 +00:00
fix(heartbeat): apply provider override from heartbeat config
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.
This commit is contained in:
1 parent
175e05287a
commit
5c8e0b7af2
4 files changed
+62
-31
No files matched your search
+8
-6
@@ -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))
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -427,8 +427,9 @@ type RunRequest struct {
|
||||
TraceName string // override trace name (default: "chat <agentID>")
|
||||
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
|
||||
|
||||
@@ -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"},
|
||||
|
||||
Reference in new issue
Block a user