feat(category): implement recategorize backfill engine and fallback mapping (Spec §3.4-3.5 A4-A5)

cmd/recategorize/main.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"time"
+
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/config"
+)
+
+func main() {
+	envPath := flag.String("env", ".env", "环境配置文件路径")
+	sessionID := flag.String("session", "", "指定处理的会话ID (留空则全量回填)")
+	flag.Parse()
+
+	cfg, err := config.Load(*envPath)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "加载配置失败: %v\n", err)
+		os.Exit(1)
+	}
+
+	svc := bizdev.NewService(cfg)
+	ctxMgr := svc.ContextManager()
+	if ctxMgr == nil {
+		fmt.Fprintf(os.Stderr, "ContextManager 未初始化\n")
+		os.Exit(1)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+
+	fmt.Println("==================================================================")
+	fmt.Println("   🔄 历史消息分类回填与数据对齐工具 (Spec §3.4 A4)   ")
+	fmt.Println("==================================================================")
+	if *sessionID != "" {
+		fmt.Printf("处理目标会话: %s\n", *sessionID)
+	} else {
+		fmt.Println("处理目标: 全量历史消息")
+	}
+
+	res, err := ctxMgr.RecategorizeMessages(ctx, *sessionID)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "分类回填失败: %v\n", err)
+		os.Exit(1)
+	}
+
+	fmt.Printf("✅ 分类回填完成!\n已扫描消息数: %d\n已更新分类数: %d\n", res.TotalProcessed, res.TotalUpdated)
+}

internal/ctxmgr/recategorize.go

