diff --git a/pkg/agent/context_legacy.go b/pkg/agent/context_legacy.go index eeaba3de9a..075f21741f 100644 --- a/pkg/agent/context_legacy.go +++ b/pkg/agent/context_legacy.go @@ -24,7 +24,7 @@ func (m *legacyContextManager) Assemble(_ context.Context, req *AssembleRequest) // Legacy: read history from session, return as-is. // Budget enforcement happens in BuildMessages caller via // isOverContextBudget + forceCompression. - agent := m.al.registry.GetDefaultAgent() + agent := m.al.agentForSession(req.SessionKey) if agent == nil { return &AssembleResponse{}, nil } @@ -77,7 +77,7 @@ func (m *legacyContextManager) Clear(_ context.Context, sessionKey string) error // maybeSummarize triggers summarization if the session history exceeds thresholds. // It runs asynchronously in a goroutine. func (m *legacyContextManager) maybeSummarize(sessionKey string) { - agent := m.al.registry.GetDefaultAgent() + agent := m.al.agentForSession(sessionKey) if agent == nil { return } @@ -115,7 +115,7 @@ type compressionResult struct { // It drops the oldest ~50% of Turns (a Turn is a complete user→LLM→response // cycle, as defined in #1316), so tool-call sequences are never split. func (m *legacyContextManager) forceCompression(sessionKey string) (compressionResult, bool) { - agent := m.al.registry.GetDefaultAgent() + agent := m.al.agentForSession(sessionKey) if agent == nil { return compressionResult{}, false } diff --git a/pkg/agent/context_manager_test.go b/pkg/agent/context_manager_test.go index bdccf4f693..afd8e3b1ac 100644 --- a/pkg/agent/context_manager_test.go +++ b/pkg/agent/context_manager_test.go @@ -15,6 +15,7 @@ import ( "github.com/sipeed/picoclaw/pkg/config" runtimeevents "github.com/sipeed/picoclaw/pkg/events" "github.com/sipeed/picoclaw/pkg/providers" + "github.com/sipeed/picoclaw/pkg/session" ) // --------------------------------------------------------------------------- @@ -874,3 +875,207 @@ func newCMTestAgentLoop(cfg *config.Config) *AgentLoop { msgBus := bus.NewMessageBus() return NewAgentLoop(cfg, msgBus, &simpleMockProvider{response: "test"}) } + +// --------------------------------------------------------------------------- +// Routed-agent regression tests +// --------------------------------------------------------------------------- +// +// These tests mirror the dispatch setup from +// TestClearCommandRoutedAgentCallsContextManagerClear: a default agent ("main") +// and a routed agent ("support"), with history seeded ONLY into the routed +// agent's session store. The legacy context manager must resolve session +// ownership via agentForSession instead of assuming GetDefaultAgent(), or the +// routed store is invisible to Assemble / maybeSummarize / forceCompression. + +// newRoutedCMTestAgentLoop builds an AgentLoop with a default agent (main) and +// a routed agent (support). defaults may override AgentDefaults (e.g. a low +// SummarizeMessageThreshold); nil means the standard test defaults. +func newRoutedCMTestAgentLoop(t *testing.T, defaults *config.AgentDefaults) *AgentLoop { + t.Helper() + workspace := t.TempDir() + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: filepath.Join(workspace, "main"), + 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")}, + }, + }, + Session: config.SessionConfig{ + Dimensions: []string{"chat"}, + }, + } + if defaults != nil { + cfg.Agents.Defaults = *defaults + } + msgBus := bus.NewMessageBus() + return NewAgentLoop(cfg, msgBus, &simpleMockProvider{response: "summary"}) +} + +// routedSessionKey builds an opaque session key scoped to the given agent and +// registers its scope metadata on that agent's store so agentForSession can +// resolve ownership (the same resolution Clear() relies on). +func routedSessionKey(t *testing.T, al *AgentLoop, agentID string) string { + t.Helper() + agent, ok := al.registry.GetAgent(agentID) + if !ok || agent == nil { + t.Fatalf("expected agent %q in registry", agentID) + } + scope := &session.SessionScope{ + Version: session.ScopeVersionV1, + AgentID: agentID, + Channel: "cli", + Account: "default", + Dimensions: []string{"chat"}, + Values: map[string]string{ + "chat": "direct:user1", + }, + } + key := session.BuildSessionKey(*scope) + ensureSessionMetadata(agent.Sessions, key, scope, nil) + return key +} + +// Assemble must read history/summary from the routed agent's session store, +// not the default agent's store. +func TestLegacyAssemble_RoutedAgent(t *testing.T) { + al := newRoutedCMTestAgentLoop(t, nil) + support, ok := al.registry.GetAgent("support") + if !ok || support == nil { + t.Fatal("expected support agent") + } + key := routedSessionKey(t, al, "support") + + history := []providers.Message{ + {Role: "user", Content: "what did I want before ice cream?"}, + {Role: "assistant", Content: "you wanted a hot dog"}, + } + support.Sessions.SetHistory(key, history) + support.Sessions.SetSummary(key, "early summary") + if err := support.Sessions.Save(key); err != nil { + t.Fatalf("Save: %v", err) + } + + resp, err := al.contextManager.Assemble(context.Background(), &AssembleRequest{SessionKey: key}) + if err != nil { + t.Fatalf("Assemble() error = %v", err) + } + if len(resp.History) != len(history) { + t.Fatalf("Assemble() history = %d messages, want %d (routed agent history must be loaded)", + len(resp.History), len(history)) + } + if resp.History[0].Content != history[0].Content { + t.Fatalf("history[0] = %q, want %q", resp.History[0].Content, history[0].Content) + } + if resp.Summary != "early summary" { + t.Fatalf("Assemble() summary = %q, want %q", resp.Summary, "early summary") + } +} + +// maybeSummarize must count messages in the routed +// agent's store and summarize against the routed agent. +func TestLegacyCompact_Summarize_RoutedAgent(t *testing.T) { + al := newRoutedCMTestAgentLoop(t, &config.AgentDefaults{ + ContextWindow: 8000, + SummarizeMessageThreshold: 2, + SummarizeTokenPercent: 75, + }) + support, ok := al.registry.GetAgent("support") + if !ok || support == nil { + t.Fatal("expected support agent") + } + key := routedSessionKey(t, al, "support") + + // 6 messages > threshold of 2 + history := []providers.Message{ + {Role: "user", Content: "q1"}, + {Role: "assistant", Content: "a1"}, + {Role: "user", Content: "q2"}, + {Role: "assistant", Content: "a2"}, + {Role: "user", Content: "q3"}, + {Role: "assistant", Content: "a3"}, + } + support.Sessions.SetHistory(key, history) + + runtimeCh, closeRuntimeEvents := subscribeRuntimeEventsForTest( + t, + al, + 16, + runtimeevents.KindAgentSessionSummarize, + ) + defer closeRuntimeEvents() + + err := al.contextManager.Compact(context.Background(), &CompactRequest{ + SessionKey: key, + Reason: ContextCompressReasonSummarize, + }) + if err != nil { + t.Fatalf("Compact() error = %v", err) + } + + waitForRuntimeEvent(t, runtimeCh, 5*time.Second, func(evt runtimeevents.Event) bool { + return evt.Kind == runtimeevents.KindAgentSessionSummarize + }) + + newHistory := support.Sessions.GetHistory(key) + if len(newHistory) >= len(history) { + t.Fatalf("expected summarization to reduce routed history from %d messages, got %d", + len(history), len(newHistory)) + } + if summary := support.Sessions.GetSummary(key); summary == "" { + t.Fatal("expected summary written to routed agent's store") + } +} + +// TestLegacyCompact_Overflow_RoutedAgent guards forceCompression: overflow +// compression must drop oldest turns in the routed agent's store and leave the +// default agent's store untouched. +func TestLegacyCompact_Overflow_RoutedAgent(t *testing.T) { + al := newRoutedCMTestAgentLoop(t, nil) + defaultAgent := al.registry.GetDefaultAgent() + if defaultAgent == nil { + t.Fatal("expected default agent") + } + support, ok := al.registry.GetAgent("support") + if !ok || support == nil { + t.Fatal("expected support agent") + } + key := routedSessionKey(t, al, "support") + + history := []providers.Message{ + {Role: "user", Content: "msg 1"}, + {Role: "assistant", Content: "resp 1"}, + {Role: "user", Content: "msg 2"}, + {Role: "assistant", Content: "resp 2"}, + {Role: "user", Content: "msg 3"}, + } + support.Sessions.SetHistory(key, history) + + err := al.contextManager.Compact(context.Background(), &CompactRequest{ + SessionKey: key, + Reason: ContextCompressReasonRetry, + }) + if err != nil { + t.Fatalf("Compact() error = %v", err) + } + + newHistory := support.Sessions.GetHistory(key) + if len(newHistory) >= len(history) { + t.Fatalf("expected compressed routed history, got %d messages (was %d)", + len(newHistory), len(history)) + } + summary := support.Sessions.GetSummary(key) + if !strings.Contains(summary, "Emergency compression") { + t.Fatalf("expected compression note in routed summary, got %q", summary) + } + + // The default agent's store must not be touched by routed compression. + if h := defaultAgent.Sessions.GetHistory(key); len(h) != 0 { + t.Fatalf("default agent store unexpectedly has %d messages for routed key", len(h)) + } +} diff --git a/pkg/agent/context_seahorse.go b/pkg/agent/context_seahorse.go index 4e4c55fd90..f0b4fe7672 100644 --- a/pkg/agent/context_seahorse.go +++ b/pkg/agent/context_seahorse.go @@ -56,11 +56,20 @@ func newSeahorseContextManager(_ json.RawMessage, al *AgentLoop) (ContextManager al.RegisterTool(seahorse.NewGrepTool(retrieval)) al.RegisterTool(seahorse.NewExpandTool(retrieval)) - // Bootstrap all existing sessions at startup - if agent.Sessions != nil { - ctx := context.Background() - for _, sessionKey := range agent.Sessions.ListSessions() { - mgr.bootstrapSession(ctx, sessionKey) + // Bootstrap all existing sessions at startup, for ALL registered agents. + // Routed agents keep history in their own session stores; a seahorse + // turn for a routed session would otherwise start with empty context + // because only the default agent's store was imported. + ctx := context.Background() + if al.registry != nil { + for _, agentID := range al.registry.ListAgentIDs() { + agent, ok := al.registry.GetAgent(agentID) + if !ok || agent == nil || agent.Sessions == nil { + continue + } + for _, sessionKey := range agent.Sessions.ListSessions() { + mgr.bootstrapSession(ctx, agent.Sessions, sessionKey) + } } } @@ -179,12 +188,16 @@ func (m *seahorseContextManager) Clear(ctx context.Context, sessionKey string) e } // bootstrapSession reconciles JSONL session history into seahorse SQLite. -func (m *seahorseContextManager) bootstrapSession(ctx context.Context, sessionKey string) { - if m.sessions == nil { +func (m *seahorseContextManager) bootstrapSession( + ctx context.Context, + sessions session.SessionStore, + sessionKey string, +) { + if sessions == nil { return } - history := m.sessions.GetHistory(sessionKey) + history := sessions.GetHistory(sessionKey) if len(history) == 0 { return } diff --git a/pkg/agent/context_seahorse_test.go b/pkg/agent/context_seahorse_test.go index 497e1fc449..7fb2932e73 100644 --- a/pkg/agent/context_seahorse_test.go +++ b/pkg/agent/context_seahorse_test.go @@ -3,6 +3,7 @@ package agent import ( "context" "fmt" + "path/filepath" "strings" "testing" "time" @@ -1165,3 +1166,78 @@ func TestSeahorseSummarizeSkipsCondensedWhenBelowThreshold(t *testing.T) { t.Errorf("BUG: condensed created when tokens (%d) < threshold (%d)", tokensBefore, threshold) } } + +// --------------------------------------------------------------------------- +// Routed-agent regression tests +// --------------------------------------------------------------------------- + +// routedSeahorseConfig returns a two-agent config (main default + support +// routed) sharing the given workspace root. +// ctxManager "" selects the legacy manager; "seahorse" selects the seahorse +// context manager. +func routedSeahorseConfig(workspace, ctxManager string) *config.Config { + return &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: filepath.Join(workspace, "main"), + ModelName: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + ContextManager: ctxManager, + }, + List: []config.AgentConfig{ + {ID: "main", Default: true, Workspace: filepath.Join(workspace, "main")}, + {ID: "support", Workspace: filepath.Join(workspace, "support")}, + }, + }, + } +} + +// The constructor must import pre-existing history for ALL registered agents, +// not just the default agent's store. +func TestSeahorseBootstrap_RoutedAgent(t *testing.T) { + workspace := t.TempDir() + + // Phase 1: seed routed-agent history on disk (legacy manager). + seedCfg := routedSeahorseConfig(workspace, "") + al1 := NewAgentLoop(seedCfg, bus.NewMessageBus(), &seahorseTestProvider{}) + support1, ok := al1.registry.GetAgent("support") + if !ok || support1 == nil { + t.Fatal("expected support agent") + } + key := routedSessionKey(t, al1, "support") + + history := []providers.Message{ + {Role: "user", Content: "what did I want before ice cream?"}, + {Role: "assistant", Content: "you wanted a hot dog"}, + } + support1.Sessions.SetHistory(key, history) + if err := support1.Sessions.Save(key); err != nil { + t.Fatalf("Save: %v", err) + } + + // Phase 2: construct a seahorse-backed loop against the same workspaces. + // The constructor bootstraps pre-existing sessions into SQLite. + seahorseCfg := routedSeahorseConfig(workspace, "seahorse") + al2 := NewAgentLoop(seahorseCfg, bus.NewMessageBus(), &seahorseTestProvider{}) + seahorseCM, ok := al2.contextManager.(*seahorseContextManager) + if !ok { + t.Fatal("expected seahorseContextManager") + } + + resp, err := seahorseCM.Assemble(context.Background(), &AssembleRequest{ + SessionKey: key, + Budget: 100000, + MaxTokens: 4096, + }) + if err != nil { + t.Fatalf("Assemble: %v", err) + } + if len(resp.History) != len(history) { + t.Fatalf("seahorse history = %d messages, want %d (routed agent sessions must be bootstrapped)", + len(resp.History), len(history)) + } + if resp.History[0].Content != history[0].Content { + t.Fatalf("history[0] = %q, want %q", resp.History[0].Content, history[0].Content) + } +}