From 04bfa2c7a74b09ac471ea63981ce918365f2b771 Mon Sep 17 00:00:00 2001 From: Aska Date: Sun, 6 Sep 2026 14:05:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway):=20admin=20=E5=88=97=E8=A1=A8?= =?UTF-8?q?=E7=AB=AF=E7=82=B9=E6=8E=A5=E7=9C=9F=E5=AE=9E=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E2=80=94=E2=80=94persona/session/memory=20=E8=A1=A5=20List()?= =?UTF-8?q?=20=E5=B9=B6=E6=B3=A8=E5=85=A5=20gateway.Deps?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因：gateway.Server 定义 SessionStore/PersonaStore/MemoryStore 接口但生产从未满足： - 底层 manager 无 List() 方法； - wireFrontend 的 gateway.Deps{} 未传 Session/Personas/Memories。 → /api/v1/sessions|personas 与 /api/v1/admin/{sessions,personas,memories} 恒返回 []。 改动： - persona/registry：Registry.List() []Character（并发安全、值拷贝）。 - channel/session：SessionManager.List(ctx)（查 channel_sessions 表+按 actor_role 重建）。 - runtime/memory：Store.List(ctx)（全表，复用 scanItems）。 - gateway 三 Store 接口改为返回域类型（Character/MemoryItem），映射统一放 collect* 侧。 - cmd/cyrene/wire.go：Deps 补传 Session/Personas/Memories。 - .gitignore：cyrene→/cyrene（锚定根，不再误伤 cmd/cyrene 目录）。 实机验证：/api/v1/admin/personas 返回 {id:cyrene,name:昔涟,isDefault:true}。 --- .gitignore | 4 +- cmd/cyrene/wire.go | 6 +- internal/channel/session/manager.go | 58 +++++++++++++- internal/channel/session/manager_test.go | 88 ++++++++++++++++++++++ internal/persona/registry/registry.go | 18 +++++ internal/persona/registry/registry_test.go | 47 ++++++++++++ internal/runtime/memory/store.go | 15 ++++ internal/runtime/memory/store_test.go | 39 ++++++++++ internal/services/gateway/admin.go | 4 +- internal/services/gateway/rest.go | 25 +++++- internal/services/gateway/server.go | 26 ++++--- 11 files changed, 312 insertions(+), 18 deletions(-) diff --git a/.gitignore b/.gitignore index 946c8e2..defa9e4 100644 --- a/.gitignore +++ b/.gitignore @@ -49,6 +49,6 @@ node_modules/ dist/ # 开发/运行本地产出（不入库） -# data/ = SQLite 数据目录（DATA_DIR=./data）；cyrene = 根目录构建产物 +# data/ = SQLite 数据目录（DATA_DIR=./data）；/cyrene = 根目录构建产物（锚定，不误伤 cmd/cyrene） data/ -cyrene +/cyrene diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 0ec8679..847edca 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -399,7 +399,8 @@ func (a *app) wireChannels() error { } // wireFrontend builds the gateway with the embedded/external SPA filesystem and -// the JWT/RBAC authenticator. +// the JWT/RBAC authenticator. Session/Persona/Memories are wired to the concrete +// managers so the admin/user list endpoints return real data. func (a *app) wireFrontend() error { jwt := auth.New(a.cfg.Auth) gw, err := gateway.New(a.cfg, gateway.Deps{ @@ -408,6 +409,9 @@ func (a *app) wireFrontend() error { Bus: a.bus, Registry: a.caps, Loop: a.loop, + Session: a.sessions, + Personas: a.personas, + Memories: a.memStore, IoT: a.iot, Frontend: resolveFrontendFS(a.log), }) diff --git a/internal/channel/session/manager.go b/internal/channel/session/manager.go index e8fe7ad..8d3e166 100644 --- a/internal/channel/session/manager.go +++ b/internal/channel/session/manager.go @@ -5,6 +5,7 @@ import ( "database/sql" "fmt" "log/slog" + "strings" "sync" "time" @@ -40,6 +41,18 @@ VALUES (?, ?, ?, ?, ?, ?)` const selectSessionSQL = ` SELECT session_id FROM channel_sessions WHERE platform = ? AND channel_id = ?` +// listSessionsSQL selects every persisted channel mapping, newest first. Group +// sessions key on a channel_id carrying groupChannelPrefix; direct sessions use +// their channel id. +const listSessionSQL = ` +SELECT platform, channel_id, session_id, actor_role FROM channel_sessions +ORDER BY created_at DESC` + +// groupChannelPrefix marks a channel id as a group conversation so a session +// reconstructed from the backing table (where no IsGroup flag is persisted) can +// still report group membership. +const groupChannelPrefix = "group:" + // SessionManager owns the channel↔session mapping. It backs the map with the // SQLite table channel_sessions so a channel keeps its internal session across // restarts, and it caches lookups in memory to avoid a query per message. The @@ -109,6 +122,49 @@ func (m *SessionManager) GetOrCreate(ctx context.Context, msg adapter.UnifiedMes return s, nil } +// List returns every known session, newest first, across the backing table and +// the in-memory cache. Rows that are still in the cache are returned with their +// full derived metadata (Owner/IsGroup/MemWritable from the original resolve); +// rows recovered from the table (for example from a previous process) are +// reconstructed from the persisted actor role. The returned slice is non-nil. It +// is safe for concurrent use. +func (m *SessionManager) List(ctx context.Context) ([]Session, error) { + rows, err := m.db.QueryContext(ctx, listSessionSQL) + if err != nil { + return nil, fmt.Errorf("session: list: %w", err) + } + defer func() { _ = rows.Close() }() + + sessions := make([]Session, 0) + for rows.Next() { + var platform, cid, sessionID, roleStr string + if err := rows.Scan(&platform, &cid, &sessionID, &roleStr); err != nil { + return nil, fmt.Errorf("session: scan list: %w", err) + } + key := platform + "\x00" + cid + if cached, ok := m.lookupCached(key); ok { + sessions = append(sessions, cached) + continue + } + role := adapter.Role(roleStr) + owner := role == adapter.RoleOwner + isGroup := strings.HasPrefix(cid, groupChannelPrefix) + sessions = append(sessions, Session{ + Platform: platform, + ChannelID: cid, + SessionID: sessionID, + ActorRole: role, + Owner: owner, + IsGroup: isGroup, + MemWritable: owner && !isGroup, + }) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("session: iterate list: %w", err) + } + return sessions, nil +} + // lookupCached returns the cached session for key under a read lock. func (m *SessionManager) lookupCached(key string) (Session, bool) { m.mu.RLock() @@ -144,7 +200,7 @@ func (m *SessionManager) persist(ctx context.Context, platform, cid string, acto func channelID(msg adapter.UnifiedMessage) string { if msg.IsGroup() { if msg.GroupID != "" { - return "group:" + msg.GroupID + return groupChannelPrefix + msg.GroupID } return msg.ChannelID } diff --git a/internal/channel/session/manager_test.go b/internal/channel/session/manager_test.go index da564b1..f2cb45f 100644 --- a/internal/channel/session/manager_test.go +++ b/internal/channel/session/manager_test.go @@ -207,6 +207,94 @@ func TestGetOrCreateConcurrent(t *testing.T) { } } +func TestList(t *testing.T) { + t.Parallel() + + m, _ := newTestManager(t) + ctx := context.Background() + + if _, err := m.GetOrCreate(ctx, directMsg("o-100"), testOwner()); err != nil { + t.Fatalf("GetOrCreate(owner): %v", err) + } + if _, err := m.GetOrCreate(ctx, directMsg("stranger-1"), testOwner()); err != nil { + t.Fatalf("GetOrCreate(stranger): %v", err) + } + if _, err := m.GetOrCreate(ctx, groupMsg("g-member-1"), testOwner()); err != nil { + t.Fatalf("GetOrCreate(group): %v", err) + } + + got, err := m.List(ctx) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(got) != 3 { + t.Fatalf("List = %d sessions, want 3", len(got)) + } + var owner, stranger, group bool + for _, s := range got { + switch s.ChannelID { + case "direct-o-100": + if !s.Owner || !s.MemWritable || s.IsGroup { + t.Errorf("owner session flags = %+v, want owner/writable/not-group", s) + } + owner = true + case "direct-stranger-1": + if s.Owner || s.MemWritable || s.IsGroup { + t.Errorf("stranger session flags = %+v, want non-owner/read-only/not-group", s) + } + stranger = true + case groupChannelPrefix + "g1": + if !s.IsGroup || s.MemWritable { + t.Errorf("group session flags = %+v, want group/read-only", s) + } + group = true + } + } + if !owner || !stranger || !group { + t.Fatalf("missing sessions: owner=%v stranger=%v group=%v", owner, stranger, group) + } +} + +func TestListRecoveredFromTable(t *testing.T) { + t.Parallel() + + ctx := context.Background() + handle, err := db.Open(ctx, ":memory:", nil) + if err != nil { + t.Fatalf("db.Open = %v", err) + } + t.Cleanup(func() { _ = handle.Close() }) + + m1, err := New(ctx, handle, nil) + if err != nil { + t.Fatalf("New#1 = %v", err) + } + if _, err := m1.GetOrCreate(ctx, directMsg("o-100"), testOwner()); err != nil { + t.Fatalf("m1.GetOrCreate = %v", err) + } + + // A brand-new manager over the same DB has an empty cache: List must + // reconstruct the session from the backing table (cross-restart path). + m2, err := New(ctx, handle, nil) + if err != nil { + t.Fatalf("New#2 = %v", err) + } + got, err := m2.List(ctx) + if err != nil { + t.Fatalf("List = %v", err) + } + if len(got) != 1 { + t.Fatalf("List = %d sessions, want 1", len(got)) + } + s := got[0] + if s.ChannelID != "direct-o-100" || s.SessionID == "" { + t.Fatalf("recovered session = %+v, want direct-o-100", s) + } + if !s.Owner || !s.MemWritable { + t.Errorf("recovered owner flags = %+v, want owner/writable", s) + } +} + func TestGetOrCreateInvalidChannel(t *testing.T) { t.Parallel() diff --git a/internal/persona/registry/registry.go b/internal/persona/registry/registry.go index 82b357b..478fa72 100644 --- a/internal/persona/registry/registry.go +++ b/internal/persona/registry/registry.go @@ -7,6 +7,7 @@ package registry import ( "fmt" + "sort" "sync" "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/emotion" @@ -184,3 +185,20 @@ func (r *Registry) CountCharacters() int { defer r.mu.RUnlock() return len(r.characters) } + +// List returns every registered character, sorted by ID. Each returned Character +// is a copy so callers cannot mutate the registry's internal state. It is safe +// for concurrent use. +func (r *Registry) List() []Character { + r.mu.RLock() + defer r.mu.RUnlock() + + out := make([]Character, 0, len(r.characters)) + for _, c := range r.characters { + ch := *c + ch.Modes = append([]string(nil), c.Modes...) + out = append(out, ch) + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out +} diff --git a/internal/persona/registry/registry_test.go b/internal/persona/registry/registry_test.go index 0b93567..e3c5cd0 100644 --- a/internal/persona/registry/registry_test.go +++ b/internal/persona/registry/registry_test.go @@ -167,6 +167,53 @@ func TestRegistry_EmotionsFollowPersona(t *testing.T) { } } +func TestRegistry_List(t *testing.T) { + t.Parallel() + + r := New() + // An empty registry lists nothing but stays non-nil. + if got := r.List(); got == nil || len(got) != 0 { + t.Fatalf("List() on empty registry = %#v, want empty non-nil", got) + } + + if err := r.RegisterPersona(mkPersona("pb", "c", "")); err != nil { + t.Fatalf("RegisterPersona(pb): %v", err) + } + if err := r.RegisterPersona(mkPersona("pa", "c", "")); err != nil { + t.Fatalf("RegisterPersona(pa): %v", err) + } + if err := r.RegisterCharacter(mkChar("cb", "pb", false)); err != nil { + t.Fatalf("RegisterCharacter(cb): %v", err) + } + if err := r.RegisterCharacter(mkChar("ca", "pa", true)); err != nil { + t.Fatalf("RegisterCharacter(ca): %v", err) + } + + got := r.List() + if len(got) != 2 { + t.Fatalf("List() = %d characters, want 2", len(got)) + } + // Sorted by ID: ca before cb. + if got[0].ID != "ca" || got[1].ID != "cb" { + t.Fatalf("List() order = %q,%q, want ca,cb", got[0].ID, got[1].ID) + } + if got[0].PersonaID != "pa" || !got[0].Default || got[1].Default { + t.Fatalf("default/persona refs wrong: %+v", got) + } + if len(got[0].Modes) != 2 || got[0].Modes[0] != "chat" { + t.Fatalf("Modes = %v, want [chat home]", got[0].Modes) + } + + // The returned slice is a copy: mutating it must not touch the registry. + got[0].Modes[0] = "hacked" + got[0].Name = "hacked" + // Re-read and assert the registry is unchanged. + ch, ok := r.Character("ca") + if !ok || ch.Name != "ca" || ch.Modes[0] != "chat" { + t.Fatalf("registry mutated through List copy: %+v", ch) + } +} + func TestRegistry_ConcurrentLifecycle(t *testing.T) { t.Parallel() diff --git a/internal/runtime/memory/store.go b/internal/runtime/memory/store.go index 7803b61..acd2911 100644 --- a/internal/runtime/memory/store.go +++ b/internal/runtime/memory/store.go @@ -246,6 +246,21 @@ func (s *Store) ListByPersona(ctx context.Context, personaID string) ([]MemoryIt return scanItems(rows, "persona item") } +// List returns every memory item, newest first, across all personas. It is the +// aggregate/admin view that enumerates all persisted memory regardless of the +// persona-isolation boundary; callers that must respect isolation should use +// ListByPersona or ListScope instead. The returned slice is non-nil. +func (s *Store) List(ctx context.Context) ([]MemoryItem, error) { + rows, err := s.db.QueryContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind, + content, tags, confidence, source_conversation, created_at, updated_at, + decay_weight FROM memory_items + ORDER BY created_at DESC`) + if err != nil { + return nil, fmt.Errorf("memory: list items: %w", err) + } + return scanItems(rows, "item") +} + // scanItems reads every row into a MemoryItem slice, closing rows and wrapping // scan/iterate failures with the operation name. A non-nil, possibly empty slice // is always returned on success. diff --git a/internal/runtime/memory/store_test.go b/internal/runtime/memory/store_test.go index 161b48a..b219659 100644 --- a/internal/runtime/memory/store_test.go +++ b/internal/runtime/memory/store_test.go @@ -171,6 +171,45 @@ func TestStore_ListByPersona_MissingPersona_异常(t *testing.T) { } } +func TestStore_List_正常(t *testing.T) { + t.Parallel() + ctx := context.Background() + s := newTestStore(t) + + // An empty store lists nothing but stays non-nil. + if got, err := s.List(ctx); err != nil || got == nil || len(got) != 0 { + t.Fatalf("List() on empty store = %#v, err=%v; want empty non-nil", got, err) + } + + for _, item := range []MemoryItem{ + {PersonaID: "persona.a", Content: "A 的记忆", Scope: ScopePersona, Kind: KindRelationship}, + {PersonaID: "persona.b", Content: "B 的记忆", Scope: ScopePersona, Kind: KindWorldEvent}, + {PersonaID: "persona.a", Content: "项目记忆", Scope: ScopeProject, Kind: KindWorldEvent, ProjectID: "proj.1"}, + } { + if _, err := s.Add(ctx, item); err != nil { + t.Fatalf("Add(%s): %v", item.Content, err) + } + } + + got, err := s.List(ctx) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(got) != 3 { + t.Fatalf("List = %d items, want 3", len(got)) + } + // Newest first (created_at DESC). + if got[0].Content != "项目记忆" { + t.Errorf("got[0].Content = %q, want the latest added (项目记忆)", got[0].Content) + } + if got[2].Content != "A 的记忆" { + t.Errorf("got[2].Content = %q, want the oldest (A 的记忆)", got[2].Content) + } + if got[0].Scope != ScopeProject || got[0].PersonaID != "persona.a" { + t.Errorf("project item fields = %+v", got[0]) + } +} + func TestStore_Delete_正常(t *testing.T) { t.Parallel() ctx := context.Background() diff --git a/internal/services/gateway/admin.go b/internal/services/gateway/admin.go index ace2468..85bd2c6 100644 --- a/internal/services/gateway/admin.go +++ b/internal/services/gateway/admin.go @@ -156,7 +156,9 @@ func (s *Server) collectMemories(ctx context.Context) []MemorySummary { s.log.Error("list memories", "err", err) return out } - out = append(out, list...) + for _, it := range list { + out = append(out, MemorySummary{ID: it.ID, Scope: string(it.Scope), Text: it.Content}) + } return out } diff --git a/internal/services/gateway/rest.go b/internal/services/gateway/rest.go index 8c81c30..3033c03 100644 --- a/internal/services/gateway/rest.go +++ b/internal/services/gateway/rest.go @@ -158,12 +158,33 @@ func (s *Server) handleListPersonas(w http.ResponseWriter, r *http.Request) { // collectPersonas returns a non-nil persona slice from the configured store. func (s *Server) collectPersonas() []PersonaSummary { out := make([]PersonaSummary, 0) - if s.personas != nil { - out = append(out, s.personas.List()...) + if s.personas == nil { + return out + } + for _, c := range s.personas.List() { + id := c.PersonaID + if id == "" { + id = c.ID + } + out = append(out, PersonaSummary{ + ID: id, + Name: c.Name, + Character: c.Name, + IsDefault: c.Default, + Mode: firstMode(c.Modes), + }) } return out } +// firstMode returns the first declared behavior mode, or "" when none are set. +func firstMode(modes []string) string { + if len(modes) == 0 { + return "" + } + return modes[0] +} + // summaryOfSession maps a session.Session to its wire shape. func summaryOfSession(sess session.Session) SessionSummary { return SessionSummary{ diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 7d86099..bca5307 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -34,7 +34,9 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/config" "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) @@ -69,23 +71,24 @@ type Chatter interface { Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error) } -// SessionStore lists known channel sessions. The concrete session manager does -// not yet expose a list method, so production wiring may leave Sessions nil and -// the handlers return []. +// SessionStore lists known channel sessions. *session.SessionManager is the +// production implementation. type SessionStore interface { List(ctx context.Context) ([]session.Session, error) } -// PersonaStore lists persona summaries. The concrete persona registry does not -// yet expose a list method, so production wiring may leave Personas nil. +// PersonaStore lists registered characters (each bound one-to-one to a persona). +// *persona/registry.Registry is the production implementation. The gateway maps +// each Character to its wire summary; the interface deliberately returns the +// domain type so the wire shape stays out of the store contract. type PersonaStore interface { - List() []PersonaSummary + List() []personareg.Character } -// MemoryStore lists memory entries. The concrete memory store does not yet -// expose a list method, so production wiring may leave Memories nil. +// MemoryStore lists memory entries. *runtime/memory.Store is the production +// implementation. The gateway maps each MemoryItem to its wire summary. type MemoryStore interface { - List(ctx context.Context) ([]MemorySummary, error) + List(ctx context.Context) ([]memory.MemoryItem, error) } // IoTStore lists IoT devices. *iot.Manager is the production implementation. @@ -96,8 +99,9 @@ type IoTStore interface { // Deps carries the gateway's injected dependencies. Every field is nil-tolerant // except Auth, which is required (authentication backs login and every RBAC // decision). Registry/Loop are satisfied by the concrete *capability.Registry and -// *loop.Loop; Sessions/Personas/Memories remain nil until their managers expose a -// list method, in which case the handlers return []. +// *loop.Loop; Sessions/Personas/Memories are satisfied by the concrete session +// manager, persona registry and memory store, and the handlers return [] only +// when a dependency is left nil. type Deps struct { // Log is the base logger. Nil uses a discard logger. Log *slog.Logger