@@ -0,0 +1,154 @@
+package ctxmgr
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+// RecategorizeResult 存储重分类批处理结果统计
+type RecategorizeResult struct {
+	TotalProcessed int `json:"total_processed"`
+	TotalUpdated   int `json:"total_updated"`
+}
+
+type recatMsg struct {
+	ID        int64
+	SessionID string
+	Role      string
+	Intent    string
+	Branch    string
+	Metadata  map[string]any
+	Category  string
+	CreatedAt time.Time
+}
+
+// RecategorizeMessages 存量消息重分类回填引擎 (Spec §3.4 A4)
+//
+// 1. assistant 消息：
+//   - metadata 含有 turn_intent -> 直接按 IntentCategoryMap 映射；
+//   - metadata 无 turn_intent -> 回溯同轮用户消息（时间间隔 < 60s 且 role=user 的前序消息）推导其意图；
+//   - 兜底按现有 DeriveMessageCategory 判定；
+//
+// 2. user 消息：
+//   - 按最新意图词表重新判定，纠偏遗漏映射（如 CONFIRM_ACTION、EXECUTE_STAGE_ADVANCE 等）；
+//
+// 3. 幂等更新：仅当计算出的 newCategory != 当前 category 时才执行持久化。
+func (m *ContextManager) RecategorizeMessages(ctx context.Context, sessionID string) (*RecategorizeResult, error) {
+	result := &RecategorizeResult{}
+
+	if m.pool != nil {
+		query := `
+			SELECT id, session_id, role, intent, branch, metadata, category, created_at
+			FROM chat_messages
+			WHERE ($1 = '' OR session_id = $1) AND deleted_at IS NULL
+			ORDER BY session_id, created_at ASC, id ASC
+		`
+		rows, err := m.pool.Query(ctx, query, sessionID)
+		if err != nil {
+			return nil, fmt.Errorf("query messages for recategorize failed: %w", err)
+		}
+		defer rows.Close()
+
+		var msgs []recatMsg
+		for rows.Next() {
+			var r recatMsg
+			var metaJSON []byte
+			if err := rows.Scan(&r.ID, &r.SessionID, &r.Role, &r.Intent, &r.Branch, &metaJSON, &r.Category, &r.CreatedAt); err != nil {
+				return nil, fmt.Errorf("scan recategorize row failed: %w", err)
+			}
+			r.Metadata = make(map[string]any)
+			if len(metaJSON) > 0 {
+				_ = json.Unmarshal(metaJSON, &r.Metadata)
+			}
+			msgs = append(msgs, r)
+		}
+
+		var lastUserMsg *recatMsg
+		for _, msg := range msgs {
+			result.TotalProcessed++
+
+			var targetIntent = msg.Intent
+			var targetMeta = msg.Metadata
+			if targetMeta == nil {
+				targetMeta = make(map[string]any)
+			}
+
+			if msg.Role == "user" {
+				copyMsg := msg
+				lastUserMsg = &copyMsg
+			} else if msg.Role == "assistant" {
+				// 若元数据中无 turn_intent，尝试回溯 60s 内的同轮用户消息
+				if _, ok := targetMeta["turn_intent"]; !ok && lastUserMsg != nil && lastUserMsg.SessionID == msg.SessionID {
+					if msg.CreatedAt.Sub(lastUserMsg.CreatedAt) <= 60*time.Second {
+						if lastUserMsg.Intent != "" {
+							targetMeta["turn_intent"] = lastUserMsg.Intent
+						}
+					}
+				}
+			}
+
+			newCategory := DeriveMessageCategory(msg.Role, targetIntent, msg.Branch, targetMeta)
+
+			if newCategory != msg.Category {
+				_, err := m.pool.Exec(ctx, `
+					UPDATE chat_messages
+					SET category = $1, metadata = $2
+					WHERE id = $3
+				`, newCategory, targetMeta, msg.ID)
+				if err != nil {
+					return nil, fmt.Errorf("update message %d category failed: %w", msg.ID, err)
+				}
+				result.TotalUpdated++
+			}
+		}
+
+		return result, nil
+	}
+
+	// 内存模式 fallback
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	for sid, list := range m.messages {
+		if sessionID != "" && sid != sessionID {
+			continue
+		}
+
+		var lastUserMsg *ChatMessageRecord
+		for i := range list {
+			result.TotalProcessed++
+			msg := &list[i]
+			if msg.DeletedAt != nil {
+				continue
+			}
+
+			meta := msg.Metadata
+			if meta == nil {
+				meta = make(map[string]any)
+			}
+
+			if msg.Role == "user" {
+				lastUserMsg = msg
+			} else if msg.Role == "assistant" {
+				if _, ok := meta["turn_intent"]; !ok && lastUserMsg != nil && lastUserMsg.SessionID == msg.SessionID {
+					if msg.CreatedAt.Sub(lastUserMsg.CreatedAt) <= 60*time.Second {
+						if lastUserMsg.Intent != "" {
+							meta["turn_intent"] = lastUserMsg.Intent
+						}
+					}
+				}
+			}
+
+			newCategory := DeriveMessageCategory(msg.Role, msg.Intent, msg.Branch, meta)
+			if newCategory != msg.Category {
+				msg.Category = newCategory
+				msg.Metadata = meta
+				result.TotalUpdated++
+			}
+		}
+	}
+
+	return result, nil
+}

internal/ctxmgr/recategorize_test.go

