opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(session): implement SQLite SessionStore + LegacyAdapter (TASKS-3 Phase 0)

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
dj-oyu 2026-03-04 21:55:13 +09:00
parent e6de7d0d62
commit af5ed21f5b
11 changed files with 2689 additions and 58 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
CLAUDE.md
View file

@ -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` 参照
## コードの匂い — チェックリスト
新しいコードを書くとき・レビューするときの確認事項:

239
pkg/agent/instance.go
View file

@ -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: <workspace>/.worktrees/<branch-basename>/
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
}

496
pkg/session/legacy_adapter.go Normal file
View file

@ -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
}
}
}

484
pkg/session/legacy_adapter_test.go Normal file
View file

@ -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))
}
})
}

123
pkg/session/migrate.go Normal file
View file

@ -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
}

276
pkg/session/migrate_test.go Normal file
View file

@ -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())
}
}
}

497
pkg/session/sqlite.go Normal file
View file

@ -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
}

481
pkg/session/sqlite_test.go Normal file
View file

@ -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)
}
}

31
pkg/session/store.go Normal file
View file

@ -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
}

84
pkg/session/types.go Normal file
View file

@ -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
}

27
todo/TASKS-3.md
View file

@ -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` は変更なし — メソッドシグネチャ同一
---