fix(storage): 归一化历史前缀累积污染的会话 ID（一次性迁移）

- 早期版本重复拼接通道前缀产生 websocket:websocket:<uuid> 一类的污染会话记录，会话列表因此出现多条前缀逐层累积的重复会话

- 新增 user_version=2 迁移：子表数据改指归一化 ID，污染会话并入既有会话或重命名，seq 按时间重编号，整体事务保证原子性

- 新增回归测试覆盖合并、重命名、seq 连续性与幂等重跑

src/storage/migrations.rs

@@ -5,6 +5,7 @@
 //! every [`super::SessionStore`] construction and are idempotent.
 
 use rusqlite::Connection;
+use rusqlite::params;
 
 use super::StorageError;
 
@@ -213,6 +214,175 @@ pub(super) fn ensure_memory_scope_key_migration(conn: &Connection) -> Result<(),
     Ok(())
 }
 
+/// 修复历史 bug 污染的会话 ID（user_version 2，一次性迁移）。
+///
+/// 早期版本在 chat_id 已带通道前缀时会再次拼接 `{channel}:{chat_id}`，
+/// 产生形如 `websocket:websocket:<uuid>` 的污染会话记录，且每次重连都会
+/// 多累积一层前缀、多出一条孤儿会话。`persistent_session_id` 现已幂等，
+/// 不会再产生新污染，但存量污染记录会在会话列表中显示为多个重复会话。
+///
+/// 本迁移将污染 ID 归一化：子表数据（消息/话题/技能事件/待办/子代理/记忆溯源）
+/// 改指归一化 ID，污染会话记录本身被删除或重命名。通过 PRAGMA user_version
+/// 追踪，仅执行一次；整体包在事务中，失败自动回滚，重启后可安全重跑。
+pub(super) fn repair_session_id_prefix_pollution(
+    conn: &mut Connection,
+) -> Result<(), StorageError> {
+    const SESSION_ID_REPAIR_VERSION: i64 = 2;
+
+    let current_version: i64 = conn.query_row("PRAGMA user_version", [], |row| row.get(0))?;
+    if current_version >= SESSION_ID_REPAIR_VERSION {
+        return Ok(());
+    }
+
+    let tx = conn.transaction()?;
+    repair_session_id_prefix_pollution_inner(&tx)?;
+    tx.execute(
+        &format!("PRAGMA user_version = {SESSION_ID_REPAIR_VERSION}"),
+        [],
+    )?;
+    tx.commit()?;
+
+    Ok(())
+}
+
+fn repair_session_id_prefix_pollution_inner(conn: &Connection) -> Result<(), StorageError> {
+    let candidates: Vec<(String, String, String)> = {
+        let mut stmt = conn.prepare("SELECT id, channel_name, chat_id FROM sessions")?;
+        let rows = stmt.query_map([], |row| {
+            Ok((
+                row.get::<_, String>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, String>(2)?,
+            ))
+        })?;
+        rows.collect::<Result<Vec<_>, _>>()?
+    };
+
+    let mut repaired = 0usize;
+    for (polluted_id, channel_name, chat_id) in candidates {
+        let normalized = super::persistent_session_id(&channel_name, &polluted_id);
+        if normalized == polluted_id {
+            continue;
+        }
+
+        let target_exists: bool = conn
+            .query_row(
+                "SELECT COUNT(*) FROM sessions WHERE id = ?1",
+                params![normalized],
+                |row| row.get::<_, i64>(0),
+            )
+            .map(|count| count > 0)?;
+
+        if target_exists {
+            // 并入既有归一化会话：先把污染消息的 seq 偏移大数，
+            // 避免改指 session_id 时违反 UNIQUE(session_id, seq)，合并后统一重编号。
+            conn.execute(
+                "UPDATE messages SET seq = seq + 1000000000 WHERE session_id = ?1",
+                params![polluted_id],
+            )?;
+        } else {
+            // 无归一化会话：先建归一化 ID 的记录（子表改指时父记录必须已存在，
+            // 否则会违反外键约束），最后删除污染记录。
+            // websocket/cli 的 chat_id 与 session id 同源，顺带剥离累积前缀。
+            let clean_chat_id = if channel_name == "cli" || channel_name == "websocket" {
+                super::persistent_session_id(&channel_name, &chat_id)
+            } else {
+                chat_id
+            };
+            conn.execute(
+                "INSERT INTO sessions (
+                    id, title, channel_name, chat_id, summary,
+                    created_at, updated_at, last_active_at, archived_at, deleted_at,
+                    message_count, user_turn_count, agent_prompt_reinjection_count
+                 )
+                 SELECT ?2, title, channel_name, ?3, summary,
+                        created_at, updated_at, last_active_at, archived_at, deleted_at,
+                        message_count, user_turn_count, agent_prompt_reinjection_count
+                 FROM sessions WHERE id = ?1",
+                params![polluted_id, normalized, clean_chat_id],
+            )?;
+        }
+
+        conn.execute(
+            "UPDATE messages SET session_id = ?2 WHERE session_id = ?1",
+            params![polluted_id, normalized],
+        )?;
+        conn.execute(
+            "UPDATE topics SET session_id = ?2 WHERE session_id = ?1",
+            params![polluted_id, normalized],
+        )?;
+        conn.execute(
+            "UPDATE skill_events SET session_id = ?2 WHERE session_id = ?1",
+            params![polluted_id, normalized],
+        )?;
+        conn.execute(
+            "UPDATE todos SET session_id = ?2 WHERE session_id = ?1",
+            params![polluted_id, normalized],
+        )?;
+        conn.execute(
+            "UPDATE pending_subagents SET parent_session_id = ?2 WHERE parent_session_id = ?1",
+            params![polluted_id, normalized],
+        )?;
+        conn.execute(
+            "UPDATE memories SET source_session_id = ?2 WHERE source_session_id = ?1",
+            params![polluted_id, normalized],
+        )?;
+
+        if target_exists {
+            conn.execute(
+                "UPDATE sessions SET
+                    created_at = MIN(created_at, (SELECT created_at FROM sessions WHERE id = ?2)),
+                    updated_at = MAX(updated_at, (SELECT updated_at FROM sessions WHERE id = ?2)),
+                    last_active_at = MAX(last_active_at, (SELECT last_active_at FROM sessions WHERE id = ?2)),
+                    message_count = message_count + (SELECT message_count FROM sessions WHERE id = ?2),
+                    user_turn_count = user_turn_count + (SELECT user_turn_count FROM sessions WHERE id = ?2)
+                 WHERE id = ?1",
+                params![normalized, polluted_id],
+            )?;
+        }
+
+        conn.execute("DELETE FROM sessions WHERE id = ?1", params![polluted_id])?;
+
+        if target_exists {
+            // 重编号：按 created_at 恢复连续 seq。原地 UPDATE 会在行级更新过程中
+            // 与尚未更新的行冲突 UNIQUE(session_id, seq)，因此分两步：
+            // 先写入"大偏移 + 行号"（互不冲突），再统一减去偏移还原为 1..N。
+            const SEQ_RENUMBER_OFFSET: i64 = 2_000_000_000;
+            conn.execute(
+                "UPDATE messages SET seq = ?2 + (
+                    SELECT t.rn FROM (
+                        SELECT id AS mid, ROW_NUMBER() OVER (ORDER BY created_at ASC, seq ASC) AS rn
+                        FROM messages WHERE session_id = ?1
+                    ) t WHERE t.mid = messages.id
+                 )
+                 WHERE session_id = ?1",
+                params![normalized, SEQ_RENUMBER_OFFSET],
+            )?;
+            conn.execute(
+                "UPDATE messages SET seq = seq - ?2 WHERE session_id = ?1",
+                params![normalized, SEQ_RENUMBER_OFFSET],
+            )?;
+        }
+
+        repaired += 1;
+        tracing::info!(
+            polluted_id = %polluted_id,
+            normalized_id = %normalized,
+            merged_into_existing = target_exists,
+            "Repaired polluted session id"
+        );
+    }
+
+    if repaired > 0 {
+        tracing::info!(
+            repaired_count = repaired,
+            "Session id prefix pollution repair complete"
+        );
+    }
+
+    Ok(())
+}
+
 pub(super) fn ensure_todos_schema(conn: &Connection) -> Result<(), StorageError> {
     let table_exists: bool = conn
         .query_row(

src/storage/mod.rs

@@ -65,7 +65,7 @@ impl SessionStore {
     /// Initialize a SessionStore from a connection and its file path.
     /// The connection is used for schema initialization only; the pool
     /// manages subsequent connections using the same file path.
-    fn from_connection(conn: Connection, db_uri: &str) -> Result<Self, StorageError> {
+    fn from_connection(mut conn: Connection, db_uri: &str) -> Result<Self, StorageError> {
         conn.busy_timeout(std::time::Duration::from_secs(30))?;
         conn.execute_batch(
             "
@@ -237,6 +237,7 @@ impl SessionStore {
         ensure_memory_scope_key_migration(&conn)?;
         ensure_todos_schema(&conn)?;
         ensure_pending_subagents_schema(&conn)?;
+        repair_session_id_prefix_pollution(&mut conn)?;
 
         drop(conn);
 

src/storage/tests.rs

@@ -717,3 +717,80 @@ fn test_get_topic_message_count_uses_count_query() {
         0
     );
 }
+
+#[test]
+fn test_repair_session_id_prefix_pollution() {
+    let store = SessionStore::in_memory().unwrap();
+
+    let clean = store.ensure_channel_session("websocket", "abc").unwrap();
+    assert_eq!(clean.id, "abc");
+
+    store
+        .append_message(&clean.id, &ChatMessage::user("clean msg"))
+        .unwrap();
+
+    let mut conn = store.pool.get().unwrap();
+    conn.execute("PRAGMA user_version = 1", []).unwrap();
+
+    conn.execute(
+        "INSERT INTO sessions (id, title, channel_name, chat_id, summary, created_at, updated_at, last_active_at, archived_at, deleted_at, message_count, user_turn_count, agent_prompt_reinjection_count) VALUES ('websocket:websocket:abc', 'Polluted', 'websocket', 'websocket:websocket:abc', NULL, 100, 100, 100, NULL, NULL, 1, 1, 0)",
+        [],
+    )
+    .unwrap();
+    conn.execute(
+        "INSERT INTO messages (id, session_id, seq, role, content, media_refs_json, created_at) VALUES ('m-polluted', 'websocket:websocket:abc', 1, 'user', 'old polluted msg', '[]', 100)",
+        [],
+    )
+    .unwrap();
+
+    conn.execute(
+        "INSERT INTO sessions (id, title, channel_name, chat_id, summary, created_at, updated_at, last_active_at, archived_at, deleted_at, message_count, user_turn_count, agent_prompt_reinjection_count) VALUES ('websocket:xyz', 'PollutedXyz', 'websocket', 'websocket:xyz', NULL, 200, 200, 200, NULL, NULL, 1, 1, 0)",
+        [],
+    )
+    .unwrap();
+    conn.execute(
+        "INSERT INTO messages (id, session_id, seq, role, content, media_refs_json, created_at) VALUES ('m-xyz', 'websocket:xyz', 1, 'user', 'xyz msg', '[]', 200)",
+        [],
+    )
+    .unwrap();
+
+    super::migrations::repair_session_id_prefix_pollution(&mut conn).unwrap();
+
+    assert!(
+        store
+            .get_session("websocket:websocket:abc")
+            .unwrap()
+            .is_none()
+    );
+    assert!(store.get_session("websocket:xyz").unwrap().is_none());
+
+    let merged = store.get_session("abc").unwrap().unwrap();
+    assert_eq!(merged.message_count, 2);
+    assert_eq!(merged.user_turn_count, 2);
+
+    let msgs = store.load_messages("abc").unwrap();
+    assert_eq!(msgs.len(), 2);
+    assert_eq!(msgs[0].content, "old polluted msg");
+
+    {
+        let mut stmt = conn
+            .prepare("SELECT seq FROM messages WHERE session_id = 'abc' ORDER BY seq")
+            .unwrap();
+        let seqs: Vec<i64> = stmt
+            .query_map([], |row| row.get(0))
+            .unwrap()
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+        assert_eq!(seqs, vec![1, 2]);
+    }
+
+    let xyz = store.get_session("xyz").unwrap().unwrap();
+    assert_eq!(xyz.chat_id, "xyz");
+    assert_eq!(store.load_messages("xyz").unwrap().len(), 1);
+
+    assert_eq!(store.list_sessions("websocket", false).unwrap().len(), 2);
+
+    super::migrations::repair_session_id_prefix_pollution(&mut conn).unwrap();
+    assert_eq!(store.get_session("abc").unwrap().unwrap().message_count, 2);
+    assert_eq!(store.list_sessions("websocket", false).unwrap().len(), 2);
+}