@@ -0,0 +1,61 @@
+package ctxmgr
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestRecategorizeMessages_MemoryMode(t *testing.T) {
+	mgr := NewContextManager(nil, nil, DefaultTimeDecayConfig())
+	ctx := context.Background()
+	sessID := "sess-recat-test"
+
+	// 构造测试数据：
+	// 1. 用户发送咨询请求 (intent=QUERY_REGULATIONS) -> 当前被错填或历史未填分类
+	now := time.Now()
+	_, _ = mgr.RecordUserMessageWithID(ctx, sessID, 1, "user", "请问NMPA注册要求", "QUERY_REGULATIONS", "", map[string]any{
+		"created_at": now,
+	})
+
+	// 2. 助手在 10s 后回复，由于历史数据缺失 turn_intent，被初始化成了 uncategorized
+	asstID, _ := mgr.RecordUserMessageWithID(ctx, sessID, 0, "assistant", "NMPA申报需准备以下资料...", "", "", map[string]any{
+		"created_at":        now.Add(10 * time.Second),
+		"category_override": CategoryUncategorized, // 人为模拟历史错误分类
+	})
+
+	// 执行重分类
+	res, err := mgr.RecategorizeMessages(ctx, sessID)
+	if err != nil {
+		t.Fatalf("RecategorizeMessages failed: %v", err)
+	}
+
+	if res.TotalProcessed != 2 {
+		t.Errorf("expected 2 processed, got %d", res.TotalProcessed)
+	}
+	if res.TotalUpdated < 1 {
+		t.Errorf("expected at least 1 updated, got %d", res.TotalUpdated)
+	}
+
+	// 验证 assistant 消息已被回填为 consult
+	msgs := mgr.messages[sessID]
+	for _, m := range msgs {
+		if m.ID == asstID {
+			if m.Category != CategoryConsult {
+				t.Errorf("expected assistant message category to be %s, got %s", CategoryConsult, m.Category)
+			}
+			if m.Metadata["turn_intent"] != "QUERY_REGULATIONS" {
+				t.Errorf("expected turn_intent to be backfilled as QUERY_REGULATIONS, got %v", m.Metadata["turn_intent"])
+			}
+		}
+	}
+
+	// 再次执行，幂等性验证（不应产生任何新更新）
+	res2, err := mgr.RecategorizeMessages(ctx, sessID)
+	if err != nil {
+		t.Fatalf("RecategorizeMessages idempotent failed: %v", err)
+	}
+	if res2.TotalUpdated != 0 {
+		t.Errorf("expected 0 updated on second run, got %d", res2.TotalUpdated)
+	}
+}

internal/server/chat_recategorize_api.go

@@ -0,0 +1,39 @@
+package server
+
+// chat_recategorize_api.go — 管理端点 POST /api/v1/chat/recategorize (Spec §3.4 A4)
+
+import (
+	"encoding/json"
+	"net/http"
+)
+
+type recategorizeReq struct {
+	SessionID string `json:"session_id"`
+}
+
+func (s *Server) handleChatRecategorize(w http.ResponseWriter, r *http.Request) {
+	if s.svc == nil || s.svc.ContextManager() == nil {
+		http.Error(w, "context manager unavailable", http.StatusServiceUnavailable)
+		return
+	}
+
+	var req recategorizeReq
+	if r.Body != nil {
+		_ = json.NewDecoder(r.Body).Decode(&req)
+	}
+
+	res, err := s.svc.ContextManager().RecategorizeMessages(r.Context(), req.SessionID)
+	if err != nil {
+		writeJSONResponse(w, http.StatusInternalServerError, map[string]any{
+			"success": false,
+			"error":   err.Error(),
+		})
+		return
+	}
+
+	writeJSONResponse(w, http.StatusOK, map[string]any{
+		"success":         true,
+		"total_processed": res.TotalProcessed,
+		"total_updated":   res.TotalUpdated,
+	})
+}

internal/server/chat_recategorize_api_test.go

@@ -0,0 +1,56 @@
+package server_test
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/config"
+	"chat-agent-system/internal/server"
+)
+
+func TestServer_ChatRecategorizeAPI(t *testing.T) {
+	cfg := &config.Config{
+		Env:      "dev",
+		HTTPHost: "127.0.0.1",
+		HTTPPort: 8892,
+		DataDir:  t.TempDir(),
+	}
+	svc := bizdev.NewService(cfg)
+	srv := server.NewServer(cfg, svc)
+	handler := srv.Routes()
+
+	sessID := "sess_recategorize_api_001"
+	ctx := t.Context()
+	ctxMgr := svc.ContextManager()
+
+	// 注入种子消息
+	_, _ = ctxMgr.RecordUserMessageWithID(ctx, sessID, 1, "user", "请问二类注册审批周期？", "QUERY_REGULATIONS", "", nil)
+	_, _ = ctxMgr.RecordUserMessageWithID(ctx, sessID, 0, "assistant", "根据现有规定，一般需 6-9 个月。", "", "", nil)
+
+	// 调用 POST /api/v1/chat/recategorize
+	reqBody := `{"session_id":"` + sessID + `"}`
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/chat/recategorize", strings.NewReader(reqBody))
+	req.Header.Set("Content-Type", "application/json")
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("expected status 200, got %d: %s", rec.Code, rec.Body.String())
+	}
+
+	var res map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &res); err != nil {
+		t.Fatalf("json unmarshal failed: %v", err)
+	}
+
+	if success, ok := res["success"].(bool); !ok || !success {
+		t.Fatalf("expected success: true, got %v", res)
+	}
+	if totalProcessed, ok := res["total_processed"].(float64); !ok || totalProcessed < 2 {
+		t.Fatalf("expected total_processed >= 2, got %v", res["total_processed"])
+	}
+}

