mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 06:17:41 +00:00
refactor(workers): per-tenant provider resolution for background workers
Replace static provider/model fields with per-tenant resolution at processing time. Fixes architectural mismatch where singleton workers used tenant-specific system_configs. Changes: - Add shared ResolveBackgroundProvider() in providerresolve package - Refactor vault enrichWorker, episodic, dreaming workers to resolve provider per-event using registry + systemConfigs - Remove hot-reload machinery (no longer needed) - Update ConsolidationDeps to use Registry/SystemConfigs Fallback chain: background.provider → agent.default_provider → first registered provider.
This commit is contained in:
1 parent
e021934e68
commit
3fa3fb5ebf
17 files changed
+259
-191
No files matched your search
+16
-15
@@ -198,8 +198,8 @@ func runGateway() {
|
||||
KGStore: pgStores.KnowledgeGraph,
|
||||
SessionStore: pgStores.Sessions,
|
||||
EventBus: domainBus,
|
||||
Provider: bgProvider,
|
||||
Model: bgModel,
|
||||
SystemConfigs: pgStores.SystemConfigs,
|
||||
Registry: providerRegistry,
|
||||
Extractor: kgExtractor,
|
||||
AgentStore: pgStores.Agents,
|
||||
})
|
||||
@@ -211,21 +211,22 @@ func runGateway() {
|
||||
}
|
||||
|
||||
// V3: Wire vault enrichment worker (async summary + embedding + auto-linking).
|
||||
// Provider is resolved per-tenant at runtime — no static provider needed.
|
||||
var enrichProgress *vault.EnrichProgress
|
||||
var updateVaultProvider vault.ProviderUpdater
|
||||
if pgStores.Vault != nil && bgProvider != nil {
|
||||
cleanupVaultEnrich, ep, updater := vault.RegisterEnrichWorker(vault.EnrichWorkerDeps{
|
||||
VaultStore: pgStores.Vault,
|
||||
Provider: bgProvider,
|
||||
Model: bgModel,
|
||||
EventBus: domainBus,
|
||||
MsgBus: msgBus,
|
||||
TeamStore: pgStores.Teams, // Phase 04 task-based auto-linking
|
||||
var enrichWorker *vault.EnrichWorker
|
||||
if pgStores.Vault != nil && providerRegistry != nil {
|
||||
cleanupVaultEnrich, ep, ew := vault.RegisterEnrichWorker(vault.EnrichWorkerDeps{
|
||||
VaultStore: pgStores.Vault,
|
||||
SystemConfigs: pgStores.SystemConfigs,
|
||||
Registry: providerRegistry,
|
||||
EventBus: domainBus,
|
||||
MsgBus: msgBus,
|
||||
TeamStore: pgStores.Teams,
|
||||
})
|
||||
enrichProgress = ep
|
||||
updateVaultProvider = updater
|
||||
enrichWorker = ew
|
||||
defer cleanupVaultEnrich()
|
||||
slog.Info("vault enrichment worker registered", "provider", bgProvider.Name(), "model", bgModel)
|
||||
slog.Info("vault enrichment worker registered (per-tenant provider resolution)")
|
||||
}
|
||||
|
||||
loadBootstrapFiles(pgStores, workspace, agentCfg)
|
||||
@@ -298,8 +299,8 @@ func runGateway() {
|
||||
agentRouter: agentRouter,
|
||||
toolsReg: toolsReg,
|
||||
skillsLoader: skillsLoader,
|
||||
enrichProgress: enrichProgress,
|
||||
updateVaultProvider: updateVaultProvider,
|
||||
enrichProgress: enrichProgress,
|
||||
enrichWorker: enrichWorker,
|
||||
workspace: workspace,
|
||||
dataDir: dataDir,
|
||||
domainBus: domainBus,
|
||||
|
||||
+2
-2
@@ -28,8 +28,8 @@ type gatewayDeps struct {
|
||||
toolsReg *tools.Registry
|
||||
skillsLoader *skills.Loader // optional: enables skill creation in evolution approval
|
||||
permCache *cache.PermissionCache // nil if no tenant store; closed on shutdown to stop sweep goroutines
|
||||
enrichProgress *vault.EnrichProgress // nil if enrichment worker not registered
|
||||
updateVaultProvider vault.ProviderUpdater // nil if enrichment worker not registered; hot-swaps LLM
|
||||
enrichProgress *vault.EnrichProgress // nil if enrichment worker not registered
|
||||
enrichWorker *vault.EnrichWorker // nil if enrichment worker not registered; for stop/enqueue
|
||||
workspace string
|
||||
dataDir string
|
||||
domainBus eventbus.DomainEventBus
|
||||
|
||||
@@ -118,12 +118,8 @@ func (d *gatewayDeps) wireHTTPHandlersOnServer(
|
||||
pgMem.UpdateChunkConfig(mem.MaxChunkLen, mem.ChunkOverlap)
|
||||
}
|
||||
}
|
||||
// Hot-swap vault enrichment provider/model if config changed
|
||||
if d.updateVaultProvider != nil {
|
||||
if p, m := resolveBackgroundProvider(d.cfg, d.providerRegistry); p != nil {
|
||||
d.updateVaultProvider(p, m)
|
||||
}
|
||||
}
|
||||
// Note: vault enrichment provider is resolved per-tenant at runtime,
|
||||
// no hot-reload needed here
|
||||
slog.Debug("system_configs refreshed to in-memory config", "keys", len(sysConfigs))
|
||||
}
|
||||
})
|
||||
|
||||
@@ -106,21 +106,8 @@ func (d *gatewayDeps) runLifecycle(
|
||||
slog.Info("tts config reloaded", "provider", newMgr.PrimaryProvider(), "auto", string(newMgr.AutoMode()))
|
||||
})
|
||||
|
||||
// Hot-swap vault enrichment provider/model on config changes via pub/sub.
|
||||
if d.updateVaultProvider != nil {
|
||||
d.msgBus.Subscribe("vault-enrich-config-reload", func(evt bus.Event) {
|
||||
if evt.Name != bus.TopicConfigChanged {
|
||||
return
|
||||
}
|
||||
updatedCfg, ok := evt.Payload.(*config.Config)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if p, m := resolveBackgroundProvider(updatedCfg, d.providerRegistry); p != nil {
|
||||
d.updateVaultProvider(p, m)
|
||||
}
|
||||
})
|
||||
}
|
||||
// Note: vault enrichment provider is resolved per-tenant at runtime,
|
||||
// no hot-reload handler needed here
|
||||
|
||||
// Log orphaned providers on agent deletion. Auto-delete is unsafe because
|
||||
// providers can be referenced by heartbeats (FK), OAuth tokens, media chains.
|
||||
|
||||
Reference in new issue
Block a user