Merge pull request '实现会话生命周期管理，让 WS handler 能追踪会话、存储对话历史。' (#35) from feature/phase2 into develop

Reviewed-on: http://8.161.227.145:3000/XEngineers/CamTalk/pulls/35
2026-06-13 15:34:04 +08:00
parent 688e13a491 fe4dedec71
commit 33effe161a
9 changed files with 1048 additions and 39 deletions
--- a/backend/cmd/server/main.go
+++ b/backend/cmd/server/main.go
@@ -12,6 +12,7 @@ import (

 	"github.com/hhs/camtalk/internal/config"
 	"github.com/hhs/camtalk/internal/logger"
+	"github.com/hhs/camtalk/internal/session"
 	"github.com/hhs/camtalk/internal/ws"
 )

@@ -33,6 +34,12 @@ func main() {
 		"addr", cfg.Server.Addr(),
 	)

+	// 初始化 Session Manager（MVP 默认内存实现）
+	var sessionMgr session.Manager
+	// TODO: 当 Redis 配置非空时切换为 RedisManager
+	sessionMgr = session.NewMemoryManager(30*time.Minute, 20)
+	defer sessionMgr.(*session.MemoryManager).Stop()
+
 	// Gin 模式
 	if cfg.App.Env == "prod" {
 		gin.SetMode(gin.ReleaseMode)
@@ -44,11 +51,11 @@ func main() {
 	// REST API
 	api := r.Group("/api")
 	{
-		api.GET("/health", healthHandler)
+		api.GET("/health", healthHandler(sessionMgr))
 	}

 	// WebSocket
-	r.GET("/ws", ws.ServeWS)
+	r.GET("/ws", ws.ServeWS(sessionMgr))

 	// HTTP Server
 	srv := &http.Server{
@@ -82,11 +89,13 @@ func main() {
 }

 // healthHandler 健康检查。
-func healthHandler(c *gin.Context) {
-	c.JSON(200, gin.H{
-		"status":          "ok",
-		"version":         "0.1.0",
-		"uptime":          time.Since(startTime).String(),
-		"active_sessions": 0, // TODO: 接入 Session Manager
-	})
+func healthHandler(sessionMgr session.Manager) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		c.JSON(200, gin.H{
+			"status":          "ok",
+			"version":         "0.1.0",
+			"uptime":          time.Since(startTime).String(),
+			"active_sessions": sessionMgr.ActiveCount(),
+		})
+	}
 }
--- a/backend/go.mod
+++ b/backend/go.mod
@@ -1,6 +1,6 @@
 module github.com/hhs/camtalk

-go 1.23.0
+go 1.24

 require (
 	github.com/gin-gonic/gin v1.10.0
@@ -13,6 +13,7 @@ require (
 require (
 	github.com/bytedance/sonic v1.11.6 // indirect
 	github.com/bytedance/sonic/loader v0.1.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.4 // indirect
 	github.com/cloudwego/iasm v0.2.0 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
@@ -24,12 +25,13 @@ require (
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
 	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
-	github.com/klauspost/cpuid/v2 v2.2.7 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.10 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
+	github.com/redis/go-redis/v9 v9.20.1 // indirect
 	github.com/sagikazarmark/locafero v0.11.0 // indirect
 	github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect
 	github.com/spf13/afero v1.15.0 // indirect
@@ -38,12 +40,13 @@ require (
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.12 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/multierr v1.10.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/arch v0.8.0 // indirect
 	golang.org/x/crypto v0.23.0 // indirect
 	golang.org/x/net v0.25.0 // indirect
-	golang.org/x/sys v0.29.0 // indirect
+	golang.org/x/sys v0.30.0 // indirect
 	golang.org/x/text v0.28.0 // indirect
 	google.golang.org/protobuf v1.34.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
--- a/backend/go.sum
+++ b/backend/go.sum
@@ -2,6 +2,8 @@ github.com/bytedance/sonic v1.11.6 h1:oUp34TzMlL+OY1OUWxHqsdkgC/Zfc85zGqw9siXjrc
 github.com/bytedance/sonic v1.11.6/go.mod h1:LysEHSvpvDySVdC2f87zGWf6CIKJcAvqab1ZaiQtds4=
 github.com/bytedance/sonic/loader v0.1.1 h1:c+e5Pt1k/cy5wMveRDyk2X4B9hF4g7an8N3zCYjJFNM=
 github.com/bytedance/sonic/loader v0.1.1/go.mod h1:ncP89zfokxS5LZrJxl5z0UJcsk4M4yY2JpfqGeCtNLU=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cloudwego/base64x v0.1.4 h1:jwCgWpFanWmN8xoIUHa2rtzmkd5J2plF/dnLS6Xd/0Y=
 github.com/cloudwego/base64x v0.1.4/go.mod h1:0zlkT4Wn5C6NdauXdJRhSKRlJvmclQ1hhJgA0rcu/8w=
 github.com/cloudwego/iasm v0.2.0 h1:1KNIy1I1H9hNNFEEH3DVnI4UujN+1zjpuk6gwHLTssg=
@@ -43,6 +45,8 @@ github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHm
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.7 h1:ZWSB3igEs+d0qvnxR/ZBzXVmxkgt8DdzP6m9pfuVLDM=
 github.com/klauspost/cpuid/v2 v2.2.7/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
+github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
+github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
 github.com/knz/go-libedit v1.10.1/go.mod h1:MZTVkCWyz0oBc7JOWP3wNAzd002ZbM/5hgShxwh4x8M=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
@@ -61,6 +65,8 @@ github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0
 github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/redis/go-redis/v9 v9.20.1 h1:sfCU6A8P3dXbKyWes02uxA2baehGux9dZHfEKtsTB1w=
+github.com/redis/go-redis/v9 v9.20.1/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
 github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
 github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc=
@@ -91,6 +97,8 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
 github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
+go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
+go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ=
@@ -110,6 +118,8 @@ golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
 golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng=
 golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU=
 google.golang.org/protobuf v1.34.1 h1:9ddQBjfCyZPOHPUiPxpYESBLc+T8P3E+Vo4IbKZgFWg=
--- a/backend/internal/models/models.go
+++ b/backend/internal/models/models.go
@@ -21,6 +21,26 @@ func DefaultConfig() SessionConfig {
 	return SessionConfig{TTSEnabled: true, DetailLevel: "low", Language: "zh-CN"}
 }

+// SessionConfigPatch 会话配置增量更新（指针字段表示"未传则不更新"）。
+type SessionConfigPatch struct {
+	TTSEnabled  *bool   `json:"tts_enabled,omitempty"`
+	DetailLevel *string `json:"detail_level,omitempty"`
+	Language    *string `json:"language,omitempty"`
+}
+
+// Apply 将 patch 中的非 nil 字段覆盖到 cfg。
+func (p SessionConfigPatch) Apply(cfg *SessionConfig) {
+	if p.TTSEnabled != nil {
+		cfg.TTSEnabled = *p.TTSEnabled
+	}
+	if p.DetailLevel != nil {
+		cfg.DetailLevel = *p.DetailLevel
+	}
+	if p.Language != nil {
+		cfg.Language = *p.Language
+	}
+}
+
 // Message 对话消息。
 type Message struct {
 	Role    string `json:"role"` // "user" | "assistant"
--- a/backend/internal/session/manager.go
+++ b/backend/internal/session/manager.go
@@ -0,0 +1,49 @@
+// Package session 提供会话生命周期管理能力。
+package session
+
+import (
+	"context"
+	"errors"
+
+	"github.com/hhs/camtalk/internal/models"
+)
+
+// ErrSessionNotFound 会话不存在或已过期。
+var ErrSessionNotFound = errors.New("session not found")
+
+// Manager 会话管理器接口。
+// WebSocket Handler 通过此接口操作会话，不直接接触存储层。
+type Manager interface {
+	// Create 创建新会话，返回 session ID。
+	Create(ctx context.Context, config models.SessionConfig) (string, error)
+
+	// Get 获取会话（含 config）。不存在返回 ErrSessionNotFound。
+	Get(ctx context.Context, sessionID string) (*models.Session, error)
+
+	// UpdateConfig 更新会话配置（config 消息触发）。
+	UpdateConfig(ctx context.Context, sessionID string, patch models.SessionConfigPatch) error
+
+	// GetHistory 获取最近 N 轮对话历史（供 Orchestrator 构建 LLM 上下文）。
+	GetHistory(ctx context.Context, sessionID string, limit int) ([]models.Message, error)
+
+	// AppendMessage 追加一条对话消息，同时刷新 TTL。
+	AppendMessage(ctx context.Context, sessionID string, msg models.Message) error
+
+	// SetActiveRequest 标记当前正在处理的请求 ID（interrupt 用）。
+	SetActiveRequest(ctx context.Context, sessionID string, requestID string) error
+
+	// GetActiveRequestID 获取当前活跃请求 ID。
+	GetActiveRequestID(ctx context.Context, sessionID string) (string, error)
+
+	// ClearActiveRequest 清除活跃请求标记（请求完成或中断后）。
+	ClearActiveRequest(ctx context.Context, sessionID string) error
+
+	// Touch 刷新 TTL（心跳时调用）。
+	Touch(ctx context.Context, sessionID string) error
+
+	// Destroy 显式销毁会话（REST API DELETE 或连接断开清理）。
+	Destroy(ctx context.Context, sessionID string) error
+
+	// ActiveCount 返回当前活跃会话数（健康检查用）。
+	ActiveCount() int
+}
--- a/backend/internal/session/memory.go
+++ b/backend/internal/session/memory.go
@@ -0,0 +1,275 @@
+package session
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/hhs/camtalk/internal/logger"
+	"github.com/hhs/camtalk/internal/models"
+)
+
+const (
+	defaultTTL         = 30 * time.Minute
+	defaultHistorySize = 20
+)
+
+// sessionEntry 内部会话条目。
+type sessionEntry struct {
+	session     models.Session
+	history     []models.Message
+	activeReqID string
+	lastActive  time.Time
+}
+
+// MemoryManager 基于内存的 SessionManager 实现。
+// 适用于 MVP 和无 Redis 的开发环境。
+type MemoryManager struct {
+	mu          sync.RWMutex
+	sessions    map[string]*sessionEntry
+	ttl         time.Duration
+	maxHistory  int
+	stopCleaner chan struct{}
+}
+
+// NewMemoryManager 创建内存版 SessionManager。
+// ttl 为会话过期时间，maxHistory 为对话历史上限（0 表示使用默认值 20）。
+func NewMemoryManager(ttl time.Duration, maxHistory int) *MemoryManager {
+	if ttl <= 0 {
+		ttl = defaultTTL
+	}
+	if maxHistory <= 0 {
+		maxHistory = defaultHistorySize
+	}
+
+	m := &MemoryManager{
+		sessions:    make(map[string]*sessionEntry),
+		ttl:         ttl,
+		maxHistory:  maxHistory,
+		stopCleaner: make(chan struct{}),
+	}
+
+	// 启动后台清理 goroutine，每分钟清除过期会话。
+	go m.cleanLoop()
+
+	return m
+}
+
+// cleanLoop 后台定期清理过期会话。
+func (m *MemoryManager) cleanLoop() {
+	ticker := time.NewTicker(1 * time.Minute)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			m.cleanExpired()
+		case <-m.stopCleaner:
+			return
+		}
+	}
+}
+
+// cleanExpired 清除所有过期会话。
+func (m *MemoryManager) cleanExpired() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	now := time.Now()
+	for id, entry := range m.sessions {
+		if now.Sub(entry.lastActive) > m.ttl {
+			delete(m.sessions, id)
+			logger.Log.Debugw("session expired (cleaner)", "session", id)
+		}
+	}
+}
+
+// Stop 停止后台清理 goroutine。应用退出前调用。
+func (m *MemoryManager) Stop() {
+	close(m.stopCleaner)
+}
+
+// isExpired 检查会话是否过期（调用方需持锁或在已知 entry 存在时调用）。
+func (m *MemoryManager) isExpired(entry *sessionEntry) bool {
+	return time.Since(entry.lastActive) > m.ttl
+}
+
+// Create 创建新会话。
+func (m *MemoryManager) Create(_ context.Context, config models.SessionConfig) (string, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	id := uuid.New().String()
+	now := time.Now()
+	m.sessions[id] = &sessionEntry{
+		session: models.Session{
+			ID:        id,
+			CreatedAt: now,
+			Config:    config,
+		},
+		history:    make([]models.Message, 0),
+		lastActive: now,
+	}
+
+	logger.Log.Debugw("session created", "session", id)
+	return id, nil
+}
+
+// Get 获取会话。
+func (m *MemoryManager) Get(_ context.Context, sessionID string) (*models.Session, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	entry, ok := m.sessions[sessionID]
+	if !ok || m.isExpired(entry) {
+		return nil, ErrSessionNotFound
+	}
+
+	sess := entry.session // 复制一份返回
+	return &sess, nil
+}
+
+// UpdateConfig 更新会话配置。
+func (m *MemoryManager) UpdateConfig(_ context.Context, sessionID string, patch models.SessionConfigPatch) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	entry, ok := m.sessions[sessionID]
+	if !ok || m.isExpired(entry) {
+		return ErrSessionNotFound
+	}
+
+	patch.Apply(&entry.session.Config)
+	entry.lastActive = time.Now()
+
+	logger.Log.Debugw("session config updated", "session", sessionID)
+	return nil
+}
+
+// GetHistory 获取最近 N 轮对话历史。
+func (m *MemoryManager) GetHistory(_ context.Context, sessionID string, limit int) ([]models.Message, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	entry, ok := m.sessions[sessionID]
+	if !ok || m.isExpired(entry) {
+		return nil, ErrSessionNotFound
+	}
+
+	if limit <= 0 || limit > len(entry.history) {
+		limit = len(entry.history)
+	}
+
+	// 返回最近 limit 条的副本
+	result := make([]models.Message, limit)
+	copy(result, entry.history[len(entry.history)-limit:])
+	return result, nil
+}
+
+// AppendMessage 追加一条对话消息，同时刷新 TTL。
+func (m *MemoryManager) AppendMessage(_ context.Context, sessionID string, msg models.Message) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	entry, ok := m.sessions[sessionID]
+	if !ok || m.isExpired(entry) {
+		return ErrSessionNotFound
+	}
+
+	entry.history = append(entry.history, msg)
+
+	// 超过上限时裁剪，保留最新的 maxHistory 条
+	if len(entry.history) > m.maxHistory {
+		entry.history = entry.history[len(entry.history)-m.maxHistory:]
+	}
+
+	entry.lastActive = time.Now()
+	return nil
+}
+
+// SetActiveRequest 标记当前正在处理的请求 ID。
+func (m *MemoryManager) SetActiveRequest(_ context.Context, sessionID string, requestID string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	entry, ok := m.sessions[sessionID]
+	if !ok || m.isExpired(entry) {
+		return ErrSessionNotFound
+	}
+
+	entry.activeReqID = requestID
+	entry.lastActive = time.Now()
+	return nil
+}
+
+// GetActiveRequestID 获取当前活跃请求 ID。
+func (m *MemoryManager) GetActiveRequestID(_ context.Context, sessionID string) (string, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	entry, ok := m.sessions[sessionID]
+	if !ok || m.isExpired(entry) {
+		return "", ErrSessionNotFound
+	}
+
+	return entry.activeReqID, nil
+}
+
+// ClearActiveRequest 清除活跃请求标记。
+func (m *MemoryManager) ClearActiveRequest(_ context.Context, sessionID string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	entry, ok := m.sessions[sessionID]
+	if !ok || m.isExpired(entry) {
+		return ErrSessionNotFound
+	}
+
+	entry.activeReqID = ""
+	entry.lastActive = time.Now()
+	return nil
+}
+
+// Touch 刷新 TTL。
+func (m *MemoryManager) Touch(_ context.Context, sessionID string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	entry, ok := m.sessions[sessionID]
+	if !ok || m.isExpired(entry) {
+		return ErrSessionNotFound
+	}
+
+	entry.lastActive = time.Now()
+	return nil
+}
+
+// Destroy 显式销毁会话。
+func (m *MemoryManager) Destroy(_ context.Context, sessionID string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if _, ok := m.sessions[sessionID]; !ok {
+		return ErrSessionNotFound
+	}
+
+	delete(m.sessions, sessionID)
+	logger.Log.Debugw("session destroyed", "session", sessionID)
+	return nil
+}
+
+// ActiveCount 返回当前活跃会话数。
+func (m *MemoryManager) ActiveCount() int {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	now := time.Now()
+	count := 0
+	for _, entry := range m.sessions {
+		if now.Sub(entry.lastActive) <= m.ttl {
+			count++
+		}
+	}
+	return count
+}
--- a/backend/internal/session/memory_test.go
+++ b/backend/internal/session/memory_test.go
@@ -0,0 +1,286 @@
+package session
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/hhs/camtalk/internal/logger"
+	"github.com/hhs/camtalk/internal/models"
+)
+
+func init() {
+	logger.Init("debug", "console")
+}
+
+func TestCreateAndGet(t *testing.T) {
+	m := NewMemoryManager(30*time.Minute, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	config := models.DefaultConfig()
+	id, err := m.Create(ctx, config)
+	if err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+	if id == "" {
+		t.Fatal("Create returned empty ID")
+	}
+
+	sess, err := m.Get(ctx, id)
+	if err != nil {
+		t.Fatalf("Get: %v", err)
+	}
+	if sess.ID != id {
+		t.Errorf("ID = %q, want %q", sess.ID, id)
+	}
+	if sess.Config.Language != "zh-CN" {
+		t.Errorf("Language = %q, want %q", sess.Config.Language, "zh-CN")
+	}
+}
+
+func TestGetNotFound(t *testing.T) {
+	m := NewMemoryManager(30*time.Minute, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	_, err := m.Get(ctx, "nonexistent")
+	if err != ErrSessionNotFound {
+		t.Errorf("Get nonexistent: err = %v, want ErrSessionNotFound", err)
+	}
+}
+
+func TestExpire(t *testing.T) {
+	// 使用极短 TTL 测试过期
+	m := NewMemoryManager(50*time.Millisecond, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	id, _ := m.Create(ctx, models.DefaultConfig())
+
+	// 未过期时应能获取
+	_, err := m.Get(ctx, id)
+	if err != nil {
+		t.Fatalf("Get before expire: %v", err)
+	}
+
+	// 等待过期
+	time.Sleep(80 * time.Millisecond)
+
+	_, err = m.Get(ctx, id)
+	if err != ErrSessionNotFound {
+		t.Errorf("Get after expire: err = %v, want ErrSessionNotFound", err)
+	}
+}
+
+func TestDestroy(t *testing.T) {
+	m := NewMemoryManager(30*time.Minute, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	id, _ := m.Create(ctx, models.DefaultConfig())
+
+	if err := m.Destroy(ctx, id); err != nil {
+		t.Fatalf("Destroy: %v", err)
+	}
+
+	_, err := m.Get(ctx, id)
+	if err != ErrSessionNotFound {
+		t.Errorf("Get after Destroy: err = %v, want ErrSessionNotFound", err)
+	}
+}
+
+func TestDestroyNotFound(t *testing.T) {
+	m := NewMemoryManager(30*time.Minute, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	err := m.Destroy(ctx, "nonexistent")
+	if err != ErrSessionNotFound {
+		t.Errorf("Destroy nonexistent: err = %v, want ErrSessionNotFound", err)
+	}
+}
+
+func TestAppendMessageAndGetHistory(t *testing.T) {
+	m := NewMemoryManager(30*time.Minute, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	id, _ := m.Create(ctx, models.DefaultConfig())
+
+	msgs := []models.Message{
+		{Role: "user", Content: "你好"},
+		{Role: "assistant", Content: "你好！有什么可以帮你的吗？"},
+		{Role: "user", Content: "这是什么？"},
+		{Role: "assistant", Content: "这是一朵花。"},
+	}
+
+	for _, msg := range msgs {
+		if err := m.AppendMessage(ctx, id, msg); err != nil {
+			t.Fatalf("AppendMessage: %v", err)
+		}
+	}
+
+	history, err := m.GetHistory(ctx, id, 0)
+	if err != nil {
+		t.Fatalf("GetHistory: %v", err)
+	}
+	if len(history) != 4 {
+		t.Fatalf("GetHistory len = %d, want 4", len(history))
+	}
+	if history[0].Content != "你好" {
+		t.Errorf("history[0] = %q, want %q", history[0].Content, "你好")
+	}
+}
+
+func TestGetHistoryLimit(t *testing.T) {
+	m := NewMemoryManager(30*time.Minute, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	id, _ := m.Create(ctx, models.DefaultConfig())
+
+	for i := 0; i < 10; i++ {
+		m.AppendMessage(ctx, id, models.Message{Role: "user", Content: "msg"})
+	}
+
+	history, err := m.GetHistory(ctx, id, 3)
+	if err != nil {
+		t.Fatalf("GetHistory: %v", err)
+	}
+	if len(history) != 3 {
+		t.Fatalf("GetHistory limit=3: len = %d, want 3", len(history))
+	}
+}
+
+func TestHistoryLimit(t *testing.T) {
+	const maxHistory = 5
+	m := NewMemoryManager(30*time.Minute, maxHistory)
+	defer m.Stop()
+	ctx := context.Background()
+
+	id, _ := m.Create(ctx, models.DefaultConfig())
+
+	// 插入超过上限的消息
+	for i := 0; i < 10; i++ {
+		m.AppendMessage(ctx, id, models.Message{Role: "user", Content: "msg"})
+	}
+
+	history, err := m.GetHistory(ctx, id, 0)
+	if err != nil {
+		t.Fatalf("GetHistory: %v", err)
+	}
+	if len(history) != maxHistory {
+		t.Fatalf("GetHistory after overflow: len = %d, want %d", len(history), maxHistory)
+	}
+}
+
+func TestUpdateConfig(t *testing.T) {
+	m := NewMemoryManager(30*time.Minute, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	id, _ := m.Create(ctx, models.DefaultConfig())
+
+	ttsEnabled := false
+	detailLevel := "high"
+	patch := models.SessionConfigPatch{
+		TTSEnabled:  &ttsEnabled,
+		DetailLevel: &detailLevel,
+	}
+
+	if err := m.UpdateConfig(ctx, id, patch); err != nil {
+		t.Fatalf("UpdateConfig: %v", err)
+	}
+
+	sess, _ := m.Get(ctx, id)
+	if sess.Config.TTSEnabled != false {
+		t.Errorf("TTSEnabled = %v, want false", sess.Config.TTSEnabled)
+	}
+	if sess.Config.DetailLevel != "high" {
+		t.Errorf("DetailLevel = %q, want %q", sess.Config.DetailLevel, "high")
+	}
+	// Language 未传，应保持原值
+	if sess.Config.Language != "zh-CN" {
+		t.Errorf("Language = %q, want %q", sess.Config.Language, "zh-CN")
+	}
+}
+
+func TestActiveRequest(t *testing.T) {
+	m := NewMemoryManager(30*time.Minute, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	id, _ := m.Create(ctx, models.DefaultConfig())
+
+	// 初始应为空
+	reqID, err := m.GetActiveRequestID(ctx, id)
+	if err != nil {
+		t.Fatalf("GetActiveRequestID: %v", err)
+	}
+	if reqID != "" {
+		t.Errorf("initial active request = %q, want empty", reqID)
+	}
+
+	// 设置
+	if err := m.SetActiveRequest(ctx, id, "req-123"); err != nil {
+		t.Fatalf("SetActiveRequest: %v", err)
+	}
+	reqID, _ = m.GetActiveRequestID(ctx, id)
+	if reqID != "req-123" {
+		t.Errorf("active request = %q, want %q", reqID, "req-123")
+	}
+
+	// 清除
+	if err := m.ClearActiveRequest(ctx, id); err != nil {
+		t.Fatalf("ClearActiveRequest: %v", err)
+	}
+	reqID, _ = m.GetActiveRequestID(ctx, id)
+	if reqID != "" {
+		t.Errorf("active request after clear = %q, want empty", reqID)
+	}
+}
+
+func TestTouchRefreshesTTL(t *testing.T) {
+	m := NewMemoryManager(100*time.Millisecond, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	id, _ := m.Create(ctx, models.DefaultConfig())
+
+	// 50ms 后 Touch，应重置 TTL
+	time.Sleep(50 * time.Millisecond)
+	if err := m.Touch(ctx, id); err != nil {
+		t.Fatalf("Touch: %v", err)
+	}
+
+	// 再等 70ms（距创建 120ms，但距 Touch 只有 70ms），不应过期
+	time.Sleep(70 * time.Millisecond)
+	_, err := m.Get(ctx, id)
+	if err != nil {
+		t.Errorf("Get after Touch: %v, want nil (should not expire yet)", err)
+	}
+
+	// 再等 50ms（距 Touch 120ms），应过期
+	time.Sleep(50 * time.Millisecond)
+	_, err = m.Get(ctx, id)
+	if err != ErrSessionNotFound {
+		t.Errorf("Get after TTL: err = %v, want ErrSessionNotFound", err)
+	}
+}
+
+func TestActiveCount(t *testing.T) {
+	m := NewMemoryManager(30*time.Minute, 20)
+	defer m.Stop()
+	ctx := context.Background()
+
+	if m.ActiveCount() != 0 {
+		t.Errorf("initial ActiveCount = %d, want 0", m.ActiveCount())
+	}
+
+	m.Create(ctx, models.DefaultConfig())
+	m.Create(ctx, models.DefaultConfig())
+	if m.ActiveCount() != 2 {
+		t.Errorf("ActiveCount = %d, want 2", m.ActiveCount())
+	}
+}
--- a/backend/internal/session/redis.go
+++ b/backend/internal/session/redis.go
@@ -0,0 +1,324 @@
+package session
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/redis/go-redis/v9"
+
+	"github.com/hhs/camtalk/internal/logger"
+	"github.com/hhs/camtalk/internal/models"
+)
+
+// RedisManager 基于 Redis 的 SessionManager 实现。
+// 数据结构：
+//   - session:{id}:meta    → Hash（会话元数据）
+//   - session:{id}:history → List（对话历史）
+type RedisManager struct {
+	rdb        *redis.Client
+	ttl        time.Duration
+	maxHistory int
+}
+
+// NewRedisManager 创建 Redis 版 SessionManager。
+func NewRedisManager(rdb *redis.Client, ttl time.Duration, maxHistory int) *RedisManager {
+	if ttl <= 0 {
+		ttl = defaultTTL
+	}
+	if maxHistory <= 0 {
+		maxHistory = defaultHistorySize
+	}
+	return &RedisManager{rdb: rdb, ttl: ttl, maxHistory: maxHistory}
+}
+
+func metaKey(id string) string   { return fmt.Sprintf("session:%s:meta", id) }
+func histKey(id string) string   { return fmt.Sprintf("session:%s:history", id) }
+
+// Create 创建新会话。
+func (m *RedisManager) Create(ctx context.Context, config models.SessionConfig) (string, error) {
+	id := uuidNew()
+	now := time.Now().UTC()
+
+	pipe := m.rdb.Pipeline()
+
+	// 写入 meta Hash
+	pipe.HSet(ctx, metaKey(id), map[string]interface{}{
+		"session_id":         id,
+		"config.tts_enabled": strconv.FormatBool(config.TTSEnabled),
+		"config.detail_level": config.DetailLevel,
+		"config.language":    config.Language,
+		"created_at":         now.Format(time.RFC3339),
+		"last_active":        now.Format(time.RFC3339),
+		"active_request_id":  "",
+	})
+	pipe.Expire(ctx, metaKey(id), m.ttl)
+
+	// 初始化空 history List
+	pipe.RPush(ctx, histKey(id), placeholderHistoryMark)
+	pipe.Expire(ctx, histKey(id), m.ttl)
+
+	if _, err := pipe.Exec(ctx); err != nil {
+		return "", fmt.Errorf("redis create session: %w", err)
+	}
+
+	logger.Log.Debugw("redis session created", "session", id)
+	return id, nil
+}
+
+// placeholderHistoryMark 占位符，避免 Redis 对空 key 的特殊行为。
+const placeholderHistoryMark = "__placeholder__"
+
+// Get 获取会话。
+func (m *RedisManager) Get(ctx context.Context, sessionID string) (*models.Session, error) {
+	vals, err := m.rdb.HGetAll(ctx, metaKey(sessionID)).Result()
+	if err != nil {
+		return nil, fmt.Errorf("redis get session: %w", err)
+	}
+	if len(vals) == 0 {
+		return nil, ErrSessionNotFound
+	}
+
+	sess := &models.Session{
+		ID: vals["session_id"],
+	}
+	sess.CreatedAt, _ = time.Parse(time.RFC3339, vals["created_at"])
+	sess.Config.TTSEnabled, _ = strconv.ParseBool(vals["config.tts_enabled"])
+	sess.Config.DetailLevel = vals["config.detail_level"]
+	sess.Config.Language = vals["config.language"]
+
+	return sess, nil
+}
+
+// UpdateConfig 更新会话配置。
+func (m *RedisManager) UpdateConfig(ctx context.Context, sessionID string, patch models.SessionConfigPatch) error {
+	// 先检查会话是否存在
+	exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result()
+	if err != nil {
+		return fmt.Errorf("redis check session: %w", err)
+	}
+	if exists == 0 {
+		return ErrSessionNotFound
+	}
+
+	fields := map[string]interface{}{
+		"last_active": time.Now().UTC().Format(time.RFC3339),
+	}
+	if patch.TTSEnabled != nil {
+		fields["config.tts_enabled"] = strconv.FormatBool(*patch.TTSEnabled)
+	}
+	if patch.DetailLevel != nil {
+		fields["config.detail_level"] = *patch.DetailLevel
+	}
+	if patch.Language != nil {
+		fields["config.language"] = *patch.Language
+	}
+
+	if err := m.rdb.HSet(ctx, metaKey(sessionID), fields).Err(); err != nil {
+		return fmt.Errorf("redis update config: %w", err)
+	}
+
+	// 刷新 TTL
+	m.rdb.Expire(ctx, metaKey(sessionID), m.ttl)
+	logger.Log.Debugw("redis session config updated", "session", sessionID)
+	return nil
+}
+
+// GetHistory 获取最近 N 轮对话历史。
+func (m *RedisManager) GetHistory(ctx context.Context, sessionID string, limit int) ([]models.Message, error) {
+	// 检查会话是否存在
+	exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result()
+	if err != nil {
+		return nil, fmt.Errorf("redis check session: %w", err)
+	}
+	if exists == 0 {
+		return nil, ErrSessionNotFound
+	}
+
+	if limit <= 0 {
+		limit = m.maxHistory
+	}
+
+	// LRANGE 0 {limit-1}，最新在前（LPUSH），需要反转为时间顺序
+	raws, err := m.rdb.LRange(ctx, histKey(sessionID), 0, int64(limit)).Result()
+	if err != nil {
+		return nil, fmt.Errorf("redis get history: %w", err)
+	}
+
+	var msgs []models.Message
+	for _, raw := range raws {
+		if raw == placeholderHistoryMark {
+			continue
+		}
+		var msg models.Message
+		if err := json.Unmarshal([]byte(raw), &msg); err != nil {
+			logger.Log.Warnw("invalid history entry", "session", sessionID, "raw", raw)
+			continue
+		}
+		msgs = append(msgs, msg)
+	}
+
+	// 反转为时间顺序（LPUSH 最新在前 → 需要最旧在前）
+	for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 {
+		msgs[i], msgs[j] = msgs[j], msgs[i]
+	}
+
+	return msgs, nil
+}
+
+// AppendMessage 追加一条对话消息，同时刷新 TTL。
+func (m *RedisManager) AppendMessage(ctx context.Context, sessionID string, msg models.Message) error {
+	// 检查会话是否存在
+	exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result()
+	if err != nil {
+		return fmt.Errorf("redis check session: %w", err)
+	}
+	if exists == 0 {
+		return ErrSessionNotFound
+	}
+
+	data, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("marshal message: %w", err)
+	}
+
+	pipe := m.rdb.Pipeline()
+	// LPUSH 新消息到左头（最新在前）
+	pipe.LPush(ctx, histKey(sessionID), string(data))
+	// LTRIM 保留最近 maxHistory 条（+1 是因为有占位符）
+	pipe.LTrim(ctx, histKey(sessionID), 0, int64(m.maxHistory))
+	// 刷新 TTL
+	pipe.Expire(ctx, histKey(sessionID), m.ttl)
+	pipe.Expire(ctx, metaKey(sessionID), m.ttl)
+	// 更新 last_active
+	pipe.HSet(ctx, metaKey(sessionID), "last_active", time.Now().UTC().Format(time.RFC3339))
+
+	if _, err := pipe.Exec(ctx); err != nil {
+		return fmt.Errorf("redis append message: %w", err)
+	}
+
+	return nil
+}
+
+// SetActiveRequest 标记当前正在处理的请求 ID。
+func (m *RedisManager) SetActiveRequest(ctx context.Context, sessionID string, requestID string) error {
+	exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result()
+	if err != nil {
+		return fmt.Errorf("redis check session: %w", err)
+	}
+	if exists == 0 {
+		return ErrSessionNotFound
+	}
+
+	pipe := m.rdb.Pipeline()
+	pipe.HSet(ctx, metaKey(sessionID), "active_request_id", requestID)
+	pipe.HSet(ctx, metaKey(sessionID), "last_active", time.Now().UTC().Format(time.RFC3339))
+	pipe.Expire(ctx, metaKey(sessionID), m.ttl)
+
+	if _, err := pipe.Exec(ctx); err != nil {
+		return fmt.Errorf("redis set active request: %w", err)
+	}
+	return nil
+}
+
+// GetActiveRequestID 获取当前活跃请求 ID。
+func (m *RedisManager) GetActiveRequestID(ctx context.Context, sessionID string) (string, error) {
+	val, err := m.rdb.HGet(ctx, metaKey(sessionID), "active_request_id").Result()
+	if err == redis.Nil {
+		return "", ErrSessionNotFound
+	}
+	if err != nil {
+		return "", fmt.Errorf("redis get active request: %w", err)
+	}
+	return val, nil
+}
+
+// ClearActiveRequest 清除活跃请求标记。
+func (m *RedisManager) ClearActiveRequest(ctx context.Context, sessionID string) error {
+	exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result()
+	if err != nil {
+		return fmt.Errorf("redis check session: %w", err)
+	}
+	if exists == 0 {
+		return ErrSessionNotFound
+	}
+
+	pipe := m.rdb.Pipeline()
+	pipe.HSet(ctx, metaKey(sessionID), "active_request_id", "")
+	pipe.HSet(ctx, metaKey(sessionID), "last_active", time.Now().UTC().Format(time.RFC3339))
+	pipe.Expire(ctx, metaKey(sessionID), m.ttl)
+
+	if _, err := pipe.Exec(ctx); err != nil {
+		return fmt.Errorf("redis clear active request: %w", err)
+	}
+	return nil
+}
+
+// Touch 刷新 TTL。
+func (m *RedisManager) Touch(ctx context.Context, sessionID string) error {
+	exists, err := m.rdb.Exists(ctx, metaKey(sessionID)).Result()
+	if err != nil {
+		return fmt.Errorf("redis check session: %w", err)
+	}
+	if exists == 0 {
+		return ErrSessionNotFound
+	}
+
+	pipe := m.rdb.Pipeline()
+	pipe.Expire(ctx, metaKey(sessionID), m.ttl)
+	pipe.Expire(ctx, histKey(sessionID), m.ttl)
+	pipe.HSet(ctx, metaKey(sessionID), "last_active", time.Now().UTC().Format(time.RFC3339))
+
+	if _, err := pipe.Exec(ctx); err != nil {
+		return fmt.Errorf("redis touch: %w", err)
+	}
+	return nil
+}
+
+// Destroy 显式销毁会话。
+func (m *RedisManager) Destroy(ctx context.Context, sessionID string) error {
+	deleted, err := m.rdb.Del(ctx, metaKey(sessionID), histKey(sessionID)).Result()
+	if err != nil {
+		return fmt.Errorf("redis destroy session: %w", err)
+	}
+	if deleted == 0 {
+		return ErrSessionNotFound
+	}
+
+	logger.Log.Debugw("redis session destroyed", "session", sessionID)
+	return nil
+}
+
+// ActiveCount 返回当前活跃会话数。
+// Redis 实现通过 SCAN 遍历 meta key，适用于中等规模。
+// 大规模部署建议维护独立的活跃会话集合。
+func (m *RedisManager) ActiveCount() int {
+	ctx := context.Background()
+	count := 0
+	var cursor uint64
+	for {
+		keys, nextCursor, err := m.rdb.Scan(ctx, cursor, "session:*:meta", 100).Result()
+		if err != nil {
+			break
+		}
+		for _, key := range keys {
+			exists, _ := m.rdb.Exists(ctx, key).Result()
+			if exists > 0 {
+				count++
+			}
+		}
+		cursor = nextCursor
+		if cursor == 0 {
+			break
+		}
+	}
+	return count
+}
+
+// uuidNew 生成 UUID，便于测试时 mock。
+var uuidNew = func() string {
+	return uuid.New().String()
+}
--- a/backend/internal/ws/handler.go
+++ b/backend/internal/ws/handler.go
@@ -1,17 +1,19 @@
 package ws

 import (
+	"context"
 	"encoding/json"
 	"net/http"
 	"sync"
 	"time"

 	"github.com/gin-gonic/gin"
-	"github.com/google/uuid"
 	"github.com/gorilla/websocket"

+	"github.com/hhs/camtalk/internal/errors"
 	"github.com/hhs/camtalk/internal/logger"
 	"github.com/hhs/camtalk/internal/models"
+	"github.com/hhs/camtalk/internal/session"
 )

 var upgrader = websocket.Upgrader{
@@ -25,14 +27,21 @@ type Client struct {
 	mu        sync.Mutex
 }

-func (c *Client) sendJSON(v any) error {
+// SendJSON 向客户端发送 JSON 消息（公开以便 errors 包调用）。
+func (c *Client) SendJSON(v any) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	return c.conn.WriteJSON(v)
 }

 // ServeWS 处理 WebSocket 升级请求。
-func ServeWS(c *gin.Context) {
+func ServeWS(sessionMgr session.Manager) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		serveWS(c, sessionMgr)
+	}
+}
+
+func serveWS(c *gin.Context, sessionMgr session.Manager) {
 	conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
 	if err != nil {
 		logger.Log.Errorw("websocket upgrade failed", "error", err)
@@ -40,11 +49,17 @@ func ServeWS(c *gin.Context) {
 	}
 	defer conn.Close()

-	sessionID := uuid.New().String()
+	// 创建会话
+	sessionID, err := sessionMgr.Create(context.Background(), models.DefaultConfig())
+	if err != nil {
+		logger.Log.Errorw("create session failed", "error", err)
+		return
+	}
+
 	client := &Client{conn: conn, sessionID: sessionID}

 	// 发送 connected 消息
-	_ = client.sendJSON(models.WsConnected{
+	_ = client.SendJSON(models.WsConnected{
 		Type:          "connected",
 		SessionID:     sessionID,
 		ServerVersion: "0.1.0",
@@ -92,51 +107,67 @@ func ServeWS(c *gin.Context) {
 			Type string `json:"type"`
 		}
 		if err := json.Unmarshal(message, &envelope); err != nil {
-			_ = client.sendJSON(models.WsError{
-				Type:    "error",
-				Code:    "INVALID_MESSAGE",
-				Message: "invalid JSON",
-			})
+			errors.SendWSError(client, errors.CodeInvalidMessage, "", err)
 			continue
 		}

 		switch envelope.Type {
 		case "ping":
-			_ = client.sendJSON(models.WsPong{Type: "pong"})
+			_ = client.SendJSON(models.WsPong{Type: "pong"})

 		case "query":
 			var msg models.WsQuery
 			if err := json.Unmarshal(message, &msg); err != nil {
-				_ = client.sendJSON(models.WsError{
-					Type:      "error",
-					Code:      "INVALID_MESSAGE",
-					Message:   "invalid query message",
-					RequestID: msg.RequestID,
-				})
+				errors.SendWSError(client, errors.CodeInvalidMessage, msg.RequestID, err)
 				continue
 			}
 			logger.Log.Infow("query received", "session", sessionID, "request", msg.RequestID)
-			// TODO: 调用 AI 编排流程（STT → LLM → TTS）
+
+			// 刷新会话 TTL
+			if err := sessionMgr.Touch(context.Background(), sessionID); err != nil {
+				logger.Log.Warnw("touch session failed", "session", sessionID, "error", err)
+			}
+
+			// 标记活跃请求
+			if err := sessionMgr.SetActiveRequest(context.Background(), sessionID, msg.RequestID); err != nil {
+				logger.Log.Warnw("set active request failed", "session", sessionID, "error", err)
+			}
+
+			// 获取对话历史（供后续 Orchestrator 使用）
+			_, _ = sessionMgr.GetHistory(context.Background(), sessionID, 20)
+
+			// TODO: 解码 audio Base64 → 启动 orchestrator.ProcessQuery goroutine

 		case "config":
 			var msg models.WsConfig
 			if err := json.Unmarshal(message, &msg); err != nil {
-				_ = client.sendJSON(models.WsError{
-					Type:    "error",
-					Code:    "INVALID_MESSAGE",
-					Message: "invalid config message",
-				})
+				errors.SendWSError(client, errors.CodeInvalidMessage, "", err)
 				continue
 			}
-			logger.Log.Infow("config update", "session", sessionID)
-			// TODO: 更新会话配置
+
+			patch := models.SessionConfigPatch{
+				TTSEnabled:  msg.Payload.TTSEnabled,
+				DetailLevel: msg.Payload.DetailLevel,
+				Language:    msg.Payload.Language,
+			}
+			if err := sessionMgr.UpdateConfig(context.Background(), sessionID, patch); err != nil {
+				errors.SendWSError(client, errors.CodeInternalError, "", err)
+				continue
+			}
+			logger.Log.Infow("config updated", "session", sessionID)

 		case "interrupt":
 			logger.Log.Infow("interrupt received", "session", sessionID)
-			// TODO: 中断当前 AI 响应
+
+			// 获取活跃请求 ID（实际 cancel 在 Phase 5 接入 orchestrator 后实现）
+			reqID, _ := sessionMgr.GetActiveRequestID(context.Background(), sessionID)
+			if reqID != "" {
+				_ = sessionMgr.ClearActiveRequest(context.Background(), sessionID)
+				// TODO: 取消对应 context cancel func
+			}

 		default:
-			_ = client.sendJSON(models.WsError{
+			_ = client.SendJSON(models.WsError{
 				Type:    "error",
 				Code:    "INVALID_MESSAGE",
 				Message: "unknown message type: " + envelope.Type,
@@ -145,5 +176,7 @@ func ServeWS(c *gin.Context) {
 	}

 	close(done)
+
+	// 断开连接时不销毁会话，让其自然过期（支持重连恢复）
 	logger.Log.Infow("client disconnected", "session", sessionID)
 }