diff --git a/internal/gateway/methods/chat.go b/internal/gateway/methods/chat.go index 7c0bab22..3b43c974 100644 --- a/internal/gateway/methods/chat.go +++ b/internal/gateway/methods/chat.go @@ -119,7 +119,7 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re runID := uuid.NewString() sessionKey := params.SessionKey if sessionKey == "" { - sessionKey = sessions.SessionKey(params.AgentID, "ws-"+client.ID()) + sessionKey = sessions.BuildWSSessionKey(params.AgentID, uuid.NewString()) } // Detach from HTTP request context so agent runs survive page navigation/reconnect. @@ -190,7 +190,7 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re Message: message, Media: mediaFiles, Channel: "ws", - ChatID: client.ID(), + ChatID: userID, // use stable userID for team/workspace isolation (not ephemeral client.ID()) RunID: runID, UserID: userID, Stream: params.Stream, @@ -237,7 +237,7 @@ func (m *ChatMethods) handleHistory(ctx context.Context, client *gateway.Client, sessionKey := params.SessionKey if sessionKey == "" { - sessionKey = sessions.SessionKey(params.AgentID, "ws-"+client.ID()) + sessionKey = sessions.BuildWSSessionKey(params.AgentID, uuid.NewString()) } history := m.sessions.GetHistory(sessionKey) diff --git a/internal/gateway/methods/sessions.go b/internal/gateway/methods/sessions.go index 3f5f3b96..11f6290b 100644 --- a/internal/gateway/methods/sessions.go +++ b/internal/gateway/methods/sessions.go @@ -31,6 +31,7 @@ func (m *SessionsMethods) Register(router *gateway.MethodRouter) { type sessionsListParams struct { AgentID string `json:"agentId"` + Channel string `json:"channel"` // optional: filter by channel prefix ("ws", "telegram") Limit int `json:"limit"` Offset int `json:"offset"` } @@ -45,11 +46,19 @@ func (m *SessionsMethods) handleList(_ context.Context, client *gateway.Client, params.Limit = 20 } - result := m.sessions.ListPagedRich(store.SessionListOpts{ + opts := store.SessionListOpts{ AgentID: params.AgentID, + Channel: params.Channel, Limit: params.Limit, Offset: params.Offset, - }) + } + // Only filter by UserID when a channel filter is specified (e.g. chat sidebar sends channel="ws"). + // Sessions admin page omits channel → sees all sessions unfiltered. + if params.Channel != "" { + opts.UserID = client.UserID() + } + + result := m.sessions.ListPagedRich(opts) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "sessions": result.Sessions, "total": result.Total, diff --git a/internal/sessions/key.go b/internal/sessions/key.go index ccd17833..0b19952f 100644 --- a/internal/sessions/key.go +++ b/internal/sessions/key.go @@ -183,6 +183,19 @@ func IsHeartbeatSession(key string) bool { return strings.HasPrefix(rest, "heartbeat") } +// BuildWSSessionKey builds the canonical WS session key for a web conversation. +// +// agent:{agentId}:ws:direct:{conversationId} +func BuildWSSessionKey(agentID, conversationID string) string { + return BuildSessionKey(agentID, "ws", PeerDirect, conversationID) +} + +// IsWSSession checks if a session key is a WS session (new or legacy format). +func IsWSSession(key string) bool { + _, rest := ParseSessionKey(key) + return strings.HasPrefix(rest, "ws:") || strings.HasPrefix(rest, "ws-") +} + // PeerKindFromGroup returns PeerGroup if isGroup is true, PeerDirect otherwise. func PeerKindFromGroup(isGroup bool) PeerKind { if isGroup { diff --git a/internal/store/pg/sessions_list.go b/internal/store/pg/sessions_list.go index 524a05da..05e63005 100644 --- a/internal/store/pg/sessions_list.go +++ b/internal/store/pg/sessions_list.go @@ -3,6 +3,7 @@ package pg import ( "database/sql" "encoding/json" + "fmt" "strings" "time" @@ -12,6 +13,41 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/store" ) +// buildSessionFilter builds a dynamic WHERE clause from SessionListOpts. +// Returns the WHERE string (with leading " WHERE ") and the positional args. +// The tableAlias is prepended to column names (e.g. "s" → "s.session_key"). +func buildSessionFilter(opts store.SessionListOpts, tableAlias string) (string, []any) { + prefix := "" + if tableAlias != "" { + prefix = tableAlias + "." + } + var conditions []string + var args []any + idx := 1 + + if opts.AgentID != "" { + conditions = append(conditions, fmt.Sprintf("%ssession_key LIKE $%d", prefix, idx)) + args = append(args, "agent:"+opts.AgentID+":%") + idx++ + } + if opts.Channel != "" { + // Match canonical format: agent:X:{channel}:... + conditions = append(conditions, fmt.Sprintf("%ssession_key LIKE $%d", prefix, idx)) + args = append(args, "agent:%:"+opts.Channel+":%") + idx++ + } + if opts.UserID != "" { + conditions = append(conditions, fmt.Sprintf("%suser_id = $%d", prefix, idx)) + args = append(args, opts.UserID) + idx++ + } + + if len(conditions) == 0 { + return "", nil + } + return " WHERE " + strings.Join(conditions, " AND "), args +} + func (s *PGSessionStore) List(agentID string) []store.SessionInfo { var rows *sql.Rows var err error @@ -65,13 +101,7 @@ func (s *PGSessionStore) ListPaged(opts store.SessionListOpts) store.SessionList } offset := max(opts.Offset, 0) - var where string - var whereArgs []any - - if opts.AgentID != "" { - where = " WHERE session_key LIKE $1" - whereArgs = append(whereArgs, "agent:"+opts.AgentID+":%") - } + where, whereArgs := buildSessionFilter(opts, "") // Count total var total int @@ -81,18 +111,10 @@ func (s *PGSessionStore) ListPaged(opts store.SessionListOpts) store.SessionList } // Fetch page using jsonb_array_length to avoid loading full messages - var selectQ string - var selectArgs []any - - if opts.AgentID != "" { - selectQ = `SELECT session_key, jsonb_array_length(messages), created_at, updated_at, label, channel, user_id, COALESCE(metadata, '{}') - FROM sessions WHERE session_key LIKE $1 ORDER BY updated_at DESC LIMIT $2 OFFSET $3` - selectArgs = []any{whereArgs[0], limit, offset} - } else { - selectQ = `SELECT session_key, jsonb_array_length(messages), created_at, updated_at, label, channel, user_id, COALESCE(metadata, '{}') - FROM sessions ORDER BY updated_at DESC LIMIT $1 OFFSET $2` - selectArgs = []any{limit, offset} - } + nextIdx := len(whereArgs) + 1 + selectQ := fmt.Sprintf(`SELECT session_key, jsonb_array_length(messages), created_at, updated_at, label, channel, user_id, COALESCE(metadata, '{}') + FROM sessions%s ORDER BY updated_at DESC LIMIT $%d OFFSET $%d`, where, nextIdx, nextIdx+1) + selectArgs := append(append([]any{}, whereArgs...), limit, offset) rows, err := s.db.Query(selectQ, selectArgs...) if err != nil { @@ -139,13 +161,7 @@ func (s *PGSessionStore) ListPagedRich(opts store.SessionListOpts) store.Session } offset := max(opts.Offset, 0) - var where string - var whereArgs []any - - if opts.AgentID != "" { - where = " WHERE s.session_key LIKE $1" - whereArgs = append(whereArgs, "agent:"+opts.AgentID+":%") - } + where, whereArgs := buildSessionFilter(opts, "s") // Count total var total int @@ -155,9 +171,6 @@ func (s *PGSessionStore) ListPagedRich(opts store.SessionListOpts) store.Session } // Fetch page with agent name via LEFT JOIN - var selectQ string - var selectArgs []any - const richCols = `s.session_key, jsonb_array_length(s.messages), s.created_at, s.updated_at, s.label, s.channel, s.user_id, COALESCE(s.metadata, '{}'), s.model, s.provider, s.input_tokens, s.output_tokens, @@ -166,17 +179,11 @@ func (s *PGSessionStore) ListPagedRich(opts store.SessionListOpts) store.Session COALESCE(a.context_window, 200000), -- config.DefaultContextWindow s.compaction_count` - if opts.AgentID != "" { - selectQ = `SELECT ` + richCols + ` - FROM sessions s LEFT JOIN agents a ON s.agent_id = a.id - WHERE s.session_key LIKE $1 ORDER BY s.updated_at DESC LIMIT $2 OFFSET $3` - selectArgs = []any{whereArgs[0], limit, offset} - } else { - selectQ = `SELECT ` + richCols + ` - FROM sessions s LEFT JOIN agents a ON s.agent_id = a.id - ORDER BY s.updated_at DESC LIMIT $1 OFFSET $2` - selectArgs = []any{limit, offset} - } + nextIdx := len(whereArgs) + 1 + selectQ := fmt.Sprintf(`SELECT %s + FROM sessions s LEFT JOIN agents a ON s.agent_id = a.id + %s ORDER BY s.updated_at DESC LIMIT $%d OFFSET $%d`, richCols, where, nextIdx, nextIdx+1) + selectArgs := append(append([]any{}, whereArgs...), limit, offset) rows, err := s.db.Query(selectQ, selectArgs...) if err != nil { diff --git a/internal/store/session_store.go b/internal/store/session_store.go index 9b4d08e1..1fcf5b6a 100644 --- a/internal/store/session_store.go +++ b/internal/store/session_store.go @@ -53,6 +53,8 @@ type SessionInfo struct { // SessionListOpts holds pagination options for ListPaged. type SessionListOpts struct { AgentID string + Channel string // optional: filter by channel prefix ("ws", "telegram", etc.) + UserID string // optional: filter by user_id Limit int Offset int }