feat: 管理面板补齐(RAG导入/记忆添加/会话接history/诚实表述) + design/16 §4.6 IM多会话隔离+第三方身份+黑白名单
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

agent-98 面板批次:
- 会话管理面板改接 history.ListSessions(主web+次级) + 新建次级会话按钮
- 记忆面板手动添加入口 POST /admin/memories(owner)
- 知识库真正实现 IndexDocument(分块+落projectindex) + POST /admin/rag 导入端点
- 三面板空态诚实表述(去除'后端M10+提供')
- RAG嵌入降级到keyword/line(无直属embedder,不伪造语义,符合红线)

agent-99 design/16 §4.6:
- model.Message.SenderName + 迁移0006 sender_name 列
- 第三方身份边界: session.ResolveActor判定(非LLM猜) + [第三方·昵称]前缀(不改源content)
- 群聊/陌生人落独立sess-xxx(persistIsolated),不写主会话,不广播
- 黑/白名单(allowGroups/denyGroups/allowUsers/denyUsers)最前拦截,不进LLM
- 多账号 CHANNELS_JSON 配置 + 每平台多实例
- 管理只读 GET /admin/channels/{platform}/messages(该平台非主人消息流)

实机验证(JWT): 会话管理返回主会话web / 记忆添加成功 / RAG导入documents非空+Cyrene命中; 后端build/vet/test全绿 + 前端build绿
This commit is contained in:
Aska committed 2026-09-07 05:13:36 +08:00
1 parent 37e98ec584
commit 9c2a37c4ac
32 files changed
+1768 -136

No files matched your search

+34
View File
@@ -0,0 +1,34 @@
package config
import (
"encoding/json"
"fmt"
"strings"
)
// ParseChannelAccounts decodes the CHANNELS_JSON value into a per-platform list
// of ChannelAccount. The wire shape is map[platform]→[]ChannelAccount, for
// example:
//
// {"telegram":[{"id":"conf0","config":{"bot_token":"..."},"denyUsers":["12345"]}]}
//
// An empty/blank input yields an empty (non-nil) map so a caller can iterate it
// unconditionally. A malformed JSON or a non-array account value returns an
// error so a bad config fails loudly at boot rather than silently dropping
// channels.
func ParseChannelAccounts(raw string) (map[string][]ChannelAccount, error) {
if strings.TrimSpace(raw) == "" {
return map[string][]ChannelAccount{}, nil
}
var m map[string][]ChannelAccount
dec := json.NewDecoder(strings.NewReader(raw))
if err := dec.Decode(&m); err != nil {
return nil, fmt.Errorf("config: decode CHANNELS_JSON: %w", err)
}
// Tolerate a top-level array of accounts that all share a blank platform name
// by normalizing a nil map to an empty one.
if m == nil {
return map[string][]ChannelAccount{}, nil
}
return m, nil
}
+61
View File
@@ -0,0 +1,61 @@
package config
import (
"reflect"
"testing"
)
func TestParseChannelAccounts_Empty(t *testing.T) {
t.Parallel()
got, err := ParseChannelAccounts("")
if err != nil {
t.Fatalf("ParseChannelAccounts(empty): %v", err)
}
if len(got) != 0 {
t.Fatalf("empty input = %d platforms, want 0", len(got))
}
}
func TestParseChannelAccounts_MultiAccountWithAllowDeny(t *testing.T) {
t.Parallel()
raw := `{
"telegram": [
{"id":"conf0","config":{"bot_token":"t0"},"denyUsers":["12345"],"denyGroups":["-100"]},
{"id":"conf1","config":{"bot_token":"t1"},"allowUsers":["u1","u2"],"allowGroups":["g1"]}
],
"qq-obv11": [
{"id":"bot0","config":{"remote_url":"http://x","mode":"client"}}
]
}`
got, err := ParseChannelAccounts(raw)
if err != nil {
t.Fatalf("ParseChannelAccounts: %v", err)
}
if len(got) != 2 {
t.Fatalf("platforms = %d, want 2", len(got))
}
tg := got["telegram"]
if len(tg) != 2 {
t.Fatalf("telegram accounts = %d, want 2", len(tg))
}
if tg[0].ID != "conf0" || tg[0].Config["bot_token"] != "t0" {
t.Fatalf("account[0] = %+v", tg[0])
}
if !reflect.DeepEqual(tg[0].DenyUsers, []string{"12345"}) || !reflect.DeepEqual(tg[0].DenyGroups, []string{"-100"}) {
t.Fatalf("account[0] deny lists = %+v", tg[0])
}
if !reflect.DeepEqual(tg[1].AllowUsers, []string{"u1", "u2"}) || !reflect.DeepEqual(tg[1].AllowGroups, []string{"g1"}) {
t.Fatalf("account[1] allow lists = %+v", tg[1])
}
qq := got["qq-obv11"]
if len(qq) != 1 || qq[0].ID != "bot0" || qq[0].Config["mode"] != "client" {
t.Fatalf("qq accounts = %+v", qq)
}
}
func TestParseChannelAccounts_Malformed(t *testing.T) {
t.Parallel()
if _, err := ParseChannelAccounts(`{`); err == nil {
t.Fatal("expected error for malformed JSON")
}
}
+41
View File
@@ -33,6 +33,7 @@ type Config struct {
ExecSandbox ExecSandbox
MCP MCPConfig
Capacity CapacityConfig
Channels ChannelsConfig
}
// ServerConfig holds the single gateway listener settings.
@@ -154,6 +155,46 @@ type MCPServer struct {
Enabled bool `json:"enabled"`
}
// ChannelsConfig holds optional multi-account IM channel settings (design/16
// §4.6 §5). The single-account credentials are read from CHANNEL_<NAME>_* env
// vars in cmd/cyrene/channels.go; this block carries the optional multi-account
// registry plus the per-account allow/deny black-or-white lists via a JSON
// string (the reflection-based loader only supports scalars and []string, so a
// structured value is passed as JSON — matching the MCP and Capacity configs).
type ChannelsConfig struct {
// AccountsJSON is the JSON shape map[platform]→[]ChannelAccount, e.g.
// {"telegram":[{"id":"conf0","config":{"bot_token":"..."},
// "denyUsers":["12345"]}]}. Empty means the single-account env path applies.
AccountsJSON string `env:"CHANNELS_JSON"`
}
// ChannelAccount is one IM access config (one account) for a platform. It pairs
// an id (used to name the adapter instance and to build the platform:confId
// account key) with the platform-specific credential map and the allow/deny
// black-or-white lists applied at the channelbridge ingress (design/16 §4.6), so
// a blacklisted or non-whitelisted group/user never reaches the LLM.
type ChannelAccount struct {
// ID is the local config id, unique within its platform. It names the adapter
// instance and the platform:confId account key (design/16 §4.5 deviceID).
ID string `json:"id"`
// Config is the platform-specific credential/feature map decoded into the
// adapter's typed Config (bot_token, base_url, remote_url, ...). Missing keys
// fall back to the adapter's own zero values.
Config map[string]any `json:"config,omitempty"`
// AllowGroups, when non-empty, is a white list of group ids: a group message
// is dropped unless its GroupID is in this list.
AllowGroups []string `json:"allowGroups,omitempty"`
// DenyGroups is a black list of group ids: a group message whose GroupID is in
// this list is dropped at ingress.
DenyGroups []string `json:"denyGroups,omitempty"`
// AllowUsers, when non-empty, is a white list of user uids: a direct (private)
// message from a user not in this list is dropped.
AllowUsers []string `json:"allowUsers,omitempty"`
// DenyUsers is a black list of user uids: a direct message from a user in this
// list is dropped at ingress.
DenyUsers []string `json:"denyUsers,omitempty"`
}
// Load reads configuration, merging an optional .env file (development only)
// with process environment variables. Process environment variables always take
// precedence over .env values, which take precedence over code defaults.
@@ -0,0 +1 @@
ALTER TABLE chat_messages DROP COLUMN sender_name;
@@ -0,0 +1,13 @@
-- 0006_chat_sender_name: tag every chat message with the human-readable sender
-- display name (DevDocs/design/16 §4.6). Group chats and stranger private chats
-- carry the sender's nickname so the agent can see who is speaking and the IM
-- adapter log page can label the message stream; the owner's private chat on the
-- main session is intentionally left blank (the owner is never labeled).
--
-- The 0005 chat_messages table already exists at this point, so ALTER ADD COLUMN is
-- the correct, non-destructive step. The canonical schema (including this column) is
-- also re-issued idempotently by internal/runtime/history.NewStore so a fresh
-- in-memory store used by tests stays self-contained; the migration is the
-- authoritative path for an existing database.
ALTER TABLE chat_messages ADD COLUMN sender_name TEXT NOT NULL DEFAULT '';
+61 -10
View File
@@ -86,8 +86,15 @@ type Message struct {
// that sent the message (design/16 §4.5) — e.g. "web:chrome:a1b2c3" for a
// browser, "weixin-ilink:conf0:wx_uid_123" for IM. It complements Platform: two
// clients on the same platform share Platform but differ in DeviceID.
DeviceID string
CreatedAt int64
DeviceID string
// SenderName is the human-readable display name of the message's sender
// (design/16 §4.6). It is set for third-party messages (group chats and
// stranger private chats) so the agent and the IM adapter log page can label
// who is speaking; it is intentionally blank for the owner's private chat on
// the main session (the owner is never labeled). It is display metadata only —
// identity is never decided from this field.
SenderName string
CreatedAt int64
}
// SessionMeta is the wire-facing metadata of one persistent session.
@@ -117,6 +124,12 @@ type HistoryStore interface {
// 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)
// ListPlatformLog returns up to limit messages for an IM platform across every
// non-main (isolated) session, newest first, for the adapter log page (design/16
// §4.6 §4). It excludes the fixed main web session, so the platform's non-owner
// message stream (groups + stranger private chats) is surfaced without the
// owner-private mirror. before is a millisecond cursor; 0 means newest page.
ListPlatformLog(ctx context.Context, platform string, before int64, limit 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
@@ -199,10 +212,10 @@ func (s *Store) insert(ctx context.Context, msg Message) (Message, error) {
msg.Platform = PlatformWeb
}
_, err := s.db.ExecContext(ctx, `INSERT INTO chat_messages
(id, session_id, role, content, mode, mood, image, thinking, platform, device_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
(id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
msg.ID, msg.SessionID, msg.Role, msg.Content, msg.Mode, msg.Mood,
msg.Image, msg.Thinking, msg.Platform, msg.DeviceID, msg.CreatedAt)
msg.Image, msg.Thinking, msg.Platform, msg.DeviceID, msg.SenderName, msg.CreatedAt)
if err != nil {
return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err)
}
@@ -218,7 +231,7 @@ func (s *Store) insert(ctx context.Context, msg Message) (Message, error) {
// limit is clamped to [1, maxPageLimit].
func (s *Store) List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]Message, error) {
limit = clampLimit(limit)
query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, created_at
query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, created_at
FROM chat_messages WHERE session_id = ?`
args := []any{sessionID}
if before > 0 {
@@ -252,13 +265,50 @@ func (s *Store) List(ctx context.Context, sessionID string, before int64, limit
return msgs, nil
}
// ListPlatformLog returns up to limit messages for platform across every non-main
// (isolated) session, newest first. Main-session messages are excluded so only the
// platform's own message stream (groups + stranger private chats) is surfaced to the
// adapter log page (design/16 §4.6 §4). before is a millisecond cursor; when nonzero
// only strictly older messages are returned. The returned slice is non-nil.
func (s *Store) ListPlatformLog(ctx context.Context, platform string, before int64, limit int) ([]Message, error) {
limit = clampLimit(limit)
query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, created_at
FROM chat_messages WHERE platform = ? AND session_id != ?`
args := []any{platform, MainSessionID}
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: platform log %s: %w", platform, 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 platform log %s: %w", platform, err)
}
msgs = append(msgs, m)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("history: iterate platform log %s: %w", platform, 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, platform, device_id, created_at
rows, err := s.db.QueryContext(ctx, `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, 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)
@@ -372,11 +422,11 @@ func (s *Store) ensureMainSession(ctx context.Context) error {
// 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, platform, deviceID string
if err := sc.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &mode, &mood, &image, &thinking, &platform, &deviceID, &m.CreatedAt); err != nil {
var mode, mood, platform, deviceID, senderName string
if err := sc.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &mode, &mood, &m.Image, &m.Thinking, &platform, &deviceID, &senderName, &m.CreatedAt); err != nil {
return Message{}, err
}
m.Mode, m.Mood, m.Image, m.Thinking, m.Platform, m.DeviceID = mode, mood, image, thinking, platform, deviceID
m.Mode, m.Mood, m.Platform, m.DeviceID, m.SenderName = mode, mood, platform, deviceID, senderName
return m, nil
}
@@ -433,6 +483,7 @@ CREATE TABLE IF NOT EXISTS chat_messages (
thinking TEXT,
platform TEXT NOT NULL DEFAULT 'web',
device_id TEXT NOT NULL DEFAULT '',
sender_name TEXT NOT NULL DEFAULT '',
created_at INTEGER NOT NULL
)`
+56
View File
@@ -367,3 +367,59 @@ func TestDeleteSession(t *testing.T) {
t.Fatalf("delete main err = %v, want ErrImmutableMain", err)
}
}
func TestSenderNameRoundTripOnListRecentAndPlatformLog(t *testing.T) {
t.Parallel()
st := newStore(t)
ctx := context.Background()
// An isolated group session carries the sender display name (design/16 §4.6).
groupID := "sess-g"
if err := st.Append(ctx, Message{SessionID: groupID, Role: RoleUser, Content: "hello all", Platform: "telegram", SenderName: "张三", DeviceID: "telegram:-20:9", CreatedAt: 1000}); err != nil {
t.Fatalf("append group msg: %v", err)
}
// An owner-private message on the main session carries no sender name.
if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "hi", Platform: "telegram", CreatedAt: 900}); err != nil {
t.Fatalf("append main msg: %v", err)
}
// An unrelated platform on the main session should not leak into the log.
if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "web msg", Platform: "web", CreatedAt: 800}); err != nil {
t.Fatalf("append web msg: %v", err)
}
// List reads the sender_name back.
ls, err := st.List(ctx, groupID, 0, 50, "")
if err != nil {
t.Fatalf("list: %v", err)
}
if len(ls) != 1 || ls[0].SenderName != "张三" || ls[0].Content != "hello all" {
t.Fatalf("list = %+v, want SenderName 张三", ls)
}
// Recent reads it too.
rs, err := st.Recent(ctx, groupID, 5)
if err != nil {
t.Fatalf("recent: %v", err)
}
if len(rs) != 1 || rs[0].SenderName != "张三" {
t.Fatalf("recent = %+v, want SenderName 张三", rs)
}
// Platform log surfaces only the isolated (non-main) messages of the platform,
// newest first, excluding the owner-private mirror on the main session.
log, err := st.ListPlatformLog(ctx, "telegram", 0, 50)
if err != nil {
t.Fatalf("platform log: %v", err)
}
if len(log) != 1 {
t.Fatalf("platform log = %+v, want only the isolated group message", log)
}
if log[0].SessionID != groupID || log[0].SenderName != "张三" {
t.Fatalf("platform log[0] = %+v, want group session with SenderName 张三", log[0])
}
// A platform with only main-session messages yields an empty log.
if log, err := st.ListPlatformLog(ctx, "web", 0, 50); err != nil || len(log) != 0 {
t.Fatalf("web platform log = %+v, err=%v; want empty", log, err)
}
}
+125 -12
View File
@@ -10,28 +10,36 @@
// so a configured vector index (pgvector / SQLite vector) is genuinely used and
// never a shell.
//
// The store is read-only for this milestone: upload/chunk/embed and delete land
// in M10. The write methods return ErrWriteUnsupported so the admin handler can
// answer honestly instead of fabricating a success.
// The store persists knowledge documents by indexing them into the project index
// (项目索引 → 项目描述 → 章节内容): IndexDocument chunks a document's content and
// stores it as a project plus its chapters, making it text/keyword-searchable
// immediately. Semantic/vector embedding of an imported document is a graceful
// no-op here (this data source holds no vector index); the existing semantic path
// still ranks project-linked memories via the retriever. Deletion remains
// unimplemented (M10) and returns ErrWriteUnsupported so the admin handler
// answers honestly instead of fabricating a success.
package knowledge
import (
"context"
"crypto/rand"
"encoding/hex"
"errors"
"fmt"
"io"
"log/slog"
"strings"
"unicode/utf8"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/projectindex"
)
// ErrWriteUnsupported is returned by the write operations. Upload/chunk/embed and
// delete are milestone M10; the admin handler maps it to a clear 501 so the
// frontend can explain rather than show a placeholder alert.
var ErrWriteUnsupported = errors.New("knowledge: write operations (upload/chunk/embed/delete) land in M10")
// ErrWriteUnsupported is returned by DeleteDocument. Deletion is milestone M10;
// the admin handler maps it to a clear 501 so the frontend can explain rather
// than show a placeholder alert.
var ErrWriteUnsupported = errors.New("knowledge: delete is not implemented in this milestone")
// defaultTopK bounds the semantic retrieval path.
const defaultTopK = 8
@@ -54,8 +62,12 @@ type Doc struct {
Chunks int
// Embedded reports whether a persistent vector index is wired for this
// deployment. This is the deployment-level semantic capability; per-document
// embedding state is not tracked yet (M10).
// embedding state is not tracked yet.
Embedded bool
// Content is the document body supplied to IndexDocument. It is empty for
// documents returned by ListDocuments (they carry metadata only) and is
// chunked into chapters by IndexDocument.
Content string
}
// Hit is one retrieval result surfaced to the admin panel.
@@ -253,10 +265,61 @@ func describeProject(p *projectindex.Project) string {
return p.Title
}
// IndexDocument is not implemented in this milestone. It returns
// ErrWriteUnsupported so the handler answers honestly (upload/chunk/embed = M10).
func (s *Store) IndexDocument(_ context.Context, _ Doc) error {
return ErrWriteUnsupported
// IndexDocument persists a document into the project index: it creates a project
// entry (title + description), chunks the content into chapters and adds each
// chapter, so the document becomes immediately text/keyword-searchable via
// projectindex.Search. A blank name or content returns an error. When the
// project-index store is not wired there is nowhere to persist, so it returns an
// error rather than a fabricated success.
//
// Semantic embedding is a graceful no-op: this data source does not own a vector
// index, so a document imported here is keyword-searchable, and the existing
// semantic path (project-linked memories ranked by the retriever) is untouched.
// Per the red line that a configured vector enhancement must be genuinely used
// and never a shell, we do not pretend to embed; the "Embedded" flag still
// reflects the deployment-level vector capability reported by ListDocuments.
func (s *Store) IndexDocument(ctx context.Context, doc Doc) error {
name := strings.TrimSpace(doc.Name)
if name == "" {
return errors.New("knowledge: index document: name is required")
}
content := strings.TrimSpace(doc.Content)
if content == "" {
return errors.New("knowledge: index document: content is required")
}
if s.projects == nil {
return errors.New("knowledge: index document: project index store is not configured")
}
projectID := strings.TrimSpace(doc.ID)
if projectID == "" {
id, err := newProjectID()
if err != nil {
return fmt.Errorf("knowledge: index document: %w", err)
}
projectID = id
}
if err := s.projects.PutProject(ctx, projectindex.Project{
ID: projectID,
Title: name,
Desc: strings.TrimSpace(doc.Desc),
}); err != nil {
return fmt.Errorf("knowledge: index project: %w", err)
}
chunks := chunkDocument(content)
for i, ch := range chunks {
if err := s.projects.AddChapter(ctx, projectID, projectindex.Chapter{
ProjectID: projectID,
Title: fmt.Sprintf("第 %d 节", i+1),
Content: ch,
}); err != nil {
return fmt.Errorf("knowledge: index chapter %d: %w", i+1, err)
}
}
s.log.Info("indexed document", "project", projectID, "name", name, "chapters", len(chunks))
return nil
}
// DeleteDocument is not implemented in this milestone. It returns
@@ -264,3 +327,53 @@ func (s *Store) IndexDocument(_ context.Context, _ Doc) error {
func (s *Store) DeleteDocument(_ context.Context, _ string) error {
return ErrWriteUnsupported
}
// maxChapterRunes bounds a single chapter's content length. Paragraphs longer
// than this are split into fixed-run pieces so no chapter is overwhelmingly large.
const maxChapterRunes = 500
// chunkDocument splits content into chapter-sized strings. Paragraphs are
// split on blank lines (a run of two or more newlines); a paragraph longer than
// maxChapterRunes is cut into fixed-run pieces at a rune boundary. Empty
// paragraphs are dropped and a non-nil (possibly empty) slice is returned.
func chunkDocument(content string) []string {
paragraphs := strings.Split(content, "\n\n")
chunks := make([]string, 0, len(paragraphs))
for _, p := range paragraphs {
p = strings.TrimSpace(p)
if p == "" {
continue
}
if utf8.RuneCountInString(p) <= maxChapterRunes {
chunks = append(chunks, p)
continue
}
chunks = append(chunks, splitRunes(p, maxChapterRunes)...)
}
return chunks
}
// splitRunes breaks s into pieces of at most n runes each.
func splitRunes(s string, n int) []string {
rs := []rune(s)
out := make([]string, 0, (len(rs)+n-1)/n)
for len(rs) > 0 {
if len(rs) <= n {
out = append(out, string(rs))
break
}
out = append(out, string(rs[:n]))
rs = rs[n:]
}
return out
}
// newProjectID returns a 128-bit cryptographically random id, hex-encoded. It
// avoids a UUID dependency so the package stays self-contained.
func newProjectID() (string, error) {
b := make([]byte, 16)
if _, err := rand.Read(b); err != nil {
return "", fmt.Errorf("knowledge: generate id: %w", err)
}
return hex.EncodeToString(b), nil
}
+108
View File
@@ -0,0 +1,108 @@
package knowledge
import (
"context"
"errors"
"io"
"log/slog"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/projectindex"
)
func testLogger() *slog.Logger {
return slog.New(slog.NewTextHandler(io.Discard, nil))
}
// newTestStore builds a Store backed by an isolated in-memory SQLite project index.
func newTestStore(t *testing.T) *Store {
t.Helper()
ctx := context.Background()
d, err := db.Open(ctx, ":memory:", testLogger())
if err != nil {
t.Fatalf("open in-memory db: %v", err)
}
t.Cleanup(func() { _ = d.Close() })
ps, err := projectindex.New(ctx, d, testLogger())
if err != nil {
t.Fatalf("projectindex.New: %v", err)
}
return New(Deps{Projects: ps, Log: testLogger()})
}
func TestIndexDocumentPersistsProjectAndChapters(t *testing.T) {
t.Parallel()
ctx := context.Background()
s := newTestStore(t)
doc := Doc{Name: "Cyrene 设计", Desc: "v2 设计文档", Content: "这是项目描述段落。\n\n第二段:内部实现要点。\n第三行补充说明。"}
if err := s.IndexDocument(ctx, doc); err != nil {
t.Fatalf("IndexDocument: %v", err)
}
docs, err := s.ListDocuments(ctx)
if err != nil {
t.Fatalf("ListDocuments: %v", err)
}
if len(docs) != 1 {
t.Fatalf("documents = %d, want 1", len(docs))
}
if docs[0].Name != "Cyrene 设计" || docs[0].Desc != "v2 设计文档" {
t.Fatalf("doc = %+v, want name/desc set", docs[0])
}
if docs[0].Chunks != 2 {
t.Fatalf("chunks = %d, want 2 (two paragraphs)", docs[0].Chunks)
}
// The imported content is keyword-searchable via the project index.
hits, err := s.Search(ctx, "内部实现")
if err != nil {
t.Fatalf("Search: %v", err)
}
if len(hits) == 0 {
t.Fatalf("search returned no hits for imported content")
}
found := false
for _, h := range hits {
if h.DocName == "Cyrene 设计" && h.Chapter == "第 2 节" {
found = true
break
}
}
if !found {
t.Fatalf("hits = %+v, want chapter 第 2 节 under Cyrene 设计", hits)
}
}
func TestIndexDocumentRejectsBlankNameOrContent(t *testing.T) {
t.Parallel()
ctx := context.Background()
s := newTestStore(t)
if err := s.IndexDocument(ctx, Doc{Content: "有内容没名字"}); err == nil {
t.Fatal("IndexDocument with blank name should error")
}
if err := s.IndexDocument(ctx, Doc{Name: "有名字没内容"}); err == nil {
t.Fatal("IndexDocument with blank content should error")
}
}
func TestIndexDocumentRequiresProjectStore(t *testing.T) {
t.Parallel()
ctx := context.Background()
s := New(Deps{Log: testLogger()}) // projects == nil
err := s.IndexDocument(ctx, Doc{Name: "x", Content: "y"})
if err == nil {
t.Fatal("IndexDocument without a project index should error")
}
}
func TestDeleteDocumentStillUnsupported(t *testing.T) {
t.Parallel()
s := New(Deps{Log: testLogger()})
if err := s.DeleteDocument(context.Background(), "any"); !errors.Is(err, ErrWriteUnsupported) {
t.Fatalf("DeleteDocument err = %v, want ErrWriteUnsupported", err)
}
}
+7
View File
@@ -55,6 +55,13 @@ type Message struct {
// Content carries one or more parts. A text-only message uses a single
// ContentTypeText part.
Content []Content
// SenderName is the human-readable display name of the message's sender
// (design/16 §4.6). It is transport metadata used by the context-assembly
// layer to label third-party messages for the model (e.g. "[第三方 · 昵称]"),
// never a substitute for the strongly typed sender identity carried on the
// channel layer. Empty means the sender is unlabeled (the owner, or an
// unknown/unresolved sender), and callers may leave it unset.
SenderName string
// ToolCallID and ToolCallName are reserved as interface-level placeholders
// for the M2 tool loop; they are ignored by current adapters.
ToolCallID string
+210 -28
View File
@@ -86,6 +86,11 @@ type Options struct {
Owner *user.OwnerProfile
History history.HistoryStore
OnPersisted func(history.Message)
// OnRejected, when non-nil, is called after a message is dropped by the
// allow/deny black-or-white lists (design/16 §4.6 §6), so the composition root
// can route it into an audit trail. It is invoked with the rejected message and
// a short reason token ("group_denied", "user_not_allowed", ...). Nil = log only.
OnRejected func(adapter.UnifiedMessage, string)
}
// Bridge owns the channel adapter lifecycle and the inbound message pipeline. It
@@ -97,6 +102,7 @@ type Bridge struct {
owner *user.OwnerProfile
history history.HistoryStore
onPersisted func(history.Message)
onRejected func(adapter.UnifiedMessage, string)
mu sync.RWMutex
adapters map[string]adapter.PlatformAdapter
@@ -130,6 +136,7 @@ func New(opts Options) (*Bridge, error) {
owner: opts.Owner,
history: opts.History,
onPersisted: opts.OnPersisted,
onRejected: opts.OnRejected,
adapters: make(map[string]adapter.PlatformAdapter),
connected: make(map[string]bool),
cancel: func() {},
@@ -205,7 +212,12 @@ func (b *Bridge) Start() {
b.connected[a.Name()] = true
b.mu.Unlock()
a.SetMessageHandler(b.onMessage)
// Close over the adapter so the inbound callback knows which account
// received the message — required for multi-account reply routing and for
// reading that account's allow/deny lists (design/16 §4.6 §5).
a.SetMessageHandler(func(ctx context.Context, msg adapter.UnifiedMessage) {
b.onMessage(ctx, msg, a)
})
b.wg.Add(1)
go b.readLoop(a)
b.log.Info("channel connected", "channel", a.Name())
@@ -274,7 +286,7 @@ func (b *Bridge) snapshot() []adapter.PlatformAdapter {
// when no bot token is present.
func (b *Bridge) configured(a adapter.PlatformAdapter) bool {
cfg := a.Config()
switch a.Name() {
switch basePlatform(a.Name()) {
case "telegram", "weixin-ilink":
return cfgStr(cfg, "bot_token") != ""
case "qq-obv11":
@@ -290,6 +302,16 @@ func (b *Bridge) configured(a adapter.PlatformAdapter) bool {
}
}
// basePlatform returns the bare platform name of a (possibly multi-account)
// adapter name like "telegram:conf0" → "telegram", so the credential gate and the
// reply-routing fallback treat every account of a platform identically.
func basePlatform(name string) string {
if base, _, ok := strings.Cut(name, ":"); ok {
return base
}
return name
}
// readLoop runs an adapter's blocking read path until the bridge stops. The emit
// sink is wired to onMessage so poll-style adapters (Telegram getUpdates, weixin
// long poll) deliver through the same pipeline as push-style adapters.
@@ -297,13 +319,16 @@ func (b *Bridge) readLoop(a adapter.PlatformAdapter) {
defer b.wg.Done()
ctx := b.ctx
a.ReadMessages(ctx, func(msg adapter.UnifiedMessage) {
b.onMessage(ctx, msg)
b.onMessage(ctx, msg, a)
})
}
// onMessage is the adapter-facing inbound callback. It runs the pipeline in a
// bounded goroutine so the adapter's read loop never blocks on a slow agent turn.
func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage) {
// The adapter is carried explicitly so the pipeline knows which account received
// the message (multi-account reply routing + per-account allow/deny, design/16
// §4.6 §5).
func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) {
b.mu.Lock()
if b.closed {
b.mu.Unlock()
@@ -337,15 +362,24 @@ func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage) {
go func() {
defer b.wg.Done()
defer func() { <-b.sem }()
b.process(ctx, msg)
b.processFrom(ctx, msg, a)
}()
}
// process runs the full inbound pipeline: resolve the session/actor, drive the
// agent loop with the chat text, and send the loop's reply back through the
// channel. Identity is resolved structurally by the session layer; the bridge
// never reads message content to decide who the sender is.
// process runs the inbound pipeline using the adapter whose platform matches the
// message (single-account convenience path; the adapter may be a bare
// platform-named registration).
func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) {
b.processFrom(ctx, msg, b.adapter(msg.Platform))
}
// processFrom runs the full inbound pipeline against the named account adapter:
// resolve the session/actor, apply the allow/deny black-or-white lists, drive the
// agent loop with the chat text, and send the loop's reply back through the same
// adapter that received the message. Identity is resolved structurally by the
// session layer; the bridge never reads message content to decide who the sender
// is (a project red line).
func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) {
text := strings.TrimSpace(msg.Content)
if text == "" {
b.log.Debug("inbound message has no chat text, skipped",
@@ -353,6 +387,22 @@ func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) {
return
}
// design/16 §4.6 §6: the black-or-white lists are enforced at the very head of
// the pipeline, before a session is resolved or the loop is driven, so a
// blacklisted/non-whitelisted group or user message never reaches the LLM. The
// owner is always allowed (structural, via the resolved actor).
actor := session.ResolveActor(msg, b.owner)
if actor.IsOwner() {
// fall through: owner messages are never blacklisted
} else if denied, reason := b.blacklisted(msg, a); denied {
b.log.Warn("inbound message blacklisted, dropped",
"platform", msg.Platform, "channel", msg.ChannelID, "reason", reason)
if b.onRejected != nil {
b.onRejected(msg, reason)
}
return
}
sess, err := b.sessions.GetOrCreate(ctx, msg, b.owner)
if err != nil {
b.log.Warn("session resolve failed", "platform", msg.Platform,
@@ -368,7 +418,7 @@ func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) {
// design/16 §4.4 + design/02 §3: an owner private chat is routed INTO the single
// main web session (shared across platforms) and persisted to chat_messages with
// the channel platform, so the Web client mirrors it; group chats and strangers
// keep their isolated sess-xxx context and are NOT persisted to the main session.
// keep their isolated sess-xxx context and their own platform log.
sessID := sess.SessionID
isOwnerPrivate := !sess.IsGroup && sess.Owner
platform := ""
@@ -377,19 +427,26 @@ func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) {
platform = msg.Platform
}
// Persist the inbound user message into the main web session before the turn, so
// the Web client sees it even if the channel reply fails (design/16 §4.4).
// The third-party boundary only applies in a restricted (group/stranger)
// context: memory is read-only (MemWritable=false), and the sender name is
// carried so the agent can see who is speaking (design/16 §4.6 §1-2). The
// resolved actor decides who is the owner vs a third party structurally — never
// from prompt text.
runEnv := loop.SessionEnv{ActorEnv: sess.Env(), MemWritable: sess.MemWritable}
senderName := senderDisplayName(msg, actor)
loopMsgs := b.buildLoopMessages(ctx, text, runEnv, actor, senderName)
// Persist the inbound user message before the turn so it is durable even if the
// channel reply fails. Owner-private messages land in the main web session and
// are broadcast to the Web client; restricted messages land in their isolated
// session (the adapter log page) and are NOT broadcast to the main session.
if isOwnerPrivate {
b.persistToMain(ctx, history.RoleUser, text, platform, deviceIDFor(msg))
} else {
b.persistIsolated(ctx, history.RoleUser, text, deviceIDFor(msg), sess, msg.Platform, senderName)
}
// Thread the resolved isolation boundary into the loop's memory gates: the
// owner's private conversation reads+writes memory, group chats and strangers
// are read-only isolation (design/02 §3, design/11 §2).
runEnv := loop.SessionEnv{ActorEnv: sess.Env(), MemWritable: sess.MemWritable}
res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), sessID, []model.Message{
model.TextMessage(model.RoleUser, text),
})
res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), sessID, loopMsgs)
if err != nil {
b.log.Warn("agent run failed", "platform", msg.Platform,
"session", sessID, "error", err)
@@ -401,13 +458,14 @@ func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) {
b.log.Debug("no reply produced", "platform", msg.Platform, "session", sessID)
return
}
// Persist the assistant reply into the main web session (design/16 §4.4) and
// broadcast, then deliver it back to the channel — so the Web client mirrors the
// IM reply too.
// Persist the assistant reply (owner-private → main session + broadcast;
// restricted → isolated log) then deliver it back to the channel.
if isOwnerPrivate {
b.persistToMain(ctx, history.RoleAssistant, reply, platform, deviceIDFor(msg))
} else {
b.persistIsolated(ctx, history.RoleAssistant, reply, "", sess, msg.Platform, "")
}
if err := b.sendReply(ctx, msg, reply); err != nil {
if err := b.sendReply(ctx, msg, a, reply); err != nil {
if errors.Is(err, adapter.ErrUnsupported) {
b.log.Debug("channel does not support outbound reply, skipped",
"platform", msg.Platform, "session", sessID)
@@ -417,11 +475,35 @@ func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) {
}
}
// sendReply delivers the loop's text back to the channel's source conversation.
// It mirrors the inbound message's channel targeting fields so the adapter
// resolves the same peer the message came from.
func (b *Bridge) sendReply(ctx context.Context, in adapter.UnifiedMessage, text string) error {
a := b.adapter(in.Platform)
// buildLoopMessages builds the model-visible message list for one inbound turn. In
// a restricted (group/stranger) context it prepends a single system message that
// states the third-party identity boundary — "every IM non-admin account's message
// is a third party, unrelated to the owner" (design/16 §4.6 §1). The boundary is
// decided structurally from the resolved SessionEnv, never inferred by the LLM.
// The raw message text is left untouched (a copy is built with the prefix), so
// persisted content stays consistent with what the source platform delivered.
func (b *Bridge) buildLoopMessages(_ context.Context, text string, env loop.SessionEnv, actor adapter.Actor, senderName string) []model.Message {
msg := model.TextMessage(model.RoleUser, text)
thirdParty := env.Restricted() && !actor.IsOwner() && senderName != ""
if thirdParty {
msg.SenderName = senderName
msg.Content[0].Text = fmt.Sprintf("[第三方 · %s] %s", senderName, text)
}
if !env.Restricted() {
return []model.Message{msg}
}
boundary := model.TextMessage(model.RoleSystem, thirdPartyBoundaryPrompt)
return []model.Message{boundary, msg}
}
// sendReply delivers the loop's text back to the channel's source conversation
// through the account adapter that received the message (multi-account safe,
// design/16 §4.6 §5). It mirrors the inbound message's channel targeting fields so
// the adapter resolves the same peer the message came from.
func (b *Bridge) sendReply(ctx context.Context, in adapter.UnifiedMessage, a adapter.PlatformAdapter, text string) error {
if a == nil {
a = b.adapter(in.Platform)
}
if a == nil {
return fmt.Errorf("channelbridge: no adapter for platform %q", in.Platform)
}
@@ -475,3 +557,103 @@ func deviceIDFor(msg adapter.UnifiedMessage) string {
}
return msg.Platform + ":" + msg.ChannelID
}
// thirdPartyBoundaryPrompt is the system message prepended to a restricted
// (group-chat / stranger) context so the model knows the identity boundary. It is
// only injected when the session is structurally restricted (design/16 §4.6 §1);
// the LLM never decides who is a third party — that was already resolved from the
// actor metadata before this message exists.
const thirdPartyBoundaryPrompt = "你正在 IM 平台的群聊/第三方会话中。每个 IM 的「非管理员账户」发送的消息 = 第三方人员,和主人没有半毛钱关系,不要把它们当成主人的话、也不要按主人对待。只有未标注昵称的消息才是管理员(主人)发言。"
// senderDisplayName returns the human-readable sender name used to label a
// third-party message. It prefers the resolved actor's display name (which the
// channel layer derived from channel metadata + the raw SenderName), falling back
// to the message's own SenderName, so an empty SenderName always means "the
// owner/unlabeled" rather than a nickname that resolved to nothing.
func senderDisplayName(msg adapter.UnifiedMessage, actor adapter.Actor) string {
if actor.DisplayName != "" {
return actor.DisplayName
}
return msg.SenderName
}
// blacklisted applies the per-account allow/deny black-or-white lists (design/16
// §4.6 §6) to a non-owner inbound message. Group messages are matched by GroupID
// against allowGroups/denyGroups; direct messages are matched by the sender's user
// id against allowUsers/denyUsers. An empty allow list means "allow all" (only the
// deny list applies); a non-empty allow list becomes a white list. The owner is
// never blacklisted (the caller skips this for owner messages).
func (b *Bridge) blacklisted(msg adapter.UnifiedMessage, a adapter.PlatformAdapter) (bool, string) {
if a == nil {
return false, ""
}
allowGroups, denyGroups, allowUsers, denyUsers := channelLists(a.Config())
if msg.IsGroup() {
if sliceContains(denyGroups, msg.GroupID) {
return true, "group_denied"
}
if len(allowGroups) > 0 && !sliceContains(allowGroups, msg.GroupID) {
return true, "group_not_allowed"
}
return false, ""
}
uid := firstNonEmptyStr(msg.SenderID, msg.Sender.UserID)
if sliceContains(denyUsers, uid) {
return true, "user_denied"
}
if len(allowUsers) > 0 && !sliceContains(allowUsers, uid) {
return true, "user_not_allowed"
}
return false, ""
}
// channelLists reads the allow/deny string lists off an adapter's Config snapshot.
// The lists are stored as plain []string values (set by the multi-account wiring);
// an absent key yields an empty list.
func channelLists(cfg map[string]any) (allowGroups, denyGroups, allowUsers, denyUsers []string) {
if cfg == nil {
return nil, nil, nil, nil
}
return cfgStrings(cfg, "allowGroups"), cfgStrings(cfg, "denyGroups"),
cfgStrings(cfg, "allowUsers"), cfgStrings(cfg, "denyUsers")
}
// cfgStrings reads a []string field from an adapter config map, returning nil when
// absent or of a different type.
func cfgStrings(m map[string]any, key string) []string {
if m == nil {
return nil
}
switch v := m[key].(type) {
case []string:
return v
case []any:
out := make([]string, 0, len(v))
for _, e := range v {
if s, ok := e.(string); ok {
out = append(out, s)
}
}
return out
default:
return nil
}
}
// sliceContains reports whether s is present in list.
func sliceContains(list []string, s string) bool {
for _, v := range list {
if v == s {
return true
}
}
return false
}
// firstNonEmptyStr returns a when non-empty, otherwise b.
func firstNonEmptyStr(a, b string) string {
if a != "" {
return a
}
return b
}
@@ -2,6 +2,7 @@ package channelbridge
import (
"context"
"strings"
"sync"
"testing"
@@ -18,6 +19,7 @@ import (
type fakeAdapter struct {
name string
connected bool
cfg map[string]any
sends []adapter.UnifiedMessage
handler adapter.MessageHandler
sendErr error
@@ -25,6 +27,9 @@ type fakeAdapter struct {
func (f *fakeAdapter) Name() string { return f.name }
func (f *fakeAdapter) Config() map[string]any {
if f.cfg != nil {
return f.cfg
}
return map[string]any{"bot_token": "configured"}
}
func (f *fakeAdapter) Connect(context.Context) error { f.connected = true; return nil }
@@ -236,6 +241,9 @@ func (c *captureHistory) List(context.Context, string, int64, int, string) ([]hi
func (c *captureHistory) Recent(context.Context, string, int) ([]history.Message, error) {
return nil, nil
}
func (c *captureHistory) ListPlatformLog(context.Context, string, int64, int) ([]history.Message, error) {
return nil, nil
}
func (c *captureHistory) DeleteSession(context.Context, string) error { return nil }
func (c *captureHistory) NewSubSession(context.Context, string) (string, error) {
return "", nil
@@ -350,16 +358,177 @@ func TestGroupMessageNotPersistedToMainSession(t *testing.T) {
}
b.process(context.Background(), msg)
// Group chat stays isolated: nothing is written to the main web session and the
// loop runs against the isolated session id.
if recs := hist.records(); len(recs) != 0 {
t.Fatalf("group appends = %+v, want none persisted", recs)
// design/16 §4.6: a group/stranger conversation stays isolated — it is NOT
// persisted into the main web session, and the loop runs against the isolated
// session id. Messages land only in the isolated session's own platform log
// (the adapter log page), so nothing reaches the Web main session.
recs := hist.records()
if len(recs) != 2 {
t.Fatalf("group appends = %d, want 2 (isolated user + assistant)", len(recs))
}
for i, rec := range recs {
if rec.SessionID != "sess-g" {
t.Fatalf("record[%d] SessionID = %q, want isolated sess-g (not web)", i, rec.SessionID)
}
if rec.Platform != "telegram" {
t.Fatalf("record[%d] Platform = %q, want telegram", i, rec.Platform)
}
}
if chat.sessionID != "sess-g" {
t.Fatalf("group loop session id = %q, want sess-g", chat.sessionID)
}
}
func TestBlacklistGroupMessageDroppedBeforeLoop(t *testing.T) {
t.Parallel()
chat := &stubChatter{res: &loop.Result{Text: "不该回复"}}
// A stranger's group session that would otherwise be processed.
sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: false, MemWritable: false}}
ad := &fakeAdapter{name: "telegram", cfg: map[string]any{
"bot_token": "configured",
"denyGroups": []string{"-100"},
}}
var rejected []string
b, err := New(Options{
Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: &captureHistory{},
OnRejected: func(_ adapter.UnifiedMessage, reason string) { rejected = append(rejected, reason) },
})
if err != nil {
t.Fatalf("New: %v", err)
}
b.Add(ad)
msg := adapter.UnifiedMessage{
Platform: "telegram", ChannelID: "-100", ChannelType: adapter.ChannelGroup,
GroupID: "-100", SenderID: "9", SenderName: "张三",
Sender: adapter.Actor{Role: adapter.RoleMember}, Content: "spam",
ContentType: adapter.ContentText,
}
b.process(context.Background(), msg)
// The blacklisted group message never reaches the agent loop and never sends a
// reply; the rejection reason is surfaced for the audit trail.
if chat.sessionID != "" || len(chat.messages) != 0 {
t.Fatalf("loop ran for blacklisted message: session=%q messages=%+v", chat.sessionID, chat.messages)
}
if len(ad.sends) != 0 {
t.Fatalf("blacklisted message sent %d replies, want 0", len(ad.sends))
}
if len(rejected) != 1 || rejected[0] != "group_denied" {
t.Fatalf("rejection reasons = %v, want [group_denied]", rejected)
}
}
func TestBlacklistUserMessageDroppedBeforeLoop(t *testing.T) {
t.Parallel()
chat := &stubChatter{res: &loop.Result{Text: "不该回复"}}
sess := &stubSessions{sess: session.Session{SessionID: "sess-s", IsGroup: false, Owner: false, MemWritable: false}}
ad := &fakeAdapter{name: "telegram", cfg: map[string]any{
"bot_token": "configured",
"allowUsers": []string{"111"},
}}
b, err := New(Options{Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: &captureHistory{}})
if err != nil {
t.Fatalf("New: %v", err)
}
b.Add(ad)
msg := adapter.UnifiedMessage{
Platform: "telegram", ChannelID: "200", ChannelType: adapter.ChannelDirect,
SenderID: "200", Sender: adapter.Actor{Role: adapter.RoleMember}, Content: "hi",
ContentType: adapter.ContentText,
}
b.process(context.Background(), msg)
// A direct message from a user NOT in the allow whitelist is dropped; the owner
// is never affected (this is a stranger).
if chat.sessionID != "" || len(chat.messages) != 0 {
t.Fatalf("loop ran for non-whitelisted user: session=%q messages=%+v", chat.sessionID, chat.messages)
}
if len(ad.sends) != 0 {
t.Fatalf("non-whitelisted user sent %d replies, want 0", len(ad.sends))
}
}
func TestGroupMessageThirdPartySenderNameAndBoundary(t *testing.T) {
t.Parallel()
chat := &stubChatter{res: &loop.Result{Text: "收到"}}
sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: false, MemWritable: false}}
hist := &captureHistory{}
b, err := New(Options{Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist})
if err != nil {
t.Fatalf("New: %v", err)
}
msg := adapter.UnifiedMessage{
Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup,
GroupID: "-20", GroupName: "Gang", SenderID: "9", SenderName: "张三",
Sender: adapter.Actor{Role: adapter.RoleMember}, Content: "hello all",
ContentType: adapter.ContentText,
}
b.process(context.Background(), msg)
// The loop sees a leading third-party boundary system message plus the user
// message already prefixed with [第三方 · 张三] (the raw content is preserved in
// history, so no source data is lost).
if len(chat.messages) != 2 {
t.Fatalf("loop messages = %d, want 2 (boundary + user)", len(chat.messages))
}
if chat.messages[0].Role != model.RoleSystem || !strings.Contains(chat.messages[0].Content[0].Text, "非管理员账户") {
t.Fatalf("expected third-party boundary system message, got %+v", chat.messages[0])
}
user := chat.messages[1]
if user.SenderName != "张三" || user.Content[0].Text != "[第三方 · 张三] hello all" {
t.Fatalf("loop user message = %+v, want SenderName 张三 and [第三方 · 张三] prefix", user)
}
// History keeps the RAW content and carries the sender name so the adapter log
// page can label the stream without altering the source message.
recs := hist.records()
if len(recs) != 2 || recs[0].Content != "hello all" || recs[0].SenderName != "张三" {
t.Fatalf("isolated records = %+v, want raw content with SenderName 张三", recs)
}
}
func TestOwnerInGroupIsIsolatedButNotLabeledThirdParty(t *testing.T) {
t.Parallel()
chat := &stubChatter{res: &loop.Result{Text: "收到"}}
// Owner speaking inside a group: the session stays restricted (mem read-only),
// but the owner is never labeled as a third party (design/16 §4.6 §1-2).
sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: true, MemWritable: false}}
hist := &captureHistory{}
b, err := New(Options{Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist})
if err != nil {
t.Fatalf("New: %v", err)
}
msg := adapter.UnifiedMessage{
Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup,
GroupID: "-20", SenderID: "owner_uid", SenderName: "主人",
Sender: adapter.Actor{Role: adapter.RoleOwner}, Content: "我在群里",
ContentType: adapter.ContentText,
}
b.process(context.Background(), msg)
// Restricted context → a third-party boundary system message is still injected.
if len(chat.messages) != 2 || chat.messages[0].Role != model.RoleSystem {
t.Fatalf("loop messages = %d, want 2 (boundary + user)", len(chat.messages))
}
// Owner's message is NOT prefixed as third-party and carries no SenderName.
user := chat.messages[1]
if user.SenderName != "" || user.Content[0].Text != "我在群里" {
t.Fatalf("owner group message = %+v, want no SenderName and no [第三方] prefix", user)
}
// Owner group speech still lands in the isolated session (not the main web session).
recs := hist.records()
if len(recs) != 2 || recs[0].SessionID != "sess-g" {
t.Fatalf("owner group records = %+v, want isolated sess-g, not web", recs)
}
}
func TestNoHistoryStoreKeepsBridgeUnchanged(t *testing.T) {
t.Parallel()
// A bridge built without a history store persists nothing but still runs/replies.
@@ -3,6 +3,8 @@ package channelbridge
import (
"context"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
)
@@ -23,6 +25,16 @@ func (b *Bridge) SetOnPersisted(fn func(history.Message)) {
b.onPersisted = fn
}
// SetOnRejected wires (or replaces) the hook called after a message is dropped by
// the allow/deny black-or-white lists (design/16 §4.6 §6). The composition root
// routes it into the gateway audit trail. It is a no-op-safe call and is safe for
// concurrent use with in-flight processing.
func (b *Bridge) SetOnRejected(fn func(adapter.UnifiedMessage, string)) {
b.mu.Lock()
defer b.mu.Unlock()
b.onRejected = fn
}
// notifyPersisted hands a just-persisted message to the transport hook when one is
// wired, so the gateway can broadcast `message_persisted` for cross-end sync.
func (b *Bridge) notifyPersisted(msg history.Message) {
@@ -53,3 +65,27 @@ func (b *Bridge) persistToMain(ctx context.Context, role, content, platform, dev
}
b.notifyPersisted(full)
}
// persistIsolated writes one message into a restricted (group / stranger) session's
// own platform log (design/16 §4.6 §4): the message lands in the isolated sess-xxx
// session (NOT the main web session) and carries the channel platform, the client
// deviceID and the sender's display name, so the IM adapter log page can render the
// non-owner message stream. It is deliberately NOT broadcast to the Web main session
// (only owner-private messages mirror to the Web). It is a no-op when no history
// store is wired, so the bridge runs unchanged without storage.
func (b *Bridge) persistIsolated(ctx context.Context, role, content, deviceID string, sess session.Session, platform, senderName string) {
if b.history == nil {
return
}
full, err := b.history.AppendFull(ctx, history.Message{
SessionID: sess.SessionID, Role: role, Content: content, Platform: platform,
DeviceID: deviceID, SenderName: senderName,
})
if err != nil {
b.log.Warn("persist isolated session", "err", err, "role", role,
"session", sess.SessionID, "platform", platform, "sender_name", senderName)
return
}
// No broadcast: third-party messages must not reach the Web main session (design/16 §4.6 §3).
_ = full
}
+9 -1
View File
@@ -96,8 +96,16 @@ func (s *Server) adminListPersonas(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.collectPersonas())
}
// adminListSessions lists channel sessions (admin read).
// adminListSessions lists the persistent chat-history sessions (main + sub) when a
// history store is wired, otherwise falls back to the channel-session view. It is
// the admin counterpart of handleListSessions: the panel shows the real main and
// secondary sessions (design/16 §2) rather than stale channel-session rows, and
// the fallback keeps the pre-design/16 contract for servers built without history.
func (s *Server) adminListSessions(w http.ResponseWriter, r *http.Request) {
if s.history != nil {
s.handleHistorySessions(w, r)
return
}
writeJSON(w, http.StatusOK, s.collectSessions(r.Context()))
}
@@ -0,0 +1,127 @@
package gateway
import (
"encoding/json"
"net/http"
"strings"
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
)
// Admin memory + knowledge-base write endpoints (owner-only, audited). The memory
// add writes a real item into the memory store (owner-private, safe), and the RAG
// index imports a document into the project index so it becomes searchable. Both
// degrade to a clear 501 when their backing store is not wired.
// memoryAddRequest is the POST /api/v1/admin/memories body.
type memoryAddRequest struct {
PersonaID string `json:"personaId"`
Content string `json:"content"`
Scope string `json:"scope"`
}
// adminAddMemory writes a single memory item for the owner. It is the manual
// "添加记忆" path: the item is scoped to the persona (default), so it behaves like
// the "记住 X" path the loop already writes. Scope may be omitted (default
// persona) or one of user|persona|project; an invalid non-empty scope is rejected.
func (s *Server) adminAddMemory(w http.ResponseWriter, r *http.Request) {
var req memoryAddRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body"))
return
}
personaID := strings.TrimSpace(req.PersonaID)
if personaID == "" {
personaID = "cyrene"
}
content := strings.TrimSpace(req.Content)
if content == "" {
s.auditClaims(r, "memory.add", "error")
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "content is required"))
return
}
rawScope := strings.TrimSpace(req.Scope)
if rawScope != "" && !isValidMemoryScope(rawScope) {
s.auditClaims(r, "memory.add", "error")
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "scope must be one of user|persona|project"))
return
}
store, ok := s.memories.(*memory.Store)
if !ok {
s.auditClaims(r, "memory.add", "error")
writeError(w, &apperr.AppError{Code: "not_implemented", Message: "记忆存储未配置(未注入 memory.Store)", HTTPCode: http.StatusNotImplemented})
return
}
stored, err := store.Add(r.Context(), memory.MemoryItem{
PersonaID: personaID,
Content: content,
Scope: memory.Scope(rawScope),
})
if err != nil {
s.auditClaims(r, "memory.add", "error")
s.log.Error("memory add", "err", err)
writeError(w, apperr.Wrap(apperr.ErrInternal, "memory_add_failed", "failed to add memory"))
return
}
s.auditClaims(r, "memory.add", "ok")
writeJSON(w, http.StatusCreated, map[string]any{"id": stored.ID, "scope": string(stored.Scope), "text": stored.Content})
}
// isValidMemoryScope reports whether raw is one of the defined memory scopes.
func isValidMemoryScope(raw string) bool {
switch memory.Scope(strings.TrimSpace(raw)) {
case memory.ScopeUser, memory.ScopePersona, memory.ScopeProject:
return true
default:
return false
}
}
// ragIndexRequest is the POST /api/v1/admin/rag body.
type ragIndexRequest struct {
Name string `json:"name"`
Desc string `json:"desc"`
Content string `json:"content"`
}
// adminIndexRag imports one knowledge document: it hands the name/description and
// content to the knowledge store, which chunks it and indexes it into the project
// index so the document becomes keyword/searchable. It is the admin "导入文档" path.
func (s *Server) adminIndexRag(w http.ResponseWriter, r *http.Request) {
var req ragIndexRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body"))
return
}
name := strings.TrimSpace(req.Name)
if name == "" {
s.auditClaims(r, "rag.index", "error")
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "name is required"))
return
}
content := strings.TrimSpace(req.Content)
if content == "" {
s.auditClaims(r, "rag.index", "error")
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "content is required"))
return
}
if s.rag == nil {
s.auditClaims(r, "rag.index", "error")
writeError(w, &apperr.AppError{Code: "not_implemented", Message: "知识库 RAG 服务未配置(未注入 RagStore)", HTTPCode: http.StatusNotImplemented})
return
}
if err := s.rag.IndexDocument(r.Context(), knowledge.Doc{
Name: name,
Desc: strings.TrimSpace(req.Desc),
Content: content,
}); err != nil {
s.auditClaims(r, "rag.index", "error")
s.log.Error("rag index", "err", err)
writeError(w, apperr.Wrap(apperr.ErrInternal, "rag_index_failed", "failed to index document"))
return
}
s.auditClaims(r, "rag.index", "ok")
writeJSON(w, http.StatusCreated, map[string]any{"name": name})
}
@@ -0,0 +1,233 @@
package gateway
import (
"context"
"encoding/json"
"io"
"log/slog"
"net/http"
"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"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/projectindex"
)
func gwTestLogger() *slog.Logger {
return slog.New(slog.NewTextHandler(io.Discard, nil))
}
// newMemoryStoreForGateway opens an isolated in-memory SQLite DB and builds a
// real memory.Store over it so the owner write path (which type-asserts to
// *memory.Store) exercises the real Add.
func newMemoryStoreForGateway(t *testing.T) *memory.Store {
t.Helper()
ctx := context.Background()
d, err := db.Open(ctx, ":memory:", gwTestLogger())
if err != nil {
t.Fatalf("open in-memory db: %v", err)
}
t.Cleanup(func() { _ = d.Close() })
s, err := memory.NewStore(ctx, d, gwTestLogger())
if err != nil {
t.Fatalf("memory.NewStore: %v", err)
}
return s
}
func newRagStoreForGateway(t *testing.T) *knowledge.Store {
t.Helper()
ctx := context.Background()
d, err := db.Open(ctx, ":memory:", gwTestLogger())
if err != nil {
t.Fatalf("open in-memory db: %v", err)
}
t.Cleanup(func() { _ = d.Close() })
ps, err := projectindex.New(ctx, d, gwTestLogger())
if err != nil {
t.Fatalf("projectindex.New: %v", err)
}
return knowledge.New(knowledge.Deps{Projects: ps, Log: gwTestLogger()})
}
func newHistoryStoreForGateway(t *testing.T) *history.Store {
t.Helper()
ctx := context.Background()
d, err := db.Open(ctx, ":memory:", gwTestLogger())
if err != nil {
t.Fatalf("open in-memory db: %v", err)
}
t.Cleanup(func() { _ = d.Close() })
s, err := history.NewStore(ctx, d, gwTestLogger())
if err != nil {
t.Fatalf("history.NewStore: %v", err)
}
return s
}
func TestAdminAddMemoryWritesAndLists(t *testing.T) {
t.Parallel()
store := newMemoryStoreForGateway(t)
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Memories: store})
_ = srv
tok := signAny(t, jwt, "acct", auth.RoleOwner)
resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/memories",
map[string]any{"personaId": "cyrene", "content": "她喜欢喝冰美式", "scope": "persona"}, tok)
if resp.StatusCode != http.StatusCreated {
t.Fatalf("add status = %d, want 201 (body=%s)", resp.StatusCode, body)
}
var added struct {
ID string `json:"id"`
Scope string `json:"scope"`
Text string `json:"text"`
}
if err := json.Unmarshal(decodeBody(t, body)["data"], &added); err != nil {
t.Fatalf("decode add: %v (body=%s)", err, body)
}
if added.ID == "" || added.Scope != "persona" || added.Text != "她喜欢喝冰美式" {
t.Fatalf("added = %+v, want persona/她喜欢喝冰美式", added)
}
// The freshly added item is now listed.
listResp, listBody := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/memories", nil, tok)
if listResp.StatusCode != http.StatusOK {
t.Fatalf("list status = %d, want 200 (body=%s)", listResp.StatusCode, listBody)
}
var items []MemorySummary
if err := json.Unmarshal(decodeBody(t, listBody)["data"], &items); err != nil {
t.Fatalf("decode list: %v", err)
}
if len(items) != 1 || items[0].Text != "她喜欢喝冰美式" || items[0].Scope != "persona" {
t.Fatalf("items = %+v, want one persona 她喜欢喝冰美式", items)
}
}
func TestAdminAddMemoryRejectsBadInput(t *testing.T) {
t.Parallel()
store := newMemoryStoreForGateway(t)
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{Memories: store})
tok := signAny(t, jwt, "acct", auth.RoleOwner)
// Blank content.
resp, _ := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/memories",
map[string]any{"personaId": "cyrene", "content": " "}, tok)
if resp.StatusCode != http.StatusBadRequest {
t.Fatalf("blank content status = %d, want 400", resp.StatusCode)
}
// Invalid scope.
resp2, _ := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/memories",
map[string]any{"personaId": "cyrene", "content": "x", "scope": "bogus"}, tok)
if resp2.StatusCode != http.StatusBadRequest {
t.Fatalf("bogus scope status = %d, want 400", resp2.StatusCode)
}
}
func TestAdminAddMemoryUnwired501(t *testing.T) {
t.Parallel()
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{}) // no memory store
tok := signAny(t, jwt, "acct", auth.RoleOwner)
resp, _ := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/memories",
map[string]any{"content": "x"}, tok)
if resp.StatusCode != http.StatusNotImplemented {
t.Fatalf("unwired status = %d, want 501", resp.StatusCode)
}
}
func TestAdminIndexRagImportsAndLists(t *testing.T) {
t.Parallel()
rag := newRagStoreForGateway(t)
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: rag})
_ = srv
tok := signAny(t, jwt, "acct", auth.RoleOwner)
resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag",
map[string]any{"name": "内部知识库.md", "desc": "团队规范", "content": "第一段:命名规范。\n\n第二段:评审流程。"}, tok)
if resp.StatusCode != http.StatusCreated {
t.Fatalf("index status = %d, want 201 (body=%s)", resp.StatusCode, body)
}
// The document now shows up in the list (imported + chunked).
listResp, listBody := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/rag", nil, tok)
if listResp.StatusCode != http.StatusOK {
t.Fatalf("list status = %d, want 200 (body=%s)", listResp.StatusCode, listBody)
}
var res AdminRagResponse
if err := json.Unmarshal(decodeBody(t, listBody)["data"], &res); err != nil {
t.Fatalf("decode list: %v", err)
}
if len(res.Documents) != 1 || res.Documents[0].Name != "内部知识库.md" || res.Documents[0].Chunks != 2 {
t.Fatalf("documents = %+v, want one 内部知识库.md with 2 chunks", res.Documents)
}
// It is immediately queryable (keyword/line path).
qResp, qBody := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag/query",
map[string]any{"query": "评审流程"}, tok)
if qResp.StatusCode != http.StatusOK {
t.Fatalf("query status = %d, want 200 (body=%s)", qResp.StatusCode, qBody)
}
var qres struct {
Hits []AdminRagHit `json:"hits"`
}
if err := json.Unmarshal(decodeBody(t, qBody)["data"], &qres); err != nil {
t.Fatalf("decode query: %v", err)
}
if len(qres.Hits) == 0 || qres.Hits[0].DocName != "内部知识库.md" {
t.Fatalf("hits = %+v, want a hit on 内部知识库.md", qres.Hits)
}
}
func TestAdminIndexRagRejectsBlankNameOrContent(t *testing.T) {
t.Parallel()
rag := newRagStoreForGateway(t)
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: rag})
tok := signAny(t, jwt, "acct", auth.RoleOwner)
resp, _ := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag",
map[string]any{"name": " ", "content": "x"}, tok)
if resp.StatusCode != http.StatusBadRequest {
t.Fatalf("blank name status = %d, want 400", resp.StatusCode)
}
resp2, _ := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag",
map[string]any{"name": "x", "content": " "}, tok)
if resp2.StatusCode != http.StatusBadRequest {
t.Fatalf("blank content status = %d, want 400", resp2.StatusCode)
}
}
func TestAdminIndexRagUnwired501(t *testing.T) {
t.Parallel()
_, ts, jwt := buildEnv(t, cfgDefault(), Deps{})
tok := signAny(t, jwt, "acct", auth.RoleOwner)
resp, _ := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag",
map[string]any{"name": "x", "content": "y"}, tok)
if resp.StatusCode != http.StatusNotImplemented {
t.Fatalf("unwired status = %d, want 501", resp.StatusCode)
}
}
func TestAdminListSessionsHistoryFirst(t *testing.T) {
t.Parallel()
h := newHistoryStoreForGateway(t)
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{History: h})
_ = srv
tok := signAny(t, jwt, "acct", auth.RoleOwner)
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/sessions", nil, tok)
if resp.StatusCode != http.StatusOK {
t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
}
var items []historySessionSummary
if err := json.Unmarshal(decodeBody(t, body)["data"], &items); err != nil {
t.Fatalf("decode: %v", err)
}
if len(items) == 0 {
t.Fatal("admin sessions list is empty; main session should always exist")
}
if items[0].Type != "main" || items[0].ID == "" {
t.Fatalf("first session = %+v, want a main session with id", items[0])
}
}
+10
View File
@@ -20,6 +20,16 @@ func (s *Server) auditClaims(r *http.Request, what, result string) {
s.audit(r.Context(), who, what, result)
}
// AuditEvent records an external (non-HTTP) audit event with an explicit caller
// identity, so programmatic sinks outside the request path (for example the
// channel bridge's blacklist rejection hook, design/16 §4.6 §6) can append to the
// same who/what/when/result trail. who is a free-form actor label ("channel" or
// "bridge:telegram"); what is the operation and result its outcome. It is safe for
// concurrent use.
func (s *Server) AuditEvent(who, what, result string) {
s.audit(context.Background(), who, what, result)
}
// audit emits one structured audit record on the gateway logger and persists it to
// the configured AuditStore (falling back to an in-memory store when none is
// injected). A persist failure is logged but never blocks the admin write, so a
+55 -4
View File
@@ -34,7 +34,11 @@ type historyMessage struct {
Thinking string `json:"thinking,omitempty"`
Platform string `json:"platform,omitempty"`
DeviceID string `json:"deviceId,omitempty"`
CreatedAt int64 `json:"createdAt"`
// SenderName is the human-readable sender display name of a third-party
// message (group/stranger, design/16 §4.6). It is blank for the owner's
// private chat on the main session, so empty means "labeled as the owner".
SenderName string `json:"senderName,omitempty"`
CreatedAt int64 `json:"createdAt"`
}
// handleHistorySessions returns the main session first, then every sub session (oldest
@@ -125,6 +129,52 @@ func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, out)
}
// handleChannelPlatformLog returns up to limit messages for an IM platform across its
// isolated (group/stranger) sessions, newest first, for the admin adapter log page
// (design/16 §4.6 §4). It excludes the fixed main web session, so only the platform's
// non-owner message stream is surfaced; the owner-private mirror stays out of this view.
func (s *Server) handleChannelPlatformLog(w http.ResponseWriter, r *http.Request) {
if s.history == nil {
writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置(未注入 HistoryStore)", HTTPCode: http.StatusNotImplemented})
return
}
platform := r.PathValue("platform")
if platform == "" {
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "platform 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.ListPlatformLog(r.Context(), platform, before, limit)
if err != nil {
s.log.Error("history platform log", "err", err, "platform", platform)
writeError(w, apperr.Wrap(apperr.ErrInternal, "platform_log_failed", "failed to list platform log"))
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) {
@@ -160,9 +210,10 @@ func toHistoryMessage(m history.Message) historyMessage {
Mood: m.Mood,
Image: m.Image,
Thinking: m.Thinking,
Platform: m.Platform,
DeviceID: m.DeviceID,
CreatedAt: m.CreatedAt,
Platform: m.Platform,
DeviceID: m.DeviceID,
SenderName: m.SenderName,
CreatedAt: m.CreatedAt,
}
}
+4
View File
@@ -115,6 +115,7 @@ type HistoryStore interface {
AppendFull(ctx context.Context, msg history.Message) (history.Message, error)
List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]history.Message, error)
Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error)
ListPlatformLog(ctx context.Context, platform string, before int64, limit 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)
@@ -347,6 +348,7 @@ func (s *Server) Handler() http.Handler {
mux.Handle("GET /api/v1/admin/memories", adminAuth(http.HandlerFunc(s.adminListMemories)))
mux.Handle("GET /api/v1/admin/plugins", adminAuth(http.HandlerFunc(s.adminListPlugins)))
mux.Handle("GET /api/v1/admin/channels", adminAuth(http.HandlerFunc(s.adminListChannels)))
mux.Handle("GET /api/v1/admin/channels/{platform}/messages", adminAuth(http.HandlerFunc(s.handleChannelPlatformLog)))
mux.Handle("GET /api/v1/admin/iot", adminAuth(http.HandlerFunc(s.adminListIoT)))
mux.Handle("GET /api/v1/admin/llm", adminAuth(http.HandlerFunc(s.adminGetLlm)))
mux.Handle("GET /api/v1/admin/automations", adminAuth(http.HandlerFunc(s.adminListAutomations)))
@@ -383,7 +385,9 @@ func (s *Server) Handler() http.Handler {
mux.Handle("POST /api/v1/admin/channels/test", ownerAuth(http.HandlerFunc(s.adminChannelTest)))
mux.Handle("POST /api/v1/admin/plugins/install", ownerAuth(http.HandlerFunc(s.adminPluginInstall)))
mux.Handle("POST /api/v1/admin/plugins/uninstall", ownerAuth(http.HandlerFunc(s.adminPluginUninstall)))
mux.Handle("POST /api/v1/admin/memories", ownerAuth(http.HandlerFunc(s.adminAddMemory)))
mux.Handle("POST /api/v1/admin/memories/search", ownerAuth(http.HandlerFunc(s.adminMemorySearch)))
mux.Handle("POST /api/v1/admin/rag", ownerAuth(http.HandlerFunc(s.adminIndexRag)))
mux.Handle("GET /api/v1/admin/sessions/{id}/export", adminAuth(http.HandlerFunc(s.adminExportSession)))
// External webhook callbacks (e.g. the webhook channel) are mounted without
@@ -183,6 +183,10 @@ func (f *fakeHistory) Recent(context.Context, string, int) ([]history.Message, e
return nil, nil
}
func (f *fakeHistory) ListPlatformLog(context.Context, string, int64, int) ([]history.Message, error) {
return nil, nil
}
func (f *fakeHistory) DeleteSession(context.Context, string) error {
return nil
}