feat(telegram): implement user-based isolation for persistent per-user memory across chats
This commit is contained in:
parent
31906c8f19
commit
b6d1ae637a
6 changed files with 36 additions and 28 deletions
|
|
@ -465,7 +465,7 @@ func TestLegacyCompact_PostTurn_ExceedsMessageThreshold(t *testing.T) {
|
|||
},
|
||||
}
|
||||
msgBus := bus.NewMessageBus()
|
||||
al := NewAgentLoop(cfg, msgBus, &simpleMockProvider{response: "summary"})
|
||||
al := NewAgentLoop(cfg, "", msgBus, &simpleMockProvider{response: "summary"})
|
||||
|
||||
defaultAgent := al.registry.GetDefaultAgent()
|
||||
if defaultAgent == nil {
|
||||
|
|
@ -617,7 +617,7 @@ func TestIngestCalledDuringTurn(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
al := NewAgentLoop(cfg, msgBus, &simpleMockProvider{response: "done"})
|
||||
al := NewAgentLoop(cfg, "", msgBus, &simpleMockProvider{response: "done"})
|
||||
defaultAgent := al.registry.GetDefaultAgent()
|
||||
if defaultAgent == nil {
|
||||
t.Fatal("expected default agent")
|
||||
|
|
@ -760,5 +760,5 @@ func testConfig(t *testing.T) *config.Config {
|
|||
|
||||
func newCMTestAgentLoop(cfg *config.Config) *AgentLoop {
|
||||
msgBus := bus.NewMessageBus()
|
||||
return NewAgentLoop(cfg, msgBus, &simpleMockProvider{response: "test"})
|
||||
return NewAgentLoop(cfg, "", msgBus, &simpleMockProvider{response: "test"})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -454,7 +454,7 @@ func TestAgentLoop_EmitsSessionSummarizeEvent(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
al := NewAgentLoop(cfg, msgBus, &simpleMockProvider{response: "summary text"})
|
||||
al := NewAgentLoop(cfg, "", msgBus, &simpleMockProvider{response: "summary text"})
|
||||
defaultAgent := al.registry.GetDefaultAgent()
|
||||
if defaultAgent == nil {
|
||||
t.Fatal("expected default agent")
|
||||
|
|
|
|||
|
|
@ -176,7 +176,9 @@ func TestProcessMessage_IsolatedTenant_UsesPrivateWorkspace(t *testing.T) {
|
|||
fmt.Printf("Agent Response: %s\n", resp)
|
||||
|
||||
// Verify the file was written to the ISOLATED workspace, NOT the global one
|
||||
isolatedPath := filepath.Join(tmpDir, "sessions", isolationID, "workspace", "secret.txt")
|
||||
// Since we now prefer SenderID for isolation, the workspace is under "user1"
|
||||
expectedIsoID := "user1"
|
||||
isolatedPath := filepath.Join(tmpDir, "sessions", expectedIsoID, "workspace", "secret.txt")
|
||||
globalPath := filepath.Join(tmpDir, "secret.txt")
|
||||
|
||||
// Debug: Print all files in tmpDir
|
||||
|
|
@ -195,9 +197,9 @@ func TestProcessMessage_IsolatedTenant_UsesPrivateWorkspace(t *testing.T) {
|
|||
t.Errorf("expected file at %s to NOT exist (leaked to global workspace)", globalPath)
|
||||
}
|
||||
|
||||
// Verify history is in the base sessions directory with the isolated key
|
||||
// agent:main:tenant-A becomes agent_main_tenant-A
|
||||
isoSessionPath := filepath.Join(tmpDir, "sessions", "agent_main_tenant-A.jsonl")
|
||||
// Verify history is in the base sessions directory with the session key
|
||||
// Based on resolveScopeKey(isolationID="user1"), it should be agent:main:user1
|
||||
isoSessionPath := filepath.Join(tmpDir, "sessions", "agent_main_user1.jsonl")
|
||||
if _, err := os.Stat(isoSessionPath); os.IsNotExist(err) {
|
||||
t.Errorf("expected history at %s to exist", isoSessionPath)
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -1478,7 +1478,13 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
|
|||
return "", routeErr
|
||||
}
|
||||
|
||||
agent, err := al.getOrCreateIsolatedAgent(route.AgentID, msg.Channel, msg.ChatID)
|
||||
// Prefer SenderID for isolation to ensure per-user workspaces that follow
|
||||
// individuals across different chat rooms (e.g. personal memory in groups).
|
||||
isolationID := msg.ChatID
|
||||
if msg.SenderID != "" {
|
||||
isolationID = msg.SenderID
|
||||
}
|
||||
agent, err := al.getOrCreateIsolatedAgent(route.AgentID, msg.Channel, isolationID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
|
@ -1491,8 +1497,8 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
|
|||
}
|
||||
|
||||
// Resolve session key from route, while preserving explicit agent-scoped keys.
|
||||
// If caller provides a session key, respect it. Otherwise, derive from chatID for isolation.
|
||||
scopeKey := resolveScopeKey(route, msg.SessionKey, msg.ChatID, agent.ID)
|
||||
// If caller provides a session key, respect it. Otherwise, derive from isolationID.
|
||||
scopeKey := resolveScopeKey(route, msg.SessionKey, isolationID, agent.ID)
|
||||
sessionKey := scopeKey
|
||||
|
||||
logger.InfoCF("agent", "Routed message",
|
||||
|
|
|
|||
|
|
@ -19,7 +19,6 @@ import (
|
|||
"github.com/sipeed/picoclaw/pkg/config"
|
||||
"github.com/sipeed/picoclaw/pkg/media"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/routing"
|
||||
"github.com/sipeed/picoclaw/pkg/tools"
|
||||
)
|
||||
|
||||
|
|
@ -670,7 +669,7 @@ func TestProcessMessage_MediaToolHandledSkipsFollowUpLLMAndFinalText(t *testing.
|
|||
if err != nil {
|
||||
t.Fatalf("resolveMessageRoute() error = %v", err)
|
||||
}
|
||||
sessionKey := resolveScopeKey(route, "", "chat1", route.AgentID)
|
||||
sessionKey := resolveScopeKey(route, "", "user1", route.AgentID)
|
||||
history := defaultAgent.Sessions.GetHistory(sessionKey)
|
||||
if len(history) == 0 {
|
||||
t.Fatal("expected session history to be saved")
|
||||
|
|
@ -1399,8 +1398,8 @@ func TestProcessMessage_UsesRouteSessionKey(t *testing.T) {
|
|||
},
|
||||
}
|
||||
|
||||
// With chatID isolation, session key is derived from chatID
|
||||
sessionKey := fmt.Sprintf("agent:main:%s", msg.ChatID)
|
||||
// With SenderID isolation, session key is derived from SenderID
|
||||
sessionKey := fmt.Sprintf("agent:main:%s", msg.SenderID)
|
||||
|
||||
defaultAgent := al.registry.GetDefaultAgent()
|
||||
if defaultAgent == nil {
|
||||
|
|
@ -2084,9 +2083,14 @@ func TestAgentLoop_ToolLimitUsesDedicatedFallback(t *testing.T) {
|
|||
al := NewAgentLoop(cfg, "", msgBus, provider)
|
||||
al.RegisterTool(&toolLimitTestTool{})
|
||||
|
||||
response, err := al.ProcessDirectWithChannel(context.Background(), "hello", "tool-limit", "test", "direct")
|
||||
msg := bus.InboundMessage{
|
||||
Channel: "test",
|
||||
ChatID: "direct",
|
||||
Content: "hello",
|
||||
}
|
||||
response, err := al.processMessage(context.Background(), msg)
|
||||
if err != nil {
|
||||
t.Fatalf("ProcessDirectWithChannel failed: %v", err)
|
||||
t.Fatalf("processMessage failed: %v", err)
|
||||
}
|
||||
if response != toolLimitResponse {
|
||||
t.Fatalf("response = %q, want %q", response, toolLimitResponse)
|
||||
|
|
@ -2096,14 +2100,10 @@ func TestAgentLoop_ToolLimitUsesDedicatedFallback(t *testing.T) {
|
|||
if defaultAgent == nil {
|
||||
t.Fatal("No default agent found")
|
||||
}
|
||||
route := al.registry.ResolveRoute(routing.RouteInput{
|
||||
Channel: "test",
|
||||
Peer: &routing.RoutePeer{
|
||||
Kind: "direct",
|
||||
ID: "cron",
|
||||
},
|
||||
})
|
||||
history := defaultAgent.Sessions.GetHistory(route.SessionKey)
|
||||
|
||||
// For unisolated "direct" chat, the session key defaults to agent:main:main
|
||||
sessionKey := "agent:main:main"
|
||||
history := defaultAgent.Sessions.GetHistory(sessionKey)
|
||||
if len(history) != 4 {
|
||||
t.Fatalf("history len = %d, want 4", len(history))
|
||||
}
|
||||
|
|
@ -2296,7 +2296,7 @@ func TestHandleReasoning(t *testing.T) {
|
|||
},
|
||||
}
|
||||
msgBus := bus.NewMessageBus()
|
||||
return NewAgentLoop(cfg, msgBus, &mockProvider{}), msgBus
|
||||
return NewAgentLoop(cfg, "", msgBus, &mockProvider{}), msgBus
|
||||
}
|
||||
|
||||
t.Run("skips when any required field is empty", func(t *testing.T) {
|
||||
|
|
|
|||
|
|
@ -362,7 +362,7 @@ func TestDrainBusToSteering_RequeuesDifferentScopeMessage(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
al := NewAgentLoop(cfg, msgBus, &mockProvider{})
|
||||
al := NewAgentLoop(cfg, "", msgBus, &mockProvider{})
|
||||
|
||||
activeMsg := bus.InboundMessage{
|
||||
Channel: "telegram",
|
||||
|
|
@ -1511,7 +1511,7 @@ func TestAgentLoop_Steering_SkippedToolsHaveErrorResults(t *testing.T) {
|
|||
}
|
||||
|
||||
msgBus := bus.NewMessageBus()
|
||||
al := NewAgentLoop(cfg, msgBus, wrappedProvider)
|
||||
al := NewAgentLoop(cfg, "", msgBus, wrappedProvider)
|
||||
al.RegisterTool(tool1)
|
||||
al.RegisterTool(tool2)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue