From be2b9507ddca75597a43fe5eb5c6d405baa6aa01 Mon Sep 17 00:00:00 2001 From: mingmxren Date: Wed, 4 Mar 2026 01:38:12 +0800 Subject: [PATCH] feat(session): add session lifecycle management [Phase 2/2] Port scope-aware session management onto the centralized command registry (Phase 1). Key additions: - SessionOps interface on Runtime for session operations - ScopeKey on Request for routing-resolved scope identity - /new command: starts new session with automatic backlog pruning - /session list|resume: SubCommand pattern for session management - SessionManager: scope-aware index with persistence, self-healing, deferred delete retry, ResolveActive/StartNew/List/Resume/Prune - BacklogLimit config with EffectiveBacklogLimit() validation - Comprehensive test coverage for all session handlers and manager Co-Authored-By: Claude Opus 4.6 --- pkg/agent/loop.go | 19 +- pkg/agent/loop_test.go | 7 +- pkg/commands/builtin.go | 2 + pkg/commands/builtin_test.go | 7 +- pkg/commands/cmd_new.go | 48 ++ pkg/commands/cmd_session.go | 91 ++++ pkg/commands/request.go | 16 +- pkg/commands/runtime.go | 34 +- pkg/commands/session_handlers_test.go | 284 ++++++++++ pkg/config/config.go | 38 +- pkg/config/config_test.go | 48 ++ pkg/config/defaults.go | 3 +- pkg/session/manager.go | 746 ++++++++++++++++++++++++-- pkg/session/manager_test.go | 417 ++++++++++++++ 14 files changed, 1700 insertions(+), 60 deletions(-) create mode 100644 pkg/commands/cmd_new.go create mode 100644 pkg/commands/cmd_session.go create mode 100644 pkg/commands/session_handlers_test.go diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 966668227..a2975caff 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -568,7 +568,7 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) // Global commands (/help, /show, /switch) work even when routing fails; // context-dependent commands check their own Runtime fields and report // "unavailable" when the required capability is nil. - if response, handled := al.handleCommand(ctx, msg, agent); handled { + if response, handled := al.handleCommand(ctx, msg, route, agent); handled { return response, nil } @@ -583,9 +583,12 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) } } - // Resolve session key from route, while preserving explicit agent-scoped keys. + // Resolve active session from the routing scope. scopeKey := resolveScopeKey(route, msg.SessionKey) - sessionKey := scopeKey + 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{ @@ -1540,6 +1543,7 @@ func (al *AgentLoop) estimateTokens(messages []providers.Message) int { func (al *AgentLoop) handleCommand( ctx context.Context, msg bus.InboundMessage, + route routing.ResolvedRoute, agent *AgentInstance, ) (string, bool) { if !commands.HasCommandPrefix(msg.Content) { @@ -1550,7 +1554,7 @@ func (al *AgentLoop) handleCommand( return "", false } - rt := al.buildCommandsRuntime(agent) + rt := al.buildCommandsRuntime(route, agent) executor := commands.NewExecutor(al.cmdRegistry, rt) var commandReply string @@ -1558,6 +1562,7 @@ func (al *AgentLoop) handleCommand( Channel: msg.Channel, ChatID: msg.ChatID, SenderID: msg.SenderID, + ScopeKey: resolveScopeKey(route, msg.SessionKey), Text: msg.Content, Reply: func(text string) error { commandReply = text @@ -1579,7 +1584,10 @@ func (al *AgentLoop) handleCommand( } } -func (al *AgentLoop) buildCommandsRuntime(agent *AgentInstance) *commands.Runtime { +func (al *AgentLoop) buildCommandsRuntime( + route routing.ResolvedRoute, + agent *AgentInstance, +) *commands.Runtime { rt := &commands.Runtime{ Config: al.cfg, ListAgentIDs: al.registry.ListAgentIDs, @@ -1609,6 +1617,7 @@ func (al *AgentLoop) buildCommandsRuntime(agent *AgentInstance) *commands.Runtim agent.Model = value return oldModel, nil } + rt.SessionOps = agent.Sessions } return rt } diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 2e456fa60..e7dacc284 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -503,6 +503,7 @@ func TestProcessMessage_CommandOutcomes(t *testing.T) { t.Fatalf("LLM should be called exactly once after /foo passthrough, calls=%d", provider.calls) } + // /new is now a handled session command (starts a new session) newResp := helper.executeAndGetResponse(t, context.Background(), bus.InboundMessage{ Channel: baseMsg.Channel, SenderID: baseMsg.SenderID, @@ -510,11 +511,11 @@ func TestProcessMessage_CommandOutcomes(t *testing.T) { Content: "/new", Peer: baseMsg.Peer, }) - if newResp != "LLM reply" { + if !strings.Contains(newResp, "Started new session") { t.Fatalf("unexpected /new reply: %q", newResp) } - if provider.calls != 2 { - t.Fatalf("LLM should be called for passthrough /new command, calls=%d", provider.calls) + if provider.calls != 1 { + t.Fatalf("LLM should NOT be called for handled /new command, calls=%d", provider.calls) } } diff --git a/pkg/commands/builtin.go b/pkg/commands/builtin.go index a36dd3eba..8dac5e1b9 100644 --- a/pkg/commands/builtin.go +++ b/pkg/commands/builtin.go @@ -8,6 +8,8 @@ func BuiltinDefinitions() []Definition { return []Definition{ startCommand(), helpCommand(), + newCommand(), + sessionCommand(), showCommand(), listCommand(), switchCommand(), diff --git a/pkg/commands/builtin_test.go b/pkg/commands/builtin_test.go index 66a84825e..844b65123 100644 --- a/pkg/commands/builtin_test.go +++ b/pkg/commands/builtin_test.go @@ -35,13 +35,18 @@ func TestBuiltinHelpHandler_ReturnsFormattedMessage(t *testing.T) { if err != nil { t.Fatalf("/help handler error: %v", err) } - // Now uses auto-generated EffectiveUsage which includes agents if !strings.Contains(reply, "/show [model|channel|agents]") { t.Fatalf("/help reply missing /show usage, got %q", reply) } if !strings.Contains(reply, "/list [models|channels|agents]") { t.Fatalf("/help reply missing /list usage, got %q", reply) } + if !strings.Contains(reply, "/new") { + t.Fatalf("/help reply missing /new, got %q", reply) + } + if !strings.Contains(reply, "/session [list|resume") { + t.Fatalf("/help reply missing /session usage, got %q", reply) + } } func TestBuiltinShowChannel_PreservesUserVisibleBehavior(t *testing.T) { diff --git a/pkg/commands/cmd_new.go b/pkg/commands/cmd_new.go new file mode 100644 index 000000000..10afc3f52 --- /dev/null +++ b/pkg/commands/cmd_new.go @@ -0,0 +1,48 @@ +package commands + +import ( + "context" + "fmt" + "strings" + + "github.com/sipeed/picoclaw/pkg/config" +) + +func newCommand() Definition { + return Definition{ + Name: "new", + Aliases: []string{"reset"}, + Description: "Start a new chat session", + Usage: "/new", + Handler: func(_ context.Context, req Request, rt *Runtime) error { + if rt == nil || rt.SessionOps == nil || strings.TrimSpace(req.ScopeKey) == "" { + return req.Reply(unavailableMsg) + } + + newSessionKey, err := rt.SessionOps.StartNew(req.ScopeKey) + if err != nil { + return req.Reply(fmt.Sprintf("Failed to start new session: %v", err)) + } + + backlogLimit := config.DefaultSessionBacklogLimit + if rt.Config != nil { + backlogLimit = rt.Config.Session.EffectiveBacklogLimit() + } + + pruned, err := rt.SessionOps.Prune(req.ScopeKey, backlogLimit) + if err != nil { + return req.Reply(fmt.Sprintf( + "Started new session (%s), but pruning old sessions failed: %v", + newSessionKey, err, + )) + } + + if len(pruned) == 0 { + return req.Reply(fmt.Sprintf("Started new session: %s", newSessionKey)) + } + return req.Reply( + fmt.Sprintf("Started new session: %s (pruned %d old session(s))", newSessionKey, len(pruned)), + ) + }, + } +} diff --git a/pkg/commands/cmd_session.go b/pkg/commands/cmd_session.go new file mode 100644 index 000000000..087d9570e --- /dev/null +++ b/pkg/commands/cmd_session.go @@ -0,0 +1,91 @@ +package commands + +import ( + "context" + "fmt" + "strconv" + "strings" + + "github.com/sipeed/picoclaw/pkg/session" +) + +func sessionCommand() Definition { + return Definition{ + Name: "session", + Description: "Manage chat sessions", + SubCommands: []SubCommand{ + { + Name: "list", + Description: "List sessions for current chat", + Handler: sessionListHandler(), + }, + { + Name: "resume", + Description: "Resume a previous session", + ArgsUsage: "", + Handler: sessionResumeHandler(), + }, + }, + } +} + +func sessionListHandler() Handler { + return func(_ context.Context, req Request, rt *Runtime) error { + if rt == nil || rt.SessionOps == nil || strings.TrimSpace(req.ScopeKey) == "" { + return req.Reply(unavailableMsg) + } + + list, err := rt.SessionOps.List(req.ScopeKey) + if err != nil { + return req.Reply(fmt.Sprintf("Failed to list sessions: %v", err)) + } + if len(list) == 0 { + return req.Reply("No sessions found for current chat.") + } + return req.Reply(formatSessionList(list)) + } +} + +func sessionResumeHandler() Handler { + return func(_ context.Context, req Request, rt *Runtime) error { + if rt == nil || rt.SessionOps == nil || strings.TrimSpace(req.ScopeKey) == "" { + return req.Reply(unavailableMsg) + } + + // tokens: [/session, resume, ] + indexStr := nthToken(req.Text, 2) + if indexStr == "" { + return req.Reply("Usage: /session resume ") + } + index, err := strconv.Atoi(indexStr) + if err != nil || index < 1 { + return req.Reply("Usage: /session resume ") + } + + sessionKey, err := rt.SessionOps.Resume(req.ScopeKey, index) + if err != nil { + return req.Reply(fmt.Sprintf("Failed to resume session %d: %v", index, err)) + } + return req.Reply(fmt.Sprintf("Resumed session %d: %s", index, sessionKey)) + } +} + +func formatSessionList(list []session.SessionMeta) string { + 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 | updated: %s | messages: %d", + item.Ordinal, activeMarker, item.SessionKey, updated, item.MessageCnt, + )) + } + return strings.Join(lines, "\n") +} diff --git a/pkg/commands/request.go b/pkg/commands/request.go index 62ee600f2..4f1507568 100644 --- a/pkg/commands/request.go +++ b/pkg/commands/request.go @@ -7,12 +7,20 @@ import ( type Handler func(ctx context.Context, req Request, rt *Runtime) error +// Request describes an incoming command invocation — the "what / who / where." +// It carries identity and context derived from the inbound message and its +// routing resolution. Fields here answer "where did this come from?" and +// "what scope does it belong to?" +// +// Contrast with [Runtime], which carries capabilities and services ("what can +// I do?"). A handler reads context from Request and calls services on Runtime. type Request struct { - Channel string - ChatID string - SenderID string - Text string + Channel string // platform the message arrived on + ChatID string // conversation identifier + SenderID string // who sent the message + Text string // raw command text Reply func(text string) error + ScopeKey string // routing-resolved scope for session operations } const unavailableMsg = "Command unavailable in current context." diff --git a/pkg/commands/runtime.go b/pkg/commands/runtime.go index 227d495f4..d98d1ede2 100644 --- a/pkg/commands/runtime.go +++ b/pkg/commands/runtime.go @@ -1,10 +1,35 @@ package commands -import "github.com/sipeed/picoclaw/pkg/config" +import ( + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/session" +) -// Runtime provides runtime dependencies to command handlers. It is constructed -// per-request by the agent loop so that per-request state (like session scope) -// can coexist with long-lived callbacks (like GetModelInfo). +// SessionOps defines the session lifecycle operations that command handlers +// rely on. Implementations are expected to be scope-aware and deterministic +// for a given scopeKey. SessionManager satisfies this interface. +type SessionOps interface { + ResolveActive(scopeKey string) (string, error) + StartNew(scopeKey string) (string, error) + List(scopeKey string) ([]session.SessionMeta, error) + Resume(scopeKey string, index int) (string, error) + Prune(scopeKey string, limit int) ([]string, error) +} + +// Runtime provides capabilities and services to command handlers — the +// "what can I do?" side of the handler contract. +// +// It is constructed per-request by the agent loop so that late-bound +// dependencies (e.g. session manager for the resolved agent) are always +// current. Fields here are either read-only accessors or mutation callbacks; +// none describe the identity of the incoming message (that belongs in +// [Request]). +// +// Guideline for placing a new field: +// +// Request — identity / context: channel, chat, sender, scope key, text. +// Runtime — service / capability: config access, query funcs, mutation funcs, +// service interfaces (SessionOps, etc.). type Runtime struct { Config *config.Config GetModelInfo func() (name, provider string) @@ -13,4 +38,5 @@ type Runtime struct { GetEnabledChannels func() []string SwitchModel func(value string) (oldModel string, err error) SwitchChannel func(value string) error + SessionOps SessionOps // nil when session commands are unavailable } diff --git a/pkg/commands/session_handlers_test.go b/pkg/commands/session_handlers_test.go new file mode 100644 index 000000000..e2ccf00b2 --- /dev/null +++ b/pkg/commands/session_handlers_test.go @@ -0,0 +1,284 @@ +package commands + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/session" +) + +type fakeSessionOps struct { + startNewScopeKeys []string + startNewValue string + startNewErr error + + pruneScopeKeys []string + pruneLimits []int + pruneValue []string + pruneErr error + + listScopeKeys []string + listValue []session.SessionMeta + listErr error + + resumeScopeKeys []string + resumeIndices []int + resumeValue string + resumeErr error +} + +func (f *fakeSessionOps) ResolveActive(scopeKey string) (string, error) { return "", nil } + +func (f *fakeSessionOps) StartNew(scopeKey string) (string, error) { + f.startNewScopeKeys = append(f.startNewScopeKeys, scopeKey) + return f.startNewValue, f.startNewErr +} + +func (f *fakeSessionOps) List(scopeKey string) ([]session.SessionMeta, error) { + f.listScopeKeys = append(f.listScopeKeys, scopeKey) + return f.listValue, f.listErr +} + +func (f *fakeSessionOps) Resume(scopeKey string, index int) (string, error) { + f.resumeScopeKeys = append(f.resumeScopeKeys, scopeKey) + f.resumeIndices = append(f.resumeIndices, index) + return f.resumeValue, f.resumeErr +} + +func (f *fakeSessionOps) Prune(scopeKey string, limit int) ([]string, error) { + f.pruneScopeKeys = append(f.pruneScopeKeys, scopeKey) + f.pruneLimits = append(f.pruneLimits, limit) + return f.pruneValue, f.pruneErr +} + +func TestSessionHandlers_New_UsesRuntimeSessionOps(t *testing.T) { + ops := &fakeSessionOps{ + startNewValue: "scope#2", + pruneValue: []string{"scope#1"}, + } + rt := &Runtime{ + SessionOps: ops, + Config: &config.Config{Session: config.SessionConfig{BacklogLimit: 7}}, + } + ex := NewExecutor(NewRegistry(BuiltinDefinitions()), rt) + + var reply string + res := ex.Execute(context.Background(), Request{ + Channel: "whatsapp", + ScopeKey: "scope", + Text: "/new", + Reply: func(text string) error { reply = text; return nil }, + }) + + if res.Outcome != OutcomeHandled { + t.Fatalf("outcome=%v, want=%v", res.Outcome, OutcomeHandled) + } + if len(ops.startNewScopeKeys) != 1 || ops.startNewScopeKeys[0] != "scope" { + t.Fatalf("startNew calls=%v, want [scope]", ops.startNewScopeKeys) + } + if len(ops.pruneLimits) != 1 || ops.pruneLimits[0] != 7 { + t.Fatalf("prune limits=%v, want [7]", ops.pruneLimits) + } + if reply != "Started new session: scope#2 (pruned 1 old session(s))" { + t.Fatalf("reply=%q", reply) + } +} + +func TestSessionHandlers_SessionResume(t *testing.T) { + ops := &fakeSessionOps{resumeValue: "scope#3"} + rt := &Runtime{SessionOps: ops} + ex := NewExecutor(NewRegistry(BuiltinDefinitions()), rt) + + var reply string + res := ex.Execute(context.Background(), Request{ + ScopeKey: "scope", + Text: "/session resume 3", + Reply: func(text string) error { reply = text; return nil }, + }) + + if res.Outcome != OutcomeHandled { + t.Fatalf("outcome=%v, want=%v", res.Outcome, OutcomeHandled) + } + if len(ops.resumeIndices) != 1 || ops.resumeIndices[0] != 3 { + t.Fatalf("resume indices=%v, want [3]", ops.resumeIndices) + } + if reply != "Resumed session 3: scope#3" { + t.Fatalf("reply=%q", reply) + } +} + +func TestSessionHandlers_SessionList(t *testing.T) { + ops := &fakeSessionOps{ + listValue: []session.SessionMeta{ + { + Ordinal: 1, + SessionKey: "scope#3", + UpdatedAt: time.Date(2026, 3, 1, 9, 7, 0, 0, time.UTC), + MessageCnt: 4, + Active: true, + }, + }, + } + rt := &Runtime{SessionOps: ops} + ex := NewExecutor(NewRegistry(BuiltinDefinitions()), rt) + + var reply string + res := ex.Execute(context.Background(), Request{ + ScopeKey: "scope", + Text: "/session list", + Reply: func(text string) error { reply = text; return nil }, + }) + + if res.Outcome != OutcomeHandled { + t.Fatalf("outcome=%v, want=%v", res.Outcome, OutcomeHandled) + } + if reply != "Sessions for current chat:\n1. [*] scope#3 | updated: 2026-03-01 09:07 | messages: 4" { + t.Fatalf("reply=%q", reply) + } +} + +func TestSessionHandlers_NilRuntime_Unavailable(t *testing.T) { + ex := NewExecutor(NewRegistry(BuiltinDefinitions()), nil) + + for _, text := range []string{"/new", "/session list"} { + var reply string + res := ex.Execute(context.Background(), Request{ + ScopeKey: "scope", + Text: text, + Reply: func(text string) error { reply = text; return nil }, + }) + if res.Outcome != OutcomeHandled { + t.Fatalf("text=%q outcome=%v, want=%v", text, res.Outcome, OutcomeHandled) + } + if reply != unavailableMsg { + t.Fatalf("text=%q reply=%q, want unavailable", text, reply) + } + } +} + +func TestSessionHandlers_NilSessionOps_Unavailable(t *testing.T) { + rt := &Runtime{} // SessionOps is nil + ex := NewExecutor(NewRegistry(BuiltinDefinitions()), rt) + + for _, text := range []string{"/new", "/session list"} { + var reply string + res := ex.Execute(context.Background(), Request{ + ScopeKey: "scope", + Text: text, + Reply: func(text string) error { reply = text; return nil }, + }) + if res.Outcome != OutcomeHandled { + t.Fatalf("text=%q outcome=%v, want=%v", text, res.Outcome, OutcomeHandled) + } + if reply != unavailableMsg { + t.Fatalf("text=%q reply=%q, want unavailable", text, reply) + } + } +} + +func TestSessionHandlers_EmptyScopeKey_Unavailable(t *testing.T) { + ops := &fakeSessionOps{} + rt := &Runtime{SessionOps: ops} + ex := NewExecutor(NewRegistry(BuiltinDefinitions()), rt) + + for _, text := range []string{"/new", "/session list"} { + var reply string + res := ex.Execute(context.Background(), Request{ + ScopeKey: " ", // whitespace-only + Text: text, + Reply: func(text string) error { reply = text; return nil }, + }) + if res.Outcome != OutcomeHandled { + t.Fatalf("text=%q outcome=%v, want=%v", text, res.Outcome, OutcomeHandled) + } + if reply != unavailableMsg { + t.Fatalf("text=%q reply=%q, want unavailable", text, reply) + } + } +} + +func TestSessionHandlers_ErrorReplies(t *testing.T) { + tests := []struct { + name string + text string + scopeKey string + ops *fakeSessionOps + wantReply string + }{ + { + name: "start new error", + text: "/new", + scopeKey: "scope", + ops: &fakeSessionOps{startNewErr: errors.New("boom")}, + wantReply: "Failed to start new session: boom", + }, + { + name: "prune error", + text: "/new", + scopeKey: "scope", + ops: &fakeSessionOps{ + startNewValue: "scope#2", + pruneErr: errors.New("prune failed"), + }, + wantReply: "Started new session (scope#2), but pruning old sessions failed: prune failed", + }, + { + name: "list error", + text: "/session list", + scopeKey: "scope", + ops: &fakeSessionOps{listErr: errors.New("list failed")}, + wantReply: "Failed to list sessions: list failed", + }, + { + name: "resume error", + text: "/session resume 2", + scopeKey: "scope", + ops: &fakeSessionOps{resumeErr: errors.New("resume failed")}, + wantReply: "Failed to resume session 2: resume failed", + }, + { + name: "resume missing index", + text: "/session resume", + scopeKey: "scope", + ops: &fakeSessionOps{}, + wantReply: "Usage: /session resume ", + }, + { + name: "resume non numeric index", + text: "/session resume abc", + scopeKey: "scope", + ops: &fakeSessionOps{}, + wantReply: "Usage: /session resume ", + }, + { + name: "resume zero index", + text: "/session resume 0", + scopeKey: "scope", + ops: &fakeSessionOps{}, + wantReply: "Usage: /session resume ", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + rt := &Runtime{SessionOps: tc.ops} + ex := NewExecutor(NewRegistry(BuiltinDefinitions()), rt) + + var reply string + res := ex.Execute(context.Background(), Request{ + ScopeKey: tc.scopeKey, + Text: tc.text, + Reply: func(text string) error { reply = text; return nil }, + }) + if res.Outcome != OutcomeHandled { + t.Fatalf("outcome=%v, want=%v", res.Outcome, OutcomeHandled) + } + if reply != tc.wantReply { + t.Fatalf("reply=%q, want=%q", reply, tc.wantReply) + } + }) + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index cff81a3a7..1dc76fe50 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 string `json:"dm_scope,omitempty"` + // 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 } // RoutingConfig controls the intelligent model routing feature. @@ -728,6 +753,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 10ebc7c90..284417723 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" @@ -479,3 +480,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 c4c04d41a..feec30167 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -39,7 +39,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..d74ffb318 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,325 @@ type Session struct { Updated time.Time `json:"updated"` } -type SessionManager struct { - sessions map[string]*Session - mu sync.RWMutex - storage string +const ( + sessionIndexFilename = "session_index.json" + legacySessionIndexFilename = "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, err := sm.ensureScopePersistedLocked(scopeKey, now) + if 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.RLock() + scope := sm.index.Scopes[scopeKey] + if scope != nil && len(scope.OrderedSessions) > 0 && scope.ActiveSessionKey != "" { + list := sm.buildSessionMetaListLocked(scope) + sm.mu.RUnlock() + return list, nil + } + sm.mu.RUnlock() + + sm.mu.Lock() + defer sm.mu.Unlock() + scope, err := sm.ensureScopePersistedLocked(scopeKey, time.Now()) + if err != nil { + return nil, err + } + return sm.buildSessionMetaListLocked(scope), 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, err := sm.ensureScopePersistedLocked(scopeKey, now) + if 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, err := sm.ensureScopePersistedLocked(scopeKey, now) + if 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 +462,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 +470,307 @@ 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 || file.Name() == legacySessionIndexFilename { + 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 + } + + indexPath := sm.indexPath + data, err := os.ReadFile(indexPath) + loadedFromLegacy := false + if err != nil { + if !os.IsNotExist(err) { + return err + } + legacyPath := filepath.Join(sm.storage, legacySessionIndexFilename) + data, err = os.ReadFile(legacyPath) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return err + } + indexPath = legacyPath + loadedFromLegacy = true + } + + 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 || loadedFromLegacy { + if err := sm.saveIndexLocked(); err != nil { + return err + } + } + if loadedFromLegacy && indexPath != sm.indexPath { + _ = removeFile(indexPath) + } + 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 (sm *SessionManager) ensureScopePersistedLocked(scopeKey string, now time.Time) (*scopeIndex, error) { + scope, changed := sm.ensureScopeLocked(scopeKey, now) + if changed { + if err := sm.saveIndexLocked(); err != nil { + return nil, err + } + } + return scope, nil +} + +func (sm *SessionManager) buildSessionMetaListLocked(scope *scopeIndex) []SessionMeta { + 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 +} + +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 +783,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 +893,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..0dd4d4516 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,416 @@ 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, sessionIndexFilename) + raw, err := os.ReadFile(indexPath) + if err != nil { + t.Fatalf("read index: %v", err) + } + + var decoded map[string]any + if unmarshalErr := json.Unmarshal(raw, &decoded); unmarshalErr != nil { + t.Fatalf("unmarshal index: %v", unmarshalErr) + } + + 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 TestLoadIndex_MigratesLegacyIndexFileName(t *testing.T) { + dir := t.TempDir() + scope := "agent:main:telegram:direct:user1" + + legacyPath := filepath.Join(dir, legacySessionIndexFilename) + raw := []byte(`{ + "version": 1, + "scopes": { + "agent:main:telegram:direct:user1": { + "active_session_key": "agent:main:telegram:direct:user1", + "ordered_sessions": ["agent:main:telegram:direct:user1"], + "updated_at": "2026-03-01T00:00:00Z" + } + } +}`) + if err := os.WriteFile(legacyPath, raw, 0o644); err != nil { + t.Fatalf("write legacy index: %v", err) + } + + sm := NewSessionManager(dir) + active, err := sm.ResolveActive(scope) + if err != nil { + t.Fatalf("ResolveActive failed: %v", err) + } + if active != scope { + t.Fatalf("active=%q, want=%q", active, scope) + } + + if _, err := os.Stat(filepath.Join(dir, sessionIndexFilename)); err != nil { + t.Fatalf("expected migrated index file to exist: %v", err) + } + if _, err := os.Stat(legacyPath); !os.IsNotExist(err) { + t.Fatalf("expected legacy index file removed, stat err=%v", err) + } +} + +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 _, statErr := os.Stat(sessionPath); statErr != nil { + t.Fatalf("expected %s to exist: %v", sessionPath, statErr) + } + + 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 _, statErr := os.Stat(removedFile); !os.IsNotExist(statErr) { + t.Fatalf("expected %s deleted, stat err=%v", removedFile, statErr) + } + + 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 writeErr := os.WriteFile(filepath.Join(dir, sessionIndexFilename), raw, 0o644); writeErr != nil { + t.Fatalf("write stale index: %v", writeErr) + } + + 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 unmarshalErr := json.Unmarshal(indexRaw, &healed); unmarshalErr != nil { + t.Fatalf("unmarshal healed index: %v", unmarshalErr) + } + + 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 delErr := sm.DeleteSession(sessionKey); delErr != nil { + t.Fatalf("DeleteSession(%q) returned unexpected error: %v", sessionKey, delErr) + } + 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 _, statErr := os.Stat(sessionPath); statErr != nil { + t.Fatalf("expected deferred file %q to still exist, err=%v", sessionPath, statErr) + } + + 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) + } +}