diff --git a/README.md b/README.md index 7a88912c6..48d0b12dc 100644 --- a/README.md +++ b/README.md @@ -340,7 +340,7 @@ picoclaw gateway **4. Telegram command menu (auto-registered at startup)** -PicoClaw now keeps command definitions in one shared registry. On startup, Telegram will automatically register supported bot commands (for example `/start`, `/help`, `/show`, `/list`) so command menu and runtime behavior stay in sync. +PicoClaw now keeps command definitions in one shared registry. On startup, Telegram will automatically register supported bot commands (for example `/start`, `/help`, `/new`, `/session`, `/show`, `/list`) so command menu and runtime behavior stay in sync. If command registration fails (network/API transient errors), the channel still starts and PicoClaw retries registration in the background. @@ -727,6 +727,21 @@ PicoClaw stores data in your configured workspace (default: `~/.picoclaw/workspa └── USER.md # User preferences ``` +### Session Scope and Backlog + +Use `session.dm_scope` to control DM session isolation and `session.backlog_limit` to cap how many sessions are retained per scope: + +```json +{ + "session": { + "dm_scope": "per-channel-peer", + "backlog_limit": 20 + } +} +``` + +`/new` (or `/reset`) starts a fresh active session for the current scope. `/session list` and `/session resume ` operate within that same scope. + ### 🔒 Security Sandbox PicoClaw runs in a sandboxed environment by default. The agent can only access files and execute commands within the configured workspace. @@ -1182,6 +1197,10 @@ picoclaw agent -m "Hello" "model": "anthropic/claude-opus-4-5" } }, + "session": { + "dm_scope": "per-channel-peer", + "backlog_limit": 20 + }, "providers": { "openrouter": { "api_key": "sk-or-v1-xxx" diff --git a/README.zh.md b/README.zh.md index 608f183ad..bf5161cf0 100644 --- a/README.zh.md +++ b/README.zh.md @@ -309,7 +309,7 @@ PicoClaw 支持多种聊天平台,使您的 Agent 能够连接到任何地方 ### Telegram 命令注册(启动时自动同步) -PicoClaw 现在使用统一的命令定义来源。启动时会自动将 Telegram 支持的命令(例如 `/start`、`/help`、`/show`、`/list`)注册到 Bot 命令菜单,确保菜单展示与实际行为一致。 +PicoClaw 现在使用统一的命令定义来源。启动时会自动将 Telegram 支持的命令(例如 `/start`、`/help`、`/new`、`/session`、`/show`、`/list`)注册到 Bot 命令菜单,确保菜单展示与实际行为一致。 如果注册因网络或 API 短暂异常失败,不会阻塞 channel 启动;系统会在后台自动重试。 @@ -368,6 +368,21 @@ PicoClaw 将数据存储在您配置的工作区中(默认:`~/.picoclaw/work ``` +### 会话作用域与保留上限 + +使用 `session.dm_scope` 控制私聊会话隔离粒度,使用 `session.backlog_limit` 控制每个作用域保留多少历史会话: + +```json +{ + "session": { + "dm_scope": "per-channel-peer", + "backlog_limit": 20 + } +} +``` + +`/new`(或 `/reset`)会在当前作用域创建新会话;`/session list` 和 `/session resume ` 只在当前作用域内生效。 + ### 心跳 / 周期性任务 (Heartbeat) PicoClaw 可以自动执行周期性任务。在工作区创建 `HEARTBEAT.md` 文件: @@ -707,6 +722,10 @@ picoclaw agent -m "你好" "model": "anthropic/claude-opus-4-5" } }, + "session": { + "dm_scope": "per-channel-peer", + "backlog_limit": 20 + }, "providers": { "openrouter": { "api_key": "sk-or-v1-xxx" diff --git a/config/config.example.json b/config/config.example.json index fe3740289..dd156a188 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -9,6 +9,10 @@ "max_tool_iterations": 20 } }, + "session": { + "dm_scope": "per-channel-peer", + "backlog_limit": 20 + }, "model_list": [ { "model_name": "gpt4", diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index ac9b449a2..521f1806c 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -12,6 +12,7 @@ import ( "errors" "fmt" "path/filepath" + "strconv" "strings" "sync" "sync/atomic" @@ -449,12 +450,52 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) return al.processSystemMessage(ctx, msg) } + route, agent, err := al.resolveMessageRoute(msg) + if err != nil { + return "", err + } + // Check for commands - if response, handled := al.handleCommand(ctx, msg); handled { + if response, handled := al.handleCommand(ctx, msg, route, agent); handled { return response, nil } - // Route to determine agent and session key + // Reset message-tool state for this round so we don't skip publishing due to a previous round. + if tool, ok := agent.Tools.Get("message"); ok { + if mt, ok := tool.(tools.ContextualTool); ok { + mt.SetContext(msg.Channel, msg.ChatID) + } + } + + // Resolve active session from the routing scope. + scopeKey := resolveScopeKey(route, msg.SessionKey) + sessionKey, err := agent.Sessions.ResolveActive(scopeKey) + if err != nil { + return "", fmt.Errorf("resolve active session: %w", err) + } + + logger.InfoCF("agent", "Routed message", + map[string]any{ + "agent_id": agent.ID, + "scope_key": scopeKey, + "session_key": sessionKey, + "matched_by": route.MatchedBy, + "route_agent": route.AgentID, + "route_channel": route.Channel, + }) + + return al.runAgentLoop(ctx, agent, processOptions{ + SessionKey: sessionKey, + Channel: msg.Channel, + ChatID: msg.ChatID, + UserMessage: msg.Content, + DefaultResponse: defaultResponse, + EnableSummary: true, + SendResponse: false, + }) +} + +func (al *AgentLoop) resolveMessageRoute(msg bus.InboundMessage) (routing.ResolvedRoute, *AgentInstance, error) { route := al.registry.ResolveRoute(routing.RouteInput{ Channel: msg.Channel, AccountID: msg.Metadata["account_id"], @@ -469,38 +510,17 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) agent = al.registry.GetDefaultAgent() } if agent == nil { - return "", fmt.Errorf("no agent available for route (agent_id=%s)", route.AgentID) + return routing.ResolvedRoute{}, nil, fmt.Errorf("no agent available for route (agent_id=%s)", route.AgentID) } - // Reset message-tool state for this round so we don't skip publishing due to a previous round. - if tool, ok := agent.Tools.Get("message"); ok { - if mt, ok := tool.(tools.ContextualTool); ok { - mt.SetContext(msg.Channel, msg.ChatID) - } + return route, agent, nil +} + +func resolveScopeKey(route routing.ResolvedRoute, msgSessionKey string) string { + if msgSessionKey != "" && strings.HasPrefix(msgSessionKey, "agent:") { + return msgSessionKey } - - // Use routed session key, but honor pre-set agent-scoped keys (for ProcessDirect/cron) - sessionKey := route.SessionKey - if msg.SessionKey != "" && strings.HasPrefix(msg.SessionKey, "agent:") { - sessionKey = msg.SessionKey - } - - logger.InfoCF("agent", "Routed message", - map[string]any{ - "agent_id": agent.ID, - "session_key": sessionKey, - "matched_by": route.MatchedBy, - }) - - return al.runAgentLoop(ctx, agent, processOptions{ - SessionKey: sessionKey, - Channel: msg.Channel, - ChatID: msg.ChatID, - UserMessage: msg.Content, - DefaultResponse: defaultResponse, - EnableSummary: true, - SendResponse: false, - }) + return route.SessionKey } func (al *AgentLoop) processSystemMessage( @@ -1362,7 +1382,14 @@ func (al *AgentLoop) estimateTokens(messages []providers.Message) int { return totalChars * 2 / 5 } -func (al *AgentLoop) handleCommand(ctx context.Context, msg bus.InboundMessage) (string, bool) { +func (al *AgentLoop) handleCommand( + ctx context.Context, + msg bus.InboundMessage, + route routing.ResolvedRoute, + agent *AgentInstance, +) (string, bool) { + // Scope-aware command routing: session-affecting commands are resolved + // against route-derived scope keys so each chat/group keeps isolated history. content := strings.TrimSpace(msg.Content) if !strings.HasPrefix(content, "/") { return "", false @@ -1374,9 +1401,96 @@ func (al *AgentLoop) handleCommand(ctx context.Context, msg bus.InboundMessage) } cmd := parts[0] + if at := strings.Index(cmd, "@"); at > 0 { + cmd = cmd[:at] + } args := parts[1:] switch cmd { + case "/new", "/reset": + // Create and rotate session state only inside this scope. + scopeKey := resolveScopeKey(route, msg.SessionKey) + newSessionKey, err := agent.Sessions.StartNew(scopeKey) + if err != nil { + return fmt.Sprintf("Failed to start new session: %v", err), true + } + + backlogLimit := config.DefaultSessionBacklogLimit + if al.cfg != nil { + backlogLimit = al.cfg.Session.EffectiveBacklogLimit() + } + + pruned, err := agent.Sessions.Prune(scopeKey, backlogLimit) + if err != nil { + return fmt.Sprintf( + "Started new session (%s), but pruning old sessions failed: %v", + newSessionKey, + err, + ), true + } + + if len(pruned) == 0 { + return fmt.Sprintf("Started new session: %s", newSessionKey), true + } + return fmt.Sprintf("Started new session: %s (pruned %d old session(s))", newSessionKey, len(pruned)), true + + case "/session": + if len(args) < 1 { + return "Usage: /session [list|resume ]", true + } + + // List/resume operate on the same scope-local ordering used by /new. + scopeKey := resolveScopeKey(route, msg.SessionKey) + switch args[0] { + case "list": + list, err := agent.Sessions.List(scopeKey) + if err != nil { + return fmt.Sprintf("Failed to list sessions: %v", err), true + } + if len(list) == 0 { + return "No sessions found for current chat.", true + } + + lines := make([]string, 0, len(list)+1) + lines = append(lines, "Sessions for current chat:") + for _, item := range list { + activeMarker := " " + if item.Active { + activeMarker = "*" + } + updated := "-" + if !item.UpdatedAt.IsZero() { + updated = item.UpdatedAt.Format("2006-01-02 15:04") + } + lines = append(lines, fmt.Sprintf( + "%d. [%s] %s (%d msgs, updated %s)", + item.Ordinal, + activeMarker, + item.SessionKey, + item.MessageCnt, + updated, + )) + } + return strings.Join(lines, "\n"), true + + case "resume": + if len(args) != 2 { + return "Usage: /session resume ", true + } + index, err := strconv.Atoi(args[1]) + if err != nil || index < 1 { + return "Usage: /session resume ", true + } + sessionKey, err := agent.Sessions.Resume(scopeKey, index) + if err != nil { + return fmt.Sprintf("Failed to resume session %d: %v", index, err), true + } + return fmt.Sprintf("Resumed session %d: %s", index, sessionKey), true + + default: + return "Usage: /session [list|resume ]", true + } + case "/show": if len(args) < 1 { return "Usage: /show [model|channel|agents]", true diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 3565314fe..503471667 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "slices" + "strings" "testing" "time" @@ -13,6 +14,7 @@ import ( "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/providers" + "github.com/sipeed/picoclaw/pkg/routing" "github.com/sipeed/picoclaw/pkg/tools" ) @@ -406,6 +408,174 @@ func (h testHelper) executeAndGetResponse(tb testing.TB, ctx context.Context, ms const responseTimeout = 3 * time.Second +func TestProcessMessage_UsesResolvedActiveSession(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &simpleMockProvider{response: "ok"} + al := NewAgentLoop(cfg, msgBus, provider) + + msg := bus.InboundMessage{ + Channel: "telegram", + SenderID: "user1", + ChatID: "chat1", + Content: "hello", + Peer: bus.Peer{ + Kind: "direct", + ID: "user1", + }, + } + + route := al.registry.ResolveRoute(routing.RouteInput{ + Channel: msg.Channel, + Peer: extractPeer(msg), + }) + scopeKey := route.SessionKey + + defaultAgent := al.registry.GetDefaultAgent() + if defaultAgent == nil { + t.Fatal("No default agent found") + } + + rotated, err := defaultAgent.Sessions.StartNew(scopeKey) + if err != nil { + t.Fatalf("StartNew(%q) failed: %v", scopeKey, err) + } + + helper := testHelper{al: al} + _ = helper.executeAndGetResponse(t, context.Background(), msg) + + if got := len(defaultAgent.Sessions.GetHistory(scopeKey)); got != 0 { + t.Fatalf("expected base scope history len=0, got %d", got) + } + + rotatedHistory := defaultAgent.Sessions.GetHistory(rotated) + if len(rotatedHistory) != 2 { + t.Fatalf("expected rotated history len=2, got %d", len(rotatedHistory)) + } + if rotatedHistory[0].Role != "user" || rotatedHistory[0].Content != "hello" { + t.Fatalf("unexpected first message in rotated session: %+v", rotatedHistory[0]) + } +} + +func TestHandleCommand_NewAndSessionCommands(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + Session: config.SessionConfig{ + DMScope: "per-channel-peer", + BacklogLimit: 20, + }, + } + + msgBus := bus.NewMessageBus() + provider := &simpleMockProvider{response: "ok"} + al := NewAgentLoop(cfg, msgBus, provider) + helper := testHelper{al: al} + + baseMsg := bus.InboundMessage{ + Channel: "telegram", + SenderID: "user1", + ChatID: "chat1", + Peer: bus.Peer{ + Kind: "direct", + ID: "user1", + }, + } + + route := al.registry.ResolveRoute(routing.RouteInput{ + Channel: baseMsg.Channel, + Peer: extractPeer(baseMsg), + }) + scopeKey := route.SessionKey + defaultAgent := al.registry.GetDefaultAgent() + if defaultAgent == nil { + t.Fatal("No default agent found") + } + + respNew1 := helper.executeAndGetResponse(t, context.Background(), bus.InboundMessage{ + Channel: baseMsg.Channel, + SenderID: baseMsg.SenderID, + ChatID: baseMsg.ChatID, + Content: "/new", + Peer: baseMsg.Peer, + }) + if !strings.Contains(respNew1, scopeKey+"#2") { + t.Fatalf("/new response missing new session key, got: %q", respNew1) + } + + respNew2 := helper.executeAndGetResponse(t, context.Background(), bus.InboundMessage{ + Channel: baseMsg.Channel, + SenderID: baseMsg.SenderID, + ChatID: baseMsg.ChatID, + Content: "/new", + Peer: baseMsg.Peer, + }) + if !strings.Contains(respNew2, scopeKey+"#3") { + t.Fatalf("second /new response missing session #3, got: %q", respNew2) + } + + listResp := helper.executeAndGetResponse(t, context.Background(), bus.InboundMessage{ + Channel: baseMsg.Channel, + SenderID: baseMsg.SenderID, + ChatID: baseMsg.ChatID, + Content: "/session list", + Peer: baseMsg.Peer, + }) + if !strings.Contains(listResp, "1. [*] "+scopeKey+"#3") { + t.Fatalf("/session list response missing active session #3, got:\n%s", listResp) + } + if !strings.Contains(listResp, "3. [ ] "+scopeKey) { + t.Fatalf("/session list response missing base session ordinal, got:\n%s", listResp) + } + + resumeResp := helper.executeAndGetResponse(t, context.Background(), bus.InboundMessage{ + Channel: baseMsg.Channel, + SenderID: baseMsg.SenderID, + ChatID: baseMsg.ChatID, + Content: "/session resume 3", + Peer: baseMsg.Peer, + }) + if !strings.Contains(resumeResp, scopeKey) { + t.Fatalf("/session resume response missing target session key, got: %q", resumeResp) + } + + active, err := defaultAgent.Sessions.ResolveActive(scopeKey) + if err != nil { + t.Fatalf("ResolveActive failed: %v", err) + } + if active != scopeKey { + t.Fatalf("active session = %q, want %q after resume", active, scopeKey) + } +} + // TestToolResult_SilentToolDoesNotSendUserMessage verifies silent tools don't trigger outbound func TestToolResult_SilentToolDoesNotSendUserMessage(t *testing.T) { tmpDir, err := os.MkdirTemp("", "agent-test-*") diff --git a/pkg/commands/builtin.go b/pkg/commands/builtin.go index e80e63821..c1db0fd55 100644 --- a/pkg/commands/builtin.go +++ b/pkg/commands/builtin.go @@ -28,6 +28,17 @@ func BuiltinDefinitions(cfg *config.Config) []Definition { return req.Reply(FormatHelpMessage(defs)) }, }, + { + Name: "new", + Aliases: []string{"reset"}, + Description: "Start a new chat session", + Usage: "/new", + }, + { + Name: "session", + Description: "Manage chat sessions", + Usage: "/session [list|resume ]", + }, { Name: "show", Description: "Show current configuration", diff --git a/pkg/commands/builtin_test.go b/pkg/commands/builtin_test.go index adbbd835b..5e04008db 100644 --- a/pkg/commands/builtin_test.go +++ b/pkg/commands/builtin_test.go @@ -37,11 +37,28 @@ func TestBuiltinHelpHandler_ReturnsFormattedMessage(t *testing.T) { if err != nil { t.Fatalf("/help handler error: %v", err) } - if !strings.Contains(reply, "/show [model|channel] - Show current configuration") { - t.Fatalf("/help reply missing /show usage, got %q", reply) + if !strings.Contains(reply, "/new - Start a new chat session") { + t.Fatalf("/help reply missing /new usage, got %q", reply) } - if !strings.Contains(reply, "/list [models|channels] - List available options") { - t.Fatalf("/help reply missing /list usage, got %q", reply) + if !strings.Contains(reply, "/session [list|resume ] - Manage chat sessions") { + t.Fatalf("/help reply missing /session usage, got %q", reply) + } +} + +func TestBuiltinDefinitions_SessionCommandsRemainPassthroughWithoutRuntime(t *testing.T) { + defs := BuiltinDefinitions(nil) + + newDef := findDefinitionByName(t, defs, "new") + if !contains(newDef.Aliases, "reset") { + t.Fatalf("/new aliases=%v, want alias reset", newDef.Aliases) + } + if newDef.Handler != nil { + t.Fatalf("/new should remain passthrough without runtime handler") + } + + sessionDef := findDefinitionByName(t, defs, "session") + if sessionDef.Handler != nil { + t.Fatalf("/session should remain passthrough without runtime handler") } } diff --git a/pkg/commands/dispatcher.go b/pkg/commands/dispatcher.go index 177419415..ca617ded6 100644 --- a/pkg/commands/dispatcher.go +++ b/pkg/commands/dispatcher.go @@ -58,7 +58,9 @@ func (d *Dispatcher) Dispatch(ctx context.Context, req Request) Result { continue } if def.Handler == nil { - return Result{Matched: true, Handled: false, Command: def.Name} + // Definition-only command (for menu registration / discovery). + // Let the inbound message continue to the agent loop. + return Result{Matched: false, Handled: false, Command: def.Name} } err := def.Handler(ctx, req) return Result{Matched: true, Handled: true, Command: def.Name, Err: err} diff --git a/pkg/commands/dispatcher_test.go b/pkg/commands/dispatcher_test.go index d4877f5b7..3c1c0d7dd 100644 --- a/pkg/commands/dispatcher_test.go +++ b/pkg/commands/dispatcher_test.go @@ -101,3 +101,17 @@ func TestDispatcher_CommandMatchingIsCaseInsensitive(t *testing.T) { t.Fatalf("dispatch result = %+v, called=%v", res, called) } } + +func TestDispatcher_PassThroughDefinitionWithoutHandler(t *testing.T) { + d := NewDispatcher(NewRegistry([]Definition{ + {Name: "session"}, // menu-only / pass-through definition + })) + + res := d.Dispatch(context.Background(), Request{ + Channel: "telegram", + Text: "/session list", + }) + if res.Matched { + t.Fatalf("expected pass-through unmatched result, got %+v", res) + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 305ae67e3..03c3f0282 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -3,6 +3,7 @@ package config import ( "encoding/json" "fmt" + "io" "os" "sync/atomic" @@ -14,6 +15,15 @@ import ( // rrCounter is a global counter for round-robin load balancing across models. var rrCounter atomic.Uint64 +var warningWriter io.Writer = os.Stderr + +func warnf(format string, args ...any) { + if warningWriter == nil { + return + } + _, _ = fmt.Fprintf(warningWriter, "warning: "+format+"\n", args...) +} + // FlexibleStringSlice is a []string that also accepts JSON numbers, // so allow_from can contain both "123" and 123. type FlexibleStringSlice []string @@ -78,7 +88,7 @@ func (c Config) MarshalJSON() ([]byte, error) { } // Only include session if not empty - if c.Session.DMScope != "" || len(c.Session.IdentityLinks) > 0 { + if c.Session.DMScope != "" || len(c.Session.IdentityLinks) > 0 || c.Session.BacklogLimit > 0 { aux.Session = &c.Session } @@ -163,8 +173,23 @@ type AgentBinding struct { } type SessionConfig struct { + // DMScope controls how direct-message sessions are partitioned. + // Example values: "channel", "user", or "channel_user" depending on desired isolation. DMScope string `json:"dm_scope,omitempty"` + // IdentityLinks allows multiple platform IDs to share one logical scope. IdentityLinks map[string][]string `json:"identity_links,omitempty"` + // BacklogLimit is the max number of recent sessions kept per scope. + BacklogLimit int `json:"backlog_limit,omitempty"` +} + +const DefaultSessionBacklogLimit = 20 + +// EffectiveBacklogLimit guarantees a positive value for pruning logic. +func (s SessionConfig) EffectiveBacklogLimit() int { + if s.BacklogLimit < 1 { + return DefaultSessionBacklogLimit + } + return s.BacklogLimit } type AgentDefaults struct { @@ -658,6 +683,15 @@ func LoadConfig(path string) (*Config, error) { return nil, err } + if cfg.Session.BacklogLimit < 1 { + warnf( + "invalid session.backlog_limit=%d, fallback to default=%d", + cfg.Session.BacklogLimit, + DefaultSessionBacklogLimit, + ) + cfg.Session.BacklogLimit = DefaultSessionBacklogLimit + } + // Migrate legacy channel config fields to new unified structures cfg.migrateChannelConfigs() diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 6af7c209e..da129eead 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -1,6 +1,7 @@ package config import ( + "bytes" "encoding/json" "os" "path/filepath" @@ -467,3 +468,50 @@ func TestDefaultConfig_WorkspacePath_WithPicoclawHome(t *testing.T) { t.Errorf("Workspace path with PICOCLAW_HOME = %q, want %q", cfg.Agents.Defaults.Workspace, want) } } + +func TestDefaultConfig_SessionBacklogLimit(t *testing.T) { + cfg := DefaultConfig() + + if cfg.Session.BacklogLimit != DefaultSessionBacklogLimit { + t.Errorf( + "Session.BacklogLimit = %d, want %d", + cfg.Session.BacklogLimit, + DefaultSessionBacklogLimit, + ) + } +} + +func TestLoadConfig_InvalidBacklogLimitFallsBackToDefault(t *testing.T) { + tempDir := t.TempDir() + configPath := filepath.Join(tempDir, "config.json") + var warnings bytes.Buffer + oldWarningWriter := warningWriter + warningWriter = &warnings + t.Cleanup(func() { + warningWriter = oldWarningWriter + }) + + configJSON := `{ + "agents": {"defaults":{"workspace":"./workspace","model":"gpt4","max_tokens":8192,"max_tool_iterations":20}}, + "session": {"backlog_limit": 0}, + "model_list": [{"model_name":"gpt4","model":"openai/gpt-5.2","api_key":"x"}] +}` + if err := os.WriteFile(configPath, []byte(configJSON), 0o600); err != nil { + t.Fatalf("os.WriteFile() error: %v", err) + } + + cfg, err := LoadConfig(configPath) + if err != nil { + t.Fatalf("LoadConfig() error: %v", err) + } + if cfg.Session.BacklogLimit != DefaultSessionBacklogLimit { + t.Fatalf( + "Session.BacklogLimit = %d, want %d", + cfg.Session.BacklogLimit, + DefaultSessionBacklogLimit, + ) + } + if got := warnings.String(); !strings.Contains(got, "invalid session.backlog_limit=0") { + t.Fatalf("expected warning about invalid backlog_limit, got: %q", got) + } +} diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index 9fc09c5f1..50f3d02fc 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -37,7 +37,8 @@ func DefaultConfig() *Config { }, Bindings: []AgentBinding{}, Session: SessionConfig{ - DMScope: "per-channel-peer", + DMScope: "per-channel-peer", + BacklogLimit: DefaultSessionBacklogLimit, }, Channels: ChannelsConfig{ WhatsApp: WhatsAppConfig{ diff --git a/pkg/session/manager.go b/pkg/session/manager.go index 08f0b0ad2..ab22a8bce 100644 --- a/pkg/session/manager.go +++ b/pkg/session/manager.go @@ -2,8 +2,12 @@ package session import ( "encoding/json" + "errors" + "fmt" + "io" "os" "path/filepath" + "strconv" "strings" "sync" "time" @@ -19,26 +23,338 @@ type Session struct { Updated time.Time `json:"updated"` } -type SessionManager struct { - sessions map[string]*Session - mu sync.RWMutex - storage string +const sessionIndexFilename = "index.json" + +var ( + removeFile = os.Remove + warningWriter io.Writer = os.Stderr +) + +type scopeIndex struct { + ActiveSessionKey string `json:"active_session_key"` + OrderedSessions []string `json:"ordered_sessions"` + UpdatedAt time.Time `json:"updated_at"` } +type sessionIndex struct { + Version int `json:"version"` + Scopes map[string]*scopeIndex `json:"scopes"` + PendingDeletes []string `json:"pending_deletes,omitempty"` +} + +type SessionMeta struct { + Ordinal int `json:"ordinal"` + SessionKey string `json:"session_key"` + UpdatedAt time.Time `json:"updated_at"` + MessageCnt int `json:"message_cnt"` + Active bool `json:"active"` +} + +type SessionManager struct { + sessions map[string]*Session + mu sync.RWMutex + storage string + index sessionIndex + indexPath string +} + +// NewSessionManager initializes a scope-aware session manager. +// +// Phase 2 design summary: +// - keep one ordered session list per scope (dm/group/agent route key); +// - persist both session payloads and an index file for deterministic resume/list; +// - self-heal malformed index entries during load so command handlers stay robust. func NewSessionManager(storage string) *SessionManager { sm := &SessionManager{ sessions: make(map[string]*Session), storage: storage, + index: sessionIndex{ + Version: 1, + Scopes: make(map[string]*scopeIndex), + }, } if storage != "" { os.MkdirAll(storage, 0o755) + sm.indexPath = filepath.Join(storage, sessionIndexFilename) sm.loadSessions() + sm.loadIndex() } return sm } +func (sm *SessionManager) ResolveActive(scopeKey string) (string, error) { + sm.mu.Lock() + defer sm.mu.Unlock() + + now := time.Now() + scope, changed := sm.ensureScopeLocked(scopeKey, now) + if changed { + if err := sm.saveIndexLocked(); err != nil { + return "", err + } + } + return scope.ActiveSessionKey, nil +} + +// StartNew creates a new session within one scope, persists the session file first, +// then updates the scope index. If index persistence fails, it rolls back both memory +// and file-side effects to avoid half-written session state. +func (sm *SessionManager) StartNew(scopeKey string) (string, error) { + sm.mu.Lock() + defer sm.mu.Unlock() + + now := time.Now() + prevScopeEntryExists := false + prevScopeEntryWasNil := false + var prevScopeSnapshot *scopeIndex + if sm.index.Scopes != nil { + if existingScope, ok := sm.index.Scopes[scopeKey]; ok { + prevScopeEntryExists = true + if existingScope == nil { + prevScopeEntryWasNil = true + } else { + prevScopeSnapshot = cloneScopeIndex(existingScope) + } + } + } + + orderedForOrdinal := []string{scopeKey} + if prevScopeSnapshot != nil && len(prevScopeSnapshot.OrderedSessions) > 0 { + orderedForOrdinal = prevScopeSnapshot.OrderedSessions + } + + newOrdinal := 2 + for _, existing := range orderedForOrdinal { + ordinal, ok := sessionOrdinal(scopeKey, existing) + if !ok { + continue + } + if ordinal >= newOrdinal { + newOrdinal = ordinal + 1 + } + } + + newSessionKey := scopeKey + "#" + strconv.Itoa(newOrdinal) + + created := false + if _, ok := sm.sessions[newSessionKey]; !ok { + sm.sessions[newSessionKey] = &Session{ + Key: newSessionKey, + Messages: []providers.Message{}, + Created: now, + Updated: now, + } + created = true + } + if err := sm.saveSessionLocked(newSessionKey); err != nil { + if created { + delete(sm.sessions, newSessionKey) + } + return "", err + } + + scope, _ := sm.ensureScopeLocked(scopeKey, now) + scope.ActiveSessionKey = newSessionKey + scope.OrderedSessions = prependSessionUnique(scope.OrderedSessions, newSessionKey) + scope.UpdatedAt = now + if err := sm.saveIndexLocked(); err != nil { + if prevScopeEntryExists { + if sm.index.Scopes == nil { + sm.index.Scopes = make(map[string]*scopeIndex) + } + if prevScopeEntryWasNil { + sm.index.Scopes[scopeKey] = nil + } else { + sm.index.Scopes[scopeKey] = cloneScopeIndex(prevScopeSnapshot) + } + } else if sm.index.Scopes != nil { + delete(sm.index.Scopes, scopeKey) + } + if created { + delete(sm.sessions, newSessionKey) + _ = sm.deleteSessionFile(newSessionKey) + } + return "", err + } + return newSessionKey, nil +} + +// List returns stable, newest-first metadata for one scope. +func (sm *SessionManager) List(scopeKey string) ([]SessionMeta, error) { + sm.mu.Lock() + defer sm.mu.Unlock() + + now := time.Now() + scope, changed := sm.ensureScopeLocked(scopeKey, now) + if changed { + if err := sm.saveIndexLocked(); err != nil { + return nil, err + } + } + + list := make([]SessionMeta, 0, len(scope.OrderedSessions)) + for i, key := range scope.OrderedSessions { + meta := SessionMeta{ + Ordinal: i + 1, + SessionKey: key, + Active: key == scope.ActiveSessionKey, + } + + if session, ok := sm.sessions[key]; ok { + meta.UpdatedAt = session.Updated + meta.MessageCnt = len(session.Messages) + } + list = append(list, meta) + } + return list, nil +} + +// Resume switches active session by 1-based position in the scope-specific order. +func (sm *SessionManager) Resume(scopeKey string, index int) (string, error) { + if index < 1 { + return "", fmt.Errorf("session index must be >= 1") + } + + sm.mu.Lock() + defer sm.mu.Unlock() + + now := time.Now() + scope, changed := sm.ensureScopeLocked(scopeKey, now) + if changed { + if err := sm.saveIndexLocked(); err != nil { + return "", err + } + } + + if index > len(scope.OrderedSessions) { + return "", fmt.Errorf("session index %d out of range", index) + } + + scope.ActiveSessionKey = scope.OrderedSessions[index-1] + scope.UpdatedAt = now + if err := sm.saveIndexLocked(); err != nil { + return "", err + } + return scope.ActiveSessionKey, nil +} + +func (sm *SessionManager) DeleteSession(sessionKey string) error { + sm.mu.Lock() + changed := false + delete(sm.sessions, sessionKey) + + now := time.Now() + for scopeKey, scope := range sm.index.Scopes { + if scope == nil { + delete(sm.index.Scopes, scopeKey) + changed = true + continue + } + + filtered := scope.OrderedSessions[:0] + removed := false + for _, key := range scope.OrderedSessions { + if key == sessionKey { + removed = true + changed = true + continue + } + filtered = append(filtered, key) + } + scope.OrderedSessions = filtered + if !removed { + continue + } + + if scope.ActiveSessionKey == sessionKey { + if len(scope.OrderedSessions) > 0 { + scope.ActiveSessionKey = scope.OrderedSessions[0] + } else { + scope.ActiveSessionKey = "" + } + } + + if len(scope.OrderedSessions) == 0 { + delete(sm.index.Scopes, scopeKey) + continue + } + scope.UpdatedAt = now + } + + if changed { + if err := sm.saveIndexLocked(); err != nil { + sm.mu.Unlock() + return err + } + } + sm.mu.Unlock() + + if err := sm.deleteSessionFile(sessionKey); err != nil { + sm.warnf("failed to delete session file for %q, deferred retry on startup: %v", sessionKey, err) + sm.mu.Lock() + pendingChanged := sm.addPendingDeleteLocked(sessionKey) + if pendingChanged { + if err := sm.saveIndexLocked(); err != nil { + sm.mu.Unlock() + sm.warnf("failed to persist deferred delete for %q: %v", sessionKey, err) + return nil + } + } + sm.mu.Unlock() + return nil + } + + sm.mu.Lock() + pendingChanged := sm.removePendingDeleteLocked(sessionKey) + if pendingChanged { + if err := sm.saveIndexLocked(); err != nil { + sm.mu.Unlock() + sm.warnf("failed to persist cleanup of deferred delete for %q: %v", sessionKey, err) + return nil + } + } + sm.mu.Unlock() + + return nil +} + +// Prune keeps the newest `limit` sessions in one scope and removes older ones. +// Deletion uses the same safe path as DeleteSession so index/file consistency is preserved. +func (sm *SessionManager) Prune(scopeKey string, limit int) ([]string, error) { + if limit < 1 { + return nil, fmt.Errorf("limit must be >= 1") + } + + sm.mu.Lock() + now := time.Now() + scope, changed := sm.ensureScopeLocked(scopeKey, now) + if changed { + if err := sm.saveIndexLocked(); err != nil { + sm.mu.Unlock() + return nil, err + } + } + + if len(scope.OrderedSessions) <= limit { + sm.mu.Unlock() + return []string{}, nil + } + + candidates := append([]string(nil), scope.OrderedSessions[limit:]...) + sm.mu.Unlock() + + pruned := make([]string, 0, len(candidates)) + for _, sessionKey := range candidates { + if err := sm.DeleteSession(sessionKey); err != nil { + return pruned, err + } + pruned = append(pruned, sessionKey) + } + return pruned, nil +} + func (sm *SessionManager) GetOrCreate(key string) *Session { sm.mu.Lock() defer sm.mu.Unlock() @@ -159,16 +475,6 @@ func (sm *SessionManager) Save(key string) error { return nil } - filename := sanitizeFilename(key) - - // filepath.IsLocal rejects empty names, "..", absolute paths, and - // OS-reserved device names (NUL, COM1 … on Windows). - // The extra checks reject "." and any directory separators so that - // the session file is always written directly inside sm.storage. - if filename == "." || !filepath.IsLocal(filename) || strings.ContainsAny(filename, `/\`) { - return os.ErrInvalid - } - // Snapshot under read lock, then perform slow file I/O after unlock. sm.mu.RLock() stored, ok := sm.sessions[key] @@ -177,6 +483,263 @@ func (sm *SessionManager) Save(key string) error { return nil } + snapshot := cloneSession(stored) + sm.mu.RUnlock() + + return sm.writeSessionSnapshot(snapshot) +} + +func (sm *SessionManager) loadSessions() error { + files, err := os.ReadDir(sm.storage) + if err != nil { + return err + } + + for _, file := range files { + if file.IsDir() { + continue + } + + if filepath.Ext(file.Name()) != ".json" { + continue + } + if file.Name() == sessionIndexFilename { + continue + } + + sessionPath := filepath.Join(sm.storage, file.Name()) + data, err := os.ReadFile(sessionPath) + if err != nil { + continue + } + + var session Session + if err := json.Unmarshal(data, &session); err != nil { + continue + } + if session.Key == "" { + continue + } + + sm.sessions[session.Key] = &session + } + + return nil +} + +func (sm *SessionManager) loadIndex() error { + if sm.storage == "" { + return nil + } + + data, err := os.ReadFile(sm.indexPath) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return err + } + + var loaded sessionIndex + if err := json.Unmarshal(data, &loaded); err != nil { + return err + } + if loaded.Version == 0 { + loaded.Version = 1 + } + if loaded.Scopes == nil { + loaded.Scopes = make(map[string]*scopeIndex) + } + + changed := false + seenPending := make(map[string]struct{}, len(loaded.PendingDeletes)) + retryPending := make([]string, 0, len(loaded.PendingDeletes)) + for _, sessionKey := range loaded.PendingDeletes { + if sessionKey == "" { + changed = true + continue + } + if _, dup := seenPending[sessionKey]; dup { + changed = true + continue + } + seenPending[sessionKey] = struct{}{} + + // Deferred-delete sessions should not be visible even if stale files remain. + delete(sm.sessions, sessionKey) + + if err := sm.deleteSessionFile(sessionKey); err != nil { + // Invalid paths are unrecoverable; drop them from retry queue. + if errors.Is(err, os.ErrInvalid) { + changed = true + sm.warnf("dropping invalid deferred delete key %q: %v", sessionKey, err) + continue + } + sm.warnf("retry deferred session delete failed for %q: %v", sessionKey, err) + retryPending = append(retryPending, sessionKey) + continue + } + changed = true + } + if len(retryPending) != len(loaded.PendingDeletes) { + changed = true + } + loaded.PendingDeletes = retryPending + + for scopeKey, scope := range loaded.Scopes { + if scope == nil { + delete(loaded.Scopes, scopeKey) + changed = true + continue + } + + filtered := make([]string, 0, len(scope.OrderedSessions)) + seen := make(map[string]struct{}, len(scope.OrderedSessions)) + for _, sessionKey := range scope.OrderedSessions { + if sessionKey == "" { + changed = true + continue + } + if _, exists := sm.sessions[sessionKey]; !exists { + changed = true + continue + } + if _, dup := seen[sessionKey]; dup { + changed = true + continue + } + seen[sessionKey] = struct{}{} + filtered = append(filtered, sessionKey) + } + + if len(filtered) == 0 { + delete(loaded.Scopes, scopeKey) + changed = true + continue + } + + if len(filtered) != len(scope.OrderedSessions) { + changed = true + } + scope.OrderedSessions = filtered + if _, ok := seen[scope.ActiveSessionKey]; !ok { + scope.ActiveSessionKey = scope.OrderedSessions[0] + changed = true + } + } + + sm.index = loaded + if changed { + return sm.saveIndexLocked() + } + return nil +} + +func (sm *SessionManager) saveIndexLocked() error { + if sm.storage == "" { + return nil + } + if sm.index.Scopes == nil { + sm.index.Scopes = make(map[string]*scopeIndex) + } + sm.index.Version = 1 + + data, err := json.MarshalIndent(sm.index, "", " ") + if err != nil { + return err + } + + tmpFile, err := os.CreateTemp(sm.storage, "index-*.tmp") + if err != nil { + return err + } + + tmpPath := tmpFile.Name() + cleanup := true + defer func() { + if cleanup { + _ = os.Remove(tmpPath) + } + }() + + if _, err := tmpFile.Write(data); err != nil { + _ = tmpFile.Close() + return err + } + if err := tmpFile.Chmod(0o644); err != nil { + _ = tmpFile.Close() + return err + } + if err := tmpFile.Sync(); err != nil { + _ = tmpFile.Close() + return err + } + if err := tmpFile.Close(); err != nil { + return err + } + if err := os.Rename(tmpPath, sm.indexPath); err != nil { + return err + } + cleanup = false + return nil +} + +func (sm *SessionManager) ensureScopeLocked(scopeKey string, now time.Time) (*scopeIndex, bool) { + if sm.index.Scopes == nil { + sm.index.Scopes = make(map[string]*scopeIndex) + } + + scope, ok := sm.index.Scopes[scopeKey] + if !ok || scope == nil { + scope = &scopeIndex{ + ActiveSessionKey: scopeKey, + OrderedSessions: []string{scopeKey}, + UpdatedAt: now, + } + sm.index.Scopes[scopeKey] = scope + return scope, true + } + + changed := false + if len(scope.OrderedSessions) == 0 { + scope.OrderedSessions = []string{scopeKey} + changed = true + } + if scope.ActiveSessionKey == "" { + scope.ActiveSessionKey = scope.OrderedSessions[0] + changed = true + } + if changed { + scope.UpdatedAt = now + } + return scope, changed +} + +func prependSessionUnique(ordered []string, sessionKey string) []string { + next := make([]string, 0, len(ordered)+1) + next = append(next, sessionKey) + for _, existing := range ordered { + if existing == sessionKey { + continue + } + next = append(next, existing) + } + return next +} + +func cloneScopeIndex(scope *scopeIndex) *scopeIndex { + if scope == nil { + return nil + } + cloned := &scopeIndex{ + ActiveSessionKey: scope.ActiveSessionKey, + UpdatedAt: scope.UpdatedAt, + } + cloned.OrderedSessions = append([]string(nil), scope.OrderedSessions...) + return cloned +} + +func cloneSession(stored *Session) Session { snapshot := Session{ Key: stored.Key, Summary: stored.Summary, @@ -189,7 +752,73 @@ func (sm *SessionManager) Save(key string) error { } else { snapshot.Messages = []providers.Message{} } - sm.mu.RUnlock() + return snapshot +} + +func (sm *SessionManager) warnf(format string, args ...any) { + if warningWriter == nil { + return + } + _, _ = fmt.Fprintf(warningWriter, "warning: "+format+"\n", args...) +} + +func (sm *SessionManager) addPendingDeleteLocked(sessionKey string) bool { + if sessionKey == "" { + return false + } + for _, existing := range sm.index.PendingDeletes { + if existing == sessionKey { + return false + } + } + sm.index.PendingDeletes = append(sm.index.PendingDeletes, sessionKey) + return true +} + +func (sm *SessionManager) removePendingDeleteLocked(sessionKey string) bool { + if len(sm.index.PendingDeletes) == 0 { + return false + } + filtered := sm.index.PendingDeletes[:0] + removed := false + for _, existing := range sm.index.PendingDeletes { + if existing == sessionKey { + removed = true + continue + } + filtered = append(filtered, existing) + } + sm.index.PendingDeletes = filtered + return removed +} + +func (sm *SessionManager) saveSessionLocked(key string) error { + if sm.storage == "" { + return nil + } + + stored, ok := sm.sessions[key] + if !ok { + return fmt.Errorf("session %q not found", key) + } + + return sm.writeSessionSnapshot(cloneSession(stored)) +} + +func (sm *SessionManager) writeSessionSnapshot(snapshot Session) error { + if sm.storage == "" { + return nil + } + + filename := sanitizeFilename(snapshot.Key) + + // filepath.IsLocal rejects empty names, "..", absolute paths, and + // OS-reserved device names (NUL, COM1 … on Windows). + // The extra checks reject "." and any directory separators so that + // the session file is always written directly inside sm.storage. + if filename == "." || !filepath.IsLocal(filename) || strings.ContainsAny(filename, `/\`) { + return os.ErrInvalid + } data, err := json.MarshalIndent(snapshot, "", " ") if err != nil { @@ -233,36 +862,42 @@ func (sm *SessionManager) Save(key string) error { return nil } -func (sm *SessionManager) loadSessions() error { - files, err := os.ReadDir(sm.storage) - if err != nil { +func (sm *SessionManager) deleteSessionFile(sessionKey string) error { + if sm.storage == "" { + return nil + } + + filename := sanitizeFilename(sessionKey) + if filename == "." || !filepath.IsLocal(filename) || strings.ContainsAny(filename, `/\`) { + return os.ErrInvalid + } + + sessionPath := filepath.Join(sm.storage, filename+".json") + if err := removeFile(sessionPath); err != nil { + if os.IsNotExist(err) { + return nil + } return err } + return nil +} - for _, file := range files { - if file.IsDir() { - continue - } - - if filepath.Ext(file.Name()) != ".json" { - continue - } - - sessionPath := filepath.Join(sm.storage, file.Name()) - data, err := os.ReadFile(sessionPath) - if err != nil { - continue - } - - var session Session - if err := json.Unmarshal(data, &session); err != nil { - continue - } - - sm.sessions[session.Key] = &session +func sessionOrdinal(scopeKey, sessionKey string) (int, bool) { + if sessionKey == scopeKey { + return 1, true } - return nil + prefix := scopeKey + "#" + if !strings.HasPrefix(sessionKey, prefix) { + return 0, false + } + + n, err := strconv.Atoi(strings.TrimPrefix(sessionKey, prefix)) + if err != nil || n < 2 { + return 0, false + } + + return n, true } // SetHistory updates the messages of a session. diff --git a/pkg/session/manager_test.go b/pkg/session/manager_test.go index 5ef5f4349..60af0d6cc 100644 --- a/pkg/session/manager_test.go +++ b/pkg/session/manager_test.go @@ -1,8 +1,12 @@ package session import ( + "bytes" + "encoding/json" + "errors" "os" "path/filepath" + "strings" "testing" ) @@ -72,3 +76,380 @@ func TestSave_RejectsPathTraversal(t *testing.T) { } } } + +func TestSessionIndex_BootstrapScopeAndPersist(t *testing.T) { + tmp := t.TempDir() + sm := NewSessionManager(tmp) + scope := "agent:main:telegram:direct:user1" + + active, err := sm.ResolveActive(scope) + if err != nil { + t.Fatal(err) + } + if active != scope { + t.Fatalf("active=%q, want %q", active, scope) + } + + indexPath := filepath.Join(tmp, "index.json") + raw, err := os.ReadFile(indexPath) + if err != nil { + t.Fatalf("read index: %v", err) + } + + var decoded map[string]any + if err := json.Unmarshal(raw, &decoded); err != nil { + t.Fatalf("unmarshal index: %v", err) + } + + sm2 := NewSessionManager(tmp) + active2, err := sm2.ResolveActive(scope) + if err != nil { + t.Fatal(err) + } + if active2 != active { + t.Fatalf("active2=%q, want %q", active2, active) + } +} + +func TestStartNew_CreatesMonotonicSessionKeys(t *testing.T) { + sm := NewSessionManager(t.TempDir()) + scope := "agent:main:telegram:direct:user1" + + if _, err := sm.ResolveActive(scope); err != nil { + t.Fatal(err) + } + + s2, err := sm.StartNew(scope) + if err != nil { + t.Fatal(err) + } + if s2 != scope+"#2" { + t.Fatalf("s2=%q, want %q", s2, scope+"#2") + } + + s3, err := sm.StartNew(scope) + if err != nil { + t.Fatal(err) + } + if s3 != scope+"#3" { + t.Fatalf("s3=%q, want %q", s3, scope+"#3") + } +} + +func TestStartNew_PersistsSessionFileWithoutManualSave(t *testing.T) { + dir := t.TempDir() + sm := NewSessionManager(dir) + scope := "agent:main:telegram:direct:user1" + + if _, err := sm.ResolveActive(scope); err != nil { + t.Fatal(err) + } + + s2, err := sm.StartNew(scope) + if err != nil { + t.Fatal(err) + } + + sessionPath := filepath.Join(dir, sanitizeFilename(s2)+".json") + if _, err := os.Stat(sessionPath); err != nil { + t.Fatalf("expected %s to exist: %v", sessionPath, err) + } + + indexPath := filepath.Join(dir, sessionIndexFilename) + raw, err := os.ReadFile(indexPath) + if err != nil { + t.Fatalf("read index: %v", err) + } + + var idx sessionIndex + if err := json.Unmarshal(raw, &idx); err != nil { + t.Fatalf("unmarshal index: %v", err) + } + + scoped := idx.Scopes[scope] + if scoped == nil { + t.Fatalf("expected scope %q in index", scope) + } + if scoped.ActiveSessionKey != s2 { + t.Fatalf("active=%q, want %q", scoped.ActiveSessionKey, s2) + } + if len(scoped.OrderedSessions) == 0 || scoped.OrderedSessions[0] != s2 { + t.Fatalf("ordered_sessions=%v, want first=%q", scoped.OrderedSessions, s2) + } +} + +func TestStartNew_DoesNotMutateIndexWhenSessionPersistFails(t *testing.T) { + dir := t.TempDir() + sm := NewSessionManager(dir) + scope := "../invalid/scope" + + _, err := sm.StartNew(scope) + if err == nil { + t.Fatalf("expected StartNew to fail for invalid persisted session key") + } + + if _, exists := sm.index.Scopes[scope]; exists { + t.Fatalf("scope %q should not be added to index on session persist failure", scope) + } + + if _, exists := sm.sessions[scope+"#2"]; exists { + t.Fatalf("session %q should not remain in memory on session persist failure", scope+"#2") + } + + files, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("read dir %q: %v", dir, err) + } + if len(files) != 0 { + t.Fatalf("storage should stay untouched, found files: %v", files) + } +} + +func TestListAndResume_ByScopeOrdinal(t *testing.T) { + sm := NewSessionManager(t.TempDir()) + scope := "agent:main:telegram:direct:user1" + + if _, err := sm.ResolveActive(scope); err != nil { + t.Fatal(err) + } + if _, err := sm.StartNew(scope); err != nil { // #2 + t.Fatal(err) + } + if _, err := sm.StartNew(scope); err != nil { // #3 (active) + t.Fatal(err) + } + + list, err := sm.List(scope) + if err != nil { + t.Fatal(err) + } + if len(list) != 3 { + t.Fatalf("len(list)=%d, want 3", len(list)) + } + if list[0].Ordinal != 1 || list[0].SessionKey != scope+"#3" || !list[0].Active { + t.Fatalf("list[0]=%+v", list[0]) + } + if list[2].Ordinal != 3 || list[2].SessionKey != scope { + t.Fatalf("list[2]=%+v", list[2]) + } + + resumed, err := sm.Resume(scope, 3) + if err != nil { + t.Fatal(err) + } + if resumed != scope { + t.Fatalf("resumed=%q, want %q", resumed, scope) + } + + listAfter, err := sm.List(scope) + if err != nil { + t.Fatal(err) + } + if !listAfter[2].Active { + t.Fatalf("listAfter[2] should be active: %+v", listAfter[2]) + } +} + +func TestPrune_RemovesOldestFromMemoryAndDisk(t *testing.T) { + dir := t.TempDir() + sm := NewSessionManager(dir) + scope := "agent:main:telegram:direct:user1" + + if _, err := sm.ResolveActive(scope); err != nil { + t.Fatal(err) + } + if _, err := sm.StartNew(scope); err != nil { // #2 + t.Fatal(err) + } + if _, err := sm.StartNew(scope); err != nil { // #3 (active) + t.Fatal(err) + } + + keys := []string{scope, scope + "#2", scope + "#3"} + for _, key := range keys { + sm.AddMessage(key, "user", "hello") + if err := sm.Save(key); err != nil { + t.Fatalf("save %q: %v", key, err) + } + } + + pruned, err := sm.Prune(scope, 2) + if err != nil { + t.Fatal(err) + } + if len(pruned) != 1 || pruned[0] != scope { + t.Fatalf("pruned=%v, want [%s]", pruned, scope) + } + + if got := len(sm.GetHistory(scope)); got != 0 { + t.Fatalf("expected deleted session history len=0, got %d", got) + } + + removedFile := filepath.Join(dir, sanitizeFilename(scope)+".json") + if _, err := os.Stat(removedFile); !os.IsNotExist(err) { + t.Fatalf("expected %s deleted, stat err=%v", removedFile, err) + } + + list, err := sm.List(scope) + if err != nil { + t.Fatal(err) + } + if len(list) != 2 { + t.Fatalf("len(list)=%d, want 2", len(list)) + } + if list[0].SessionKey != scope+"#3" || list[1].SessionKey != scope+"#2" { + t.Fatalf("list order after prune = %+v", list) + } +} + +func TestLoadIndex_SelfHealsStaleReferences(t *testing.T) { + dir := t.TempDir() + scopeA := "agent:main:telegram:direct:user1" + scopeB := "agent:main:telegram:direct:user2" + validNewest := scopeA + "#3" + validOlder := scopeA + "#2" + + seed := NewSessionManager(dir) + seed.AddMessage(validNewest, "user", "hello") + seed.AddMessage(validOlder, "user", "hello") + if err := seed.Save(validNewest); err != nil { + t.Fatalf("save %q: %v", validNewest, err) + } + if err := seed.Save(validOlder); err != nil { + t.Fatalf("save %q: %v", validOlder, err) + } + + stale := sessionIndex{ + Version: 1, + Scopes: map[string]*scopeIndex{ + scopeA: { + ActiveSessionKey: scopeA + "#999", + OrderedSessions: []string{ + validNewest, + validNewest, + scopeA + "#404", + validOlder, + }, + }, + scopeB: { + ActiveSessionKey: scopeB, + OrderedSessions: []string{scopeB}, + }, + }, + } + + raw, err := json.MarshalIndent(stale, "", " ") + if err != nil { + t.Fatalf("marshal stale index: %v", err) + } + if err := os.WriteFile(filepath.Join(dir, sessionIndexFilename), raw, 0o644); err != nil { + t.Fatalf("write stale index: %v", err) + } + + reloaded := NewSessionManager(dir) + + indexRaw, err := os.ReadFile(filepath.Join(dir, sessionIndexFilename)) + if err != nil { + t.Fatalf("read healed index: %v", err) + } + + var healed sessionIndex + if err := json.Unmarshal(indexRaw, &healed); err != nil { + t.Fatalf("unmarshal healed index: %v", err) + } + + scopeAHealed := healed.Scopes[scopeA] + if scopeAHealed == nil { + t.Fatalf("expected scope %q in healed index", scopeA) + } + if scopeAHealed.ActiveSessionKey != validNewest { + t.Fatalf("active=%q, want %q", scopeAHealed.ActiveSessionKey, validNewest) + } + if len(scopeAHealed.OrderedSessions) != 2 { + t.Fatalf("ordered_sessions=%v, want len=2", scopeAHealed.OrderedSessions) + } + if scopeAHealed.OrderedSessions[0] != validNewest || scopeAHealed.OrderedSessions[1] != validOlder { + t.Fatalf("ordered_sessions=%v, want [%s %s]", scopeAHealed.OrderedSessions, validNewest, validOlder) + } + if _, exists := healed.Scopes[scopeB]; exists { + t.Fatalf("expected stale scope %q removed", scopeB) + } + + list, err := reloaded.List(scopeA) + if err != nil { + t.Fatal(err) + } + if len(list) != 2 { + t.Fatalf("len(list)=%d, want 2", len(list)) + } + if !list[0].Active || list[0].SessionKey != validNewest { + t.Fatalf("list[0]=%+v, want active newest", list[0]) + } +} + +func TestDeleteSession_FileDeleteFailureIsDeferredAndRetriedOnStartup(t *testing.T) { + dir := t.TempDir() + sm := NewSessionManager(dir) + scope := "agent:main:telegram:direct:user1" + + if _, err := sm.ResolveActive(scope); err != nil { + t.Fatal(err) + } + sessionKey, err := sm.StartNew(scope) + if err != nil { + t.Fatal(err) + } + + oldRemoveFile := removeFile + oldWarningWriter := warningWriter + var warnings bytes.Buffer + warningWriter = &warnings + removeFile = func(path string) error { + if strings.HasSuffix(path, sanitizeFilename(sessionKey)+".json") { + return errors.New("permission denied") + } + return oldRemoveFile(path) + } + t.Cleanup(func() { + removeFile = oldRemoveFile + warningWriter = oldWarningWriter + }) + + if err := sm.DeleteSession(sessionKey); err != nil { + t.Fatalf("DeleteSession(%q) returned unexpected error: %v", sessionKey, err) + } + if got := warnings.String(); !strings.Contains(got, "deferred retry on startup") { + t.Fatalf("expected deferred-delete warning, got: %q", got) + } + + sessionPath := filepath.Join(dir, sanitizeFilename(sessionKey)+".json") + if _, err := os.Stat(sessionPath); err != nil { + t.Fatalf("expected deferred file %q to still exist, err=%v", sessionPath, err) + } + + list, err := sm.List(scope) + if err != nil { + t.Fatal(err) + } + for _, item := range list { + if item.SessionKey == sessionKey { + t.Fatalf("deleted session key %q should not remain in index list", sessionKey) + } + } + + if len(sm.index.PendingDeletes) != 1 || sm.index.PendingDeletes[0] != sessionKey { + t.Fatalf("pending_deletes=%v, want [%q]", sm.index.PendingDeletes, sessionKey) + } + + removeFile = oldRemoveFile + reloaded := NewSessionManager(dir) + if len(reloaded.index.PendingDeletes) != 0 { + t.Fatalf("pending_deletes should be drained on startup retry, got %v", reloaded.index.PendingDeletes) + } + if _, err := os.Stat(sessionPath); !os.IsNotExist(err) { + t.Fatalf("expected %q to be removed by startup retry, stat err=%v", sessionPath, err) + } + if _, exists := reloaded.sessions[sessionKey]; exists { + t.Fatalf("session %q should not be present in memory after startup retry", sessionKey) + } +}