diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index fd2ecba..47e3937 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -23,6 +23,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader" personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry" "git.yeij.top/AskaEth/Cyrene_v2/internal/platform" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/capacity" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" @@ -92,6 +93,7 @@ type app struct { backthink *backthink.Backthink vindex vector.Index owner *user.OwnerProfile + history *history.Store gw *gateway.Server channelBridge *channelbridge.Bridge } @@ -510,6 +512,11 @@ func (a *app) wireFrontend() error { a.log.Warn("audit store unavailable, falling back to in-memory", "error", aerr) auditStore = nil } + hist, herr := history.NewStore(context.Background(), a.db, a.log) + if herr != nil { + return fmt.Errorf("history store: %w", herr) + } + a.history = hist gw, err := gateway.New(a.cfg, gateway.Deps{ Log: logger.WithSource(a.log, "gateway"), Auth: jwt, @@ -519,6 +526,7 @@ func (a *app) wireFrontend() error { Session: a.sessions, Personas: a.personas, Memories: a.memStore, + History: a.history, IoT: a.iot, Scheduler: a.scheduler, Tasks: a.tasks, diff --git a/internal/db/migrations/0003_chat_history.down.sql b/internal/db/migrations/0003_chat_history.down.sql new file mode 100644 index 0000000..0ae457e --- /dev/null +++ b/internal/db/migrations/0003_chat_history.down.sql @@ -0,0 +1,4 @@ +-- 0003_chat_history: rolls back the chat history tables created by the up migration. +DROP TABLE IF EXISTS chat_messages; +DROP INDEX IF EXISTS idx_chat_msgs_session_created; +DROP TABLE IF EXISTS chat_sessions; diff --git a/internal/db/migrations/0003_chat_history.up.sql b/internal/db/migrations/0003_chat_history.up.sql new file mode 100644 index 0000000..3d16370 --- /dev/null +++ b/internal/db/migrations/0003_chat_history.up.sql @@ -0,0 +1,31 @@ +-- 0003_chat_history: persistent chat sessions + message stream (DevDocs/design/16 §2.1, §3.1). +-- chat_sessions records the persistent sessions: one fixed main session plus arbitrarily +-- many secondary (sub) sessions. chat_messages is the raw, durable message stream in +-- millisecond-timestamp cursor order. The two are the backend truth for the Web frontend; +-- they are NOT memory (design/16 §3.2 separates memory from the raw stream). +-- +-- Its canonical schema is owned by internal/runtime/history, which reuses these tables. The +-- migration is the authoritative path; internal/runtime/history.NewStore re-issues the same +-- DDL idempotently so the store is self-contained (matching memory/session). + +CREATE TABLE IF NOT EXISTS chat_sessions ( + id TEXT PRIMARY KEY, -- main = fixed id ("web"); sub = "web:" + type TEXT NOT NULL, -- 'main' | 'sub' + title TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL -- ms timestamp +); + +CREATE TABLE IF NOT EXISTS chat_messages ( + id TEXT PRIMARY KEY, -- message unique id (uuid) + session_id TEXT NOT NULL, -- owning session (main or sub) + role TEXT NOT NULL, -- user / assistant / system + content TEXT NOT NULL, -- raw message body (incl.

/) + mode TEXT, -- behavior mode (optional, bubble render) + mood TEXT, -- mood (optional) + image TEXT, -- attachment base64/URL (optional) + thinking TEXT, -- chain of thought (optional, stored, not echoed by default) + created_at INTEGER NOT NULL -- ms timestamp (pagination cursor) +); + +CREATE INDEX IF NOT EXISTS idx_chat_msgs_session_created + ON chat_messages (session_id, created_at DESC); diff --git a/internal/runtime/history/store.go b/internal/runtime/history/store.go new file mode 100644 index 0000000..76d80af --- /dev/null +++ b/internal/runtime/history/store.go @@ -0,0 +1,390 @@ +// Package history implements the persistent chat-history session model described in +// DevDocs/design/16-会话持久化与主次级会话.md. +// +// It is the storage layer, not the orchestration/context layer: it persists the raw +// message stream (chat_messages) and the persistent session records (chat_sessions) +// so the Web frontend can request + display history, and secondary (sub) sessions can +// be created, listed and deleted. It intentionally does not touch memory (design/16 +// §3.2: memory is persona-scoped and lives in runtime/memory), does not build context +// (runtime/loop) and does not schedule turns (gateway/session_orchestrator). +// +// Session model (design/16 §2.1): there is exactly one fixed main session (id "web") and +// any number of sub sessions (id "web:"). Main + sub share the same persona and +// therefore the same memory; this package only models sessions + messages. +package history + +import ( + "context" + "crypto/rand" + "database/sql" + "encoding/hex" + "errors" + "fmt" + "io" + "log/slog" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +// Session roles and types. These are the values persisted in the role / type columns. +const ( + // RoleUser, RoleAssistant, RoleSystem classify a message's author. + RoleUser = "user" + RoleAssistant = "assistant" + RoleSystem = "system" + + // SessionMain and SessionSub classify a persistent session. + SessionMain = "main" + SessionSub = "sub" +) + +// MainSessionID is the fixed id of the one main session (design/16 §2.1). Sub session +// ids are MainSessionID + ":" + uuid. +const MainSessionID = "web" + +// defaultPageLimit is the page size used when the caller does not pass one. +const defaultPageLimit = 50 + +// maxPageLimit bounds a client-supplied limit so a single request cannot drain memory. +const maxPageLimit = 200 + +// Sentinels distinguishing history package failures. Callers wrap them with %w so +// errors.Is works through the chain. +var ( + // ErrNilDB is returned when NewStore receives a nil *sql.DB. + ErrNilDB = errors.New("history: nil db") + + // ErrImmutableMain is returned when a caller tries to delete the fixed main session. + ErrImmutableMain = errors.New("history: the main session cannot be deleted") +) + +// Message is one persisted chat message. CreatedAt is a Unix millisecond timestamp and +// doubles as the pagination cursor (design/16 §3.1): pages are ordered newest-first and +// "before" is the created_at of the oldest message on the previous page. +type Message struct { + ID string + SessionID string + Role string + Content string + Mode string + Mood string + Image string + Thinking string + CreatedAt int64 +} + +// SessionMeta is the wire-facing metadata of one persistent session. +type SessionMeta struct { + ID string + Type string // "main" | "sub" + Title string + CreatedAt int64 +} + +// HistoryStore abstracts the chat-history store so callers (the gateway) and tests can +// hold an interface rather than a concrete *Store. +type HistoryStore interface { + // Append persists msg, assigning an id and timestamp when the caller omitted them. + Append(ctx context.Context, msg Message) error + // List returns up to limit messages for sessionID, newest first. before is a + // millisecond timestamp cursor: when nonzero, only messages older than before are + // returned (inclusive-free, matching the "before" bound on the frontend). + List(ctx context.Context, sessionID string, before int64, limit int) ([]Message, error) + // Recent returns the most recent n messages for sessionID in oldest-first order, + // for WS replay / context tail usage. + Recent(ctx context.Context, sessionID string, n int) ([]Message, error) + // DeleteSession removes a sub session and all its messages. The main session is + // immutable and returns ErrImmutableMain. + DeleteSession(ctx context.Context, sessionID string) error + // NewSubSession creates a new sub session and returns its id (MainSessionID:uuid). + NewSubSession(ctx context.Context, ownerID string) (string, error) + // ListSessions returns the main session first then every sub session, oldest first. + ListSessions(ctx context.Context) ([]SessionMeta, error) +} + +// Store is the SQLite-backed chat-history store. It owns the chat_sessions and +// chat_messages tables. Concurrency is delegated to *sql.DB (safe for concurrent use), +// so Store holds no further lock. Every query is parameterized and reads list columns +// explicitly (no SELECT *); reads take a context. +type Store struct { + db *sql.DB + log *slog.Logger +} + +// rowScanner is satisfied by both *sql.Row and *sql.Rows, so a single scanMessage +// helper serves List and Recent. +type rowScanner interface { + Scan(dest ...any) error +} + +// NewStore builds a Store on db, ensuring the chat_sessions/chat_messages schema exists +// and that the fixed main session row is present. The DDL mirrors the 0003 migration; +// it is idempotent so the store is self-contained and usable from any *sql.DB, matching +// the memory and session packages. db must be live; nil returns ErrNilDB. log may be nil. +func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) { + if db == nil { + return nil, ErrNilDB + } + log = logger.WithSource(discardLogger(log), "history") + for _, ddl := range []string{ + createSessionsTable, + createMessagesTable, + createMessagesIndex, + } { + if _, err := db.ExecContext(ctx, ddl); err != nil { + return nil, fmt.Errorf("history: apply schema: %w", err) + } + } + s := &Store{db: db, log: log} + if err := s.ensureMainSession(ctx); err != nil { + return nil, err + } + return s, nil +} + +// Append persists msg, assigning a fresh id and a now-ms CreatedAt when the caller left +// either empty/zero. It returns the error wrapped with the operation name. +func (s *Store) Append(ctx context.Context, msg Message) error { + if msg.ID == "" { + id, err := newID() + if err != nil { + return fmt.Errorf("history: append id: %w", err) + } + msg.ID = id + } + if msg.CreatedAt == 0 { + msg.CreatedAt = time.Now().UnixMilli() + } + _, err := s.db.ExecContext(ctx, `INSERT INTO chat_messages + (id, session_id, role, content, mode, mood, image, thinking, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + msg.ID, msg.SessionID, msg.Role, msg.Content, msg.Mode, msg.Mood, + msg.Image, msg.Thinking, msg.CreatedAt) + if err != nil { + return fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) + } + return nil +} + +// List returns up to limit messages for sessionID, newest first. When before is nonzero +// it bounds the query to messages strictly older than before, so a client can page upward +// by passing the oldest message's created_at from the previous page. The returned slice +// is non-nil. limit is clamped to [1, maxPageLimit]. +func (s *Store) List(ctx context.Context, sessionID string, before int64, limit int) ([]Message, error) { + limit = clampLimit(limit) + query := `SELECT id, session_id, role, content, mode, mood, image, thinking, created_at + FROM chat_messages WHERE session_id = ?` + args := []any{sessionID} + if before > 0 { + query += ` AND created_at < ?` + args = append(args, before) + } + query += ` ORDER BY created_at DESC LIMIT ?` + args = append(args, limit) + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("history: list %s: %w", sessionID, err) + } + defer func() { _ = rows.Close() }() + + msgs := make([]Message, 0) + for rows.Next() { + m, err := scanMessage(rows) + if err != nil { + return nil, fmt.Errorf("history: scan list %s: %w", sessionID, err) + } + msgs = append(msgs, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate list %s: %w", sessionID, err) + } + return msgs, nil +} + +// Recent returns the most recent n messages for sessionID in oldest-first order (the +// tail-most window), for WS replay / context tail. The returned slice is non-nil. +func (s *Store) Recent(ctx context.Context, sessionID string, n int) ([]Message, error) { + if n < 1 { + n = defaultPageLimit + } + rows, err := s.db.QueryContext(ctx, `SELECT id, session_id, role, content, mode, mood, image, thinking, created_at + FROM chat_messages WHERE session_id = ? ORDER BY created_at DESC LIMIT ?`, sessionID, n) + if err != nil { + return nil, fmt.Errorf("history: recent %s: %w", sessionID, err) + } + defer func() { _ = rows.Close() }() + + // Recent returns newest-last so the window is in chronological order; build it and + // reverse in place. + msgs := make([]Message, 0) + for rows.Next() { + m, err := scanMessage(rows) + if err != nil { + return nil, fmt.Errorf("history: scan recent %s: %w", sessionID, err) + } + msgs = append(msgs, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate recent %s: %w", sessionID, err) + } + for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 { + msgs[i], msgs[j] = msgs[j], msgs[i] + } + return msgs, nil +} + +// DeleteSession removes a sub session and all its messages in one transaction. The fixed +// main session is immutable and returns ErrImmutableMain (design/16 §6). +func (s *Store) DeleteSession(ctx context.Context, sessionID string) error { + if sessionID == MainSessionID { + return ErrImmutableMain + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("history: begin delete %s: %w", sessionID, err) + } + defer func() { _ = tx.Rollback() }() + + if _, err := tx.ExecContext(ctx, `DELETE FROM chat_messages WHERE session_id = ?`, sessionID); err != nil { + return fmt.Errorf("history: delete messages %s: %w", sessionID, err) + } + if _, err := tx.ExecContext(ctx, `DELETE FROM chat_sessions WHERE id = ? AND type = ?`, sessionID, SessionSub); err != nil { + return fmt.Errorf("history: delete session %s: %w", sessionID, err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("history: commit delete %s: %w", sessionID, err) + } + return nil +} + +// NewSubSession creates a sub session and returns its id (MainSessionID:uuid). ownerID is +// accepted for interface completeness (the Web owner is identified by JWT, not stored per +// session); it is logged at debug and not persisted. +func (s *Store) NewSubSession(ctx context.Context, ownerID string) (string, error) { + id, err := newID() + if err != nil { + return "", fmt.Errorf("history: new sub id: %w", err) + } + sessionID := MainSessionID + ":" + id + now := time.Now().UnixMilli() + if _, err := s.db.ExecContext(ctx, `INSERT INTO chat_sessions (id, type, title, created_at) + VALUES (?, ?, ?, ?)`, sessionID, SessionSub, "", now); err != nil { + return "", fmt.Errorf("history: insert sub session: %w", err) + } + if s.log != nil { + s.log.Debug("sub session created", "session_id", sessionID, "owner_id", ownerID) + } + return sessionID, nil +} + +// ListSessions returns the main session first, then every sub session, oldest first. The +// returned slice is non-nil. +func (s *Store) ListSessions(ctx context.Context) ([]SessionMeta, error) { + rows, err := s.db.QueryContext(ctx, `SELECT id, type, title, created_at FROM chat_sessions + ORDER BY (type = ?) DESC, created_at ASC`, SessionMain) + if err != nil { + return nil, fmt.Errorf("history: list sessions: %w", err) + } + defer func() { _ = rows.Close() }() + + sessions := make([]SessionMeta, 0) + for rows.Next() { + var ( + meta SessionMeta + typeStr string + createdAt int64 + ) + if err := rows.Scan(&meta.ID, &typeStr, &meta.Title, &createdAt); err != nil { + return nil, fmt.Errorf("history: scan session: %w", err) + } + meta.Type = typeStr + meta.CreatedAt = createdAt + sessions = append(sessions, meta) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate sessions: %w", err) + } + return sessions, nil +} + +// ensureMainSession inserts the fixed main session row when it is absent, so ListSessions +// always reports the main session even before any message is written. +func (s *Store) ensureMainSession(ctx context.Context) error { + _, err := s.db.ExecContext(ctx, `INSERT OR IGNORE INTO chat_sessions (id, type, title, created_at) + VALUES (?, ?, ?, ?)`, MainSessionID, SessionMain, "", time.Now().UnixMilli()) + if err != nil { + return fmt.Errorf("history: ensure main session: %w", err) + } + return nil +} + +// scanMessage reads one chat_messages row. message columns are scanned positionally. +func scanMessage(sc rowScanner) (Message, error) { + var m Message + var mode, mood, image, thinking string + if err := sc.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &mode, &mood, &image, &thinking, &m.CreatedAt); err != nil { + return Message{}, err + } + m.Mode, m.Mood, m.Image, m.Thinking = mode, mood, image, thinking + return m, nil +} + +// clampLimit bounds limit to [1, maxPageLimit]; a caller-supplied 0 or negative falls back +// to defaultPageLimit. +func clampLimit(limit int) int { + if limit < 1 { + return defaultPageLimit + } + if limit > maxPageLimit { + return maxPageLimit + } + return limit +} + +// newID returns a 128-bit cryptographically random id, hex-encoded. It avoids a UUID +// dependency so the package stays self-contained. +func newID() (string, error) { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + return "", fmt.Errorf("history: generate id: %w", err) + } + return hex.EncodeToString(b), nil +} + +// discardLogger returns log if non-nil, otherwise a logger that writes nowhere. +func discardLogger(log *slog.Logger) *slog.Logger { + if log != nil { + return log + } + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// Schema DDL. The 0003 migration in internal/db is the authoritative mechanism for +// domain tables, but it is intentionally also re-issued here idempotently so the store is +// self-contained and usable from any *sql.DB (matching the memory and session packages). +const createSessionsTable = ` +CREATE TABLE IF NOT EXISTS chat_sessions ( + id TEXT PRIMARY KEY, + type TEXT NOT NULL, + title TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL +)` + +const createMessagesTable = ` +CREATE TABLE IF NOT EXISTS chat_messages ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + role TEXT NOT NULL, + content TEXT NOT NULL, + mode TEXT, + mood TEXT, + image TEXT, + thinking TEXT, + created_at INTEGER NOT NULL +)` + +const createMessagesIndex = ` +CREATE INDEX IF NOT EXISTS idx_chat_msgs_session_created + ON chat_messages (session_id, created_at DESC)` diff --git a/internal/runtime/history/store_test.go b/internal/runtime/history/store_test.go new file mode 100644 index 0000000..cb66e30 --- /dev/null +++ b/internal/runtime/history/store_test.go @@ -0,0 +1,204 @@ +package history + +import ( + "context" + "errors" + "io" + "log/slog" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" +) + +// testLogger returns a logger that writes nowhere, keeping test output clean. +func testLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// newStore opens an isolated in-memory DB and returns a Store on it. +func newStore(t *testing.T) *Store { + t.Helper() + ctx := context.Background() + conn, err := db.Open(ctx, ":memory:", testLogger()) + if err != nil { + t.Fatalf("open memory db: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + st, err := NewStore(ctx, conn, testLogger()) + if err != nil { + t.Fatalf("new store: %v", err) + } + return st +} + +func TestNewStoreNilDB(t *testing.T) { + t.Parallel() + if _, err := NewStore(context.Background(), nil, testLogger()); !errors.Is(err, ErrNilDB) { + t.Fatalf("err = %v, want wrap of ErrNilDB", err) + } +} + +func TestAppendAndListNewestFirst(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + for i, ts := range []int64{1000, 2000, 3000} { + m := Message{SessionID: MainSessionID, Role: RoleUser, Content: "hi", CreatedAt: ts} + if err := st.Append(ctx, m); err != nil { + t.Fatalf("append %d: %v", i, err) + } + } + + msgs, err := st.List(ctx, MainSessionID, 0, 50) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 3 { + t.Fatalf("len = %d, want 3", len(msgs)) + } + // Newest first. + if msgs[0].CreatedAt != 3000 || msgs[1].CreatedAt != 2000 || msgs[2].CreatedAt != 1000 { + t.Fatalf("order = %+v, want [3000 2000 1000]", []int64{msgs[0].CreatedAt, msgs[1].CreatedAt, msgs[2].CreatedAt}) + } + // Id/timestamp assigned when the caller omits them. + empty := Message{SessionID: MainSessionID, Role: RoleAssistant, Content: "ok"} + if err := st.Append(ctx, empty); err != nil { + t.Fatalf("append no ts: %v", err) + } + msgs, err = st.List(ctx, MainSessionID, 0, 50) + if err != nil { + t.Fatalf("list after append: %v", err) + } + if msgs[0].ID == "" || msgs[0].CreatedAt == 0 { + t.Fatalf("append did not fill id/created_at: %+v", msgs[0]) + } +} + +func TestListPaginationBefore(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + for _, ts := range []int64{1000, 2000, 3000, 4000, 5000, 6000} { + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "x", CreatedAt: ts}); err != nil { + t.Fatalf("append %d: %v", ts, err) + } + } + + // First page: newest 2. + page, err := st.List(ctx, MainSessionID, 0, 2) + if err != nil { + t.Fatalf("first page: %v", err) + } + if len(page) != 2 || page[0].CreatedAt != 6000 || page[1].CreatedAt != 5000 { + t.Fatalf("first page = %+v, want [6000 5000]", page) + } + // Cursor on the page's oldest (5000) → older page [4000 3000]. + next, err := st.List(ctx, MainSessionID, page[1].CreatedAt, 2) + if err != nil { + t.Fatalf("next page: %v", err) + } + if len(next) != 2 || next[0].CreatedAt != 4000 || next[1].CreatedAt != 3000 { + t.Fatalf("next page = %+v, want [4000 3000]", next) + } + // Past the start → empty. + tail, err := st.List(ctx, MainSessionID, 1000, 50) + if err != nil { + t.Fatalf("tail: %v", err) + } + if len(tail) != 0 { + t.Fatalf("tail = %+v, want empty", tail) + } +} + +func TestRecentChronological(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + for _, ts := range []int64{1000, 2000, 3000, 4000, 5000} { + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "x", CreatedAt: ts}); err != nil { + t.Fatalf("append %d: %v", ts, err) + } + } + recent, err := st.Recent(ctx, MainSessionID, 3) + if err != nil { + t.Fatalf("recent: %v", err) + } + if len(recent) != 3 || recent[0].CreatedAt != 3000 || recent[1].CreatedAt != 4000 || recent[2].CreatedAt != 5000 { + t.Fatalf("recent = %+v, want [3000 4000 5000]", recent) + } +} + +func TestNewSubSessionAndListSessions(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + // Main session exists by default. + sessions, err := st.ListSessions(ctx) + if err != nil { + t.Fatalf("list sessions initial: %v", err) + } + if len(sessions) != 1 || sessions[0].ID != MainSessionID || sessions[0].Type != SessionMain { + t.Fatalf("initial sessions = %+v, want [main web]", sessions) + } + + id, err := st.NewSubSession(ctx, "owner") + if err != nil { + t.Fatalf("new sub: %v", err) + } + if !strings.HasPrefix(id, MainSessionID+":") { + t.Fatalf("sub id = %q, want prefix web:", id) + } + + sessions, err = st.ListSessions(ctx) + if err != nil { + t.Fatalf("list sessions: %v", err) + } + if len(sessions) != 2 || sessions[0].ID != MainSessionID || sessions[0].Type != SessionMain { + t.Fatalf("sessions[0] = %+v, want main web first", sessions[0]) + } + if sessions[1].ID != id || sessions[1].Type != SessionSub { + t.Fatalf("sessions[1] = %+v, want sub %s", sessions[1], id) + } +} + +func TestDeleteSession(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + sub, err := st.NewSubSession(ctx, "owner") + if err != nil { + t.Fatalf("new sub: %v", err) + } + if err := st.Append(ctx, Message{SessionID: sub, Role: RoleUser, Content: "x", CreatedAt: 1000}); err != nil { + t.Fatalf("append: %v", err) + } + + if err := st.DeleteSession(ctx, sub); err != nil { + t.Fatalf("delete sub: %v", err) + } + msgs, err := st.List(ctx, sub, 0, 50) + if err != nil { + t.Fatalf("list after delete: %v", err) + } + if len(msgs) != 0 { + t.Fatalf("messages after delete = %+v, want empty", msgs) + } + sessions, err := st.ListSessions(ctx) + if err != nil { + t.Fatalf("list sessions after delete: %v", err) + } + if len(sessions) != 1 { + t.Fatalf("sessions after delete = %+v, want [main]", sessions) + } + + // Main session is immutable. + if err := st.DeleteSession(ctx, MainSessionID); !errors.Is(err, ErrImmutableMain) { + t.Fatalf("delete main err = %v, want ErrImmutableMain", err) + } +} diff --git a/internal/services/gateway/gateway_test.go b/internal/services/gateway/gateway_test.go index ba33307..e2858e1 100644 --- a/internal/services/gateway/gateway_test.go +++ b/internal/services/gateway/gateway_test.go @@ -384,6 +384,33 @@ func TestWSUpgradeAndBroadcast(t *testing.T) { } } +func TestWsEventSessionIDField(t *testing.T) { + t.Parallel() + // When set, sessionId is marshalled so the client can filter by session. + set, err := json.Marshal(wsEvent{Type: "stream_end", SessionID: "web", Data: map[string]any{"stopReason": "stop"}}) + if err != nil { + t.Fatalf("marshal set: %v", err) + } + if !strings.Contains(string(set), `"sessionId":"web"`) { + t.Fatalf("set frame = %s, want sessionId web", set) + } + // When empty, omitempty drops it so existing clients/tests see no change. + unset, err := json.Marshal(wsEvent{Type: "stream_end", Data: map[string]any{"stopReason": "stop"}}) + if err != nil { + t.Fatalf("marshal unset: %v", err) + } + if strings.Contains(string(unset), "sessionId") { + t.Fatalf("unset frame = %s, want sessionId omitted", unset) + } + var ev wsEvent + if err := json.Unmarshal(unset, &ev); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if ev.SessionID != "" { + t.Fatalf("decoded SessionID = %q, want empty", ev.SessionID) + } +} + // readWS reads one text frame and decodes it into a wsEvent. func readWS(t *testing.T, conn *websocket.Conn) wsEvent { t.Helper() diff --git a/internal/services/gateway/history.go b/internal/services/gateway/history.go new file mode 100644 index 0000000..3eb3cf8 --- /dev/null +++ b/internal/services/gateway/history.go @@ -0,0 +1,165 @@ +package gateway + +import ( + "errors" + "net/http" + "strconv" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// historySessionSummary is the wire shape of one persistent session (main or sub) in the +// GET /api/v1/sessions response. +type historySessionSummary struct { + ID string `json:"id"` + Type string `json:"type"` // "main" | "sub" + Title string `json:"title,omitempty"` + CreatedAt int64 `json:"createdAt"` +} + +// historyMessage is the wire shape of one persisted chat message in the +// GET /api/v1/sessions/{id}/messages response. Mode is carried so the frontend renders +// the bubble without guessing (design/16 §5.1); thinking is stored and returned to the +// owner/admin (the messages route is owner/admin-only, so it never reaches a viewer). +type historyMessage struct { + ID string `json:"id"` + SessionID string `json:"sessionId"` + Role string `json:"role"` + Content string `json:"content"` + Mode string `json:"mode,omitempty"` + Mood string `json:"mood,omitempty"` + Image string `json:"image,omitempty"` + Thinking string `json:"thinking,omitempty"` + CreatedAt int64 `json:"createdAt"` +} + +// handleHistorySessions returns the main session first, then every sub session (oldest +// first). It is the chat-history session list for the frontend (design/16 §2.2). +func (s *Server) handleHistorySessions(w http.ResponseWriter, r *http.Request) { + meta, err := s.history.ListSessions(r.Context()) + if err != nil { + s.log.Error("history list sessions", "err", err) + writeError(w, apperr.Wrap(apperr.ErrInternal, "sessions_failed", "failed to list sessions")) + return + } + out := make([]historySessionSummary, 0, len(meta)) + for _, m := range meta { + out = append(out, historySessionSummary{ + ID: m.ID, + Type: m.Type, + Title: m.Title, + CreatedAt: m.CreatedAt, + }) + } + writeJSON(w, http.StatusOK, out) +} + +// handleCreateSubSession creates a new sub session and returns its sessionId. It is a +// user-side write, so it is owner/admin-gated like /api/v1/chat. +func (s *Server) handleCreateSubSession(w http.ResponseWriter, r *http.Request) { + if s.history == nil { + writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置（未注入 HistoryStore）", HTTPCode: http.StatusNotImplemented}) + return + } + ownerID := "" + if claims, ok := auth.ClaimsFromContext(r.Context()); ok { + ownerID = claims.Sub + } + id, err := s.history.NewSubSession(r.Context(), ownerID) + if err != nil { + s.log.Error("history create sub session", "err", err) + writeError(w, apperr.Wrap(apperr.ErrInternal, "session_failed", "failed to create session")) + return + } + writeJSON(w, http.StatusCreated, map[string]any{"sessionId": id}) +} + +// handleListMessages returns a page of messages for a session, newest first, with cursor +// pagination. It is the hot-paging history endpoint (design/16 §2.2, §5.1): limit defaults +// to 50 (clamped to 200), before is the oldest message's created_at from the previous page. +func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) { + if s.history == nil { + writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置（未注入 HistoryStore）", HTTPCode: http.StatusNotImplemented}) + return + } + id := r.PathValue("id") + if id == "" { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "session id is required")) + return + } + limitStr := r.URL.Query().Get("limit") + limit := historyDefaultLimit + if limitStr != "" { + v, err := strconv.Atoi(limitStr) + if err != nil || v < 1 { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid limit")) + return + } + limit = v + } + var before int64 + if b := r.URL.Query().Get("before"); b != "" { + v, err := strconv.ParseInt(b, 10, 64) + if err != nil { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid before")) + return + } + before = v + } + + msgs, err := s.history.List(r.Context(), id, before, limit) + if err != nil { + s.log.Error("history list messages", "err", err, "session", id) + writeError(w, apperr.Wrap(apperr.ErrInternal, "messages_failed", "failed to list messages")) + return + } + out := make([]historyMessage, 0, len(msgs)) + for _, m := range msgs { + out = append(out, toHistoryMessage(m)) + } + writeJSON(w, http.StatusOK, out) +} + +// handleDeleteSession deletes a sub session and all its messages. The main session is +// immutable (design/16 §6) and returns 403. +func (s *Server) handleDeleteSession(w http.ResponseWriter, r *http.Request) { + if s.history == nil { + writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置（未注入 HistoryStore）", HTTPCode: http.StatusNotImplemented}) + return + } + id := r.PathValue("id") + if id == "" { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "session id is required")) + return + } + if err := s.history.DeleteSession(r.Context(), id); err != nil { + if errors.Is(err, history.ErrImmutableMain) { + writeError(w, apperr.Wrap(apperr.ErrForbidden, "forbidden", "the main session cannot be deleted")) + return + } + s.log.Error("history delete session", "err", err, "session", id) + writeError(w, apperr.Wrap(apperr.ErrInternal, "session_failed", "failed to delete session")) + return + } + writeJSON(w, http.StatusOK, map[string]any{"deleted": id}) +} + +// toHistoryMessage maps a stored history.Message to its wire shape. +func toHistoryMessage(m history.Message) historyMessage { + return historyMessage{ + ID: m.ID, + SessionID: m.SessionID, + Role: m.Role, + Content: m.Content, + Mode: m.Mode, + Mood: m.Mood, + Image: m.Image, + Thinking: m.Thinking, + CreatedAt: m.CreatedAt, + } +} + +// historyDefaultLimit is the page size used when the client omits the limit query param. +const historyDefaultLimit = 50 diff --git a/internal/services/gateway/history_handler_test.go b/internal/services/gateway/history_handler_test.go new file mode 100644 index 0000000..64a4531 --- /dev/null +++ b/internal/services/gateway/history_handler_test.go @@ -0,0 +1,228 @@ +package gateway + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// testLogger returns a logger that writes nowhere, keeping test output clean. +func testLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// newHistoryEnv builds a Server wired with a real in-memory history store, returning the +// httptest server, the JWT service and the underlying history store. +func newHistoryEnv(t *testing.T) (*Server, *httptest.Server, *auth.JWTService, *history.Store) { + t.Helper() + ctx := context.Background() + conn, err := db.Open(ctx, ":memory:", testLogger()) + if err != nil { + t.Fatalf("open memory db: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + st, err := history.NewStore(ctx, conn, testLogger()) + if err != nil { + t.Fatalf("new history store: %v", err) + } + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{History: st}) + return srv, ts, jwt, st +} + +// decodeDataSlice decodes the {data:[...]} envelope into the given slice target. +func decodeDataSlice(t *testing.T, body []byte, out any) { + t.Helper() + if err := json.Unmarshal(decodeBody(t, body)["data"], out); err != nil { + t.Fatalf("decode data: %v (body=%s)", err, body) + } +} + +func TestHistoryCreateAndListSessions(t *testing.T) { + t.Parallel() + _, ts, jwt, _ := newHistoryEnv(t) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // Fresh store always lists the main session. + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("list status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var initial []historySessionSummary + decodeDataSlice(t, body, &initial) + if len(initial) != 1 || initial[0].ID != history.MainSessionID || initial[0].Type != history.SessionMain { + t.Fatalf("initial sessions = %+v, want [main web]", initial) + } + + // Create a sub session. + resp, body = doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions", nil, tok) + if resp.StatusCode != http.StatusCreated { + t.Fatalf("create status = %d, want 201 (body=%s)", resp.StatusCode, body) + } + var created struct { + SessionID string `json:"sessionId"` + } + if err := json.Unmarshal(decodeBody(t, body)["data"], &created); err != nil { + t.Fatalf("decode create data: %v (body=%s)", err, body) + } + if !strings.HasPrefix(created.SessionID, history.MainSessionID+":") { + t.Fatalf("sessionId = %q, want prefix web:", created.SessionID) + } + + // Now main first, then the sub. + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("list status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var sessions []historySessionSummary + decodeDataSlice(t, body, &sessions) + if len(sessions) != 2 || sessions[0].ID != history.MainSessionID || sessions[0].Type != history.SessionMain { + t.Fatalf("sessions[0] = %+v, want main web first", sessions[0]) + } + if sessions[1].ID != created.SessionID || sessions[1].Type != history.SessionSub { + t.Fatalf("sessions[1] = %+v, want sub %s", sessions[1], created.SessionID) + } +} + +func TestHistoryListMessagesPagination(t *testing.T) { + t.Parallel() + _, ts, jwt, st := newHistoryEnv(t) + ctx := context.Background() + // Seed the main session with messages at ascending ms timestamps. + for _, ts := range []int64{1000, 2000, 3000, 4000} { + if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "x", CreatedAt: ts}); err != nil { + t.Fatalf("seed %d: %v", ts, err) + } + } + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // Newest page of 2. + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages?limit=2", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("first page status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var page []historyMessage + decodeDataSlice(t, body, &page) + if len(page) != 2 || page[0].CreatedAt != 4000 || page[1].CreatedAt != 3000 { + t.Fatalf("first page = %+v, want [4000 3000]", page) + } + + // Cursor before the page's oldest (3000) → older page [2000 1000]. + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages?before=3000&limit=2", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("older page status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var older []historyMessage + decodeDataSlice(t, body, &older) + if len(older) != 2 || older[0].CreatedAt != 2000 || older[1].CreatedAt != 1000 { + t.Fatalf("older page = %+v, want [2000 1000]", older) + } + + // Invalid limit/before → 400. + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages?limit=abc", nil, tok) + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("invalid limit status = %d, want 400 (body=%s)", resp.StatusCode, body) + } +} + +func TestHistoryDeleteSubSession(t *testing.T) { + t.Parallel() + _, ts, jwt, st := newHistoryEnv(t) + ctx := context.Background() + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // Create a sub and seed a message. + resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions", nil, tok) + if resp.StatusCode != http.StatusCreated { + t.Fatalf("create status = %d, want 201 (body=%s)", resp.StatusCode, body) + } + var created struct { + SessionID string `json:"sessionId"` + } + if err := json.Unmarshal(decodeBody(t, body)["data"], &created); err != nil { + t.Fatalf("decode create data: %v", err) + } + if err := st.Append(ctx, history.Message{SessionID: created.SessionID, Role: history.RoleUser, Content: "x", CreatedAt: 1000}); err != nil { + t.Fatalf("seed sub: %v", err) + } + + // Delete it. + resp, _ = doReq(t, http.MethodDelete, ts.URL+"/api/v1/sessions/"+created.SessionID, nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("delete status = %d, want 200", resp.StatusCode) + } + // It is gone from the list. + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("list status = %d, want 200", resp.StatusCode) + } + var sessions []historySessionSummary + decodeDataSlice(t, body, &sessions) + if len(sessions) != 1 || sessions[0].ID != history.MainSessionID { + t.Fatalf("sessions after delete = %+v, want [main]", sessions) + } + + // Main session is immutable → 403. + resp, body = doReq(t, http.MethodDelete, ts.URL+"/api/v1/sessions/"+history.MainSessionID, nil, tok) + if resp.StatusCode != http.StatusForbidden { + t.Fatalf("delete main status = %d, want 403 (body=%s)", resp.StatusCode, body) + } + wantErrorShape(t, body) +} + +func TestHistoryNotWired(t *testing.T) { + t.Parallel() + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // Messages/create/delete answer a clear 501 when no history store is wired. + for _, tc := range []struct { + method, url string + want int + }{ + {http.MethodGet, "/api/v1/sessions/web/messages?limit=2", http.StatusNotImplemented}, + {http.MethodPost, "/api/v1/sessions", http.StatusNotImplemented}, + } { + resp, body := doReq(t, tc.method, ts.URL+tc.url, nil, tok) + if resp.StatusCode != tc.want { + t.Fatalf("%s %s status = %d, want %d (body=%s)", tc.method, tc.url, resp.StatusCode, tc.want, body) + } + if !strings.Contains(string(body), "not_implemented") { + t.Fatalf("%s %s body = %s, want not_implemented code", tc.method, tc.url, body) + } + } +} + +func TestHistoryRBAC(t *testing.T) { + t.Parallel() + _, ts, jwt, _ := newHistoryEnv(t) + viewer := signAny(t, jwt, "acct", auth.RoleViewer) + owner := signAny(t, jwt, "acct", auth.RoleOwner) + + // Viewer can list session metadata (readAuth) but not read/write history content. + resp, _ := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions", nil, viewer) + if resp.StatusCode != http.StatusOK { + t.Fatalf("viewer list sessions status = %d, want 200", resp.StatusCode) + } + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/web/messages", nil, viewer) + if resp.StatusCode != http.StatusForbidden { + t.Fatalf("viewer messages status = %d, want 403 (body=%s)", resp.StatusCode, body) + } + resp, body = doReq(t, http.MethodPost, ts.URL+"/api/v1/sessions", nil, viewer) + if resp.StatusCode != http.StatusForbidden { + t.Fatalf("viewer create status = %d, want 403 (body=%s)", resp.StatusCode, body) + } + // Owner can read messages. + resp, _ = doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/web/messages?limit=2", nil, owner) + if resp.StatusCode != http.StatusOK { + t.Fatalf("owner messages status = %d, want 200", resp.StatusCode) + } +} diff --git a/internal/services/gateway/rest.go b/internal/services/gateway/rest.go index 300d454..05657a0 100644 --- a/internal/services/gateway/rest.go +++ b/internal/services/gateway/rest.go @@ -175,8 +175,15 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, chatResponse{Reply: res.Text, Steps: res.Steps, StopReason: res.StopReason}) } -// handleListSessions returns the known sessions (or [] when no store is wired). +// handleListSessions returns the persistent chat-history sessions (main + sub) when a +// history store is wired, otherwise falls back to the channel-session view. The fallback +// keeps the pre-design/16 contract: a server built without history still lists the known +// channel sessions (or [] when no store is wired). func (s *Server) handleListSessions(w http.ResponseWriter, r *http.Request) { + if s.history != nil { + s.handleHistorySessions(w, r) + return + } out := make([]SessionSummary, 0) if s.sessions != nil { list, err := s.sessions.List(r.Context()) diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index bdcdd3b..34c64af 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -37,6 +37,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader" personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" @@ -102,6 +103,19 @@ type MemoryStore interface { List(ctx context.Context) ([]memory.MemoryItem, error) } +// HistoryStore is the persistent chat-history store (design/16). *runtime/history.Store +// is the production implementation. When nil the session/message endpoints degrade: list +// sessions falls back to the channel-session view, and the create/messages/delete +// endpoints answer with a clear 501. +type HistoryStore interface { + Append(ctx context.Context, msg history.Message) error + List(ctx context.Context, sessionID string, before int64, limit int) ([]history.Message, error) + Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error) + DeleteSession(ctx context.Context, sessionID string) error + NewSubSession(ctx context.Context, ownerID string) (string, error) + ListSessions(ctx context.Context) ([]history.SessionMeta, error) +} + // IoTStore lists IoT devices. *iot.Manager is the production implementation. type IoTStore interface { List(ctx context.Context) ([]iot.Device, error) @@ -179,6 +193,9 @@ type Deps struct { Personas PersonaStore // Memories lists memory entries. Optional. Memories MemoryStore + // History is the persistent chat-history store (design/16). Optional; when nil the + // session/message endpoints degrade (list falls back to channel sessions, others 501). + History HistoryStore // IoT lists IoT devices. Optional. IoT IoTStore // Scheduler audits sub-agent context for the subagent_ask WS route. @@ -229,6 +246,7 @@ type Server struct { sessions SessionStore personas PersonaStore memories MemoryStore + history HistoryStore iot IoTStore auditor SubagentAuditor taskOutputs TaskOutputStore @@ -270,6 +288,7 @@ func New(cfg *config.Config, deps Deps) (*Server, error) { sessions: deps.Session, personas: deps.Personas, memories: deps.Memories, + history: deps.History, iot: deps.IoT, auditor: deps.Scheduler, taskOutputs: deps.Tasks, @@ -305,6 +324,15 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /api/v1/sessions", readAuth(http.HandlerFunc(s.handleListSessions))) mux.Handle("GET /api/v1/personas", readAuth(http.HandlerFunc(s.handleListPersonas))) + // Session/model reads and writes are the owner's private history; owner/admin only + // (chat uses the same roles). Session metadata in GET /sessions stays readAuth for + // any authenticated role, but the message content (and thinking) is owner/admin-only + // per design/16 §6. Creating and deleting a sub session mutate persistent state. + historyAuth := s.auth.Authorize(auth.RoleOwner, auth.RoleAdmin) + mux.Handle("GET /api/v1/sessions/{id}/messages", historyAuth(http.HandlerFunc(s.handleListMessages))) + mux.Handle("POST /api/v1/sessions", historyAuth(http.HandlerFunc(s.handleCreateSubSession))) + mux.Handle("DELETE /api/v1/sessions/{id}", historyAuth(http.HandlerFunc(s.handleDeleteSession))) + // Realtime. mux.HandleFunc("/ws/chat", s.handleWS) @@ -425,9 +453,10 @@ func newOrchestrator(s *Server) *Orchestrator { return s.hub.runTurn(ctx, sessionID, messages, env) }, Emit: func(sessionID, typ string, data map[string]any) { - s.hub.broadcast(wsEvent{Type: typ, Data: data}) + s.hub.broadcast(wsEvent{Type: typ, SessionID: sessionID, Data: data}) }, - Log: s.log, + Log: s.log, + History: s.history, }) } diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index a8ad7e7..61204de 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -104,6 +104,10 @@ type OrchestratorOptions struct { Log *slog.Logger MaxPending int MaxTail int + // History is the optional persistent chat-history store (design/16 §3.3). + // When nil every Append is skipped, so the orchestrator runs without storage + // (tests / a gateway built without a history store). + History HistoryStore } // sessionState is the per-session message state (design/15 §2). Every mutation @@ -135,6 +139,7 @@ type Orchestrator struct { run TurnRunner emit StateSink log *slog.Logger + history HistoryStore maxPending int maxTail int } @@ -160,6 +165,7 @@ func NewOrchestrator(opts OrchestratorOptions) *Orchestrator { run: opts.Run, emit: opts.Emit, log: opts.Log, + history: opts.History, maxPending: maxPending, maxTail: maxTail, } @@ -212,6 +218,7 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes depth := len(st.pending) mode := st.mode o.mu.Unlock() + o.persistUserMessages(ctx, sessionID, messages) o.emitState(sessionID, EventPendingAck, map[string]any{"sessionId": sessionID, "mode": mode, "queueDepth": depth}) return SubmitResult{Accepted: true, Pending: true, QueueDepth: depth}, nil } @@ -225,6 +232,7 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes o.mu.Unlock() o.emitState(sessionID, EventSessionBusy, map[string]any{"sessionId": sessionID, "mode": ModeThinking}) + o.persistUserMessages(ctx, sessionID, messages) if sync { res, err := o.run(runCtx, sessionID, messages, env) @@ -257,6 +265,7 @@ func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Mes // (design/15 §2.2: incremental, not full replay). A stale generation (an // interrupt or a newer top-level turn superseded this one) is a no-op. func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Result, runErr error) error { + o.persistAssistantReply(sessionID, res) o.mu.Lock() st := o.states[sessionID] if st == nil || st.gen != gen { diff --git a/internal/services/gateway/session_orchestrator_test.go b/internal/services/gateway/session_orchestrator_test.go index df6d224..218de0a 100644 --- a/internal/services/gateway/session_orchestrator_test.go +++ b/internal/services/gateway/session_orchestrator_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) @@ -149,6 +150,89 @@ func readOnlyEnv() loop.SessionEnv { return loop.SessionEnv{MemWritable: false} } +// fakeHistory records every Append so tests assert the orchestrator persists the +// user stream and the assistant reply into the chat-history store. +type fakeHistory struct { + mu sync.Mutex + appends []history.Message +} + +func (f *fakeHistory) Append(_ context.Context, msg history.Message) error { + f.mu.Lock() + f.appends = append(f.appends, msg) + f.mu.Unlock() + return nil +} + +func (f *fakeHistory) List(context.Context, string, int64, int) ([]history.Message, error) { + return nil, nil +} + +func (f *fakeHistory) Recent(context.Context, string, int) ([]history.Message, error) { + return nil, nil +} + +func (f *fakeHistory) DeleteSession(context.Context, string) error { + return nil +} + +func (f *fakeHistory) NewSubSession(context.Context, string) (string, error) { + return "", nil +} + +func (f *fakeHistory) ListSessions(context.Context) ([]history.SessionMeta, error) { + return nil, nil +} + +func (f *fakeHistory) records() []history.Message { + f.mu.Lock() + defer f.mu.Unlock() + return append([]history.Message(nil), f.appends...) +} + +func TestSubmitPersistsUserAndAssistant(t *testing.T) { + t.Parallel() + run := newGatedRunner(1) + hist := &fakeHistory{} + o := NewOrchestrator(OrchestratorOptions{ + Run: run.Run, + Log: slog.Default(), + History: hist, + }) + run.tickets <- struct{}{} + sr, res := o.SubmitSync(context.Background(), "web", []model.Message{userMsg("你好")}, loop.OwnerEnv()) + if !sr.Accepted || !sr.Immediate { + t.Fatalf("submit = %+v, want Accepted+Immediate", sr) + } + if res == nil { + t.Fatal("sync result is nil") + } + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs)) + } + if recs[0].SessionID != "web" || recs[0].Role != history.RoleUser || recs[0].Content != "你好" { + t.Fatalf("user record = %+v", recs[0]) + } + if recs[1].SessionID != "web" || recs[1].Role != history.RoleAssistant || recs[1].Content == "" { + t.Fatalf("assistant record = %+v", recs[1]) + } +} + +func TestSubmitDoesNotPersistRejectedTurn(t *testing.T) { + t.Parallel() + hist := &fakeHistory{} + o := NewOrchestrator(OrchestratorOptions{Run: newGatedRunner(0).Run, Log: slog.Default(), History: hist, MaxPending: 1}) + // Empty turn is rejected before any persistence. + sr := o.Submit(context.Background(), "web", nil, loop.OwnerEnv()) + if sr.Accepted || !sr.Empty { + t.Fatalf("empty submit = %+v, want Empty reject", sr) + } + if n := len(hist.records()); n != 0 { + t.Fatalf("appends = %d, want 0 for a rejected turn", n) + } +} + func TestSubmitIdleStartsImmediate(t *testing.T) { t.Parallel() run := newGatedRunner(1) diff --git a/internal/services/gateway/session_persist.go b/internal/services/gateway/session_persist.go new file mode 100644 index 0000000..d507c04 --- /dev/null +++ b/internal/services/gateway/session_persist.go @@ -0,0 +1,62 @@ +package gateway + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// This file owns the orchestrator's chat-history persistence hooks (design/16 +// §3.3): the user stream is written at submit time and the assistant reply when a +// turn settles. Both are silent no-ops when no history store is wired, so the +// orchestrator runs unchanged without a store (tests / an unwired gateway). + +// persistUserMessages writes each RoleUser message of a submitted turn to the +// history store so the raw user stream is durable before the turn resolves. It is +// a no-op when no history store is wired. Rejected turns (empty, pending-full) +// never reach it, so nothing vacuous is persisted. +func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string, messages []model.Message) { + if o.history == nil { + return + } + for _, m := range messages { + if m.Role != model.RoleUser { + continue + } + text, image := messageTextImage(m) + if text == "" && image == "" { + continue + } + if err := o.history.Append(ctx, history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image}); err != nil { + o.log.Error("orchestrator persist user", "err", err, "session", sessionID) + } + } +} + +// persistAssistantReply writes the final assistant text of a settled turn to the +// history store. It is a no-op when no history store is wired or the turn +// produced no text (interrupted/errored turns leave res nil). The write runs +// under a fresh background context so a canceled turn context never aborts it. +func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result) { + if o.history == nil || res == nil || res.Text == "" { + return + } + if err := o.history.Append(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text}); err != nil { + o.log.Error("orchestrator persist assistant", "err", err, "session", sessionID) + } +} + +// messageTextImage returns the concatenated text and the first image URL of a +// message's content parts, for persistence into the chat_messages columns. +func messageTextImage(m model.Message) (text, image string) { + for _, c := range m.Content { + if c.Type == model.ContentTypeText { + text += c.Text + } else if c.Type == model.ContentTypeImage && image == "" { + image = c.ImageURL + } + } + return text, image +} diff --git a/internal/services/gateway/ws_history.go b/internal/services/gateway/ws_history.go new file mode 100644 index 0000000..67a77eb --- /dev/null +++ b/internal/services/gateway/ws_history.go @@ -0,0 +1,34 @@ +package gateway + +import ( + "context" +) + +// wsReplayLimit is the number of most-recent messages replayed to a connecting +// client. It is intentionally small so a reconnect is cheap; the frontend pulls +// the full active-session history via REST (design/16 §4.1) and the replay is +// only there to give a brand-new connection immediate context. +const wsReplayLimit = 50 + +// replayHistory sends the most recent messages for sessionID to a single, +// freshly-registered connection as a ws "history" event. It is best-effort: a +// nil store or a read error skips the replay silently. Thinking is NOT included +// (design/16 §6) so the frame never leaks chain-of-thought to a connection whose +// role the gateway did not restrict on upgrade. +func (h *hub) replayHistory(c *wsConn, sessionID string) { + if h.history == nil { + return + } + msgs, err := h.history.Recent(context.Background(), sessionID, wsReplayLimit) + if err != nil { + h.log.Warn("history replay", "err", err, "session", sessionID) + return + } + out := make([]historyMessage, 0, len(msgs)) + for _, m := range msgs { + wire := toHistoryMessage(m) + wire.Thinking = "" + out = append(out, wire) + } + c.enqueueJSON(wsEvent{Type: "history", SessionID: sessionID, Data: map[string]any{"messages": out}}) +} diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index 08d3ac8..f30c517 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -17,6 +17,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) @@ -41,10 +42,11 @@ const ( // used by the streaming path so a client can correlate a burst of streamed // events to one assistant turn and keep them in order. type wsEvent struct { - Type string `json:"type"` - TurnID string `json:"turnId,omitempty"` - Seq int `json:"seq,omitempty"` - Data any `json:"data,omitempty"` + Type string `json:"type"` + TurnID string `json:"turnId,omitempty"` + Seq int `json:"seq,omitempty"` + SessionID string `json:"sessionId,omitempty"` + Data any `json:"data,omitempty"` } // wsInbound is the client→server frame. Data carries the per-type payload. @@ -68,15 +70,16 @@ type wsConn struct { // hub owns the connection set and relays bus events to every client. type hub struct { - mu sync.RWMutex - conns map[*wsConn]struct{} - bus bus.Bus - runner Chatter - auditor SubagentAuditor - taskOutputs TaskOutputStore + mu sync.RWMutex + conns map[*wsConn]struct{} + bus bus.Bus + runner Chatter + auditor SubagentAuditor + taskOutputs TaskOutputStore voice VoiceService streaming bool orchestrator *Orchestrator + history HistoryStore log *slog.Logger upgrader websocket.Upgrader origins map[string]bool @@ -96,6 +99,7 @@ func newHub(s *Server) *hub { taskOutputs: s.taskOutputs, voice: s.voice, streaming: s.streaming, + history: s.history, log: s.log, origins: origins, allowAll: allowAll, @@ -222,7 +226,7 @@ func (h *hub) streamChat(ctx context.Context, payload any) { _ = json.Unmarshal(b, &req) } if h.runner == nil || (strings.TrimSpace(req.Text) == "" && strings.TrimSpace(req.Image) == "") { - h.broadcast(wsEvent{Type: "stream_end", TurnID: newTurnID(), Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": "idle"}}) + h.broadcast(wsEvent{Type: "stream_end", TurnID: newTurnID(), Seq: int(h.streamSeq.Add(1)), SessionID: req.SessionID, Data: map[string]any{"stopReason": "idle"}}) return } if req.SessionID == "" { @@ -248,19 +252,19 @@ func (h *hub) streamChat(ctx context.Context, payload any) { func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) { turnID := newTurnID() if h.runner == nil || len(msgs) == 0 { - h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": "idle"}}) + h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": "idle"}}) return &loop.Result{StopReason: "idle"}, nil } res, err := h.runner.Run(loop.WithSessionEnv(ctx, env), sessionID, msgs) if err != nil { h.log.Error("chat run", "err", err, "session", sessionID) - h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": "error", "error": err.Error()}}) + h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": "error", "error": err.Error()}}) return nil, err } if !h.streaming { - h.broadcast(wsEvent{Type: "stream_chunk", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"delta": res.Text}}) + h.broadcast(wsEvent{Type: "stream_chunk", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"delta": res.Text}}) } - h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": res.StopReason, "steps": res.Steps}}) + h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": res.StopReason, "steps": res.Steps}}) return res, nil } @@ -330,6 +334,7 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { claims: claims, } s.hub.register(c) + s.hub.replayHistory(c, history.MainSessionID) go c.writePump() go c.readPump() }