internal/server/server.go

@@ -106,6 +106,7 @@ func (s *Server) Routes() http.Handler {
 	mux.HandleFunc("GET /api/v1/chat/unread", s.handleChatUnread)
 	mux.HandleFunc("POST /api/v1/chat/read", s.handleChatRead)
 	mux.HandleFunc("GET /api/v1/chat/export/messages", s.handleChatExportMessages)
+	mux.HandleFunc("POST /api/v1/chat/recategorize", s.handleChatRecategorize)
 	mux.HandleFunc("GET /api/v1/context/cases/recommend", s.handleRecommendCases)
 
 	// 8. 受控文档编制、审核、批准生命周期与认领治理

web/src/data/storage.ts

@@ -3,6 +3,7 @@ import { RoleDefinition } from '../types/rbac';
 import { UserProfile } from '../types/docgov';
 import { buildPrivateSessionId, dedupePrivateSessions } from './sessionPrivacy';
 import { NEUROPULSE_SESSIONS, NEUROPULSE_MESSAGES, NEUROPULSE_ACTIVE_SESSION_ID } from './neuropulseDemo';
+import { enrichMessagesWithCategories } from '../engine/categoryHelper';
 
 // v6：修正 NeuroPulse 驳回追问误挂「疲劳寿命」旧答复；对话 LLM 每轮独立工作区
 const STORAGE_PREFIX = 'smdf_agent_workbench_v6_';
@@ -493,14 +494,15 @@ export function loadMessages(): ChatMessage[] {
   const cleaned = loaded.filter((m) => !legacyIds.has(m.sessionId));
   const existingIds = new Set(cleaned.map((m) => m.id));
   const missing = DEFAULT_MESSAGES.filter((m) => !existingIds.has(m.id));
+  let result: ChatMessage[];
   if (missing.length > 0) {
-    return [...cleaned, ...missing];
+    result = [...cleaned, ...missing];
+  } else if (cleaned.length === 0) {
+    result = [...DEFAULT_MESSAGES];
+  } else {
+    result = cleaned;
   }
-  // 若清洗后几乎为空，回退默认预设
-  if (cleaned.length === 0) {
-    return [...DEFAULT_MESSAGES];
-  }
-  return cleaned;
+  return enrichMessagesWithCategories(result);
 }
 
 export function saveMessages(messages: ChatMessage[]): void {

web/src/engine/categoryHelper.test.ts

@@ -0,0 +1,77 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { deriveFallbackCategory, enrichMessagesWithCategories } from './categoryHelper';
+import { ChatMessage } from '../types/workbench';
+
+test('deriveFallbackCategory identifies categories correctly', () => {
+  assert.equal(deriveFallbackCategory({ role: 'system', content: '系统启动' }), 'system');
+
+  assert.equal(
+    deriveFallbackCategory({
+      role: 'assistant',
+      intentTag: '🎯 项目初始画像与责任人就绪',
+      content: '初始化完成',
+    }),
+    'milestone'
+  );
+
+  assert.equal(
+    deriveFallbackCategory({
+      role: 'assistant',
+      intentTag: '📚 监管法规与核心参数咨询',
+      content: '法规要点解析',
+    }),
+    'consult'
+  );
+
+  assert.equal(
+    deriveFallbackCategory({
+      role: 'assistant',
+      phaseEvents: [{ step: '1', icon: '👥', label: '组装小队', status: 'completed' }],
+      content: '开始执行声学建模任务',
+    }),
+    'task'
+  );
+
+  assert.equal(
+    deriveFallbackCategory({
+      role: 'assistant',
+      intentTag: '驳回门禁',
+      meta: { interactive_card: true },
+      content: '请重新审核',
+    }),
+    'review'
+  );
+});
+
+test('enrichMessagesWithCategories backfills categories without overwriting existing ones', () => {
+  const input: ChatMessage[] = [
+    {
+      id: 'm-1',
+      sessionId: 'sess-1',
+      role: 'user',
+      content: '请教三类超声设备的注册审评要求和关键工程参数',
+      timestamp: '10:00',
+    },
+    {
+      id: 'm-2',
+      sessionId: 'sess-1',
+      role: 'assistant',
+      content: '三类超声器械审评要求如下：1. 强标符合性...',
+      timestamp: '10:01',
+    },
+    {
+      id: 'm-3',
+      sessionId: 'sess-1',
+      role: 'user',
+      category: 'task',
+      content: '启动声学仿真任务',
+      timestamp: '10:02',
+    },
+  ];
+
+  const enriched = enrichMessagesWithCategories(input);
+  assert.equal(enriched[0].category, 'consult');
+  assert.equal(enriched[1].category, 'consult'); // inherited / derived
+  assert.equal(enriched[2].category, 'task'); // retained original
+});

web/src/engine/categoryHelper.ts

@@ -0,0 +1,99 @@
+import { ChatMessage, MessageCategory } from '../types/workbench';
+
+/**
+ * deriveFallbackCategory
+ * 依据消息的 role、intentTag、meta、phaseEvents 以及上下文推导分类 (Spec §3.5 A5)
+ * 避免在纯前端/离线回退模式下出现大量 'uncategorized'。
+ */
+export function deriveFallbackCategory(
+  msg: Partial<ChatMessage>,
+  prevUserMsg?: Partial<ChatMessage>
+): MessageCategory {
+  if (msg.category && msg.category !== 'uncategorized') {
+    return msg.category;
+  }
+
+  if (msg.role === 'system') {
+    return 'system';
+  }
+
+  // 1. milestone: 里程碑与阶段标记
+  if (msg.meta && msg.meta.milestone) {
+    return 'milestone';
+  }
+
+  // 2. file: 文件/附件收发
+  if ((msg.attachments && msg.attachments.length > 0) || (msg.meta && (msg.meta.attachments || msg.meta.file_op))) {
+    return 'file';
+  }
+
+  // 3. review: 审批流转与交互卡片
+  if (msg.interactiveCard || (msg.meta && msg.meta.interactive_card)) {
+    return 'review';
+  }
+
+  // 4. task: 包含阶段进度/执行追踪
+  if ((msg.phaseEvents && msg.phaseEvents.length > 0) || msg.taskRequest || (msg.meta && msg.meta.task_request)) {
+    return 'task';
+  }
+
+  const intentTag = msg.intentTag || '';
+  const content = msg.content || '';
+
+  // 5. 意图标签与关键词启发式推导
+  if (
+    /法规|咨询|参数|研判|建议|查询|要点|指南/.test(intentTag) ||
+    /法规|参数|规范|请教|咨询/.test(content.slice(0, 100))
+  ) {
+    return 'consult';
+  }
+
+  if (/立项|责任人|画像|阶段|里程碑|拍板|就绪/.test(intentTag)) {
+    return 'milestone';
+  }
+
+  if (/任务|执行|推进|生成|建模|仿真|门禁/.test(intentTag)) {
+    return 'task';
+  }
+
+  if (/审批|审核|驳回|认领|签字|提交/.test(intentTag)) {
+    return 'review';
+  }
+
+  if (/闲聊|问候|你好|早安|晚安/.test(intentTag)) {
+    return 'casual';
+  }
+
+  // 6. 若助手回复缺失分类，尝试继承本轮上一条用户消息的推导分类
+  if (msg.role === 'assistant' && prevUserMsg) {
+    const prevCat = deriveFallbackCategory(prevUserMsg);
+    if (prevCat && prevCat !== 'uncategorized' && prevCat !== 'system') {
+      return prevCat;
+    }
+  }
+
+  // 7. 有实质内容的提问/回复默认归入咨询，短句问候归入闲聊
+  if (content.length > 20) {
+    return 'consult';
+  }
+
+  return 'casual';
+}
+
+/**
+ * enrichMessagesWithCategories
+ * 遍历并填充缺失分类的消息列表
+ */
+export function enrichMessagesWithCategories(messages: ChatMessage[]): ChatMessage[] {
+  let lastUserMsg: ChatMessage | undefined;
+  return messages.map((m) => {
+    const enrichedCat = deriveFallbackCategory(m, lastUserMsg);
+    if (m.role === 'user') {
+      lastUserMsg = m;
+    }
+    if (!m.category || m.category === 'uncategorized') {
+      return { ...m, category: enrichedCat };
+    }
+    return m;
+  });
+}

web/src/hooks/useChatWindow.ts

@@ -10,6 +10,7 @@ import {
   loadSessionDraft,
   saveSessionDraft,
 } from '../data/storage';
+import { enrichMessagesWithCategories } from '../engine/categoryHelper';
 
 export type WindowPhase =
   | 'loading_window'
@@ -92,16 +93,17 @@ export function useChatWindow({
         setOldestLoadedDay(res.oldestLoadedDay);
         saveSessionWindowCache(sessionId, res.days);
       } else if (initialFallbackMessages && initialFallbackMessages.length > 0) {
-        // 若服务端尚无历史数据（如纯本地 Demo 会话），将 fallback 消息包装为今日 bucket
+        // 若服务端尚无历史数据（如纯本地 Demo 会话），将 fallback 消息包装为今日 bucket (Spec §3.5 A5)
+        const enrichedFallback = enrichMessagesWithCategories(initialFallbackMessages);
         const todayStr = getTodayKey();
         const fallbackBucket: ChatDayBucket = {
           dayKey: todayStr,
-          count: initialFallbackMessages.length,
-          loadedCount: initialFallbackMessages.length,
+          count: enrichedFallback.length,
+          loadedCount: enrichedFallback.length,
           truncated: false,
           firstMessageId: 0,
           lastMessageId: 0,
-          messages: initialFallbackMessages,
+          messages: enrichedFallback,
         };
         setDayStack([fallbackBucket]);
         setHasMoreBefore(false);
@@ -115,16 +117,17 @@ export function useChatWindow({
       // 若已有本地快照则维持
       const cached = loadSessionWindowCache(sessionId);
       if (!cached && initialFallbackMessages && initialFallbackMessages.length > 0) {
+        const enrichedFallback = enrichMessagesWithCategories(initialFallbackMessages);
         const todayStr = getTodayKey();
         setDayStack([
           {
             dayKey: todayStr,
-            count: initialFallbackMessages.length,
-            loadedCount: initialFallbackMessages.length,
+            count: enrichedFallback.length,
+            loadedCount: enrichedFallback.length,
             truncated: false,
             firstMessageId: 0,
             lastMessageId: 0,
-            messages: initialFallbackMessages,
+            messages: enrichedFallback,
           },
         ]);
       }

web/src/types/workbench.ts

@@ -139,6 +139,7 @@ export interface ChatMessage {
   category?: MessageCategory;
   quotes?: QuotedMessage[];
   createdAt?: string;
+  meta?: Record<string, any>;
 }
 
 export type MessageCategory =