feat(session): Session management [Phase 2/3] scope-aware session rotation
This commit is contained in:
mingmxren
parent
b4471d6394
commit
c92712aa2c
14 changed files with 1550 additions and 81 deletions
21
README.md
21
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 <index>` 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"
|
||||
|
|
|
|||
21
README.zh.md
21
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 <index>` 只在当前作用域内生效。
|
||||
|
||||
### 心跳 / 周期性任务 (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"
|
||||
|
|
|
|||
|
|
@ -9,6 +9,10 @@
|
|||
"max_tool_iterations": 20
|
||||
}
|
||||
},
|
||||
"session": {
|
||||
"dm_scope": "per-channel-peer",
|
||||
"backlog_limit": 20
|
||||
},
|
||||
"model_list": [
|
||||
{
|
||||
"model_name": "gpt4",
|
||||
|
|
|
|||
|
|
@ -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 <index>]", 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 <index>", true
|
||||
}
|
||||
index, err := strconv.Atoi(args[1])
|
||||
if err != nil || index < 1 {
|
||||
return "Usage: /session resume <index>", 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 <index>]", true
|
||||
}
|
||||
|
||||
case "/show":
|
||||
if len(args) < 1 {
|
||||
return "Usage: /show [model|channel|agents]", true
|
||||
|
|
|
|||
|
|
@ -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-*")
|
||||
|
|
|
|||
|
|
@ -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 <index>]",
|
||||
},
|
||||
{
|
||||
Name: "show",
|
||||
Description: "Show current configuration",
|
||||
|
|
|
|||
|
|
@ -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 <index>] - 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")
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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{
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue