From 2e044d1d1906dff264f1a6c4605b2fff27c64a1e Mon Sep 17 00:00:00 2001 From: Aska Date: Mon, 7 Sep 2026 17:25:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(channels):=20=E7=AE=A1=E7=90=86=E9=9D=A2?= =?UTF-8?q?=E6=9D=BF=E6=B6=88=E6=81=AF=E6=97=A5=E5=BF=97=E2=80=94=E2=80=94?= =?UTF-8?q?chat=5Fmessages=20=E5=8A=A0=20adapter=5Fid=20+=20=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E7=BA=A7/=E5=B9=B3=E5=8F=B0/=E6=80=BB=E8=A7=88?= =?UTF-8?q?=E4=B8=89=E7=B2=92=E5=BA=A6=E7=AB=AF=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端： - 迁移 0007：chat_messages 加 adapter_id 列 + 索引 - history.Message 加 AdapterID；persistToMain/persistIsolated 传适配器名(a.Name()) - channelbridge: 4 个持久化点 + registerSilenced 传 a；light.go registerSilenced 签名加 a - Store 新增 ListAdapterLog(?adapter=, 不排除主会话)/ListAllLog(adapter_id!='', 总览)；ListPlatformLog 保留 - 新端点 GET /api/v1/admin/channels/messages：?adapter= 配置级 / ?platform= 平台 / 无参=总览，historyMessage 加 adapterId - 测试：ListAdapterLog/ListAllLog 过滤用例 + 端点粒度用例 + 桥接持久化断言 前端：渠道管理每配置(账号)「消息日志」按钮 + 页顶「适配器总览日志」入口 + ChannelLogPanel 弹窗(总览/配置级)，Miuix 风格，build 绿。 --- frontend/web/src/admin/modules/Channels.tsx | 100 ++++++++++++++- frontend/web/src/lib/api.ts | 13 ++ frontend/web/src/lib/types.ts | 17 +++ .../db/migrations/0007_chat_adapter.down.sql | 6 + .../db/migrations/0007_chat_adapter.up.sql | 15 +++ internal/runtime/history/store.go | 115 ++++++++++++++++-- internal/runtime/history/store_test.go | 113 +++++++++++++++++ .../services/channelbridge/channelbridge.go | 22 +++- .../channelbridge/channelbridge_test.go | 19 ++- internal/services/channelbridge/light.go | 9 +- internal/services/channelbridge/persist.go | 27 ++-- internal/services/gateway/history.go | 63 +++++++++- .../services/gateway/history_handler_test.go | 71 +++++++++++ internal/services/gateway/server.go | 3 + .../gateway/session_orchestrator_test.go | 8 ++ 15 files changed, 560 insertions(+), 41 deletions(-) create mode 100644 internal/db/migrations/0007_chat_adapter.down.sql create mode 100644 internal/db/migrations/0007_chat_adapter.up.sql diff --git a/frontend/web/src/admin/modules/Channels.tsx b/frontend/web/src/admin/modules/Channels.tsx index 89e6270..ce4168e 100644 --- a/frontend/web/src/admin/modules/Channels.tsx +++ b/frontend/web/src/admin/modules/Channels.tsx @@ -5,9 +5,11 @@ import { adminChannelOpen, adminChannelClose, adminChannelTest, + adminChannelMessages, + adminChannelLogOverview, adminDeleteChannelAccount, } from "../../lib/api"; -import type { AdminChannelRecord, ChannelTestResult } from "../../lib/types"; +import type { AdminChannelMessage, AdminChannelRecord, ChannelTestResult } from "../../lib/types"; import { useAutoReload } from "../hooks"; import { useAuditStore } from "../audit"; import { AdminTable, Notice, PageHeader, type AdminColumn } from "../components"; @@ -20,6 +22,14 @@ function platformOf(name: string): string { return name.split(":")[0]; } +/** Adapter (platform:account) name for one account config. The bridge registers a + * channels.json account as "kook:conf0" (env single-account = bare platform), so a + * config id of "default"/"" maps to the bare platform name. */ +function adapterNameOf(platform: string, accountId?: string): string { + const id = accountId?.trim(); + return id && id !== "default" ? `${platform}:${id}` : platform; +} + /** Open-config target: the platform + the prefill bundle for its form. */ interface CfgTarget { platform: string; @@ -34,6 +44,7 @@ export function ChannelsModule(): React.ReactNode { const [err, setErr] = useState(null); const [note, setNote] = useState(null); const [cfg, setCfg] = useState(null); + const [logFor, setLogFor] = useState<{ adapter: string | null } | null>(null); const [ownerId, setOwnerId] = useState(""); const [platform, setPlatform] = useState(""); @@ -157,6 +168,14 @@ export function ChannelsModule(): React.ReactNode { {tokenMasked ? "token 已配置" : "token 未填"} + {a.id ? ( + + + + } /> @@ -266,6 +290,10 @@ export function ChannelsModule(): React.ReactNode { /> ) : null} + {logFor ? ( + setLogFor(null)} /> + ) : null} +

身份识别测试（结构性判定 · 绝非提示词）

@@ -304,3 +332,67 @@ export function ChannelsModule(): React.ReactNode { ); } + +/** Overlay that renders one adapter config's send/receive message log, or the + * adapter overview (adapter == null) across every adapter. */ +function ChannelLogPanel({ adapter, onClose }: { adapter: string | null; onClose: () => void }): React.ReactNode { + const isOverview = adapter == null; + const platform = adapter ? platformOf(adapter) : ""; + const { data, loading, error, reload } = useAsyncData( + () => (adapter ? adminChannelMessages(adapter, 200) : adminChannelLogOverview(200)), + [adapter], + ); + // Backend returns newest-first; display oldest→newest for a natural read. + const msgs = data ? [...data].reverse() : data; + + return ( +

+
e.stopPropagation()}> +
+ {platform ? {PLATFORM_LABEL[platform] ?? platform} : 总览} + {adapter ? {adapter} : null} +

{isOverview ? "适配器总览日志" : "消息日志"}

+ GET /api/v1/admin/channels/messages + + +
+
+ {loading ? ( + 加载中… + ) : error ? ( + {error} + ) : !msgs || msgs.length === 0 ? ( + 暂无消息。向该渠道发送消息、或接入 /ws/chat 发起会话后，这里会显示收发记录。 + ) : ( + msgs.map((m) => ) + )} +
+
+
+ ); +} + +/** One message line in the adapter log. */ +function ChannelLogRow({ m, showAdapter }: { m: AdminChannelMessage; showAdapter?: boolean }): React.ReactNode { + const isAssistant = m.role === "assistant"; + return ( +
+
+ {isAssistant ? "她 / 昔涟" : "用户"} + {m.senderName ? {m.senderName} : null} + {showAdapter && m.adapterId ? {m.adapterId} : null} + {m.deviceId ? {m.deviceId} : null} + {new Date(m.createdAt).toLocaleString()} +
+
{m.content}
+
+ ); +} diff --git a/frontend/web/src/lib/api.ts b/frontend/web/src/lib/api.ts index 825328a..0d28d5d 100644 --- a/frontend/web/src/lib/api.ts +++ b/frontend/web/src/lib/api.ts @@ -1,5 +1,6 @@ import type { AdminAutomationRecord, + AdminChannelMessage, AdminChannelRecord, AdminIoTDevice, AdminLlmConfig, @@ -546,6 +547,18 @@ export function adminChannelTest(req: { }): Promise { return request("/admin/channels/test", { method: "POST", body: JSON.stringify(req) }); } +/** GET /api/v1/admin/channels/messages?adapter= — one adapter config's + * (platform:account) send/receive message log, newest first. */ +export function adminChannelMessages(adapter: string, limit?: number): Promise { + const q = new URLSearchParams({ adapter }); + if (limit && limit > 0) q.set("limit", String(limit)); + return request(`/admin/channels/messages?${q.toString()}`); +} +/** GET /api/v1/admin/channels/messages — adapter overview log across all adapters. */ +export function adminChannelLogOverview(limit?: number): Promise { + const q = limit && limit > 0 ? `?limit=${limit}` : ""; + return request(`/admin/channels/messages${q}`); +} /** PUT /api/v1/admin/channels/{platform}/config — save an account config + hot-reload. */ export function adminSaveChannelConfig( platform: string, diff --git a/frontend/web/src/lib/types.ts b/frontend/web/src/lib/types.ts index 590c796..eb63404 100644 --- a/frontend/web/src/lib/types.ts +++ b/frontend/web/src/lib/types.ts @@ -275,6 +275,23 @@ export interface AdminChannelAccount { botNickname?: string; } +/** One message in the platform (IM channel) send/receive log. Mirrors the + * historyMessage wire shape from GET /api/v1/admin/channels/{platform}/messages. */ +export interface AdminChannelMessage { + id: string; + sessionId: string; + role: string; // "user" | "assistant" + content: string; + mode?: string; + platform?: string; + deviceId?: string; + /** Adapter (platform:account) that produced the message; blank for Web. */ + adapterId?: string; + /** Human sender display name for group/stranger messages; blank = owner. */ + senderName?: string; + createdAt: number; +} + /** An IoT device as returned by GET /api/v1/admin/iot. */ export interface AdminIoTDevice { id: string; diff --git a/internal/db/migrations/0007_chat_adapter.down.sql b/internal/db/migrations/0007_chat_adapter.down.sql new file mode 100644 index 0000000..61cadce --- /dev/null +++ b/internal/db/migrations/0007_chat_adapter.down.sql @@ -0,0 +1,6 @@ +-- 0007_chat_adapter down: drop the adapter_id column and its index. +-- SQLite DROP COLUMN (3.35+) cannot drop a column that is still referenced by an +-- index, so the adapter index is dropped first, then the column. + +DROP INDEX IF EXISTS idx_chat_msgs_adapter_created; +ALTER TABLE chat_messages DROP COLUMN adapter_id; diff --git a/internal/db/migrations/0007_chat_adapter.up.sql b/internal/db/migrations/0007_chat_adapter.up.sql new file mode 100644 index 0000000..6fb65c7 --- /dev/null +++ b/internal/db/migrations/0007_chat_adapter.up.sql @@ -0,0 +1,15 @@ +-- 0007_chat_adapter: tag every chat message with the adapter (platform + account) +-- that received it, e.g. "kook:3bef..." for a multi-account channels.json registration +-- or a bare platform name (e.g. "kook") when the account comes from env. It lets the +-- admin panel query the message send/receive log per independent adapter config, and +-- the full adapter overview (adapter_id != ''). +-- +-- IM adapter-originated messages carry this tag; Web main-session messages stay empty, +-- so only IM adapter sources are surfaced. The 0006 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 adapter_id TEXT NOT NULL DEFAULT ''; +CREATE INDEX IF NOT EXISTS idx_chat_msgs_adapter_created ON chat_messages (adapter_id, created_at DESC); diff --git a/internal/runtime/history/store.go b/internal/runtime/history/store.go index 44396e2..8905d84 100644 --- a/internal/runtime/history/store.go +++ b/internal/runtime/history/store.go @@ -94,6 +94,12 @@ type Message struct { // the main session (the owner is never labeled). It is display metadata only — // identity is never decided from this field. SenderName string + // AdapterID is the adapter (platform + account) registration name of the IM + // source that produced the message, e.g. "kook:3bef..." for a multi-account + // channels.json registration or a bare platform name (e.g. "kook") when the + // account comes from env. IM adapter-originated messages carry it; the Web main + // session leaves it empty, so it is always '' for Web messages. + AdapterID string CreatedAt int64 } @@ -130,6 +136,18 @@ type HistoryStore interface { // 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) + // ListAdapterLog returns up to limit messages for a single IM adapter (platform + + // account registration name, e.g. "kook:3bef..."), newest first, for the + // per-adapter log page. Unlike ListPlatformLog it does NOT exclude the main web + // session: an owner-private IM chat lives in the main session but still carries + // the adapter_id, so a per-adapter log includes the owner mirror. before is a + // millisecond cursor; 0 means the newest page. An empty adapterID returns an + // empty slice. + ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]Message, error) + // ListAllLog returns up to limit messages across every IM adapter source + // (adapter_id != ''), newest first, for the adapter overview page. before is a + // millisecond cursor; 0 means the newest page. + ListAllLog(ctx context.Context, 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 @@ -212,10 +230,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, sender_name, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + (id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, msg.ID, msg.SessionID, msg.Role, msg.Content, msg.Mode, msg.Mood, - msg.Image, msg.Thinking, msg.Platform, msg.DeviceID, msg.SenderName, msg.CreatedAt) + msg.Image, msg.Thinking, msg.Platform, msg.DeviceID, msg.SenderName, msg.AdapterID, msg.CreatedAt) if err != nil { return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) } @@ -231,7 +249,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, sender_name, created_at + query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at FROM chat_messages WHERE session_id = ?` args := []any{sessionID} if before > 0 { @@ -272,7 +290,7 @@ func (s *Store) List(ctx context.Context, sessionID string, before int64, limit // 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 + query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at FROM chat_messages WHERE platform = ? AND session_id != ?` args := []any{platform, MainSessionID} if before > 0 { @@ -302,13 +320,91 @@ func (s *Store) ListPlatformLog(ctx context.Context, platform string, before int return msgs, nil } +// ListAdapterLog returns up to limit messages for a single IM adapter (platform + +// account registration name, e.g. "kook:3bef..."), newest first, for the per-adapter +// log page. It does NOT exclude the main web session: an owner-private IM chat lives +// in the main session but still carries the adapter_id, so a per-adapter log includes +// the owner mirror. An empty adapterID returns an empty (non-nil) slice. before is a +// millisecond cursor; when nonzero only strictly older messages are returned. The +// returned slice is non-nil. +func (s *Store) ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]Message, error) { + limit = clampLimit(limit) + if adapterID == "" { + return make([]Message, 0), nil + } + query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at + FROM chat_messages WHERE adapter_id = ?` + args := []any{adapterID} + 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: adapter log %s: %w", adapterID, 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 adapter log %s: %w", adapterID, err) + } + msgs = append(msgs, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate adapter log %s: %w", adapterID, err) + } + return msgs, nil +} + +// ListAllLog returns up to limit messages across every IM adapter source +// (adapter_id != ''), newest first, for the adapter overview page. before is a +// millisecond cursor; when nonzero only strictly older messages are returned. The +// returned slice is non-nil. +func (s *Store) ListAllLog(ctx context.Context, 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, adapter_id, created_at + FROM chat_messages WHERE adapter_id != ''` + var args []any + 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: all adapter log: %w", 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 all adapter log: %w", err) + } + msgs = append(msgs, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate all adapter log: %w", 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, sender_name, created_at + rows, err := s.db.QueryContext(ctx, `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, 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) @@ -422,11 +518,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, 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 { + var mode, mood, platform, deviceID, senderName, adapterID string + if err := sc.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &mode, &mood, &m.Image, &m.Thinking, &platform, &deviceID, &senderName, &adapterID, &m.CreatedAt); err != nil { return Message{}, err } - m.Mode, m.Mood, m.Platform, m.DeviceID, m.SenderName = mode, mood, platform, deviceID, senderName + m.Mode, m.Mood, m.Platform, m.DeviceID, m.SenderName, m.AdapterID = mode, mood, platform, deviceID, senderName, adapterID return m, nil } @@ -484,6 +580,7 @@ CREATE TABLE IF NOT EXISTS chat_messages ( platform TEXT NOT NULL DEFAULT 'web', device_id TEXT NOT NULL DEFAULT '', sender_name TEXT NOT NULL DEFAULT '', + adapter_id TEXT NOT NULL DEFAULT '', created_at INTEGER NOT NULL )` diff --git a/internal/runtime/history/store_test.go b/internal/runtime/history/store_test.go index 21ce773..5da14cc 100644 --- a/internal/runtime/history/store_test.go +++ b/internal/runtime/history/store_test.go @@ -423,3 +423,116 @@ func TestSenderNameRoundTripOnListRecentAndPlatformLog(t *testing.T) { t.Fatalf("web platform log = %+v, err=%v; want empty", log, err) } } + +func TestAdapterLogFiltersByAdapter(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + // Seed across two adapters, an empty-adapter web message, and a main-session + // owner-private mirror that still carries the adapter (design/16 §4.4). + for _, m := range []Message{ + {SessionID: MainSessionID, Role: RoleUser, Content: "kook-owner", Platform: "kook", AdapterID: "kook:3bef", CreatedAt: 1000}, + {SessionID: "sess-g1", Role: RoleUser, Content: "kook-group", Platform: "kook", AdapterID: "kook:3bef", SenderName: "张三", CreatedAt: 2000}, + {SessionID: "sess-g2", Role: RoleUser, Content: "tg-group", Platform: "telegram", AdapterID: "telegram:conf0", SenderName: "李四", CreatedAt: 3000}, + {SessionID: MainSessionID, Role: RoleUser, Content: "web-msg", Platform: PlatformWeb, CreatedAt: 4000}, + } { + if err := st.Append(ctx, m); err != nil { + t.Fatalf("seed %q: %v", m.Content, err) + } + } + + // Per-adapter log includes BOTH the owner-private main mirror and the group + // message, newest first. + kook, err := st.ListAdapterLog(ctx, "kook:3bef", 0, 50) + if err != nil { + t.Fatalf("adapter log kook:3bef: %v", err) + } + if len(kook) != 2 { + t.Fatalf("kook adapter log = %d, want 2 (%+v)", len(kook), kook) + } + if kook[0].Content != "kook-group" || kook[0].SessionID != "sess-g1" { + t.Fatalf("kook[0] = %+v, want newest kook-group", kook[0]) + } + if kook[1].Content != "kook-owner" || kook[1].SessionID != MainSessionID { + t.Fatalf("kook[1] = %+v, want owner mirror on main session", kook[1]) + } + + // A different adapter yields only its own lane. + tg, err := st.ListAdapterLog(ctx, "telegram:conf0", 0, 50) + if err != nil { + t.Fatalf("adapter log telegram:conf0: %v", err) + } + if len(tg) != 1 || tg[0].Content != "tg-group" { + t.Fatalf("telegram adapter log = %+v, want [tg-group]", tg) + } + + // An unknown / empty adapter yields a non-nil empty slice. + none, err := st.ListAdapterLog(ctx, "kook:unknown", 0, 50) + if err != nil { + t.Fatalf("adapter log unknown: %v", err) + } + if none == nil || len(none) != 0 { + t.Fatalf("unknown adapter log = %#v, want non-nil empty slice", none) + } + empty, err := st.ListAdapterLog(ctx, "", 0, 50) + if err != nil { + t.Fatalf("adapter log empty: %v", err) + } + if empty == nil || len(empty) != 0 { + t.Fatalf("empty adapter log = %#v, want non-nil empty slice", empty) + } + + // The before cursor pages backward through the adapter's stream. + older, err := st.ListAdapterLog(ctx, "kook:3bef", 2000, 50) + if err != nil { + t.Fatalf("adapter log before: %v", err) + } + if len(older) != 1 || older[0].Content != "kook-owner" { + t.Fatalf("kook before page = %+v, want [kook-owner]", older) + } +} + +func TestAllLogSurfacesOnlyAdapterMessages(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + for _, m := range []Message{ + {SessionID: MainSessionID, Role: RoleUser, Content: "kook-owner", Platform: "kook", AdapterID: "kook:3bef", CreatedAt: 1000}, + {SessionID: MainSessionID, Role: RoleUser, Content: "web-msg", Platform: PlatformWeb, CreatedAt: 2000}, + {SessionID: "sess-g1", Role: RoleUser, Content: "tg-group", Platform: "telegram", AdapterID: "telegram:conf0", CreatedAt: 3000}, + } { + if err := st.Append(ctx, m); err != nil { + t.Fatalf("seed %q: %v", m.Content, err) + } + } + + // The overview only surfaces IM adapter sources; the web message is excluded. + all, err := st.ListAllLog(ctx, 0, 50) + if err != nil { + t.Fatalf("all log: %v", err) + } + if len(all) != 2 { + t.Fatalf("all log = %d, want 2 (%+v)", len(all), all) + } + if all[0].Content != "tg-group" || all[1].Content != "kook-owner" { + t.Fatalf("all log order = %+v, want [tg-group kook-owner]", all) + } + // The web message never leaks into the adapter overview. + for _, m := range all { + if m.Content == "web-msg" { + t.Fatalf("web message leaked into adapter overview: %+v", m) + } + } + + // An empty store yields a non-nil empty slice. + st2 := newStore(t) + e, err := st2.ListAllLog(ctx, 0, 50) + if err != nil { + t.Fatalf("empty all log: %v", err) + } + if e == nil || len(e) != 0 { + t.Fatalf("empty all log = %#v, want non-nil empty slice", e) + } +} diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index 8086cf3..cded587 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -356,6 +356,18 @@ func basePlatform(name string) string { return name } +// adapterName returns the account-adapter registration name (e.g. "kook:3bef...") +// for the adapter that received a message, or "" when a is nil. The single-account +// convenience path (process) and a platform with no registered adapter can pass a nil +// adapter, so the persist hooks tolerate it and store an empty adapter_id rather than +// dereferencing a nil adapter. +func adapterName(a adapter.PlatformAdapter) string { + if a == nil { + return "" + } + return a.Name() +} + // readLoop runs an adapter's blocking read path until the bridge stops or the // adapter's own context is cancelled (a hot reload). The emit sink is wired to // onMessage so poll-style adapters (Telegram getUpdates, weixin long poll) @@ -485,7 +497,7 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a // owner-private chat and owner group @/high-value replies are unchanged. if !actor.IsOwner() { if ok, reason := b.shouldRespond(ctx, msg, actor); !ok { - b.registerSilenced(ctx, msg, actor, sess) + b.registerSilenced(ctx, msg, actor, sess, a) b.log.Debug("non-owner message silenced (minimal disturbance)", "platform", msg.Platform, "channel", msg.ChannelID, "session", sess.SessionID, "reason", reason) @@ -519,9 +531,9 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a // 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)) + b.persistToMain(ctx, history.RoleUser, text, platform, deviceIDFor(msg), adapterName(a)) } else { - b.persistIsolated(ctx, history.RoleUser, text, deviceIDFor(msg), sess, msg.Platform, senderName) + b.persistIsolated(ctx, history.RoleUser, text, deviceIDFor(msg), sess, msg.Platform, senderName, adapterName(a)) } res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), sessID, loopMsgs) @@ -539,9 +551,9 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a // 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)) + b.persistToMain(ctx, history.RoleAssistant, reply, platform, deviceIDFor(msg), adapterName(a)) } else { - b.persistIsolated(ctx, history.RoleAssistant, reply, "", sess, msg.Platform, "") + b.persistIsolated(ctx, history.RoleAssistant, reply, "", sess, msg.Platform, "", adapterName(a)) } if err := b.sendReply(ctx, msg, a, reply); err != nil { if errors.Is(err, adapter.ErrUnsupported) { diff --git a/internal/services/channelbridge/channelbridge_test.go b/internal/services/channelbridge/channelbridge_test.go index 367293f..01ae188 100644 --- a/internal/services/channelbridge/channelbridge_test.go +++ b/internal/services/channelbridge/channelbridge_test.go @@ -255,6 +255,12 @@ func (c *captureHistory) Recent(context.Context, string, int) ([]history.Message func (c *captureHistory) ListPlatformLog(context.Context, string, int64, int) ([]history.Message, error) { return nil, nil } +func (c *captureHistory) ListAdapterLog(context.Context, string, int64, int) ([]history.Message, error) { + return nil, nil +} +func (c *captureHistory) ListAllLog(context.Context, 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 @@ -284,6 +290,9 @@ func TestOwnerPrivatePersistsToMainSessionWithPlatform(t *testing.T) { if err != nil { t.Fatalf("New: %v", err) } + // Register the account adapter that pretends to receive the message, so the + // persisted stream carries the adapter registration name (e.g. "telegram"). + b.Add(&fakeAdapter{name: "telegram"}) msg := directMessage() // Platform = "telegram" b.process(context.Background(), msg) @@ -292,14 +301,14 @@ func TestOwnerPrivatePersistsToMainSessionWithPlatform(t *testing.T) { if len(recs) != 2 { t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs)) } - // Both land on the main web session and carry the channel platform. + // Both land on the main web session and carry the channel platform + adapter. if recs[0].SessionID != history.MainSessionID || recs[0].Role != history.RoleUser || - recs[0].Content != "hello" || recs[0].Platform != "telegram" { - t.Fatalf("user record = %+v, want web/telegram/hello", recs[0]) + recs[0].Content != "hello" || recs[0].Platform != "telegram" || recs[0].AdapterID != "telegram" { + t.Fatalf("user record = %+v, want web/telegram/hello/telegram", recs[0]) } if recs[1].SessionID != history.MainSessionID || recs[1].Role != history.RoleAssistant || - recs[1].Content != "微信回复" || recs[1].Platform != "telegram" { - t.Fatalf("assistant record = %+v, want web/telegram/微信回复", recs[1]) + recs[1].Content != "微信回复" || recs[1].Platform != "telegram" || recs[1].AdapterID != "telegram" { + t.Fatalf("assistant record = %+v, want web/telegram/微信回复/telegram", recs[1]) } // Every persisted message was handed to the broadcast hook. if len(persisted) != 2 { diff --git a/internal/services/channelbridge/light.go b/internal/services/channelbridge/light.go index 3f49e0e..1733663 100644 --- a/internal/services/channelbridge/light.go +++ b/internal/services/channelbridge/light.go @@ -168,10 +168,11 @@ func (b *Bridge) shouldRespond(ctx context.Context, msg adapter.UnifiedMessage, // registerSilenced persists a silenced non-owner inbound turn into its isolated // session's platform log so the adapter log page still renders the non-owner -// stream (design/16 §4.6 §4), then stops — no loop, no reply. It is the 登记 side -// of the minimal-disturbance path. -func (b *Bridge) registerSilenced(ctx context.Context, msg adapter.UnifiedMessage, actor adapter.Actor, sess session.Session) { - b.persistIsolated(ctx, history.RoleUser, msg.Content, deviceIDFor(msg), sess, msg.Platform, senderDisplayName(msg, actor)) +// stream (design/16 §4.6 §4), then stops — no loop, no reply. a is the account +// adapter that received the message, so the log is tagged with the adapter config. +// It is the 登记 side of the minimal-disturbance path. +func (b *Bridge) registerSilenced(ctx context.Context, msg adapter.UnifiedMessage, actor adapter.Actor, sess session.Session, a adapter.PlatformAdapter) { + b.persistIsolated(ctx, history.RoleUser, msg.Content, deviceIDFor(msg), sess, msg.Platform, senderDisplayName(msg, actor), adapterName(a)) b.log.Debug("non-owner message silenced, registered only", "platform", msg.Platform, "channel", msg.ChannelID, "session", sess.SessionID) } diff --git a/internal/services/channelbridge/persist.go b/internal/services/channelbridge/persist.go index d15d32c..0576cc7 100644 --- a/internal/services/channelbridge/persist.go +++ b/internal/services/channelbridge/persist.go @@ -50,17 +50,19 @@ func (b *Bridge) notifyPersisted(msg history.Message) { // platform and client deviceID, then notifies the transport hook so Web clients receive // a `message_persisted` event without a refresh (design/16 §4.4). deviceID is the // concrete client that sent the message (design/16 §4.5) — e.g. "weixin-ilink:conf0:wx_uid" -// for IM. It is a no-op when no history store is wired, so the bridge runs unchanged -// without storage. -func (b *Bridge) persistToMain(ctx context.Context, role, content, platform, deviceID string) { +// for IM. adapterID is the account adapter registration name that received the message +// (e.g. "kook:3bef..."), so the admin log can query a per-adapter send/receive stream; +// it is empty for Web-originated messages. It is a no-op when no history store is wired, +// so the bridge runs unchanged without storage. +func (b *Bridge) persistToMain(ctx context.Context, role, content, platform, deviceID, adapterID string) { if b.history == nil { return } full, err := b.history.AppendFull(ctx, history.Message{ - SessionID: history.MainSessionID, Role: role, Content: content, Platform: platform, DeviceID: deviceID, + SessionID: history.MainSessionID, Role: role, Content: content, Platform: platform, DeviceID: deviceID, AdapterID: adapterID, }) if err != nil { - b.log.Warn("persist main session", "err", err, "role", role, "platform", platform, "device_id", deviceID) + b.log.Warn("persist main session", "err", err, "role", role, "platform", platform, "device_id", deviceID, "adapter", adapterID) return } b.notifyPersisted(full) @@ -69,21 +71,22 @@ func (b *Bridge) persistToMain(ctx context.Context, role, content, platform, dev // 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) { +// deviceID, the sender's display name and the account adapter registration name, so +// the IM adapter log page can render the non-owner message stream per adapter config. +// 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, adapterID 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, + DeviceID: deviceID, SenderName: senderName, AdapterID: adapterID, }) if err != nil { b.log.Warn("persist isolated session", "err", err, "role", role, - "session", sess.SessionID, "platform", platform, "sender_name", senderName) + "session", sess.SessionID, "platform", platform, "sender_name", senderName, "adapter", adapterID) return } // No broadcast: third-party messages must not reach the Web main session (design/16 §4.6 §3). diff --git a/internal/services/gateway/history.go b/internal/services/gateway/history.go index 3fb74ad..c8d85b1 100644 --- a/internal/services/gateway/history.go +++ b/internal/services/gateway/history.go @@ -38,6 +38,10 @@ type historyMessage struct { // 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"` + // AdapterID is the adapter (platform + account) registration name of the IM + // source that produced the message (e.g. "kook:3bef..."). It is blank for + // Web-originated messages. + AdapterID string `json:"adapterId,omitempty"` CreatedAt int64 `json:"createdAt"` } @@ -175,8 +179,62 @@ func (s *Server) handleChannelPlatformLog(w http.ResponseWriter, r *http.Request 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. +// handleChannelMessages returns up to limit messages for the admin adapter log page, +// newest first, across one of three granularities (design/16 §4.6 §4): a single +// adapter config (?adapter=kook:3bef...), a whole platform (?platform=kook), or the +// full IM adapter overview (neither, so adapter_id != ''). before/limit reuse the +// standard cursor pagination. The owner-private mirror on the main web session is +// NOT excluded here because it carries the adapter_id; each granularity filters on +// adapter_id / platform directly. +func (s *Server) handleChannelMessages(w http.ResponseWriter, r *http.Request) { + if s.history == nil { + writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置（未注入 HistoryStore）", HTTPCode: http.StatusNotImplemented}) + 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 + } + + adapter := r.URL.Query().Get("adapter") + platform := r.URL.Query().Get("platform") + + var msgs []history.Message + var err error + switch { + case adapter != "": + msgs, err = s.history.ListAdapterLog(r.Context(), adapter, before, limit) + case platform != "": + msgs, err = s.history.ListPlatformLog(r.Context(), platform, before, limit) + default: + msgs, err = s.history.ListAllLog(r.Context(), before, limit) + } + if err != nil { + s.log.Error("history channel messages", "err", err, "adapter", adapter, "platform", platform) + writeError(w, apperr.Wrap(apperr.ErrInternal, "channel_messages_failed", "failed to list channel messages")) + return + } + out := make([]historyMessage, 0, len(msgs)) + for _, m := range msgs { + out = append(out, toHistoryMessage(m)) + } + writeJSON(w, http.StatusOK, out) +} func (s *Server) handleDeleteSession(w http.ResponseWriter, r *http.Request) { if s.history == nil { writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置（未注入 HistoryStore）", HTTPCode: http.StatusNotImplemented}) @@ -213,6 +271,7 @@ func toHistoryMessage(m history.Message) historyMessage { Platform: m.Platform, DeviceID: m.DeviceID, SenderName: m.SenderName, + AdapterID: m.AdapterID, CreatedAt: m.CreatedAt, } } diff --git a/internal/services/gateway/history_handler_test.go b/internal/services/gateway/history_handler_test.go index 677d3f5..0cde6df 100644 --- a/internal/services/gateway/history_handler_test.go +++ b/internal/services/gateway/history_handler_test.go @@ -168,6 +168,77 @@ func TestHistoryListMessagesPlatformFilter(t *testing.T) { } } +// handleChannelMessagesGranularity exercises the admin adapter log endpoint at each +// granularity: a single adapter config (?adapter=), a whole platform (?platform=), +// and the full IM adapter overview (neither). It asserts adapterId is carried on the +// DTO so the panel can label the source adapter config. +func TestHistoryChannelMessagesGranularity(t *testing.T) { + t.Parallel() + _, ts, jwt, st := newHistoryEnv(t) + ctx := context.Background() + for _, m := range []history.Message{ + {SessionID: history.MainSessionID, Role: history.RoleUser, Content: "kook-owner", Platform: "kook", AdapterID: "kook:3bef", CreatedAt: 1000}, + {SessionID: "sess-g", Role: history.RoleUser, Content: "kook-group", Platform: "kook", AdapterID: "kook:3bef", SenderName: "张三", CreatedAt: 2000}, + {SessionID: "sess-t", Role: history.RoleUser, Content: "tg-group", Platform: "telegram", AdapterID: "telegram:conf0", CreatedAt: 3000}, + {SessionID: history.MainSessionID, Role: history.RoleUser, Content: "web-msg", Platform: history.PlatformWeb, CreatedAt: 4000}, + } { + if err := st.Append(ctx, m); err != nil { + t.Fatalf("seed %q: %v", m.Content, err) + } + } + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // ?adapter=kook:3bef → only that adapter config, both the owner mirror and the + // group message, newest first, and the DTO carries adapterId. + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/channels/messages?adapter=kook:3bef", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("adapter status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var byAdapter []historyMessage + decodeDataSlice(t, body, &byAdapter) + if len(byAdapter) != 2 { + t.Fatalf("adapter page = %d, want 2 (%+v)", len(byAdapter), byAdapter) + } + if byAdapter[0].Content != "kook-group" || byAdapter[0].AdapterID != "kook:3bef" { + t.Fatalf("adapter[0] = %+v, want kook-group with adapterId", byAdapter[0]) + } + if byAdapter[1].Content != "kook-owner" || byAdapter[1].SessionID != history.MainSessionID { + t.Fatalf("adapter[1] = %+v, want kook-owner on main session", byAdapter[1]) + } + + // ?platform=kook → the platform lane across isolated sessions (owner-private + // mirror excluded, matching handleChannelPlatformLog semantics). + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/channels/messages?platform=kook", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("platform status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var byPlatform []historyMessage + decodeDataSlice(t, body, &byPlatform) + if len(byPlatform) != 1 || byPlatform[0].Content != "kook-group" { + t.Fatalf("platform page = %+v, want [kook-group]", byPlatform) + } + + // No filter → the full IM adapter overview (adapter_id != ''), web excluded. + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/channels/messages", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("all status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var all []historyMessage + decodeDataSlice(t, body, &all) + if len(all) != 3 { + t.Fatalf("all page = %d, want 3 (%+v)", len(all), all) + } + if all[0].Content != "tg-group" || all[1].Content != "kook-group" || all[2].Content != "kook-owner" { + t.Fatalf("all order = %+v, want [tg-group kook-group kook-owner]", all) + } + + // Invalid before → 400. + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/channels/messages?before=abc", nil, tok) + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("invalid before status = %d, want 400 (body=%s)", resp.StatusCode, body) + } +} + func TestHistoryDeleteSubSession(t *testing.T) { t.Parallel() _, ts, jwt, st := newHistoryEnv(t) diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 17dde9e..5874452 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -115,6 +115,8 @@ type HistoryStore interface { 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) + ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]history.Message, error) + ListAllLog(ctx context.Context, 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) @@ -380,6 +382,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/messages", adminAuth(http.HandlerFunc(s.handleChannelMessages))) 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))) diff --git a/internal/services/gateway/session_orchestrator_test.go b/internal/services/gateway/session_orchestrator_test.go index 5e07ef6..ed6daa8 100644 --- a/internal/services/gateway/session_orchestrator_test.go +++ b/internal/services/gateway/session_orchestrator_test.go @@ -187,6 +187,14 @@ func (f *fakeHistory) ListPlatformLog(context.Context, string, int64, int) ([]hi return nil, nil } +func (f *fakeHistory) ListAdapterLog(context.Context, string, int64, int) ([]history.Message, error) { + return nil, nil +} + +func (f *fakeHistory) ListAllLog(context.Context, int64, int) ([]history.Message, error) { + return nil, nil +} + func (f *fakeHistory) DeleteSession(context.Context, string) error { return nil }