feat: SQLite session store and tool-calling agent loop

internal/agent/agent.go

@@ -0,0 +1,121 @@
+// Package agent implements the LLM tool-calling loop.
+package agent
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/crypto/openclaw/internal/llm"
+	"github.com/crypto/openclaw/internal/store"
+	"github.com/crypto/openclaw/internal/tools"
+)
+
+// Event is streamed to clients while the agent works.
+type Event struct {
+	Type    string `json:"type"` // delta | tool_call | tool_result | done | error
+	Text    string `json:"text,omitempty"`
+	Tool    string `json:"tool,omitempty"`
+	Args    string `json:"args,omitempty"`
+	CallID  string `json:"call_id,omitempty"`
+	Session string `json:"session_id,omitempty"`
+}
+
+type Agent struct {
+	LLM           llm.Provider
+	Tools         *tools.Registry
+	Store         *store.Store
+	SystemPrompt  string
+	MaxIterations int
+	HistoryLimit  int
+
+	mu    sync.Mutex
+	locks map[string]*sync.Mutex
+}
+
+func (a *Agent) sessionLock(id string) *sync.Mutex {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	if a.locks == nil {
+		a.locks = map[string]*sync.Mutex{}
+	}
+	l, ok := a.locks[id]
+	if !ok {
+		l = &sync.Mutex{}
+		a.locks[id] = l
+	}
+	return l
+}
+
+// Run appends the user message to the session and loops until the model
+// answers without tool calls. Returns the final assistant text.
+func (a *Agent) Run(ctx context.Context, sessionID, userText string, emit func(Event)) (string, error) {
+	if emit == nil {
+		emit = func(Event) {}
+	}
+	l := a.sessionLock(sessionID)
+	l.Lock()
+	defer l.Unlock()
+
+	if _, err := a.Store.AppendMessage(ctx, sessionID, llm.Message{Role: llm.RoleUser, Content: userText}); err != nil {
+		return "", err
+	}
+	var specs []llm.ToolSpec
+	if a.Tools != nil {
+		specs = a.Tools.Specs()
+	}
+	for i := 0; i < a.MaxIterations; i++ {
+		msgs, err := a.history(ctx, sessionID)
+		if err != nil {
+			return "", err
+		}
+		resp, err := a.LLM.Chat(ctx, llm.Request{Messages: msgs, Tools: specs}, func(t string) {
+			emit(Event{Type: "delta", Text: t})
+		})
+		if err != nil {
+			return "", err
+		}
+		asst := llm.Message{Role: llm.RoleAssistant, Content: resp.Content, ToolCalls: resp.ToolCalls}
+		if _, err := a.Store.AppendMessage(ctx, sessionID, asst); err != nil {
+			return "", err
+		}
+		if len(resp.ToolCalls) == 0 {
+			return resp.Content, nil
+		}
+		for _, tc := range resp.ToolCalls {
+			emit(Event{Type: "tool_call", Tool: tc.Name, Args: tc.Arguments, CallID: tc.ID})
+			out := a.Tools.Call(ctx, tc.Name, tc.Arguments)
+			emit(Event{Type: "tool_result", Tool: tc.Name, Text: clip(out, 2000), CallID: tc.ID})
+			if _, err := a.Store.AppendMessage(ctx, sessionID, llm.Message{Role: llm.RoleTool, ToolCallID: tc.ID, Name: tc.Name, Content: out}); err != nil {
+				return "", err
+			}
+		}
+	}
+	return "", fmt.Errorf("stopped after %d iterations without a final answer", a.MaxIterations)
+}
+
+func (a *Agent) history(ctx context.Context, sessionID string) ([]llm.Message, error) {
+	stored, err := a.Store.Messages(ctx, sessionID, a.HistoryLimit)
+	if err != nil {
+		return nil, err
+	}
+	// A truncated window must not start with orphaned tool results.
+	for len(stored) > 0 && stored[0].Role == llm.RoleTool {
+		stored = stored[1:]
+	}
+	msgs := make([]llm.Message, 0, len(stored)+1)
+	if a.SystemPrompt != "" {
+		msgs = append(msgs, llm.Message{Role: llm.RoleSystem, Content: a.SystemPrompt})
+	}
+	for _, m := range stored {
+		msgs = append(msgs, m.Message)
+	}
+	return msgs, nil
+}
+
+func clip(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	return s[:n] + "…"
+}

