DBT/clawgo
1
0
Fork 0
mirror of https://github.com/YspCoder/clawgo.git synced 2026-04-15 01:37:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix provider config hot reload

Browse Source
This commit is contained in:
LPF
2026-03-19 22:46:37 +08:00
parent 971ac8ddb9
commit e9bd7891cc
11 changed files with 93 additions and 548 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
pkg/agent/loop.go
View File

@@ -67,6 +67,9 @@ type AgentLoop struct {
subagentDigestMu sync.Mutex
subagentDigestDelay time.Duration
subagentDigests map[string]*subagentDigestState
runMu sync.Mutex
runCancel context.CancelFunc
runWG sync.WaitGroup
}
type providerCandidate struct {
@@ -403,19 +406,34 @@ func (al *AgentLoop) readSubagentPromptFile(relPath string) string {
}
func (al *AgentLoop) Run(ctx context.Context) error {
al.runMu.Lock()
if al.runCancel != nil {
al.runMu.Unlock()
return fmt.Errorf("agent loop already running")
}
runCtx, cancel := context.WithCancel(ctx)
al.runCancel = cancel
al.running = true
al.runMu.Unlock()
defer func() {
al.runMu.Lock()
al.running = false
al.runCancel = nil
al.runMu.Unlock()
}()
shards := al.buildSessionShards(ctx)
shards := al.buildSessionShards(runCtx)
defer func() {
for _, ch := range shards {
close(ch)
}
al.runWG.Wait()
}()
for al.running {
msg, ok := al.bus.ConsumeInbound(ctx)
msg, ok := al.bus.ConsumeInbound(runCtx)
if !ok {
if ctx.Err() != nil {
if runCtx.Err() != nil {
return nil
}
continue
@@ -423,7 +441,7 @@ func (al *AgentLoop) Run(ctx context.Context) error {
idx := sessionShardIndex(msg.SessionKey, len(shards))
select {
case shards[idx] <- msg:
case <-ctx.Done():
case <-runCtx.Done():
return nil
}
}
@@ -432,7 +450,14 @@ func (al *AgentLoop) Run(ctx context.Context) error {
}
func (al *AgentLoop) Stop() {
al.runMu.Lock()
cancel := al.runCancel
al.runMu.Unlock()
if cancel != nil {
cancel()
}
al.running = false
al.runWG.Wait()
}
func (al *AgentLoop) buildSessionShards(ctx context.Context) []chan bus.InboundMessage {
@@ -440,7 +465,9 @@ func (al *AgentLoop) buildSessionShards(ctx context.Context) []chan bus.InboundM
shards := make([]chan bus.InboundMessage, count)
for i := 0; i < count; i++ {
shards[i] = make(chan bus.InboundMessage, 64)
al.runWG.Add(1)
go func(ch <-chan bus.InboundMessage) {
defer al.runWG.Done()
for msg := range ch {
al.processInbound(ctx, msg)
}

10
pkg/api/server.go
View File

@@ -48,7 +48,7 @@ type Server struct {
logFilePath string
onChat func(ctx context.Context, sessionKey, content string) (string, error)
onChatHistory func(sessionKey string) []map[string]interface{}
onConfigAfter func() error
onConfigAfter func(forceRuntimeReload bool) error
onCron func(action string, args map[string]interface{}) (interface{}, error)
onToolsCatalog func() interface{}
whatsAppBridge *channels.WhatsAppBridgeService
@@ -85,7 +85,7 @@ func (s *Server) SetChatHandler(fn func(ctx context.Context, sessionKey, content
func (s *Server) SetChatHistoryHandler(fn func(sessionKey string) []map[string]interface{}) {
s.onChatHistory = fn
}
func (s *Server) SetConfigAfterHook(fn func() error) { s.onConfigAfter = fn }
func (s *Server) SetConfigAfterHook(fn func(forceRuntimeReload bool) error) { s.onConfigAfter = fn }
func (s *Server) SetCronHandler(fn func(action string, args map[string]interface{}) (interface{}, error)) {
s.onCron = fn
}
@@ -414,7 +414,7 @@ func (s *Server) persistWebUIConfig(cfg *cfgpkg.Config) error {
return err
}
if s.onConfigAfter != nil {
return s.onConfigAfter()
return s.onConfigAfter(false)
}
return requestSelfReloadSignal()
}
@@ -978,7 +978,9 @@ func (s *Server) saveProviderConfig(cfg *cfgpkg.Config, name string, pc cfgpkg.P
return err
}
if s.onConfigAfter != nil {
if err := s.onConfigAfter(); err != nil {
// Provider updates may only change external credential file contents,
// so force a runtime rebuild even when config JSON remains identical.
if err := s.onConfigAfter(true); err != nil {
return err
}
} else {

51
pkg/api/server_test.go
View File

@@ -100,7 +100,10 @@ func TestHandleWebUIConfigPostSavesRawConfig(t *testing.T) {
srv := NewServer("127.0.0.1", 0, "")
srv.SetConfigPath(cfgPath)
hookCalled := 0
srv.SetConfigAfterHook(func() error {
srv.SetConfigAfterHook(func(forceRuntimeReload bool) error {
if forceRuntimeReload {
t.Fatalf("expected raw config save to use non-forced reload")
}
hookCalled++
return nil
})
@@ -150,7 +153,12 @@ func TestHandleWebUIConfigPostSavesNormalizedConfig(t *testing.T) {
srv := NewServer("127.0.0.1", 0, "")
srv.SetConfigPath(cfgPath)
srv.SetConfigAfterHook(func() error { return nil })
srv.SetConfigAfterHook(func(forceRuntimeReload bool) error {
if forceRuntimeReload {
t.Fatalf("expected normalized config save to use non-forced reload")
}
return nil
})
req := httptest.NewRequest(http.MethodPost, "/api/config?mode=normalized", strings.NewReader(`{"core":{"gateway":{"host":"127.0.0.1","port":18790},"tools":{"shell_enabled":false,"mcp_enabled":true}},"runtime":{"router":{"enabled":true,"strategy":"rules_first","max_hops":2,"default_timeout_sec":90},"providers":{"openai":{"api_base":"https://api.openai.com/v1","auth":"bearer","timeout_sec":150}}}}`))
req.Header.Set("Content-Type", "application/json")
@@ -172,6 +180,45 @@ func TestHandleWebUIConfigPostSavesNormalizedConfig(t *testing.T) {
}
}
func TestSaveProviderConfigForcesRuntimeReload(t *testing.T) {
t.Parallel()
tmp := t.TempDir()
cfgPath := filepath.Join(tmp, "config.json")
cfg := cfgpkg.DefaultConfig()
cfg.Logging.Enabled = false
cfg.Models.Providers["openai"] = cfgpkg.ProviderConfig{
APIBase: "https://api.openai.com/v1",
Auth: "oauth",
Models: []string{"gpt-5"},
TimeoutSec: 120,
OAuth: cfgpkg.ProviderOAuthConfig{
Provider: "codex",
CredentialFile: filepath.Join(tmp, "auth.json"),
},
}
if err := cfgpkg.SaveConfig(cfgPath, cfg); err != nil {
t.Fatalf("save config: %v", err)
}
srv := NewServer("127.0.0.1", 0, "")
srv.SetConfigPath(cfgPath)
forced := false
srv.SetConfigAfterHook(func(forceRuntimeReload bool) error {
forced = forceRuntimeReload
return nil
})
pc := cfg.Models.Providers["openai"]
if err := srv.saveProviderConfig(cfg, "openai", pc); err != nil {
t.Fatalf("save provider config: %v", err)
}
if !forced {
t.Fatalf("expected provider config save to force runtime reload")
}
}
func TestWithCORSEchoesPreflightHeaders(t *testing.T) {
t.Parallel()