AskaEth/Cyrene_v2
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(channels): 适配器日志接 WS 实时推送(落库即秒推, channel_persisted)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
Details
CI / Frontend (tsc + vite build) (push) Has been cancelled
Details

Browse Source 
后端: persistIsolated 由 _ = full(不广播)改为 notifyPersisted(full); BroadcastPersisted 按
sessionId 分流——main/web:<uuid> → message_persisted(Web聊天同步), 隔离 sess-xxx → channel_persisted
(适配器日志专用, 永不写主会话 web, 红线). payload data:{message:historyMessage}(strip thinking).
前端: ws.ts 新增 channel_persisted case→parseAdminChannelMessage→pushChannelEvent; wsParse.ts 纯函数;
useAppStore 新增 channelEvents(最新在前,按id去重,cap200,不持久化); channelLogList 订阅过滤
(单adapter按adapterId/总览取adapterId非空)merge进列表(仅追加、id去重、不整页重载),保留拉历史+10s兜底.
Docs/api-client.md §3.4 补 channel_persisted 契约.
验收: 后端 build/vet/test 全绿+windows/darwin交叉编译绿; 前端 npm run build 99模块绿.
注: 偏离=web:<uuid> 归 message_persisted(保Web副会话同步), 隔离永不带web前缀红线仍守.
This commit is contained in:
Aska committed 2026-09-07 18:51:00 +08:00
1 parent 9e7b4ebaac
commit 3386679d45
10 files changed
+180 -23
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/services/channelbridge/channelbridge.go
+8 -5
View File
@@ -79,8 +79,9 @@ type SessionStore interface {
// (design/16 §4.4): when a history store is wired, owner-private messages are
// routed into the main web session and tagged with the channel platform, and every
// persisted message is handed to OnPersisted so the gateway can broadcast a
// `message_persisted` event; when both are absent the bridge runs exactly as
// before (isolated sess-xxx, no chat_messages writes).
// `message_persisted` (web chat) or `channel_persisted` (IM adapter log) event;
// when both are absent the bridge runs exactly as before (isolated sess-xxx, no
// chat_messages writes).
type Options struct {
Log *slog.Logger
Loop Chatter
@@ -572,7 +573,8 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a
// 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.
// session (the adapter log page) and are broadcast as a `channel_persisted` event
// (never into the main/`web` session or the web chat).
if isOwnerPrivate {
b.persistToMain(ctx, history.RoleUser, text, platform, deviceIDFor(msg), adapterName(a))
} else {
@@ -591,8 +593,9 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a
b.log.Debug("no reply produced", "platform", msg.Platform, "session", sessID)
return
}
// Persist the assistant reply (owner-private → main session + broadcast;
// restricted → isolated log) then deliver it back to the channel.
// Persist the assistant reply (owner-private → main session + message_persisted
// broadcast; restricted → isolated log + channel_persisted broadcast) then
// deliver it back to the channel.
if isOwnerPrivate {
b.persistToMain(ctx, history.RoleAssistant, reply, platform, deviceIDFor(msg), adapterName(a))
} else {
internal/services/channelbridge/channelbridge_test.go
+20 -1
View File
@@ -377,7 +377,11 @@ func TestGroupMessageNotPersistedToMainSession(t *testing.T) {
chat := &stubChatter{res: &loop.Result{Text: "群聊回复"}}
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})
var persisted []history.Message
b, err := New(Options{
Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist,
OnPersisted: func(m history.Message) { persisted = append(persisted, m) },
})
if err != nil {
t.Fatalf("New: %v", err)
}
@@ -411,6 +415,21 @@ func TestGroupMessageNotPersistedToMainSession(t *testing.T) {
if chat.sessionID != "sess-g" {
t.Fatalf("group loop session id = %q, want sess-g", chat.sessionID)
}
// Each isolated write is ALSO handed to the broadcast hook so the gateway can
// emit `channel_persisted` for the IM adapter log page — but the message itself
// stays strictly in the isolated session and never leaks to the main/`web`
// session (design/16 §4.6 §3).
if len(persisted) != 2 {
t.Fatalf("isolated persisted notifications = %d, want 2", len(persisted))
}
for i, p := range persisted {
if p.SessionID != "sess-g" {
t.Fatalf("persisted[%d] SessionID = %q, want isolated sess-g", i, p.SessionID)
}
if p.SessionID == history.MainSessionID {
t.Fatalf("persisted[%d] leaked to main web session", i)
}
}
}
func TestBlacklistGroupMessageDroppedBeforeLoop(t *testing.T) {
internal/services/channelbridge/persist.go
+13 -6
View File
@@ -36,7 +36,8 @@ func (b *Bridge) SetOnRejected(fn func(adapter.UnifiedMessage, string)) {
}
// 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.
// wired, so the gateway can broadcast a `message_persisted` (web chat) or
// `channel_persisted` (IM adapter log) event for cross-end realtime sync.
func (b *Bridge) notifyPersisted(msg history.Message) {
b.mu.RLock()
fn := b.onPersisted
@@ -73,9 +74,11 @@ func (b *Bridge) persistToMain(ctx context.Context, role, content, platform, dev
// session (NOT the main web session) and carries the channel platform, the client
// 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.
// The message is still NOT mirrored into the Web main session (only owner-private
// messages mirror to the Web); it is merely handed to the transport hook so the
// gateway broadcasts a `channel_persisted` event and the IM adapter log page appends
// it without a refresh. 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
@@ -89,6 +92,10 @@ func (b *Bridge) persistIsolated(ctx context.Context, role, content, deviceID st
"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).
_ = full
// Broadcast the isolated write as a `channel_persisted` event so the IM adapter
// log page appends it in realtime. The gateway splits the event type by session:
// this is NOT the main web session, so the web chat has no handler for it and the
// message itself still lands only in the isolated sess-xxx session (never the
// main/`web` session, design/16 §4.6 §3).
b.notifyPersisted(full)
}
internal/services/gateway/extra_test.go
+49
View File
@@ -24,6 +24,7 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
"git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader"
personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
@@ -935,6 +936,54 @@ func TestServerBroadcastCapacityAwaken(t *testing.T) {
}
}
// TestServerBroadcastPersistedSplit verifies BroadcastPersisted routes the event
// type by session: the main web session and its web sub-sessions broadcast
// `message_persisted` (web chat realtime sync), while an isolated channel adapter
// session (sess-xxx) broadcasts `channel_persisted` (IM adapter log page) so it is
// naturally ignored by the web chat handler.
func TestServerBroadcastPersistedSplit(t *testing.T) {
t.Parallel()
env := newEnv(t, nil, false)
c := newFakeConn(env.srv.hub, 32)
env.srv.hub.register(c)
cases := []struct {
name string
sessionID string
wantType string
}{
{name: "main web session", sessionID: history.MainSessionID, wantType: "message_persisted"},
{name: "web sub session", sessionID: history.MainSessionID + ":abc123", wantType: "message_persisted"},
{name: "isolated channel session", sessionID: "sess-g", wantType: "channel_persisted"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
env.srv.BroadcastPersisted(history.Message{
ID: "m-" + tc.sessionID, SessionID: tc.sessionID, Role: history.RoleUser,
Content: "hello", Platform: "telegram", AdapterID: "telegram:abc", CreatedAt: 1000,
})
ev := readFakeFrame(t, c)
if ev.Type != tc.wantType {
t.Fatalf("frame type = %q, want %q (frame=%+v)", ev.Type, tc.wantType, ev)
}
if ev.SessionID != tc.sessionID {
t.Fatalf("frame sessionId = %q, want %q", ev.SessionID, tc.sessionID)
}
data, ok := ev.Data.(map[string]any)
if !ok {
t.Fatalf("frame data = %+v, want map", ev.Data)
}
msg, ok := data["message"].(map[string]any)
if !ok {
t.Fatalf("frame data.message = %+v, want map", data["message"])
}
if msg["content"] != "hello" || msg["adapterId"] != "telegram:abc" {
t.Fatalf("frame message = %+v, want content/hello + adapterId/telegram:abc", msg)
}
})
}
}
func TestHubStreamChatIdle(t *testing.T) {
t.Parallel()
hb := newTestHub(nil, nil) // nil runner
internal/services/gateway/server.go
+17 -8
View File
@@ -25,6 +25,7 @@ import (
"net"
"net/http"
"strconv"
"strings"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
@@ -514,17 +515,25 @@ func newOrchestrator(s *Server) *Orchestrator {
})
}
// BroadcastPersisted broadcasts a just-persisted chat message as a
// `message_persisted` WS event so every browser of the same account appends it
// without a manual refresh (design/16 §4.2). It is shared by the orchestrator
// (Web turns) and the channel bridge (IM owner-private messages routed into the
// main web session), so both paths emit the identical frame. Thinking is stripped
// to stay consistent with the WS history replay (design/16 §6). It is a no-op-safe
// call when the hub has broadcast wired.
// BroadcastPersisted broadcasts a just-persisted chat message as a WS event so
// every browser of the same account appends it without a manual refresh
// (design/16 §4.2). It is shared by the orchestrator (Web turns) and the channel
// bridge (IM messages), so both paths flow through one frame shape. The event type
// is split by session: the main web session and its web sub-sessions broadcast
// `message_persisted` for web-chat realtime sync, while any other session id is an
// isolated channel adapter session (sess-xxx) and broadcasts `channel_persisted`
// for the IM adapter log page — the web chat has no handler for the latter, so it
// is naturally ignored there and never pollutes the conversation. Thinking is
// stripped to stay consistent with the WS history replay (design/16 §6). It is a
// no-op-safe call when the hub has broadcast wired.
func (s *Server) BroadcastPersisted(msg history.Message) {
wire := toHistoryMessage(msg)
wire.Thinking = ""
s.hub.broadcast(wsEvent{Type: "message_persisted", SessionID: msg.SessionID, Data: map[string]any{"message": wire}})
typ := "channel_persisted"
if msg.SessionID == history.MainSessionID || strings.HasPrefix(msg.SessionID, history.MainSessionID+":") {
typ = "message_persisted"
}
s.hub.broadcast(wsEvent{Type: typ, SessionID: msg.SessionID, Data: map[string]any{"message": wire}})
}
// wsCapacityAwaken is the WS event type for a capacity state transition