internal/agent/agent_test.go

@@ -0,0 +1,93 @@
+package agent
+
+import (
+	"context"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/crypto/openclaw/internal/llm"
+	"github.com/crypto/openclaw/internal/store"
+	"github.com/crypto/openclaw/internal/tools"
+)
+
+// scripted provider returns queued responses and records requests.
+type scripted struct {
+	resps []llm.Response
+	reqs  []llm.Request
+}
+
+func (s *scripted) Chat(_ context.Context, req llm.Request, d llm.DeltaFunc) (llm.Response, error) {
+	s.reqs = append(s.reqs, req)
+	r := s.resps[0]
+	s.resps = s.resps[1:]
+	if d != nil && r.Content != "" {
+		d(r.Content)
+	}
+	return r, nil
+}
+
+func newStore(t *testing.T) *store.Store {
+	s, err := store.Open(filepath.Join(t.TempDir(), "a.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { s.Close() })
+	return s
+}
+
+func TestToolLoop(t *testing.T) {
+	st := newStore(t)
+	ss, _ := st.CreateSession(context.Background(), "")
+	p := &scripted{resps: []llm.Response{
+		{ToolCalls: []llm.ToolCall{{ID: "c1", Name: "time_now", Arguments: `{"timezone":"UTC"}`}}},
+		{Content: "It is noon."},
+	}}
+	fixed := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC)
+	a := &Agent{LLM: p, Store: st, Tools: tools.NewRegistry(tools.TimeNow{Now: func() time.Time { return fixed }}), SystemPrompt: "sys", MaxIterations: 4}
+	var events []Event
+	out, err := a.Run(context.Background(), ss.ID, "what time?", func(e Event) { events = append(events, e) })
+	if err != nil || out != "It is noon." {
+		t.Fatalf("%q %v", out, err)
+	}
+	second := p.reqs[1].Messages
+	if second[0].Role != llm.RoleSystem || second[len(second)-1].Role != llm.RoleTool || !strings.HasPrefix(second[len(second)-1].Content, "2026-05-05T12:00:00") {
+		t.Fatalf("bad second request: %+v", second)
+	}
+	types := []string{}
+	for _, e := range events {
+		types = append(types, e.Type)
+	}
+	if strings.Join(types, ",") != "tool_call,tool_result,delta" {
+		t.Fatalf("events %v", types)
+	}
+	msgs, _ := st.Messages(context.Background(), ss.ID, 0)
+	if len(msgs) != 4 {
+		t.Fatalf("stored %d msgs", len(msgs))
+	}
+}
+
+func TestMaxIterations(t *testing.T) {
+	st := newStore(t)
+	ss, _ := st.CreateSession(context.Background(), "")
+	call := llm.Response{ToolCalls: []llm.ToolCall{{ID: "c", Name: "time_now", Arguments: "{}"}}}
+	a := &Agent{LLM: &scripted{resps: []llm.Response{call, call}}, Store: st, Tools: tools.NewRegistry(tools.TimeNow{}), MaxIterations: 2}
+	if _, err := a.Run(context.Background(), ss.ID, "loop", nil); err == nil {
+		t.Fatal("expected iteration error")
+	}
+}
+
+func TestHistoryDropsOrphanToolMessages(t *testing.T) {
+	st := newStore(t)
+	ctx := context.Background()
+	ss, _ := st.CreateSession(ctx, "")
+	st.AppendMessage(ctx, ss.ID, llm.Message{Role: llm.RoleAssistant, ToolCalls: []llm.ToolCall{{ID: "x", Name: "t"}}})
+	st.AppendMessage(ctx, ss.ID, llm.Message{Role: llm.RoleTool, ToolCallID: "x", Content: "r"})
+	st.AppendMessage(ctx, ss.ID, llm.Message{Role: llm.RoleAssistant, Content: "ok"})
+	a := &Agent{Store: st, HistoryLimit: 2}
+	h, _ := a.history(ctx, ss.ID)
+	if len(h) != 1 || h[0].Content != "ok" {
+		t.Fatalf("%+v", h)
+	}
+}

internal/store/store.go

@@ -0,0 +1,181 @@
+// Package store persists sessions and messages in SQLite (pure Go driver, no CGO).
+package store
+
+import (
+	"context"
+	"crypto/rand"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/crypto/openclaw/internal/llm"
+	_ "modernc.org/sqlite"
+)
+
+var ErrNotFound = errors.New("not found")
+
+type Session struct {
+	ID        string    `json:"id"`
+	Title     string    `json:"title"`
+	CreatedAt time.Time `json:"created_at"`
+	UpdatedAt time.Time `json:"updated_at"`
+}
+
+type StoredMessage struct {
+	llm.Message
+	ID        int64     `json:"id"`
+	CreatedAt time.Time `json:"created_at"`
+}
+
+type Store struct{ db *sql.DB }
+
+const schema = `
+PRAGMA journal_mode=WAL;
+PRAGMA foreign_keys=ON;
+CREATE TABLE IF NOT EXISTS sessions(
+  id TEXT PRIMARY KEY, title TEXT NOT NULL DEFAULT '',
+  created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL);
+CREATE TABLE IF NOT EXISTS messages(
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
+  role TEXT NOT NULL, content TEXT NOT NULL DEFAULT '',
+  tool_calls TEXT NOT NULL DEFAULT '', tool_call_id TEXT NOT NULL DEFAULT '',
+  name TEXT NOT NULL DEFAULT '', created_at INTEGER NOT NULL);
+CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, id);
+`
+
+func Open(path string) (*Store, error) {
+	db, err := sql.Open("sqlite", "file:"+path+"?_pragma=busy_timeout(5000)")
+	if err != nil {
+		return nil, err
+	}
+	db.SetMaxOpenConns(1) // SQLite: serialize writers, avoids SQLITE_BUSY
+	if _, err := db.Exec(schema); err != nil {
+		db.Close()
+		return nil, fmt.Errorf("migrate: %w", err)
+	}
+	return &Store{db: db}, nil
+}
+
+func (s *Store) Close() error { return s.db.Close() }
+
+func newID() string {
+	b := make([]byte, 12)
+	rand.Read(b)
+	return hex.EncodeToString(b)
+}
+
+func (s *Store) CreateSession(ctx context.Context, title string) (Session, error) {
+	now := time.Now().UTC()
+	ss := Session{ID: newID(), Title: title, CreatedAt: now, UpdatedAt: now}
+	_, err := s.db.ExecContext(ctx, `INSERT INTO sessions(id,title,created_at,updated_at) VALUES(?,?,?,?)`,
+		ss.ID, ss.Title, now.UnixMilli(), now.UnixMilli())
+	return ss, err
+}
+
+func (s *Store) GetSession(ctx context.Context, id string) (Session, error) {
+	var ss Session
+	var c, u int64
+	err := s.db.QueryRowContext(ctx, `SELECT id,title,created_at,updated_at FROM sessions WHERE id=?`, id).Scan(&ss.ID, &ss.Title, &c, &u)
+	if errors.Is(err, sql.ErrNoRows) {
+		return ss, ErrNotFound
+	}
+	ss.CreatedAt, ss.UpdatedAt = time.UnixMilli(c).UTC(), time.UnixMilli(u).UTC()
+	return ss, err
+}
+
+func (s *Store) ListSessions(ctx context.Context, limit int) ([]Session, error) {
+	rows, err := s.db.QueryContext(ctx, `SELECT id,title,created_at,updated_at FROM sessions ORDER BY updated_at DESC LIMIT ?`, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := []Session{}
+	for rows.Next() {
+		var ss Session
+		var c, u int64
+		if err := rows.Scan(&ss.ID, &ss.Title, &c, &u); err != nil {
+			return nil, err
+		}
+		ss.CreatedAt, ss.UpdatedAt = time.UnixMilli(c).UTC(), time.UnixMilli(u).UTC()
+		out = append(out, ss)
+	}
+	return out, rows.Err()
+}
+
+func (s *Store) RenameSession(ctx context.Context, id, title string) error {
+	res, err := s.db.ExecContext(ctx, `UPDATE sessions SET title=? WHERE id=?`, title, id)
+	return affected(res, err)
+}
+
+func (s *Store) DeleteSession(ctx context.Context, id string) error {
+	res, err := s.db.ExecContext(ctx, `DELETE FROM sessions WHERE id=?`, id)
+	return affected(res, err)
+}
+
+func affected(res sql.Result, err error) error {
+	if err != nil {
+		return err
+	}
+	if n, _ := res.RowsAffected(); n == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+func (s *Store) AppendMessage(ctx context.Context, sessionID string, m llm.Message) (int64, error) {
+	tc := ""
+	if len(m.ToolCalls) > 0 {
+		b, _ := json.Marshal(m.ToolCalls)
+		tc = string(b)
+	}
+	now := time.Now().UTC().UnixMilli()
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return 0, err
+	}
+	defer tx.Rollback()
+	res, err := tx.ExecContext(ctx, `INSERT INTO messages(session_id,role,content,tool_calls,tool_call_id,name,created_at) VALUES(?,?,?,?,?,?,?)`,
+		sessionID, string(m.Role), m.Content, tc, m.ToolCallID, m.Name, now)
+	if err != nil {
+		return 0, err
+	}
+	if _, err := tx.ExecContext(ctx, `UPDATE sessions SET updated_at=? WHERE id=?`, now, sessionID); err != nil {
+		return 0, err
+	}
+	id, _ := res.LastInsertId()
+	return id, tx.Commit()
+}
+
+// Messages returns the last `limit` messages of a session in chronological order (limit<=0: all).
+func (s *Store) Messages(ctx context.Context, sessionID string, limit int) ([]StoredMessage, error) {
+	if limit <= 0 {
+		limit = -1
+	}
+	rows, err := s.db.QueryContext(ctx, `SELECT * FROM (SELECT id,role,content,tool_calls,tool_call_id,name,created_at
+		FROM messages WHERE session_id=? ORDER BY id DESC LIMIT ?) ORDER BY id ASC`, sessionID, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := []StoredMessage{}
+	for rows.Next() {
+		var m StoredMessage
+		var role, tc string
+		var c int64
+		if err := rows.Scan(&m.ID, &role, &m.Content, &tc, &m.ToolCallID, &m.Name, &c); err != nil {
+			return nil, err
+		}
+		m.Role, m.CreatedAt = llm.Role(role), time.UnixMilli(c).UTC()
+		if tc != "" {
+			if err := json.Unmarshal([]byte(tc), &m.ToolCalls); err != nil {
+				return nil, err
+			}
+		}
+		out = append(out, m)
+	}
+	return out, rows.Err()
+}

internal/store/store_test.go

@@ -0,0 +1,50 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"path/filepath"
+	"testing"
+
+	"github.com/crypto/openclaw/internal/llm"
+)
+
+func TestStoreRoundTrip(t *testing.T) {
+	s, err := Open(filepath.Join(t.TempDir(), "t.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	ctx := context.Background()
+	ss, err := s.CreateSession(ctx, "first")
+	if err != nil {
+		t.Fatal(err)
+	}
+	s.AppendMessage(ctx, ss.ID, llm.Message{Role: llm.RoleUser, Content: "hi"})
+	s.AppendMessage(ctx, ss.ID, llm.Message{Role: llm.RoleAssistant, ToolCalls: []llm.ToolCall{{ID: "c1", Name: "time_now", Arguments: "{}"}}})
+	s.AppendMessage(ctx, ss.ID, llm.Message{Role: llm.RoleTool, ToolCallID: "c1", Name: "time_now", Content: "now"})
+	all, _ := s.Messages(ctx, ss.ID, 0)
+	if len(all) != 3 || all[1].ToolCalls[0].Name != "time_now" || all[2].ToolCallID != "c1" {
+		t.Fatalf("%+v", all)
+	}
+	last, _ := s.Messages(ctx, ss.ID, 2)
+	if len(last) != 2 || last[0].Role != llm.RoleAssistant {
+		t.Fatalf("limit: %+v", last)
+	}
+	if err := s.RenameSession(ctx, ss.ID, "renamed"); err != nil {
+		t.Fatal(err)
+	}
+	list, _ := s.ListSessions(ctx, 10)
+	if len(list) != 1 || list[0].Title != "renamed" {
+		t.Fatalf("%+v", list)
+	}
+	if err := s.DeleteSession(ctx, ss.ID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.GetSession(ctx, ss.ID); !errors.Is(err, ErrNotFound) {
+		t.Fatal(err)
+	}
+	if m, _ := s.Messages(ctx, ss.ID, 0); len(m) != 0 {
+		t.Fatal("cascade delete failed")
+	}
+}