mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-15 00:17:27 +00:00
fix(cache): tenant-aware invalidation for builtin tools and skills
Tenant config changes for builtin tools and skills silently failed to invalidate cached agent Loops, leaving tenants stuck on stale tool/skill sets until the 10-minute TTL expired or an unrelated event wiped the cache. Master-level skill CRUD had the same gap in the opposite direction. - Add CacheInvalidatePayload.TenantID so events can scope to one tenant - Add Router.InvalidateTenant(tenantID) with prefix match on "tenantID:agentKey" cache keys; uuid.Nil is a no-op - Rework emitCacheInvalidate helpers in builtin_tools + skills handlers to carry tenant scope; add defense-in-depth uuid.Nil guards in four tenant-config handlers - Update TopicCacheBuiltinTools + TopicCacheSkills subscribers to branch on payload.TenantID (tenant event wipes that tenant, global event keeps the existing InvalidateAll path) - Wire emitCacheInvalidate into master skill CRUD paths (update, delete, toggle, upload, install-deps, rescan-deps, import) that previously only called BumpVersion - Document the system-owner bypass in requireTenantAdmin so handlers keep guarding uuid.Nil themselves
This commit is contained in:
1 parent
ee75d498e6
commit
d77a3664db
13 files changed
+568
-15
No files matched your search
+17
-2
@@ -427,7 +427,11 @@ func wireExtras(
|
||||
}
|
||||
})
|
||||
|
||||
// Skills cache: bump version on skill changes
|
||||
// Skills cache: bump version on every event (global listCache is
|
||||
// version-keyed, so bump invalidates every tenant's ListSkills cache —
|
||||
// cheap since rebuild is a single DB read). Then the agent router
|
||||
// receives a scoped wipe: tenant-scoped events only wipe that tenant's
|
||||
// cached Loops; master/global events wipe the entire router cache.
|
||||
if stores.Skills != nil {
|
||||
msgBus.Subscribe(bus.TopicCacheSkills, func(event bus.Event) {
|
||||
if event.Name != protocol.EventCacheInvalidate {
|
||||
@@ -438,6 +442,11 @@ func wireExtras(
|
||||
return
|
||||
}
|
||||
stores.Skills.BumpVersion()
|
||||
if payload.TenantID != uuid.Nil {
|
||||
agentRouter.InvalidateTenant(payload.TenantID)
|
||||
return
|
||||
}
|
||||
agentRouter.InvalidateAll()
|
||||
})
|
||||
}
|
||||
|
||||
@@ -507,7 +516,9 @@ func wireExtras(
|
||||
})
|
||||
}
|
||||
|
||||
// Builtin tools cache: re-apply disables on settings/enabled changes
|
||||
// Builtin tools cache: re-apply disables on settings/enabled changes.
|
||||
// Tenant-scoped events only invalidate that tenant's cached agents — the
|
||||
// global registry disables list is master-only and unaffected.
|
||||
if stores.BuiltinTools != nil {
|
||||
msgBus.Subscribe(bus.TopicCacheBuiltinTools, func(event bus.Event) {
|
||||
if event.Name != protocol.EventCacheInvalidate {
|
||||
@@ -517,6 +528,10 @@ func wireExtras(
|
||||
if !ok || payload.Kind != bus.CacheKindBuiltinTools {
|
||||
return
|
||||
}
|
||||
if payload.TenantID != uuid.Nil {
|
||||
agentRouter.InvalidateTenant(payload.TenantID)
|
||||
return
|
||||
}
|
||||
applyBuiltinToolDisables(context.Background(), stores.BuiltinTools, toolsReg)
|
||||
agentRouter.InvalidateAll()
|
||||
})
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bootstrap"
|
||||
@@ -508,6 +509,27 @@ func (r *Router) InvalidateAll() {
|
||||
slog.Debug("invalidated all agent caches")
|
||||
}
|
||||
|
||||
// InvalidateTenant clears all cached agents for a single tenant.
|
||||
// Cache keys are "tenantID:agentKey" when tenant-scoped (see agentCacheKey).
|
||||
// Non-tenant entries (bare "agentKey") are untouched. uuid.Nil is a no-op —
|
||||
// callers use InvalidateAll for global invalidation.
|
||||
func (r *Router) InvalidateTenant(tenantID uuid.UUID) {
|
||||
if tenantID == uuid.Nil {
|
||||
return
|
||||
}
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
prefix := tenantID.String() + ":"
|
||||
var deleted int
|
||||
for key := range r.agents {
|
||||
if strings.HasPrefix(key, prefix) {
|
||||
delete(r.agents, key)
|
||||
deleted++
|
||||
}
|
||||
}
|
||||
slog.Debug("invalidated tenant agent cache", "tenant", tenantID, "count", deleted)
|
||||
}
|
||||
|
||||
// resolveTenantSlug looks up the tenant slug for workspace path resolution.
|
||||
// Returns the tenant ID string as fallback if lookup fails.
|
||||
func resolveTenantSlug(ts store.TenantStore, tenantID uuid.UUID) string {
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
// TestInvalidateTenant_OnlyAffectsMatchingTenant verifies that
|
||||
// Router.InvalidateTenant deletes only cache entries whose key is prefixed
|
||||
// with the target tenant's UUID, leaving other tenants and bare
|
||||
// (non-tenant) entries untouched.
|
||||
func TestInvalidateTenant_OnlyAffectsMatchingTenant(t *testing.T) {
|
||||
r := NewRouter()
|
||||
|
||||
tenantA := uuid.New()
|
||||
tenantB := uuid.New()
|
||||
|
||||
ctxA := store.WithTenantID(context.Background(), tenantA)
|
||||
ctxB := store.WithTenantID(context.Background(), tenantB)
|
||||
|
||||
// Entries for three scopes: tenantA, tenantB, bare (no tenant).
|
||||
keyA1 := agentCacheKey(ctxA, "agent-1")
|
||||
keyA2 := agentCacheKey(ctxA, "agent-2")
|
||||
keyB1 := agentCacheKey(ctxB, "agent-1")
|
||||
keyBare := agentCacheKey(context.Background(), "agent-1")
|
||||
|
||||
r.agents[keyA1] = &agentEntry{}
|
||||
r.agents[keyA2] = &agentEntry{}
|
||||
r.agents[keyB1] = &agentEntry{}
|
||||
r.agents[keyBare] = &agentEntry{}
|
||||
|
||||
r.InvalidateTenant(tenantA)
|
||||
|
||||
if _, ok := r.agents[keyA1]; ok {
|
||||
t.Error("tenantA entry agent-1 should have been invalidated")
|
||||
}
|
||||
if _, ok := r.agents[keyA2]; ok {
|
||||
t.Error("tenantA entry agent-2 should have been invalidated")
|
||||
}
|
||||
if _, ok := r.agents[keyB1]; !ok {
|
||||
t.Error("tenantB entry must survive tenantA invalidation")
|
||||
}
|
||||
if _, ok := r.agents[keyBare]; !ok {
|
||||
t.Error("bare (non-tenant) entry must survive tenantA invalidation")
|
||||
}
|
||||
}
|
||||
|
||||
// TestInvalidateTenant_NilIsNoop verifies the uuid.Nil guard — passing Nil
|
||||
// must not wipe any entries (callers use InvalidateAll for global wipes).
|
||||
func TestInvalidateTenant_NilIsNoop(t *testing.T) {
|
||||
r := NewRouter()
|
||||
tenantA := uuid.New()
|
||||
|
||||
ctxA := store.WithTenantID(context.Background(), tenantA)
|
||||
keyA := agentCacheKey(ctxA, "agent-1")
|
||||
keyBare := agentCacheKey(context.Background(), "agent-1")
|
||||
|
||||
r.agents[keyA] = &agentEntry{}
|
||||
r.agents[keyBare] = &agentEntry{}
|
||||
|
||||
r.InvalidateTenant(uuid.Nil)
|
||||
|
||||
if len(r.agents) != 2 {
|
||||
t.Errorf("InvalidateTenant(uuid.Nil) must be a no-op, got %d entries remaining (want 2)", len(r.agents))
|
||||
}
|
||||
}
|
||||
|
||||
// TestInvalidateTenant_SubstringSafety verifies that a tenant with a UUID
|
||||
// that is a prefix (by string) of another tenant's key segment cannot
|
||||
// accidentally wipe the other. UUIDs are fixed-length so this can only
|
||||
// happen if matching is purely substring-based. The ":" suffix in the
|
||||
// prefix guards against it.
|
||||
func TestInvalidateTenant_SubstringSafety(t *testing.T) {
|
||||
r := NewRouter()
|
||||
|
||||
// Same UUID used in two roles — as a tenant key prefix and inside an
|
||||
// agent key. If the matcher incorrectly matched anywhere in the key,
|
||||
// InvalidateTenant(tenantA) would wipe the second entry too.
|
||||
tenantA := uuid.New()
|
||||
tenantB := uuid.New()
|
||||
|
||||
ctxA := store.WithTenantID(context.Background(), tenantA)
|
||||
ctxB := store.WithTenantID(context.Background(), tenantB)
|
||||
|
||||
keyA := agentCacheKey(ctxA, "agent-1")
|
||||
// Craft a pathological key under tenantB whose agent name contains
|
||||
// tenantA's UUID. Should NOT be deleted.
|
||||
keyB := agentCacheKey(ctxB, "agent-"+tenantA.String())
|
||||
|
||||
r.agents[keyA] = &agentEntry{}
|
||||
r.agents[keyB] = &agentEntry{}
|
||||
|
||||
r.InvalidateTenant(tenantA)
|
||||
|
||||
if _, ok := r.agents[keyA]; ok {
|
||||
t.Error("tenantA key should have been deleted")
|
||||
}
|
||||
if _, ok := r.agents[keyB]; !ok {
|
||||
t.Error("tenantB key containing tenantA UUID in agent name must NOT be deleted")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,177 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
|
||||
)
|
||||
|
||||
// This file mirrors the subscriber callback bodies registered in
|
||||
// cmd/gateway_managed.go (TopicCacheBuiltinTools + TopicCacheSkills). Keeping
|
||||
// the logic here — not in cmd — gives the test direct access to the router's
|
||||
// private map so we can assert precisely which entries survive tenant-scoped
|
||||
// vs global invalidation. If the real subscribers in gateway_managed.go
|
||||
// diverge from these, update both in lockstep.
|
||||
|
||||
func handleBuiltinToolsInvalidate(router *Router, onGlobal func(), event bus.Event) {
|
||||
if event.Name != protocol.EventCacheInvalidate {
|
||||
return
|
||||
}
|
||||
payload, ok := event.Payload.(bus.CacheInvalidatePayload)
|
||||
if !ok || payload.Kind != bus.CacheKindBuiltinTools {
|
||||
return
|
||||
}
|
||||
if payload.TenantID != uuid.Nil {
|
||||
router.InvalidateTenant(payload.TenantID)
|
||||
return
|
||||
}
|
||||
if onGlobal != nil {
|
||||
onGlobal()
|
||||
}
|
||||
router.InvalidateAll()
|
||||
}
|
||||
|
||||
func handleSkillsInvalidate(router *Router, onBump func(), event bus.Event) {
|
||||
if event.Name != protocol.EventCacheInvalidate {
|
||||
return
|
||||
}
|
||||
payload, ok := event.Payload.(bus.CacheInvalidatePayload)
|
||||
if !ok || payload.Kind != bus.CacheKindSkills {
|
||||
return
|
||||
}
|
||||
if onBump != nil {
|
||||
onBump()
|
||||
}
|
||||
if payload.TenantID != uuid.Nil {
|
||||
router.InvalidateTenant(payload.TenantID)
|
||||
return
|
||||
}
|
||||
router.InvalidateAll()
|
||||
}
|
||||
|
||||
func seedTenantAgents(t *testing.T, tenantA, tenantB uuid.UUID) (*Router, [3]string) {
|
||||
t.Helper()
|
||||
r := NewRouter()
|
||||
ctxA := store.WithTenantID(context.Background(), tenantA)
|
||||
ctxB := store.WithTenantID(context.Background(), tenantB)
|
||||
keys := [3]string{
|
||||
agentCacheKey(ctxA, "agent-1"),
|
||||
agentCacheKey(ctxB, "agent-1"),
|
||||
agentCacheKey(context.Background(), "agent-1"),
|
||||
}
|
||||
for _, k := range keys {
|
||||
r.agents[k] = &agentEntry{}
|
||||
}
|
||||
return r, keys
|
||||
}
|
||||
|
||||
// TestBuiltinToolsSubscriber_TenantEventOnlyInvalidatesThatTenant: the
|
||||
// tenant branch must leave other tenants and bare entries alone and must
|
||||
// NOT run the global re-apply callback.
|
||||
func TestBuiltinToolsSubscriber_TenantEventOnlyInvalidatesThatTenant(t *testing.T) {
|
||||
tenantA := uuid.New()
|
||||
tenantB := uuid.New()
|
||||
r, keys := seedTenantAgents(t, tenantA, tenantB)
|
||||
|
||||
globalCalled := false
|
||||
handleBuiltinToolsInvalidate(r, func() { globalCalled = true }, bus.Event{
|
||||
Name: protocol.EventCacheInvalidate,
|
||||
Payload: bus.CacheInvalidatePayload{
|
||||
Kind: bus.CacheKindBuiltinTools,
|
||||
Key: "tts",
|
||||
TenantID: tenantA,
|
||||
},
|
||||
})
|
||||
|
||||
if globalCalled {
|
||||
t.Error("global reapply must NOT run for tenant-scoped event")
|
||||
}
|
||||
if _, ok := r.agents[keys[0]]; ok {
|
||||
t.Error("tenantA entry should have been invalidated")
|
||||
}
|
||||
if _, ok := r.agents[keys[1]]; !ok {
|
||||
t.Error("tenantB entry must survive tenantA invalidation")
|
||||
}
|
||||
if _, ok := r.agents[keys[2]]; !ok {
|
||||
t.Error("bare entry must survive tenantA invalidation")
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuiltinToolsSubscriber_GlobalEventInvalidatesAll: the master/global
|
||||
// branch must wipe everything and run the global re-apply callback.
|
||||
func TestBuiltinToolsSubscriber_GlobalEventInvalidatesAll(t *testing.T) {
|
||||
tenantA := uuid.New()
|
||||
tenantB := uuid.New()
|
||||
r, _ := seedTenantAgents(t, tenantA, tenantB)
|
||||
|
||||
globalCalled := false
|
||||
handleBuiltinToolsInvalidate(r, func() { globalCalled = true }, bus.Event{
|
||||
Name: protocol.EventCacheInvalidate,
|
||||
Payload: bus.CacheInvalidatePayload{
|
||||
Kind: bus.CacheKindBuiltinTools,
|
||||
Key: "tts",
|
||||
},
|
||||
})
|
||||
|
||||
if !globalCalled {
|
||||
t.Error("global reapply must run for master/global event")
|
||||
}
|
||||
if len(r.agents) != 0 {
|
||||
t.Errorf("global event must wipe all entries, got %d remaining", len(r.agents))
|
||||
}
|
||||
}
|
||||
|
||||
// TestSkillsSubscriber_BumpsVersionAndInvalidatesTenant: bump runs every
|
||||
// event; tenant branch only wipes the target tenant.
|
||||
func TestSkillsSubscriber_BumpsVersionAndInvalidatesTenant(t *testing.T) {
|
||||
tenantA := uuid.New()
|
||||
tenantB := uuid.New()
|
||||
r, keys := seedTenantAgents(t, tenantA, tenantB)
|
||||
|
||||
bumped := false
|
||||
handleSkillsInvalidate(r, func() { bumped = true }, bus.Event{
|
||||
Name: protocol.EventCacheInvalidate,
|
||||
Payload: bus.CacheInvalidatePayload{
|
||||
Kind: bus.CacheKindSkills,
|
||||
Key: uuid.New().String(),
|
||||
TenantID: tenantA,
|
||||
},
|
||||
})
|
||||
|
||||
if !bumped {
|
||||
t.Error("skills version must be bumped on every invalidate event")
|
||||
}
|
||||
if _, ok := r.agents[keys[0]]; ok {
|
||||
t.Error("tenantA entry should have been invalidated")
|
||||
}
|
||||
if _, ok := r.agents[keys[1]]; !ok {
|
||||
t.Error("tenantB entry must survive")
|
||||
}
|
||||
}
|
||||
|
||||
// TestSkillsSubscriber_IgnoresMismatchedKind: events with a different Kind
|
||||
// must be ignored entirely (no bump, no router touch).
|
||||
func TestSkillsSubscriber_IgnoresMismatchedKind(t *testing.T) {
|
||||
r, keys := seedTenantAgents(t, uuid.New(), uuid.New())
|
||||
bumped := false
|
||||
|
||||
handleSkillsInvalidate(r, func() { bumped = true }, bus.Event{
|
||||
Name: protocol.EventCacheInvalidate,
|
||||
Payload: bus.CacheInvalidatePayload{
|
||||
Kind: bus.CacheKindBuiltinTools, // wrong kind for skills subscriber
|
||||
Key: "x",
|
||||
},
|
||||
})
|
||||
|
||||
if bumped {
|
||||
t.Error("skills subscriber must ignore non-skill events")
|
||||
}
|
||||
if _, ok := r.agents[keys[0]]; !ok {
|
||||
t.Error("router entries must remain untouched on mismatched kind")
|
||||
}
|
||||
}
|
||||
@@ -142,10 +142,16 @@ type AuditEventPayload struct {
|
||||
}
|
||||
|
||||
// CacheInvalidatePayload signals cache layers to evict stale entries.
|
||||
// Used with protocol.EventCacheInvalidate events.
|
||||
// Used with protocol.EventCacheInvalidate events. Events are delivered
|
||||
// in-process via MessageBus and never marshaled to the wire, so the json
|
||||
// tags are documentation-only (and omitempty on uuid.UUID is a no-op
|
||||
// because uuid.UUID is [16]byte — all-zero arrays don't count as empty).
|
||||
type CacheInvalidatePayload struct {
|
||||
Kind string `json:"kind"` // CacheKind* constants
|
||||
Key string `json:"key"` // agent_key, agent_id, etc. Empty = invalidate all
|
||||
// TenantID scopes the invalidation to a single tenant. uuid.Nil means
|
||||
// global (master admin action) — subscribers treat it as "invalidate all".
|
||||
TenantID uuid.UUID `json:"tenant_id"`
|
||||
}
|
||||
|
||||
// MessageHandler handles an inbound message from a specific channel.
|
||||
|
||||
@@ -46,13 +46,19 @@ func (h *BuiltinToolsHandler) adminAuth(next http.HandlerFunc) http.HandlerFunc
|
||||
return requireAuth(permissions.RoleAdmin, next)
|
||||
}
|
||||
|
||||
func (h *BuiltinToolsHandler) emitCacheInvalidate(key string) {
|
||||
// emitCacheInvalidate broadcasts a cache invalidation event for a builtin tool.
|
||||
// tenantID == uuid.Nil means global invalidation (master admin path).
|
||||
func (h *BuiltinToolsHandler) emitCacheInvalidate(key string, tenantID uuid.UUID) {
|
||||
if h.msgBus == nil {
|
||||
return
|
||||
}
|
||||
h.msgBus.Broadcast(bus.Event{
|
||||
Name: protocol.EventCacheInvalidate,
|
||||
Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindBuiltinTools, Key: key},
|
||||
Name: protocol.EventCacheInvalidate,
|
||||
Payload: bus.CacheInvalidatePayload{
|
||||
Kind: bus.CacheKindBuiltinTools,
|
||||
Key: key,
|
||||
TenantID: tenantID,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
@@ -141,7 +147,7 @@ func (h *BuiltinToolsHandler) handleUpdate(w http.ResponseWriter, r *http.Reques
|
||||
}
|
||||
|
||||
emitAudit(h.msgBus, r, "builtin_tool.updated", "builtin_tool", name)
|
||||
h.emitCacheInvalidate(name)
|
||||
h.emitCacheInvalidate(name, uuid.Nil)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "updated"})
|
||||
}
|
||||
|
||||
@@ -156,6 +162,14 @@ func (h *BuiltinToolsHandler) handleSetTenantConfig(w http.ResponseWriter, r *ht
|
||||
}
|
||||
name := r.PathValue("name")
|
||||
tid := store.TenantIDFromContext(r.Context())
|
||||
if tid == uuid.Nil {
|
||||
// Defense-in-depth: owner-role bypass in requireTenantAdmin could
|
||||
// otherwise reach here without a tenant scope. The DB FK would
|
||||
// reject the write, but we want the guard explicit so a nil tid
|
||||
// never flows into the cache invalidate emit as a global wipe.
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "tenant context required"})
|
||||
return
|
||||
}
|
||||
|
||||
var body struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
@@ -172,6 +186,7 @@ func (h *BuiltinToolsHandler) handleSetTenantConfig(w http.ResponseWriter, r *ht
|
||||
}
|
||||
|
||||
emitAudit(h.msgBus, r, "builtin_tool.tenant_config.set", "builtin_tool", name)
|
||||
h.emitCacheInvalidate(name, tid)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "updated"})
|
||||
}
|
||||
|
||||
@@ -186,6 +201,10 @@ func (h *BuiltinToolsHandler) handleDeleteTenantConfig(w http.ResponseWriter, r
|
||||
}
|
||||
name := r.PathValue("name")
|
||||
tid := store.TenantIDFromContext(r.Context())
|
||||
if tid == uuid.Nil {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "tenant context required"})
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.tenantCfgStore.Delete(r.Context(), tid, name); err != nil {
|
||||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
|
||||
@@ -193,5 +212,6 @@ func (h *BuiltinToolsHandler) handleDeleteTenantConfig(w http.ResponseWriter, r
|
||||
}
|
||||
|
||||
emitAudit(h.msgBus, r, "builtin_tool.tenant_config.deleted", "builtin_tool", name)
|
||||
h.emitCacheInvalidate(name, tid)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"})
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
package http
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
|
||||
)
|
||||
|
||||
// TestBuiltinToolsHandler_EmitCacheInvalidate_CarriesTenantID verifies the
|
||||
// emit helper populates TenantID in the broadcast payload so downstream
|
||||
// subscribers can branch on tenant-scoped vs global invalidation.
|
||||
func TestBuiltinToolsHandler_EmitCacheInvalidate_CarriesTenantID(t *testing.T) {
|
||||
mb := bus.New()
|
||||
defer mb.Close()
|
||||
|
||||
h := &BuiltinToolsHandler{msgBus: mb}
|
||||
|
||||
var captured bus.CacheInvalidatePayload
|
||||
var gotEvent bool
|
||||
mb.Subscribe("test-capture", func(e bus.Event) {
|
||||
if e.Name != protocol.EventCacheInvalidate {
|
||||
return
|
||||
}
|
||||
p, ok := e.Payload.(bus.CacheInvalidatePayload)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
captured = p
|
||||
gotEvent = true
|
||||
})
|
||||
|
||||
tid := uuid.New()
|
||||
h.emitCacheInvalidate("tts", tid)
|
||||
|
||||
if !gotEvent {
|
||||
t.Fatal("no cache invalidate event received")
|
||||
}
|
||||
if captured.Kind != bus.CacheKindBuiltinTools {
|
||||
t.Errorf("kind = %q, want %q", captured.Kind, bus.CacheKindBuiltinTools)
|
||||
}
|
||||
if captured.Key != "tts" {
|
||||
t.Errorf("key = %q, want %q", captured.Key, "tts")
|
||||
}
|
||||
if captured.TenantID != tid {
|
||||
t.Errorf("tenantID = %v, want %v", captured.TenantID, tid)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuiltinToolsHandler_EmitCacheInvalidate_NilTenantMeansGlobal verifies
|
||||
// passing uuid.Nil yields a payload whose subscribers read as "global".
|
||||
func TestBuiltinToolsHandler_EmitCacheInvalidate_NilTenantMeansGlobal(t *testing.T) {
|
||||
mb := bus.New()
|
||||
defer mb.Close()
|
||||
|
||||
h := &BuiltinToolsHandler{msgBus: mb}
|
||||
|
||||
var captured bus.CacheInvalidatePayload
|
||||
mb.Subscribe("test-capture", func(e bus.Event) {
|
||||
if p, ok := e.Payload.(bus.CacheInvalidatePayload); ok {
|
||||
captured = p
|
||||
}
|
||||
})
|
||||
|
||||
h.emitCacheInvalidate("tts", uuid.Nil)
|
||||
|
||||
if captured.TenantID != uuid.Nil {
|
||||
t.Errorf("tenantID = %v, want uuid.Nil for global invalidation", captured.TenantID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuiltinToolsHandler_EmitCacheInvalidate_NilMsgBusNoop verifies the
|
||||
// helper is safe to call when msgBus is unset (e.g., in tests/desktop lite).
|
||||
func TestBuiltinToolsHandler_EmitCacheInvalidate_NilMsgBusNoop(t *testing.T) {
|
||||
h := &BuiltinToolsHandler{msgBus: nil}
|
||||
// Must not panic.
|
||||
h.emitCacheInvalidate("tts", uuid.New())
|
||||
}
|
||||
+35
-4
@@ -59,14 +59,22 @@ func (h *SkillsHandler) skillUploadLock(scopeKey string) *sync.Mutex {
|
||||
return actual.(*sync.Mutex)
|
||||
}
|
||||
|
||||
// emitCacheInvalidate broadcasts a cache invalidation event if msgBus is set.
|
||||
func (h *SkillsHandler) emitCacheInvalidate(kind, key string) {
|
||||
// emitCacheInvalidate broadcasts a skill-related cache invalidation event.
|
||||
// tenantID == uuid.Nil means global invalidation (master admin path).
|
||||
// Existing grant-related callers pass tenantID == uuid.Nil since grants are
|
||||
// stored globally; tenant-aware callers (tenant_config handlers) pass the
|
||||
// caller's tenant ID so only that tenant's cached agents are invalidated.
|
||||
func (h *SkillsHandler) emitCacheInvalidate(kind, key string, tenantID uuid.UUID) {
|
||||
if h.msgBus == nil {
|
||||
return
|
||||
}
|
||||
h.msgBus.Broadcast(bus.Event{
|
||||
Name: protocol.EventCacheInvalidate,
|
||||
Payload: bus.CacheInvalidatePayload{Kind: kind, Key: key},
|
||||
Name: protocol.EventCacheInvalidate,
|
||||
Payload: bus.CacheInvalidatePayload{
|
||||
Kind: kind,
|
||||
Key: key,
|
||||
TenantID: tenantID,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
@@ -212,6 +220,7 @@ func (h *SkillsHandler) handleUpdate(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
h.skills.BumpVersion()
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, idStr, uuid.Nil)
|
||||
emitAudit(h.msgBus, r, "skill.updated", "skill", idStr)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"ok": "true"})
|
||||
}
|
||||
@@ -245,6 +254,7 @@ func (h *SkillsHandler) handleDelete(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
h.skills.BumpVersion()
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, idStr, uuid.Nil)
|
||||
emitAudit(h.msgBus, r, "skill.deleted", "skill", idStr)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"ok": "true"})
|
||||
}
|
||||
@@ -333,6 +343,7 @@ func (h *SkillsHandler) handleInstallDeps(w http.ResponseWriter, r *http.Request
|
||||
}
|
||||
if statusChanged {
|
||||
h.skills.BumpVersion()
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, "", uuid.Nil)
|
||||
}
|
||||
|
||||
if h.msgBus != nil {
|
||||
@@ -495,6 +506,11 @@ func (h *SkillsHandler) handleRescanDeps(w http.ResponseWriter, r *http.Request)
|
||||
return
|
||||
}
|
||||
updated, results := h.rescanAndUpdate()
|
||||
if updated > 0 {
|
||||
// rescanAndUpdate bumped the skills version already; emit a global
|
||||
// invalidate so cached agent Loops pick up the new status set.
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, "", uuid.Nil)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"updated": updated,
|
||||
"results": results,
|
||||
@@ -555,6 +571,7 @@ func (h *SkillsHandler) handleToggle(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
h.skills.BumpVersion()
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, idStr, uuid.Nil)
|
||||
emitAudit(h.msgBus, r, "skill.toggled", "skill", idStr)
|
||||
writeJSON(w, http.StatusOK, map[string]any{"ok": true, "enabled": body.Enabled, "status": newStatus})
|
||||
}
|
||||
@@ -570,6 +587,14 @@ func (h *SkillsHandler) handleSetTenantConfig(w http.ResponseWriter, r *http.Req
|
||||
}
|
||||
locale := store.LocaleFromContext(r.Context())
|
||||
tid := store.TenantIDFromContext(r.Context())
|
||||
if tid == uuid.Nil {
|
||||
// Defense-in-depth: owner-role bypass in requireTenantAdmin could
|
||||
// otherwise reach here without a tenant scope. Reject explicitly so
|
||||
// a nil tid never flows into the cache invalidate emit as a global
|
||||
// wipe of every tenant's agent cache.
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "tenant context required"})
|
||||
return
|
||||
}
|
||||
|
||||
idStr := r.PathValue("id")
|
||||
skillID, err := uuid.Parse(idStr)
|
||||
@@ -593,6 +618,7 @@ func (h *SkillsHandler) handleSetTenantConfig(w http.ResponseWriter, r *http.Req
|
||||
}
|
||||
|
||||
emitAudit(h.msgBus, r, "skill.tenant_config.set", "skill", idStr)
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, idStr, tid)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "updated"})
|
||||
}
|
||||
|
||||
@@ -607,6 +633,10 @@ func (h *SkillsHandler) handleDeleteTenantConfig(w http.ResponseWriter, r *http.
|
||||
}
|
||||
locale := store.LocaleFromContext(r.Context())
|
||||
tid := store.TenantIDFromContext(r.Context())
|
||||
if tid == uuid.Nil {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "tenant context required"})
|
||||
return
|
||||
}
|
||||
|
||||
idStr := r.PathValue("id")
|
||||
skillID, err := uuid.Parse(idStr)
|
||||
@@ -622,5 +652,6 @@ func (h *SkillsHandler) handleDeleteTenantConfig(w http.ResponseWriter, r *http.
|
||||
}
|
||||
|
||||
emitAudit(h.msgBus, r, "skill.tenant_config.deleted", "skill", idStr)
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, idStr, tid)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"})
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package http
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
|
||||
)
|
||||
|
||||
// TestSkillsHandler_EmitCacheInvalidate_TenantScopedPropagatesTenantID
|
||||
// verifies tenant-config handlers emit payloads carrying the caller's tenant.
|
||||
func TestSkillsHandler_EmitCacheInvalidate_TenantScopedPropagatesTenantID(t *testing.T) {
|
||||
mb := bus.New()
|
||||
defer mb.Close()
|
||||
|
||||
h := &SkillsHandler{msgBus: mb}
|
||||
|
||||
var captured bus.CacheInvalidatePayload
|
||||
var gotEvent bool
|
||||
mb.Subscribe("test-capture", func(e bus.Event) {
|
||||
if e.Name != protocol.EventCacheInvalidate {
|
||||
return
|
||||
}
|
||||
if p, ok := e.Payload.(bus.CacheInvalidatePayload); ok {
|
||||
captured = p
|
||||
gotEvent = true
|
||||
}
|
||||
})
|
||||
|
||||
tid := uuid.New()
|
||||
skillID := uuid.New().String()
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, skillID, tid)
|
||||
|
||||
if !gotEvent {
|
||||
t.Fatal("no cache invalidate event received")
|
||||
}
|
||||
if captured.Kind != bus.CacheKindSkills {
|
||||
t.Errorf("kind = %q, want %q", captured.Kind, bus.CacheKindSkills)
|
||||
}
|
||||
if captured.Key != skillID {
|
||||
t.Errorf("key = %q, want %q", captured.Key, skillID)
|
||||
}
|
||||
if captured.TenantID != tid {
|
||||
t.Errorf("tenantID = %v, want %v", captured.TenantID, tid)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSkillsHandler_EmitCacheInvalidate_GrantsGlobal verifies grant-change
|
||||
// callers pass uuid.Nil so subscribers treat the event as global.
|
||||
func TestSkillsHandler_EmitCacheInvalidate_GrantsGlobal(t *testing.T) {
|
||||
mb := bus.New()
|
||||
defer mb.Close()
|
||||
|
||||
h := &SkillsHandler{msgBus: mb}
|
||||
|
||||
var captured bus.CacheInvalidatePayload
|
||||
mb.Subscribe("test-capture", func(e bus.Event) {
|
||||
if p, ok := e.Payload.(bus.CacheInvalidatePayload); ok {
|
||||
captured = p
|
||||
}
|
||||
})
|
||||
|
||||
h.emitCacheInvalidate(bus.CacheKindSkillGrants, "", uuid.Nil)
|
||||
|
||||
if captured.Kind != bus.CacheKindSkillGrants {
|
||||
t.Errorf("kind = %q, want %q", captured.Kind, bus.CacheKindSkillGrants)
|
||||
}
|
||||
if captured.TenantID != uuid.Nil {
|
||||
t.Errorf("tenantID = %v, want uuid.Nil for grant-change global event", captured.TenantID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSkillsHandler_EmitCacheInvalidate_NilMsgBusNoop verifies the helper is
|
||||
// safe with an unset msgBus.
|
||||
func TestSkillsHandler_EmitCacheInvalidate_NilMsgBusNoop(t *testing.T) {
|
||||
h := &SkillsHandler{msgBus: nil}
|
||||
// Must not panic.
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, "x", uuid.New())
|
||||
}
|
||||
@@ -76,7 +76,7 @@ func (h *SkillsHandler) handleGrantAgent(w http.ResponseWriter, r *http.Request)
|
||||
}
|
||||
|
||||
h.skills.BumpVersion()
|
||||
h.emitCacheInvalidate(bus.CacheKindSkillGrants, "")
|
||||
h.emitCacheInvalidate(bus.CacheKindSkillGrants, "", uuid.Nil)
|
||||
emitAudit(h.msgBus, r, "skill.grant_changed", "skill", idStr)
|
||||
writeJSON(w, http.StatusCreated, map[string]string{"ok": "true"})
|
||||
}
|
||||
@@ -113,7 +113,7 @@ func (h *SkillsHandler) handleRevokeAgent(w http.ResponseWriter, r *http.Request
|
||||
}
|
||||
|
||||
h.skills.BumpVersion()
|
||||
h.emitCacheInvalidate(bus.CacheKindSkillGrants, "")
|
||||
h.emitCacheInvalidate(bus.CacheKindSkillGrants, "", uuid.Nil)
|
||||
emitAudit(h.msgBus, r, "skill.grant_changed", "skill", idStr)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"ok": "true"})
|
||||
}
|
||||
@@ -158,7 +158,7 @@ func (h *SkillsHandler) handleGrantUser(w http.ResponseWriter, r *http.Request)
|
||||
}
|
||||
|
||||
h.skills.BumpVersion()
|
||||
h.emitCacheInvalidate(bus.CacheKindSkillGrants, "")
|
||||
h.emitCacheInvalidate(bus.CacheKindSkillGrants, "", uuid.Nil)
|
||||
emitAudit(h.msgBus, r, "skill.grant_changed", "skill", idStr)
|
||||
writeJSON(w, http.StatusCreated, map[string]string{"ok": "true"})
|
||||
}
|
||||
@@ -193,7 +193,7 @@ func (h *SkillsHandler) handleRevokeUser(w http.ResponseWriter, r *http.Request)
|
||||
}
|
||||
|
||||
h.skills.BumpVersion()
|
||||
h.emitCacheInvalidate(bus.CacheKindSkillGrants, "")
|
||||
h.emitCacheInvalidate(bus.CacheKindSkillGrants, "", uuid.Nil)
|
||||
emitAudit(h.msgBus, r, "skill.grant_changed", "skill", idStr)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"ok": "true"})
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/i18n"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
@@ -54,6 +55,11 @@ func (h *SkillsHandler) handleSkillsImport(w http.ResponseWriter, r *http.Reques
|
||||
sendSSE(w, flusher, "error", ProgressEvent{Phase: "import", Status: "error", Detail: importErr.Error()})
|
||||
return
|
||||
}
|
||||
// Import affects the importer's tenant scope — invalidate that
|
||||
// tenant's cached agents so they pick up the new skill set. If
|
||||
// imported under master, tid is the master tenant UUID, which
|
||||
// still yields a correct per-tenant router wipe.
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, "", store.TenantIDFromContext(r.Context()))
|
||||
sendSSE(w, flusher, "complete", summary)
|
||||
return
|
||||
}
|
||||
@@ -64,6 +70,7 @@ func (h *SkillsHandler) handleSkillsImport(w http.ResponseWriter, r *http.Reques
|
||||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": i18n.T(locale, i18n.MsgInternalError, err.Error())})
|
||||
return
|
||||
}
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, "", store.TenantIDFromContext(r.Context()))
|
||||
writeJSON(w, http.StatusCreated, summary)
|
||||
}
|
||||
|
||||
|
||||
@@ -13,6 +13,8 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/i18n"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/skills"
|
||||
@@ -227,6 +229,7 @@ func (h *SkillsHandler) handleUpload(w http.ResponseWriter, r *http.Request) {
|
||||
response["id"] = id
|
||||
|
||||
h.skills.BumpVersion()
|
||||
h.emitCacheInvalidate(bus.CacheKindSkills, id.String(), uuid.Nil)
|
||||
emitAudit(h.msgBus, r, "skill.uploaded", "skill", slug)
|
||||
slog.Info("skill uploaded", "id", id, "slug", slug, "version", version, "size", header.Size, "status", skill.Status)
|
||||
depState.emit(h, slug)
|
||||
|
||||
@@ -12,6 +12,12 @@ import (
|
||||
// requireTenantAdmin verifies the caller has owner or admin role within the
|
||||
// specified tenant. System-wide owners (IsOwnerRole) bypass the check.
|
||||
// Returns true if authorized, false if an error response was written.
|
||||
//
|
||||
// WARNING: the system-owner bypass does NOT guarantee a non-nil tenant ID
|
||||
// downstream. A system owner can reach tenant-config handlers with
|
||||
// tid == uuid.Nil. Handlers that branch on tenant ID (e.g. emitting
|
||||
// tenant-scoped cache invalidate events) must therefore guard against
|
||||
// uuid.Nil explicitly — don't rely on this helper to enforce it.
|
||||
func requireTenantAdmin(w http.ResponseWriter, r *http.Request, ts store.TenantStore) bool {
|
||||
ctx := r.Context()
|
||||
|
||||
|
||||
Reference in new issue
Block a user