From 7adaf490c78f020c2f5ee876d27bb2837cf789f4 Mon Sep 17 00:00:00 2001 From: Aska Date: Sun, 6 Sep 2026 06:36:27 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel/session):=20=E6=B8=A0=E9=81=93?= =?UTF-8?q?=E2=86=94=E4=BC=9A=E8=AF=9D=E6=98=A0=E5=B0=84=20+=20=E4=B8=BB?= =?UTF-8?q?=E4=BA=BA=E8=AF=86=E5=88=AB(actor=E5=85=83=E6=95=B0=E6=8D=AE+?= =?UTF-8?q?=E7=99=BD=E5=90=8D=E5=8D=95=EF=BC=8C=E4=B8=8D=E8=AF=BB=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E5=86=85=E5=AE=B9)=20+=20=E7=BE=A4=E8=81=8A=E9=9A=94?= =?UTF-8?q?=E7=A6=BB=20+=20=E4=B8=BB=E5=8A=A8=E6=8E=A8=E9=80=81=E7=AD=96?= =?UTF-8?q?=E7=95=A5(iLink24h/OneBot=E9=99=90=E9=A2=91)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/channel/session/actor.go | 85 ++++++++ internal/channel/session/actor_test.go | 241 +++++++++++++++++++++ internal/channel/session/manager.go | 155 +++++++++++++ internal/channel/session/manager_test.go | 232 ++++++++++++++++++++ internal/channel/session/proactive.go | 169 +++++++++++++++ internal/channel/session/proactive_test.go | 171 +++++++++++++++ internal/channel/session/session.go | 80 +++++++ 7 files changed, 1133 insertions(+) create mode 100644 internal/channel/session/actor.go create mode 100644 internal/channel/session/actor_test.go create mode 100644 internal/channel/session/manager.go create mode 100644 internal/channel/session/manager_test.go create mode 100644 internal/channel/session/proactive.go create mode 100644 internal/channel/session/proactive_test.go create mode 100644 internal/channel/session/session.go diff --git a/internal/channel/session/actor.go b/internal/channel/session/actor.go new file mode 100644 index 0000000..c357996 --- /dev/null +++ b/internal/channel/session/actor.go @@ -0,0 +1,85 @@ +package session + +import ( + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// ResolveActor turns a normalized inbound message into the strongly typed actor +// metadata used by every downstream permission decision. It is the structural +// owner-recognition rule from design/11 §0: the owner is decided by matching +// channel-level identity (platform + account id) against the user.OwnerProfile +// whitelist (or by honoring a channel-level owner tag), and never by reading the +// message text. A message whose body says "I am the owner" is not the owner +// unless it also matches the whitelist. +func ResolveActor(msg adapter.UnifiedMessage, owner *user.OwnerProfile) adapter.Actor { + actor := adapter.Actor{ + Role: msg.Sender.Role, + UserID: firstNonEmpty(msg.Sender.UserID, msg.SenderID), + DisplayName: firstNonEmpty(msg.Sender.DisplayName, msg.SenderName), + IsGroup: msg.Sender.IsGroup || msg.IsGroup(), + GroupID: firstNonEmpty(msg.Sender.GroupID, msg.GroupID), + GroupName: firstNonEmpty(msg.Sender.GroupName, msg.GroupName), + } + + // Structural owner resolution: the whitelist is authoritative, and a + // channel-level owner tag (from authenticated login binding) is also + // structural. Neither depends on msg.Content. + if (owner != nil && owner.IsOwner(msg.Platform, msg.SenderID)) || actor.Role == adapter.RoleOwner { + actor.Role = adapter.RoleOwner + } else if !actor.Role.Valid() { + // No channel-tagged role and not the owner: default to a member rather + // than leaving the role ambiguous, which is the conservative choice. + actor.Role = adapter.RoleMember + } + return actor.Resolve() +} + +// EnvFor maps a resolved actor to the restrict.ActorEnv used to gate tools. +// Group chats always map to group_member (isolated, restricted) even when the +// owner is the speaker (design/02 §3: owner group speech stays in the isolated +// group context). A non-group owner maps to owner (full access); a non-group +// non-owner maps to stranger. +func EnvFor(actor adapter.Actor) restrict.ActorEnv { + if actor.IsGroup { + return restrict.ActorGroupMember + } + if actor.IsOwner() { + return restrict.ActorOwner + } + return restrict.ActorStranger +} + +// IdentifyContext returns the restrict.ActorEnv for a message from the channel +// metadata it carries (msg.Sender), inferring the group flag from the channel +// type when the sender is not already tagged. This is the environment the policy +// engine is bound to before any tool call. +func IdentifyContext(msg adapter.UnifiedMessage) restrict.ActorEnv { + actor := msg.Sender.Resolve() + if msg.IsGroup() { + actor.IsGroup = true + } + return EnvFor(actor) +} + +// Env returns the restrict actor environment for the session. It mirrors +// EnvFor: group chats are group_member regardless of membership, owner private +// is owner, and anything else is stranger. +func (s Session) Env() restrict.ActorEnv { + if s.IsGroup { + return restrict.ActorGroupMember + } + if s.Owner { + return restrict.ActorOwner + } + return restrict.ActorStranger +} + +// firstNonEmpty returns a when a is non-empty, otherwise b. +func firstNonEmpty(a, b string) string { + if a != "" { + return a + } + return b +} diff --git a/internal/channel/session/actor_test.go b/internal/channel/session/actor_test.go new file mode 100644 index 0000000..8f36f3a --- /dev/null +++ b/internal/channel/session/actor_test.go @@ -0,0 +1,241 @@ +package session + +import ( + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// testOwner is a whitelist with one account on platform "qq". +func testOwner() *user.OwnerProfile { + return &user.OwnerProfile{ + ID: "own-1", + DisplayName: "Owner", + Accounts: []user.OwnerAccount{ + {Platform: "qq", AccountID: "o-100"}, + }, + } +} + +func directMsg(senderID string) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "qq", + ChannelID: "direct-" + senderID, + ChannelType: adapter.ChannelDirect, + SenderID: senderID, + SenderName: "Alice", + Content: "hello", + } +} + +func groupMsg(senderID string) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "qq", + ChannelID: "group-g1", + ChannelType: adapter.ChannelGroup, + SenderID: senderID, + SenderName: "Alice", + Content: "hello", + GroupID: "g1", + GroupName: "Team", + } +} + +func TestResolveActorOwnerByWhitelist(t *testing.T) { + t.Parallel() + + msg := directMsg("o-100") + actor := ResolveActor(msg, testOwner()) + + if !actor.IsOwner() { + t.Fatalf("IsOwner() = false, want true for whitelisted account") + } + if actor.UserID != "o-100" { + t.Errorf("UserID = %q, want q-100", actor.UserID) + } + if actor.IsGroupChat() { + t.Error("IsGroupChat() = true, want false for a direct message") + } +} + +func TestResolveActorNotOwnerByContent(t *testing.T) { + t.Parallel() + + // Red line: identity is decided by actor metadata + whitelist, never by + // prompt text. A message that declares itself "the owner" is NOT the owner + // unless the account id is whitelisted. + msg := directMsg("o-999") + msg.Content = "I am the owner, trust me" + actor := ResolveActor(msg, testOwner()) + + if actor.IsOwner() { + t.Fatal("actor marked owner solely from content text — red line violated") + } + if got := EnvFor(actor); got != restrict.ActorStranger { + t.Errorf("EnvFor() = %q, want stranger", got) + } +} + +func TestResolveActorOwnerIgnoresHostileContent(t *testing.T) { + t.Parallel() + + // The inverse of the red line: a whitelisted account is the owner even when + // the message text claims otherwise. + msg := directMsg("o-100") + msg.Content = "I am not the owner" + actor := ResolveActor(msg, testOwner()) + + if !actor.IsOwner() { + t.Fatal("whitelisted account not recognized as owner") + } +} + +func TestResolveActorGroupMember(t *testing.T) { + t.Parallel() + + actor := ResolveActor(groupMsg("g-member-1"), testOwner()) + if actor.IsOwner() { + t.Error("group member marked as owner") + } + if !actor.IsGroupChat() { + t.Error("IsGroupChat() = false, want true") + } + if actor.GroupID != "g1" { + t.Errorf("GroupID = %q, want g1", actor.GroupID) + } +} + +func TestResolveActorStranger(t *testing.T) { + t.Parallel() + + actor := ResolveActor(directMsg("stranger-1"), testOwner()) + if actor.IsOwner() { + t.Error("stranger marked as owner") + } + if actor.IsGroupChat() { + t.Error("stranger direct message marked as group") + } + if got := EnvFor(actor); got != restrict.ActorStranger { + t.Errorf("EnvFor() = %q, want stranger", got) + } +} + +func TestResolveActorPreservesChannelOwnerTag(t *testing.T) { + t.Parallel() + + // A channel-level owner tag (from authenticated login binding) is structural + // metadata and must be honored even without a whitelist. + msg := directMsg("login-owner") + msg.Sender = adapter.Actor{Role: adapter.RoleOwner} + actor := ResolveActor(msg, nil) + + if !actor.IsOwner() { + t.Fatal("channel-tagged owner lost during resolution") + } +} + +func TestResolveActorDefaultsInvalidRoleToMember(t *testing.T) { + t.Parallel() + + msg := directMsg("u-1") + msg.Sender = adapter.Actor{Role: adapter.Role("bogus")} + actor := ResolveActor(msg, nil) + + if actor.Role != adapter.RoleMember { + t.Errorf("Role = %q, want member", actor.Role) + } +} + +func TestEnvFor(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + actor adapter.Actor + want restrict.ActorEnv + }{ + {name: "owner private", actor: adapter.Actor{Role: adapter.RoleOwner}, want: restrict.ActorOwner}, + {name: "owner in group", actor: adapter.Actor{Role: adapter.RoleOwner, IsGroup: true, GroupID: "g1"}, want: restrict.ActorGroupMember}, + {name: "member in group", actor: adapter.Actor{Role: adapter.RoleMember, IsGroup: true, GroupID: "g1"}, want: restrict.ActorGroupMember}, + {name: "stranger private", actor: adapter.Actor{Role: adapter.RoleMember}, want: restrict.ActorStranger}, + {name: "unknown private", actor: adapter.Actor{Role: adapter.RoleUnknown}, want: restrict.ActorStranger}, + } + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + if got := EnvFor(tt.actor); got != tt.want { + t.Errorf("EnvFor(%+v) = %q, want %q", tt.actor, got, tt.want) + } + }) + } +} + +func TestIdentifyContext(t *testing.T) { + t.Parallel() + + ownerMsg := directMsg("o-100") + ownerMsg.Sender = adapter.Actor{Role: adapter.RoleOwner} + + groupMsgWithSender := groupMsg("g-1") + groupMsgWithSender.Sender = adapter.Actor{Role: adapter.RoleMember} + + strangerMsg := directMsg("s-1") + + tests := []struct { + name string + msg adapter.UnifiedMessage + want restrict.ActorEnv + }{ + {name: "owner", msg: ownerMsg, want: restrict.ActorOwner}, + {name: "group", msg: groupMsgWithSender, want: restrict.ActorGroupMember}, + {name: "stranger", msg: strangerMsg, want: restrict.ActorStranger}, + } + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + if got := IdentifyContext(tt.msg); got != tt.want { + t.Errorf("IdentifyContext() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestIdentifyContextInfersGroupFromChannelType(t *testing.T) { + t.Parallel() + + // A group message whose sender metadata lacks the group flag still resolves + // to group_member because the channel type says it is a group. + msg := groupMsg("g-1") + msg.Sender = adapter.Actor{Role: adapter.RoleOwner} + if got := IdentifyContext(msg); got != restrict.ActorGroupMember { + t.Errorf("IdentifyContext() = %q, want group_member for owner in group", got) + } +} + +func TestSessionEnv(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + session Session + want restrict.ActorEnv + }{ + {name: "owner private", session: Session{Owner: true}, want: restrict.ActorOwner}, + {name: "owner group", session: Session{Owner: true, IsGroup: true}, want: restrict.ActorGroupMember}, + {name: "group member", session: Session{IsGroup: true}, want: restrict.ActorGroupMember}, + {name: "stranger", session: Session{}, want: restrict.ActorStranger}, + } + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + if got := tt.session.Env(); got != tt.want { + t.Errorf("Session.Env() = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/internal/channel/session/manager.go b/internal/channel/session/manager.go new file mode 100644 index 0000000..e8fe7ad --- /dev/null +++ b/internal/channel/session/manager.go @@ -0,0 +1,155 @@ +package session + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// DDL for the channel↔session mapping. The composite primary key over +// (platform, channel_id) is the unique key the design requires: each real-world +// channel maps to exactly one internal session_id. created_at/updated_at follow +// the project's timestamp convention; actor_role records the identity lane so the +// mapping survives restarts. +const createTableSQL = ` +CREATE TABLE IF NOT EXISTS channel_sessions ( + platform TEXT NOT NULL, + channel_id TEXT NOT NULL, + session_id TEXT NOT NULL, + actor_role TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + PRIMARY KEY (platform, channel_id) +)` + +const createIndexSQL = ` +CREATE INDEX IF NOT EXISTS idx_channel_sessions_actor_role ON channel_sessions(actor_role)` + +const insertOrIgnoreSessionSQL = ` +INSERT OR IGNORE INTO channel_sessions + (platform, channel_id, session_id, actor_role, created_at, updated_at) +VALUES (?, ?, ?, ?, ?, ?)` + +const selectSessionSQL = ` +SELECT session_id FROM channel_sessions WHERE platform = ? AND channel_id = ?` + +// SessionManager owns the channel↔session mapping. It backs the map with the +// SQLite table channel_sessions so a channel keeps its internal session across +// restarts, and it caches lookups in memory to avoid a query per message. The +// cache is guarded by an RWMutex for concurrent adapter use. +type SessionManager struct { + mu sync.RWMutex + db *sql.DB + log *slog.Logger + cache map[string]Session +} + +// New builds a SessionManager, ensuring the channel_sessions table and its index +// exist. It performs no migration bookkeeping: the schema is created here so the +// module is self-contained. A nil db is rejected. ctx is used for DDL. +func New(ctx context.Context, db *sql.DB, log *slog.Logger) (*SessionManager, error) { + if db == nil { + return nil, ErrNilDB + } + log = logger.WithSource(log, "channel/session") + if _, err := db.ExecContext(ctx, createTableSQL); err != nil { + return nil, fmt.Errorf("session: ensure channel_sessions: %w", err) + } + if _, err := db.ExecContext(ctx, createIndexSQL); err != nil { + return nil, fmt.Errorf("session: ensure channel_sessions index: %w", err) + } + return &SessionManager{db: db, log: log, cache: make(map[string]Session)}, nil +} + +// GetOrCreate resolves the actor for msg and returns the Session bound to the +// (platform, channel) the message represents, creating the mapping on first +// sight. The owner whitelist is applied here (ResolveActor), so the owner flag +// is set structurally rather than from prompt text. Concurrent calls for the +// same channel return the same session. +func (m *SessionManager) GetOrCreate(ctx context.Context, msg adapter.UnifiedMessage, owner *user.OwnerProfile) (Session, error) { + actor := ResolveActor(msg, owner) + cid := channelID(msg) + if cid == "" { + return Session{}, fmt.Errorf("session: get-or-create %s: %w", msg.Platform, ErrInvalidChannel) + } + key := msg.Platform + "\x00" + cid + + if s, ok := m.lookupCached(key); ok { + return s, nil + } + + m.mu.Lock() + defer m.mu.Unlock() + if s, ok := m.cache[key]; ok { + return s, nil + } + + sessionID, err := m.persist(ctx, msg.Platform, cid, actor) + if err != nil { + return Session{}, err + } + + s := Session{ + Platform: msg.Platform, + ChannelID: cid, + SessionID: sessionID, + ActorRole: actor.Role, + Owner: actor.IsOwner(), + IsGroup: actor.IsGroupChat(), + MemWritable: actor.IsOwner() && !actor.IsGroupChat(), + } + m.cache[key] = s + return s, nil +} + +// lookupCached returns the cached session for key under a read lock. +func (m *SessionManager) lookupCached(key string) (Session, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + s, ok := m.cache[key] + return s, ok +} + +// persist writes the channel mapping if it does not exist yet and reads back the +// canonical session_id. INSERT OR IGNORE makes it safe under concurrency: the +// winning writer's session_id becomes the binding. It is called under the write +// lock so the in-memory cache and the table stay in agreement. +func (m *SessionManager) persist(ctx context.Context, platform, cid string, actor adapter.Actor) (string, error) { + now := time.Now().UTC().Format(time.RFC3339Nano) + if _, err := m.db.ExecContext(ctx, insertOrIgnoreSessionSQL, platform, cid, newSessionID(), string(actor.Role), now, now); err != nil { + return "", fmt.Errorf("session: persist %s/%s: %w", platform, cid, err) + } + var sessionID string + if err := m.db.QueryRowContext(ctx, selectSessionSQL, platform, cid).Scan(&sessionID); err != nil { + return "", fmt.Errorf("session: read back %s/%s: %w", platform, cid, err) + } + if m.log != nil { + m.log.Debug("channel session bound", + "platform", platform, "channel_id", cid, "session_id", sessionID, "actor_role", actor.Role) + } + return sessionID, nil +} + +// channelID derives the stable channel key used for session mapping. Group +// messages key on the group (prefixed so a group never collides with a direct +// channel of the same id); direct messages key on their channel id, falling back +// to the sender id when no explicit channel id is present. +func channelID(msg adapter.UnifiedMessage) string { + if msg.IsGroup() { + if msg.GroupID != "" { + return "group:" + msg.GroupID + } + return msg.ChannelID + } + if msg.ChannelID != "" { + return msg.ChannelID + } + return msg.SenderID +} diff --git a/internal/channel/session/manager_test.go b/internal/channel/session/manager_test.go new file mode 100644 index 0000000..da564b1 --- /dev/null +++ b/internal/channel/session/manager_test.go @@ -0,0 +1,232 @@ +package session + +import ( + "context" + "database/sql" + "errors" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" +) + +// newTestManager opens a private in-memory SQLite handle and builds a manager +// over it. Each caller gets an isolated, non-shared database so parallel tests +// never contend. +func newTestManager(t *testing.T) (*SessionManager, *sql.DB) { + t.Helper() + ctx := context.Background() + handle, err := db.Open(ctx, ":memory:", nil) + if err != nil { + t.Fatalf("db.Open(:memory:) = %v", err) + } + t.Cleanup(func() { _ = handle.Close() }) + + m, err := New(ctx, handle, nil) + if err != nil { + t.Fatalf("New() = %v", err) + } + return m, handle +} + +func TestGetOrCreateCreatesAndReuses(t *testing.T) { + t.Parallel() + + m, _ := newTestManager(t) + ctx := context.Background() + msg := directMsg("o-100") + + first, err := m.GetOrCreate(ctx, msg, testOwner()) + if err != nil { + t.Fatalf("GetOrCreate() = %v", err) + } + if first.SessionID == "" { + t.Fatal("first SessionID empty") + } + if !first.Owner || !first.MemWritable { + t.Fatalf("owner session Owner=%v MemWritable=%v, want true/true", first.Owner, first.MemWritable) + } + + second, err := m.GetOrCreate(ctx, msg, testOwner()) + if err != nil { + t.Fatalf("second GetOrCreate() = %v", err) + } + if second.SessionID != first.SessionID { + t.Errorf("SessionID changed across calls: %q != %q", first.SessionID, second.SessionID) + } + if second.ChannelID != first.ChannelID { + t.Errorf("ChannelID changed: %q != %q", first.ChannelID, second.ChannelID) + } +} + +func TestGetOrCreateDifferentChannelsDiffer(t *testing.T) { + t.Parallel() + + m, _ := newTestManager(t) + ctx := context.Background() + + s1, err := m.GetOrCreate(ctx, directMsg("o-100"), testOwner()) + if err != nil { + t.Fatalf("GetOrCreate(sender1) = %v", err) + } + s2, err := m.GetOrCreate(ctx, directMsg("o-100"), testOwner()) + if err != nil { + t.Fatalf("GetOrCreate(sender2) = %v", err) + } + // Same real account maps to the same session, so the channel id must be the + // same; a different account must differ. + s3, err := m.GetOrCreate(ctx, directMsg("stranger-1"), testOwner()) + if err != nil { + t.Fatalf("GetOrCreate(stranger) = %v", err) + } + if s1.SessionID != s2.SessionID { + t.Error("same account produced different sessions") + } + if s3.SessionID == s1.SessionID { + t.Error("different account collided onto the same session") + } +} + +func TestGetOrCreateGroupIsolation(t *testing.T) { + t.Parallel() + + m, _ := newTestManager(t) + ctx := context.Background() + + // Owner speaking in a group: isolated context, memory read-only. + sess, err := m.GetOrCreate(ctx, groupMsg("o-100"), testOwner()) + if err != nil { + t.Fatalf("GetOrCreate(owner group) = %v", err) + } + if !sess.IsGroup { + t.Error("group message not flagged IsGroup") + } + if !sess.Owner { + t.Error("owner in group should still be Owner") + } + if sess.MemWritable { + t.Error("group session must be read-only (MemWritable=false) even for owner") + } + if got := sess.Env(); got != restrict.ActorGroupMember { + t.Errorf("Env() = %q, want group_member", got) + } +} + +func TestGetOrCreateStrangerIsolation(t *testing.T) { + t.Parallel() + + m, _ := newTestManager(t) + ctx := context.Background() + + sess, err := m.GetOrCreate(ctx, directMsg("stranger-1"), testOwner()) + if err != nil { + t.Fatalf("GetOrCreate(stranger) = %v", err) + } + if sess.Owner { + t.Error("stranger marked Owner") + } + if sess.MemWritable { + t.Error("stranger session must be read-only (MemWritable=false)") + } + if sess.IsGroup { + t.Error("stranger direct message marked IsGroup") + } +} + +func TestGetOrCreatePersistenceAcrossManagers(t *testing.T) { + t.Parallel() + + ctx := context.Background() + handle, err := db.Open(ctx, ":memory:", nil) + if err != nil { + t.Fatalf("db.Open = %v", err) + } + t.Cleanup(func() { _ = handle.Close() }) + + m1, err := New(ctx, handle, nil) + if err != nil { + t.Fatalf("New#1 = %v", err) + } + first, err := m1.GetOrCreate(ctx, directMsg("o-100"), testOwner()) + if err != nil { + t.Fatalf("m1.GetOrCreate = %v", err) + } + + // A brand-new manager over the same database must recover the same binding. + m2, err := New(ctx, handle, nil) + if err != nil { + t.Fatalf("New#2 = %v", err) + } + second, err := m2.GetOrCreate(ctx, directMsg("o-100"), testOwner()) + if err != nil { + t.Fatalf("m2.GetOrCreate = %v", err) + } + if second.SessionID != first.SessionID { + t.Errorf("session binding lost across managers: %q != %q", first.SessionID, second.SessionID) + } +} + +func TestGetOrCreateConcurrent(t *testing.T) { + t.Parallel() + + m, _ := newTestManager(t) + ctx := context.Background() + msg := directMsg("o-100") + + const n = 20 + ids := make([]string, n) + errs := make([]error, n) + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + i := i + go func() { + defer wg.Done() + sess, err := m.GetOrCreate(ctx, msg, testOwner()) + if err != nil { + errs[i] = err + return + } + ids[i] = sess.SessionID + }() + } + wg.Wait() + + for i := 0; i < n; i++ { + if errs[i] != nil { + t.Fatalf("goroutine %d: GetOrCreate = %v", i, errs[i]) + } + if ids[i] == "" { + t.Fatalf("goroutine %d produced empty session id", i) + } + if ids[i] != ids[0] { + t.Errorf("goroutine %d session id %q != %q", i, ids[i], ids[0]) + } + } +} + +func TestGetOrCreateInvalidChannel(t *testing.T) { + t.Parallel() + + m, _ := newTestManager(t) + ctx := context.Background() + + msg := adapter.UnifiedMessage{ + Platform: "qq", + ChannelType: adapter.ChannelDirect, + } + _, err := m.GetOrCreate(ctx, msg, testOwner()) + if !errors.Is(err, ErrInvalidChannel) { + t.Fatalf("GetOrCreate(invalid) error = %v, want ErrInvalidChannel", err) + } +} + +func TestNewRejectsNilDB(t *testing.T) { + t.Parallel() + + if _, err := New(context.Background(), nil, nil); !errors.Is(err, ErrNilDB) { + t.Fatalf("New(nil db) error = %v, want ErrNilDB", err) + } +} diff --git a/internal/channel/session/proactive.go b/internal/channel/session/proactive.go new file mode 100644 index 0000000..6d3dfef --- /dev/null +++ b/internal/channel/session/proactive.go @@ -0,0 +1,169 @@ +package session + +import ( + "context" + "log/slog" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +// Proactive push per-platform defaults (design/08 §3): weixin-ilink allows push +// only within 24h of the last inbound message; OneBot allows push but must be +// rate-limited to avoid wind-control flags; telegram/webhook push freely. +const ( + // iLinkWindow is the weixin-ilink proactive push window. + iLinkWindow = 24 * time.Hour + // oneBotBurstLimit is the maximum OneBot pushes within the window. + oneBotBurstLimit = 3 + // oneBotBurstWindow is the OneBot rate-limit window. + oneBotBurstWindow = 10 * time.Second +) + +// ProactiveOptions configures a ProactivePolicy. +type ProactiveOptions struct { + // Window is the inactivity window after which a push is refused. Zero + // disables the window check (unrestricted channels). + Window time.Duration + // RateLimit is the maximum push count within RateWindow. Zero disables the + // window check and rate limiting together. + RateLimit int + // RateWindow is the sliding window over which RateLimit is enforced. Zero + // disables rate limiting. + RateWindow time.Duration +} + +// ProactivePolicy decides whether a proactive push to a channel is allowed. It +// encodes the per-platform rules from design/08 §3: a capability gate (the +// adapter must declare Capabilities.Proactive), an optional inactivity window +// (weixin-ilink), and an optional burst rate limit (OneBot). The policy is +// immutable after construction and safe for concurrent use. +type ProactivePolicy struct { + log *slog.Logger + window time.Duration + limiter *RateLimiter +} + +// NewProactivePolicy builds a policy from opts. A zero Window and a zero +// RateLimit yield an unrestricted policy (telegram/webhook). WithLogger can be +// chained to tag denial records with the channel/session source. +func NewProactivePolicy(opts ProactiveOptions) *ProactivePolicy { + p := &ProactivePolicy{window: opts.Window} + if opts.RateLimit > 0 && opts.RateWindow > 0 { + p.limiter = NewRateLimiter(opts.RateLimit, opts.RateWindow) + } + return p +} + +// ProactivePolicyForPlatform returns the default proactive policy for a known +// platform identifier, following design/08 §3. Unknown platforms get an +// unrestricted policy, which is the conservative-outcome of "strong" channels. +func ProactivePolicyForPlatform(platform string) *ProactivePolicy { + switch platform { + case "weixin-ilink": + return NewProactivePolicy(ProactiveOptions{Window: iLinkWindow}) + case "qq-obv11", "qq-onebot", "onebot": + return NewProactivePolicy(ProactiveOptions{RateLimit: oneBotBurstLimit, RateWindow: oneBotBurstWindow}) + default: + return NewProactivePolicy(ProactiveOptions{}) + } +} + +// WithLogger returns a copy of p that logs denials through l with the +// channel/session source tag. It does not mutate p. +func (p *ProactivePolicy) WithLogger(l *slog.Logger) *ProactivePolicy { + if p == nil || l == nil { + return p + } + cp := *p + cp.log = logger.WithSource(l, "channel/session") + return &cp +} + +// Allow reports whether a proactive push may be sent now. It fails closed: a nil +// policy, a canceled ctx, or an adapter that does not declare proactive support +// all deny. Then it applies the configured inactivity window and burst limit. +func (p *ProactivePolicy) Allow(ctx context.Context, caps adapter.Capabilities, lastMessageAt time.Time) bool { + if p == nil { + return false + } + if err := ctx.Err(); err != nil { + p.logDeny("context canceled", err.Error()) + return false + } + if !caps.Proactive { + p.logDeny("capability off", "capture.proactive=false") + return false + } + if p.window > 0 { + if lastMessageAt.IsZero() || time.Since(lastMessageAt) > p.window { + p.logDeny("outside window", "last_message_at="+lastMessageAt.UTC().Format(time.RFC3339)) + return false + } + } + if p.limiter != nil && !p.limiter.Allow() { + p.logDeny("rate limited", "") + return false + } + return true +} + +func (p *ProactivePolicy) logDeny(reason, detail string) { + if p == nil || p.log == nil { + return + } + args := []any{"reason", reason} + if detail != "" { + args = append(args, "detail", detail) + } + p.log.Warn("proactive push denied by policy", args...) +} + +// RateLimiter is a concurrency-safe sliding-window rate limiter. It keeps event +// timestamps in memory and reports whether a new event may proceed. It is +// independent of any bus: the sliding window is a purely in-process counter that +// is safe to share across goroutines. +type RateLimiter struct { + mu sync.Mutex + limit int + window time.Duration + events []time.Time +} + +// NewRateLimiter builds a limiter that allows at most limit events per window. +// A limit <= 0 or window <= 0 yields an unlimited (always-allow) limiter, which +// is the safe "no rule configured" default. +func NewRateLimiter(limit int, window time.Duration) *RateLimiter { + return &RateLimiter{limit: limit, window: window} +} + +// Allow records a new event if the sliding window has capacity, returning true +// when the event may proceed, and false when it would exceed the limit. Events +// older than window are dropped before counting. +func (l *RateLimiter) Allow() bool { + if l == nil || l.limit <= 0 || l.window <= 0 { + return true + } + now := time.Now() + cutoff := now.Add(-l.window) + + l.mu.Lock() + defer l.mu.Unlock() + + keep := 0 + for _, t := range l.events { + if t.After(cutoff) { + l.events[keep] = t + keep++ + } + } + l.events = l.events[:keep] + + if len(l.events) >= l.limit { + return false + } + l.events = append(l.events, now) + return true +} diff --git a/internal/channel/session/proactive_test.go b/internal/channel/session/proactive_test.go new file mode 100644 index 0000000..bc790a9 --- /dev/null +++ b/internal/channel/session/proactive_test.go @@ -0,0 +1,171 @@ +package session + +import ( + "context" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func capsProactive() adapter.Capabilities { + return adapter.Capabilities{Proactive: true} +} + +func capsNoProactive() adapter.Capabilities { + return adapter.Capabilities{Proactive: false} +} + +func TestProactivePolicyCapabilityGate(t *testing.T) { + t.Parallel() + + p := NewProactivePolicy(ProactiveOptions{}) + // A channel that does not declare proactive support is denied regardless of + // how fresh the last message is. + if p.Allow(context.Background(), capsNoProactive(), time.Now()) { + t.Error("Allow() = true with Proactive=false, want false") + } +} + +func TestProactivePolicyWindowWithin(t *testing.T) { + t.Parallel() + + p := NewProactivePolicy(ProactiveOptions{Window: time.Hour}) + last := time.Now().Add(-30 * time.Minute) + if !p.Allow(context.Background(), capsProactive(), last) { + t.Error("Allow() = false within window, want true") + } +} + +func TestProactivePolicyWindowBeyond(t *testing.T) { + t.Parallel() + + p := NewProactivePolicy(ProactiveOptions{Window: time.Hour}) + last := time.Now().Add(-90 * time.Minute) + if p.Allow(context.Background(), capsProactive(), last) { + t.Error("Allow() = true beyond window, want false") + } +} + +func TestProactivePolicyWindowZeroLastMessage(t *testing.T) { + t.Parallel() + + p := NewProactivePolicy(ProactiveOptions{Window: time.Hour}) + if p.Allow(context.Background(), capsProactive(), time.Time{}) { + t.Error("Allow() = true with no last message, want false") + } +} + +func TestProactivePolicyRateLimit(t *testing.T) { + t.Parallel() + + p := NewProactivePolicy(ProactiveOptions{RateLimit: 3, RateWindow: 10 * time.Second}) + ctx := context.Background() + for i := 1; i <= 3; i++ { + if !p.Allow(ctx, capsProactive(), time.Now()) { + t.Fatalf("Allow() denied on attempt %d, want allowed under limit", i) + } + } + if p.Allow(ctx, capsProactive(), time.Now()) { + t.Error("Allow() allowed past burst limit, want denied") + } +} + +func TestProactivePolicyUnrestrictedChannel(t *testing.T) { + t.Parallel() + + // telegram/webhook-like channel: proactive capability on, no window, no + // rate limit → always allowed. + p := NewProactivePolicy(ProactiveOptions{}) + old := time.Now().Add(-48 * time.Hour) + for i := 0; i < 5; i++ { + if !p.Allow(context.Background(), capsProactive(), old) { + t.Fatalf("Allow() denied on unrestricted channel, want true") + } + } +} + +func TestProactivePolicyForPlatform(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + iLink := ProactivePolicyForPlatform("weixin-ilink") + now := time.Now() + if !iLink.Allow(ctx, capsProactive(), now.Add(-23*time.Hour)) { + t.Error("weixin-ilink allowed only within 24h; 23h old should allow") + } + if iLink.Allow(ctx, capsProactive(), now.Add(-25*time.Hour)) { + t.Error("weixin-ilink 25h old last message should deny") + } + + oneBot := ProactivePolicyForPlatform("qq-obv11") + for i := 0; i < oneBotBurstLimit; i++ { + if !oneBot.Allow(ctx, capsProactive(), now) { + t.Fatalf("onebot denied on attempt %d, want allowed", i+1) + } + } + if oneBot.Allow(ctx, capsProactive(), now) { + t.Error("onebot allowed past burst limit, want denied") + } + + tg := ProactivePolicyForPlatform("telegram") + if !tg.Allow(ctx, capsProactive(), now.Add(-72*time.Hour)) { + t.Error("telegram must push freely, want true") + } +} + +func TestProactivePolicyNilPolicyAndCanceledCtx(t *testing.T) { + t.Parallel() + + var p *ProactivePolicy + if p.Allow(context.Background(), capsProactive(), time.Now()) { + t.Error("nil policy Allow() = true, want false (fail closed)") + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + policy := NewProactivePolicy(ProactiveOptions{}) + if policy.Allow(ctx, capsProactive(), time.Now()) { + t.Error("Allow() with canceled ctx = true, want false") + } +} + +func TestRateLimiterSlidingWindow(t *testing.T) { + t.Parallel() + + l := NewRateLimiter(2, 50*time.Millisecond) + if !l.Allow() { + t.Fatal("first Allow() = false, want true") + } + if !l.Allow() { + t.Fatal("second Allow() = false, want true") + } + if l.Allow() { + t.Fatal("third Allow() = true, want false (limit reached)") + } + + time.Sleep(60 * time.Millisecond) + if !l.Allow() { + t.Fatal("Allow() after window elapsed = false, want true") + } +} + +func TestRateLimiterDisabledConfig(t *testing.T) { + t.Parallel() + + // A limiter with no limit/window configured is the safe permissive default. + configs := []*RateLimiter{ + NewRateLimiter(0, 0), + NewRateLimiter(-1, time.Second), + NewRateLimiter(0, time.Second), + } + for i, l := range configs { + if !l.Allow() || !l.Allow() { + t.Errorf("config %d: Allow() = false, want true (unlimited)", i) + } + } + if !(*RateLimiter)(nil).Allow() { + t.Error("nil rate limiter Allow() = false, want true (permissive)") + } +} diff --git a/internal/channel/session/session.go b/internal/channel/session/session.go new file mode 100644 index 0000000..88cfc04 --- /dev/null +++ b/internal/channel/session/session.go @@ -0,0 +1,80 @@ +// Package session maps IM channels to internal sessions and resolves the +// strongly typed actor identity that gates every downstream permission decision. +// +// It is the M6 home of three cross-cutting concerns wired together at the +// channel layer, per DevDocs/design/11 §0/§2: +// +// - Channel↔session mapping: a persistent (platform, channel_id) → session_id +// map (SQLite table channel_sessions), so the same conversation always maps +// to the same internal session across restarts. +// - Owner recognition: a structural, never-prompt rule. ResolveActor decides +// owner vs group_member vs stranger purely from channel-level actor metadata +// matched against the user.OwnerProfile account whitelist. Nothing here reads +// message text to infer identity — that is a project red line. +// - Proactive push policy: per-platform rules from design/08 §3 (weixin-ilink +// 24h window, OneBot burst rate limit, telegram/webhook unrestricted). +// +// The package is concurrency-safe: the manager guards its cache with an +// RWMutex and the rate limiter uses its own mutex, so multiple adapters may +// resolve sessions and issue pushes concurrently. +package session + +import ( + "crypto/rand" + "encoding/hex" + "errors" + "fmt" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// Package-level sentinel errors. Each is wrapped with %w at the call site so +// callers can use errors.Is. +var ( + // ErrNilDB is returned when SessionManager is built with a nil database. + ErrNilDB = errors.New("session: nil database handle") + // ErrInvalidChannel is returned when a message has no usable channel id. + ErrInvalidChannel = errors.New("session: empty channel id") +) + +// sessionIDPrefix distinguishes generated session ids from foreign identifiers. +const sessionIDPrefix = "sess-" + +// Session is the resolved channel↔session binding for one inbound message. It +// carries the identity metadata (ActorRole, Owner, IsGroup) that the policy +// engine and memory layer read, plus the isolation flag MemWritable which says +// whether this conversation may write to shared memory. +// +// The memory rule follows design/02 §3 and design/11 §2: only the owner's +// private conversation may read/write memory shared with the Web main session; +// group chats and stranger conversations may read memory read-only and may +// never persist anything that could leak owner-sensitive information. +type Session struct { + // Platform is the normalized platform identifier (e.g. "weixin-ilink"). + Platform string + // ChannelID is the stable channel key used for session mapping. + ChannelID string + // SessionID is the internal session binding the channel maps to. + SessionID string + // ActorRole is the strongly typed role of the resolved actor. + ActorRole adapter.Role + // Owner reports whether the actor is the bound owner account. + Owner bool + // IsGroup reports whether this is a group/room conversation. + IsGroup bool + // MemWritable reports whether this conversation may write shared memory. + // It is true only for the owner's private conversation. + MemWritable bool +} + +// newSessionID returns a random, collision-resistant session identifier. +func newSessionID() string { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + // rand.Read should never fail; fall back to a nanosecond-named id so a + // session can still be created rather than failing the whole call. + return fmt.Sprintf("%s%d", sessionIDPrefix, time.Now().UnixNano()) + } + return sessionIDPrefix + hex.EncodeToString(b[:]) +}