feat(gateway): admin 列表端点接真实数据——persona/session/memory 补 List() 并注入 gateway.Deps
根因: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}。
This commit is contained in:
1 parent
5dae2b9741
commit
04bfa2c7a7
11 files changed
+312
-18
No files matched your search
+2
-2
@@ -49,6 +49,6 @@ node_modules/
|
||||
dist/
|
||||
|
||||
# 开发/运行本地产出(不入库)
|
||||
# data/ = SQLite 数据目录(DATA_DIR=./data);cyrene = 根目录构建产物
|
||||
# data/ = SQLite 数据目录(DATA_DIR=./data);/cyrene = 根目录构建产物(锚定,不误伤 cmd/cyrene)
|
||||
data/
|
||||
cyrene
|
||||
/cyrene
|
||||
+5
-1
@@ -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),
|
||||
})
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user