Merge pull request #3224 from Ethan1918/fix/clear-routed-agent-session
fix(agent): clear routed agent session
This commit is contained in:
commit
0f4a997b47
6 changed files with 220 additions and 7 deletions
|
|
@ -333,6 +333,15 @@ func (al *AgentLoop) buildCommandsRuntime(
|
||||||
if opts == nil {
|
if opts == nil {
|
||||||
return fmt.Errorf("process options not available")
|
return fmt.Errorf("process options not available")
|
||||||
}
|
}
|
||||||
|
// /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)
|
return al.contextManager.Clear(ctx, opts.SessionKey)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -3365,6 +3365,105 @@ func TestProcessMessage_CommandOutcomes(t *testing.T) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestProcessMessage_ClearCommandClearsRoutedAgentSession(t *testing.T) {
|
||||||
|
workspace := t.TempDir()
|
||||||
|
cfg := &config.Config{
|
||||||
|
Agents: config.AgentsConfig{
|
||||||
|
Defaults: config.AgentDefaults{
|
||||||
|
Workspace: filepath.Join(workspace, "default"),
|
||||||
|
ModelName: "test-model",
|
||||||
|
MaxTokens: 4096,
|
||||||
|
MaxToolIterations: 10,
|
||||||
|
},
|
||||||
|
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(), &countingMockProvider{response: "LLM reply"})
|
||||||
|
mainAgent, ok := al.registry.GetAgent("main")
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("expected main agent")
|
||||||
|
}
|
||||||
|
supportAgent, ok := al.registry.GetAgent("support")
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("expected support agent")
|
||||||
|
}
|
||||||
|
|
||||||
|
msg := testInboundMessage(bus.InboundMessage{
|
||||||
|
Context: bus.InboundContext{
|
||||||
|
Channel: "dingtalk",
|
||||||
|
ChatID: "chat1",
|
||||||
|
ChatType: "direct",
|
||||||
|
SenderID: "user1",
|
||||||
|
},
|
||||||
|
Content: "/clear",
|
||||||
|
})
|
||||||
|
route, routedAgent, err := al.resolveMessageRoute(msg)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("resolveMessageRoute() error = %v", err)
|
||||||
|
}
|
||||||
|
if routedAgent != supportAgent {
|
||||||
|
t.Fatalf("routed agent = %s, want support", routedAgent.ID)
|
||||||
|
}
|
||||||
|
sessionKey := al.allocateRouteSession(route, msg).SessionKey
|
||||||
|
|
||||||
|
mainHistory := []providers.Message{{Role: "user", Content: "main history"}}
|
||||||
|
supportHistory := []providers.Message{{Role: "user", Content: "support history"}}
|
||||||
|
mainAgent.Sessions.SetHistory(sessionKey, mainHistory)
|
||||||
|
mainAgent.Sessions.SetSummary(sessionKey, "main summary")
|
||||||
|
supportAgent.Sessions.SetHistory(sessionKey, supportHistory)
|
||||||
|
supportAgent.Sessions.SetSummary(sessionKey, "support summary")
|
||||||
|
|
||||||
|
response, err := al.processMessage(context.Background(), msg)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("processMessage() error = %v", err)
|
||||||
|
}
|
||||||
|
if response != "Chat history cleared!" {
|
||||||
|
t.Fatalf("response = %q, want clear confirmation", response)
|
||||||
|
}
|
||||||
|
|
||||||
|
if got := supportAgent.Sessions.GetHistory(sessionKey); len(got) != 0 {
|
||||||
|
t.Fatalf("support history len = %d, want 0", len(got))
|
||||||
|
}
|
||||||
|
if got := supportAgent.Sessions.GetSummary(sessionKey); got != "" {
|
||||||
|
t.Fatalf("support summary = %q, want empty", got)
|
||||||
|
}
|
||||||
|
if got := mainAgent.Sessions.GetHistory(sessionKey); len(got) != len(mainHistory) {
|
||||||
|
t.Fatalf("main history len = %d, want %d", len(got), len(mainHistory))
|
||||||
|
} else if got[0].Role != mainHistory[0].Role {
|
||||||
|
t.Fatalf("main history[0].Role = %q, want %q", got[0].Role, mainHistory[0].Role)
|
||||||
|
} else if got[0].Content != mainHistory[0].Content {
|
||||||
|
t.Fatalf("main history[0].Content = %q, want %q", got[0].Content, mainHistory[0].Content)
|
||||||
|
}
|
||||||
|
if got := mainAgent.Sessions.GetSummary(sessionKey); got != "main summary" {
|
||||||
|
t.Fatalf("main summary = %q, want %q", got, "main summary")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestProcessMessage_MCPCommandsHandledWithoutLLMCall(t *testing.T) {
|
func TestProcessMessage_MCPCommandsHandledWithoutLLMCall(t *testing.T) {
|
||||||
tmpDir, err := os.MkdirTemp("", "agent-test-*")
|
tmpDir, err := os.MkdirTemp("", "agent-test-*")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
@ -160,10 +162,18 @@ func (m *seahorseContextManager) Clear(ctx context.Context, sessionKey string) e
|
||||||
if err := m.engine.ClearSession(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
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -7,7 +7,6 @@ import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"log"
|
|
||||||
"maps"
|
"maps"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue