From af5ed21f5b3cd0b43f13f2f881f727af9059436d Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Wed, 4 Mar 2026 21:55:13 +0900 Subject: [PATCH] feat(session): implement SQLite SessionStore + LegacyAdapter (TASKS-3 Phase 0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace JSON file-based session storage with SQLite backend while maintaining full backward compatibility through LegacyAdapter. New files: - pkg/session/types.go: Turn, SessionInfo, CreateOpts, ListFilter types - pkg/session/store.go: SessionStore interface (15 methods) - pkg/session/sqlite.go: SQLite implementation (WAL mode, modernc.org/sqlite) - pkg/session/legacy_adapter.go: wraps SessionStore with SessionManager API - pkg/session/migrate.go: JSON → SQLite migration at startup Changes: - pkg/agent/instance.go: Sessions type changed to *LegacyAdapter - loop.go / loop_test.go: zero changes (method signatures identical) Co-Authored-By: Claude Opus 4.6 --- CLAUDE.md | 9 + pkg/agent/instance.go | 239 +++++++++++--- pkg/session/legacy_adapter.go | 496 ++++++++++++++++++++++++++++ pkg/session/legacy_adapter_test.go | 484 ++++++++++++++++++++++++++++ pkg/session/migrate.go | 123 +++++++ pkg/session/migrate_test.go | 276 ++++++++++++++++ pkg/session/sqlite.go | 497 +++++++++++++++++++++++++++++ pkg/session/sqlite_test.go | 481 ++++++++++++++++++++++++++++ pkg/session/store.go | 31 ++ pkg/session/types.go | 84 +++++ todo/TASKS-3.md | 27 +- 11 files changed, 2689 insertions(+), 58 deletions(-) create mode 100644 pkg/session/legacy_adapter.go create mode 100644 pkg/session/legacy_adapter_test.go create mode 100644 pkg/session/migrate.go create mode 100644 pkg/session/migrate_test.go create mode 100644 pkg/session/sqlite.go create mode 100644 pkg/session/sqlite_test.go create mode 100644 pkg/session/store.go create mode 100644 pkg/session/types.go diff --git a/CLAUDE.md b/CLAUDE.md index 26e08fa1d..6576f96b5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -26,6 +26,15 @@ Lint: `golangci-lint run` - **Sandbox/Spawn**: `pkg/tools/sandbox.go`, `pkg/tools/spawn.go` 実装済み。 - **AgentReporter**: `orch.AgentReporter` / `orch.Noop` / `orch.Broadcaster` で統一。main/heartbeat/subagent 全セッションが同一 Broadcaster に発火。Mini App は `agentLoop.GetOrchBroadcaster()` → `handler.SetOrchBroadcaster()` で受信。 +## Session DAG (Phase 0 実装済み) + +- **SQLite SessionStore**: `pkg/session/sqlite.go` — `modernc.org/sqlite` (CGO不要)、WAL モード、`sessions` + `turns` テーブル +- **SessionStore interface**: `pkg/session/store.go` — Create/Get/List/Append/Turns/Compact/Fork/Prune 等15メソッド +- **LegacyAdapter**: `pkg/session/legacy_adapter.go` — SessionStore をラップし SessionManager と同一 API を提供。`loop.go` 変更なし +- **JSON → SQLite migration**: `pkg/session/migrate.go` — 起動時に `sessions/*.json` を検出 → SQLite import → `.json.migrated` にリネーム +- **配線**: `pkg/agent/instance.go` の `Sessions` 型が `*LegacyAdapter` に変更。`sessions.db` を workspace 直下に生成 +- **Phase 1以降**: Fork/Report ターン導入、SessionGraph 直接呼び出し、Mini App 可視化 → `todo/TASKS-3.md` 参照 + ## コードの匂い — チェックリスト 新しいコードを書くとき・レビューするときの確認事項: diff --git a/pkg/agent/instance.go b/pkg/agent/instance.go index e0cf3ebf1..03f308ffb 100644 --- a/pkg/agent/instance.go +++ b/pkg/agent/instance.go @@ -17,90 +17,156 @@ import ( ) // AgentInstance represents a fully configured agent with its own workspace, + // session manager, context builder, and tool registry. + type AgentInstance struct { - ID string - Name string - Model string - Fallbacks []string - Workspace string - MaxIterations int + ID string + + Name string + + Model string + + Fallbacks []string + + Workspace string + + MaxIterations int + TaskReminderInterval int - MaxTokens int - Temperature float64 - ContextWindow int - Provider providers.LLMProvider - Sessions *session.SessionManager - ContextBuilder *ContextBuilder - Tools *tools.ToolRegistry - Subagents *config.SubagentsConfig - SkillsFilter []string - Candidates []providers.FallbackCandidate - PlanModel string - PlanFallbacks []string - PlanCandidates []providers.FallbackCandidate + + MaxTokens int + + Temperature float64 + + ContextWindow int + + Provider providers.LLMProvider + + Sessions *session.LegacyAdapter + + ContextBuilder *ContextBuilder + + Tools *tools.ToolRegistry + + Subagents *config.SubagentsConfig + + SkillsFilter []string + + Candidates []providers.FallbackCandidate + + PlanModel string + + PlanFallbacks []string + + PlanCandidates []providers.FallbackCandidate // SubagentMgr is set during registerSharedTools when orchestration is enabled. + // Used by runAgentLoop to wait for spawned subagents before worktree cleanup. + SubagentMgr *tools.SubagentManager // Interview staleness tracking: consecutive turns where MEMORY.md was not updated. + interviewStaleCount int - interviewMemoryLen int + + interviewMemoryLen int // Per-session worktree isolation - worktrees map[string]*git.WorktreeInfo // sessionKey → worktree + + worktrees map[string]*git.WorktreeInfo // sessionKey → worktree + worktreeMu sync.RWMutex } // NewAgentInstance creates an agent instance from config. + func NewAgentInstance( agentCfg *config.AgentConfig, + defaults *config.AgentDefaults, + cfg *config.Config, + provider providers.LLMProvider, ) *AgentInstance { workspace := resolveAgentWorkspace(agentCfg, defaults) + os.MkdirAll(workspace, 0o755) model := resolveAgentModel(agentCfg, defaults) + fallbacks := resolveAgentFallbacks(agentCfg, defaults) restrict := defaults.RestrictToWorkspace + toolsRegistry := tools.NewToolRegistry() + toolsRegistry.Register(tools.NewReadFileTool(workspace, restrict)) + toolsRegistry.Register(tools.NewWriteFileTool(workspace, restrict)) + toolsRegistry.Register(tools.NewListDirTool(workspace, restrict)) + execTool, err := tools.NewExecToolWithConfig(workspace, restrict, cfg) if err != nil { log.Fatalf("Critical error: unable to initialize exec tool: %v", err) } + toolsRegistry.Register(execTool) + toolsRegistry.Register(tools.NewBgMonitorTool(execTool)) + toolsRegistry.Register(tools.NewEditFileTool(workspace, restrict)) + toolsRegistry.Register(tools.NewAppendFileTool(workspace, restrict)) + toolsRegistry.Register(tools.NewLogsTool()) + toolsRegistry.Register(tools.NewGitPushTool()) + toolsRegistry.Register(tools.NewCreatePRTool()) - sessionsDir := filepath.Join(workspace, "sessions") - sessionsManager := session.NewSessionManager(sessionsDir) + dbPath := filepath.Join(workspace, "sessions.db") + + store, err := session.OpenSQLiteStore(dbPath) + if err != nil { + log.Fatalf("open session store: %v", err) + } + + jsonDir := filepath.Join(workspace, "sessions") + + if n, merr := session.MigrateJSONSessions(jsonDir, store); merr != nil { + log.Printf("session migration: %d migrated, error: %v", n, merr) + } else if n > 0 { + log.Printf("session migration: %d sessions migrated to SQLite", n) + } + + sessionsManager := session.NewLegacyAdapter(store) contextBuilder := NewContextBuilder(workspace) agentID := routing.DefaultAgentID + agentName := "" + var subagents *config.SubagentsConfig + var skillsFilter []string if agentCfg != nil { agentID = routing.NormalizeAgentID(agentCfg.ID) + agentName = agentCfg.Name + subagents = agentCfg.Subagents + skillsFilter = agentCfg.Skills } // Apply defaults.Orchestration: if the flag is set, ensure orchestration is enabled. + if defaults.Orchestration { if subagents == nil { subagents = &config.SubagentsConfig{Enabled: true} @@ -110,43 +176,54 @@ func NewAgentInstance( } maxIter := defaults.MaxToolIterations + if maxIter == 0 { maxIter = 20 } reminderInterval := defaults.TaskReminderInterval + if reminderInterval == 0 { reminderInterval = 5 } maxTokens := defaults.MaxTokens + if maxTokens == 0 { maxTokens = 8192 } temperature := 0.7 + if defaults.Temperature != nil { temperature = *defaults.Temperature } // Resolve fallback candidates + modelCfg := providers.ModelConfig{ - Primary: model, + Primary: model, + Fallbacks: fallbacks, } + resolveFromModelList := func(raw string) (string, bool) { ensureProtocol := func(model string) string { model = strings.TrimSpace(model) + if model == "" { return "" } + if strings.Contains(model, "/") { return model } + return "openai/" + model } raw = strings.TrimSpace(raw) + if raw == "" { return "", false } @@ -158,13 +235,17 @@ func NewAgentInstance( for i := range cfg.ModelList { fullModel := strings.TrimSpace(cfg.ModelList[i].Model) + if fullModel == "" { continue } + if fullModel == raw { return ensureProtocol(fullModel), true } + _, modelID := providers.ExtractProtocol(fullModel) + if modelID == raw { return ensureProtocol(fullModel), true } @@ -177,107 +258,155 @@ func NewAgentInstance( candidates := providers.ResolveCandidatesWithLookup(modelCfg, defaults.Provider, resolveFromModelList) // Resolve plan model (for interviewing/review phases) + planModel := resolvePlanModel(agentCfg, defaults) + planFallbacks := resolvePlanFallbacks(agentCfg, defaults) + var planCandidates []providers.FallbackCandidate + if planModel != "" { planModelCfg := providers.ModelConfig{ - Primary: planModel, + Primary: planModel, + Fallbacks: planFallbacks, } + planCandidates = providers.ResolveCandidates(planModelCfg, defaults.Provider) } // Startup cleanup: prune orphaned worktrees + worktreesDir := filepath.Join(workspace, ".worktrees") + if repoRoot := git.FindRepoRoot(workspace); repoRoot != "" { git.PruneOrphaned(repoRoot, worktreesDir) } return &AgentInstance{ - ID: agentID, - Name: agentName, - Model: model, - Fallbacks: fallbacks, - Workspace: workspace, - MaxIterations: maxIter, + ID: agentID, + + Name: agentName, + + Model: model, + + Fallbacks: fallbacks, + + Workspace: workspace, + + MaxIterations: maxIter, + TaskReminderInterval: reminderInterval, - MaxTokens: maxTokens, - Temperature: temperature, - ContextWindow: maxTokens, - Provider: provider, - Sessions: sessionsManager, - ContextBuilder: contextBuilder, - Tools: toolsRegistry, - Subagents: subagents, - SkillsFilter: skillsFilter, - Candidates: candidates, - PlanModel: planModel, - PlanFallbacks: planFallbacks, - PlanCandidates: planCandidates, + + MaxTokens: maxTokens, + + Temperature: temperature, + + ContextWindow: maxTokens, + + Provider: provider, + + Sessions: sessionsManager, + + ContextBuilder: contextBuilder, + + Tools: toolsRegistry, + + Subagents: subagents, + + SkillsFilter: skillsFilter, + + Candidates: candidates, + + PlanModel: planModel, + + PlanFallbacks: planFallbacks, + + PlanCandidates: planCandidates, } } // resolveAgentWorkspace determines the workspace directory for an agent. + func resolveAgentWorkspace(agentCfg *config.AgentConfig, defaults *config.AgentDefaults) string { if agentCfg != nil && strings.TrimSpace(agentCfg.Workspace) != "" { return expandHome(strings.TrimSpace(agentCfg.Workspace)) } + if agentCfg == nil || agentCfg.Default || agentCfg.ID == "" || routing.NormalizeAgentID(agentCfg.ID) == "main" { return expandHome(defaults.Workspace) } + home, _ := os.UserHomeDir() + id := routing.NormalizeAgentID(agentCfg.ID) + return filepath.Join(home, ".picoclaw", "workspace-"+id) } // resolveAgentModel resolves the primary model for an agent. + func resolveAgentModel(agentCfg *config.AgentConfig, defaults *config.AgentDefaults) string { if agentCfg != nil && agentCfg.Model != nil && strings.TrimSpace(agentCfg.Model.Primary) != "" { return strings.TrimSpace(agentCfg.Model.Primary) } + return defaults.GetModelName() } // resolveAgentFallbacks resolves the fallback models for an agent. + func resolveAgentFallbacks(agentCfg *config.AgentConfig, defaults *config.AgentDefaults) []string { if agentCfg != nil && agentCfg.Model != nil && agentCfg.Model.Fallbacks != nil { return agentCfg.Model.Fallbacks } + return defaults.ModelFallbacks } // resolvePlanModel resolves the plan model for an agent (used during interviewing/review phases). + func resolvePlanModel(agentCfg *config.AgentConfig, defaults *config.AgentDefaults) string { if agentCfg != nil && agentCfg.PlanModel != nil && strings.TrimSpace(agentCfg.PlanModel.Primary) != "" { return strings.TrimSpace(agentCfg.PlanModel.Primary) } + return defaults.PlanModel } // resolvePlanFallbacks resolves the plan model fallbacks for an agent. + func resolvePlanFallbacks(agentCfg *config.AgentConfig, defaults *config.AgentDefaults) []string { if agentCfg != nil && agentCfg.PlanModel != nil && agentCfg.PlanModel.Fallbacks != nil { return agentCfg.PlanModel.Fallbacks } + return defaults.PlanModelFallbacks } // ActivateWorktree creates a worktree for a session. + // projectDir is the git repository to create the worktree in. + // If empty, falls back to ai.Workspace. + // Worktree path: /.worktrees// + func (ai *AgentInstance) ActivateWorktree(sessionKey, taskName, projectDir string) (*git.WorktreeInfo, error) { if projectDir == "" { projectDir = ai.Workspace } + repoRoot := git.FindRepoRoot(projectDir) + if repoRoot == "" { return nil, fmt.Errorf("directory is not a git repository: %s", projectDir) } branchName := git.SanitizeBranchName(taskName) + baseName := git.BranchBaseName(branchName) + wtPath := filepath.Join(ai.Workspace, ".worktrees", baseName) wt, err := git.CreateWorktree(repoRoot, wtPath, branchName) @@ -286,22 +415,29 @@ func (ai *AgentInstance) ActivateWorktree(sessionKey, taskName, projectDir strin } ai.worktreeMu.Lock() + if ai.worktrees == nil { ai.worktrees = make(map[string]*git.WorktreeInfo) } + ai.worktrees[sessionKey] = wt + ai.worktreeMu.Unlock() return wt, nil } // DeactivateWorktree safe-disposes the session's worktree. + func (ai *AgentInstance) DeactivateWorktree(sessionKey, commitMsg string, discard bool) (*git.DisposeResult, error) { ai.worktreeMu.Lock() + wt, ok := ai.worktrees[sessionKey] + if ok { delete(ai.worktrees, sessionKey) } + ai.worktreeMu.Unlock() if !ok || wt == nil { @@ -309,44 +445,55 @@ func (ai *AgentInstance) DeactivateWorktree(sessionKey, commitMsg string, discar } repoRoot := git.FindRepoRoot(ai.Workspace) + if repoRoot == "" { return nil, fmt.Errorf("workspace is not a git repository") } // Even on discard, SafeDispose auto-commits first for safety + if commitMsg != "" && git.HasUncommittedChanges(wt.Path) { _ = git.AutoCommit(wt.Path, commitMsg) } result := git.SafeDispose(repoRoot, wt) + return &result, nil } // GetWorktree returns the session's active worktree, or nil. + func (ai *AgentInstance) GetWorktree(sessionKey string) *git.WorktreeInfo { ai.worktreeMu.RLock() + defer ai.worktreeMu.RUnlock() + return ai.worktrees[sessionKey] } // IsInWorktree returns true if the session has an active worktree. + func (ai *AgentInstance) IsInWorktree(sessionKey string) bool { return ai.GetWorktree(sessionKey) != nil } // EffectiveWorkspace returns worktree path for session, or original Workspace. + func (ai *AgentInstance) EffectiveWorkspace(sessionKey string) string { if wt := ai.GetWorktree(sessionKey); wt != nil { return wt.Path } + return ai.Workspace } // GetWorktreeBranch returns the branch name for the session's worktree, or "". + func (ai *AgentInstance) GetWorktreeBranch(sessionKey string) string { if wt := ai.GetWorktree(sessionKey); wt != nil { return wt.Branch } + return "" } @@ -354,12 +501,16 @@ func expandHome(path string) string { if path == "" { return path } + if path[0] == '~' { home, _ := os.UserHomeDir() + if len(path) > 1 && path[1] == '/' { return home + path[1:] } + return home } + return path } diff --git a/pkg/session/legacy_adapter.go b/pkg/session/legacy_adapter.go new file mode 100644 index 000000000..9440a5915 --- /dev/null +++ b/pkg/session/legacy_adapter.go @@ -0,0 +1,496 @@ +package session + +import ( + "sync" + "time" + + "github.com/sipeed/picoclaw/pkg/providers" +) + +// LegacyAdapter wraps a SessionStore and exposes the same public API as + +// SessionManager so that all existing call sites (loop.go, etc.) work + +// without modification. + +type LegacyAdapter struct { + store SessionStore + + mu sync.RWMutex + + cache map[string]*sessionCache + + dirtyMu sync.Mutex + + dirtyKeys map[string]bool + + done chan struct{} +} + +type sessionCache struct { + messages []providers.Message + + summary string + + created time.Time + + updated time.Time + + dirty bool + + replaced bool // SetHistory/TruncateHistory set this; Save does full rewrite + + stored int // number of messages already persisted in the store +} + +// NewLegacyAdapter creates a LegacyAdapter backed by the given store. + +func NewLegacyAdapter(store SessionStore) *LegacyAdapter { + la := &LegacyAdapter{ + store: store, + + cache: make(map[string]*sessionCache), + + dirtyKeys: make(map[string]bool), + + done: make(chan struct{}), + } + + go la.flushLoop() + + return la +} + +// getOrLoad returns the cache entry for key, loading from the store if needed. + +// Caller must hold la.mu (write lock). + +func (la *LegacyAdapter) getOrLoad(key string) *sessionCache { + if c, ok := la.cache[key]; ok { + return c + } + + // Try loading from store + + info, err := la.store.Get(key) + + if err != nil || info == nil { + // Create in store + + _ = la.store.Create(key, nil) + + now := time.Now() + + c := &sessionCache{ + messages: []providers.Message{}, + + created: now, + + updated: now, + } + + la.cache[key] = c + + return c + } + + // Load all turns and reconstruct messages + + turns, _ := la.store.Turns(key, 0) + + var msgs []providers.Message + + for _, t := range turns { + msgs = append(msgs, t.Messages...) + } + + if msgs == nil { + msgs = []providers.Message{} + } + + c := &sessionCache{ + messages: msgs, + + summary: info.Summary, + + created: info.CreatedAt, + + updated: info.UpdatedAt, + + stored: len(msgs), + } + + la.cache[key] = c + + return c +} + +// GetOrCreate returns a Session-compatible object for the given key. + +// Creates the session if it doesn't exist. + +func (la *LegacyAdapter) GetOrCreate(key string) *Session { + la.mu.Lock() + + defer la.mu.Unlock() + + c := la.getOrLoad(key) + + return &Session{ + Key: key, + + Messages: c.messages, + + Summary: c.summary, + + Created: c.created, + + Updated: c.updated, + } +} + +// AddMessage adds a simple message to the session. + +func (la *LegacyAdapter) AddMessage(sessionKey, role, content string) { + la.AddFullMessage(sessionKey, providers.Message{ + Role: role, + + Content: content, + }) +} + +// AddFullMessage adds a complete message with tool calls to the session. + +func (la *LegacyAdapter) AddFullMessage(sessionKey string, msg providers.Message) { + la.mu.Lock() + + defer la.mu.Unlock() + + c := la.getOrLoad(sessionKey) + + c.messages = append(c.messages, msg) + + c.updated = time.Now() + + c.dirty = true +} + +// GetHistory returns a defensive copy of the session messages. + +func (la *LegacyAdapter) GetHistory(key string) []providers.Message { + la.mu.RLock() + + c, ok := la.cache[key] + + la.mu.RUnlock() + + if !ok { + // Try lazy load + + la.mu.Lock() + + c, ok = la.cache[key] + + if !ok { + // Check if it exists in the store + + info, _ := la.store.Get(key) + + if info == nil { + la.mu.Unlock() + + return []providers.Message{} + } + + c = la.getOrLoad(key) + } + + la.mu.Unlock() + } + + la.mu.RLock() + + defer la.mu.RUnlock() + + history := make([]providers.Message, len(c.messages)) + + copy(history, c.messages) + + return history +} + +// SetHistory replaces the session's message history entirely. + +func (la *LegacyAdapter) SetHistory(key string, history []providers.Message) { + la.mu.Lock() + + defer la.mu.Unlock() + + c, ok := la.cache[key] + + if !ok { + return + } + + msgs := make([]providers.Message, len(history)) + + copy(msgs, history) + + c.messages = msgs + + c.updated = time.Now() + + c.replaced = true + + c.dirty = true +} + +// GetSummary returns the session summary. + +func (la *LegacyAdapter) GetSummary(key string) string { + la.mu.RLock() + + c, ok := la.cache[key] + + la.mu.RUnlock() + + if !ok { + la.mu.Lock() + + c, ok = la.cache[key] + + if !ok { + info, _ := la.store.Get(key) + + if info == nil { + la.mu.Unlock() + + return "" + } + + c = la.getOrLoad(key) + } + + la.mu.Unlock() + } + + la.mu.RLock() + + defer la.mu.RUnlock() + + return c.summary +} + +// SetSummary updates the session summary in cache and store. + +func (la *LegacyAdapter) SetSummary(key string, summary string) { + la.mu.Lock() + + defer la.mu.Unlock() + + c, ok := la.cache[key] + + if !ok { + return + } + + c.summary = summary + + c.updated = time.Now() + + _ = la.store.SetSummary(key, summary) +} + +// TruncateHistory keeps only the last n messages. + +func (la *LegacyAdapter) TruncateHistory(key string, keepLast int) { + la.mu.Lock() + + defer la.mu.Unlock() + + c, ok := la.cache[key] + + if !ok { + return + } + + if keepLast <= 0 { + c.messages = []providers.Message{} + + c.updated = time.Now() + + c.replaced = true + + c.dirty = true + + return + } + + if len(c.messages) <= keepLast { + return + } + + c.messages = c.messages[len(c.messages)-keepLast:] + + c.updated = time.Now() + + c.replaced = true + + c.dirty = true +} + +// MarkDirty marks a session key for deferred persistence. + +func (la *LegacyAdapter) MarkDirty(key string) { + la.dirtyMu.Lock() + + la.dirtyKeys[key] = true + + la.dirtyMu.Unlock() +} + +// FlushDirty writes all dirty sessions to the store. + +func (la *LegacyAdapter) FlushDirty() { + la.dirtyMu.Lock() + + keys := make([]string, 0, len(la.dirtyKeys)) + + for k := range la.dirtyKeys { + keys = append(keys, k) + } + + la.dirtyKeys = make(map[string]bool) + + la.dirtyMu.Unlock() + + for _, k := range keys { + la.Save(k) + } +} + +// Save persists the session to the store. + +func (la *LegacyAdapter) Save(key string) error { + la.mu.RLock() + + c, ok := la.cache[key] + + if !ok { + la.mu.RUnlock() + + return nil + } + + // Snapshot under read lock + + replaced := c.replaced + + stored := c.stored + + msgs := make([]providers.Message, len(c.messages)) + + copy(msgs, c.messages) + + la.mu.RUnlock() + + if replaced { + // Full rewrite: compact all existing turns then write the whole history + + if err := la.store.Compact(key, 1<<31, ""); err != nil { + return err + } + + if len(msgs) > 0 { + turn := &Turn{ + SessionKey: key, + + Kind: TurnNormal, + + Messages: msgs, + } + + if err := la.store.Append(key, turn); err != nil { + return err + } + } + + la.mu.Lock() + + if cc, ok := la.cache[key]; ok { + cc.replaced = false + + cc.stored = len(msgs) + + cc.dirty = false + } + + la.mu.Unlock() + } else { + // Incremental: only append new messages + + newMsgs := msgs[stored:] + + if len(newMsgs) > 0 { + turn := &Turn{ + SessionKey: key, + + Kind: TurnNormal, + + Messages: newMsgs, + } + + if err := la.store.Append(key, turn); err != nil { + return err + } + } + + la.mu.Lock() + + if cc, ok := la.cache[key]; ok { + cc.stored = len(msgs) + + cc.dirty = false + } + + la.mu.Unlock() + } + + return nil +} + +// Close stops the background flush loop and persists all dirty sessions. + +func (la *LegacyAdapter) Close() { + select { + case <-la.done: + + return // already closed + + default: + } + + close(la.done) + + la.FlushDirty() + + la.store.Close() +} + +func (la *LegacyAdapter) flushLoop() { + ticker := time.NewTicker(5 * time.Minute) + + defer ticker.Stop() + + for { + select { + case <-ticker.C: + + la.FlushDirty() + + case <-la.done: + + return + } + } +} diff --git a/pkg/session/legacy_adapter_test.go b/pkg/session/legacy_adapter_test.go new file mode 100644 index 000000000..216875b30 --- /dev/null +++ b/pkg/session/legacy_adapter_test.go @@ -0,0 +1,484 @@ +package session + +import ( + "path/filepath" + "testing" + + "github.com/sipeed/picoclaw/pkg/providers" +) + +// sessionBackend abstracts the common API shared by SessionManager and LegacyAdapter. + +type sessionBackend interface { //nolint:interfacebloat // test helper mirrors SessionManager API + + GetOrCreate(key string) *Session + + AddMessage(sessionKey, role, content string) + + AddFullMessage(sessionKey string, msg providers.Message) + + GetHistory(key string) []providers.Message + + SetHistory(key string, history []providers.Message) + + GetSummary(key string) string + + SetSummary(key string, summary string) + + TruncateHistory(key string, keepLast int) + + MarkDirty(key string) + + FlushDirty() + + Save(key string) error + + Close() +} + +func backends(t *testing.T) map[string]sessionBackend { + t.Helper() + + jsonDir := t.TempDir() + + sm := NewSessionManager(jsonDir) + + t.Cleanup(func() { sm.Close() }) + + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, err := OpenSQLiteStore(dbPath) + if err != nil { + t.Fatalf("OpenSQLiteStore: %v", err) + } + + la := NewLegacyAdapter(store) + + t.Cleanup(func() { la.Close() }) + + return map[string]sessionBackend{ + "json": sm, + + "sqlite": la, + } +} + +func TestBackend_GetOrCreate(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + s := be.GetOrCreate("k1") + + if s.Key != "k1" { + t.Errorf("expected key k1, got %s", s.Key) + } + + if len(s.Messages) != 0 { + t.Errorf("expected empty messages, got %d", len(s.Messages)) + } + + // Second call returns existing + + be.AddMessage("k1", "user", "hello") + + s2 := be.GetOrCreate("k1") + + if s2.Key != "k1" { + t.Errorf("expected key k1 on second call") + } + }) + } +} + +func TestBackend_AddMessageAndGetHistory(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + be.GetOrCreate("k1") + + be.AddMessage("k1", "user", "hello") + + be.AddMessage("k1", "assistant", "hi") + + history := be.GetHistory("k1") + + if len(history) != 2 { + t.Fatalf("expected 2 messages, got %d", len(history)) + } + + if history[0].Role != "user" || history[0].Content != "hello" { + t.Errorf("unexpected first message: %+v", history[0]) + } + + if history[1].Role != "assistant" || history[1].Content != "hi" { + t.Errorf("unexpected second message: %+v", history[1]) + } + }) + } +} + +func TestBackend_AddFullMessage(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + be.GetOrCreate("k1") + + be.AddFullMessage("k1", providers.Message{ + Role: "assistant", + + Content: "sure", + + ToolCalls: []providers.ToolCall{ + {ID: "call_1", Type: "function", Function: &providers.FunctionCall{Name: "exec", Arguments: `{}`}}, + }, + }) + + be.AddFullMessage("k1", providers.Message{ + Role: "tool", + + Content: "ok", + + ToolCallID: "call_1", + }) + + history := be.GetHistory("k1") + + if len(history) != 2 { + t.Fatalf("expected 2, got %d", len(history)) + } + + if history[0].ToolCalls[0].ID != "call_1" { + t.Errorf("tool call ID mismatch") + } + + if history[1].ToolCallID != "call_1" { + t.Errorf("tool call result ID mismatch") + } + }) + } +} + +func TestBackend_AddFullMessage_AutoCreates(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + // AddFullMessage without prior GetOrCreate should still work + + be.AddFullMessage("auto", providers.Message{Role: "user", Content: "hi"}) + + history := be.GetHistory("auto") + + if len(history) != 1 { + t.Fatalf("expected 1, got %d", len(history)) + } + }) + } +} + +func TestBackend_SetHistory(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + be.GetOrCreate("k1") + + be.AddMessage("k1", "user", "old") + + newHistory := []providers.Message{ + {Role: "user", Content: "new1"}, + + {Role: "assistant", Content: "new2"}, + } + + be.SetHistory("k1", newHistory) + + got := be.GetHistory("k1") + + if len(got) != 2 || got[0].Content != "new1" || got[1].Content != "new2" { + t.Errorf("unexpected history after SetHistory: %+v", got) + } + }) + } +} + +func TestBackend_GetSetSummary(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + be.GetOrCreate("k1") + + if s := be.GetSummary("k1"); s != "" { + t.Errorf("expected empty summary, got %q", s) + } + + be.SetSummary("k1", "test summary") + + if s := be.GetSummary("k1"); s != "test summary" { + t.Errorf("expected 'test summary', got %q", s) + } + }) + } +} + +func TestBackend_TruncateHistory(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + be.GetOrCreate("k1") + + for i := range 10 { + be.AddMessage("k1", "user", string(rune('a'+i))) + } + + be.TruncateHistory("k1", 3) + + got := be.GetHistory("k1") + + if len(got) != 3 { + t.Fatalf("expected 3, got %d", len(got)) + } + + if got[0].Content != "h" { + t.Errorf("expected 'h', got %q", got[0].Content) + } + }) + } +} + +func TestBackend_TruncateHistory_Zero(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + be.GetOrCreate("k1") + + be.AddMessage("k1", "user", "hello") + + be.TruncateHistory("k1", 0) + + got := be.GetHistory("k1") + + if len(got) != 0 { + t.Errorf("expected 0, got %d", len(got)) + } + }) + } +} + +func TestBackend_TruncateHistory_LargerThanLen(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + be.GetOrCreate("k1") + + be.AddMessage("k1", "user", "hello") + + be.TruncateHistory("k1", 100) + + got := be.GetHistory("k1") + + if len(got) != 1 { + t.Errorf("expected 1, got %d", len(got)) + } + }) + } +} + +func TestBackend_GetHistory_DefensiveCopy(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + be.GetOrCreate("k1") + + be.AddMessage("k1", "user", "hello") + + h1 := be.GetHistory("k1") + + h1[0].Content = "modified" + + h2 := be.GetHistory("k1") + + if h2[0].Content != "hello" { + t.Errorf("defensive copy failed: %q", h2[0].Content) + } + }) + } +} + +func TestBackend_GetHistory_NonExistent(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + got := be.GetHistory("nope") + + if got == nil || len(got) != 0 { + t.Errorf("expected empty slice, got %v", got) + } + }) + } +} + +func TestBackend_MarkDirtyAndFlush(t *testing.T) { + for name, be := range backends(t) { + t.Run(name, func(t *testing.T) { + be.GetOrCreate("k1") + + be.AddMessage("k1", "user", "hello") + + be.MarkDirty("k1") + + be.FlushDirty() + + // Should not panic or error + }) + } +} + +func TestBackend_SaveAndReload(t *testing.T) { + // Test that Save persists data that can be reloaded. + + // For JSON backend, we reload via new SessionManager. + + // For SQLite, we reload via new LegacyAdapter on same DB. + + t.Run("json", func(t *testing.T) { + dir := t.TempDir() + + sm := NewSessionManager(dir) + + sm.GetOrCreate("k1") + + sm.AddMessage("k1", "user", "hello") + + sm.SetSummary("k1", "test") + + sm.Save("k1") + + sm.Close() + + sm2 := NewSessionManager(dir) + + defer sm2.Close() + + h := sm2.GetHistory("k1") + + if len(h) != 1 || h[0].Content != "hello" { + t.Errorf("json reload: expected [hello], got %+v", h) + } + + if s := sm2.GetSummary("k1"); s != "test" { + t.Errorf("json reload summary: expected 'test', got %q", s) + } + }) + + t.Run("sqlite", func(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, _ := OpenSQLiteStore(dbPath) + + la := NewLegacyAdapter(store) + + la.GetOrCreate("k1") + + la.AddMessage("k1", "user", "hello") + + la.SetSummary("k1", "test") + + la.Save("k1") + + la.Close() + + store2, _ := OpenSQLiteStore(dbPath) + + la2 := NewLegacyAdapter(store2) + + defer la2.Close() + + h := la2.GetHistory("k1") + + if len(h) != 1 || h[0].Content != "hello" { + t.Errorf("sqlite reload: expected [hello], got %+v", h) + } + + if s := la2.GetSummary("k1"); s != "test" { + t.Errorf("sqlite reload summary: expected 'test', got %q", s) + } + }) +} + +func TestBackend_SaveAfterSetHistory(t *testing.T) { + // Verify that Save after SetHistory (full replacement) works correctly + + t.Run("sqlite", func(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, _ := OpenSQLiteStore(dbPath) + + la := NewLegacyAdapter(store) + + la.GetOrCreate("k1") + + la.AddMessage("k1", "user", "old1") + + la.AddMessage("k1", "user", "old2") + + la.Save("k1") + + // Replace history + + la.SetHistory("k1", []providers.Message{ + {Role: "user", Content: "new1"}, + }) + + la.Save("k1") + + la.Close() + + // Reload and verify + + store2, _ := OpenSQLiteStore(dbPath) + + la2 := NewLegacyAdapter(store2) + + defer la2.Close() + + h := la2.GetHistory("k1") + + if len(h) != 1 || h[0].Content != "new1" { + t.Errorf("expected [new1], got %+v", h) + } + }) +} + +func TestBackend_IncrementalSave(t *testing.T) { + // Verify that incremental saves only add new messages + + t.Run("sqlite", func(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, _ := OpenSQLiteStore(dbPath) + + la := NewLegacyAdapter(store) + + la.GetOrCreate("k1") + + la.AddMessage("k1", "user", "msg1") + + la.Save("k1") + + la.AddMessage("k1", "user", "msg2") + + la.Save("k1") + + la.Close() + + // Verify 2 turns were created (one per save) + + store2, _ := OpenSQLiteStore(dbPath) + + defer store2.Close() + + turns, _ := store2.Turns("k1", 0) + + if len(turns) != 2 { + t.Errorf("expected 2 turns (incremental), got %d", len(turns)) + } + + // But total messages should be 2 + + la2 := NewLegacyAdapter(store2) + + h := la2.GetHistory("k1") + + if len(h) != 2 { + t.Errorf("expected 2 messages total, got %d", len(h)) + } + }) +} diff --git a/pkg/session/migrate.go b/pkg/session/migrate.go new file mode 100644 index 000000000..eb5f55305 --- /dev/null +++ b/pkg/session/migrate.go @@ -0,0 +1,123 @@ +package session + +import ( + "encoding/json" + "log" + "os" + "path/filepath" + "strings" +) + +// MigrateJSONSessions reads JSON session files from jsonDir and imports them + +// into the given SessionStore. Successfully migrated files are renamed to + +// .json.migrated so they are skipped on subsequent runs. + +// + +// Individual file errors are logged and skipped (the file remains for retry). + +// Returns the number of sessions migrated and the first error encountered, if any. + +func MigrateJSONSessions(jsonDir string, store SessionStore) (int, error) { + entries, err := os.ReadDir(jsonDir) + if err != nil { + if os.IsNotExist(err) { + return 0, nil + } + + return 0, err + } + + migrated := 0 + + for _, entry := range entries { + if entry.IsDir() { + continue + } + + name := entry.Name() + + if !strings.HasSuffix(name, ".json") { + continue + } + + path := filepath.Join(jsonDir, name) + + data, err := os.ReadFile(path) + if err != nil { + log.Printf("session migrate: read %s: %v", name, err) + + continue + } + + var sess Session + + if err := json.Unmarshal(data, &sess); err != nil { + log.Printf("session migrate: parse %s: %v", name, err) + + continue + } + + if sess.Key == "" { + log.Printf("session migrate: skip %s: empty key", name) + + continue + } + + // Create session in store (skip if already exists) + + if existing, _ := store.Get(sess.Key); existing != nil { + // Already migrated (perhaps from a previous partial run) + + _ = os.Rename(path, path+".migrated") + + migrated++ + + continue + } + + if err := store.Create(sess.Key, nil); err != nil { + log.Printf("session migrate: create %s: %v", sess.Key, err) + + continue + } + + // Import messages as a single turn + + if len(sess.Messages) > 0 { + turn := &Turn{ + SessionKey: sess.Key, + + Kind: TurnNormal, + + Messages: sess.Messages, + + CreatedAt: sess.Created, + } + + if err := store.Append(sess.Key, turn); err != nil { + log.Printf("session migrate: append %s: %v", sess.Key, err) + + continue + } + } + + // Set summary if present + + if sess.Summary != "" { + _ = store.SetSummary(sess.Key, sess.Summary) + } + + // Mark as migrated + + if err := os.Rename(path, path+".migrated"); err != nil { + log.Printf("session migrate: rename %s: %v", name, err) + } + + migrated++ + } + + return migrated, nil +} diff --git a/pkg/session/migrate_test.go b/pkg/session/migrate_test.go new file mode 100644 index 000000000..22e2ff1c3 --- /dev/null +++ b/pkg/session/migrate_test.go @@ -0,0 +1,276 @@ +package session + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/sipeed/picoclaw/pkg/providers" +) + +func writeJSONSession(t *testing.T, dir string, sess Session) { + t.Helper() + + data, err := json.MarshalIndent(sess, "", " ") + if err != nil { + t.Fatal(err) + } + + filename := sanitizeFilename(sess.Key) + ".json" + + if err := os.WriteFile(filepath.Join(dir, filename), data, 0o644); err != nil { + t.Fatal(err) + } +} + +func TestMigrate_Basic(t *testing.T) { + jsonDir := t.TempDir() + + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, err := OpenSQLiteStore(dbPath) + if err != nil { + t.Fatal(err) + } + + defer store.Close() + + writeJSONSession(t, jsonDir, Session{ + Key: "telegram:123", + + Messages: []providers.Message{ + {Role: "user", Content: "hello"}, + + {Role: "assistant", Content: "hi"}, + }, + + Summary: "greeting", + }) + + writeJSONSession(t, jsonDir, Session{ + Key: "discord:456", + + Messages: []providers.Message{ + {Role: "user", Content: "test"}, + }, + }) + + migrated, err := MigrateJSONSessions(jsonDir, store) + if err != nil { + t.Fatalf("MigrateJSONSessions: %v", err) + } + + if migrated != 2 { + t.Errorf("expected 2 migrated, got %d", migrated) + } + + // Verify sessions exist + + info, _ := store.Get("telegram:123") + + if info == nil || info.Summary != "greeting" { + t.Errorf("telegram:123 not found or wrong summary") + } + + turns, _ := store.Turns("telegram:123", 0) + + if len(turns) != 1 || len(turns[0].Messages) != 2 { + t.Errorf("expected 1 turn with 2 messages, got %+v", turns) + } + + info2, _ := store.Get("discord:456") + + if info2 == nil { + t.Error("discord:456 not found") + } + + // Verify .json files were renamed + + entries, _ := os.ReadDir(jsonDir) + + for _, e := range entries { + if filepath.Ext(e.Name()) == ".json" { + t.Errorf("expected .json.migrated, found %s", e.Name()) + } + } +} + +func TestMigrate_EmptyMessages(t *testing.T) { + jsonDir := t.TempDir() + + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, _ := OpenSQLiteStore(dbPath) + + defer store.Close() + + writeJSONSession(t, jsonDir, Session{ + Key: "empty:1", + + Messages: []providers.Message{}, + }) + + migrated, err := MigrateJSONSessions(jsonDir, store) + if err != nil { + t.Fatal(err) + } + + if migrated != 1 { + t.Errorf("expected 1, got %d", migrated) + } + + // Should exist but have no turns + + count, _ := store.TurnCount("empty:1") + + if count != 0 { + t.Errorf("expected 0 turns, got %d", count) + } +} + +func TestMigrate_EmptySummary(t *testing.T) { + jsonDir := t.TempDir() + + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, _ := OpenSQLiteStore(dbPath) + + defer store.Close() + + writeJSONSession(t, jsonDir, Session{ + Key: "nosummary:1", + + Messages: []providers.Message{{Role: "user", Content: "hi"}}, + }) + + MigrateJSONSessions(jsonDir, store) + + info, _ := store.Get("nosummary:1") + + if info.Summary != "" { + t.Errorf("expected empty summary, got %q", info.Summary) + } +} + +func TestMigrate_InvalidJSON(t *testing.T) { + jsonDir := t.TempDir() + + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, _ := OpenSQLiteStore(dbPath) + + defer store.Close() + + // Write invalid JSON + + os.WriteFile(filepath.Join(jsonDir, "bad.json"), []byte("{invalid"), 0o644) + + // Write a valid one too + + writeJSONSession(t, jsonDir, Session{ + Key: "good:1", + + Messages: []providers.Message{{Role: "user", Content: "hi"}}, + }) + + migrated, err := MigrateJSONSessions(jsonDir, store) + if err != nil { + t.Fatal(err) + } + + if migrated != 1 { + t.Errorf("expected 1 (skipped bad), got %d", migrated) + } + + // Bad file should still be .json (not renamed) + + if _, err := os.Stat(filepath.Join(jsonDir, "bad.json")); os.IsNotExist(err) { + t.Error("bad.json should still exist") + } +} + +func TestMigrate_Idempotent(t *testing.T) { + jsonDir := t.TempDir() + + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, _ := OpenSQLiteStore(dbPath) + + defer store.Close() + + writeJSONSession(t, jsonDir, Session{ + Key: "k1", + + Messages: []providers.Message{{Role: "user", Content: "hello"}}, + }) + + n1, _ := MigrateJSONSessions(jsonDir, store) + + if n1 != 1 { + t.Fatalf("first run: expected 1, got %d", n1) + } + + // Second run should find no .json files (all renamed) + + n2, _ := MigrateJSONSessions(jsonDir, store) + + if n2 != 0 { + t.Errorf("second run: expected 0, got %d", n2) + } +} + +func TestMigrate_NonExistentDir(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, _ := OpenSQLiteStore(dbPath) + + defer store.Close() + + n, err := MigrateJSONSessions("/nonexistent/path", store) + if err != nil { + t.Fatalf("expected nil error for non-existent dir, got %v", err) + } + + if n != 0 { + t.Errorf("expected 0, got %d", n) + } +} + +func TestMigrate_AlreadyExistsInStore(t *testing.T) { + jsonDir := t.TempDir() + + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, _ := OpenSQLiteStore(dbPath) + + defer store.Close() + + // Pre-create session in store + + store.Create("k1", nil) + + // Write JSON for same key + + writeJSONSession(t, jsonDir, Session{ + Key: "k1", + + Messages: []providers.Message{{Role: "user", Content: "hello"}}, + }) + + n, _ := MigrateJSONSessions(jsonDir, store) + + if n != 1 { + t.Errorf("expected 1, got %d", n) + } + + // File should still be renamed + + entries, _ := os.ReadDir(jsonDir) + + for _, e := range entries { + if filepath.Ext(e.Name()) == ".json" { + t.Errorf("expected .json.migrated, found %s", e.Name()) + } + } +} diff --git a/pkg/session/sqlite.go b/pkg/session/sqlite.go new file mode 100644 index 000000000..9af6ce909 --- /dev/null +++ b/pkg/session/sqlite.go @@ -0,0 +1,497 @@ +package session + +import ( + "database/sql" + "encoding/json" + "fmt" + "time" + + "github.com/google/uuid" + _ "modernc.org/sqlite" +) + +const sqliteDriver = "sqlite" + +const schema = ` + +CREATE TABLE IF NOT EXISTS sessions ( + + key TEXT PRIMARY KEY, + + parent_key TEXT NOT NULL DEFAULT '', + + fork_turn_id TEXT NOT NULL DEFAULT '', + + status TEXT NOT NULL DEFAULT 'active', + + label TEXT NOT NULL DEFAULT '', + + summary TEXT NOT NULL DEFAULT '', + + created_at TEXT NOT NULL, + + updated_at TEXT NOT NULL + +); + +CREATE TABLE IF NOT EXISTS turns ( + + id TEXT PRIMARY KEY, + + session_key TEXT NOT NULL REFERENCES sessions(key) ON DELETE CASCADE, + + seq INTEGER NOT NULL, + + kind INTEGER NOT NULL DEFAULT 0, + + messages TEXT NOT NULL DEFAULT '[]', + + origin_key TEXT NOT NULL DEFAULT '', + + summary TEXT NOT NULL DEFAULT '', + + author TEXT NOT NULL DEFAULT '', + + created_at TEXT NOT NULL, + + meta TEXT NOT NULL DEFAULT '{}' + +); + +CREATE INDEX IF NOT EXISTS idx_turns_session_seq ON turns(session_key, seq); + +CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_key); + +` + +// SQLiteStore implements SessionStore backed by a single SQLite file. + +type SQLiteStore struct { + db *sql.DB +} + +// OpenSQLiteStore opens (or creates) a SQLite session database at dbPath. + +func OpenSQLiteStore(dbPath string) (*SQLiteStore, error) { + connStr := "file:" + dbPath + "?_journal_mode=WAL&_foreign_keys=on&_busy_timeout=5000" + + db, err := sql.Open(sqliteDriver, connStr) + if err != nil { + return nil, fmt.Errorf("open session store: %w", err) + } + + db.SetMaxOpenConns(1) + + db.SetMaxIdleConns(1) + + // Ensure foreign keys are enabled (connection string param may not suffice). + + if _, err := db.Exec("PRAGMA foreign_keys = ON"); err != nil { + _ = db.Close() + + return nil, fmt.Errorf("enable foreign keys: %w", err) + } + + if _, err := db.Exec(schema); err != nil { + _ = db.Close() + + return nil, fmt.Errorf("create schema: %w", err) + } + + return &SQLiteStore{db: db}, nil +} + +func nowUTC() string { + return time.Now().UTC().Format(time.RFC3339Nano) +} + +func parseTime(s string) time.Time { + t, _ := time.Parse(time.RFC3339Nano, s) + + return t +} + +// --- Session CRUD --- + +func (s *SQLiteStore) Create(key string, opts *CreateOpts) error { + now := nowUTC() + + parentKey, forkTurnID, label := "", "", "" + + if opts != nil { + parentKey = opts.ParentKey + + forkTurnID = opts.ForkTurnID + + label = opts.Label + } + + _, err := s.db.Exec( + + `INSERT INTO sessions (key, parent_key, fork_turn_id, label, created_at, updated_at) + + VALUES (?, ?, ?, ?, ?, ?)`, + + key, parentKey, forkTurnID, label, now, now, + ) + + return err +} + +func (s *SQLiteStore) Get(key string) (*SessionInfo, error) { + row := s.db.QueryRow( + + `SELECT key, parent_key, fork_turn_id, status, label, summary, created_at, updated_at + + FROM sessions WHERE key = ?`, key, + ) + + return scanSessionInfo(row) +} + +func scanSessionInfo(row *sql.Row) (*SessionInfo, error) { + var info SessionInfo + + var createdAt, updatedAt string + + err := row.Scan(&info.Key, &info.ParentKey, &info.ForkTurnID, &info.Status, + + &info.Label, &info.Summary, &createdAt, &updatedAt) + + if err == sql.ErrNoRows { + return nil, nil + } + + if err != nil { + return nil, err + } + + info.CreatedAt = parseTime(createdAt) + + info.UpdatedAt = parseTime(updatedAt) + + return &info, nil +} + +func (s *SQLiteStore) List(filter *ListFilter) ([]*SessionInfo, error) { + query := `SELECT key, parent_key, fork_turn_id, status, label, summary, created_at, updated_at FROM sessions WHERE 1=1` + + var args []any + + if filter != nil { + if filter.ParentKey != "" { + query += ` AND parent_key = ?` + + args = append(args, filter.ParentKey) + } + + if filter.Status != "" { + query += ` AND status = ?` + + args = append(args, filter.Status) + } + } + + query += ` ORDER BY created_at` + + rows, err := s.db.Query(query, args...) + if err != nil { + return nil, err + } + + defer rows.Close() + + var result []*SessionInfo + + for rows.Next() { + var info SessionInfo + + var createdAt, updatedAt string + + if err := rows.Scan(&info.Key, &info.ParentKey, &info.ForkTurnID, &info.Status, + + &info.Label, &info.Summary, &createdAt, &updatedAt); err != nil { + return nil, err + } + + info.CreatedAt = parseTime(createdAt) + + info.UpdatedAt = parseTime(updatedAt) + + result = append(result, &info) + } + + return result, rows.Err() +} + +func (s *SQLiteStore) SetStatus(key, status string) error { + res, err := s.db.Exec(`UPDATE sessions SET status = ?, updated_at = ? WHERE key = ?`, status, nowUTC(), key) + if err != nil { + return err + } + + return checkRowAffected(res, key) +} + +func (s *SQLiteStore) SetSummary(key, summary string) error { + res, err := s.db.Exec(`UPDATE sessions SET summary = ?, updated_at = ? WHERE key = ?`, summary, nowUTC(), key) + if err != nil { + return err + } + + return checkRowAffected(res, key) +} + +func (s *SQLiteStore) Delete(key string) error { + _, err := s.db.Exec(`DELETE FROM sessions WHERE key = ?`, key) + + return err +} + +func (s *SQLiteStore) Children(key string) ([]*SessionInfo, error) { + return s.List(&ListFilter{ParentKey: key}) +} + +// --- Turn operations --- + +func (s *SQLiteStore) Append(sessionKey string, turn *Turn) error { + if turn.ID == "" { + turn.ID = uuid.New().String() + } + + if turn.CreatedAt.IsZero() { + turn.CreatedAt = time.Now().UTC() + } + + messagesJSON, err := json.Marshal(turn.Messages) + if err != nil { + return fmt.Errorf("marshal messages: %w", err) + } + + metaJSON, err := json.Marshal(turn.Meta) + if err != nil { + return fmt.Errorf("marshal meta: %w", err) + } + + // Auto-assign seq if not set + + if turn.Seq == 0 { + var maxSeq sql.NullInt64 + + _ = s.db.QueryRow(`SELECT MAX(seq) FROM turns WHERE session_key = ?`, sessionKey).Scan(&maxSeq) + + if maxSeq.Valid { + turn.Seq = int(maxSeq.Int64) + 1 + } else { + turn.Seq = 1 + } + } + + _, err = s.db.Exec( + + `INSERT INTO turns (id, session_key, seq, kind, messages, origin_key, summary, author, created_at, meta) + + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + + turn.ID, sessionKey, turn.Seq, int(turn.Kind), + + string(messagesJSON), turn.OriginKey, turn.Summary, turn.Author, + + turn.CreatedAt.UTC().Format(time.RFC3339Nano), string(metaJSON), + ) + if err != nil { + return err + } + + // Update session's updated_at + + _, _ = s.db.Exec(`UPDATE sessions SET updated_at = ? WHERE key = ?`, nowUTC(), sessionKey) + + return nil +} + +func (s *SQLiteStore) Turns(sessionKey string, sinceSeq int) ([]*Turn, error) { + rows, err := s.db.Query( + + `SELECT id, session_key, seq, kind, messages, origin_key, summary, author, created_at, meta + + FROM turns WHERE session_key = ? AND seq > ? ORDER BY seq`, + + sessionKey, sinceSeq, + ) + if err != nil { + return nil, err + } + + defer rows.Close() + + var result []*Turn + + for rows.Next() { + t, err := scanTurn(rows) + if err != nil { + return nil, err + } + + result = append(result, t) + } + + return result, rows.Err() +} + +type scanner interface { + Scan(dest ...any) error +} + +func scanTurn(row scanner) (*Turn, error) { + var t Turn + + var kind int + + var messagesJSON, metaJSON, createdAt string + + err := row.Scan(&t.ID, &t.SessionKey, &t.Seq, &kind, &messagesJSON, + + &t.OriginKey, &t.Summary, &t.Author, &createdAt, &metaJSON) + if err != nil { + return nil, err + } + + t.Kind = TurnKind(kind) + + t.CreatedAt = parseTime(createdAt) + + if err := json.Unmarshal([]byte(messagesJSON), &t.Messages); err != nil { + return nil, fmt.Errorf("unmarshal messages: %w", err) + } + + if metaJSON != "" && metaJSON != "{}" { + if err := json.Unmarshal([]byte(metaJSON), &t.Meta); err != nil { + return nil, fmt.Errorf("unmarshal meta: %w", err) + } + } + + return &t, nil +} + +func (s *SQLiteStore) LastTurn(sessionKey string) (*Turn, error) { + row := s.db.QueryRow( + + `SELECT id, session_key, seq, kind, messages, origin_key, summary, author, created_at, meta + + FROM turns WHERE session_key = ? ORDER BY seq DESC LIMIT 1`, + + sessionKey, + ) + + var t Turn + + var kind int + + var messagesJSON, metaJSON, createdAt string + + err := row.Scan(&t.ID, &t.SessionKey, &t.Seq, &kind, &messagesJSON, + + &t.OriginKey, &t.Summary, &t.Author, &createdAt, &metaJSON) + + if err == sql.ErrNoRows { + return nil, nil + } + + if err != nil { + return nil, err + } + + t.Kind = TurnKind(kind) + + t.CreatedAt = parseTime(createdAt) + + if err := json.Unmarshal([]byte(messagesJSON), &t.Messages); err != nil { + return nil, fmt.Errorf("unmarshal messages: %w", err) + } + + if metaJSON != "" && metaJSON != "{}" { + if err := json.Unmarshal([]byte(metaJSON), &t.Meta); err != nil { + return nil, fmt.Errorf("unmarshal meta: %w", err) + } + } + + return &t, nil +} + +func (s *SQLiteStore) TurnCount(sessionKey string) (int, error) { + var count int + + err := s.db.QueryRow(`SELECT COUNT(*) FROM turns WHERE session_key = ?`, sessionKey).Scan(&count) + + return count, err +} + +func (s *SQLiteStore) Compact(sessionKey string, upToSeq int, summary string) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + + defer tx.Rollback() + + _, err = tx.Exec(`DELETE FROM turns WHERE session_key = ? AND seq <= ?`, sessionKey, upToSeq) + if err != nil { + return err + } + + if summary != "" { + _, err = tx.Exec(`UPDATE sessions SET summary = ?, updated_at = ? WHERE key = ?`, + + summary, nowUTC(), sessionKey) + if err != nil { + return err + } + } + + return tx.Commit() +} + +// --- DAG operations --- + +func (s *SQLiteStore) Fork(parentKey, childKey string, opts *CreateOpts) error { + if opts == nil { + opts = &CreateOpts{} + } + + opts.ParentKey = parentKey + + return s.Create(childKey, opts) +} + +// --- Maintenance --- + +func (s *SQLiteStore) Prune(olderThan time.Duration) (int, error) { + cutoff := time.Now().UTC().Add(-olderThan).Format(time.RFC3339Nano) + + res, err := s.db.Exec(`DELETE FROM sessions WHERE updated_at < ?`, cutoff) + if err != nil { + return 0, err + } + + n, _ := res.RowsAffected() + + return int(n), nil +} + +func (s *SQLiteStore) Close() error { + return s.db.Close() +} + +func checkRowAffected(res sql.Result, key string) error { + n, err := res.RowsAffected() + if err != nil { + return err + } + + if n == 0 { + return fmt.Errorf("session not found: %s", key) + } + + return nil +} diff --git a/pkg/session/sqlite_test.go b/pkg/session/sqlite_test.go new file mode 100644 index 000000000..33e573e67 --- /dev/null +++ b/pkg/session/sqlite_test.go @@ -0,0 +1,481 @@ +package session + +import ( + "path/filepath" + "testing" + "time" + + "github.com/sipeed/picoclaw/pkg/providers" +) + +func newTestStore(t *testing.T) *SQLiteStore { + t.Helper() + + dbPath := filepath.Join(t.TempDir(), "test.db") + + store, err := OpenSQLiteStore(dbPath) + if err != nil { + t.Fatalf("OpenSQLiteStore: %v", err) + } + + t.Cleanup(func() { store.Close() }) + + return store +} + +func TestSQLite_CreateGetDelete(t *testing.T) { + store := newTestStore(t) + + // Create + + if err := store.Create("s1", nil); err != nil { + t.Fatalf("Create: %v", err) + } + + // Get + + info, err := store.Get("s1") + if err != nil { + t.Fatalf("Get: %v", err) + } + + if info == nil { + t.Fatal("expected session, got nil") + } + + if info.Key != "s1" || info.Status != "active" { + t.Errorf("unexpected session: %+v", info) + } + + // Get non-existent + + info, err = store.Get("nope") + if err != nil { + t.Fatalf("Get non-existent: %v", err) + } + + if info != nil { + t.Errorf("expected nil for non-existent session") + } + + // Delete + + if delErr := store.Delete("s1"); delErr != nil { + t.Fatalf("Delete: %v", delErr) + } + + info, err = store.Get("s1") + if err != nil { + t.Fatalf("Get after delete: %v", err) + } + + if info != nil { + t.Errorf("expected nil after delete") + } +} + +func TestSQLite_CreateWithOpts(t *testing.T) { + store := newTestStore(t) + + if err := store.Create("parent", nil); err != nil { + t.Fatalf("Create parent: %v", err) + } + + if err := store.Create("child", &CreateOpts{ + ParentKey: "parent", + + ForkTurnID: "turn-1", + + Label: "test child", + }); err != nil { + t.Fatalf("Create child: %v", err) + } + + info, _ := store.Get("child") + + if info.ParentKey != "parent" || info.ForkTurnID != "turn-1" || info.Label != "test child" { + t.Errorf("unexpected opts: %+v", info) + } +} + +func TestSQLite_List(t *testing.T) { + store := newTestStore(t) + + store.Create("a", nil) + + store.Create("b", &CreateOpts{ParentKey: "a"}) + + store.Create("c", nil) + + all, _ := store.List(nil) + + if len(all) != 3 { + t.Fatalf("expected 3 sessions, got %d", len(all)) + } + + children, _ := store.List(&ListFilter{ParentKey: "a"}) + + if len(children) != 1 || children[0].Key != "b" { + t.Errorf("unexpected children: %+v", children) + } + + store.SetStatus("c", "archived") + + active, _ := store.List(&ListFilter{Status: "active"}) + + if len(active) != 2 { + t.Errorf("expected 2 active, got %d", len(active)) + } +} + +func TestSQLite_SetStatusSummary(t *testing.T) { + store := newTestStore(t) + + store.Create("s1", nil) + + if err := store.SetStatus("s1", "archived"); err != nil { + t.Fatalf("SetStatus: %v", err) + } + + info, _ := store.Get("s1") + + if info.Status != "archived" { + t.Errorf("expected archived, got %s", info.Status) + } + + if err := store.SetSummary("s1", "test summary"); err != nil { + t.Fatalf("SetSummary: %v", err) + } + + info, _ = store.Get("s1") + + if info.Summary != "test summary" { + t.Errorf("expected 'test summary', got %q", info.Summary) + } + + // Non-existent session + + if err := store.SetStatus("nope", "active"); err == nil { + t.Error("expected error for non-existent session") + } +} + +func TestSQLite_Children(t *testing.T) { + store := newTestStore(t) + + store.Create("p", nil) + + store.Create("c1", &CreateOpts{ParentKey: "p"}) + + store.Create("c2", &CreateOpts{ParentKey: "p"}) + + store.Create("other", nil) + + children, err := store.Children("p") + if err != nil { + t.Fatalf("Children: %v", err) + } + + if len(children) != 2 { + t.Errorf("expected 2 children, got %d", len(children)) + } +} + +func TestSQLite_AppendAndTurns(t *testing.T) { + store := newTestStore(t) + + store.Create("s1", nil) + + turn1 := &Turn{ + SessionKey: "s1", + + Kind: TurnNormal, + + Messages: []providers.Message{{Role: "user", Content: "hello"}}, + + Author: "user", + } + + if err := store.Append("s1", turn1); err != nil { + t.Fatalf("Append: %v", err) + } + + if turn1.ID == "" { + t.Error("expected ID to be assigned") + } + + if turn1.Seq != 1 { + t.Errorf("expected seq 1, got %d", turn1.Seq) + } + + turn2 := &Turn{ + SessionKey: "s1", + + Kind: TurnNormal, + + Messages: []providers.Message{{Role: "assistant", Content: "hi"}}, + + Author: "assistant", + } + + store.Append("s1", turn2) + + if turn2.Seq != 2 { + t.Errorf("expected seq 2, got %d", turn2.Seq) + } + + // Get all turns + + turns, err := store.Turns("s1", 0) + if err != nil { + t.Fatalf("Turns: %v", err) + } + + if len(turns) != 2 { + t.Fatalf("expected 2 turns, got %d", len(turns)) + } + + // sinceSeq filter + + turns, _ = store.Turns("s1", 1) + + if len(turns) != 1 || turns[0].Seq != 2 { + t.Errorf("expected 1 turn with seq 2, got %+v", turns) + } +} + +func TestSQLite_LastTurn(t *testing.T) { + store := newTestStore(t) + + store.Create("s1", nil) + + // No turns + + last, err := store.LastTurn("s1") + if err != nil { + t.Fatalf("LastTurn empty: %v", err) + } + + if last != nil { + t.Error("expected nil for empty session") + } + + store.Append("s1", &Turn{Messages: []providers.Message{{Role: "user", Content: "a"}}}) + + store.Append("s1", &Turn{Messages: []providers.Message{{Role: "user", Content: "b"}}}) + + last, _ = store.LastTurn("s1") + + if last == nil || last.Messages[0].Content != "b" { + t.Errorf("expected last message 'b', got %+v", last) + } +} + +func TestSQLite_TurnCount(t *testing.T) { + store := newTestStore(t) + + store.Create("s1", nil) + + count, _ := store.TurnCount("s1") + + if count != 0 { + t.Errorf("expected 0, got %d", count) + } + + store.Append("s1", &Turn{Messages: []providers.Message{{Role: "user", Content: "a"}}}) + + store.Append("s1", &Turn{Messages: []providers.Message{{Role: "user", Content: "b"}}}) + + count, _ = store.TurnCount("s1") + + if count != 2 { + t.Errorf("expected 2, got %d", count) + } +} + +func TestSQLite_Compact(t *testing.T) { + store := newTestStore(t) + + store.Create("s1", nil) + + for i := range 5 { + store.Append("s1", &Turn{ + Messages: []providers.Message{{Role: "user", Content: string(rune('a' + i))}}, + }) + } + + // Compact up to seq 3 + + if err := store.Compact("s1", 3, "summary of first 3 turns"); err != nil { + t.Fatalf("Compact: %v", err) + } + + turns, _ := store.Turns("s1", 0) + + if len(turns) != 2 { + t.Errorf("expected 2 remaining turns, got %d", len(turns)) + } + + if turns[0].Seq != 4 { + t.Errorf("expected first remaining seq 4, got %d", turns[0].Seq) + } + + info, _ := store.Get("s1") + + if info.Summary != "summary of first 3 turns" { + t.Errorf("expected compacted summary, got %q", info.Summary) + } +} + +func TestSQLite_Fork(t *testing.T) { + store := newTestStore(t) + + store.Create("parent", nil) + + store.Append("parent", &Turn{ + Messages: []providers.Message{{Role: "user", Content: "hello"}}, + }) + + last, _ := store.LastTurn("parent") + + if err := store.Fork("parent", "child", &CreateOpts{ForkTurnID: last.ID}); err != nil { + t.Fatalf("Fork: %v", err) + } + + child, _ := store.Get("child") + + if child.ParentKey != "parent" || child.ForkTurnID != last.ID { + t.Errorf("unexpected fork result: %+v", child) + } + + children, _ := store.Children("parent") + + if len(children) != 1 || children[0].Key != "child" { + t.Errorf("expected 1 child, got %+v", children) + } +} + +func TestSQLite_Prune(t *testing.T) { + store := newTestStore(t) + + // Create an old session by manipulating updated_at directly + + store.Create("old", nil) + + store.Create("new", nil) + + old := time.Now().UTC().Add(-48 * time.Hour).Format(time.RFC3339Nano) + + store.db.Exec(`UPDATE sessions SET updated_at = ? WHERE key = ?`, old, "old") + + pruned, err := store.Prune(24 * time.Hour) + if err != nil { + t.Fatalf("Prune: %v", err) + } + + if pruned != 1 { + t.Errorf("expected 1 pruned, got %d", pruned) + } + + info, _ := store.Get("old") + + if info != nil { + t.Error("expected old session to be pruned") + } + + info, _ = store.Get("new") + + if info == nil { + t.Error("expected new session to survive prune") + } +} + +func TestSQLite_MessagesRoundTrip(t *testing.T) { + store := newTestStore(t) + + store.Create("s1", nil) + + msgs := []providers.Message{ + {Role: "user", Content: "hello"}, + + { + Role: "assistant", + + Content: "sure", + + ToolCalls: []providers.ToolCall{ + { + ID: "call_1", + + Type: "function", + + Function: &providers.FunctionCall{ + Name: "exec", + + Arguments: `{"cmd":"ls"}`, + }, + }, + }, + }, + + {Role: "tool", Content: "file1\nfile2", ToolCallID: "call_1"}, + + {Role: "assistant", Content: "done"}, + } + + store.Append("s1", &Turn{Messages: msgs}) + + turns, _ := store.Turns("s1", 0) + + if len(turns) != 1 { + t.Fatalf("expected 1 turn, got %d", len(turns)) + } + + got := turns[0].Messages + + if len(got) != 4 { + t.Fatalf("expected 4 messages, got %d", len(got)) + } + + // Check tool call round-trip + + if got[1].ToolCalls[0].ID != "call_1" { + t.Errorf("tool call ID mismatch: %s", got[1].ToolCalls[0].ID) + } + + if got[1].ToolCalls[0].Function.Name != "exec" { + t.Errorf("tool call function name mismatch: %s", got[1].ToolCalls[0].Function.Name) + } + + if got[1].ToolCalls[0].Function.Arguments != `{"cmd":"ls"}` { + t.Errorf("tool call arguments mismatch: %s", got[1].ToolCalls[0].Function.Arguments) + } + + if got[2].ToolCallID != "call_1" { + t.Errorf("tool call ID mismatch on result: %s", got[2].ToolCallID) + } +} + +func TestSQLite_CascadeDelete(t *testing.T) { + store := newTestStore(t) + + store.Create("s1", nil) + + store.Append("s1", &Turn{Messages: []providers.Message{{Role: "user", Content: "a"}}}) + + store.Append("s1", &Turn{Messages: []providers.Message{{Role: "user", Content: "b"}}}) + + count, _ := store.TurnCount("s1") + + if count != 2 { + t.Fatalf("expected 2 turns before delete, got %d", count) + } + + store.Delete("s1") + + count, _ = store.TurnCount("s1") + + if count != 0 { + t.Errorf("expected 0 turns after cascade delete, got %d", count) + } +} diff --git a/pkg/session/store.go b/pkg/session/store.go new file mode 100644 index 000000000..244deedd2 --- /dev/null +++ b/pkg/session/store.go @@ -0,0 +1,31 @@ +package session + +import "time" + +// SessionStore is the storage interface for sessions and turns. +// Phase 0 provides a SQLite implementation; LegacyAdapter wraps it to +// expose the same API as SessionManager. +type SessionStore interface { //nolint:interfacebloat // storage facade — methods are logically grouped + // Session CRUD + Create(key string, opts *CreateOpts) error + Get(key string) (*SessionInfo, error) + List(filter *ListFilter) ([]*SessionInfo, error) + SetStatus(key, status string) error + SetSummary(key, summary string) error + Delete(key string) error + Children(key string) ([]*SessionInfo, error) + + // Turn operations + Append(sessionKey string, turn *Turn) error + Turns(sessionKey string, sinceSeq int) ([]*Turn, error) + LastTurn(sessionKey string) (*Turn, error) + TurnCount(sessionKey string) (int, error) + Compact(sessionKey string, upToSeq int, summary string) error + + // DAG operations + Fork(parentKey, childKey string, opts *CreateOpts) error + + // Maintenance + Prune(olderThan time.Duration) (int, error) + Close() error +} diff --git a/pkg/session/types.go b/pkg/session/types.go new file mode 100644 index 000000000..32ecdcf39 --- /dev/null +++ b/pkg/session/types.go @@ -0,0 +1,84 @@ +package session + +import ( + "time" + + "github.com/sipeed/picoclaw/pkg/providers" +) + +// TurnKind classifies a turn within a session. + +type TurnKind int + +const ( + TurnNormal TurnKind = iota // Regular conversation turn + + TurnReport // Subagent report turn + + TurnForkPoint // Fork point for child sessions + +) + +// Turn represents a single conversation turn persisted in the store. + +type Turn struct { + ID string + + SessionKey string + + OriginKey string + + Summary string + + Author string + + Seq int + + Kind TurnKind + + Messages []providers.Message + + CreatedAt time.Time + + Meta map[string]string +} + +// SessionInfo holds metadata about a session. + +type SessionInfo struct { + Key string + + ParentKey string + + ForkTurnID string + + Status string + + Label string + + Summary string + + TurnCount int + + CreatedAt time.Time + + UpdatedAt time.Time +} + +// CreateOpts are options for creating a new session. + +type CreateOpts struct { + ParentKey string + + ForkTurnID string + + Label string +} + +// ListFilter constrains which sessions are returned by List. + +type ListFilter struct { + ParentKey string + + Status string +} diff --git a/todo/TASKS-3.md b/todo/TASKS-3.md index 24d508753..c87060af9 100644 --- a/todo/TASKS-3.md +++ b/todo/TASKS-3.md @@ -147,28 +147,27 @@ func (tw *TurnWriter) Discard() ## タスク一覧 -### Phase 0: SQLite SessionStore + LegacyAdapter +### Phase 0: SQLite SessionStore + LegacyAdapter ✅ 既存動作を維持したまま裏側を差し替える。 -1. **SQLite SessionStore 実装** - - `pkg/session/sqlite.go` (新規): SessionStore interface の SQLite 実装 +1. ✅ **SQLite SessionStore 実装** + - `pkg/session/sqlite.go`: SessionStore interface の SQLite 実装 - WAL モード、`modernc.org/sqlite` (CGO なし、ARM クロスコンパイル容易) - schema migration (CREATE TABLE IF NOT EXISTS) - - `go:build` タグなしで常に利用可能 -2. **LegacyAdapter 実装** - - `pkg/session/legacy_adapter.go` (新規) - - 既存の `GetHistory` / `SetHistory` / `AddMessage` / `MarkDirty` を SessionGraph 経由で実装 - - 既存テストが全パスすること +2. ✅ **LegacyAdapter 実装** + - `pkg/session/legacy_adapter.go` + - 既存の `GetHistory` / `SetHistory` / `AddMessage` / `MarkDirty` を SessionStore 経由で実装 + - 既存テスト全パス + テーブル駆動で JSON/SQLite 両方を同一アサーションで検証 -3. **JSON → SQLite lazy migration** - - 起動時に `sessions/*.json` を検出 → SQLite に import → JSON ファイルをリネーム (.migrated) - - エラー時は JSON にフォールバック +3. ✅ **JSON → SQLite lazy migration** + - `pkg/session/migrate.go`: 起動時に `sessions/*.json` を検出 → SQLite に import → `.json.migrated` にリネーム + - 個別ファイル失敗はログ警告して続行 (次回起動で再試行) -4. **AgentLoop 配線** - - `AgentInstance` が `SessionStore` を保持、`LegacyAdapter` 経由で既存コードに注入 - - 既存の `SessionManager` は Phase 2 で廃止 +4. ✅ **AgentLoop 配線** + - `pkg/agent/instance.go`: `Sessions` 型を `*LegacyAdapter` に変更 + - `loop.go` / `loop_test.go` は変更なし — メソッドシグネチャ同一 ---