mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-13 02:14:39 +00:00
fix: invalidate ContextFileInterceptor cache on agents.files.set (#29)
* fix: invalidate ContextFileInterceptor cache on agents.files.set When agents.files.set (or agents.update) wrote context files to agent_context_files in Postgres, only the agent router cache was invalidated via agents.InvalidateAgent(). The ContextFileInterceptor maintains its own separate in-memory cache with a 5-minute TTL which was never cleared, causing the agent to serve stale file content (e.g. wizard-written SOUL.md/IDENTITY.md) for up to 5 minutes. Fix: inject *tools.ContextFileInterceptor into AgentsMethods and call intc.InvalidateAgent(ag.ID) immediately after each successful DB write in handleFilesSet and handleUpdate. Wire path: wireManagedExtras now returns the interceptor it creates; gateway.go captures it and passes it through registerAllMethods → NewAgentsMethods. The interceptor is nil in standalone mode so all nil guards are in place. * test: add unit tests for ContextFileInterceptor cache invalidation Covers: - Cache hit: second read does not call store again - InvalidateAgent clears agent cache → fresh content served from store - InvalidateAgent clears user cache for that agent - InvalidateAgent does not affect other agents' cache entries - TTL expiry causes re-fetch from store
This commit is contained in:
1 parent
82ce20a029
commit
8d513763f2
6 files changed
+288
-14
No files matched your search
+7
-2
@@ -579,6 +579,11 @@ func runGateway() {
|
||||
server.SetPolicyEngine(permPE)
|
||||
server.SetPairingService(pairingStore)
|
||||
|
||||
// contextFileInterceptor is created inside wireManagedExtras (managed mode only).
|
||||
// Declared here so it can be passed to registerAllMethods → AgentsMethods
|
||||
// for immediate cache invalidation on agents.files.set.
|
||||
var contextFileInterceptor *tools.ContextFileInterceptor
|
||||
|
||||
// Managed mode: set agent store for tools_invoke context injection + wire extras
|
||||
if managedStores != nil && managedStores.Agents != nil {
|
||||
server.SetAgentStore(managedStores.Agents)
|
||||
@@ -593,7 +598,7 @@ func runGateway() {
|
||||
}
|
||||
}
|
||||
|
||||
wireManagedExtras(managedStores, agentRouter, providerRegistry, msgBus, sessStore, toolsReg, toolPE, skillsLoader, hasMemory, traceCollector, workspace, cfg.Gateway.InjectionAction, cfg, sandboxMgr, dynamicLoader)
|
||||
contextFileInterceptor = wireManagedExtras(managedStores, agentRouter, providerRegistry, msgBus, sessStore, toolsReg, toolPE, skillsLoader, hasMemory, traceCollector, workspace, cfg.Gateway.InjectionAction, cfg, sandboxMgr, dynamicLoader)
|
||||
agentsH, skillsH, tracesH, mcpH, customToolsH, channelInstancesH, providersH, delegationsH, builtinToolsH := wireManagedHTTP(managedStores, cfg.Gateway.Token, msgBus, toolsReg, providerRegistry, permPE.IsOwner)
|
||||
if agentsH != nil {
|
||||
server.SetAgentsHandler(agentsH)
|
||||
@@ -654,7 +659,7 @@ func runGateway() {
|
||||
teamStoreForRPC = managedStores.Teams
|
||||
}
|
||||
|
||||
pairingMethods := registerAllMethods(server, agentRouter, sessStore, cronStore, pairingStore, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, agentStoreForRPC, isManaged, skillStore, configSecretsStore, teamStoreForRPC)
|
||||
pairingMethods := registerAllMethods(server, agentRouter, sessStore, cronStore, pairingStore, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, agentStoreForRPC, isManaged, skillStore, configSecretsStore, teamStoreForRPC, contextFileInterceptor)
|
||||
|
||||
// Channel manager
|
||||
channelMgr := channels.NewManager(msgBus)
|
||||
|
||||
@@ -25,6 +25,8 @@ import (
|
||||
// agent resolver (lazy-creates Loops from DB), virtual FS interceptors, memory tools,
|
||||
// and cache invalidation event subscribers.
|
||||
// PG store creation and tracing are handled in gateway.go before this is called.
|
||||
// Returns the ContextFileInterceptor so callers can pass it to AgentsMethods
|
||||
// for immediate cache invalidation on agents.files.set.
|
||||
func wireManagedExtras(
|
||||
stores *store.Stores,
|
||||
agentRouter *agent.Router,
|
||||
@@ -41,7 +43,7 @@ func wireManagedExtras(
|
||||
appCfg *config.Config,
|
||||
sandboxMgr sandbox.Manager,
|
||||
dynamicLoader *tools.DynamicToolLoader,
|
||||
) {
|
||||
) *tools.ContextFileInterceptor {
|
||||
// 1. Context file interceptor (created before resolver so callbacks can reference it)
|
||||
var contextFileInterceptor *tools.ContextFileInterceptor
|
||||
if stores.Agents != nil {
|
||||
@@ -354,6 +356,7 @@ func wireManagedExtras(
|
||||
}
|
||||
|
||||
slog.Info("managed mode: resolver + interceptors + cache subscribers wired")
|
||||
return contextFileInterceptor
|
||||
}
|
||||
|
||||
// wireManagedHTTP creates managed-mode HTTP handlers (agents + skills + traces + MCP + custom tools + channel instances + providers + delegations + builtin tools).
|
||||
|
||||
@@ -12,12 +12,12 @@ import (
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tools"
|
||||
)
|
||||
|
||||
func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore store.SessionStore, cronStore store.CronStore, pairingStore store.PairingStore, cfg *config.Config, cfgPath, workspace, dataDir string, msgBus *bus.MessageBus, execApprovalMgr *tools.ExecApprovalManager, agentStore store.AgentStore, isManaged bool, skillStore store.SkillStore, configSecretsStore store.ConfigSecretsStore, teamStore store.TeamStore) *methods.PairingMethods {
|
||||
func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore store.SessionStore, cronStore store.CronStore, pairingStore store.PairingStore, cfg *config.Config, cfgPath, workspace, dataDir string, msgBus *bus.MessageBus, execApprovalMgr *tools.ExecApprovalManager, agentStore store.AgentStore, isManaged bool, skillStore store.SkillStore, configSecretsStore store.ConfigSecretsStore, teamStore store.TeamStore, contextFileInterceptor *tools.ContextFileInterceptor) *methods.PairingMethods {
|
||||
router := server.Router()
|
||||
|
||||
// Phase 1: Core methods
|
||||
methods.NewChatMethods(agents, sessStore, isManaged, server.RateLimiter()).Register(router)
|
||||
methods.NewAgentsMethods(agents, cfg, cfgPath, workspace, agentStore, isManaged).Register(router)
|
||||
methods.NewAgentsMethods(agents, cfg, cfgPath, workspace, agentStore, isManaged, contextFileInterceptor).Register(router)
|
||||
methods.NewSessionsMethods(sessStore).Register(router)
|
||||
methods.NewConfigMethods(cfg, cfgPath, isManaged, configSecretsStore).Register(router)
|
||||
|
||||
|
||||
@@ -13,22 +13,24 @@ import (
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/gateway"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/tools"
|
||||
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
|
||||
)
|
||||
|
||||
// AgentsMethods handles agents.list, agents.create, agents.update, agents.delete,
|
||||
// agents.files.list/get/set, agent.identity.get.
|
||||
type AgentsMethods struct {
|
||||
agents *agent.Router
|
||||
cfg *config.Config
|
||||
cfgPath string
|
||||
workspace string
|
||||
agentStore store.AgentStore // nil in standalone mode
|
||||
isManaged bool
|
||||
agents *agent.Router
|
||||
cfg *config.Config
|
||||
cfgPath string
|
||||
workspace string
|
||||
agentStore store.AgentStore // nil in standalone mode
|
||||
interceptor *tools.ContextFileInterceptor // nil in standalone mode; invalidated on file writes
|
||||
isManaged bool
|
||||
}
|
||||
|
||||
func NewAgentsMethods(agents *agent.Router, cfg *config.Config, cfgPath, workspace string, agentStore store.AgentStore, isManaged bool) *AgentsMethods {
|
||||
return &AgentsMethods{agents: agents, cfg: cfg, cfgPath: cfgPath, workspace: workspace, agentStore: agentStore, isManaged: isManaged}
|
||||
func NewAgentsMethods(agents *agent.Router, cfg *config.Config, cfgPath, workspace string, agentStore store.AgentStore, isManaged bool, interceptor *tools.ContextFileInterceptor) *AgentsMethods {
|
||||
return &AgentsMethods{agents: agents, cfg: cfg, cfgPath: cfgPath, workspace: workspace, agentStore: agentStore, isManaged: isManaged, interceptor: interceptor}
|
||||
}
|
||||
|
||||
func (m *AgentsMethods) Register(router *gateway.MethodRouter) {
|
||||
@@ -331,6 +333,10 @@ func (m *AgentsMethods) handleUpdate(_ context.Context, client *gateway.Client,
|
||||
if err := m.agentStore.SetAgentContextFile(ctx, ag.ID, "IDENTITY.md", content); err != nil {
|
||||
slog.Warn("failed to update IDENTITY.md", "agent", params.AgentID, "error", err)
|
||||
}
|
||||
// Invalidate interceptor cache so updated IDENTITY.md is served immediately
|
||||
if m.interceptor != nil {
|
||||
m.interceptor.InvalidateAgent(ag.ID)
|
||||
}
|
||||
}
|
||||
|
||||
m.agents.InvalidateAgent(params.AgentID)
|
||||
|
||||
@@ -242,8 +242,12 @@ func (m *AgentsMethods) handleFilesSet(_ context.Context, client *gateway.Client
|
||||
return
|
||||
}
|
||||
|
||||
// Invalidate agent cache so new bootstrap content takes effect
|
||||
// Invalidate both caches so the new content is served immediately
|
||||
// without waiting for the ContextFileInterceptor's 5-minute TTL to expire.
|
||||
m.agents.InvalidateAgent(params.AgentID)
|
||||
if m.interceptor != nil {
|
||||
m.interceptor.InvalidateAgent(ag.ID)
|
||||
}
|
||||
|
||||
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{
|
||||
"agentId": params.AgentID,
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
package tools
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
// ---- minimal AgentStore stub for interceptor tests ----
|
||||
|
||||
type stubAgentStore struct {
|
||||
agentFiles []store.AgentContextFileData
|
||||
userFiles []store.UserContextFileData
|
||||
agentCallsN atomic.Int32 // counts GetAgentContextFiles calls
|
||||
setAgentCallN atomic.Int32
|
||||
setUserCallN atomic.Int32
|
||||
}
|
||||
|
||||
func (s *stubAgentStore) GetAgentContextFiles(_ context.Context, _ uuid.UUID) ([]store.AgentContextFileData, error) {
|
||||
s.agentCallsN.Add(1)
|
||||
return s.agentFiles, nil
|
||||
}
|
||||
func (s *stubAgentStore) SetAgentContextFile(_ context.Context, _ uuid.UUID, _, _ string) error {
|
||||
s.setAgentCallN.Add(1)
|
||||
return nil
|
||||
}
|
||||
func (s *stubAgentStore) GetUserContextFiles(_ context.Context, _ uuid.UUID, _ string) ([]store.UserContextFileData, error) {
|
||||
return s.userFiles, nil
|
||||
}
|
||||
func (s *stubAgentStore) SetUserContextFile(_ context.Context, _ uuid.UUID, _, _, _ string) error {
|
||||
s.setUserCallN.Add(1)
|
||||
return nil
|
||||
}
|
||||
func (s *stubAgentStore) DeleteUserContextFile(_ context.Context, _ uuid.UUID, _, _ string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Remaining interface methods — not exercised in these tests.
|
||||
func (s *stubAgentStore) Create(_ context.Context, _ *store.AgentData) error { return nil }
|
||||
func (s *stubAgentStore) GetByKey(_ context.Context, _ string) (*store.AgentData, error) { return nil, nil }
|
||||
func (s *stubAgentStore) GetByID(_ context.Context, _ uuid.UUID) (*store.AgentData, error) { return nil, nil }
|
||||
func (s *stubAgentStore) Update(_ context.Context, _ uuid.UUID, _ map[string]any) error { return nil }
|
||||
func (s *stubAgentStore) Delete(_ context.Context, _ uuid.UUID) error { return nil }
|
||||
func (s *stubAgentStore) List(_ context.Context, _ string) ([]store.AgentData, error) { return nil, nil }
|
||||
func (s *stubAgentStore) ShareAgent(_ context.Context, _ uuid.UUID, _, _, _ string) error { return nil }
|
||||
func (s *stubAgentStore) RevokeShare(_ context.Context, _ uuid.UUID, _ string) error { return nil }
|
||||
func (s *stubAgentStore) ListShares(_ context.Context, _ uuid.UUID) ([]store.AgentShareData, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (s *stubAgentStore) CanAccess(_ context.Context, _ uuid.UUID, _ string) (bool, string, error) {
|
||||
return true, "admin", nil
|
||||
}
|
||||
func (s *stubAgentStore) ListAccessible(_ context.Context, _ string) ([]store.AgentData, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (s *stubAgentStore) GetUserOverride(_ context.Context, _ uuid.UUID, _ string) (*store.UserAgentOverrideData, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (s *stubAgentStore) SetUserOverride(_ context.Context, _ *store.UserAgentOverrideData) error {
|
||||
return nil
|
||||
}
|
||||
func (s *stubAgentStore) GetOrCreateUserProfile(_ context.Context, _ uuid.UUID, _, _ string) (bool, error) {
|
||||
return false, nil
|
||||
}
|
||||
func (s *stubAgentStore) IsGroupFileWriter(_ context.Context, _ uuid.UUID, _, _ string) (bool, error) {
|
||||
return false, nil
|
||||
}
|
||||
func (s *stubAgentStore) AddGroupFileWriter(_ context.Context, _ uuid.UUID, _, _, _, _ string) error {
|
||||
return nil
|
||||
}
|
||||
func (s *stubAgentStore) RemoveGroupFileWriter(_ context.Context, _ uuid.UUID, _, _ string) error {
|
||||
return nil
|
||||
}
|
||||
func (s *stubAgentStore) ListGroupFileWriters(_ context.Context, _ uuid.UUID, _ string) ([]store.GroupFileWriterData, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// ---- Tests ----
|
||||
|
||||
// TestInterceptor_CacheHit verifies that a second read does NOT call GetAgentContextFiles again.
|
||||
func TestInterceptor_CacheHit(t *testing.T) {
|
||||
agentID := uuid.New()
|
||||
as := &stubAgentStore{
|
||||
agentFiles: []store.AgentContextFileData{
|
||||
{AgentID: agentID, FileName: "SOUL.md", Content: "you are helpful"},
|
||||
},
|
||||
}
|
||||
intc := NewContextFileInterceptor(as, "")
|
||||
|
||||
ctx := store.WithAgentID(context.Background(), agentID)
|
||||
|
||||
// First read — cache miss → goes to store
|
||||
content1, handled1, err := intc.readAgentFile(ctx, agentID, "SOUL.md")
|
||||
if err != nil || !handled1 || content1 != "you are helpful" {
|
||||
t.Fatalf("first read: want ('you are helpful', true, nil), got (%q, %v, %v)", content1, handled1, err)
|
||||
}
|
||||
if n := as.agentCallsN.Load(); n != 1 {
|
||||
t.Fatalf("expected 1 store call, got %d", n)
|
||||
}
|
||||
|
||||
// Second read — cache hit → should NOT call store again
|
||||
content2, _, _ := intc.readAgentFile(ctx, agentID, "SOUL.md")
|
||||
if content2 != "you are helpful" {
|
||||
t.Errorf("second read: expected cached content, got %q", content2)
|
||||
}
|
||||
if n := as.agentCallsN.Load(); n != 1 {
|
||||
t.Errorf("cache hit should not call store again, got %d calls", n)
|
||||
}
|
||||
}
|
||||
|
||||
// TestInterceptor_InvalidateAgent_ClearsCache verifies that after InvalidateAgent,
|
||||
// the next read fetches fresh content from the store (not cached stale content).
|
||||
func TestInterceptor_InvalidateAgent_ClearsCache(t *testing.T) {
|
||||
agentID := uuid.New()
|
||||
as := &stubAgentStore{
|
||||
agentFiles: []store.AgentContextFileData{
|
||||
{AgentID: agentID, FileName: "SOUL.md", Content: "old content"},
|
||||
},
|
||||
}
|
||||
intc := NewContextFileInterceptor(as, "")
|
||||
ctx := store.WithAgentID(context.Background(), agentID)
|
||||
|
||||
// Warm up cache with old content
|
||||
intc.readAgentFile(ctx, agentID, "SOUL.md")
|
||||
if n := as.agentCallsN.Load(); n != 1 {
|
||||
t.Fatalf("expected 1 store call after warm-up, got %d", n)
|
||||
}
|
||||
|
||||
// Simulate wizard writing new content to the store
|
||||
as.agentFiles = []store.AgentContextFileData{
|
||||
{AgentID: agentID, FileName: "SOUL.md", Content: "new wizard content"},
|
||||
}
|
||||
|
||||
// WITHOUT invalidation: stale cache is still served
|
||||
content, _, _ := intc.readAgentFile(ctx, agentID, "SOUL.md")
|
||||
if content != "old content" {
|
||||
t.Errorf("expected stale cached content before invalidation, got %q", content)
|
||||
}
|
||||
if n := as.agentCallsN.Load(); n != 1 {
|
||||
t.Errorf("expected no extra store call (cache hit), got %d", n)
|
||||
}
|
||||
|
||||
// Invalidate — this is what agents.files.set must call
|
||||
intc.InvalidateAgent(agentID)
|
||||
|
||||
// AFTER invalidation: must fetch from store and return fresh content
|
||||
content, _, err := intc.readAgentFile(ctx, agentID, "SOUL.md")
|
||||
if err != nil {
|
||||
t.Fatalf("read after invalidation: unexpected error: %v", err)
|
||||
}
|
||||
if content != "new wizard content" {
|
||||
t.Errorf("after invalidation expected fresh content, got %q", content)
|
||||
}
|
||||
if n := as.agentCallsN.Load(); n != 2 {
|
||||
t.Errorf("expected 2 store calls total (1 warm-up + 1 post-invalidate), got %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
// TestInterceptor_InvalidateAgent_ClearsUserCache verifies that InvalidateAgent
|
||||
// also evicts per-user cache entries for that agent.
|
||||
func TestInterceptor_InvalidateAgent_ClearsUserCache(t *testing.T) {
|
||||
agentID := uuid.New()
|
||||
userID := "user-42"
|
||||
as := &stubAgentStore{
|
||||
userFiles: []store.UserContextFileData{
|
||||
{AgentID: agentID, UserID: userID, FileName: "USER.md", Content: "old user content"},
|
||||
},
|
||||
}
|
||||
intc := NewContextFileInterceptor(as, "")
|
||||
|
||||
// Warm user cache
|
||||
intc.readUserFile(context.Background(), agentID, userID, "USER.md")
|
||||
|
||||
// Update store content
|
||||
as.userFiles = []store.UserContextFileData{
|
||||
{AgentID: agentID, UserID: userID, FileName: "USER.md", Content: "new user content"},
|
||||
}
|
||||
|
||||
// Verify cache is served before invalidation
|
||||
content, _, _ := intc.readUserFile(context.Background(), agentID, userID, "USER.md")
|
||||
if content != "old user content" {
|
||||
t.Errorf("expected stale user cache before invalidation, got %q", content)
|
||||
}
|
||||
|
||||
// Invalidate the agent — must also clear user cache
|
||||
intc.InvalidateAgent(agentID)
|
||||
|
||||
// Now fresh content should come through
|
||||
content, _, err := intc.readUserFile(context.Background(), agentID, userID, "USER.md")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error after user cache invalidation: %v", err)
|
||||
}
|
||||
if content != "new user content" {
|
||||
t.Errorf("after InvalidateAgent expected fresh user content, got %q", content)
|
||||
}
|
||||
}
|
||||
|
||||
// TestInterceptor_InvalidateAgent_DoesNotAffectOtherAgents verifies that
|
||||
// invalidating one agent's cache does not evict another agent's entries.
|
||||
func TestInterceptor_InvalidateAgent_DoesNotAffectOtherAgents(t *testing.T) {
|
||||
agentA := uuid.New()
|
||||
agentB := uuid.New()
|
||||
|
||||
as := &stubAgentStore{}
|
||||
as.agentFiles = []store.AgentContextFileData{
|
||||
{AgentID: agentA, FileName: "SOUL.md", Content: "agent A soul"},
|
||||
}
|
||||
|
||||
intc := NewContextFileInterceptor(as, "")
|
||||
ctx := context.Background()
|
||||
|
||||
// Warm agent A cache
|
||||
intc.readAgentFile(ctx, agentA, "SOUL.md")
|
||||
callsAfterWarmup := as.agentCallsN.Load()
|
||||
|
||||
// Invalidate agent B — must not touch agent A's cache
|
||||
intc.InvalidateAgent(agentB)
|
||||
|
||||
// Agent A should still be served from cache (no extra store call)
|
||||
intc.readAgentFile(ctx, agentA, "SOUL.md")
|
||||
if n := as.agentCallsN.Load(); n != callsAfterWarmup {
|
||||
t.Errorf("invalidating agent B should not evict agent A's cache: got %d store calls", n)
|
||||
}
|
||||
}
|
||||
|
||||
// TestInterceptor_TTLExpiry verifies that entries older than TTL are re-fetched.
|
||||
func TestInterceptor_TTLExpiry(t *testing.T) {
|
||||
agentID := uuid.New()
|
||||
as := &stubAgentStore{
|
||||
agentFiles: []store.AgentContextFileData{
|
||||
{AgentID: agentID, FileName: "SOUL.md", Content: "soul content"},
|
||||
},
|
||||
}
|
||||
intc := NewContextFileInterceptor(as, "")
|
||||
intc.ttl = 10 * time.Millisecond // very short TTL for testing
|
||||
ctx := context.Background()
|
||||
|
||||
intc.readAgentFile(ctx, agentID, "SOUL.md")
|
||||
if n := as.agentCallsN.Load(); n != 1 {
|
||||
t.Fatalf("expected 1 store call, got %d", n)
|
||||
}
|
||||
|
||||
// Wait for TTL to expire
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
// Should fetch from store again after TTL
|
||||
intc.readAgentFile(ctx, agentID, "SOUL.md")
|
||||
if n := as.agentCallsN.Load(); n != 2 {
|
||||
t.Errorf("after TTL expiry expected 2 store calls, got %d", n)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user