refactor(agent): route /clear through ContextManager.Clear for all agents
Address review feedback: the routed-agent /clear path bypassed the public ContextManager.Clear contract via an unexported hook. Restore the single Clear call in the command path and resolve the session's owning agent inside the built-in implementations instead: - legacy: Clear resolves the owning agent via agentForSession instead of assuming the default agent - seahorse: drop ClearContextStore; Clear wipes the engine state and the owning agent's session store - command path: persist session scope metadata before Clear so ownership resolves even when /clear is the session's first message - add coverage that a custom ContextManager receives Clear for routed agents Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
c4fb7a2001
commit
ffffb6a0cb
4 changed files with 123 additions and 46 deletions
|
|
@ -333,7 +333,16 @@ func (al *AgentLoop) buildCommandsRuntime(
|
||||||
if opts == nil {
|
if opts == nil {
|
||||||
return fmt.Errorf("process options not available")
|
return fmt.Errorf("process options not available")
|
||||||
}
|
}
|
||||||
return al.clearAgentSessionContext(ctx, agent, opts.SessionKey)
|
// /clear can arrive before any turn has persisted session scope
|
||||||
|
// metadata (runAgentLoop records it per turn), so record it here to
|
||||||
|
// let the ContextManager resolve which agent owns the session.
|
||||||
|
ensureSessionMetadata(
|
||||||
|
agent.Sessions,
|
||||||
|
opts.Dispatch.SessionKey,
|
||||||
|
opts.Dispatch.SessionScope,
|
||||||
|
opts.Dispatch.SessionAliases,
|
||||||
|
)
|
||||||
|
return al.contextManager.Clear(ctx, opts.SessionKey)
|
||||||
}
|
}
|
||||||
|
|
||||||
rt.AskSideQuestion = func(ctx context.Context, question string) (string, error) {
|
rt.AskSideQuestion = func(ctx context.Context, question string) (string, error) {
|
||||||
|
|
@ -363,40 +372,6 @@ func (al *AgentLoop) buildCommandsRuntime(
|
||||||
return rt
|
return rt
|
||||||
}
|
}
|
||||||
|
|
||||||
type contextStoreClearer interface {
|
|
||||||
ClearContextStore(ctx context.Context, sessionKey string) error
|
|
||||||
}
|
|
||||||
|
|
||||||
func (al *AgentLoop) clearAgentSessionContext(
|
|
||||||
ctx context.Context,
|
|
||||||
agent *AgentInstance,
|
|
||||||
sessionKey string,
|
|
||||||
) error {
|
|
||||||
if agent == nil || agent.Sessions == nil {
|
|
||||||
return fmt.Errorf("sessions not initialized")
|
|
||||||
}
|
|
||||||
if al != nil && al.registry != nil && agent == al.registry.GetDefaultAgent() {
|
|
||||||
if al.contextManager != nil {
|
|
||||||
return al.contextManager.Clear(ctx, sessionKey)
|
|
||||||
}
|
|
||||||
return clearAgentSessionStore(agent, sessionKey)
|
|
||||||
}
|
|
||||||
if al != nil && al.contextManager != nil {
|
|
||||||
if clearer, ok := al.contextManager.(contextStoreClearer); ok {
|
|
||||||
if err := clearer.ClearContextStore(ctx, sessionKey); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return clearAgentSessionStore(agent, sessionKey)
|
|
||||||
}
|
|
||||||
|
|
||||||
func clearAgentSessionStore(agent *AgentInstance, sessionKey string) error {
|
|
||||||
agent.Sessions.SetHistory(sessionKey, []providers.Message{})
|
|
||||||
agent.Sessions.SetSummary(sessionKey, "")
|
|
||||||
return agent.Sessions.Save(sessionKey)
|
|
||||||
}
|
|
||||||
|
|
||||||
func summarizeMCPToolParameters(schema any) []commands.MCPToolParameterInfo {
|
func summarizeMCPToolParameters(schema any) []commands.MCPToolParameterInfo {
|
||||||
schemaMap := normalizeMCPSchema(schema)
|
schemaMap := normalizeMCPSchema(schema)
|
||||||
properties, ok := schemaMap["properties"].(map[string]any)
|
properties, ok := schemaMap["properties"].(map[string]any)
|
||||||
|
|
|
||||||
|
|
@ -63,7 +63,9 @@ func (m *legacyContextManager) Ingest(_ context.Context, _ *IngestRequest) error
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *legacyContextManager) Clear(_ context.Context, sessionKey string) error {
|
func (m *legacyContextManager) Clear(_ context.Context, sessionKey string) error {
|
||||||
agent := m.al.registry.GetDefaultAgent()
|
// Routed (non-default) agents keep history in their own session store,
|
||||||
|
// so resolve the owning agent instead of assuming the default one.
|
||||||
|
agent := m.al.agentForSession(sessionKey)
|
||||||
if agent == nil || agent.Sessions == nil {
|
if agent == nil || agent.Sessions == nil {
|
||||||
return fmt.Errorf("sessions not initialized")
|
return fmt.Errorf("sessions not initialized")
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -4,6 +4,7 @@ import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"os"
|
"os"
|
||||||
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
|
|
@ -658,6 +659,91 @@ func TestIngestCalledDuringTurn(t *testing.T) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestClearCommandRoutedAgentCallsContextManagerClear(t *testing.T) {
|
||||||
|
cleanup := resetCMRegistry()
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
mock := &trackingContextManager{}
|
||||||
|
factory := func(cfg json.RawMessage, al *AgentLoop) (ContextManager, error) {
|
||||||
|
return mock, nil
|
||||||
|
}
|
||||||
|
if err := RegisterContextManager("clear_track_cm", factory); err != nil {
|
||||||
|
t.Fatalf("register failed: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
workspace := t.TempDir()
|
||||||
|
cfg := &config.Config{
|
||||||
|
Agents: config.AgentsConfig{
|
||||||
|
Defaults: config.AgentDefaults{
|
||||||
|
Workspace: filepath.Join(workspace, "default"),
|
||||||
|
ModelName: "test-model",
|
||||||
|
MaxTokens: 4096,
|
||||||
|
MaxToolIterations: 10,
|
||||||
|
ContextManager: "clear_track_cm",
|
||||||
|
},
|
||||||
|
List: []config.AgentConfig{
|
||||||
|
{
|
||||||
|
ID: "main",
|
||||||
|
Default: true,
|
||||||
|
Workspace: filepath.Join(workspace, "main"),
|
||||||
|
},
|
||||||
|
{
|
||||||
|
ID: "support",
|
||||||
|
Workspace: filepath.Join(workspace, "support"),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
Dispatch: &config.DispatchConfig{
|
||||||
|
Rules: []config.DispatchRule{
|
||||||
|
{
|
||||||
|
Name: "support-dingtalk",
|
||||||
|
Agent: "support",
|
||||||
|
When: config.DispatchSelector{
|
||||||
|
Channel: "dingtalk",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
Session: config.SessionConfig{
|
||||||
|
Dimensions: []string{"chat"},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
al := NewAgentLoop(cfg, bus.NewMessageBus(), &simpleMockProvider{response: "done"})
|
||||||
|
if al.contextManager != mock {
|
||||||
|
t.Fatalf("expected mock context manager, got %T", al.contextManager)
|
||||||
|
}
|
||||||
|
|
||||||
|
msg := testInboundMessage(bus.InboundMessage{
|
||||||
|
Context: bus.InboundContext{
|
||||||
|
Channel: "dingtalk",
|
||||||
|
ChatID: "chat1",
|
||||||
|
ChatType: "direct",
|
||||||
|
SenderID: "user1",
|
||||||
|
},
|
||||||
|
Content: "/clear",
|
||||||
|
})
|
||||||
|
route, _, err := al.resolveMessageRoute(msg)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("resolveMessageRoute() error = %v", err)
|
||||||
|
}
|
||||||
|
sessionKey := al.allocateRouteSession(route, msg).SessionKey
|
||||||
|
|
||||||
|
if _, err := al.processMessage(context.Background(), msg); err != nil {
|
||||||
|
t.Fatalf("processMessage() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if got := mock.clearCalls.Load(); got != 1 {
|
||||||
|
t.Fatalf("Clear calls = %d, want 1", got)
|
||||||
|
}
|
||||||
|
mock.mu.Lock()
|
||||||
|
gotKey := mock.lastClearKey
|
||||||
|
mock.mu.Unlock()
|
||||||
|
if gotKey != sessionKey {
|
||||||
|
t.Fatalf("Clear session key = %q, want %q", gotKey, sessionKey)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
// forceCompression edge cases (via legacy Compact)
|
// forceCompression edge cases (via legacy Compact)
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|
@ -712,10 +798,12 @@ type trackingContextManager struct {
|
||||||
assembleCalls atomic.Int64
|
assembleCalls atomic.Int64
|
||||||
compactCalls atomic.Int64
|
compactCalls atomic.Int64
|
||||||
ingestCalls atomic.Int64
|
ingestCalls atomic.Int64
|
||||||
|
clearCalls atomic.Int64
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
lastAssemble *AssembleRequest
|
lastAssemble *AssembleRequest
|
||||||
lastCompact *CompactRequest
|
lastCompact *CompactRequest
|
||||||
lastIngest *IngestRequest
|
lastIngest *IngestRequest
|
||||||
|
lastClearKey string
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *trackingContextManager) Assemble(_ context.Context, req *AssembleRequest) (*AssembleResponse, error) {
|
func (m *trackingContextManager) Assemble(_ context.Context, req *AssembleRequest) (*AssembleResponse, error) {
|
||||||
|
|
@ -742,7 +830,13 @@ func (m *trackingContextManager) Ingest(_ context.Context, req *IngestRequest) e
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *trackingContextManager) Clear(_ context.Context, _ string) error { return nil }
|
func (m *trackingContextManager) Clear(_ context.Context, sessionKey string) error {
|
||||||
|
m.clearCalls.Add(1)
|
||||||
|
m.mu.Lock()
|
||||||
|
m.lastClearKey = sessionKey
|
||||||
|
m.mu.Unlock()
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// resetCMRegistry clears the global factory registry and returns a cleanup
|
// resetCMRegistry clears the global factory registry and returns a cleanup
|
||||||
// function that restores the original state after the test.
|
// function that restores the original state after the test.
|
||||||
|
|
|
||||||
|
|
@ -20,6 +20,7 @@ import (
|
||||||
type seahorseContextManager struct {
|
type seahorseContextManager struct {
|
||||||
engine *seahorse.Engine
|
engine *seahorse.Engine
|
||||||
sessions session.SessionStore // for startup bootstrap
|
sessions session.SessionStore // for startup bootstrap
|
||||||
|
al *AgentLoop // for resolving the agent that owns a session
|
||||||
}
|
}
|
||||||
|
|
||||||
// newSeahorseContextManager creates a seahorse-backed ContextManager.
|
// newSeahorseContextManager creates a seahorse-backed ContextManager.
|
||||||
|
|
@ -47,6 +48,7 @@ func newSeahorseContextManager(_ json.RawMessage, al *AgentLoop) (ContextManager
|
||||||
mgr := &seahorseContextManager{
|
mgr := &seahorseContextManager{
|
||||||
engine: engine,
|
engine: engine,
|
||||||
sessions: agent.Sessions,
|
sessions: agent.Sessions,
|
||||||
|
al: al,
|
||||||
}
|
}
|
||||||
|
|
||||||
// Register seahorse tools with the agent's tool registry
|
// Register seahorse tools with the agent's tool registry
|
||||||
|
|
@ -157,21 +159,25 @@ func (m *seahorseContextManager) Ingest(ctx context.Context, req *IngestRequest)
|
||||||
|
|
||||||
// Clear removes all stored context for a session (seahorse DB + JSONL).
|
// Clear removes all stored context for a session (seahorse DB + JSONL).
|
||||||
func (m *seahorseContextManager) Clear(ctx context.Context, sessionKey string) error {
|
func (m *seahorseContextManager) Clear(ctx context.Context, sessionKey string) error {
|
||||||
if err := m.ClearContextStore(ctx, sessionKey); err != nil {
|
if err := m.engine.ClearSession(ctx, sessionKey); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if m.sessions != nil {
|
// The session may belong to a routed (non-default) agent whose JSONL
|
||||||
m.sessions.SetHistory(sessionKey, []providers.Message{})
|
// store differs from the bootstrap store, so clear the owner's store.
|
||||||
m.sessions.SetSummary(sessionKey, "")
|
sessions := m.sessions
|
||||||
return m.sessions.Save(sessionKey)
|
if m.al != nil {
|
||||||
|
if agent := m.al.agentForSession(sessionKey); agent != nil && agent.Sessions != nil {
|
||||||
|
sessions = agent.Sessions
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if sessions != nil {
|
||||||
|
sessions.SetHistory(sessionKey, []providers.Message{})
|
||||||
|
sessions.SetSummary(sessionKey, "")
|
||||||
|
return sessions.Save(sessionKey)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *seahorseContextManager) ClearContextStore(ctx context.Context, sessionKey string) error {
|
|
||||||
return m.engine.ClearSession(ctx, sessionKey)
|
|
||||||
}
|
|
||||||
|
|
||||||
// bootstrapSession reconciles JSONL session history into seahorse SQLite.
|
// bootstrapSession reconciles JSONL session history into seahorse SQLite.
|
||||||
func (m *seahorseContextManager) bootstrapSession(ctx context.Context, sessionKey string) {
|
func (m *seahorseContextManager) bootstrapSession(ctx context.Context, sessionKey string) {
|
||||||
if m.sessions == nil {
|
if m.sessions == nil {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue