feat(history): 会话消息持久化 + 主/次级会话(design/16)

- runtime/history: Store + HistoryStore 接口(Append/List 游标分页/Recent/DeleteSession/NewSubSession/ListSessions),主会话固定 web、次级 web:<uuid>
- db 迁移 0003: chat_sessions + chat_messages + idx_chat_msgs_session_created(事务幂等)
- gateway: GET/POST /sessions、GET /sessions/{id}/messages 热分页、DELETE /sessions/{id}(主不可删 403);Deps.History 可空注入
- 编排层落库: session_persist.go(submitInternal 写用户消息、completeTurn 写助手消息,拒绝的消息不落库)
- WS: wsEvent 加 sessionId 定向 + history 回放帧(握手上行后回放主会话最新 50 条,thinking 剥离)
- wire.go 装配 History 到 gateway.Deps.History
- 端到端实测: POST /sessions 建次级 201 / GET /sessions 列表 / POST /chat 后 /messages 读到 user+assistant / DELETE 次级成功 / DELETE 主会话 403
This commit is contained in:
Aska committed 2026-09-07 03:52:43 +08:00
1 parent cc3eecacac
commit 48421923f6
15 files changed
+1305 -18

No files matched your search

+8
View File
@@ -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,
@@ -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;
@@ -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:<uuid>"
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. <p>/<action>)
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);
+390
View File
@@ -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:<uuid>"). 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)`
+204
View File
@@ -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)
}
}
+27
View File
@@ -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()
+165
View File
@@ -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
@@ -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)
}
}
+8 -1
View File
@@ -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())
+31 -2
View File
@@ -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,
})
}
@@ -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 {
@@ -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)
@@ -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
}
+34
View File
@@ -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}})
}
+20 -15
View File
@@ -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()
}